diff --git a/.kiro/specs/plane-on-kubernetes/tasks.md b/.kiro/specs/plane-on-kubernetes/tasks.md index 817f757..7453605 100644 --- a/.kiro/specs/plane-on-kubernetes/tasks.md +++ b/.kiro/specs/plane-on-kubernetes/tasks.md @@ -93,7 +93,7 @@ go-ahead** with a ceiling; steps that need a hand on a cluster that does not exi - `tests/test_plane_egress.py`: for every committed template under `plane/tasks/` the rendered policy is the expected set; the IP refusal. - _Requirements: 3.2_ -- [ ] 2.4 Reconciler on `Delivery` +- [x] 2.4 Reconciler on `Delivery` - `src/openultrasast/plane/reconciler.py`: `run()` opens the store first (no fallback), builds `Delivery`, passes `delivery` to `router.starter`/`start_task` (body extended in `router.py`), `_await` on `Delivery.delivered`, outcome from the collected `summary.json`; the receiver thread is kept behind diff --git a/src/openultrasast/plane/delivery.py b/src/openultrasast/plane/delivery.py index 1e704db..420dcf3 100644 --- a/src/openultrasast/plane/delivery.py +++ b/src/openultrasast/plane/delivery.py @@ -31,9 +31,10 @@ from pathlib import Path from typing import Any from urllib.parse import urlsplit +from .manifests import Run from .memory import RUNS_PREFIX, MemoryStore -__all__ = ["DONE", "GRACE_SECONDS", "OUTPUT_TAR", "STATE", "Delivery", "DeliveryError", "expiry", "input_var"] +__all__ = ["DONE", "GRACE_SECONDS", "OUTPUT_TAR", "STATE", "Delivery", "DeliveryError", "declared", "expiry", "input_var"] OUTPUT_TAR = "output.tar" DONE = "done.json" @@ -62,6 +63,18 @@ def _now() -> str: return datetime.now(UTC).isoformat(timespec="seconds") +def declared(run: Run) -> dict[str, list[str]]: + """Per Run task, the artifacts to publish as single objects: its ``outputs[]`` plus every ``/`` + a consumer declares as an input on it (``summary.json`` of a producer is a usual one), in declaration order.""" + wanted: dict[str, list[str]] = {entry.name: list(entry.outputs) for entry in run.tasks} + for entry in run.tasks: + for ref in entry.inputs.values(): + producer, _, artifact = ref.partition("/") + if producer in wanted and artifact and artifact not in wanted[producer]: + wanted[producer].append(artifact) + return wanted + + class Delivery: """Presigned URLs, polling, collection and state mirroring for one Run on one store.""" @@ -98,6 +111,12 @@ class Delivery: # --- completion ---------------------------------------------------------------------------------------------- + def reset(self, task: str) -> None: + """Before an attempt: remove the previous attempt's tar and ``done.json`` so a rerun (an unfinished task + resumed with a larger budget, an interrupted attempt) never completes on a stale object.""" + for name in (OUTPUT_TAR, DONE): + self.store._delete(self.key(task, name)) + def delivered(self, task: str) -> bool: """Whether the runner's tar is in the store (an exact-key listing, the store's HEAD).""" key = self.key(task, OUTPUT_TAR) diff --git a/src/openultrasast/plane/reconciler.py b/src/openultrasast/plane/reconciler.py index efab978..279d893 100644 --- a/src/openultrasast/plane/reconciler.py +++ b/src/openultrasast/plane/reconciler.py @@ -6,22 +6,21 @@ env extended, its Workspaces, the bound Model), ``ax apply -f``, ``ax resume tas 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. +credential from this process's environment, never from a file (Req 4.6), and the presigned store URLs of +``delivery.py`` (plane-on-kubernetes design section 3). Completion is the runner's output tar appearing in the +memory store (ax has no Completed phase), which the reconciler polls, downloads, extracts and re-publishes per +declared output; ``Failed``, the task vanishing, a refused start or ``OUSAST_TASK_TIMEOUT`` s (7200) without +delivery fail it; ``ax delete task`` follows. ``OUSAST_DELIVERY=receiver`` keeps the host receiver of the kind +profile for group 2 only (task 2.6 removes it). 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 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); ``state.json`` -is mirrored to ``runs//state.json`` on the memory store after every change and a rerun with no local state -seeds it from there (``delivery.py``; plane-on-kubernetes design section 2). The store is opened first and is -required: there is no fallback. +``/``, fetched after the start by the presigned GET of the start body to ``OUSAST_INPUT_`` +under ``/workspace/.ousast-in/``. 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), mirrored to ``runs//state.json`` on the store after every change and seeded from there by a rerun with no +local state (design section 2); the store is opened first and required, there is no fallback. """ from __future__ import annotations @@ -45,7 +44,7 @@ from urllib.parse import urlsplit import yaml -from .delivery import Delivery +from .delivery import Delivery, DeliveryError, declared from .doctor import doctor from .egress import RECEIVER_PORT, Egress, Receiver, policy_for from .manifests import AX_API_VERSION, Manifests, Model, Run, RunTask, Task, Workspace, load_manifests @@ -143,15 +142,16 @@ def _model_doc(model: Model) -> dict[str, Any]: return _doc("Model", model.metadata.name, spec, model.metadata.atespace) -def render_task(run: Run, entry: RunTask, manifests: Manifests, artifact_url: str, dial: str | None = None) -> list[dict[str, Any]]: +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.""" + inputs are never rendered (Substrate caps one env value at 32768 characters): the runner fetches them + (``OUSAST_INPUTS``) after the start by presigned URL, or from the receiver when ``artifact_url`` is given.""" task: Task = manifests.tasks[entry.task] ax_name = _ax_name(run.metadata.name, entry.name, limit=TASK_NAME_LIMIT) env = {e.name: e.value for e in task.env} env.update(OUSAST_RUN=run.metadata.name, OUSAST_TASK=entry.name, OUSAST_OUTPUT_DIR=f"{OUTPUT_ROOT}/{entry.name}") - env.update(OUSAST_ARTIFACT_URL=artifact_url, **({"OUSAST_ARTIFACT_DIAL": dial} if dial else {})) + if artifact_url: # receiver mode only (task 2.6 removes it): the store path carries no URL in the env + 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()}) inputs = {re.sub(r"[^A-Z0-9]+", "_", name.upper()): ref for name, ref in entry.inputs.items()} @@ -333,25 +333,46 @@ def _await(ax: Ax, name: str, delivered: Callable[[], bool], start: Callable[[], def _execute( - ax: Ax, egress: Egress, run: Run, entry: RunTask, manifests: Manifests, base: Path, receiver: Receiver, router: Router + ax: Ax, + egress: Egress, + run: Run, + entry: RunTask, + manifests: Manifests, + base: Path, + delivery: Delivery, + router: Router, + receiver: Receiver | None = None, ) -> tuple[str, str]: ax_name, task = _ax_name(run.metadata.name, entry.name, limit=TASK_NAME_LIMIT), 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, receiver.url, receiver.dial) + docs = render_task(run, entry, manifests, receiver.url if receiver else "", receiver.dial if receiver else None) 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, None, urlsplit(receiver.url).hostname) + workspaces = [manifests.workspaces[b.name] for b in task.workspaces] + body = None if receiver else delivery.body(entry.name, entry.inputs) # presigned URLs ride in the start body + store_host, receiver_host = (None, urlsplit(receiver.url).hostname) if receiver else (delivery.host, None) + if receiver is None: + delivery.reset(entry.name) # a previous attempt's tar must not complete this one + + def delivered() -> bool: + return entry.name in receiver.delivered if receiver is not None else delivery.delivered(entry.name) + try: # Req 4.6: the bound Model's credential is read here and only ever sent in the start request - start = router.starter(atespace, ax_name, run.metadata.name, credentials(model)) + policy = policy_for(task, workspaces, model, store_host, receiver_host) + start = router.starter(atespace, ax_name, run.metadata.name, credentials(model), body) ax.apply(manifest) # creates the Substrate actor; its egress is denied until the policy exists egress.apply(ax_name, atespace, policy) egress.settle() - except RuntimeError as exc: # StartError included: a Model-bound task whose secret is unset + except (RuntimeError, ValueError) as exc: # StartError (a Model's secret unset) and a policy that cannot be granted ax.delete(ax_name) return "failed", str(exc) try: - failure = _await(ax, ax_name, lambda: entry.name in receiver.delivered, start) + failure = _await(ax, ax_name, delivered, start) + if failure is None and receiver is None: + delivery.collect(entry.name, declared(run)[entry.name]) # download, extract, re-publish what is declared + except DeliveryError as exc: + failure = str(exc) finally: ax.delete(ax_name) leftover = egress.delete(ax_name, atespace) # the policy goes with its actor; a survivor is reported @@ -387,13 +408,17 @@ def run( _read_json(base / "state.json") or {"run": run_spec.metadata.name, "started": _now(), "tasks": {}}, mirror=delivery.save_state, ) - port = os.environ.get("OUSAST_ARTIFACT_PORT") or (0 if os.environ.get("OUSAST_ARTIFACT_HOST") else RECEIVER_PORT) - receiver = Receiver(run_spec.metadata.name, base, int(port), profile.kube_context) # the Service targets RECEIVER_PORT - threading.Thread(target=receiver.serve_forever, daemon=True).start() + receiver: Receiver | None = None + if os.environ.get("OUSAST_DELIVERY") == "receiver": # the kind profile's host receiver, group 2 only (2.6 removes it) + port = os.environ.get("OUSAST_ARTIFACT_PORT") or (0 if os.environ.get("OUSAST_ARTIFACT_HOST") else RECEIVER_PORT) + receiver = Receiver(run_spec.metadata.name, base, int(port), profile.kube_context) # the Service targets RECEIVER_PORT + threading.Thread(target=receiver.serve_forever, daemon=True).start() cli, egress, finished = Ax(ax), Egress(kubectl_ate, profile.kube_context), queue.Queue[tuple[str, str, str]]() pending = [t for t in run_spec.tasks if state.status(t.name) != "done"] for entry in run_spec.tasks: state.set(entry.name, **({"status": "pending"} if entry in pending else {})) + if entry not in pending: + delivery.publish(entry.name, declared(run_spec)[entry.name]) # a seeded or earlier task's artifacts reach its consumers running: dict[str, RunTask] = {} halted = False @@ -401,7 +426,11 @@ def run( if state.status(entry.name) in ("running", "suspended"): cli.delete(_ax_name(run_spec.metadata.name, entry.name, limit=TASK_NAME_LIMIT)) # an interrupted attempt; ax must not keep it state.set(entry.name, status="running", started=_now(), finished=None) - finished.put((entry.name, *_execute(cli, egress, run_spec, entry, manifests, base, receiver, router))) + try: + outcome = _execute(cli, egress, run_spec, entry, manifests, base, delivery, router, receiver) + except Exception as exc: # noqa: BLE001 - a worker that dies silently would hang the run on the queue + outcome = ("failed", f"{type(exc).__name__}: {exc}") + finished.put((entry.name, *outcome)) try: with open_router(profile.kube_context, url=profile.router_url or None) as router: # one tunnel per run (ax-tunnel) @@ -415,24 +444,27 @@ def run( pending.remove(entry) running[entry.name] = entry blocked.add(entry.serialize or "") - 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()) + if receiver is not None: + 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) + if receiver is not None: + 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" if status == "unfinished": pending = [t for t in pending if name not in t.producers] finally: - receiver.shutdown() + if receiver is not None: + receiver.shutdown() lock.unlink(missing_ok=True) statuses = [state.status(t.name) for t in run_spec.tasks] result = "failed" if "failed" in statuses else "done" if all(s == "done" for s in statuses) else "unfinished" diff --git a/tests/test_plane_delivery.py b/tests/test_plane_delivery.py index e3d64d8..507c4ba 100644 --- a/tests/test_plane_delivery.py +++ b/tests/test_plane_delivery.py @@ -15,7 +15,7 @@ from pathlib import Path import pytest from plane_fake_store import HttpObjectStore -from openultrasast.plane.delivery import GRACE_SECONDS, Delivery, DeliveryError, expiry, input_var +from openultrasast.plane.delivery import GRACE_SECONDS, Delivery, DeliveryError, declared, expiry, input_var @pytest.fixture @@ -92,6 +92,9 @@ def test_delivered_polls_the_store_and_collect_extracts_republishes_and_marks_do with urllib.request.urlopen(urls["FACTS"], timeout=10) as response: # noqa: S310 assert response.read() == b'{"a": 1}' assert "X-Amz" not in caplog.text and "fake-bucket" not in caplog.text, "URLs never appear in logs" + delivery.reset("facts") # the next attempt must not complete on this attempt's tar + assert delivery.delivered("facts") is False and "runs/r/facts/done.json" not in store.objects + assert "runs/r/facts/facts.json" in store.objects, "a re-put output stays for its consumers until it is re-published" @pytest.mark.parametrize("bad", ["../escape.json", "/abs.json"]) @@ -136,6 +139,22 @@ def test_put_input_stores_a_file_no_task_produces(store: HttpObjectStore, tmp_pa assert response.read() == b"[1, 2, 3]" +def test_declared_artifacts_are_outputs_plus_what_consumers_name(tmp_path: Path) -> None: + """A consumer may declare a producer's summary.json (every task writes one) without the producer listing it.""" + from openultrasast.plane.manifests import load_manifests + + text = ( + "apiVersion: ax.io/v1alpha1\nkind: Task\nmetadata: {name: t}\nspec: {command: [repo-facts]}\n---\n" + "apiVersion: openultrasast.io/v1alpha1\nkind: Run\nmetadata: {name: r}\nspec:\n tasks:\n" + " - {name: facts, task: t, outputs: [facts.json]}\n" + " - {name: verify, task: t, inputs: {facts: facts/facts.json, facts_summary: facts/summary.json}, outputs: [units.jsonl]}\n" + " - {name: agree, task: t, inputs: {pass: verify/units.jsonl, pass_summary: verify/summary.json}}\n" + ) + (tmp_path / "run.yaml").write_text(text, encoding="utf-8") + run = next(iter(load_manifests([tmp_path / "run.yaml"]).runs.values())) + assert declared(run) == {"facts": ["facts.json", "summary.json"], "verify": ["units.jsonl", "summary.json"], "agree": []} + + def test_publish_reputs_only_declared_files_under_the_task_directory(store: HttpObjectStore, tmp_path: Path) -> None: delivery = Delivery(store, "r", tmp_path / "r") (tmp_path / "r" / "seeded").mkdir(parents=True) diff --git a/tests/test_plane_reconciler.py b/tests/test_plane_reconciler.py index 2e716b2..9375a8a 100644 --- a/tests/test_plane_reconciler.py +++ b/tests/test_plane_reconciler.py @@ -4,9 +4,10 @@ The fake is a Python script the tests write into ``tmp_path`` that follows ax's internal/controller/reconciler.go, internal/server/server.go): ``apply`` creates the task ``Suspended``; ``resume`` fails with DeadlineExceeded the first time (ax then reports ``Failed``, as while Agent Substrate builds a new image's golden snapshot) and succeeds after; ``get task`` answers ``Running`` forever -- ax has no Completed phase --- and on the scripted poll it POSTs a scripted tar to the ``OUSAST_ARTIFACT_URL`` in the applied Task's env, or -turns ``Failed`` with a condition message, or vanishes. Every outcome comes from a scripted delivery, and a -delivery happens only after the reconciler's start request reached the fake router (Req 4.5). +-- and on the scripted poll it PUTs a scripted tar to the presigned ``delivery.put`` URL the start request carried +(plane-on-kubernetes 2.4; or POSTs it to ``OUSAST_ARTIFACT_URL`` in receiver mode), or turns ``Failed`` with a +condition message, or vanishes. Every outcome comes from a scripted delivery, and a delivery happens only after +the reconciler's start request reached the fake router (Req 4.5). The store is ``plane_fake_store.HttpObjectStore``. """ from __future__ import annotations @@ -28,6 +29,7 @@ from pathlib import Path import pytest import yaml +from plane_fake_store import HttpObjectStore from openultrasast.plane import reconciler from openultrasast.plane.workspaces import workspaces @@ -146,11 +148,15 @@ elif argv[:2] == ["get", "task"]: started = (HERE / "started" / argv[2]).exists() # the runner runs its command only after the start request if rec["phase"] == "Running" and rec["polls"] >= script.get("polls", 1) and started and not rec.get("sent"): rec["sent"] = True + body = json.loads((HERE / "started" / argv[2]).read_text() or "{{}}") # the start body the router saw + delivery = body.get("delivery") or {{}} 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/" + base = env.get("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}}) + presigned = (delivery.get("inputs") or {{}}).get(name) + fetch = urllib.request.Request(presigned) if presigned else 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 @@ -171,7 +177,11 @@ elif argv[:2] == ["get", "task"]: tar.addfile(info, io.BytesIO(data)) headers = {{"Content-Type": "application/x-tar", "X-Ousast-Run": env["OUSAST_RUN"]}} headers["X-Ousast-Task"] = script.get("as_task", task) - req = urllib.request.Request(env["OUSAST_ARTIFACT_URL"], data=buf.getvalue(), method="POST", headers=headers) + if delivery.get("put"): # the store path: PUT to the presigned URL, no receiver headers + tar_headers = {{"Content-Type": "application/x-tar"}} + req = urllib.request.Request(delivery["put"], data=buf.getvalue(), method="PUT", headers=tar_headers) + else: + req = urllib.request.Request(env["OUSAST_ARTIFACT_URL"], data=buf.getvalue(), method="POST", headers=headers) try: with urllib.request.urlopen(req) as resp: (HERE / "ax.log").open("a").write(f"{{time.monotonic():.4f}} delivered {{task}} {{resp.status}}\\n") @@ -283,6 +293,7 @@ SCRIPT = { class Fake: router: FakeRouter + store: HttpObjectStore def __init__(self, root: Path) -> None: self.root = root @@ -343,7 +354,8 @@ class _RouterHandler(BaseHTTPRequestHandler): self.server.starts.append({"at": time.monotonic(), "path": self.path, "target": target, "body": body, "code": code}) if code == 202: (self.server.root / "started").mkdir(exist_ok=True) - (self.server.root / "started" / name).touch() + without_secrets = {k: v for k, v in body.items() if k != "credentials"} # the fake ax reads delivery from it + (self.server.root / "started" / name).write_text(json.dumps(without_secrets)) with (self.server.root / "ax.log").open("a") as log: log.write(f"{time.monotonic():.4f} started {name} {code}\n") self.send_response(code) @@ -366,12 +378,15 @@ def fake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, kubectl_ate: Path, pla fake = Fake(tmp_path) fake.script(SCRIPT) fake.router = FakeRouter(tmp_path) + fake.store = HttpObjectStore() # the memory store the reconciler opens first: S3-shaped, presigned URLs served locally + monkeypatch.setattr(reconciler, "open_store", lambda spec=None, environ=None: fake.store) monkeypatch.setenv("OUSAST_ROUTER_URL", f"http://127.0.0.1:{fake.router.server_address[1]}") try: yield fake finally: fake.router.shutdown() fake.router.server_close() + fake.store.close() def write_run(tmp_path: Path, name: str, tasks: str = CHAIN) -> Path: @@ -405,13 +420,17 @@ def test_chain_runs_in_order_with_inputs_env_and_model(fake: Fake, tmp_path: Pat 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 "OUSAST_ARTIFACT_URL" not in env, "the store path renders no receiver address into the Task" 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"), ("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 [n for _, n in fake.events("fetched")] == ["verify FACTS 200", "agree PASS 200"], "inputs come by their presigned GET" + assert ("GET", "runs/chain/repo-facts/facts.json") in fake.store.requests and ( + "PUT", + "runs/chain/verify/output.tar", + ) in fake.store.requests 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" @@ -570,7 +589,15 @@ def test_start_is_sent_once_after_resume_with_the_models_credential(fake: Fake, assert [s["target"] for s in starts] == ["default/sig-repo-facts", "default/sig-verify", "default/sig-agree"], "once each" assert all(s["path"] == "/ousast/v1/start" and s["code"] == 202 for s in starts) by_task = {s["body"]["task"]: s for s in starts} - assert by_task["sig-verify"]["body"] == {"run": "sig", "task": "sig-verify", "credentials": {"DEEPSEEK_API_KEY": SECRET}} + body = by_task["sig-verify"]["body"] + assert {k: body[k] for k in ("run", "task", "credentials")} == { + "run": "sig", + "task": "sig-verify", + "credentials": {"DEEPSEEK_API_KEY": SECRET}, + } + assert set(body["delivery"]) == {"put", "inputs"} and "/runs/sig/verify/output.tar?" in body["delivery"]["put"] + assert set(body["delivery"]["inputs"]) == {"FACTS"} and "/runs/sig/repo-facts/facts.json?" in body["delivery"]["inputs"]["FACTS"] + assert all("X-Amz" not in line for line in (tmp_path / "ax.log").read_text().splitlines()), "no presigned URL in the fake's log" assert by_task["sig-repo-facts"]["body"]["credentials"] == {} and by_task["sig-agree"]["body"]["credentials"] == {}, "no Model" accepted = {} for stamp, words in fake.log(): # the last resume of each task is the one ax accepted @@ -584,8 +611,10 @@ def test_start_is_sent_once_after_resume_with_the_models_credential(fake: Fake, assert verify_model["spec"]["secretKey"] == {"name": "deepseek", "key": "DEEPSEEK_API_KEY"}, "the manifest names the key" assert "DEEPSEEK_API_KEY" not in fake.applied("sig", "verify")["env"], "the rendered Task carries no credential" rendered = str(reconciler.run_dir("sig") / "verify" / "task.yaml") - assert rendered in files_containing(tmp_path, "OUSAST_ARTIFACT_URL"), "the scan reads the rendered files it vouches for" + assert rendered in files_containing(tmp_path, "OUSAST_INPUTS"), "the scan reads the rendered files it vouches for" assert files_containing(tmp_path, SECRET) == [], "no rendered task.yaml, state, log or applied record holds the value" + leaked = [p for p in files_containing(tmp_path, "X-Amz-Signature") if "/started/" not in p] # the fake router's own records + assert leaked == [], "nor a presigned URL in any product file (the fake store signs with that name)" def test_a_model_bound_task_without_its_secret_fails_before_ax_sees_it(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -649,6 +678,7 @@ def test_receiver_rejects_traversal_unknown_tasks_and_wrong_types(tmp_path: Path def test_delivery_for_a_task_that_is_not_running_is_rejected(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OUSAST_DELIVERY", "receiver") # the receiver path, kept for group 2 (task 2.6 removes this test) monkeypatch.setenv("OUSAST_TASK_TIMEOUT", "1") # the rejected delivery is no delivery: the task times out script = json.loads(json.dumps(SCRIPT)) script["repo-facts"]["as_task"] = "agree" # the runner claims to be a task that has not started @@ -895,10 +925,10 @@ def test_egress_policy_between_apply_and_resume_and_checked_after_delete(fake: F contexts = {line.split()[-1] for line in lines if line.startswith("egress ")} assert contexts == {PROFILE_CONTEXT}, "kubectl-ate received the profile's context, never a built one (1.2)" written = json.loads((tmp_path / "policies" / "default_eg-verify.written.json").read_text()) - tls = {"hostnames": ["api.deepseek.com", "example.invalid"], "ports": {"numbers": [443]}} - assert written["rules"] == [{"tlsPassthrough": tls}], "the Model host and the Git host, nothing else" + tls = {"hostnames": ["api.deepseek.com", "example.invalid", "localhost"], "ports": {"numbers": [443]}} + assert written["rules"] == [{"tlsPassthrough": tls}], "the Model host, the Git host and the store host, nothing else" facts = json.loads((tmp_path / "policies" / "default_eg-repo-facts.written.json").read_text()) - assert facts["rules"] == [{"tlsPassthrough": {"hostnames": ["example.invalid"], "ports": {"numbers": [443]}}}] + assert facts["rules"] == [{"tlsPassthrough": {"hostnames": ["example.invalid", "localhost"], "ports": {"numbers": [443]}}}] def test_a_refused_egress_policy_fails_the_task_before_resume(fake: Fake, tmp_path: Path, kubectl_ate: Path) -> None: @@ -992,9 +1022,8 @@ def test_state_is_mirrored_to_the_store_and_seeds_a_rerun_from_an_empty_results_ fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: assert reconciler.run(write_run(tmp_path, "mir"), ax=str(fake.ax)) == "done" - mirrored = tmp_path / "memory" / "runs" / "mir" / "state.json" # the profile's file:// store - assert mirrored.is_file() and json.loads(mirrored.read_text()) == state_of("mir"), "mirrored after the last change" - assert json.loads(mirrored.read_text())["status"] == "done" + mirrored = json.loads(fake.store.objects["runs/mir/state.json"].decode()) + assert mirrored == state_of("mir") and mirrored["status"] == "done", "mirrored after the last change" applies = len(fake.events("apply")) monkeypatch.setenv("OUSAST_RESULTS", str(tmp_path / "elsewhere")) # another machine: no local state at all assert reconciler.run(write_run(tmp_path, "mir"), ax=str(fake.ax)) == "done" @@ -1005,11 +1034,61 @@ def test_state_is_mirrored_to_the_store_and_seeds_a_rerun_from_an_empty_results_ def test_a_store_that_cannot_open_stops_the_run_before_ax_and_leaves_no_lock( fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: + from openultrasast.plane import memory from openultrasast.plane.memory import MemoryStoreError + monkeypatch.setattr(reconciler, "open_store", memory.open_store) # the real one, not the fixture's fake monkeypatch.setenv("OUSAST_MEMORY", "s3://nowhere") # the profile's override; no S3_* in the test environment for key in ("S3_ENDPOINT", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"): monkeypatch.delenv(key, raising=False) with pytest.raises(MemoryStoreError, match="S3_ENDPOINT"): reconciler.run(write_run(tmp_path, "nostore"), ax=str(fake.ax)) assert fake.events("apply") == [] and not (reconciler.run_dir("nostore") / "lock").exists() + + +# --- the store path end to end, and the receiver path kept for group 2 (plane-on-kubernetes 2.4) -------------------- + + +def test_the_store_holds_the_tar_the_declared_outputs_and_done_json(fake: Fake, tmp_path: Path) -> None: + """Completion is the tar in the store: downloaded, extracted under the run dir, declared outputs re-put as single + objects (the consumer's GET names one artifact), done.json with the summary's status and the tar's size.""" + assert reconciler.run(write_run(tmp_path, "st"), ax=str(fake.ax)) == "done" + keys = sorted(k for k in fake.store.objects if k.startswith("runs/st/")) + assert keys == [ + "runs/st/agree/done.json", + "runs/st/agree/output.tar", + "runs/st/repo-facts/done.json", + "runs/st/repo-facts/facts.json", + "runs/st/repo-facts/output.tar", + "runs/st/state.json", + "runs/st/verify/agreed.json", + "runs/st/verify/done.json", + "runs/st/verify/output.tar", + ], "declared outputs only (summary.json is extracted locally, not declared by this Run)" + done = json.loads(fake.store.objects["runs/st/repo-facts/done.json"]) + assert done["status"] == "done" and done["bytes"] == len(fake.store.objects["runs/st/repo-facts/output.tar"]) and done["files"] == 2 + assert json.loads((reconciler.run_dir("st") / "repo-facts" / "facts.json").read_text()) == {"a.py": {"functions": ["f"]}} + assert fake.store.objects["runs/st/repo-facts/facts.json"] == (reconciler.run_dir("st") / "repo-facts" / "facts.json").read_bytes() + + +def test_a_tar_the_store_path_cannot_accept_fails_the_task(fake: Fake, tmp_path: Path) -> None: + script = json.loads(json.dumps(SCRIPT)) + script["repo-facts"]["files"] = {"../escape.json": "{}", "summary.json": FACTS_SUMMARY} + fake.script(script) + assert reconciler.run(write_run(tmp_path, "esc"), ax=str(fake.ax)) == "failed" + assert "rejected member ../escape.json" in state_of("esc")["tasks"]["repo-facts"]["reason"] + assert not (reconciler.run_dir("esc").parent / "escape.json").exists() + + +def test_receiver_mode_still_runs_through_the_host_receiver(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """OUSAST_DELIVERY=receiver (group 2 only): the old path, with its http rule and OUSAST_ARTIFACT_URL, until 2.6.""" + monkeypatch.setenv("OUSAST_DELIVERY", "receiver") + assert reconciler.run(write_run(tmp_path, "rcv"), ax=str(fake.ax)) == "done" + assert fake.applied("rcv", "verify")["env"]["OUSAST_ARTIFACT_URL"].startswith("http://127.0.0.1:") + assert [s["body"].get("delivery") for s in fake.router.starts] == [None, None, None], "no presigned URLs in receiver mode" + assert not any(k.startswith("runs/rcv/") and k.endswith("output.tar") for k in fake.store.objects), "nothing PUT to the store" + assert "runs/rcv/state.json" in fake.store.objects, "the state is still mirrored" + written = json.loads((tmp_path / "policies" / "default_rcv-repo-facts.written.json").read_text()) + assert written["rules"] == [{"tlsPassthrough": {"hostnames": ["example.invalid"], "ports": {"numbers": [443]}}}], ( + "an IP receiver: no http rule" + )