feat(plane): runner contract with artifact delivery (task 3)

ai-service-plane task 3, Requirements 4.1-4.4 (with 2.2 and 2.3 honoured by the
exit-code contract):

- src/openultrasast/plane/runner.py: the ax-task-runner entrypoint. Reads the Task
  from AX_TASK_YAML and the bound Workspaces from AX_WORKSPACES_YAML (task-1 loaders),
  serves /healthz and /readyz on port 80 (AX_RUNNER_PORT; 0 picks a free port,
  AX_RUNNER_HTTP=0 disables it for unit tests), materialises each workspace at its
  bound path (spec.files verbatim; spec.git at <path>/<name>, exported from the
  OUSAST_CASE_CACHE checkout whose origin matches, else a shallow clone at the pin;
  a repo already present on a resumed volume is kept) and only then answers 200 on
  /readyz. Runs spec.command by importing openultrasast.plane.tasks.<name> with the
  remaining elements as argv; sets OUSAST_WORKSPACE_DIR from the first binding when
  the Task's env leaves it unset. A crash writes summary.json with status failed and
  the traceback. When OUSAST_ARTIFACT_URL is set, the whole OUSAST_OUTPUT_DIR is
  POSTed as an uncompressed tar (Content-Type application/x-tar, X-Ousast-Task,
  X-Ousast-Run) with three attempts and backoff; a failed delivery rewrites
  summary.json to failed and exits 2. Exit codes: done 0, failed 2, unfinished 3.
- plane/Dockerfile.runner: python:3.12-slim + git, `pip install .` (no extras), a
  shell wrapper at /usr/local/bin/ax-task-runner exec'ing the module; built and
  loaded into kind by ops/ax/up.sh, not by the tests.
- tests/test_plane_runner.py (+ tests/plane_stub_task.py registered as the task
  module): env contract on a files workspace, readyz 503 then 200, delivery to a
  local receiver with the headers and tar checked, delivery failure -> failed and
  exit 2 after three attempts, crash -> failed with traceback, exit codes per
  status, case-cache export at the pin, shallow-clone fallback, resume keeps the
  checkout.
- model/audit.py: SPEC_OWNED row for plane.runner; module-audit measurement
  regenerated (122 modules, no orphans, note kept).

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

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

@@ -1,7 +1,7 @@
 {
-  "modules": 121,
+  "modules": 122,
   "counts": {
-    "load_bearing": 112,
+    "load_bearing": 113,
     "standalone_capability": 9
   },
   "orphaned": [],
@@ -653,7 +653,9 @@
       "module": "plane",
       "classification": "load_bearing",
       "reason": "ai-service-plane Req 1-4: the service plane package; runner and reconciler integration in tasks 3-4",
-      "importers": []
+      "importers": [
+        "plane.runner"
+      ]
     },
     {
       "module": "plane.budget",
@@ -665,6 +667,14 @@
       "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": [
+        "plane.runner"
+      ]
+    },
+    {
+      "module": "plane.runner",
+      "classification": "load_bearing",
+      "reason": "ai-service-plane Req 4.1-4.4: the ax-task-runner entrypoint (PID 1 of the task image); delivers the output directory",
       "importers": []
     },
     {

plane/Dockerfile.runner

@@ -0,0 +1,32 @@
+# The ax task-runner image for this project (ai-service-plane Requirement 4.2).
+#
+# ax's default runner image is gcr.io/ax-substrate/ate-images/ax-task-runner@sha256:3a0dea6a... (Python 3.12,
+# git, curl, ax's Go runner as PID 1). This image keeps the same base and the same contract -- PID 1 is
+# /usr/local/bin/ax-task-runner, fed AX_TASK_YAML / AX_WORKSPACES_YAML, serving /healthz and /readyz on port 80 --
+# but the runner is this package's `openultrasast.plane.runner`, so a task's spec.command names a module under
+# `openultrasast.plane.tasks` and the output directory is delivered to the reconciler at exit.
+#
+# Build from the repository root (the package is the context):
+#     docker build -f plane/Dockerfile.runner -t openultrasast-runner:dev .
+# `ops/ax/up.sh` builds it and loads it into kind; nothing here is exercised by the unit tests.
+
+FROM python:3.12-slim
+
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends git ca-certificates \
+ && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /opt/openultrasast
+COPY pyproject.toml README.md ./
+COPY src ./src
+# Core install only: PyYAML for the manifests, no extras (no HarnessX, no tree-sitter).
+RUN pip install --no-cache-dir . \
+ && printf '#!/bin/sh\nexec python -m openultrasast.plane.runner "$@"\n' > /usr/local/bin/ax-task-runner \
+ && chmod 0755 /usr/local/bin/ax-task-runner
+
+# ax mounts the persistent volume here; it is also the command's working directory in ax's own runner.
+RUN mkdir -p /workspace
+WORKDIR /workspace
+
+EXPOSE 80
+ENTRYPOINT ["/usr/local/bin/ax-task-runner"]

src/openultrasast/model/audit.py

@@ -57,6 +57,7 @@ SPEC_OWNED = {
     "plane": "ai-service-plane Req 1-4: the service plane package; runner and reconciler integration in tasks 3-4",
     "plane.budget": "ai-service-plane Req 2.3/2.4: metered chat client with usd and calls ceilings; task integration in tasks 3, 6",
     "plane.manifests": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs",
+    "plane.runner": "ai-service-plane Req 4.1-4.4: the ax-task-runner entrypoint (PID 1 of the task image); delivers the output directory",
 }
 
 STANDALONE = {

src/openultrasast/plane/runner.py

@@ -0,0 +1,477 @@
+"""ax's runner contract for a task container (ai-service-plane Requirement 4).
+
+This module is ``/usr/local/bin/ax-task-runner`` in the runner image, PID 1 of the task container. ax hands it
+the Task as YAML in ``AX_TASK_YAML`` and the bound Workspaces as a multi-document stream in ``AX_WORKSPACES_YAML``;
+nothing else comes from the command line (Requirement 4.1). It serves ``/healthz`` and ``/readyz`` on port 80,
+``/readyz`` answering 503 until every workspace is materialised (Requirement 4.2), then runs ``spec.command``:
+its first element names ``openultrasast.plane.tasks.<name>`` whose ``main(argv)`` gets the rest.
+
+ax has no artifact channel, so after the task returns the runner posts the whole ``OUSAST_OUTPUT_DIR`` as a tar
+stream to ``OUSAST_ARTIFACT_URL`` before it reports completion; a failed delivery is reported as ``failed``, never
+as done (Requirement 4.4). The exit code follows the task's ``summary.json``: 0 ``done``, 2 ``failed``, 3
+``unfinished``.
+
+Environment the runner reads, all optional except the two ax variables:
+
+    AX_TASK_YAML, AX_WORKSPACES_YAML   ax's contract
+    AX_RUNNER_HTTP                     "0" disables the health server (unit tests only)
+    AX_RUNNER_PORT                     port of the health server, default 80; "0" picks a free port
+    OUSAST_CASE_CACHE                  a directory of git checkouts (``benchmarks/independent`` cache layout)
+    OUSAST_ARTIFACT_URL, OUSAST_RUN    where to deliver, and the run name carried in ``X-Ousast-Run``
+    OUSAST_DELIVERY_BACKOFF            seconds before the second delivery attempt, doubled per retry
+
+The runner exports ``OUSAST_WORKSPACE_DIR`` (the first bound workspace) when the Task's env leaves it unset, and
+``AX_RUNNER_BOUND_PORT`` with the health server's port.
+"""
+
+from __future__ import annotations
+
+import http.client
+import importlib
+import io
+import json
+import logging
+import os
+import shutil
+import subprocess
+import sys
+import tarfile
+import threading
+import time
+import traceback
+import urllib.error
+import urllib.request
+from collections.abc import Mapping, MutableMapping, Sequence
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from openultrasast.plane.manifests import GitSource, ManifestError, Task, Workspace, parse_manifest
+
+__all__ = [
+    "DEFAULT_PORT",
+    "DELIVERY_ATTEMPTS",
+    "EXIT_BY_STATUS",
+    "TASK_PACKAGE",
+    "DeliveryError",
+    "HealthServer",
+    "RunnerError",
+    "build_tar",
+    "deliver",
+    "find_cached_checkout",
+    "load_task",
+    "load_workspaces",
+    "main",
+    "materialise",
+    "read_summary",
+    "run_task_module",
+    "start_health_server",
+    "write_summary",
+]
+
+DEFAULT_PORT = 80
+DELIVERY_ATTEMPTS = 3
+TASK_PACKAGE = "openultrasast.plane.tasks"
+EXIT_BY_STATUS = {"done": 0, "failed": 2, "unfinished": 3}
+SUMMARY = "summary.json"
+
+log = logging.getLogger("ousast.plane.runner")
+
+
+class RunnerError(RuntimeError):
+    """The runner itself cannot proceed (bad manifests, a workspace that cannot be materialised)."""
+
+
+class DeliveryError(RuntimeError):
+    """Every delivery attempt failed; the message names the last cause."""
+
+
+# --- manifests ---------------------------------------------------------------------------------------------------
+
+
+def load_task(text: str) -> Task:
+    """Parse ``AX_TASK_YAML``; raise :class:`RunnerError` unless it is exactly one Task."""
+    try:
+        documents = [d for d in yaml.safe_load_all(text) if d is not None]
+    except yaml.YAMLError as exc:
+        raise RunnerError(f"AX_TASK_YAML is not valid YAML: {exc}") from exc
+    if len(documents) != 1:
+        raise RunnerError(f"AX_TASK_YAML must hold exactly one Task document, found {len(documents)}")
+    try:
+        manifest = parse_manifest(documents[0], source="AX_TASK_YAML")
+    except ManifestError as exc:
+        raise RunnerError(str(exc)) from exc
+    if not isinstance(manifest, Task):
+        raise RunnerError(f"AX_TASK_YAML holds a {type(manifest).__name__}, not a Task")
+    return manifest
+
+
+def load_workspaces(text: str) -> dict[str, Workspace]:
+    """Parse ``AX_WORKSPACES_YAML`` (multi-document, empty allowed) into Workspaces by name."""
+    try:
+        documents = [d for d in yaml.safe_load_all(text) if d is not None]
+    except yaml.YAMLError as exc:
+        raise RunnerError(f"AX_WORKSPACES_YAML is not valid YAML: {exc}") from exc
+    workspaces: dict[str, Workspace] = {}
+    for index, document in enumerate(documents):
+        try:
+            manifest = parse_manifest(document, source=f"AX_WORKSPACES_YAML[{index}]")
+        except ManifestError as exc:
+            raise RunnerError(str(exc)) from exc
+        if not isinstance(manifest, Workspace):
+            raise RunnerError(f"AX_WORKSPACES_YAML[{index}] holds a {type(manifest).__name__}, not a Workspace")
+        workspaces[manifest.metadata.name] = manifest
+    return workspaces
+
+
+# --- health server -----------------------------------------------------------------------------------------------
+
+
+class _HealthHandler(BaseHTTPRequestHandler):
+    server: HealthServer
+
+    def do_GET(self) -> None:  # noqa: N802 - http.server's naming
+        if self.path == "/healthz":
+            self._answer(200, "ok")
+        elif self.path == "/readyz":
+            ready = self.server.ready.is_set()
+            self._answer(200 if ready else 503, "ready" if ready else "workspaces not materialised")
+        else:
+            self._answer(404, "not found")
+
+    def _answer(self, status: int, body: str) -> None:
+        payload = body.encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "text/plain; charset=utf-8")
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def log_message(self, format: str, *args: object) -> None:  # noqa: A002 - http.server's signature
+        log.debug("health server: " + format, *args)
+
+
+class HealthServer(ThreadingHTTPServer):
+    """``/healthz`` always 200; ``/readyz`` 503 until ``ready`` is set. Serves in a daemon thread."""
+
+    daemon_threads = True
+
+    def __init__(self, port: int, ready: threading.Event) -> None:
+        super().__init__(("0.0.0.0", port), _HealthHandler)
+        self.ready = ready
+        self._thread = threading.Thread(target=self.serve_forever, name="ax-health", daemon=True)
+
+    @property
+    def port(self) -> int:
+        return int(self.server_address[1])
+
+    def start(self) -> None:
+        self._thread.start()
+
+    def stop(self) -> None:
+        self.shutdown()
+        self.server_close()
+        self._thread.join(timeout=5)
+
+
+def start_health_server(port: int, ready: threading.Event) -> HealthServer:
+    server = HealthServer(port, ready)
+    server.start()
+    return server
+
+
+# --- workspaces --------------------------------------------------------------------------------------------------
+
+
+def _git(*args: str, cwd: Path | None = None) -> str:
+    done = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, check=False)
+    if done.returncode != 0:
+        raise RunnerError(f"git {' '.join(args)} failed ({done.returncode}): {done.stderr.strip()[:400]}")
+    return done.stdout
+
+
+def _repo_key(url: str) -> str:
+    key = url.strip().lower().rstrip("/")
+    if key.startswith("git@") and ":" in key:
+        host, _, path = key[4:].partition(":")
+        key = f"https://{host}/{path}"
+    key = key.removeprefix("ssh://git@").removeprefix("http://").removeprefix("https://").removeprefix("git://")
+    return key.removesuffix(".git")
+
+
+def find_cached_checkout(cache: Path, repo: str) -> Path | None:
+    """The checkout under ``cache`` (or ``cache`` itself) whose ``origin`` is ``repo``, or None.
+
+    ``benchmarks/independent`` keeps one blob-less clone per case at ``<cache>/<case-id>``; the case id says
+    nothing about the URL, so the match is on ``git remote get-url origin``.
+    """
+    if not cache.is_dir():
+        return None
+    wanted = _repo_key(repo)
+    candidates = [cache, *sorted(p for p in cache.iterdir() if p.is_dir())]
+    for candidate in candidates:
+        if not (candidate / ".git").exists() and not (candidate / "HEAD").exists():
+            continue
+        done = subprocess.run(["git", "-C", str(candidate), "remote", "get-url", "origin"], capture_output=True, text=True, check=False)
+        if done.returncode == 0 and _repo_key(done.stdout) == wanted:
+            return candidate
+    return None
+
+
+def _pinned_ref(source: GitSource) -> str:
+    return source.commit or source.branch or "HEAD"
+
+
+def _export_from_cache(checkout: Path, source: GitSource, dest: Path) -> None:
+    """``git archive`` the pinned ref out of the cache into ``dest`` (no ``.git``), as ``evaluate.export`` does."""
+    ref = source.commit or (f"origin/{source.branch}" if source.branch else "HEAD")
+    done = subprocess.run(["git", "-C", str(checkout), "archive", ref], capture_output=True, check=False)
+    if done.returncode != 0:
+        raise RunnerError(f"git archive {ref} in {checkout} failed: {done.stderr.decode(errors='replace').strip()[:400]}")
+    dest.mkdir(parents=True, exist_ok=True)
+    with tarfile.open(fileobj=io.BytesIO(done.stdout), mode="r:") as archive:
+        if hasattr(tarfile, "data_filter"):  # 3.11.4+ / 3.12; the image is 3.12
+            archive.extractall(dest, filter="data")
+        else:  # pragma: no cover
+            archive.extractall(dest)  # noqa: S202
+    if not any(dest.iterdir()):
+        raise RunnerError(f"{source.repo}@{ref}: the export from {checkout} is empty")
+
+
+def _shallow_clone(source: GitSource, dest: Path) -> None:
+    """Shallow clone at the branch, then make sure the pinned commit is checked out."""
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    args = ["clone", "--quiet", "--depth", "1"]
+    if source.branch:
+        args += ["--branch", source.branch]
+    _git(*args, source.repo, str(dest))
+    if source.commit:
+        present = subprocess.run(["git", "cat-file", "-e", f"{source.commit}^{{commit}}"], cwd=dest, capture_output=True, check=False)
+        if present.returncode != 0:
+            _git("fetch", "--quiet", "--depth", "1", "origin", source.commit, cwd=dest)
+        _git("checkout", "--quiet", "--detach", source.commit, cwd=dest)
+
+
+def _materialise_git(source: GitSource, root: Path, cache: Path | None) -> Path:
+    dest = root / source.name
+    if dest.is_dir() and any(dest.iterdir()):
+        log.info("workspace repo %s already present at %s (resumed volume); keeping it", source.repo, dest)
+        return dest
+    checkout = find_cached_checkout(cache, source.repo) if cache is not None else None
+    if checkout is not None:
+        log.info("exporting %s@%s from cache %s to %s", source.repo, _pinned_ref(source), checkout, dest)
+        _export_from_cache(checkout, source, dest)
+    else:
+        log.info("cloning %s@%s to %s", source.repo, _pinned_ref(source), dest)
+        if dest.exists():
+            shutil.rmtree(dest)
+        _shallow_clone(source, dest)
+    return dest
+
+
+def materialise(task: Task, workspaces: Mapping[str, Workspace], cache: Path | None) -> list[Path]:
+    """Write every bound workspace at its path (``spec.files`` verbatim, ``spec.git`` at ``<path>/<name>``).
+
+    Returns the bound paths in binding order; raises :class:`RunnerError` for a binding without a Workspace.
+    """
+    paths: list[Path] = []
+    for binding in task.workspaces:
+        workspace = workspaces.get(binding.name)
+        if workspace is None:
+            raise RunnerError(f"Task/{task.metadata.name}: workspace {binding.name!r} is bound but not in AX_WORKSPACES_YAML")
+        root = Path(binding.path)
+        root.mkdir(parents=True, exist_ok=True)
+        for entry in workspace.files:
+            target = root / entry.path
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(entry.content, encoding="utf-8")
+        for source in workspace.git:
+            _materialise_git(source, root, cache)
+        paths.append(root)
+    return paths
+
+
+# --- the task ----------------------------------------------------------------------------------------------------
+
+
+def read_summary(output_dir: Path) -> dict[str, Any] | None:
+    path = output_dir / SUMMARY
+    if not path.is_file():
+        return None
+    try:
+        loaded = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    return loaded if isinstance(loaded, dict) else None
+
+
+def write_summary(output_dir: Path, summary: Mapping[str, Any]) -> None:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    (output_dir / SUMMARY).write_text(json.dumps(dict(summary), indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def _failed(reason: str, base: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    summary = dict(base or {})
+    summary.update({"status": "failed", "reason": reason})
+    return summary
+
+
+def run_task_module(command: Sequence[str], output_dir: Path) -> dict[str, Any]:
+    """Import ``openultrasast.plane.tasks.<command[0]>`` and call ``main(command[1:])``; return the summary.
+
+    A crash (any exception, or ``sys.exit`` with a non-zero code) writes ``summary.json`` with ``status: failed``
+    and the traceback; a task that returns without writing ``summary.json`` is failed too.
+    """
+    name = f"{TASK_PACKAGE}.{command[0].replace('-', '_')}"
+    argv = list(command[1:])
+    try:
+        try:
+            module = importlib.import_module(name)
+        except ImportError as exc:
+            raise RunnerError(f"task module {name} cannot be imported: {exc}") from exc
+        entry = getattr(module, "main", None)
+        if not callable(entry):
+            raise RunnerError(f"{name} has no main()")
+        entry(argv)
+    except RunnerError as exc:
+        log.error("%s", exc)
+        summary = _failed(str(exc), read_summary(output_dir))
+        write_summary(output_dir, summary)
+        return summary
+    except SystemExit as exc:
+        if exc.code not in (None, 0):
+            summary = _failed(f"task {command[0]} exited with {exc.code!r}", read_summary(output_dir))
+            write_summary(output_dir, summary)
+            return summary
+    except BaseException:
+        reason = f"task {command[0]} crashed:\n{traceback.format_exc()}"
+        log.error("%s", reason)
+        summary = _failed(reason, read_summary(output_dir))
+        write_summary(output_dir, summary)
+        return summary
+    written = read_summary(output_dir)
+    if written is not None and written.get("status") in EXIT_BY_STATUS:
+        return written
+    found = None if written is None else written.get("status")
+    summary = _failed(f"task {command[0]} wrote no valid {SUMMARY} (status {found!r})", written)
+    write_summary(output_dir, summary)
+    return summary
+
+
+# --- artifact delivery -----------------------------------------------------------------------------------------
+
+
+def build_tar(output_dir: Path) -> bytes:
+    """The output directory as an uncompressed tar with paths relative to it, in sorted order."""
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w") as archive:
+        for path in sorted(p for p in output_dir.rglob("*") if p.is_file()):
+            archive.add(path, arcname=path.relative_to(output_dir).as_posix())
+    return buffer.getvalue()
+
+
+def deliver(url: str, payload: bytes, headers: Mapping[str, str], *, attempts: int = DELIVERY_ATTEMPTS, backoff: float = 1.0) -> int:
+    """POST ``payload`` to ``url`` with ``headers``; retry with doubling backoff; return the 2xx status."""
+    last = "no attempt made"
+    for attempt in range(1, attempts + 1):
+        request = urllib.request.Request(url, data=payload, method="POST", headers=dict(headers))
+        try:
+            with urllib.request.urlopen(request, timeout=60) as response:  # noqa: S310 - the reconciler's URL
+                status = int(response.status)
+                if 200 <= status < 300:
+                    return status
+                last = f"HTTP {status}"
+        except urllib.error.HTTPError as exc:
+            last = f"HTTP {exc.code}"
+        except (urllib.error.URLError, http.client.HTTPException, OSError, ValueError) as exc:
+            last = f"{type(exc).__name__}: {exc}"
+        log.warning("artifact delivery attempt %d/%d to %s failed: %s", attempt, attempts, url, last)
+        if attempt < attempts:
+            time.sleep(backoff * (2 ** (attempt - 1)))
+    raise DeliveryError(f"{last} after {attempts} attempts to {url}")
+
+
+def _deliver_output(task: Task, output_dir: Path, environ: Mapping[str, str]) -> None:
+    url = environ.get("OUSAST_ARTIFACT_URL")
+    if not url:
+        return
+    headers = {"Content-Type": "application/x-tar", "X-Ousast-Task": task.metadata.name}
+    run = environ.get("OUSAST_RUN")
+    if run:
+        headers["X-Ousast-Run"] = run
+    backoff = float(environ.get("OUSAST_DELIVERY_BACKOFF", "1.0"))
+    deliver(url, build_tar(output_dir), headers, backoff=backoff)
+
+
+# --- entrypoint --------------------------------------------------------------------------------------------------
+
+
+@dataclass
+class _Session:
+    task: Task
+    workspaces: dict[str, Workspace]
+    output_dir: Path
+
+
+def _prepare(environ: MutableMapping[str, str]) -> _Session:
+    task_yaml = environ.get("AX_TASK_YAML")
+    if not task_yaml:
+        raise RunnerError("AX_TASK_YAML is not set")
+    task = load_task(task_yaml)
+    workspaces = load_workspaces(environ.get("AX_WORKSPACES_YAML", ""))
+    for var in task.env:
+        environ[var.name] = var.value
+    if "OUSAST_WORKSPACE_DIR" not in environ and task.workspaces:
+        environ["OUSAST_WORKSPACE_DIR"] = task.workspaces[0].path
+    if "OUSAST_OUTPUT_DIR" not in environ:
+        base = environ.get("OUSAST_WORKSPACE_DIR") or os.getcwd()
+        environ["OUSAST_OUTPUT_DIR"] = str(Path(base) / "output")
+    output_dir = Path(environ["OUSAST_OUTPUT_DIR"])
+    output_dir.mkdir(parents=True, exist_ok=True)
+    return _Session(task, workspaces, output_dir)
+
+
+def main(argv: Sequence[str] | None = None, environ: MutableMapping[str, str] | None = None) -> int:
+    """The ``ax-task-runner`` entrypoint; returns the exit code (0 done, 2 failed, 3 unfinished)."""
+    env = os.environ if environ is None else environ
+    if argv:
+        log.warning("ax-task-runner ignores command-line arguments %r; the Task manifest is the only input", list(argv))
+    ready = threading.Event()
+    server: HealthServer | None = None
+    if env.get("AX_RUNNER_HTTP", "1") != "0":
+        server = start_health_server(int(env.get("AX_RUNNER_PORT", str(DEFAULT_PORT))), ready)
+        env["AX_RUNNER_BOUND_PORT"] = str(server.port)
+        log.info("health server on port %d", server.port)
+    session: _Session | None = None
+    try:
+        session = _prepare(env)
+        cache_dir = env.get("OUSAST_CASE_CACHE")
+        materialise(session.task, session.workspaces, Path(cache_dir) if cache_dir else None)
+        ready.set()
+        log.info("workspaces ready; running task %s: %s", session.task.metadata.name, list(session.task.command))
+        summary = run_task_module(session.task.command, session.output_dir)
+        try:
+            _deliver_output(session.task, session.output_dir, env)
+        except DeliveryError as exc:
+            log.error("artifact delivery failed: %s", exc)
+            summary = _failed(f"artifact delivery failed: {exc}", summary)
+            write_summary(session.output_dir, summary)
+        status = str(summary.get("status"))
+        log.info("task %s finished with status %s", session.task.metadata.name, status)
+        return EXIT_BY_STATUS.get(status, EXIT_BY_STATUS["failed"])
+    except RunnerError as exc:
+        log.error("runner failed: %s", exc)
+        if session is not None:
+            write_summary(session.output_dir, _failed(f"runner failed: {exc}", read_summary(session.output_dir)))
+        return EXIT_BY_STATUS["failed"]
+    finally:
+        if server is not None:
+            server.stop()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s", stream=sys.stderr)
+    sys.exit(main(sys.argv[1:]))

tests/plane_stub_task.py

@@ -0,0 +1,39 @@
+"""A stub task module for the runner tests: records its contract, writes one artifact and ``summary.json``.
+
+Registered by the tests as ``openultrasast.plane.tasks.stub_task``; ``STUB_MODE`` picks the outcome
+(``done``, ``failed``, ``unfinished``, ``crash``, ``nosummary``, ``exit``).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.request
+from pathlib import Path
+
+
+def main(argv: list[str]) -> None:
+    out = Path(os.environ["OUSAST_OUTPUT_DIR"])
+    out.mkdir(parents=True, exist_ok=True)
+    mode = os.environ.get("STUB_MODE", "done")
+    workspace = Path(os.environ["OUSAST_WORKSPACE_DIR"])
+    readyz: int | None = None
+    port = os.environ.get("AX_RUNNER_BOUND_PORT")
+    if port:
+        with urllib.request.urlopen(f"http://127.0.0.1:{port}/readyz", timeout=5) as response:
+            readyz = int(response.status)
+    record = {
+        "argv": argv,
+        "env": {k: v for k, v in os.environ.items() if k.startswith("OUSAST_")},
+        "workspace_files": sorted(p.relative_to(workspace).as_posix() for p in workspace.rglob("*") if p.is_file()),
+        "readyz": readyz,
+    }
+    (out / "facts.json").write_text(json.dumps(record, indent=2, sort_keys=True), encoding="utf-8")
+    if mode == "crash":
+        raise RuntimeError("stub exploded")
+    if mode == "exit":
+        raise SystemExit(7)
+    if mode == "nosummary":
+        return
+    summary = {"status": mode, "units_done": 1, "units_total": 1, "usd": 0.0, "calls": 0, "usage": {}}
+    (out / "summary.json").write_text(json.dumps(summary), encoding="utf-8")

tests/test_plane_runner.py

@@ -0,0 +1,364 @@
+"""ax runner contract (ai-service-plane task 3, Requirements 4.1-4.4): env contract, readiness, delivery, exit codes."""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+import os
+import subprocess
+import sys
+import tarfile
+import threading
+import urllib.error
+import urllib.request
+from collections.abc import Iterator
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+import pytest
+
+from openultrasast.plane import runner
+from openultrasast.plane.runner import EXIT_BY_STATUS, DeliveryError, deliver, find_cached_checkout, start_health_server
+
+STUB = Path(__file__).parent / "plane_stub_task.py"
+STUB_MODULE = f"{runner.TASK_PACKAGE}.stub_task"
+
+
+@pytest.fixture(autouse=True)
+def stub_task() -> Iterator[None]:
+    """Register the stub as ``openultrasast.plane.tasks.stub_task`` for the runner's import; no src package needed."""
+    spec = importlib.util.spec_from_file_location(STUB_MODULE, STUB)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[STUB_MODULE] = module
+    spec.loader.exec_module(module)
+    try:
+        yield
+    finally:
+        sys.modules.pop(STUB_MODULE, None)
+
+
+@pytest.fixture
+def env(monkeypatch: pytest.MonkeyPatch) -> Iterator[dict[str, str]]:
+    """A private environment for one runner invocation, restored whole afterwards.
+
+    The runner and the stub read and write ``os.environ`` directly (that is the contract), so monkeypatch alone
+    would not undo what the runner exports; the snapshot does.
+    """
+    saved = dict(os.environ)
+    for key in list(os.environ):
+        if key.startswith(("OUSAST_", "AX_", "STUB_")):
+            monkeypatch.delenv(key)
+    monkeypatch.setenv("AX_RUNNER_HTTP", "0")
+    monkeypatch.setenv("OUSAST_DELIVERY_BACKOFF", "0")
+    try:
+        yield os.environ
+    finally:
+        os.environ.clear()
+        os.environ.update(saved)
+
+
+def task_yaml(output: Path, workspace: Path, *, command: list[str] | None = None, name: str = "stub") -> str:
+    cmd = command or ["stub-task", "--pass", "a"]
+    env_lines = "\n".join(
+        f"    - name: {k}\n      value: {json.dumps(v)}"
+        for k, v in {
+            "OUSAST_OUTPUT_DIR": str(output),
+            "OUSAST_BUDGET_USD": "0.5",
+            "OUSAST_BUDGET_CALLS": "40",
+            "OUSAST_INPUT_CANDIDATES": str(workspace / "candidates.json"),
+        }.items()
+    )
+    return (
+        "apiVersion: ax.io/v1alpha1\nkind: Task\nmetadata:\n  name: " + name + "\nspec:\n  command: " + json.dumps(cmd) + "\n"
+        "  env:\n" + env_lines + "\n  workspaces:\n    - name: case\n      path: " + json.dumps(str(workspace)) + "\n"
+    )
+
+
+def files_workspace_yaml() -> str:
+    return (
+        "apiVersion: ax.io/v1alpha1\nkind: Workspace\nmetadata:\n  name: case\nspec:\n  files:\n"
+        '    - path: candidates.json\n      content: \'{"sinks": ["a"]}\'\n'
+        "    - path: src/app.py\n      content: |\n        def handler():\n            return 1\n"
+    )
+
+
+def run_main(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, workspace_yaml: str, **extra: str) -> tuple[int, Path]:
+    output = tmp_path / "out"
+    workspace = tmp_path / "ws"
+    monkeypatch.setenv("AX_TASK_YAML", task_yaml(output, workspace))
+    monkeypatch.setenv("AX_WORKSPACES_YAML", workspace_yaml)
+    for key, value in extra.items():
+        monkeypatch.setenv(key, value)
+    return runner.main(), output
+
+
+# --- env contract on a files workspace (Requirement 4.1, 4.3) --------------------------------------------------------
+
+
+def test_env_contract_on_a_files_workspace(env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml())
+    assert code == 0
+    facts = json.loads((output / "facts.json").read_text())
+    assert facts["argv"] == ["--pass", "a"], "the command's tail is the task's argv"
+    assert facts["workspace_files"] == ["candidates.json", "src/app.py"], "spec.files written verbatim at the bound path"
+    assert (tmp_path / "ws" / "src" / "app.py").read_text() == "def handler():\n    return 1\n"
+    contract = facts["env"]
+    assert contract["OUSAST_OUTPUT_DIR"] == str(output)
+    assert contract["OUSAST_WORKSPACE_DIR"] == str(tmp_path / "ws"), "set by the runner from the first binding"
+    assert contract["OUSAST_INPUT_CANDIDATES"] == str(tmp_path / "ws" / "candidates.json")
+    assert contract["OUSAST_BUDGET_USD"] == "0.5" and contract["OUSAST_BUDGET_CALLS"] == "40"
+    assert json.loads((output / "summary.json").read_text())["status"] == "done"
+
+
+def test_workspace_dir_from_the_task_env_is_kept(env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("OUSAST_WORKSPACE_DIR", str(tmp_path / "ws"))
+    code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml())
+    assert code == 0
+    assert json.loads((output / "facts.json").read_text())["env"]["OUSAST_WORKSPACE_DIR"] == str(tmp_path / "ws")
+
+
+def test_a_bound_workspace_missing_from_the_stream_fails_before_the_task(
+    env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    code, output = run_main(monkeypatch, tmp_path, "")
+    assert code == EXIT_BY_STATUS["failed"]
+    summary = json.loads((output / "summary.json").read_text())
+    assert summary["status"] == "failed" and "case" in summary["reason"]
+    assert not (output / "facts.json").exists(), "the task never ran"
+
+
+# --- readiness (Requirement 4.2) -----------------------------------------------------------------------------------------
+
+
+def get(url: str) -> int:
+    try:
+        with urllib.request.urlopen(url, timeout=5) as response:
+            return int(response.status)
+    except urllib.error.HTTPError as exc:
+        return exc.code
+
+
+def test_readyz_is_503_until_workspaces_are_ready_then_200() -> None:
+    ready = threading.Event()
+    server = start_health_server(0, ready)
+    try:
+        base = f"http://127.0.0.1:{server.port}"
+        assert get(f"{base}/healthz") == 200
+        assert get(f"{base}/readyz") == 503
+        assert get(f"{base}/other") == 404
+        ready.set()
+        assert get(f"{base}/readyz") == 200
+    finally:
+        server.stop()
+
+
+def test_main_serves_readiness_on_a_free_port_and_the_task_sees_200(
+    env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), AX_RUNNER_HTTP="1", AX_RUNNER_PORT="0")
+    assert code == 0
+    facts = json.loads((output / "facts.json").read_text())
+    assert facts["readyz"] == 200, "workspaces were materialised before the task started"
+    port = os.environ["AX_RUNNER_BOUND_PORT"]
+    with pytest.raises(urllib.error.URLError):
+        urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz", timeout=2)
+
+
+# --- artifact delivery (Requirement 4.4) --------------------------------------------------------------------------------
+
+
+class Receiver(ThreadingHTTPServer):
+    """Records every POST (headers and body) and answers with ``status``."""
+
+    def __init__(self, status: int) -> None:
+        super().__init__(("127.0.0.1", 0), _ReceiverHandler)
+        self.status = status
+        self.posts: list[tuple[dict[str, str], bytes]] = []
+        self.thread = threading.Thread(target=self.serve_forever, daemon=True)
+        self.thread.start()
+
+    @property
+    def url(self) -> str:
+        return f"http://127.0.0.1:{self.server_address[1]}/artifacts"
+
+    def stop(self) -> None:
+        self.shutdown()
+        self.server_close()
+
+
+class _ReceiverHandler(BaseHTTPRequestHandler):
+    server: Receiver
+
+    def do_POST(self) -> None:  # noqa: N802
+        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
+        self.server.posts.append(({k: v for k, v in self.headers.items()}, body))
+        self.send_response(self.server.status)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def log_message(self, format: str, *args: object) -> None:  # noqa: A002
+        pass
+
+
+@pytest.fixture
+def receiver() -> Iterator[Receiver]:
+    server = Receiver(200)
+    try:
+        yield server
+    finally:
+        server.stop()
+
+
+def test_output_dir_is_delivered_as_a_tar_with_the_headers(
+    env: dict[str, str], receiver: Receiver, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=receiver.url, OUSAST_RUN="validation-46")
+    assert code == 0
+    assert len(receiver.posts) == 1
+    headers, body = receiver.posts[0]
+    assert headers["Content-Type"] == "application/x-tar"
+    assert headers["X-Ousast-Task"] == "stub"
+    assert headers["X-Ousast-Run"] == "validation-46"
+    with tarfile.open(fileobj=io.BytesIO(body), mode="r:") as archive:
+        names = sorted(archive.getnames())
+        summary = archive.extractfile("summary.json")
+        assert summary is not None
+        assert json.loads(summary.read())["status"] == "done"
+    assert names == ["facts.json", "summary.json"]
+
+
+def test_run_header_is_absent_without_ousast_run(
+    env: dict[str, str], receiver: Receiver, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    code, _ = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=receiver.url)
+    assert code == 0
+    assert "X-Ousast-Run" not in receiver.posts[0][0]
+
+
+def test_delivery_failure_is_reported_as_failed_with_exit_2(env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    broken = Receiver(500)
+    try:
+        code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=broken.url)
+    finally:
+        broken.stop()
+    assert code == EXIT_BY_STATUS["failed"] == 2
+    assert len(broken.posts) == 3, "three attempts"
+    summary = json.loads((output / "summary.json").read_text())
+    assert summary["status"] == "failed"
+    assert summary["reason"].startswith("artifact delivery failed: HTTP 500")
+    assert summary["units_done"] == 1 and summary["units_total"] == 1, "the task's other fields are kept"
+
+
+def test_deliver_retries_on_connection_errors_then_raises() -> None:
+    closed = Receiver(200)
+    url = closed.url
+    closed.stop()
+    with pytest.raises(DeliveryError, match="3 attempts"):
+        deliver(url, b"x", {"Content-Type": "application/x-tar"}, backoff=0)
+
+
+# --- exit codes (design: error handling) ------------------------------------------------------------------------------------
+
+
+def test_a_crash_yields_failed_with_the_traceback(env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), STUB_MODE="crash")
+    assert code == 2
+    summary = json.loads((output / "summary.json").read_text())
+    assert summary["status"] == "failed"
+    assert "Traceback" in summary["reason"] and "stub exploded" in summary["reason"]
+    assert (output / "facts.json").exists(), "artifacts written before the crash stay for resume"
+
+
+@pytest.mark.parametrize(("mode", "code"), [("done", 0), ("failed", 2), ("unfinished", 3), ("nosummary", 2), ("exit", 2)])
+def test_exit_code_follows_the_summary_status(
+    env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, mode: str, code: int
+) -> None:
+    got, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), STUB_MODE=mode)
+    assert got == code
+    summary = json.loads((output / "summary.json").read_text())
+    assert summary["status"] == (mode if mode in EXIT_BY_STATUS else "failed")
+
+
+def test_an_unknown_task_module_is_failed(env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    output, workspace = tmp_path / "out", tmp_path / "ws"
+    monkeypatch.setenv("AX_TASK_YAML", task_yaml(output, workspace, command=["no-such-task"]))
+    monkeypatch.setenv("AX_WORKSPACES_YAML", files_workspace_yaml())
+    assert runner.main() == 2
+    assert "no_such_task" in json.loads((output / "summary.json").read_text())["reason"]
+
+
+# --- git workspaces from the case cache and by shallow clone ----------------------------------------------------------------
+
+
+def git(*args: str, cwd: Path) -> str:
+    return subprocess.run(
+        ["git", "-c", "user.name=t", "-c", "[email protected]", *args], cwd=cwd, capture_output=True, text=True, check=True
+    ).stdout.strip()
+
+
+@pytest.fixture
+def upstream(tmp_path: Path) -> tuple[Path, str]:
+    """A local repository with two commits; returns it and the first commit (the pin)."""
+    repo = tmp_path / "upstream"
+    repo.mkdir()
+    git("init", "-q", "-b", "main", cwd=repo)
+    (repo / "hello.py").write_text("print('v1')\n")
+    git("add", ".", cwd=repo)
+    git("commit", "-q", "-m", "v1", cwd=repo)
+    pin = git("rev-parse", "HEAD", cwd=repo)
+    (repo / "hello.py").write_text("print('v2')\n")
+    git("commit", "-q", "-am", "v2", cwd=repo)
+    return repo, pin
+
+
+def git_workspace_yaml(repo_url: str, pin: str) -> str:
+    return (
+        "apiVersion: ax.io/v1alpha1\nkind: Workspace\nmetadata:\n  name: case\nspec:\n  git:\n"
+        f"    - name: repo\n      repo: {repo_url}\n      branch: main\n      commit: {pin}\n"
+        "  files:\n    - path: candidates.json\n      content: '{\"sinks\": []}'\n"
+    )
+
+
+def test_git_entry_is_exported_from_the_case_cache_at_the_pin(
+    env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
+) -> None:
+    repo, pin = upstream
+    url = "https://example.invalid/acme/Repo.git"
+    cache = tmp_path / "cache"
+    case = cache / "acme-case-sqli"
+    subprocess.run(["git", "clone", "-q", "--no-checkout", str(repo), str(case)], check=True)
+    git("remote", "set-url", "origin", url, cwd=case)
+    assert find_cached_checkout(cache, "https://example.invalid/acme/repo") == case
+    assert find_cached_checkout(cache, "https://example.invalid/other/repo.git") is None
+    code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(url, pin), OUSAST_CASE_CACHE=str(cache))
+    assert code == 0, (output / "summary.json").read_text()
+    checkout = tmp_path / "ws" / "repo"
+    assert (checkout / "hello.py").read_text() == "print('v1')\n", "the pinned commit, not the branch tip"
+    assert not (checkout / ".git").exists(), "an export, as evaluate.export makes"
+    facts = json.loads((output / "facts.json").read_text())
+    assert facts["workspace_files"] == ["candidates.json", "repo/hello.py"]
+
+
+def test_git_entry_without_a_cache_is_shallow_cloned_at_the_pin(
+    env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
+) -> None:
+    repo, pin = upstream
+    code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri(), pin), OUSAST_CASE_CACHE=str(tmp_path / "no-cache"))
+    assert code == 0, (output / "summary.json").read_text()
+    checkout = tmp_path / "ws" / "repo"
+    assert (checkout / "hello.py").read_text() == "print('v1')\n"
+    assert git("rev-parse", "HEAD", cwd=checkout) == pin
+
+
+def test_a_materialised_repo_survives_a_resume(
+    env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
+) -> None:
+    repo, pin = upstream
+    checkout = tmp_path / "ws" / "repo"
+    checkout.mkdir(parents=True)
+    (checkout / "hello.py").write_text("kept\n")
+    code, _ = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri(), pin))
+    assert code == 0
+    assert (checkout / "hello.py").read_text() == "kept\n", "/workspace persists across suspend and resume; no re-clone"