From 89d000c704131d1b723a533597a5ae8260e555de Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 29 Sep 2026 21:35:28 +0200 Subject: [PATCH] fix(plane): fetch task-produced inputs from the receiver after the start A producer's artifact rendered as spec.files of a generated -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 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- .kiro/specs/ai-service-plane/design.md | 7 +- .kiro/specs/ai-service-plane/tasks.md | 3 + src/openultrasast/plane/egress.py | 64 +++++++++++++++-- src/openultrasast/plane/reconciler.py | 59 +++++++-------- src/openultrasast/plane/runner.py | 92 ++++++++++++++++++++++-- tests/test_plane_increment.py | 2 +- tests/test_plane_reconciler.py | 99 ++++++++++++++++++++++++-- tests/test_plane_runner.py | 83 ++++++++++++++++++++- 8 files changed, 352 insertions(+), 57 deletions(-) diff --git a/.kiro/specs/ai-service-plane/design.md b/.kiro/specs/ai-service-plane/design.md index a1fc2e3..5c104da 100644 --- a/.kiro/specs/ai-service-plane/design.md +++ b/.kiro/specs/ai-service-plane/design.md @@ -56,7 +56,12 @@ container the same module is PID 1 (`ax-task-runner`), built as a target of the for tasks that must not overlap; engine-bearing tasks carry `engine`). The reconciler passes budgets and artifact paths to a task as `env` entries (`OUSAST_BUDGET_USD`, `OUSAST_BUDGET_CALLS`, `OUSAST_INPUT_`, `OUSAST_OUTPUT_DIR`), which is ax's own mechanism for task configuration, so a Run never adds a field to a - Task. + Task. Inputs another task produced are never rendered into a Workspace (ax puts every bound Workspace into one + env value, `AX_WORKSPACES_YAML`, and Agent Substrate rejects values over 32768 characters): the Task carries + `OUSAST_INPUTS` = `{"": "/"}`, and after the start request the runner fetches each + from the receiver's `GET /inputs//` (same headers, dial and retries as delivery; only a + running task's declared inputs are served) to its `OUSAST_INPUT_` path; a failed fetch delivers a + failed summary naming the input and the command never runs. ### Runner (`runner.py`) diff --git a/.kiro/specs/ai-service-plane/tasks.md b/.kiro/specs/ai-service-plane/tasks.md index 68b6356..0883454 100644 --- a/.kiro/specs/ai-service-plane/tasks.md +++ b/.kiro/specs/ai-service-plane/tasks.md @@ -76,6 +76,9 @@ mypy are clean, and the commit is gated on pytest's own exit code. Script-level `ate-target-actor: /`; 502/503/504 and connection errors retried up to `OUSAST_START_TIMEOUT`, a 409 final), carrying the bound Model's `secretKey.key` variable from the reconciler's environment; a Model-bound task whose variable is unset fails before `ax apply`. + Task-produced inputs travel as `OUSAST_INPUTS`, not as a rendered Workspace: the runner fetches them after + the start from the receiver's authorised `GET /inputs//` (a 182 KB `facts.json` in + `AX_WORKSPACES_YAML` exceeded Substrate's 32768-character env limit on the live cluster, 2026-09-29). - `ousast plane run `, `ousast plane status ` (token attribution table per task and Model from the tasks' `summary.json` usage; `--units` from `units.jsonl`; also written to `attribution.json`), `ousast plane workspaces ` (generates Workspace manifests per case pin). diff --git a/src/openultrasast/plane/egress.py b/src/openultrasast/plane/egress.py index 23b64bc..6bc35de 100644 --- a/src/openultrasast/plane/egress.py +++ b/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 /inputs//: 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 ``//``.""" + """Accepts one tar per running task and extracts it under ``//`` (POST); serves a running task + the producer artifacts it declared as inputs (``GET /inputs//``, ``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 ``/`` 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 diff --git a/src/openultrasast/plane/reconciler.py b/src/openultrasast/plane/reconciler.py index 7dca7ed..37a0488 100644 --- a/src/openultrasast/plane/reconciler.py +++ b/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_`` points into ``/workspace/.ousast-in/``. 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 +``/``, which the runner fetches after the start from the receiver's ``GET /inputs/...`` to +``OUSAST_INPUT_`` under ``/workspace/.ousast-in/`` (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//`` (``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" diff --git a/src/openultrasast/plane/runner.py b/src/openultrasast/plane/runner.py index 85c7ae3..22555f1 100644 --- a/src/openultrasast/plane/runner.py +++ b/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 ``{"": "/"}``: after the start and before the + command each is fetched from ``/inputs/`` (same + dial, headers and retries as delivery) to ``OUSAST_INPUT_``; 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_`` 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 /") + 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: diff --git a/tests/test_plane_increment.py b/tests/test_plane_increment.py index c37f6ad..8fbfaec 100644 --- a/tests/test_plane_increment.py +++ b/tests/test_plane_increment.py @@ -73,7 +73,7 @@ def test_validation_run_resolves_and_renders_every_task(tmp_path: Path) -> None: assert model == "deepseek-flash" and entry.budget and entry.budget.usd and entry.budget.calls env = {e.name: e.value for e in task.env} assert env["OUSAST_PASS"] == entry.name[-1] - docs = reconciler.render_task(run, entry, manifests, tmp_path, "http://127.0.0.1:1/") + docs = reconciler.render_task(run, entry, manifests, "http://127.0.0.1:1/") for doc in docs: assert not non_ax_fields(doc["spec"], AX_FIELDS[doc["kind"]], doc["kind"]) assert set(doc["metadata"]) <= {"name", "atespace"} diff --git a/tests/test_plane_reconciler.py b/tests/test_plane_reconciler.py index a6ac04f..5e995ff 100644 --- a/tests/test_plane_reconciler.py +++ b/tests/test_plane_reconciler.py @@ -19,6 +19,8 @@ import sys import tarfile import threading import time +import urllib.error +import urllib.request from collections.abc import Iterator from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -123,9 +125,22 @@ elif argv[:2] == ["get", "task"]: if rec["phase"] == "Running" and rec["polls"] >= script.get("polls", 1) and started and not rec.get("sent"): rec["sent"] = True if script.get("deliver", True) and "phase" not in script and not script.get("vanish"): + files = script.get("files", {{}}) + base = env["OUSAST_ARTIFACT_URL"].rstrip("/") + "/inputs/" + for name, ref in json.loads(env.get("OUSAST_INPUTS", "{{}}")).items(): + fetch = urllib.request.Request(base + ref, headers={{"X-Ousast-Run": env["OUSAST_RUN"], "X-Ousast-Task": task}}) + try: + with urllib.request.urlopen(fetch) as resp: + dest = HERE / "fetched" / task / ref + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(resp.read()) + (HERE / "ax.log").open("a").write(f"{{time.monotonic():.4f}} fetched {{task}} {{name}} {{resp.status}}\\n") + except urllib.error.HTTPError as exc: # the runner delivers a failed summary naming the input + (HERE / "ax.log").open("a").write(f"{{time.monotonic():.4f}} unfetched {{task}} {{name}} {{exc.code}}\\n") + files = {{"summary.json": {{"status": "failed", "reason": f"input {{name}} ({{ref}}) not fetched: HTTP {{exc.code}}"}}}} buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: - for path, content in script.get("files", {{}}).items(): + for path, content in files.items(): data = (content if isinstance(content, str) else json.dumps(content)).encode() info = tarfile.TarInfo(path) info.size = len(data) @@ -273,6 +288,8 @@ class Fake: out.append((stamp, f"{words[1]} {words[2]}")) elif verb in ("delivered", "rejected") and words[0] == verb: out.append((stamp, words[1])) + elif verb in ("fetched", "unfetched") and words[0] == verb: + out.append((stamp, f"{words[1]} {words[2]} {words[3]}")) return out @@ -363,14 +380,16 @@ def test_chain_runs_in_order_with_inputs_env_and_model(fake: Fake, tmp_path: Pat assert env["OUSAST_OUTPUT_DIR"] == "/workspace/.ousast-out/verify" assert env["OUSAST_BUDGET_USD"] == "2.5" and env["OUSAST_BUDGET_CALLS"] == "40" assert env["OUSAST_INPUT_FACTS"] == "/workspace/.ousast-in/verify/repo-facts/facts.json" + assert json.loads(env["OUSAST_INPUTS"]) == {"FACTS": "repo-facts/facts.json"} assert env["OUSAST_ARTIFACT_URL"].startswith("http://127.0.0.1:") assert env["OUSAST_MODEL"] == "deepseek-flash" assert json.loads(env["OUSAST_MODEL_PARAMS"]) == {"cache_hit_per_m": 0.07, "input_per_m": 0.27, "output_per_m": 1.1} kinds = [(d["kind"], d["metadata"]["name"]) for d in verify["docs"]] - assert kinds == [("Model", "deepseek-flash"), ("Workspace", "case-pin"), ("Workspace", "chain-verify-inputs"), ("Task", "chain-verify")] - inputs_ws = next(d for d in verify["docs"] if d["metadata"]["name"] == "chain-verify-inputs") - assert inputs_ws["spec"]["files"] == [{"path": "repo-facts/facts.json", "content": json.dumps({"a.py": {"functions": ["f"]}})}] - assert verify["docs"][-1]["spec"]["workspaces"][-1] == {"name": "chain-verify-inputs", "path": "/workspace/.ousast-in/verify"} + assert kinds == [("Model", "deepseek-flash"), ("Workspace", "case-pin"), ("Task", "chain-verify")], "no inputs Workspace" + assert verify["docs"][-1]["spec"]["workspaces"] == [{"name": "case-pin", "path": "/workspace"}] + assert [n for _, n in fake.events("fetched")] == ["verify FACTS 200", "agree PASS 200"], "inputs come over the receiver's GET" + assert json.loads((tmp_path / "fetched" / "verify" / "repo-facts" / "facts.json").read_text()) == {"a.py": {"functions": ["f"]}} + assert json.loads((tmp_path / "fetched" / "agree" / "verify" / "agreed.json").read_text()) == {"agreed": ["a.py:f"]} assert "annotations" not in verify["docs"][-1]["metadata"], "ax's ObjectMeta has no annotations" assert json.loads(env["OUSAST_GIT_PINS"]) == {"case-pin/repo": PIN}, "the Workspace annotation reaches the task as env" assert next(d for d in verify["docs"] if d["metadata"]["name"] == "case-pin")["spec"]["git"] == [ @@ -690,7 +709,7 @@ def test_rendered_documents_carry_only_ax_fields(tmp_path: Path) -> None: (tmp_path / "repo-facts").mkdir() (tmp_path / "repo-facts" / "facts.json").write_text("{}", encoding="utf-8") run, manifests = reconciler.load_run(write_run(tmp_path, "fields")) - docs = [d for entry in run.tasks for d in reconciler.render_task(run, entry, manifests, tmp_path, "http://h:1/")] + docs = [d for entry in run.tasks for d in reconciler.render_task(run, entry, manifests, "http://h:1/")] assert {d["kind"] for d in docs} == {"Task", "Workspace", "Model"} for rendered in docs: where = f"{rendered['kind']}/{rendered['metadata']['name']}" @@ -707,7 +726,7 @@ def test_the_task_is_applied_after_everything_it_binds(tmp_path: Path) -> None: (tmp_path / "repo-facts" / "facts.json").write_text("{}", encoding="utf-8") run, manifests = reconciler.load_run(write_run(tmp_path, "order")) for entry in run.tasks: - docs = reconciler.render_task(run, entry, manifests, tmp_path, "http://h:1/") + docs = reconciler.render_task(run, entry, manifests, "http://h:1/") assert [d["kind"] for d in docs].count("Task") == 1 assert docs[-1]["kind"] == "Task", [d["kind"] for d in docs] @@ -780,3 +799,69 @@ def test_a_refused_egress_policy_fails_the_task_before_resume(fake: Fake, tmp_pa assert reconciler.run(write_run(tmp_path, "egfail"), ax=str(fake.ax)) == "failed" assert "permission denied" in state_of("egfail")["tasks"]["repo-facts"]["reason"] assert fake.events("resume") == [] and [n for _, n in fake.events("delete")] == ["egfail-repo-facts"] + + +# --- inputs over the receiver's GET (Req 3.1, 4.4) ---------------------------------------------------------------- + +SUBSTRATE_ENV_LIMIT = 32768 # "actor_template.containers[0].env[11].value: Too long" on the live cluster, 2026-09-29 + + +def get_input(receiver: reconciler.Receiver, path: str, run: str = "r", task: str = "t") -> tuple[int, bytes]: + url = f"http://127.0.0.1:{receiver.server_address[1]}{path}" + request = urllib.request.Request(url, headers={"X-Ousast-Run": run, "X-Ousast-Task": task}) + try: + with urllib.request.urlopen(request, timeout=10) as response: + return response.status, response.read() + except urllib.error.HTTPError as exc: + return exc.code, exc.read() + + +def test_receiver_serves_a_running_task_its_declared_inputs_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OUSAST_ARTIFACT_HOST", "127.0.0.1") + base = tmp_path / "r" + (base / "p").mkdir(parents=True) + big = b"x" * (3 << 20) + (base / "p" / "a.json").write_bytes(big) + (base / "p" / "b.json").write_bytes(b"not declared") + (tmp_path / "secret.txt").write_text("outside the run") + receiver = reconciler.Receiver("r", base, 0) + threading.Thread(target=receiver.serve_forever, daemon=True).start() + try: + receiver.running.update({"t", "idle"}) + receiver.inputs.update({"t": frozenset({"p/a.json", "p/missing.json", "../secret.txt"}), "done": frozenset({"p/a.json"})}) + assert get_input(receiver, "/inputs/p/a.json") == (200, big), "streamed whole" + assert get_input(receiver, "/inputs/p/a.json", run="other")[0] == 403, "wrong run" + assert get_input(receiver, "/inputs/p/a.json", task="done")[0] == 403, "a task that is not running" + assert get_input(receiver, "/inputs/p/a.json", task="idle")[0] == 403, "a running task without that input" + assert get_input(receiver, "/inputs/p/b.json")[0] == 403, "not among the task's inputs" + assert get_input(receiver, "/inputs/p/missing.json")[0] == 404, "declared but not delivered" + for escape in ("/inputs/../secret.txt", "/inputs/%2e%2e/secret.txt", "/inputs/p/../../secret.txt", "/inputs//etc/passwd"): + code, body = get_input(receiver, escape) + assert code in (403, 404) and b"outside" not in body, escape + assert get_input(receiver, "/p/a.json")[0] == 404, "only /inputs/ is served" + finally: + receiver.shutdown() + receiver.server_close() + + +def test_a_missing_producer_artifact_fails_the_consumer_instead_of_running_it(fake: Fake, tmp_path: Path) -> None: + script = json.loads(json.dumps(SCRIPT)) + del script["repo-facts"]["files"]["facts.json"] # repo-facts reports done but never delivered facts.json + fake.script(script) + assert reconciler.run(write_run(tmp_path, "nofacts"), ax=str(fake.ax)) == "failed" + assert [n for _, n in fake.events("unfetched")] == ["verify FACTS 404"] + assert "input FACTS (repo-facts/facts.json) not fetched" in state_of("nofacts")["tasks"]["verify"]["reason"] + + +def test_a_megabyte_producer_artifact_keeps_every_rendered_value_under_substrates_limit(tmp_path: Path) -> None: + """Regression: a 182 KB facts.json rendered into an inputs Workspace made ax's AX_WORKSPACES_YAML too long.""" + (tmp_path / "repo-facts").mkdir() + (tmp_path / "repo-facts" / "facts.json").write_text(json.dumps({"blob": "f" * (1 << 20)}), encoding="utf-8") + run, manifests = reconciler.load_run(write_run(tmp_path, "big")) + for entry in run.tasks: + docs = reconciler.render_task(run, entry, manifests, "http://h:1/", "10.0.0.1:80") + task, workspaces = docs[-1], [d for d in docs if d["kind"] == "Workspace"] + assert all(len(e["value"]) < SUBSTRATE_ENV_LIMIT for e in task["spec"]["env"]), entry.name + assert len(yaml.safe_dump(task)) < SUBSTRATE_ENV_LIMIT, f"{entry.name}: AX_TASK_YAML" + assert len(yaml.safe_dump_all(workspaces)) < SUBSTRATE_ENV_LIMIT, f"{entry.name}: AX_WORKSPACES_YAML" + assert all(len(yaml.safe_dump(w)) < SUBSTRATE_ENV_LIMIT for w in workspaces), entry.name diff --git a/tests/test_plane_runner.py b/tests/test_plane_runner.py index 645f5fd..f2f2678 100644 --- a/tests/test_plane_runner.py +++ b/tests/test_plane_runner.py @@ -24,7 +24,7 @@ from pathlib import Path import pytest -from openultrasast.plane import runner +from openultrasast.plane import egress, runner from openultrasast.plane.manifests import GitSource from openultrasast.plane.router import Router, StartError from openultrasast.plane.runner import ( @@ -588,7 +588,7 @@ def test_git_commits_annotation_round_trips_to_the_checkout( "apiVersion: openultrasast.io/v1alpha1\nkind: Run\nmetadata: {name: rt}\nspec: {tasks: [{name: stub, task: stub}]}\n" ) run, manifests = reconciler.load_run(manifest) - *rest, task_doc = reconciler.render_task(run, run.tasks[0], manifests, tmp_path, "unused") + *rest, task_doc = reconciler.render_task(run, run.tasks[0], manifests, "unused") assert rest[0]["metadata"] == {"name": "case"}, "annotations are stripped before ax" assert rest[0]["spec"]["git"] == [{"name": "code", "repo": repo.as_uri(), "branch": "main"}], "no commit for ax" overrides = {"OUSAST_OUTPUT_DIR": str(output)} @@ -771,3 +771,82 @@ def test_workspaces_are_prepared_only_after_the_start_request(env: dict[str, str thread.join(timeout=30) assert codes == [0] assert [p for p in (tmp_path / "ws").rglob("*") if p.is_file()], "the start materialises the workspace" + + +# --- inputs other tasks produced: fetched from the receiver after the start (Requirement 3.1, 4.4) ------------------- + + +@pytest.fixture +def plane_receiver(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Iterator[egress.Receiver]: + """The reconciler's own receiver on a free port, with a producer's artifact delivered and the task ``stub`` running.""" + monkeypatch.setenv("OUSAST_ARTIFACT_HOST", "127.0.0.1") + server = egress.Receiver("r", tmp_path / "run", 0) + (tmp_path / "run" / "repo-facts").mkdir(parents=True) + (tmp_path / "run" / "repo-facts" / "facts.json").write_bytes(b'{"a.py": ["f"]}' * 5000) + server.running.add("stub") + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + yield server + finally: + server.shutdown() + server.server_close() + + +def input_env(server: egress.Receiver, tmp_path: Path) -> dict[str, str]: + return { + "OUSAST_ARTIFACT_URL": server.url, + "OUSAST_RUN": "r", + "OUSAST_TASK": "stub", + "OUSAST_INPUTS": json.dumps({"FACTS": "repo-facts/facts.json"}), + "OUSAST_INPUT_FACTS": str(tmp_path / "ws" / "in" / "repo-facts" / "facts.json"), + } + + +def test_inputs_are_fetched_from_the_receiver_before_the_command( + env: dict[str, str], plane_receiver: egress.Receiver, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + plane_receiver.inputs["stub"] = frozenset({"repo-facts/facts.json"}) + code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), **input_env(plane_receiver, tmp_path)) + assert code == 0 + fetched = tmp_path / "ws" / "in" / "repo-facts" / "facts.json" + assert fetched.read_bytes() == (tmp_path / "run" / "repo-facts" / "facts.json").read_bytes() + assert "in/repo-facts/facts.json" in json.loads((output / "facts.json").read_text())["workspace_files"], "there before the command ran" + assert "stub" in plane_receiver.delivered + + +def test_an_input_that_cannot_be_fetched_delivers_a_failed_summary_and_never_runs_the_command( + env: dict[str, str], plane_receiver: egress.Receiver, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + plane_receiver.inputs["stub"] = frozenset({"repo-facts/other.json"}) # facts.json is not among its inputs: 403 + code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), **input_env(plane_receiver, tmp_path)) + assert code == EXIT_BY_STATUS["failed"] + assert not (output / "facts.json").exists(), "the command never ran with an input missing" + delivered = json.loads((tmp_path / "run" / "stub" / "summary.json").read_text()) + assert delivered["status"] == "failed" and "input FACTS (repo-facts/facts.json) not fetched: HTTP 403" in delivered["reason"] + assert not (tmp_path / "ws" / "in" / "repo-facts").exists() or not any((tmp_path / "ws" / "in" / "repo-facts").iterdir()) + + +def test_fetch_dials_the_override_and_sends_the_url_host(tmp_path: Path) -> None: + seen: list[dict[str, str]] = [] + + class _Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 + seen.append({"path": self.path, **dict(self.headers.items())}) + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"ok") + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + dial = f"127.0.0.1:{server.server_address[1]}" + size = runner.fetch("http://receiver.example/inputs/p/a.json", {"X-Ousast-Run": "r"}, tmp_path / "x" / "a.json", dial=dial) + finally: + server.shutdown() + server.server_close() + assert size == 2 and (tmp_path / "x" / "a.json").read_bytes() == b"ok" + assert seen[0]["path"] == "/inputs/p/a.json" and seen[0]["Host"] == "receiver.example" and seen[0]["X-Ousast-Run"] == "r"