Merge branch 'worktree-agent-ad03a357c98ed488d'

# Conflicts:
#	benchmarks/measurements/2026-09-08-module-audit.json
#	src/openultrasast/model/audit.py

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

@@ -1,7 +1,7 @@
 {
-  "modules": 125,
+  "modules": 127,
   "counts": {
-    "load_bearing": 116,
+    "load_bearing": 118,
     "standalone_capability": 9
   },
   "orphaned": [],
@@ -658,7 +658,10 @@
       "classification": "load_bearing",
       "reason": "ai-service-plane Req 1-4: the service plane package; runner and reconciler integration in tasks 3-4",
       "importers": [
-        "plane.tasks.verify"
+        "cli",
+        "plane.reconciler",
+        "plane.tasks.verify",
+        "plane.workspaces"
       ]
     },
     {
@@ -673,7 +676,17 @@
       "module": "plane.manifests",
       "classification": "load_bearing",
       "reason": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs",
-      "importers": []
+      "importers": [
+        "plane.reconciler"
+      ]
+    },
+    {
+      "module": "plane.reconciler",
+      "classification": "load_bearing",
+      "reason": "ai-service-plane Req 3.1-3.5, 7.3: ax-backed Run execution, status attribution and doctor; `ousast plane`",
+      "importers": [
+        "plane.workspaces"
+      ]
     },
     {
       "module": "plane.tasks",
@@ -703,6 +716,14 @@
         "plane.tasks.agree"
       ]
     },
+    {
+      "module": "plane.workspaces",
+      "classification": "load_bearing",
+      "reason": "ai-service-plane task 4: Workspace manifests per case pin of a population; `ousast plane workspaces`",
+      "importers": [
+        "cli"
+      ]
+    },
     {
       "module": "policy",
       "classification": "load_bearing",

pyproject.toml

@@ -41,6 +41,7 @@ pythonpath = ["src"]
 markers = [
     "docker: requires a docker daemon",
     "semantic: requires openultrasast[semantic] tree-sitter extra",
+    "ax: requires the live ax cluster of ops/ax/up.sh (skipped unless `ousast plane doctor` passes)",
 ]
 
 [tool.ruff]

src/openultrasast/cli.py

@@ -238,6 +238,20 @@ def main(argv: list[str] | None = None) -> int:
     candidates.add_argument("--generator", choices=("ir", "ruleset"), default="ir", help="which pass enumerates the candidate sites")
     candidates.add_argument("--json", action="store_true")
 
+    plane = subparsers.add_parser("plane", help="maintainer: the ax-backed service plane on this host's kind cluster")
+    plane_sub = plane.add_subparsers(dest="plane_command", required=True)
+    plane_run = plane_sub.add_parser("run", help="execute a Run manifest on ax; a rerun skips done tasks")
+    plane_run.add_argument("run_manifest", type=Path)
+    plane_run.add_argument("--workers", type=int, default=1, help="tasks in flight at once (serialize labels never overlap)")
+    plane_run.add_argument("--ax", default="ax", help="path of the ax CLI")
+    plane_status = plane_sub.add_parser("status", help="per-task status and the token attribution table of a run")
+    plane_status.add_argument("run")
+    plane_status.add_argument("--units", action="store_true", help="per-unit rows from each task's units.jsonl")
+    plane_sub.add_parser("doctor", help="check kind, Agent Substrate, the ax controller and the runner image")
+    plane_ws = plane_sub.add_parser("workspaces", help="one Workspace manifest per case pin of a population file")
+    plane_ws.add_argument("population", type=Path)
+    plane_ws.add_argument("--out", type=Path, default=Path("plane/workspaces"))
+
     args = parser.parse_args(argv)
     # Diagnosis needs the stage costs, and nothing configures logging, so the default root level of
     # WARNING silently dropped every informational line. A run that reports nothing is
@@ -350,6 +364,8 @@ def main(argv: list[str] | None = None) -> int:
         return _repos(args.dir, args.repo, fetch=args.fetch, json_out=args.json)
     if args.command == "model":
         return _model_candidates(args)
+    if args.command == "plane":
+        return _plane(args)
     if args.command == "mcp":
         from .mcp import serve  # lazy: keeps the import cycle (mcp -> cli) one-directional
 
@@ -1272,6 +1288,28 @@ def _load_scan_findings(run_dir: Path) -> list[StaticFinding]:
     return load_findings(run_dir / "findings.json")
 
 
+def _plane(args: argparse.Namespace) -> int:
+    from .plane import reconciler  # lazy: the plane is a maintainer surface, not the scan path
+
+    if args.plane_command == "run":
+        result = reconciler.run(args.run_manifest, workers=args.workers, ax=args.ax)
+        print(f"run finished: {result}")
+        return 0 if result == "done" else 1
+    if args.plane_command == "status":
+        print(reconciler.status(args.run, units=args.units))
+        return 0
+    if args.plane_command == "doctor":
+        checks = reconciler.doctor()
+        for name, ok, text in checks:
+            print(f"[{'ok' if ok else 'FAIL'}] {name}: {text}")
+        return 0 if all(ok for _, ok, _ in checks) else 1
+    from .plane.workspaces import workspaces
+
+    written = workspaces(args.population, args.out)
+    print(f"{len(written)} Workspace manifests written to {args.out}")
+    return 0
+
+
 def _model_candidates(args: argparse.Namespace) -> int:
     """What the candidate enumerator can reach, per family and per slice. No model is called anywhere here.
 

src/openultrasast/model/audit.py

@@ -62,6 +62,8 @@ SPEC_OWNED = {
     "plane.tasks": "ai-service-plane Req 2.1: the tasks package, each a main() under the OUSAST_* env contract; runner wiring in task 3",
     "plane.tasks.verify": "ai-service-plane Req 5.3/2.1-2.3/7.2: the batched hunt with known callers, metered and resumable per file hunt",
     "plane.tasks.agree": "ai-service-plane Req 5.3/2.1: agreement across two verify passes with per-candidate cost, turns and site match",
+    "plane.reconciler": "ai-service-plane Req 3.1-3.5, 7.3: ax-backed Run execution, status attribution and doctor; `ousast plane`",
+    "plane.workspaces": "ai-service-plane task 4: Workspace manifests per case pin of a population; `ousast plane workspaces`",
 }
 
 STANDALONE = {

src/openultrasast/plane/reconciler.py

@@ -0,0 +1,492 @@
+"""ax-backed reconciler of a ``Run``: submission, status, resume, artifact receipt (ai-service-plane Req 3, 7).
+
+Every task of a Run is executed by ax on this host's kind cluster; nothing here runs a task as a local subprocess
+and nothing here knows a prompt, a score or a model call. For each ready task the reconciler renders one YAML
+file (the Task with its env extended, its Workspaces, one generated Workspace carrying the consumed artifacts as
+``files`` entries, and the bound Model), runs ``ax apply -f``, polls ``ax get task`` until a terminal phase and
+``ax delete task`` afterwards. The outcome is read from the delivered ``summary.json``, never from the phase alone.
+
+Model binding: a Task names its Model in ``metadata.annotations["openultrasast.io/model"]`` (ax's ObjectMeta has no
+annotations, so the rendered Task carries only ``name``/``atespace``); the task receives ``OUSAST_MODEL`` and the
+Model's ``spec.parameters`` as ``OUSAST_MODEL_PARAMS`` JSON (Req 7.1). Inputs: ``OUSAST_INPUT_<NAME>`` points into a
+generated Workspace bound at ``/workspace/.ousast-in/<task>`` (ax mounts every workspace under ``/workspace``).
+Artifacts: the runner POSTs an uncompressed tar of its output directory to ``OUSAST_ARTIFACT_URL`` with
+``X-Ousast-Run`` and ``X-Ousast-Task`` headers; the receiver listens on ``0.0.0.0`` at ``OUSAST_ARTIFACT_PORT``
+(default: a free port) and advertises ``OUSAST_ARTIFACT_HOST`` (default ``172.17.0.1``, the docker bridge address
+of this host, which kind nodes reach; ``127.0.0.1`` for a fake ax on the host). Run state lives in
+``~/ousast-results/plane/<run>/`` (``OUSAST_RESULTS`` overrides the root): ``state.json``, a PID ``lock``, one
+directory per task, ``attribution.json`` written by ``status``; ``OUSAST_POLL_SECONDS`` (default 3) paces polling.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import queue
+import re
+import subprocess
+import tarfile
+import threading
+import time
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from .manifests import AX_API_VERSION, Manifests, Model, Run, RunTask, Task, Workspace, load_manifests
+
+MODEL_ANNOTATION = "openultrasast.io/model"
+OUTPUT_ROOT = "/workspace/.ousast-out"
+INPUTS_ROOT = "/workspace/.ousast-in"
+DEFAULT_ARTIFACT_HOST = "172.17.0.1"
+FAILURE_PHASES = ("failed", "error")
+TERMINAL_PHASES = ("succeeded", "completed", *FAILURE_PHASES)
+UP = "ops/ax/up.sh brings it up"
+_ROW = "{:<28} {:<20} {:>7} {:>10} {:>10} {:>9} {:>10}"
+_COLUMNS = ("calls", "prompt", "cache_hit", "output", "usd")
+_USAGE_KEYS = {"prompt": "prompt_tokens", "cache_hit": "prompt_cache_hit_tokens", "output": "completion_tokens"}
+
+
+def results_root() -> Path:
+    return Path(os.environ.get("OUSAST_RESULTS") or Path.home() / "ousast-results") / "plane"
+
+
+def run_dir(run_name: str, root: Path | None = None) -> Path:
+    return (root or results_root()) / run_name
+
+
+def _now() -> str:
+    return datetime.now(UTC).isoformat(timespec="seconds")
+
+
+def _write_json(path: Path, payload: object) -> None:
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+    tmp.replace(path)
+
+
+def _read_json(path: Path) -> Any:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
+def _ax_name(*parts: str) -> str:
+    return re.sub(r"[^a-z0-9-]+", "-", "-".join(parts).lower()).strip("-")[:63]
+
+
+def _is(phase: str, words: tuple[str, ...]) -> bool:
+    return any(word in phase.lower() for word in words)
+
+
+def load_run(run_manifest: Path) -> tuple[Run, Manifests]:
+    """Load the Run file; a file under ``.../runs/`` also loads the sibling ``tasks``, ``models`` and ``workspaces``."""
+    paths = [run_manifest]
+    if run_manifest.parent.name == "runs":
+        for kind in ("tasks", "models", "workspaces"):
+            paths.extend(sorted((run_manifest.parent.parent / kind).glob("*.y*ml")))
+    manifests = load_manifests(paths)
+    if len(manifests.runs) != 1:
+        raise ValueError(f"{run_manifest}: expected exactly one Run, found {len(manifests.runs)}")
+    return next(iter(manifests.runs.values())), manifests
+
+
+def _doc(kind: str, name: str, spec: dict[str, Any], namespace: str | None = None) -> dict[str, Any]:
+    meta = {"name": name, **({"atespace": namespace} if namespace else {})}
+    return {"apiVersion": AX_API_VERSION, "kind": kind, "metadata": meta, "spec": spec}
+
+
+def _fields(value: object) -> dict[str, Any]:
+    return {k: v for k, v in vars(value).items() if v is not None}
+
+
+def _workspace_doc(ws: Workspace) -> dict[str, Any]:
+    spec: dict[str, Any] = {}
+    if ws.git:
+        spec["git"] = [_fields(g) for g in ws.git]
+    if ws.files:
+        spec["files"] = [_fields(f) for f in ws.files]
+    if ws.mcp:
+        spec["mcp"] = {"registries": list(ws.mcp.registries), "servers": list(ws.mcp.servers)}
+    if ws.skills:
+        spec["skills"] = {"registries": list(ws.skills.registries), **({"path": ws.skills.path} if ws.skills.path else {})}
+    return _doc("Workspace", ws.metadata.name, spec, ws.metadata.namespace)
+
+
+def _model_doc(model: Model) -> dict[str, Any]:
+    spec: dict[str, Any] = {"provider": model.provider, "model": model.model}
+    if model.secret_key:
+        spec["secretKey"] = _fields(model.secret_key)
+    if model.parameters:
+        spec["parameters"] = dict(model.parameters)
+    return _doc("Model", model.metadata.name, spec, model.metadata.namespace)
+
+
+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) -> list[dict[str, Any]]:
+    """The documents ``ax apply`` receives for one Run task: Task, its Model, its Workspaces, the inputs Workspace."""
+    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}
+    env.update(OUSAST_RUN=run.metadata.name, OUSAST_TASK=entry.name, OUSAST_OUTPUT_DIR=f"{OUTPUT_ROOT}/{entry.name}")
+    env["OUSAST_ARTIFACT_URL"] = artifact_url
+    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}"
+    docs: list[dict[str, Any]] = []
+    model_name = task.metadata.annotations.get(MODEL_ANNOTATION)
+    if model_name:
+        model = manifests.models[model_name]
+        env.update(OUSAST_MODEL=model_name, OUSAST_MODEL_PARAMS=json.dumps(dict(model.parameters), sort_keys=True))
+        docs.append(_model_doc(model))
+    bindings = [_fields(b) for b in task.workspaces]
+    docs.extend(_workspace_doc(manifests.workspaces[b.name]) for b in task.workspaces)
+    if entry.inputs:
+        docs.append(_doc("Workspace", f"{ax_name}-inputs", {"files": _input_files(entry, base)}, task.metadata.namespace))
+        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
+    if task.resources:
+        spec["resources"] = {k: _fields(v) for k, v in _fields(task.resources).items()}
+    if bindings:
+        spec["workspaces"] = bindings
+    if task.debug:
+        spec["debug"] = True
+    return [_doc("Task", ax_name, spec, task.metadata.namespace), *docs]
+
+
+class Ax:
+    """The three ``ax`` verbs the reconciler uses; the executable path is injectable so tests use a fake."""
+
+    def __init__(self, executable: str = "ax") -> None:
+        self.executable = executable
+
+    def _run(self, *args: str) -> subprocess.CompletedProcess[str]:
+        return subprocess.run([self.executable, *args], capture_output=True, text=True, check=False)
+
+    def apply(self, path: Path) -> None:
+        proc = self._run("apply", "-f", str(path))
+        if proc.returncode != 0:
+            raise RuntimeError(f"ax apply failed ({proc.returncode}): {proc.stderr.strip() or proc.stdout.strip()}")
+
+    def phase(self, name: str) -> str:
+        """``status.phase`` from ``ax get task <name>`` (YAML; Pending when unset; ``Unknown: ...`` when ax fails)."""
+        proc = self._run("get", "task", name)
+        if proc.returncode != 0:
+            return "Unknown: " + (proc.stderr.strip() or proc.stdout.strip())
+        try:
+            doc = yaml.safe_load(proc.stdout)
+        except yaml.YAMLError:
+            return "Unknown: unreadable ax output"
+        status = doc.get("status") if isinstance(doc, dict) else None
+        return str(status.get("phase") or "Pending") if isinstance(status, dict) else "Pending"
+
+    def delete(self, name: str) -> None:
+        self._run("delete", "task", name)
+
+
+class Receiver(ThreadingHTTPServer):
+    """Accepts one tar per running task and extracts it under ``<run dir>/<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.lock = threading.Lock()
+
+    def url(self) -> str:
+        return f"http://{os.environ.get('OUSAST_ARTIFACT_HOST') or DEFAULT_ARTIFACT_HOST}:{self.server_address[1]}/"
+
+    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:
+            return 403, "unknown run or task"
+        if not str(headers.get("Content-Type", "")).startswith("application/x-tar"):
+            return 415, "expected application/x-tar"
+        dest = (self.base / task).resolve()
+        try:
+            with tarfile.open(fileobj=io.BytesIO(body), 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):
+                        return 400, f"rejected member {m.name}"
+                dest.mkdir(parents=True, exist_ok=True)
+                tar.extractall(dest, members)  # every member checked above
+        except tarfile.TarError as exc:
+            return 400, f"bad tar: {exc}"
+        with self.lock:
+            self.delivered.add(task)
+        return 200, "ok"
+
+
+class _Handler(BaseHTTPRequestHandler):
+    def do_POST(self) -> None:
+        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.send_response(code)
+        self.send_header("Content-Type", "text/plain")
+        self.end_headers()
+        self.wfile.write(text.encode())
+
+    def log_message(self, format: str, *args: object) -> None:
+        return
+
+
+# --- the run -------------------------------------------------------------------------------------------------------
+
+
+@dataclass
+class _State:
+    path: Path
+    data: dict[str, Any]
+    lock: threading.Lock = field(default_factory=threading.Lock)
+
+    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)
+            _write_json(self.path, self.data)
+
+    def status(self, task: str) -> str:
+        return str(self.data["tasks"].get(task, {}).get("status", "pending"))
+
+
+def outcome_of(summary: Mapping[str, Any] | None, phase: str) -> tuple[str, str]:
+    """(status, reason) of a finished task from its delivered summary and ax's terminal phase."""
+    if _is(phase, FAILURE_PHASES):
+        return "failed", f"ax phase {phase}"
+    if summary is None:
+        return "failed", "no summary.json delivered"
+    if summary.get("status") == "done" and summary.get("units_done") == summary.get("units_total"):
+        return "done", ""
+    if summary.get("status") == "unfinished":
+        return "unfinished", str(summary.get("reason") or "budget stop")
+    return "failed", str(summary.get("reason") or summary.get("status") or "task did not report done")
+
+
+def _acquire_lock(base: Path) -> Path:
+    lock = base / "lock"
+    pid = _read_json(lock)
+    if isinstance(pid, int) and pid != os.getpid():
+        try:
+            os.kill(pid, 0)
+        except OSError:
+            pass
+        else:
+            raise RuntimeError(f"run is held by pid {pid} ({lock})")
+    lock.write_text(json.dumps(os.getpid()), encoding="utf-8")
+    return lock
+
+
+def _execute(ax: Ax, run: Run, entry: RunTask, manifests: Manifests, base: Path, receiver: Receiver, poll: float) -> tuple[str, str]:
+    ax_name = _ax_name(run.metadata.name, entry.name)
+    manifest = base / entry.name / "task.yaml"
+    manifest.parent.mkdir(parents=True, exist_ok=True)
+    docs = render_task(run, entry, manifests, base, receiver.url())
+    manifest.write_text(yaml.safe_dump_all(docs, sort_keys=False), encoding="utf-8")
+    try:
+        ax.apply(manifest)
+    except RuntimeError as exc:
+        return "failed", str(exc)
+    phase, unknown = "Pending", 0
+    try:
+        while not _is(phase, TERMINAL_PHASES):
+            time.sleep(poll)
+            phase = ax.phase(ax_name)
+            unknown = unknown + 1 if phase.startswith("Unknown") else 0
+            if unknown >= 10:  # ten consecutive unreadable polls: ax itself is broken, not the task
+                phase = "Error: " + phase
+        for _ in range(5):  # delivery may trail the phase change by a few seconds
+            if entry.name in receiver.delivered or _is(phase, FAILURE_PHASES):
+                break
+            time.sleep(poll)
+    finally:
+        ax.delete(ax_name)
+    return outcome_of(_read_json(base / entry.name / "summary.json"), phase)
+
+
+def run(run_manifest: Path, *, workers: int = 1, ax: str = "ax", results_root: Path | None = None) -> str:
+    """Execute a Run on ax and return its status: ``done``, ``unfinished`` or ``failed`` (Req 3.1-3.3, 2.2)."""
+    run_spec, manifests = load_run(Path(run_manifest))
+    base = run_dir(run_spec.metadata.name, results_root)
+    base.mkdir(parents=True, exist_ok=True)
+    lock = _acquire_lock(base)
+    poll = float(os.environ.get("OUSAST_POLL_SECONDS") or 3)
+    fresh = {"run": run_spec.metadata.name, "started": _now(), "tasks": {}}
+    state = _State(base / "state.json", _read_json(base / "state.json") or fresh)
+    receiver = Receiver(run_spec.metadata.name, base, int(os.environ.get("OUSAST_ARTIFACT_PORT") or 0))
+    threading.Thread(target=receiver.serve_forever, daemon=True).start()
+    cli, finished = Ax(ax), 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 {}))
+    running: dict[str, RunTask] = {}
+    halted = False
+
+    def worker(entry: RunTask) -> None:
+        if state.status(entry.name) in ("running", "suspended"):
+            cli.delete(_ax_name(run_spec.metadata.name, entry.name))  # an interrupted attempt; ax must not keep it
+        state.set(entry.name, status="running", started=_now(), finished=None)
+        finished.put((entry.name, *_execute(cli, run_spec, entry, manifests, base, receiver, poll)))
+
+    try:
+        while pending or running:
+            blocked = {t.serialize for t in running.values() if t.serialize}
+            for entry in list(pending):
+                if halted or len(running) >= max(workers, 1) or (entry.serialize and entry.serialize in blocked):
+                    continue
+                if not all(state.status(p) == "done" for p in entry.producers):
+                    continue
+                pending.remove(entry)
+                running[entry.name] = entry
+                blocked.add(entry.serialize or "")
+                with receiver.lock:
+                    receiver.running.add(entry.name)
+                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]
+            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()
+        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"
+    state.data.update(status=result, finished=_now())
+    _write_json(state.path, state.data)
+    return result
+
+
+# --- status and attribution (Req 7.3) ------------------------------------------------------------------------------
+
+
+def _number(value: object) -> float | None:
+    return value if isinstance(value, int | float) and not isinstance(value, bool) else None
+
+
+def _row(name: str, model: str | None, payload: Mapping[str, Any]) -> dict[str, Any]:
+    raw = payload.get("usage")
+    usage: Mapping[str, Any] = raw if isinstance(raw, Mapping) else {}
+    row: dict[str, Any] = {"task": name, "model": model, "calls": _number(payload.get("calls", payload.get("turns")))}
+    row.update({column: _number(usage.get(key)) for column, key in _USAGE_KEYS.items()})
+    row["usd"] = _number(payload.get("usd"))
+    return row
+
+
+def _sum(rows: Iterable[Mapping[str, Any]], label: str, model: str | None) -> dict[str, Any]:
+    """Column sums over rows bound to a Model; a missing value makes the sum ``n/a``, never a smaller number."""
+    priced = [r for r in rows if r.get("model")]
+    total: dict[str, Any] = {"task": label, "model": model}
+    for column in _COLUMNS:
+        values = [r.get(column) for r in priced]
+        total[column] = None if not values or any(v is None for v in values) else round(sum(v or 0 for v in values), 6)
+    return total
+
+
+def _fmt(row: Mapping[str, Any]) -> str:
+    cells = [str(row["task"])[:28], str(row.get("model") or "-")[:20]]
+    for column in _COLUMNS:
+        value = row.get(column)
+        cells.append("n/a" if value is None else f"{value:.4f}" if column == "usd" else str(int(value)))
+    return _ROW.format(*cells)
+
+
+def attribution(run_name: str, *, units: bool = False, results_root: Path | None = None) -> dict[str, Any]:
+    """Token attribution from the tasks' ``summary.json`` usage fields; ``units`` adds rows from ``units.jsonl``."""
+    base = run_dir(run_name, results_root)
+    state = _read_json(base / "state.json") or {}
+    tasks: dict[str, dict[str, Any]] = state.get("tasks") or {}
+    rows: list[dict[str, Any]] = []
+    unit_rows: dict[str, list[dict[str, Any]]] = {}
+    for name, info in tasks.items():
+        summary = _read_json(base / name / "summary.json") or {}
+        rows.append({**_row(name, summary.get("model") or info.get("model"), summary), "status": info.get("status")})
+        if units and (base / name / "units.jsonl").is_file():
+            lines = (base / name / "units.jsonl").read_text(encoding="utf-8").splitlines()
+            unit_rows[name] = [
+                _row(str(u.get("unit") or u.get("path") or u.get("candidate") or f"unit {i}"), rows[-1]["model"], u)
+                for i, u in enumerate(json.loads(line) for line in lines if line.strip())
+            ]
+    names = sorted({r["model"] for r in rows if r["model"]})
+    models = [_sum([r for r in rows if r["model"] == m], f"subtotal {m}", m) for m in names]
+    return {"run": run_name, "tasks": rows, "models": models, "total": _sum(rows, "total", None), "units": unit_rows}
+
+
+def status(run_name: str, *, units: bool = False, results_root: Path | None = None) -> str:
+    """Per-task status and the attribution table (Req 3.2, 7.3); written to ``<run dir>/attribution.json``."""
+    report = attribution(run_name, units=units, results_root=results_root)
+    base = run_dir(run_name, results_root)
+    state = _read_json(base / "state.json") or {}
+    header = _ROW.format("TASK", "MODEL", "CALLS", "PROMPT", "CACHE-HIT", "OUTPUT", "USD")
+    lines = [f"run {run_name}: {state.get('status') or 'not started'} ({base})", "", header]
+    for row in report["tasks"]:
+        lines.append(_fmt(row) + f"  [{row.get('status') or 'pending'}]")
+        lines.extend("  " + _fmt(u) for u in report["units"].get(row["task"], []))
+    lines.append("-" * len(header))
+    lines.extend(_fmt(m) for m in report["models"])
+    lines.append(_fmt(report["total"]))
+    if base.is_dir():
+        _write_json(base / "attribution.json", report)
+    return "\n".join(lines)
+
+
+# --- doctor (Req 3.5) ----------------------------------------------------------------------------------------------
+
+
+def _sh(*args: str) -> tuple[bool, str]:
+    try:
+        proc = subprocess.run(args, capture_output=True, text=True, timeout=60, check=False)
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        return False, str(exc)
+    return proc.returncode == 0, (proc.stdout + proc.stderr).strip()
+
+
+def _pods_ready(namespace: str, context: str) -> tuple[bool, str]:
+    ok, out = _sh("kubectl", "--context", context, "-n", namespace, "get", "pods", "--no-headers")
+    if not ok or not out:
+        return False, out or "no pods"
+    rows = [line.split() for line in out.splitlines()]
+    bad = [r[0] for r in rows if len(r) > 2 and r[2] not in ("Completed", "Succeeded") and r[1].split("/")[0] != r[1].split("/")[-1]]
+    return not bad, f"not ready: {', '.join(bad)}" if bad else f"{len(rows)} pods ready"
+
+
+def doctor() -> list[tuple[str, bool, str]]:
+    """Four checks of the host's ax: kind, Agent Substrate, the ax controller, the runner image (Req 3.5)."""
+    context = "kind-" + (os.environ.get("KIND_CLUSTER_NAME") or "ousast")
+    ok, out = _sh("kubectl", "--context", context, "cluster-info")
+    checks = [("kind cluster", ok, "reachable" if ok else f"context {context} unreachable ({out[:80]}); {UP}")]
+    for label, ns in (("agent substrate (ate-system)", "ate-system"), ("ax controller (ax-system)", "ax-system")):
+        ok, out = _pods_ready(ns, context)
+        checks.append((label, ok, out if ok else f"{out}; {UP}"))
+    ok, out = _sh("curl", "-s", "localhost:5001/v2/_catalog")
+    present = ok and "ousast-runner" in out
+    text = "ousast-runner present" if present else f"missing from localhost:5001 ({out[:80]}); {UP} and loads it"
+    checks.append(("runner image in kind registry", present, text))
+    return checks
+
+
+__all__ = ["Ax", "Receiver", "attribution", "doctor", "load_run", "outcome_of", "render_task", "run", "run_dir", "status"]

src/openultrasast/plane/workspaces.py

@@ -0,0 +1,33 @@
+"""Workspace manifests for the cases of an independent population file (ai-service-plane task 4).
+
+``ousast plane workspaces <population.toml> [--out DIR]`` writes one ax ``Workspace`` per case pin (vulnerable,
+fixed, benign base, benign tip) with the case's repository URL and the pinned commit, so a Run binds a case at a
+pin by name. The layout of a case is that of ``benchmarks/independent/evaluate.py``'s ``pins()``.
+"""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+
+import yaml
+
+from .reconciler import _ax_name, _doc
+
+__all__ = ["workspaces"]
+
+
+def workspaces(population: Path, out: Path) -> list[Path]:
+    """One Workspace manifest per case pin of a population file; returns the paths written."""
+    out.mkdir(parents=True, exist_ok=True)
+    written: list[Path] = []
+    for case in tomllib.loads(population.read_text(encoding="utf-8"))["case"]:
+        benign = case["benign"]
+        pins = {"vulnerable": case["vulnerable"], "fixed": case["fixed"], "benign-base": benign["base"], "benign-tip": benign["tip"]}
+        for label, commit in pins.items():
+            name = _ax_name(case["id"], label)
+            doc = _doc("Workspace", name, {"git": [{"name": "repo", "repo": case["repo"], "commit": commit}]})
+            text = f"# generated by `ousast plane workspaces` from {population.name}\n" + yaml.safe_dump(doc, sort_keys=False)
+            (out / f"{name}.yaml").write_text(text, encoding="utf-8")
+            written.append(out / f"{name}.yaml")
+    return written

tests/test_plane_ax_live.py

@@ -0,0 +1,43 @@
+"""ai-service-plane Req 3.5/3.6: one trivial Task end to end on the real ax cluster of ``ops/ax/up.sh``.
+
+Marker ``ax``; skipped unless every check of ``ousast plane doctor`` passes, so the host suite never depends on
+the cluster. Run it on purpose: ``pytest -m ax tests/test_plane_ax_live.py``.
+"""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+from openultrasast.plane.reconciler import Ax, doctor
+
+pytestmark = pytest.mark.ax
+SMOKE = Path("ops/ax/smoke-task.yaml")
+
+
+@pytest.fixture(scope="module")
+def live_ax() -> Ax:
+    checks = doctor()
+    failing = [f"{name}: {text}" for name, ok, text in checks if not ok]
+    if failing:
+        pytest.skip("no live ax cluster: " + "; ".join(failing))
+    if not SMOKE.is_file():
+        pytest.skip(f"{SMOKE} is not in this checkout")
+    return Ax()
+
+
+def test_smoke_task_completes_on_the_cluster(live_ax: Ax) -> None:
+    live_ax.apply(SMOKE)
+    deadline = time.monotonic() + 600
+    phase = "Pending"
+    try:
+        while time.monotonic() < deadline:
+            phase = live_ax.phase("ousast-smoke")
+            if any(word in phase.lower() for word in ("completed", "succeeded", "failed", "error")):
+                break
+            time.sleep(5)
+    finally:
+        live_ax.delete("ousast-smoke")
+    assert phase.lower() in ("completed", "succeeded"), f"smoke task ended in phase {phase!r}"

tests/test_plane_reconciler.py

@@ -0,0 +1,485 @@
+"""ai-service-plane Req 3.1-3.5, 2.2, 7.3: a fake ``ax`` drives a Run through the reconciler.
+
+The fake is a Python script the tests write into ``tmp_path``: it logs every argv line, parses what ``apply``
+receives, answers ``get task`` with Running until a scripted poll count and then a terminal phase, and on reaching
+it POSTs a scripted tar to the ``OUSAST_ARTIFACT_URL`` it found in the applied Task's env. Nothing here runs a
+task; every outcome comes from a scripted ``summary.json`` delivery.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import stat
+import subprocess
+import sys
+import tarfile
+import time
+from pathlib import Path
+
+import pytest
+import yaml
+
+from openultrasast.plane import reconciler
+from openultrasast.plane.workspaces import workspaces
+
+RECONCILER = Path("src/openultrasast/plane/reconciler.py")
+
+FAKE_AX = '''#!{python}
+"""Fake ax CLI for the reconciler tests: argv log, scripted phases, scripted deliveries."""
+import io, json, sys, tarfile, time, urllib.error, urllib.request
+from pathlib import Path
+import yaml
+
+HERE = Path(__file__).resolve().parent
+SCRIPT = json.loads((HERE / "script.json").read_text())
+argv = sys.argv[1:]
+(HERE / "ax.log").open("a").write(f"{{time.monotonic():.4f}} " + " ".join(argv) + "\\n")
+(HERE / "applied").mkdir(exist_ok=True)
+if argv[:2] == ["apply", "-f"]:
+    docs = list(yaml.safe_load_all(Path(argv[2]).read_text()))
+    for doc in docs:
+        if doc["kind"] == "Task":
+            env = {{e["name"]: e["value"] for e in doc["spec"].get("env", [])}}
+            (HERE / "applied" / (doc["metadata"]["name"] + ".json")).write_text(json.dumps({{"docs": docs, "env": env, "polls": 0}}))
+    print("task created")
+elif argv[:2] == ["get", "task"]:
+    record = HERE / "applied" / (argv[2] + ".json")
+    if not record.exists():
+        print("Error: not found", file=sys.stderr)
+        sys.exit(1)
+    rec = json.loads(record.read_text())
+    rec["polls"] += 1
+    env, task = rec["env"], rec["env"]["OUSAST_TASK"]
+    script = SCRIPT.get(task, {{}})
+    needed = script.get("polls", 1)
+    phase = "Running" if rec["polls"] < needed else script.get("phase", "Completed")
+    if rec["polls"] == needed and script.get("deliver", True):
+        buf = io.BytesIO()
+        with tarfile.open(fileobj=buf, mode="w") as tar:
+            for path, content in script.get("files", {{}}).items():
+                data = (content if isinstance(content, str) else json.dumps(content)).encode()
+                info = tarfile.TarInfo(path)
+                info.size = len(data)
+                tar.addfile(info, io.BytesIO(data))
+        headers = {{"Content-Type": "application/x-tar", "X-Ousast-Task": script.get("as_task", task), "X-Ousast-Run": env["OUSAST_RUN"]}}
+        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")
+        except urllib.error.HTTPError as exc:
+            (HERE / "ax.log").open("a").write(f"{{time.monotonic():.4f}} rejected {{task}} {{exc.code}}\\n")
+    record.write_text(json.dumps(rec))
+    print(yaml.safe_dump({{"apiVersion": "ax.io/v1alpha1", "kind": "Task", "metadata": {{"name": argv[2]}}, "status": {{"phase": phase}}}}))
+elif argv[:2] == ["delete", "task"]:
+    print("deleted")
+'''
+
+MANIFESTS = """
+apiVersion: ax.io/v1alpha1
+kind: Model
+metadata:
+  name: deepseek-flash
+  annotations: {openultrasast.io/provider-extension: "true"}
+spec:
+  provider: deepseek
+  model: deepseek-chat
+  parameters: {cache_hit_per_m: 0.07, input_per_m: 0.27, output_per_m: 1.1}
+---
+apiVersion: ax.io/v1alpha1
+kind: Workspace
+metadata:
+  name: case-pin
+spec:
+  git: [{name: repo, repo: https://example.invalid/repo.git, commit: abc123}]
+---
+apiVersion: ax.io/v1alpha1
+kind: Task
+metadata:
+  name: repo-facts
+spec:
+  image: ousast-runner:dev
+  command: [repo_facts]
+  workspaces: [{name: case-pin, path: /workspace}]
+---
+apiVersion: ax.io/v1alpha1
+kind: Task
+metadata:
+  name: verify
+  annotations: {openultrasast.io/model: deepseek-flash}
+spec:
+  image: ousast-runner:dev
+  command: [verify]
+  env: [{name: OUSAST_PASS, value: a}]
+  workspaces: [{name: case-pin, path: /workspace}]
+---
+apiVersion: ax.io/v1alpha1
+kind: Task
+metadata:
+  name: agree
+spec:
+  command: [agree]
+---
+apiVersion: openultrasast.io/v1alpha1
+kind: Run
+metadata:
+  name: {run}
+spec:
+  tasks:
+{tasks}
+"""
+
+CHAIN = """
+    - name: repo-facts
+      task: repo-facts
+      outputs: [facts.json]
+      serialize: engine
+    - name: verify
+      task: verify
+      inputs: {facts: repo-facts/facts.json}
+      outputs: [agreed.json]
+      budget: {usd: 2.5, calls: 40}
+    - name: agree
+      task: agree
+      dependsOn: [verify]
+      inputs: {pass: verify/agreed.json}
+"""
+
+FACTS_SUMMARY = {"status": "done", "units_done": 3, "units_total": 3, "usd": None, "calls": 0, "usage": {}}
+VERIFY_SUMMARY = {
+    "status": "done",
+    "units_done": 2,
+    "units_total": 2,
+    "usd": 0.5,
+    "calls": 4,
+    "model": "deepseek-flash",
+    "usage": {"prompt_tokens": 1000, "prompt_cache_hit_tokens": 300, "completion_tokens": 200},
+}
+AGREE_SUMMARY = {"status": "done", "units_done": 1, "units_total": 1, "usd": None, "calls": 0}
+SCRIPT = {
+    "repo-facts": {"polls": 2, "files": {"summary.json": FACTS_SUMMARY, "facts.json": {"a.py": {"functions": ["f"]}}}},
+    "verify": {"polls": 2, "files": {"summary.json": VERIFY_SUMMARY, "agreed.json": {"agreed": ["a.py:f"]}}},
+    "agree": {"polls": 1, "files": {"summary.json": AGREE_SUMMARY}},
+}
+
+
+class Fake:
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.ax = root / "ax"
+        self.ax.write_text(FAKE_AX.format(python=sys.executable), encoding="utf-8")
+        self.ax.chmod(self.ax.stat().st_mode | stat.S_IXUSR)
+
+    def script(self, script: dict) -> None:
+        (self.root / "script.json").write_text(json.dumps(script), encoding="utf-8")
+
+    def log(self) -> list[tuple[float, list[str]]]:
+        lines = (self.root / "ax.log").read_text(encoding="utf-8").splitlines() if (self.root / "ax.log").exists() else []
+        return [(float(line.split()[0]), line.split()[1:]) for line in lines]
+
+    def applied(self, run: str, task: str) -> dict:
+        return json.loads((self.root / "applied" / f"{run}-{task}.json").read_text(encoding="utf-8"))
+
+    def events(self, verb: str) -> list[tuple[float, str]]:
+        """(time, name) of every ``apply``/``delete``/``delivered``/``rejected`` line, name = run task or ax name."""
+        out = []
+        for stamp, words in self.log():
+            if verb == "apply" and words[:2] == ["apply", "-f"]:
+                out.append((stamp, Path(words[2]).parent.name))
+            elif verb == "delete" and words[:2] == ["delete", "task"]:
+                out.append((stamp, words[2]))
+            elif verb in ("delivered", "rejected") and words[0] == verb:
+                out.append((stamp, words[1]))
+        return out
+
+
+@pytest.fixture
+def fake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Fake:
+    monkeypatch.setenv("OUSAST_RESULTS", str(tmp_path / "results"))
+    monkeypatch.setenv("OUSAST_ARTIFACT_HOST", "127.0.0.1")
+    monkeypatch.setenv("OUSAST_POLL_SECONDS", "0.02")
+    fake = Fake(tmp_path)
+    fake.script(SCRIPT)
+    return fake
+
+
+def write_run(tmp_path: Path, name: str, tasks: str = CHAIN) -> Path:
+    path = tmp_path / f"{name}.yaml"
+    path.write_text(MANIFESTS.replace("{run}", name).replace("{tasks}", tasks.rstrip("\n")), encoding="utf-8")
+    return path
+
+
+def state_of(name: str) -> dict:
+    return json.loads((reconciler.run_dir(name) / "state.json").read_text(encoding="utf-8"))
+
+
+def statuses(name: str) -> dict[str, str]:
+    return {task: info["status"] for task, info in state_of(name)["tasks"].items()}
+
+
+# --- the DAG through ax -------------------------------------------------------------------------------------------
+
+
+def test_chain_runs_in_order_with_inputs_env_and_model(fake: Fake, tmp_path: Path) -> None:
+    result = reconciler.run(write_run(tmp_path, "chain"), ax=str(fake.ax))
+    assert result == "done"
+    assert statuses("chain") == {"repo-facts": "done", "verify": "done", "agree": "done"}
+    applies = [name for _, name in fake.events("apply")]
+    assert applies == ["repo-facts", "verify", "agree"]
+    assert [n for _, n in fake.events("delete")] == ["chain-repo-facts", "chain-verify", "chain-agree"]
+    verify = fake.applied("chain", "verify")
+    env = verify["env"]
+    assert env["OUSAST_PASS"] == "a" and env["OUSAST_RUN"] == "chain" and env["OUSAST_TASK"] == "verify"
+    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 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 == [("Task", "chain-verify"), ("Model", "deepseek-flash"), ("Workspace", "case-pin"), ("Workspace", "chain-verify-inputs")]
+    inputs_ws = verify["docs"][3]
+    assert inputs_ws["spec"]["files"] == [{"path": "repo-facts/facts.json", "content": json.dumps({"a.py": {"functions": ["f"]}})}]
+    assert verify["docs"][0]["spec"]["workspaces"][-1] == {"name": "chain-verify-inputs", "path": "/workspace/.ousast-in/verify"}
+    assert "annotations" not in verify["docs"][0]["metadata"], "ax's ObjectMeta has no annotations"
+    facts = fake.applied("chain", "repo-facts")
+    assert "OUSAST_MODEL" not in facts["env"] and "OUSAST_BUDGET_USD" not in facts["env"]
+    assert facts["docs"][0]["spec"]["image"] == "ousast-runner:dev"
+    delivered = reconciler.run_dir("chain") / "verify" / "agreed.json"
+    assert json.loads(delivered.read_text()) == {"agreed": ["a.py:f"]}
+    assert state_of("chain")["tasks"]["verify"]["usd"] == 0.5 and state_of("chain")["tasks"]["verify"]["model"] == "deepseek-flash"
+    assert not (reconciler.run_dir("chain") / "lock").exists()
+
+
+PAIR = """
+    - name: x
+      task: repo-facts
+      {serialize}
+    - name: y
+      task: repo-facts
+      {serialize}
+"""
+
+
+def overlap(fake: Fake) -> bool:
+    applies = dict((name, stamp) for stamp, name in fake.events("apply"))
+    deletes = dict((name.split("-", 1)[1], stamp) for stamp, name in fake.events("delete"))
+    return max(applies.values()) < min(deletes.values())
+
+
+def test_serialize_label_never_overlaps_but_others_do(fake: Fake, tmp_path: Path) -> None:
+    fake.script({"x": {"polls": 4, "files": {"summary.json": FACTS_SUMMARY}}, "y": {"polls": 4, "files": {"summary.json": FACTS_SUMMARY}}})
+    assert reconciler.run(write_run(tmp_path, "par", PAIR.replace("{serialize}", "")), ax=str(fake.ax), workers=2) == "done"
+    assert overlap(fake), "two unlabeled tasks with --workers 2 run at once"
+    (fake.root / "ax.log").unlink()
+    serialized = write_run(tmp_path, "ser", PAIR.replace("{serialize}", "serialize: engine"))
+    assert reconciler.run(serialized, ax=str(fake.ax), workers=2) == "done"
+    assert not overlap(fake), "tasks sharing a serialize label never overlap"
+
+
+def test_unfinished_stops_dependants_and_rerun_skips_done(fake: Fake, tmp_path: Path) -> None:
+    script = json.loads(json.dumps(SCRIPT))
+    script["verify"]["files"]["summary.json"] = {**VERIFY_SUMMARY, "status": "unfinished", "units_done": 1, "reason": "budget"}
+    fake.script(script)
+    manifest = write_run(tmp_path, "resume")
+    assert reconciler.run(manifest, ax=str(fake.ax)) == "unfinished"
+    assert statuses("resume") == {"repo-facts": "done", "verify": "unfinished", "agree": "pending"}
+    assert state_of("resume")["tasks"]["verify"]["reason"] == "budget"
+    fake.script(SCRIPT)
+    (fake.root / "ax.log").unlink()
+    assert reconciler.run(manifest, ax=str(fake.ax)) == "done"
+    assert [name for _, name in fake.events("apply")] == ["verify", "agree"], "done tasks are skipped on rerun (Req 2.2, 3.3)"
+    assert statuses("resume") == {"repo-facts": "done", "verify": "done", "agree": "done"}
+
+
+def test_failed_task_halts_the_run(fake: Fake, tmp_path: Path) -> None:
+    script = json.loads(json.dumps(SCRIPT))
+    failed = {"status": "failed", "units_done": 0, "units_total": 3, "reason": "402 insufficient balance"}
+    script["repo-facts"]["files"]["summary.json"] = failed
+    fake.script(script)
+    assert reconciler.run(write_run(tmp_path, "halt"), ax=str(fake.ax)) == "failed"
+    assert statuses("halt") == {"repo-facts": "failed", "verify": "pending", "agree": "pending"}
+    assert state_of("halt")["tasks"]["repo-facts"]["reason"] == "402 insufficient balance"
+    assert [name for _, name in fake.events("apply")] == ["repo-facts"]
+
+
+def test_ax_failure_phase_and_missing_delivery_are_failed(fake: Fake, tmp_path: Path) -> None:
+    script = json.loads(json.dumps(SCRIPT))
+    script["repo-facts"]["phase"] = "Failed"
+    fake.script(script)
+    assert reconciler.run(write_run(tmp_path, "phase"), ax=str(fake.ax)) == "failed"
+    assert state_of("phase")["tasks"]["repo-facts"]["reason"] == "ax phase Failed"
+    script = json.loads(json.dumps(SCRIPT))
+    script["repo-facts"]["deliver"] = False
+    fake.script(script)
+    assert reconciler.run(write_run(tmp_path, "silent"), ax=str(fake.ax)) == "failed"
+    assert state_of("silent")["tasks"]["repo-facts"]["reason"] == "no summary.json delivered"
+    assert [n for _, n in fake.events("delete")] == ["phase-repo-facts", "silent-repo-facts"], "ax delete follows every terminal task"
+
+
+def test_summary_done_requires_every_unit(fake: Fake, tmp_path: Path) -> None:
+    script = json.loads(json.dumps(SCRIPT))
+    script["repo-facts"]["files"]["summary.json"] = {**FACTS_SUMMARY, "units_done": 2}
+    fake.script(script)
+    assert reconciler.run(write_run(tmp_path, "short"), ax=str(fake.ax)) == "failed"
+    assert statuses("short")["repo-facts"] == "failed"
+
+
+# --- receiver and lock ---------------------------------------------------------------------------------------------
+
+
+def tar_of(files: dict[str, bytes]) -> bytes:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w") as tar:
+        for name, data in files.items():
+            info = tarfile.TarInfo(name)
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+    return buf.getvalue()
+
+
+def test_receiver_rejects_traversal_unknown_tasks_and_wrong_types(tmp_path: Path) -> None:
+    receiver = reconciler.Receiver("r", tmp_path / "r", 0)
+    receiver.running.add("t")
+    headers = {"X-Ousast-Run": "r", "X-Ousast-Task": "t", "Content-Type": "application/x-tar"}
+    code, text = receiver.accept(headers, tar_of({"../escape.json": b"{}"}))
+    assert code == 400 and "escape" in text and not (tmp_path / "escape.json").exists()
+    assert receiver.accept({**headers, "X-Ousast-Task": "other"}, tar_of({"summary.json": b"{}"}))[0] == 403
+    assert receiver.accept({**headers, "X-Ousast-Run": "someone"}, tar_of({"summary.json": b"{}"}))[0] == 403
+    assert receiver.accept({**headers, "Content-Type": "application/json"}, b"{}")[0] == 415
+    assert receiver.accept(headers, b"not a tar")[0] == 400
+    assert receiver.accept(headers, tar_of({"summary.json": b"{}", "sub/x.json": b"1"}))[0] == 200
+    assert (tmp_path / "r" / "t" / "sub" / "x.json").read_text() == "1" and "t" in receiver.delivered
+    receiver.server_close()
+
+
+def test_delivery_for_a_task_that_is_not_running_is_rejected(fake: Fake, tmp_path: Path) -> None:
+    script = json.loads(json.dumps(SCRIPT))
+    script["repo-facts"]["as_task"] = "agree"  # the runner claims to be a task that has not started
+    fake.script(script)
+    assert reconciler.run(write_run(tmp_path, "spoof"), ax=str(fake.ax)) == "failed"
+    assert [n for _, n in fake.events("rejected")] == ["repo-facts"]
+    assert not (reconciler.run_dir("spoof") / "agree" / "summary.json").exists()
+
+
+def test_lock_refuses_a_second_runner_and_ignores_a_dead_one(fake: Fake, tmp_path: Path) -> None:
+    manifest = write_run(tmp_path, "locked")
+    base = reconciler.run_dir("locked")
+    base.mkdir(parents=True)
+    holder = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+    try:
+        (base / "lock").write_text(str(holder.pid))
+        with pytest.raises(RuntimeError, match=f"held by pid {holder.pid}"):
+            reconciler.run(manifest, ax=str(fake.ax))
+        assert not fake.events("apply")
+    finally:
+        holder.kill()
+        holder.wait()
+    (base / "lock").write_text(str(holder.pid))  # the process is gone: a stale lock
+    assert reconciler.run(manifest, ax=str(fake.ax)) == "done"
+
+
+# --- status and attribution (Req 7.3) ------------------------------------------------------------------------------
+
+
+def usage(prompt: int, cache_hit: int, output: int) -> dict[str, int]:
+    return {"prompt_tokens": prompt, "prompt_cache_hit_tokens": cache_hit, "completion_tokens": output}
+
+
+def test_status_prints_attribution_from_usage_fields_only(fake: Fake, tmp_path: Path) -> None:
+    script = json.loads(json.dumps(SCRIPT))
+    script["agree"]["files"]["summary.json"] = {**AGREE_SUMMARY, "model": "deepseek-flash", "usd": 0.1, "calls": 1}  # no usage
+    script["verify"]["files"]["units.jsonl"] = "\n".join(
+        json.dumps(u)
+        for u in (
+            {"path": "a.py", "turns": 3, "usd": 0.3, "usage": usage(700, 300, 150)},
+            {"path": "b.py", "turns": 1, "usd": 0.2, "usage": usage(300, 0, 50)},
+        )
+    )
+    fake.script(script)
+    assert reconciler.run(write_run(tmp_path, "attr"), ax=str(fake.ax)) == "done"
+    text = reconciler.status("attr", units=True)
+    lines = {line.split()[0]: line.split() for line in text.splitlines() if line.strip() and not line.startswith("run ")}
+    assert lines["verify"][1:7] == ["deepseek-flash", "4", "1000", "300", "200", "0.5000"]
+    assert lines["agree"][1:7] == ["deepseek-flash", "1", "n/a", "n/a", "n/a", "0.1000"], "missing usage is n/a, never estimated"
+    assert lines["repo-facts"][1:7] == ["-", "0", "n/a", "n/a", "n/a", "n/a"]
+    assert lines["a.py"][1:7] == ["deepseek-flash", "3", "700", "300", "150", "0.3000"]
+    assert lines["b.py"][1:7] == ["deepseek-flash", "1", "300", "0", "50", "0.2000"]
+    assert lines["subtotal"][1:8] == ["deepseek-flash", "deepseek-flash", "5", "n/a", "n/a", "n/a", "0.6000"]
+    assert lines["total"][1:7] == ["-", "5", "n/a", "n/a", "n/a", "0.6000"]
+    report = json.loads((reconciler.run_dir("attr") / "attribution.json").read_text())
+    assert report["total"] == {"task": "total", "model": None, "calls": 5, "prompt": None, "cache_hit": None, "output": None, "usd": 0.6}
+    assert [u["task"] for u in report["units"]["verify"]] == ["a.py", "b.py"]
+    assert [row["status"] for row in report["tasks"]] == ["done", "done", "done"]
+    fake.script(SCRIPT)
+    reconciler.run(write_run(tmp_path, "attr2"), ax=str(fake.ax))
+    report = reconciler.attribution("attr2")
+    assert report["total"]["prompt"] == 1000 and report["total"]["cache_hit"] == 300 and report["total"]["output"] == 200
+    assert report["models"][0]["task"] == "subtotal deepseek-flash" and report["models"][0]["usd"] == 0.5
+
+
+def test_status_of_an_unknown_run_says_so(fake: Fake) -> None:
+    assert "not started" in reconciler.status("nothing")
+
+
+# --- doctor, workspaces, size ---------------------------------------------------------------------------------------
+
+
+def test_doctor_names_the_bring_up_script_when_nothing_is_reachable(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("PATH", str(tmp_path))  # no kubectl, no curl
+    checks = reconciler.doctor()
+    assert [name for name, _, _ in checks] == [
+        "kind cluster",
+        "agent substrate (ate-system)",
+        "ax controller (ax-system)",
+        "runner image in kind registry",
+    ]
+    assert all(not ok and "ops/ax/up.sh" in text for _, ok, text in checks)
+
+
+def test_workspaces_one_manifest_per_case_pin(tmp_path: Path) -> None:
+    population = tmp_path / "population.toml"
+    population.write_text(
+        '[[case]]\nid = "demo-sqli"\nrepo = "https://github.com/x/y"\nvulnerable = "aaa"\nfixed = "bbb"\n'
+        'benign = { base = "ccc", tip = "ddd" }\n',
+        encoding="utf-8",
+    )
+    written = workspaces(population, tmp_path / "out")
+    labels = ("vulnerable", "fixed", "benign-base", "benign-tip")
+    assert [p.name for p in written] == [f"demo-sqli-{label}.yaml" for label in labels]
+    doc = yaml.safe_load(written[0].read_text(encoding="utf-8"))
+    assert doc == {
+        "apiVersion": "ax.io/v1alpha1",
+        "kind": "Workspace",
+        "metadata": {"name": "demo-sqli-vulnerable"},
+        "spec": {"git": [{"name": "repo", "repo": "https://github.com/x/y", "commit": "aaa"}]},
+    }
+    from openultrasast.plane.manifests import load_manifests
+
+    assert set(load_manifests(written).workspaces) == {p.stem for p in written}
+
+
+def test_reconciler_is_under_500_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"
+    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"
+
+
+def test_cli_wires_plane_subcommands() -> None:
+    from openultrasast.cli import main
+
+    with pytest.raises(SystemExit) as exc:
+        main(["plane", "--help"])
+    assert exc.value.code == 0
+    with pytest.raises(SystemExit):
+        main(["plane"])
+
+
+def test_elapsed_polling_is_paced(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OUSAST_POLL_SECONDS", "0.1")
+    fake.script({"x": {"polls": 3, "files": {"summary.json": FACTS_SUMMARY}}, "y": {"polls": 1, "files": {"summary.json": FACTS_SUMMARY}}})
+    started = time.monotonic()
+    assert reconciler.run(write_run(tmp_path, "paced", PAIR.replace("{serialize}", "")), ax=str(fake.ax)) == "done"
+    assert time.monotonic() - started >= 0.4, "three polls of x plus one of y at 0.1s each"