diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index ba9c7b1..604b1bc 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 127, + "modules": 128, "counts": { - "load_bearing": 118, + "load_bearing": 119, "standalone_capability": 9 }, "orphaned": [], @@ -660,6 +660,7 @@ "importers": [ "cli", "plane.reconciler", + "plane.runner", "plane.tasks.verify", "plane.workspaces" ] @@ -677,7 +678,8 @@ "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.reconciler" + "plane.reconciler", + "plane.runner" ] }, { @@ -688,6 +690,12 @@ "plane.workspaces" ] }, + { + "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": [] + }, { "module": "plane.tasks", "classification": "load_bearing", diff --git a/plane/Dockerfile.runner b/plane/Dockerfile.runner new file mode 100644 index 0000000..45a2d78 --- /dev/null +++ b/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"] diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index e0b034c..8e0d75e 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -63,6 +63,7 @@ SPEC_OWNED = { "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`", + "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 = { diff --git a/src/openultrasast/plane/runner.py b/src/openultrasast/plane/runner.py new file mode 100644 index 0000000..5e88e28 --- /dev/null +++ b/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.`` 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 ``/``; 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 ``/``). + + 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.`` 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:])) diff --git a/tests/plane_stub_task.py b/tests/plane_stub_task.py new file mode 100644 index 0000000..a407706 --- /dev/null +++ b/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") diff --git a/tests/test_plane_runner.py b/tests/test_plane_runner.py new file mode 100644 index 0000000..51fb85a --- /dev/null +++ b/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", "user.email=t@example.invalid", *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"