fix(plane): fetch task-produced inputs from the receiver after the start

A producer's artifact rendered as spec.files of a generated <task>-inputs
Workspace ends up in ax's AX_WORKSPACES_YAML env value; Agent Substrate
rejects values over 32768 characters, so a 182 KB facts.json broke ax apply
on the live cluster. The reconciler now passes OUSAST_INPUTS instead, the
receiver serves a running task its declared inputs over GET /inputs/..., and
the runner fetches them after the start (failed fetch -> failed summary).

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/plane/egress.py

@@ -27,6 +27,7 @@ import io
 import ipaddress
 import json
 import os
+import shutil
 import subprocess
 import tarfile
 import threading
@@ -35,11 +36,12 @@ from collections.abc import Iterable, Mapping
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Any
-from urllib.parse import urlsplit
+from urllib.parse import unquote, urlsplit
 
 from .manifests import Model, Task, Workspace
 
 __all__ = [
+    "INPUTS_PATH",
     "RECEIVER_HOST",
     "RECEIVER_PORT",
     "Egress",
@@ -52,6 +54,7 @@ __all__ = [
 ]
 
 RECEIVER_HOST = "ousast-receiver.ax-system.svc.cluster.local"
+INPUTS_PATH = "/inputs/"  # GET <receiver>/inputs/<producer>/<artifact>: a consumer fetches a declared input
 RECEIVER_PORT = 18090  # the receiver Service's targetPort; ops/ax/up.sh renders the Service with the same number
 _NOT_FOUND = ("not found", "notfound")
 
@@ -206,19 +209,45 @@ class Egress:
 
 
 class Receiver(ThreadingHTTPServer):
-    """Accepts one tar per running task and extracts it under ``<run dir>/<task>/``."""
+    """Accepts one tar per running task and extracts it under ``<run dir>/<task>/`` (POST); serves a running task
+    the producer artifacts it declared as inputs (``GET /inputs/<producer>/<artifact>``, ``inputs[task]``)."""
 
     daemon_threads = True
 
     def __init__(self, run_name: str, base: Path, port: int) -> None:
         super().__init__(("0.0.0.0", port), _Handler)
         self.run_name, self.base, self.running, self.delivered = run_name, base, set[str](), set[str]()
+        self.inputs: dict[str, frozenset[str]] = {}
         self.lock = threading.Lock()
         self.url, self.dial = receiver_address(self.server_address[1])
 
+    def _task(self, headers: Any) -> str | None:
+        """The ``X-Ousast-Task`` of a request naming this run and a task that is running now, else None."""
+        task = str(headers.get("X-Ousast-Task", ""))
+        with self.lock:
+            return task if headers.get("X-Ousast-Run") == self.run_name and task in self.running else None
+
+    def input_file(self, headers: Any, ref: str) -> tuple[int, str | Path]:
+        """(200, the file) when ``ref`` is one of the requesting task's declared inputs, else (403/404, why).
+
+        ``ref`` must equal a declared ``<producer>/<artifact>`` exactly and resolve to a regular file under the run
+        directory; nothing is normalised first, so ``..`` or an absolute path never names a declared input."""
+        task = self._task(headers)
+        if task is None:
+            return 403, "unknown run or task"
+        with self.lock:
+            declared = ref in self.inputs.get(task, frozenset())
+        if not declared:
+            return 403, f"{ref!r} is not an input of {task}"
+        root = self.base.resolve()
+        path = (root / ref).resolve()
+        if not path.is_relative_to(root) or not path.is_file():
+            return 404, f"{ref} was not delivered"
+        return 200, path
+
     def accept(self, headers: Any, body: bytes) -> tuple[int, str]:
-        task = headers.get("X-Ousast-Task", "")
-        if headers.get("X-Ousast-Run") != self.run_name or task not in self.running:
+        task = self._task(headers)
+        if task is None:
             return 403, "unknown run or task"
         if not str(headers.get("Content-Type", "")).startswith("application/x-tar"):
             return 415, "expected application/x-tar"
@@ -239,14 +268,35 @@ class Receiver(ThreadingHTTPServer):
 
 
 class _Handler(BaseHTTPRequestHandler):
-    def do_POST(self) -> None:
+    def do_GET(self) -> None:  # noqa: N802 - http.server's naming
+        server: Receiver = self.server  # type: ignore[assignment]
+        path = urlsplit(self.path).path
+        if not path.startswith(INPUTS_PATH):
+            self._text(404, "not found")
+            return
+        code, found = server.input_file(self.headers, unquote(path[len(INPUTS_PATH) :]))
+        if not isinstance(found, Path):
+            self._text(code, found)
+            return
+        with found.open("rb") as source:  # streamed: a producer artifact can be far larger than any env value
+            self.send_response(200)
+            self.send_header("Content-Type", "application/octet-stream")
+            self.send_header("Content-Length", str(os.fstat(source.fileno()).st_size))
+            self.end_headers()
+            shutil.copyfileobj(source, self.wfile, 1 << 16)
+
+    def do_POST(self) -> None:  # noqa: N802 - http.server's naming
         server: Receiver = self.server  # type: ignore[assignment]
         body = self.rfile.read(int(self.headers.get("Content-Length") or 0))
-        code, text = server.accept(self.headers, body)
+        self._text(*server.accept(self.headers, body))
+
+    def _text(self, code: int, text: str) -> None:
+        data = text.encode()
         self.send_response(code)
         self.send_header("Content-Type", "text/plain")
+        self.send_header("Content-Length", str(len(data)))
         self.end_headers()
-        self.wfile.write(text.encode())
+        self.wfile.write(data)
 
     def log_message(self, format: str, *args: object) -> None:
         return

src/openultrasast/plane/reconciler.py

@@ -1,19 +1,22 @@
 """ax-backed reconciler of a ``Run``: submission, resume, start, completion by delivery (ai-service-plane Req 3, 4, 7).
 
 ax executes every task on this host's kind cluster; nothing here runs a task locally or knows a prompt, a score or
-a model call. Per ready task: one rendered YAML file (the Task with its env extended, its Workspaces, a generated
-Workspace with the consumed artifacts as ``files``, the bound Model), ``ax apply -f``, ``ax resume task`` (retried
-on DeadlineExceeded/Unavailable -- a new image's first resume waits for Agent Substrate's golden snapshot -- for
-up to ``OUSAST_RESUME_TIMEOUT`` s, 900), and once ax says ``Running`` one start request through Substrate's router
-(``router.py``, Req 4.5) carrying the bound Model's credential from this process's environment, never from a file
-(Req 4.6). Completion is the runner's artifact delivery (ax has no Completed phase); ``Failed``, the task
-vanishing, a refused start or ``OUSAST_TASK_TIMEOUT`` s (7200) without delivery fail it; ``ax delete task`` follows.
+a model call. Per ready task: one rendered YAML file (the Task with its env extended, its Workspaces, the bound
+Model), ``ax apply -f``, ``ax resume task`` (retried on DeadlineExceeded/Unavailable -- a new image's first resume
+waits for Agent Substrate's golden snapshot -- for up to ``OUSAST_RESUME_TIMEOUT`` s, 900), and once ax says
+``Running`` one start request through Substrate's router (``router.py``, Req 4.5) carrying the bound Model's
+credential from this process's environment, never from a file (Req 4.6). Completion is the runner's artifact
+delivery (ax has no Completed phase); ``Failed``, the task vanishing, a refused start or ``OUSAST_TASK_TIMEOUT`` s
+(7200) without delivery fail it; ``ax delete task`` follows.
 
 ax rejects unknown fields, so rendered documents carry only ax's fields; ``openultrasast.io/model`` reaches the
-task as ``OUSAST_MODEL`` plus ``OUSAST_MODEL_PARAMS`` (Req 7.1), each Workspace's ``openultrasast.io/git-commits``
-as ``OUSAST_GIT_PINS``; ``OUSAST_INPUT_<NAME>`` points into ``/workspace/.ousast-in/<task>``. The receiver takes
-the runner's tar (``X-Ousast-Run``, ``X-Ousast-Task``) on ``OUSAST_ARTIFACT_PORT`` (18090), reached by its Service
-name through the egress gateway; between ``ax apply`` and resume the task's egress policy is written (``egress.py``).
+task as ``OUSAST_MODEL`` plus ``OUSAST_MODEL_PARAMS`` (Req 7.1), each Workspace's pin annotation as
+``OUSAST_GIT_PINS``. Inputs other tasks produced are never rendered: ``OUSAST_INPUTS`` maps each input's name to
+``<producer>/<artifact>``, which the runner fetches after the start from the receiver's ``GET /inputs/...`` to
+``OUSAST_INPUT_<NAME>`` under ``/workspace/.ousast-in/<task>`` (the receiver serves a running task its declared
+inputs only). The receiver takes the runner's tar (``X-Ousast-Run``, ``X-Ousast-Task``) on
+``OUSAST_ARTIFACT_PORT`` (18090), reached by its Service name through the egress gateway; between ``ax apply`` and
+resume the task's egress policy is written (``egress.py``).
 State: ``~/ousast-results/plane/<run>/`` (``OUSAST_RESULTS``), polled every ``OUSAST_POLL_SECONDS`` (3).
 """
 
@@ -123,20 +126,10 @@ def _model_doc(model: Model) -> dict[str, Any]:
     return _doc("Model", model.metadata.name, spec, model.metadata.atespace)
 
 
-def _input_files(entry: RunTask, base: Path) -> list[dict[str, str]]:
-    files: list[dict[str, str]] = []
-    for ref in entry.inputs.values():
-        source = base / ref
-        for path in sorted(source.rglob("*")) if source.is_dir() else [source]:
-            if path.is_file():
-                files.append({"path": path.relative_to(base).as_posix(), "content": path.read_text(encoding="utf-8")})
-    return files
-
-
-def render_task(
-    run: Run, entry: RunTask, manifests: Manifests, base: Path, artifact_url: str, dial: str | None = None
-) -> list[dict[str, Any]]:
-    """The documents ``ax apply`` receives for one Run task: Task, its Model, its Workspaces, the inputs Workspace."""
+def render_task(run: Run, entry: RunTask, manifests: Manifests, artifact_url: str, dial: str | None = None) -> list[dict[str, Any]]:
+    """The documents ``ax apply`` receives for one Run task: its Model, its Workspaces, the Task last. Task-produced
+    inputs are never rendered: ax puts every bound Workspace into one env value, which Substrate caps at 32768
+    characters, so the runner fetches them from the receiver (``OUSAST_INPUTS``) after the start."""
     task: Task = manifests.tasks[entry.task]
     ax_name = _ax_name(run.metadata.name, entry.name)
     env = {e.name: e.value for e in task.env}
@@ -144,8 +137,9 @@ def render_task(
     env.update(OUSAST_ARTIFACT_URL=artifact_url, **({"OUSAST_ARTIFACT_DIAL": dial} if dial else {}))
     budget = _fields(entry.budget) if entry.budget else {}
     env.update({f"OUSAST_BUDGET_{k.upper()}": str(v) for k, v in budget.items()})
-    for input_name, ref in entry.inputs.items():
-        env[f"OUSAST_INPUT_{re.sub(r'[^A-Z0-9]+', '_', input_name.upper())}"] = f"{INPUTS_ROOT}/{entry.name}/{ref}"
+    inputs = {re.sub(r"[^A-Z0-9]+", "_", name.upper()): ref for name, ref in entry.inputs.items()}
+    env.update({f"OUSAST_INPUT_{name}": f"{INPUTS_ROOT}/{entry.name}/{ref}" for name, ref in inputs.items()})
+    env.update({"OUSAST_INPUTS": json.dumps(inputs, sort_keys=True)} if inputs else {})
     docs: list[dict[str, Any]] = []
     model_name = task.metadata.annotations.get(MODEL_ANNOTATION)
     if model_name:
@@ -156,9 +150,6 @@ def render_task(
     docs.extend(_workspace_doc(manifests.workspaces[b.name]) for b in task.workspaces)
     pins = {f"{b.name}/{git}": sha for b in task.workspaces for git, sha in manifests.workspaces[b.name].pins.items()}
     env.update({"OUSAST_GIT_PINS": json.dumps(pins, sort_keys=True)} if pins else {})
-    if entry.inputs:
-        docs.append(_doc("Workspace", f"{ax_name}-inputs", {"files": _input_files(entry, base)}, task.metadata.atespace))
-        bindings.append({"name": f"{ax_name}-inputs", "path": f"{INPUTS_ROOT}/{entry.name}"})
     spec: dict[str, Any] = {"command": list(task.command), "env": [{"name": k, "value": v} for k, v in env.items()]}
     if task.image:
         spec["image"] = task.image
@@ -305,7 +296,7 @@ def _execute(
     ax_name, task = _ax_name(run.metadata.name, entry.name), manifests.tasks[entry.task]
     atespace, manifest = task.metadata.atespace or "default", base / entry.name / "task.yaml"
     manifest.parent.mkdir(parents=True, exist_ok=True)
-    docs = render_task(run, entry, manifests, base, receiver.url, receiver.dial)
+    docs = render_task(run, entry, manifests, receiver.url, receiver.dial)
     manifest.write_text(yaml.safe_dump_all(docs, sort_keys=False), encoding="utf-8")
     model = manifests.models.get(task.metadata.annotations.get(MODEL_ANNOTATION) or "")
     policy = policy_for(task, [manifests.workspaces[b.name] for b in task.workspaces], model, urlsplit(receiver.url).hostname)
@@ -362,13 +353,17 @@ def run(run_manifest: Path, *, workers: int = 1, ax: str = "ax", results_root: P
                     pending.remove(entry)
                     running[entry.name] = entry
                     blocked.add(entry.serialize or "")
-                    with receiver.lock:
+                    with receiver.lock:  # the receiver serves a running task its declared inputs, nothing else
                         receiver.running.add(entry.name)
+                        receiver.inputs[entry.name] = frozenset(entry.inputs.values())
                     threading.Thread(target=worker, args=(entry,), daemon=True).start()
                 if not running:
                     break  # nothing can start: a producer stopped short
                 name, status, reason = finished.get()
                 del running[name]
+                with receiver.lock:
+                    receiver.running.discard(name)
+                    receiver.inputs.pop(name, None)
                 summary = _read_json(base / name / "summary.json") or {}
                 state.set(name, status=status, finished=_now(), usd=summary.get("usd"), model=summary.get("model"), reason=reason or None)
                 halted = halted or status == "failed"

src/openultrasast/plane/runner.py

@@ -46,6 +46,10 @@ Environment the runner reads, all optional except the two ax variables:
     OUSAST_ARTIFACT_DIAL               ``host:port`` to connect to instead of the URL's host, which is sent as
                                        ``Host`` (the egress gateway decides on and resolves that name)
     OUSAST_DELIVERY_BACKOFF            seconds before the second delivery attempt, doubled per retry
+    OUSAST_INPUTS                      JSON ``{"<NAME>": "<producer>/<artifact>"}``: after the start and before the
+                                       command each is fetched from ``<OUSAST_ARTIFACT_URL>/inputs/<ref>`` (same
+                                       dial, headers and retries as delivery) to ``OUSAST_INPUT_<NAME>``; a failed
+                                       fetch delivers a failed summary naming the input, the command never runs
 
 The runner exports ``OUSAST_WORKSPACE_DIR`` (the first bound workspace) when the Task's env leaves it unset,
 ``AX_RUNNER_BOUND_PORT`` with the health server's port and ``AX_METADATA_URL``.
@@ -60,6 +64,7 @@ import json
 import logging
 import os
 import re
+import shutil
 import signal
 import subprocess
 import sys
@@ -67,6 +72,7 @@ import tarfile
 import threading
 import time
 import urllib.error
+import urllib.parse
 import urllib.request
 from collections import deque
 from collections.abc import Mapping, MutableMapping, Sequence
@@ -97,6 +103,8 @@ __all__ = [
     "StartGate",
     "build_tar",
     "deliver",
+    "fetch",
+    "fetch_inputs",
     "clone_destination",
     "find_cached_checkout",
     "load_pins",
@@ -635,6 +643,16 @@ def build_tar(output_dir: Path) -> bytes:
     return buffer.getvalue()
 
 
+def _dialled(url: str, headers: Mapping[str, str], dial: str | None) -> tuple[str, dict[str, str]]:
+    """(URL to connect to, headers): with ``dial`` the connection goes there and the URL's host is sent as ``Host``."""
+    out = dict(headers)
+    if not dial:
+        return url, out
+    parts = urlsplit(url)
+    out["Host"] = parts.netloc
+    return parts._replace(netloc=dial).geturl(), out
+
+
 def deliver(
     url: str,
     payload: bytes,
@@ -649,10 +667,7 @@ def deliver(
     ``dial`` (``host:port``): connect there instead and send the URL's host as ``Host``, as the actor must -- it
     cannot resolve the receiver's cluster name, and the egress gateway decides on ``Host`` and resolves it itself.
     """
-    last, target, headers = "no attempt made", url, dict(headers)
-    if dial:
-        parts = urlsplit(url)
-        target, headers["Host"] = parts._replace(netloc=dial).geturl(), parts.netloc
+    last, (target, headers) = "no attempt made", _dialled(url, headers, dial)
     for attempt in range(1, attempts + 1):
         request = urllib.request.Request(target, data=payload, method="POST", headers=headers)
         try:
@@ -671,6 +686,68 @@ def deliver(
     raise DeliveryError(f"{last} after {attempts} attempts to {url}")
 
 
+def fetch(
+    url: str, headers: Mapping[str, str], dest: Path, *, attempts: int = DELIVERY_ATTEMPTS, backoff: float = 1.0, dial: str | None = None
+) -> int:
+    """GET ``url`` into ``dest`` (streamed to a sibling temp file, then renamed); retried like :func:`deliver`.
+
+    Returns the byte count; :class:`DeliveryError` names the last cause when every attempt failed."""
+    last, (target, headers) = "no attempt made", _dialled(url, headers, dial)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    partial = dest.with_name(dest.name + ".part")
+    for attempt in range(1, attempts + 1):
+        request = urllib.request.Request(target, method="GET", headers=headers)
+        try:
+            with urllib.request.urlopen(request, timeout=60) as response, partial.open("wb") as out:  # noqa: S310
+                shutil.copyfileobj(response, out, 1 << 16)
+                size = out.tell()
+            length = response.headers.get("Content-Length")
+            if length is not None and int(length) != size:
+                raise ValueError(f"short read: {size} of {length} bytes")
+            partial.replace(dest)
+            return size
+        except urllib.error.HTTPError as exc:
+            last = f"HTTP {exc.code}"
+        except (urllib.error.URLError, http.client.HTTPException, OSError, ValueError) as exc:
+            last = f"{type(exc).__name__}: {exc}"
+        partial.unlink(missing_ok=True)
+        log.warning("input fetch attempt %d/%d from %s failed: %s", attempt, attempts, url, last)
+        if attempt < attempts:
+            time.sleep(backoff * (2 ** (attempt - 1)))
+    raise DeliveryError(f"{last} after {attempts} attempts to {url}")
+
+
+def fetch_inputs(environ: Mapping[str, str]) -> list[Path]:
+    """Fetch every ``OUSAST_INPUTS`` entry from the receiver to its ``OUSAST_INPUT_<NAME>`` path.
+
+    :class:`RunnerError` names the input when one cannot be fetched: a task never runs with an input missing."""
+    raw = environ.get("OUSAST_INPUTS")
+    if not raw:
+        return []
+    try:
+        inputs = json.loads(raw)
+    except ValueError as exc:
+        raise RunnerError(f"OUSAST_INPUTS is not JSON: {exc}") from None
+    if not isinstance(inputs, dict):
+        raise RunnerError("OUSAST_INPUTS must map input names to <producer>/<artifact>")
+    url = environ.get("OUSAST_ARTIFACT_URL")
+    headers = {"X-Ousast-Run": environ.get("OUSAST_RUN", ""), "X-Ousast-Task": environ.get("OUSAST_TASK", "")}
+    backoff = float(environ.get("OUSAST_DELIVERY_BACKOFF", "1.0"))
+    fetched: list[Path] = []
+    for name, ref in sorted(inputs.items()):
+        dest = environ.get(f"OUSAST_INPUT_{name}")
+        if not url or not dest or not isinstance(ref, str):
+            raise RunnerError(f"input {name} ({ref}) not fetched: OUSAST_ARTIFACT_URL or OUSAST_INPUT_{name} is unset")
+        source = f"{url.rstrip('/')}/inputs/{urllib.parse.quote(ref)}"
+        try:
+            size = fetch(source, headers, Path(dest), backoff=backoff, dial=environ.get("OUSAST_ARTIFACT_DIAL") or None)
+        except DeliveryError as exc:
+            raise RunnerError(f"input {name} ({ref}) not fetched: {exc}") from None
+        log.info("input %s (%s): %d bytes to %s", name, ref, size, dest)
+        fetched.append(Path(dest))
+    return fetched
+
+
 def _deliver_output(task: Task, output_dir: Path, environ: Mapping[str, str]) -> bool:
     """POST the output directory; False when there is nowhere to deliver; :class:`DeliveryError` when it fails."""
     url = environ.get("OUSAST_ARTIFACT_URL")
@@ -751,8 +828,8 @@ def _finish(session: _Session, env: Mapping[str, str], exit_code: int | None, su
 
 
 def _run_once(env: MutableMapping[str, str], ready: threading.Event, stop: threading.Event, gate: StartGate) -> int:
-    """Wait for the start request, then materialise, run the command once and deliver; on a resumed volume, only
-    what the marker says is left. A boot whose marker records a delivered run refuses every start request.
+    """Wait for the start request, then materialise, fetch the inputs, run the command once and deliver; on a
+    resumed volume, only what the marker says is left. A boot whose marker records a delivered run refuses every start request.
 
     Workspaces are prepared only after the start: Substrate's golden boot never gets one, so it touches no network,
     and a started actor is RUNNING with its egress policy in force (a clone at boot failed on the live cluster
@@ -785,12 +862,13 @@ def _run_once(env: MutableMapping[str, str], ready: threading.Event, stop: threa
             return 0
         try:
             _materialise_once(session, env)
+            fetch_inputs(env)  # after the start as well: the receiver serves a task its inputs only while it runs
         except RunnerError as exc:
             log.error("runner failed: %s", exc)
             failed = _failed(f"runner failed: {exc}", read_summary(session.output_dir))
             write_summary(session.output_dir, failed)
             return _exit_of(_finish(session, env, None, failed))
-        log.info("workspaces ready; running task %s: %s", session.task.metadata.name, list(session.task.command))
+        log.info("workspaces and inputs ready; running task %s: %s", session.task.metadata.name, list(session.task.command))
         grace = float(env.get("OUSAST_TERM_GRACE") or TERM_GRACE)
         code, summary = run_command(session.task, session.output_dir, env, stop, grace, start.credentials)
         if code is None: