plane(2.1): Delivery -- presigned store URLs, collection, state mirrored to the store

src/openultrasast/plane/delivery.py: Delivery(store, run, base) mints a presigned
PUT for runs/<run>/<task>/output.tar and a GET per declared input
(runs/<run>/<producer>/<artifact>), expiry OUSAST_TASK_TIMEOUT + 600 s;
delivered() asks the store for the exact key; collect() downloads, extracts
under the run directory with the receiver's member checks (no symlink, hard
link or escape; nothing written on a rejection), re-puts the declared outputs
as single objects and writes done.json; put_input() for inputs no task
produces; save_state()/load_state() mirror state.json. The reconciler opens the
store first (no fallback; the CLI reports a store that cannot open and submits
nothing), mirrors state.json after every change and seeds a rerun with no local
state from the store (the 2.1 bullet's state mirroring is what gives the new
module its importer). The reconciler crossed 500 lines here (517) rather than at
2.4 as 1.6 projected: the bound moves to 550 with the reason in the assertion,
as 1.6 decided. tests/plane_fake_store.py is the S3-shaped fake (objects in a
dict, presigned URLs on a local server); tests/test_plane_delivery.py covers the
body, expiry, polling, rejection, declared-only re-put, state seeding and that
no URL reaches a log. Module-audit manifest regenerated (148/5, no orphan).

Gates: pytest 2020 passed, 42 skipped, exit 0; ruff check 0; ruff format 0;
mypy 0.

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

.kiro/specs/plane-on-kubernetes/tasks.md

@@ -68,7 +68,7 @@ go-ahead** with a ceiling; steps that need a hand on a cluster that does not exi
   - _Requirements: 1.1, 1.2, 1.3, 7.1_
 
 - [ ] 2. Store delivery, receiver-less egress (design section 3)
-- [ ] 2.1 Presigned URLs and the store run directory
+- [x] 2.1 Presigned URLs and the store run directory
   - `src/openultrasast/plane/delivery.py`: `Delivery(store, run)` mints presigned PUT for
     `runs/<run>/<task>/output.tar` and GET for `runs/<run>/<producer>/<artifact>` (boto3 `generate_presigned_url`
     on `S3Client.sdk`, expiry `OUSAST_TASK_TIMEOUT + 600`); `delivered(task)` HEADs the tar; `collect(task)`

benchmarks/measurements/2026-09-08-module-audit.json

@@ -1,7 +1,7 @@
 {
- "modules": 152,
+ "modules": 153,
  "counts": {
-  "load_bearing": 147,
+  "load_bearing": 148,
   "standalone_capability": 5
  },
  "orphaned": [],
@@ -848,6 +848,7 @@
     "learn.examples",
     "learn.experiments",
     "learn.program",
+    "plane.delivery",
     "plane.doctor",
     "plane.egress",
     "plane.engine_alerts",
@@ -879,6 +880,14 @@
     "plane.tasks.verify"
    ]
   },
+  {
+   "module": "plane.delivery",
+   "classification": "load_bearing",
+   "reason": "reachable from gate, map_gate, pair_gate, cli",
+   "importers": [
+    "plane.reconciler"
+   ]
+  },
   {
    "module": "plane.doctor",
    "classification": "load_bearing",
@@ -948,8 +957,10 @@
     "learn.examples",
     "learn.experiments",
     "learn.program",
+    "plane.delivery",
     "plane.doctor",
     "plane.generate",
+    "plane.reconciler",
     "plane.tasks.loop",
     "plane.tasks.remember"
    ]

src/openultrasast/cli.py

@@ -1394,8 +1394,14 @@ def _plane(args: argparse.Namespace) -> int:
         print(f"plane: {exc}", file=sys.stderr)
         return 2
     if args.plane_command == "run":
+        from .plane.memory import MemoryStoreError
+
         _plane_memory("seed", args.run_manifest)
+        try:
             result = reconciler.run(args.run_manifest, workers=args.workers, ax=args.ax, profile=profile)
+        except MemoryStoreError as exc:  # the store is the state of record: no fallback, the Run is not submitted
+            print(f"memory store {profile.memory}: {exc}", file=sys.stderr)
+            return 1
         print(f"run finished: {result}")
         _plane_memory("ingest", args.run_manifest)
         return 0 if result == "done" else 1

src/openultrasast/plane/delivery.py

@@ -0,0 +1,173 @@
+"""Task artifacts through the memory store (plane-on-kubernetes design section 3, Req 3.1).
+
+A task's runner PUTs its output tar to a presigned URL the reconciler minted, and GETs each declared input from
+another; the URLs travel in the start request's ``delivery`` field (a presigned URL is a bearer capability, so it
+goes where the credentials go, never into an ActorTemplate's env). The reconciler polls the store for the tar,
+extracts it under the run directory with the member checks the old receiver had, re-puts the task's declared
+``outputs[]`` as single objects so a consumer's URL names exactly one artifact, and writes ``done.json``. The run's
+``state.json`` is mirrored to the store after each change and seeded from it on a rerun elsewhere.
+
+Store layout under the 30-day ``runs/`` expiry rule (``memory.RUNS_RULE_ID``)::
+
+    runs/<run>/state.json                  the reconciler's state, mirrored
+    runs/<run>/<task>/output.tar           the runner's PUT
+    runs/<run>/<task>/<artifact>           each declared output, re-put by the reconciler
+    runs/<run>/<task>/done.json            status, bytes, file count, finished
+    runs/<run>/inputs/<name>               an input no task produces (``put_input``)
+
+Nothing here logs or prints a URL.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import re
+import tarfile
+from collections.abc import Iterable, Mapping
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+
+from .memory import RUNS_PREFIX, MemoryStore
+
+__all__ = ["DONE", "GRACE_SECONDS", "OUTPUT_TAR", "STATE", "Delivery", "DeliveryError", "expiry", "input_var"]
+
+OUTPUT_TAR = "output.tar"
+DONE = "done.json"
+STATE = "state.json"
+INPUTS_DIR = "inputs"
+GRACE_SECONDS = 600  # a URL outlives the task timeout by this much; a re-start after a re-resume re-mints anyway
+_NAME = re.compile(r"[^A-Z0-9]+")
+
+
+class DeliveryError(RuntimeError):
+    """A delivered tar that cannot be accepted, or an object that is not there; the message never carries a URL."""
+
+
+def expiry(environ: Mapping[str, str] | None = None) -> timedelta:
+    """``OUSAST_TASK_TIMEOUT`` (7200 s) plus :data:`GRACE_SECONDS`."""
+    env = os.environ if environ is None else environ
+    return timedelta(seconds=float(env.get("OUSAST_TASK_TIMEOUT") or 7200) + GRACE_SECONDS)
+
+
+def input_var(name: str) -> str:
+    """The ``OUSAST_INPUT_<NAME>`` suffix of a Run input name, as the reconciler renders it."""
+    return _NAME.sub("_", name.upper())
+
+
+def _now() -> str:
+    return datetime.now(UTC).isoformat(timespec="seconds")
+
+
+class Delivery:
+    """Presigned URLs, polling, collection and state mirroring for one Run on one store."""
+
+    def __init__(self, store: MemoryStore, run: str, base: Path, *, expires: timedelta | None = None) -> None:
+        self.store, self.run, self.base = store, run, base
+        self.expires = expires or expiry()
+
+    def key(self, *parts: str) -> str:
+        return f"{RUNS_PREFIX}{self.run}/" + "/".join(parts)
+
+    @property
+    def host(self) -> str | None:
+        """The hostname the runner will dial (for the task's egress policy); None for a store without one."""
+        return urlsplit(self.store.presign_get(self.key(STATE), timedelta(seconds=60))).hostname
+
+    # --- the start body -------------------------------------------------------------------------------------------
+
+    def put_url(self, task: str) -> str:
+        return self.store.presign_put(self.key(task, OUTPUT_TAR), self.expires)
+
+    def input_urls(self, inputs: Mapping[str, str]) -> dict[str, str]:
+        """``{OUSAST_INPUT_<NAME> suffix: presigned GET of runs/<run>/<producer>/<artifact>}``."""
+        out: dict[str, str] = {}
+        for name, ref in inputs.items():
+            producer, _, artifact = ref.partition("/")
+            if not producer or not artifact:
+                raise DeliveryError(f"input {name}: {ref!r} is not <producer>/<artifact>")
+            out[input_var(name)] = self.store.presign_get(self.key(producer, artifact), self.expires)
+        return out
+
+    def body(self, task: str, inputs: Mapping[str, str]) -> dict[str, Any]:
+        """The start request's ``delivery`` field for ``task``."""
+        return {"put": self.put_url(task), "inputs": self.input_urls(inputs)}
+
+    # --- completion ----------------------------------------------------------------------------------------------
+
+    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)
+        return key in self.store._keys(key)
+
+    def collect(self, task: str, outputs: Iterable[str]) -> tuple[dict[str, Any] | None, int]:
+        """Download and extract the tar under ``<base>/<task>/``, re-put the declared ``outputs``, write
+        ``done.json``; returns (the extracted ``summary.json`` or None, the tar's byte size)."""
+        got = self.store._get(self.key(task, OUTPUT_TAR))
+        if got is None:
+            raise DeliveryError(f"task {task}: output.tar is not in the store")
+        data = got[0]
+        dest = (self.base / task).resolve()
+        try:
+            with tarfile.open(fileobj=io.BytesIO(data), mode="r:") as tar:
+                members = tar.getmembers()
+                for m in members:
+                    if m.issym() or m.islnk() or not (dest / m.name).resolve().is_relative_to(dest):
+                        raise DeliveryError(f"task {task}: rejected member {m.name}")
+                dest.mkdir(parents=True, exist_ok=True)
+                tar.extractall(dest, members)  # every member checked above
+        except tarfile.TarError as exc:
+            raise DeliveryError(f"task {task}: bad tar: {exc}") from None
+        published = self.publish(task, outputs)
+        summary = _read_json(dest / "summary.json")
+        done = {
+            "task": task,
+            "status": summary.get("status") if isinstance(summary, dict) else None,
+            "bytes": len(data),
+            "files": sum(1 for m in members if m.isfile()),
+            "published": published,
+            "finished": _now(),
+        }
+        self.store._put(self.key(task, DONE), json.dumps(done, sort_keys=True).encode("utf-8"))
+        return (summary if isinstance(summary, dict) else None), len(data)
+
+    def publish(self, task: str, outputs: Iterable[str]) -> list[str]:
+        """Re-put the task's declared outputs that exist under ``<base>/<task>/`` as single objects; the names put."""
+        put: list[str] = []
+        root = (self.base / task).resolve()
+        for artifact in outputs:
+            path = (root / artifact).resolve()
+            if path.is_relative_to(root) and path.is_file():
+                self.store._put(self.key(task, artifact), path.read_bytes())
+                put.append(artifact)
+        return put
+
+    def put_input(self, name: str, path: Path) -> str:
+        """Store a file no task produces as ``runs/<run>/inputs/<name>``; returns its presigned GET URL."""
+        self.store._put(self.key(INPUTS_DIR, name), Path(path).read_bytes())
+        return self.store.presign_get(self.key(INPUTS_DIR, name), self.expires)
+
+    # --- state ---------------------------------------------------------------------------------------------------
+
+    def save_state(self, data: Mapping[str, Any]) -> None:
+        self.store._put(self.key(STATE), (json.dumps(data, indent=2) + "\n").encode("utf-8"))
+
+    def load_state(self) -> dict[str, Any] | None:
+        got = self.store._get(self.key(STATE))
+        if got is None:
+            return None
+        try:
+            loaded = json.loads(got[0].decode("utf-8"))
+        except ValueError:
+            return None
+        return loaded if isinstance(loaded, dict) else None
+
+
+def _read_json(path: Path) -> Any:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None

src/openultrasast/plane/reconciler.py

@@ -18,7 +18,10 @@ task as ``OUSAST_MODEL`` plus ``OUSAST_MODEL_PARAMS`` (Req 7.1), each Workspace'
 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).
+State: ``~/ousast-results/plane/<run>/`` (``OUSAST_RESULTS``), polled every ``OUSAST_POLL_SECONDS`` (3); ``state.json``
+is mirrored to ``runs/<run>/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.
 """
 
 from __future__ import annotations
@@ -42,9 +45,11 @@ from urllib.parse import urlsplit
 
 import yaml
 
+from .delivery import Delivery
 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
+from .memory import open_store
 from .profile import PlaneProfile, load_profile
 from .router import Router, credentials, open_router
 
@@ -226,12 +231,25 @@ class _State:
     path: Path
     data: dict[str, Any]
     lock: threading.Lock = field(default_factory=threading.Lock)
+    mirror: Callable[[Mapping[str, Any]], None] | None = None  # the store copy; a failure is reported, never fatal
 
     def set(self, task: str, **fields: object) -> None:
         with self.lock:
             self.data["tasks"].setdefault(task, {"status": "pending", "started": None, "finished": None, "usd": None, "model": None})
             self.data["tasks"][task].update(fields)
+            self._flush()
+
+    def flush(self) -> None:
+        with self.lock:
+            self._flush()
+
+    def _flush(self) -> None:
         _write_json(self.path, self.data)
+        if self.mirror is not None:
+            try:
+                self.mirror(self.data)
+            except Exception as exc:  # noqa: BLE001 - the local file is written; the mirror is reported, the run goes on
+                print(f"state mirror to the store failed: {exc}", file=sys.stderr)
 
     def status(self, task: str) -> str:
         return str(self.data["tasks"].get(task, {}).get("status", "pending"))
@@ -357,8 +375,18 @@ def run(
     run_spec, manifests = load_run(Path(run_manifest))
     base = run_dir(run_spec.metadata.name, results_root)
     base.mkdir(parents=True, exist_ok=True)
+    store = open_store(profile.memory)  # first, before the lock and before ax sees anything: no fallback
+    delivery = Delivery(store, run_spec.metadata.name, base)
+    if not (base / "state.json").exists():
+        seeded = delivery.load_state()  # a rerun from another machine or an empty results root
+        if seeded is not None:
+            _write_json(base / "state.json", seeded)
     lock = _acquire_lock(base)
-    state = _State(base / "state.json", _read_json(base / "state.json") or {"run": run_spec.metadata.name, "started": _now(), "tasks": {}})
+    state = _State(
+        base / "state.json",
+        _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()
@@ -409,7 +437,7 @@ def run(
     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"
     state.data.update(status=result, finished=_now())
-    _write_json(state.path, state.data)
+    state.flush()
     return result
 
 

tests/plane_fake_store.py

@@ -0,0 +1,109 @@
+"""An S3-shaped memory store for the delivery tests: objects in a dict, presigned URLs served by a local HTTP server
+that accepts PUT, GET and HEAD by key (``/<bucket>/<key>?X-Amz-Expires=N&X-Amz-Signature=fake``), the way a
+presigned S3 URL looks to the runner. Nothing here is imported by the product."""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Mapping
+from datetime import timedelta
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import unquote, urlsplit
+
+from openultrasast.plane.memory import MemoryStore
+
+
+class HttpObjectStore(MemoryStore):
+    """``presign_*`` hand out URLs on a local server; ``requests`` records every HTTP call (method, key)."""
+
+    def __init__(self, bucket: str = "fake-bucket", host: str = "localhost") -> None:
+        self.objects: dict[str, bytes] = {}
+        self.bucket, self.host = bucket, host
+        self.requests: list[tuple[str, str]] = []
+        self.presigned: list[tuple[str, str, int]] = []
+        self.lock = threading.Lock()
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+        self.server.store = self  # type: ignore[attr-defined]
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+
+    def close(self) -> None:
+        self.server.shutdown()
+        self.server.server_close()
+
+    def describe(self) -> str:
+        return f"fake://{self.bucket}"
+
+    def _get(self, key: str) -> tuple[bytes, str | None] | None:
+        with self.lock:
+            data = self.objects.get(key)
+        return None if data is None else (data, "v1")
+
+    def _put(self, key: str, data: bytes, labels: Mapping[str, str] | None = None) -> None:
+        with self.lock:
+            self.objects[key] = bytes(data)
+
+    def _delete(self, key: str) -> None:
+        with self.lock:
+            self.objects.pop(key, None)
+
+    def _keys(self, prefix: str) -> list[str]:
+        with self.lock:
+            return sorted(k for k in self.objects if k.startswith(prefix))
+
+    def _url(self, method: str, key: str, expires: timedelta) -> str:
+        self.presigned.append((method, key, int(expires.total_seconds())))
+        port = self.server.server_address[1]
+        return f"http://{self.host}:{port}/{self.bucket}/{key}?X-Amz-Expires={int(expires.total_seconds())}&X-Amz-Signature=fake"
+
+    def presign_put(self, key: str, expires: timedelta = timedelta(hours=1)) -> str:
+        return self._url("PUT", key, expires)
+
+    def presign_get(self, key: str, expires: timedelta = timedelta(hours=1)) -> str:
+        return self._url("GET", key, expires)
+
+
+class _Handler(BaseHTTPRequestHandler):
+    def _key(self) -> str | None:
+        store: HttpObjectStore = self.server.store  # type: ignore[attr-defined]
+        path = unquote(urlsplit(self.path).path)
+        prefix = f"/{store.bucket}/"
+        return path[len(prefix) :] if path.startswith(prefix) else None
+
+    def do_PUT(self) -> None:  # noqa: N802
+        store: HttpObjectStore = self.server.store  # type: ignore[attr-defined]
+        key = self._key()
+        body = self.rfile.read(int(self.headers.get("Content-Length") or 0))
+        store.requests.append(("PUT", key or self.path))
+        if key is None:
+            return self._answer(404)
+        store._put(key, body)
+        self._answer(200)
+
+    def do_GET(self) -> None:  # noqa: N802
+        store: HttpObjectStore = self.server.store  # type: ignore[attr-defined]
+        key = self._key()
+        store.requests.append(("GET", key or self.path))
+        got = store._get(key) if key else None
+        if got is None:
+            return self._answer(404)
+        self.send_response(200)
+        self.send_header("Content-Length", str(len(got[0])))
+        self.end_headers()
+        self.wfile.write(got[0])
+
+    def do_HEAD(self) -> None:  # noqa: N802
+        store: HttpObjectStore = self.server.store  # type: ignore[attr-defined]
+        key = self._key()
+        store.requests.append(("HEAD", key or self.path))
+        got = store._get(key) if key else None
+        self.send_response(404 if got is None else 200)
+        self.send_header("Content-Length", "0" if got is None else str(len(got[0])))
+        self.end_headers()
+
+    def _answer(self, code: int) -> None:
+        self.send_response(code)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def log_message(self, format: str, *args: object) -> None:  # noqa: A002
+        pass

tests/test_plane_delivery.py

@@ -0,0 +1,145 @@
+"""plane-on-kubernetes task 2.1 (Req 3.1, 2.1): ``Delivery`` against an S3-shaped fake -- presigned URLs in the start
+body, expiry, polling, extraction with the receiver's member checks, declared-only re-put, state seeding, and no URL
+in any log line."""
+
+from __future__ import annotations
+
+import io
+import json
+import logging
+import tarfile
+import urllib.request
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from plane_fake_store import HttpObjectStore
+
+from openultrasast.plane.delivery import GRACE_SECONDS, Delivery, DeliveryError, expiry, input_var
+
+
+@pytest.fixture
+def store() -> Iterator[HttpObjectStore]:
+    fake = HttpObjectStore()
+    try:
+        yield fake
+    finally:
+        fake.close()
+
+
+def tar_of(files: dict[str, bytes], *, symlink: str | None = None) -> bytes:
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w") as tar:
+        for name, data in files.items():
+            info = tarfile.TarInfo(name)
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+        if symlink:
+            info = tarfile.TarInfo(symlink)
+            info.type = tarfile.SYMTYPE
+            info.linkname = "/etc/passwd"
+            tar.addfile(info)
+    return buffer.getvalue()
+
+
+def put_via_url(url: str, data: bytes) -> int:
+    request = urllib.request.Request(url, data=data, method="PUT", headers={"Content-Type": "application/x-tar"})
+    with urllib.request.urlopen(request, timeout=10) as response:  # noqa: S310 - the fake's local URL
+        return int(response.status)
+
+
+def test_the_start_body_names_exactly_the_tar_and_the_declared_inputs(store: HttpObjectStore, tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("OUSAST_TASK_TIMEOUT", "100")
+    delivery = Delivery(store, "run-1", tmp_path)
+    body = delivery.body("verify", {"facts": "repo-facts/facts.json", "pass a": "va/units.jsonl"})
+    assert set(body) == {"put", "inputs"}
+    assert body["put"].startswith(f"http://localhost:{store.server.server_address[1]}/fake-bucket/runs/run-1/verify/output.tar?")
+    assert set(body["inputs"]) == {"FACTS", "PASS_A"} and input_var("pass a") == "PASS_A"
+    assert "/runs/run-1/repo-facts/facts.json?" in body["inputs"]["FACTS"] and "/runs/run-1/va/units.jsonl?" in body["inputs"]["PASS_A"]
+    assert {m for m, _, _ in store.presigned} == {"PUT", "GET"}
+    assert {e for _, _, e in store.presigned} == {100 + GRACE_SECONDS}, "expiry = OUSAST_TASK_TIMEOUT + 600"
+    assert expiry({"OUSAST_TASK_TIMEOUT": "7200"}).total_seconds() == 7800 and expiry({}).total_seconds() == 7800
+    assert delivery.host == "localhost", "the hostname the egress policy must allow"
+    with pytest.raises(DeliveryError, match="not <producer>/<artifact>"):
+        delivery.input_urls({"x": "no-slash"})
+
+
+def test_delivered_polls_the_store_and_collect_extracts_republishes_and_marks_done(
+    store: HttpObjectStore, tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+    delivery = Delivery(store, "r", tmp_path / "r")
+    assert delivery.delivered("facts") is False
+    summary = {"status": "done", "units_done": 1, "units_total": 1}
+    tar = tar_of({"facts.json": b'{"a": 1}', "summary.json": json.dumps(summary).encode(), "scratch/notes.txt": b"not declared"})
+    assert put_via_url(delivery.put_url("facts"), tar) == 200, "the runner's PUT to the presigned URL"
+    assert delivery.delivered("facts") is True
+    got, size = delivery.collect("facts", ["facts.json", "summary.json", "missing.json"])
+    assert got == summary and size == len(tar)
+    assert (tmp_path / "r" / "facts" / "facts.json").read_bytes() == b'{"a": 1}'
+    assert (tmp_path / "r" / "facts" / "scratch" / "notes.txt").exists(), "extracted locally"
+    assert store.objects["runs/r/facts/facts.json"] == b'{"a": 1}' and "runs/r/facts/summary.json" in store.objects
+    assert "runs/r/facts/scratch/notes.txt" not in store.objects and "runs/r/facts/missing.json" not in store.objects, "declared only"
+    done = json.loads(store.objects["runs/r/facts/done.json"])
+    assert (
+        done["status"] == "done"
+        and done["bytes"] == len(tar)
+        and done["files"] == 3
+        and done["published"] == ["facts.json", "summary.json"]
+    )
+    # a consumer's presigned GET names one artifact and fetches it
+    urls = delivery.input_urls({"facts": "facts/facts.json"})
+    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"
+
+
+@pytest.mark.parametrize("bad", ["../escape.json", "/abs.json"])
+def test_a_tar_member_that_escapes_is_rejected_and_nothing_is_written(store: HttpObjectStore, tmp_path: Path, bad: str) -> None:
+    delivery = Delivery(store, "r", tmp_path / "r")
+    store._put(delivery.key("t", "output.tar"), tar_of({bad: b"{}", "summary.json": b"{}"}))
+    with pytest.raises(DeliveryError, match="rejected member"):
+        delivery.collect("t", ["summary.json"])
+    assert not (tmp_path / "r" / "t").exists() and not (tmp_path / "escape.json").exists()
+    assert "runs/r/t/done.json" not in store.objects and "runs/r/t/summary.json" not in store.objects
+
+
+def test_a_symlink_member_and_a_non_tar_are_rejected(store: HttpObjectStore, tmp_path: Path) -> None:
+    delivery = Delivery(store, "r", tmp_path / "r")
+    store._put(delivery.key("t", "output.tar"), tar_of({"summary.json": b"{}"}, symlink="link"))
+    with pytest.raises(DeliveryError, match="rejected member link"):
+        delivery.collect("t", ["summary.json"])
+    store._put(delivery.key("t", "output.tar"), b"not a tar at all")
+    with pytest.raises(DeliveryError, match="bad tar"):
+        delivery.collect("t", ["summary.json"])
+    with pytest.raises(DeliveryError, match="not in the store"):
+        delivery.collect("never", [])
+
+
+def test_state_is_mirrored_and_seeds_a_rerun_elsewhere(store: HttpObjectStore, tmp_path: Path) -> None:
+    first = Delivery(store, "r", tmp_path / "machine-a")
+    assert first.load_state() is None
+    first.save_state({"run": "r", "tasks": {"facts": {"status": "done"}}})
+    second = Delivery(store, "r", tmp_path / "machine-b")  # another results root, the same store
+    assert second.load_state() == {"run": "r", "tasks": {"facts": {"status": "done"}}}
+    store._put(second.key("state.json"), b"not json")
+    assert second.load_state() is None
+
+
+def test_put_input_stores_a_file_no_task_produces(store: HttpObjectStore, tmp_path: Path) -> None:
+    delivery = Delivery(store, "r", tmp_path / "r")
+    source = tmp_path / "candidates.json"
+    source.write_bytes(b"[1, 2, 3]")
+    url = delivery.put_input("candidates.json", source)
+    assert store.objects["runs/r/inputs/candidates.json"] == b"[1, 2, 3]"
+    with urllib.request.urlopen(url, timeout=10) as response:  # noqa: S310
+        assert response.read() == b"[1, 2, 3]"
+
+
+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)
+    (tmp_path / "r" / "seeded" / "facts.json").write_bytes(b"{}")
+    (tmp_path / "r" / "secret.txt").write_bytes(b"outside")
+    assert delivery.publish("seeded", ["facts.json", "../secret.txt", "absent.json"]) == ["facts.json"]
+    assert set(store.objects) == {"runs/r/seeded/facts.json"}

tests/test_plane_reconciler.py

@@ -832,9 +832,13 @@ def test_mark_done_sets_the_state_under_the_run_lock(tmp_path: Path) -> None:
     assert not (base / "lock").exists(), "the lock is released"
 
 
-def test_reconciler_is_under_500_lines_and_holds_no_pipeline_logic() -> None:
+def test_reconciler_is_under_550_lines_and_holds_no_pipeline_logic() -> None:
     text = RECONCILER.read_text(encoding="utf-8")
-    assert len(text.splitlines()) < 500, "Req 3.4: the reconciler stays under 500 lines"
+    assert len(text.splitlines()) < 550, (
+        "Req 3.4: the reconciler stays under 500 lines; plane-on-kubernetes 1.6/2.1 moved the bound to 550 while the "
+        "store-delivery wiring (Delivery, state mirror and seed, the start-body delivery) sits beside the receiver path "
+        "that task 2.6 removes -- 2.6 records the count and returns the bound to 500 if it allows"
+    )
     for word in ("prompt(", "openai", "ChatClient", "score(", "system_prompt"):
         assert word not in text, f"Req 3.4: no pipeline logic in the reconciler ({word})"
     assert "Popen" not in text and "openultrasast.plane.tasks" not in text, "ax is the only executor"
@@ -979,3 +983,33 @@ def test_ax_task_names_leave_room_for_the_template_suffix() -> None:
     assert other != long, "no collision between tasks sharing a long prefix"
     assert reconciler._ax_name("run", "facts", limit=reconciler.TASK_NAME_LIMIT) == "run-facts"
     assert re.fullmatch(r"[a-z0-9]([a-z0-9-]*[a-z0-9])?", long)
+
+
+# --- the store is the state of record (plane-on-kubernetes 2.1, design section 2) -----------------------------------
+
+
+def test_state_is_mirrored_to_the_store_and_seeds_a_rerun_from_an_empty_results_root(
+    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"
+    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"
+    assert statuses("mir") == {"repo-facts": "done", "verify": "done", "agree": "done"}, "seeded from the store"
+    assert len(fake.events("apply")) == applies, "nothing was submitted again"
+
+
+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.memory import MemoryStoreError
+
+    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()