diff --git a/.kiro/specs/ai-service-plane/design.md b/.kiro/specs/ai-service-plane/design.md index d82e488..1c2f4f5 100644 --- a/.kiro/specs/ai-service-plane/design.md +++ b/.kiro/specs/ai-service-plane/design.md @@ -23,6 +23,7 @@ src/openultrasast/plane/ manifests.py load + validate Task / Workspace / Model / Run (PyYAML, declared dependency) reconciler.py Run execution, status, resume (<= 500 lines, test-enforced) runner.py ax runner contract: AX_TASK_YAML / AX_WORKSPACES_YAML -> entrypoint + doctor.py `ousast plane doctor`: the four checks of Requirement 3.5 budget.py spend and call metering from provider usage fields tasks/repo_facts.py source-only facts artifact tasks/verify.py one hunt per file with facts in the prompt; per-unit resume @@ -61,9 +62,19 @@ container the same module is PID 1 (`ax-task-runner`), built as a target of the Reads `AX_TASK_YAML` and `AX_WORKSPACES_YAML`, materialises each workspace's `git` entries at the pinned commit into the bound path (locally a `git archive` from the case cache, as `evaluate.export` does today), writes -`spec.files`, then executes `spec.command`, whose first element names a task module -(`openultrasast.plane.tasks.`). With `AX_RUNNER_HTTP=1` it serves `/healthz` and `/readyz` (503 until -workspaces are ready) on port 80 in a thread; locally the flag is unset (Requirement 4.2). +`spec.files`, then runs `spec.command` as ax's runner contract requires (ax `docs/runner.md`, +`docs/sandbox.md`): a child process `python -m openultrasast.plane.tasks. ` in its own process group, +with the first workspace as working directory and `AX_METADATA_URL` pointing at the runner's own server. It serves +`/healthz`, `/readyz` (503 until workspaces are ready; ax probes `/readyz?check=workspace`) and ax's two metadata +paths on port 80 (Requirement 4.2). Workspaces are prepared on the first boot only; `/workspace` survives suspend +and resume. After the child exits the runner reads `summary.json` (a crash without one becomes `failed` with the +exit code and stderr tail), delivers the output directory (Requirement 4.4), writes the completion marker +`/workspace/.ousast-state/.done.json` (`exit`, `status`, `delivered`) and then keeps running as PID 1 until +SIGTERM: an exiting runner stops the container, which the first live run showed as every resume timing out. A +boot that finds the marker does not rerun the command (a resumed actor must not repeat billed work); a marker +with a failed delivery retries the delivery only. SIGTERM (ax's stop and suspend) is forwarded to the command's +process group, followed by SIGKILL after `OUSAST_TERM_GRACE` (10 s), and the runner exits 0. +`AX_RUNNER_EXIT_AFTER_COMMAND=1` is a unit-test escape that exits with the task's code after delivery. ### Reconciler (`reconciler.py`, ax-backed) @@ -71,9 +82,15 @@ workspaces are ready) on port 80 in a thread; locally the flag is unset (Require per-task status in `pending | running | suspended | done | failed | unfinished`, timestamps, spend. Execution: topological order; for each ready task the reconciler renders the Task manifest with the run's env (`OUSAST_BUDGET_*`, `OUSAST_INPUT_*`, `OUSAST_OUTPUT_DIR`, `OUSAST_ARTIFACT_URL`), applies it and its -Workspaces through the `ax` CLI (`ax apply`, `ax get`, `ax delete`; the CLI path is injectable so tests use a -fake), and polls ax's status. Artifacts: the reconciler serves a small HTTP receiver on the host (reachable from -kind pods via the node's host gateway); the runner posts the output directory there at exit. Tasks sharing a +Workspaces through the `ax` CLI (`ax apply`, `ax resume`, `ax get`, `ax delete`; the CLI path is injectable so +tests use a fake). ax creates a Task `Suspended`; it runs only after `ax resume task`, and the first resume of a +new image can fail with `DeadlineExceeded` while Agent Substrate builds its golden snapshot (ax then reports +`Failed`), so the reconciler retries a DeadlineExceeded/Unavailable resume with backoff for up to +`OUSAST_RESUME_TIMEOUT` (900 s). ax has no Completed phase and never reports that a command exited: completion is +the artifact delivery. `Failed` after a resume went through, or the task disappearing, before delivery fails the +task with ax's condition message; no delivery within `OUSAST_TASK_TIMEOUT` (7200 s) fails it too; `ax delete task` +follows either way. Artifacts: the reconciler serves a small HTTP receiver on the host (reachable from kind pods +via the docker bridge); the runner posts the output directory there after its command exits. Tasks sharing a `serialize` label never overlap; everything else may run in parallel up to `--workers`. A task's outcome is read from its delivered `summary.json`: `done` only when the task says every unit finished; `unfinished` on a budget stop; `failed` on an account or authentication error, a crash, or a missing delivery. `status` prints @@ -140,12 +157,16 @@ plus the per-candidate cost, turns and the declared-site match (`evaluate.matche - `tests/test_plane_manifests.py`: valid ax manifests load; an unknown field, a bad provider without the annotation, and a Run with a cycle are rejected with the field named. -- `tests/test_plane_reconciler.py`: a fake `ax` CLI (a script recording `apply`/`get`/`delete` calls and - posting scripted `summary.json` deliveries to the receiver) drives a DAG: order, `serialize` non-overlap, - skip-on-rerun, `unfinished` and `failed` propagation, missing delivery, the lock, the attribution table; a +- `tests/test_plane_reconciler.py`: a fake `ax` CLI with ax's real lifecycle (apply -> `Suspended`, a first + resume failing DeadlineExceeded, then `Running` forever, and a scripted `summary.json` delivery only after a + resume went through) drives a DAG: order, `serialize` non-overlap, skip-on-rerun, `unfinished` and `failed` + propagation, resume retry and timeout, `Failed` with a condition message, a vanished task, the task timeout + without delivery, the lock, the attribution table; a test asserts the module is under 500 lines. `tests/test_plane_ax_live.py` (marker `ax`, skipped unless `ousast plane doctor` passes) applies one trivial Task on the real cluster. -- `tests/test_plane_runner.py`: the env contract on a fixture workspace; `/readyz` 503 then 200 with the flag. +- `tests/test_plane_runner.py`: the env contract on a fixture workspace; `/readyz` 503 then 200; the command as a + child in its own group in the first workspace; the runner serving on after the command; SIGTERM reaching the + command's group; the completion marker preventing a rerun and retrying a failed delivery alone. - `tests/test_plane_repo_facts.py`: a three-file tree; callers found across files, none from tests or vendor; identical output on a second run. - `tests/test_plane_verify.py`: a scripted client; the prompt carries the callers; `units.jsonl` resume asks diff --git a/.kiro/specs/ai-service-plane/tasks.md b/.kiro/specs/ai-service-plane/tasks.md index 6d29d70..2a31e93 100644 --- a/.kiro/specs/ai-service-plane/tasks.md +++ b/.kiro/specs/ai-service-plane/tasks.md @@ -33,27 +33,37 @@ mypy are clean, and the commit is gated on pytest's own exit code. Script-level - [ ] 3. Runner contract - `src/openultrasast/plane/runner.py`: read `AX_TASK_YAML` / `AX_WORKSPACES_YAML`; materialise `git` - entries at the pinned commit from the local case cache and write `spec.files`; execute `spec.command` - by importing `openultrasast.plane.tasks.`; `/healthz` `/readyz` server on port 80 (503 until - workspaces are ready; disabled only under `AX_RUNNER_HTTP=0` for unit tests); at exit, POST the output - directory as a tar stream to `OUSAST_ARTIFACT_URL` and report `failed` if delivery fails; exit codes - 0 done, 2 failed, 3 unfinished. `plane/Dockerfile.runner`: this package on Python 3.12, installed to + entries at the pinned commit from the local case cache and write `spec.files` (first boot only); run + `spec.command` as a child `python -m openultrasast.plane.tasks.` in its own process group, cwd the + first workspace, `AX_METADATA_URL` set; `/healthz` `/readyz` and ax's metadata paths on port 80 (503 until + workspaces are ready; disabled only under `AX_RUNNER_HTTP=0` for unit tests); after the command exits, + POST the output directory as a tar stream to `OUSAST_ARTIFACT_URL` (report `failed` if delivery fails), + write `/workspace/.ousast-state/.done.json`, and keep serving until SIGTERM (ax's contract for PID 1); + a boot with the marker never reruns the command and retries only a failed delivery; SIGTERM goes to the + command's group, SIGKILL after `OUSAST_TERM_GRACE`, exit 0. `AX_RUNNER_EXIT_AFTER_COMMAND=1` (tests only) + exits after delivery with 0 done, 2 failed, 3 unfinished. `plane/Dockerfile.runner`: this package on Python 3.12, installed to `/usr/local/bin/ax-task-runner`, loaded into kind by `ops/ax/up.sh`. - `tests/test_plane_runner.py`: the env contract on a fixture workspace; readyz 503 then 200; delivery to a - local receiver; delivery failure yields `failed`. + local receiver; delivery failure yields `failed`; the child's cwd and `AX_METADATA_URL`; the runner stays + up after the command; SIGTERM kills a long-running command's group; the marker prevents a rerun. - _Requirements: 4.1, 4.2, 4.3, 4.4_ - [ ] 4. ax-backed reconciler and CLI - `src/openultrasast/plane/reconciler.py` (under 500 lines, no pipeline logic, no local subprocess - executor): `state.json`, statuses, topological submission through the `ax` CLI (path injectable), status - polling, the artifact HTTP receiver, `serialize` labels never overlapping, `--workers`, budgets and + executor): `state.json`, statuses, topological submission through the `ax` CLI (path injectable), + `ax resume task` after `ax apply` (a created Task is `Suspended`), retrying DeadlineExceeded/Unavailable + with backoff up to `OUSAST_RESUME_TIMEOUT`; completion is the artifact delivery, never an ax phase (ax has + none for a finished command); `Failed` or a vanished task before delivery, or no delivery within + `OUSAST_TASK_TIMEOUT`, fails the task with ax's condition message; the artifact HTTP receiver, `serialize` labels never overlapping, `--workers`, budgets and artifact env passed in the rendered Task manifest, outcome read from the delivered `summary.json`, skip `done` on rerun, PID lock per run, `ax delete` of finished Tasks. - `ousast plane run `, `ousast plane status ` (token attribution table per task and Model from the tasks' `summary.json` usage; `--units` from `units.jsonl`; also written to `attribution.json`), `ousast plane workspaces ` (generates Workspace manifests per case pin). - - `tests/test_plane_reconciler.py`: a fake `ax` CLI drives a DAG: order, non-overlap, skip-on-rerun, - `unfinished` and `failed` propagation, missing delivery, the lock, the attribution table, and the + - `tests/test_plane_reconciler.py`: a fake `ax` CLI with ax's real lifecycle (Suspended, a first resume + failing DeadlineExceeded, Running without a Completed phase) drives a DAG: order, non-overlap, + skip-on-rerun, `unfinished` and `failed` propagation, resume retry and timeout, `Failed` with a condition + message, the task timeout without delivery, the lock, the attribution table, and the line-count assertion. `tests/test_plane_ax_live.py` (marker `ax`, skipped without a live cluster): one trivial Task end to end. - _Requirements: 3.1, 3.2, 3.3, 3.4, 2.2, 7.1, 7.3_ diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 2f6d3a5..87cd0d0 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 128, + "modules": 129, "counts": { - "load_bearing": 119, + "load_bearing": 120, "standalone_capability": 9 }, "orphaned": [], @@ -673,6 +673,14 @@ "plane.tasks.verify" ] }, + { + "module": "plane.doctor", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "plane.reconciler" + ] + }, { "module": "plane.manifests", "classification": "load_bearing", diff --git a/plane/Dockerfile.runner b/plane/Dockerfile.runner index 45a2d78..fcaa893 100644 --- a/plane/Dockerfile.runner +++ b/plane/Dockerfile.runner @@ -4,7 +4,8 @@ # 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. +# `openultrasast.plane.tasks`, run as a child process; the output directory is delivered to the reconciler when +# it exits, and the runner then keeps serving until SIGTERM, as ax requires of PID 1. # # Build from the repository root (the package is the context): # docker build -f plane/Dockerfile.runner -t openultrasast-runner:dev . diff --git a/src/openultrasast/plane/doctor.py b/src/openultrasast/plane/doctor.py new file mode 100644 index 0000000..7e067ea --- /dev/null +++ b/src/openultrasast/plane/doctor.py @@ -0,0 +1,52 @@ +"""``ousast plane doctor``: four checks of this host's ax before a Run is submitted (ai-service-plane Req 3.5). + +A sibling of ``reconciler.py`` (tasks.md task 0 allows either), which re-exports :func:`doctor`. +""" + +from __future__ import annotations + +import os +import re +import subprocess + +UP = "ops/ax/up.sh brings it up" + + +# --- checks ---------------------------------------------------------------------------------------------- + + +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 = [r for r in (line.split() for line in out.splitlines()) if len(r) > 2 and re.fullmatch(r"\d+/\d+", r[1])] + if not rows: # "No resources found" is not a ready namespace + return False, f"no pods in {namespace}" + bad = [r[0] for r in rows if 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__ = ["doctor"] diff --git a/src/openultrasast/plane/reconciler.py b/src/openultrasast/plane/reconciler.py index cf13e94..cee037d 100644 --- a/src/openultrasast/plane/reconciler.py +++ b/src/openultrasast/plane/reconciler.py @@ -1,23 +1,22 @@ -"""ax-backed reconciler of a ``Run``: submission, status, resume, artifact receipt (ai-service-plane Req 3, 7). +"""ax-backed reconciler of a ``Run``: submission, resume, completion by delivery (ai-service-plane Req 3, 4.4, 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. +``files`` entries, and the bound Model) and runs ``ax apply -f``. ax creates a Task ``Suspended``; the reconciler +resumes it (``ax resume task``), retrying with backoff while ax answers DeadlineExceeded/Unavailable (the first +resume of a new image waits for its golden snapshot) for up to ``OUSAST_RESUME_TIMEOUT`` s (default 900). ax has no +Completed phase and never reports that a command exited; completion is the runner's artifact delivery, and +``Failed``, the task vanishing, or ``OUSAST_TASK_TIMEOUT`` s (default 7200) without delivery fail the task with +ax's condition message. ``ax delete task`` follows either way. The outcome is the delivered ``summary.json``. -ax's ObjectMeta is ``name``/``atespace`` and its API server rejects unknown fields, so rendered documents carry only -ax's fields; ``openultrasast.io/*`` annotations are stripped and reach the task as env. Model binding: a Task names -its Model in ``openultrasast.io/model``; the task receives ``OUSAST_MODEL`` and the Model's ``spec.parameters`` as -``OUSAST_MODEL_PARAMS`` JSON (Req 7.1). Commit pins: each bound Workspace's ``openultrasast.io/git-commits`` becomes -``OUSAST_GIT_PINS`` = ``{"/": ""}``. Inputs: ``OUSAST_INPUT_`` points into a -generated Workspace bound at ``/workspace/.ousast-in/`` (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//`` (``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. +ax's API server rejects unknown fields, so rendered documents carry only ax's fields; ``openultrasast.io/*`` +annotations reach the task as env: ``openultrasast.io/model`` as ``OUSAST_MODEL`` plus ``OUSAST_MODEL_PARAMS`` (Req +7.1), each Workspace's ``openultrasast.io/git-commits`` as ``OUSAST_GIT_PINS``. ``OUSAST_INPUT_`` points into +a generated Workspace at ``/workspace/.ousast-in/``. The receiver takes the runner's tar (``X-Ousast-Run``, +``X-Ousast-Task``) on ``OUSAST_ARTIFACT_PORT`` and advertises ``OUSAST_ARTIFACT_HOST`` (default ``172.17.0.1``, the +docker bridge kind nodes reach). Run state: ``~/ousast-results/plane//`` (``OUSAST_RESULTS``), polls every +``OUSAST_POLL_SECONDS`` (3). """ from __future__ import annotations @@ -31,7 +30,7 @@ import subprocess import tarfile import threading import time -from collections.abc import Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -40,15 +39,14 @@ from typing import Any import yaml +from .doctor import doctor 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" +TRANSIENT = re.compile(r"deadline ?exceeded|unavailable", re.IGNORECASE) # the golden-snapshot build outlives a resume _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"} @@ -83,10 +81,6 @@ 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] @@ -177,7 +171,7 @@ def render_task(run: Run, entry: RunTask, manifests: Manifests, base: Path, arti class Ax: - """The three ``ax`` verbs the reconciler uses; the executable path is injectable so tests use a fake.""" + """The four ``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 @@ -185,24 +179,38 @@ class Ax: def _run(self, *args: str) -> subprocess.CompletedProcess[str]: return subprocess.run([self.executable, *args], capture_output=True, text=True, check=False) + def _error(self, proc: subprocess.CompletedProcess[str]) -> str: + return proc.stderr.strip() or proc.stdout.strip() or f"exit {proc.returncode}" + 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()}") + raise RuntimeError(f"ax apply failed ({proc.returncode}): {self._error(proc)}") - def phase(self, name: str) -> str: - """``status.phase`` from ``ax get task `` (YAML; Pending when unset; ``Unknown: ...`` when ax fails).""" + def resume(self, name: str) -> str | None: + """``ax resume task ``: None once ax accepted it, else ax's error text.""" + proc = self._run("resume", "task", name) + return None if proc.returncode == 0 else self._error(proc) + + def phase(self, name: str) -> tuple[str, str]: + """(``status.phase``, the Ready condition's message) from ``ax get task`` YAML; ``Pending`` while unset, + ``Gone`` when ax no longer knows the task, ``Unknown`` when ax itself fails.""" proc = self._run("get", "task", name) if proc.returncode != 0: - return "Unknown: " + (proc.stderr.strip() or proc.stdout.strip()) + return ("Gone" if re.search(r"not ?found", self._error(proc), re.IGNORECASE) else "Unknown"), self._error(proc) try: doc = yaml.safe_load(proc.stdout) except yaml.YAMLError: - return "Unknown: unreadable ax output" + 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" + if not isinstance(status, dict): + return "Pending", "" + conditions = [c for c in status.get("conditions") or [] if isinstance(c, dict)] + ready = [c for c in conditions if c.get("type") == "Ready"] or conditions + return str(status.get("phase") or "Pending"), str(ready[-1].get("message") or "") if ready else "" def delete(self, name: str) -> None: + """``ax delete task ``; ax returns once the actor is torn down.""" self._run("delete", "task", name) @@ -274,10 +282,8 @@ class _State: 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}" +def outcome_of(summary: Mapping[str, Any] | None) -> tuple[str, str]: + """(status, reason) of a delivered task from its ``summary.json``: done only when every unit is.""" if summary is None: return "failed", "no summary.json delivered" if summary.get("status") == "done" and summary.get("units_done") == summary.get("units_total"): @@ -301,6 +307,47 @@ def _acquire_lock(base: Path) -> Path: return lock +def _await(ax: Ax, name: str, delivered: Callable[[], bool], poll: float) -> str | None: + """Resume ``name`` and wait for its delivery: None once delivered, else why the task failed. + + ax sets ``Failed`` when a resume call fails, so ``Failed`` is final only once a resume went through; until then + a DeadlineExceeded/Unavailable resume is retried with doubling backoff inside ``OUSAST_RESUME_TIMEOUT``. + """ + task_limit = float(os.environ.get("OUSAST_TASK_TIMEOUT") or 7200) + resume_limit = float(os.environ.get("OUSAST_RESUME_TIMEOUT") or 900) + started = time.monotonic() + resuming: float | None = started # when the current resume spell began; None while ax runs the task + retry_at, backoff, last, unknown, phase = started, poll, "", 0, "Suspended" + while not delivered(): + now = time.monotonic() + if now - started > task_limit: + return f"no delivery within {task_limit:g}s (OUSAST_TASK_TIMEOUT); ax phase {phase}" + if resuming is not None and now >= retry_at: + if now - resuming > resume_limit: + return f"ax resume did not succeed within {resume_limit:g}s (OUSAST_RESUME_TIMEOUT): {last}" + error = ax.resume(name) + if error is None: + resuming = None + elif TRANSIENT.search(error): + last, retry_at, backoff = error, now + backoff, min(backoff * 2, 60.0) + else: + return f"ax resume failed: {error}" + time.sleep(poll) + if delivered(): + break + phase, message = ax.phase(name) + unknown = unknown + 1 if phase == "Unknown" else 0 + if phase == "Gone": + return f"the task disappeared from ax before delivery: {message}" + if unknown >= 10: # ten consecutive unreadable polls: ax itself is broken, not the task + return f"ax get task keeps failing: {message}" + if resuming is None and phase == "Failed": + return f"ax phase Failed: {message}" + if resuming is None and phase == "Suspended": + resuming, retry_at, backoff = time.monotonic(), 0.0, poll + return None + + 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" @@ -311,21 +358,11 @@ def _execute(ax: Ax, run: Run, entry: RunTask, manifests: Manifests, base: Path, 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) + failure = _await(ax, ax_name, lambda: entry.name in receiver.delivered, poll) finally: ax.delete(ax_name) - return outcome_of(_read_json(base / entry.name / "summary.json"), phase) + return ("failed", failure) if failure else outcome_of(_read_json(base / entry.name / "summary.json")) def run(run_manifest: Path, *, workers: int = 1, ax: str = "ax", results_root: Path | None = None) -> str: @@ -458,41 +495,4 @@ def status(run_name: str, *, units: bool = False, results_root: Path | None = No 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 = [r for r in (line.split() for line in out.splitlines()) if len(r) > 2 and re.fullmatch(r"\d+/\d+", r[1])] - if not rows: # "No resources found" is not a ready namespace - return False, f"no pods in {namespace}" - bad = [r[0] for r in rows if 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"] diff --git a/src/openultrasast/plane/runner.py b/src/openultrasast/plane/runner.py index 7707f3d..f3c47c5 100644 --- a/src/openultrasast/plane/runner.py +++ b/src/openultrasast/plane/runner.py @@ -2,46 +2,59 @@ 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. +nothing else comes from the command line (Requirement 4.1). It serves ``/healthz``, ``/readyz`` and ax's two +metadata paths on port 80, ``/readyz`` answering 503 until every workspace is materialised (Requirement 4.2). +Workspaces are prepared on the first boot only: ``/workspace`` survives suspend and resume, the process tree does +not (ax docs/runner.md). -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``. +The command runs as a child process, ``python -m openultrasast.plane.tasks. ``, +in its own process group, with the first workspace as its working directory and ``AX_METADATA_URL`` pointing at +this server. ax has no artifact channel and never reports that a command finished, so after the child exits the +runner reads ``summary.json`` (a crash without one becomes ``failed`` with the exit code and the stderr tail), +posts the whole ``OUSAST_OUTPUT_DIR`` as a tar to ``OUSAST_ARTIFACT_URL`` -- that delivery is the task's +completion (Requirement 4.4); a failed one is reported ``failed``, never done -- and writes the completion marker +``/.done.json`` (``exit``, ``status``, ``delivered``). Then it keeps serving until SIGTERM, +as ax requires of PID 1. A boot that finds the marker never reruns the command (a resumed actor must not repeat +billed work); when the marker says the delivery failed, it retries the delivery alone. SIGTERM (ax's stop and +suspend) goes to the command's process group; after ``OUSAST_TERM_GRACE`` seconds the rest is killed and the +runner exits 0. 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 + AX_RUNNER_EXIT_AFTER_COMMAND "1" exits after delivery with 0 done, 2 failed, 3 unfinished (unit tests only) + AX_RUNNER_TASK_PACKAGE package the command's module is looked up in (unit tests only) + OUSAST_STATE_DIR completion markers, default ``/workspace/.ousast-state`` (the durable volume) + OUSAST_TERM_GRACE seconds between SIGTERM and SIGKILL of the command's group, default 10 OUSAST_CASE_CACHE a directory of git checkouts (``benchmarks/independent`` cache layout) OUSAST_GIT_PINS JSON ``{"/": "<40-hex commit>"}`` from the reconciler 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. +The runner exports ``OUSAST_WORKSPACE_DIR`` (the first bound workspace) when the Task's env leaves it unset, +``AX_RUNNER_BOUND_PORT`` with the health server's port and ``AX_METADATA_URL``. """ from __future__ import annotations +import contextlib import http.client -import importlib import io import json import logging import os import re +import signal import subprocess import sys import tarfile import threading import time -import traceback import urllib.error import urllib.request +from collections import deque from collections.abc import Mapping, MutableMapping, Sequence from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -56,7 +69,9 @@ __all__ = [ "DEFAULT_PORT", "DELIVERY_ATTEMPTS", "EXIT_BY_STATUS", + "STATE_DIR", "TASK_PACKAGE", + "Child", "DeliveryError", "HealthServer", "RunnerError", @@ -70,19 +85,26 @@ __all__ = [ "main", "materialise", "read_summary", - "run_task_module", + "run_command", "start_health_server", + "summary_after", + "task_module", "write_summary", ] DEFAULT_PORT = 80 DELIVERY_ATTEMPTS = 3 TASK_PACKAGE = "openultrasast.plane.tasks" +STATE_DIR = "/workspace/.ousast-state" +TERM_GRACE = 10.0 # ax's own runner: SIGTERM, ten seconds, SIGKILL +STDERR_TAIL_LINES = 40 EXIT_BY_STATUS = {"done": 0, "failed": 2, "unfinished": 3} SUMMARY = "summary.json" +METADATA_PATHS = {"/metadata/v1alpha1/ax/task": "task", "/metadata/v1alpha1/ax/workspaces": "workspaces"} DEFAULT_BRANCH = "main" # ax internal/workspace/setup.go: defaultBranch _PLACEHOLDER_NAMES = ("", "repo", "origin") # setup.go: names that do not name a checkout directory _SHA = re.compile(r"[0-9a-f]{40}") +_MODULE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") log = logging.getLogger("ousast.plane.runner") @@ -140,18 +162,21 @@ class _HealthHandler(BaseHTTPRequestHandler): server: HealthServer def do_GET(self) -> None: # noqa: N802 - http.server's naming - if self.path == "/healthz": + path = self.path.split("?", 1)[0] # ax's controller probes /readyz?check=workspace + if path == "/healthz": self._answer(200, "ok") - elif self.path == "/readyz": + elif path == "/readyz": ready = self.server.ready.is_set() self._answer(200 if ready else 503, "ready" if ready else "workspaces not materialised") + elif path in METADATA_PATHS: + self._answer(200, self.server.metadata.get(METADATA_PATHS[path], ""), "application/yaml") else: self._answer(404, "not found") - def _answer(self, status: int, body: str) -> None: + def _answer(self, status: int, body: str, content_type: str = "text/plain; charset=utf-8") -> None: payload = body.encode("utf-8") self.send_response(status) - self.send_header("Content-Type", "text/plain; charset=utf-8") + self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) @@ -161,13 +186,15 @@ class _HealthHandler(BaseHTTPRequestHandler): class HealthServer(ThreadingHTTPServer): - """``/healthz`` always 200; ``/readyz`` 503 until ``ready`` is set. Serves in a daemon thread.""" + """``/healthz`` always 200; ``/readyz`` 503 until ``ready`` is set; the Task and Workspaces YAML under + ``/metadata/v1alpha1/ax/``. Serves in a daemon thread.""" daemon_threads = True - def __init__(self, port: int, ready: threading.Event) -> None: + def __init__(self, port: int, ready: threading.Event, metadata: Mapping[str, str] | None = None) -> None: super().__init__(("0.0.0.0", port), _HealthHandler) self.ready = ready + self.metadata = dict(metadata or {}) self._thread = threading.Thread(target=self.serve_forever, name="ax-health", daemon=True) @property @@ -183,8 +210,9 @@ class HealthServer(ThreadingHTTPServer): self._thread.join(timeout=5) -def start_health_server(port: int, ready: threading.Event) -> HealthServer: - server = HealthServer(port, ready) +def start_health_server(port: int, ready: threading.Event, metadata: Mapping[str, str] | None = None) -> HealthServer: + """Serve ``/healthz``, ``/readyz`` and ax's metadata paths (``metadata`` keyed ``task``/``workspaces``).""" + server = HealthServer(port, ready, metadata) server.start() return server @@ -370,48 +398,93 @@ def _failed(reason: str, base: Mapping[str, Any] | None = None) -> dict[str, Any 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. +def task_module(command: Sequence[str], package: str = TASK_PACKAGE) -> str: + """``spec.command[0]`` as a module under ``package`` (``repo-facts`` -> ``.repo_facts``).""" + if not command: + raise RunnerError("spec.command is empty") + name = command[0].replace("-", "_") + if not _MODULE.fullmatch(name): + raise RunnerError(f"spec.command[0] {command[0]!r} does not name a task module") + return f"{package}.{name}" - 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. + +class Child: + """``spec.command`` as a child in its own process group; its stderr is echoed and the tail kept.""" + + def __init__(self, argv: Sequence[str], cwd: str | None, env: Mapping[str, str]) -> None: + self.proc = subprocess.Popen(list(argv), cwd=cwd, env=dict(env), stderr=subprocess.PIPE, start_new_session=True) + self.tail: deque[str] = deque(maxlen=STDERR_TAIL_LINES) + self._reader = threading.Thread(target=self._echo, name="task-stderr", daemon=True) + self._reader.start() + + def _echo(self) -> None: + assert self.proc.stderr is not None + for raw in self.proc.stderr: + line = raw.decode("utf-8", errors="replace") + sys.stderr.write(line) + sys.stderr.flush() + self.tail.append(line.rstrip("\n")) + + def signal_group(self, signum: int) -> None: + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(self.proc.pid, signum) + + def wait(self, stop: threading.Event, grace: float) -> int | None: + """The exit code; or, once ``stop`` is set, SIGTERM to the group, ``grace`` seconds, SIGKILL, and None.""" + while self.proc.poll() is None: + if stop.wait(0.1): + log.info("stopping: SIGTERM to the command's process group %d", self.proc.pid) + self.signal_group(signal.SIGTERM) + try: + self.proc.wait(timeout=grace) + except subprocess.TimeoutExpired: + log.warning("the command outlived the %gs grace period; SIGKILL", grace) + self.signal_group(signal.SIGKILL) + self.proc.wait() + self._reader.join(timeout=5) + return None + self._reader.join(timeout=5) + return self.proc.returncode + + +def summary_after(code: int, output_dir: Path, name: str, stderr_tail: Sequence[str] = ()) -> dict[str, Any]: + """The task's ``summary.json`` after it exited with ``code``; a failed one when it wrote none that is valid. + + A task exits 2 or 3 with a ``failed``/``unfinished`` summary by design; a non-zero exit next to ``done``, or no + valid summary at all, is a crash: the failed summary keeps the task's fields and carries the exit code and the + tail of its stderr (the traceback of an uncaught exception). """ - 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) + status = None if written is None else written.get("status") + if status in EXIT_BY_STATUS and (code == 0 or status != "done"): + return dict(written or {}) + why = "reported done" if status == "done" else f"wrote no valid {SUMMARY} (status {status!r})" + reason = f"task {name} exited with {code} and {why}" + if stderr_tail: + reason += "; stderr tail:\n" + "\n".join(stderr_tail) + summary = _failed(reason, written) + summary["exit"] = code write_summary(output_dir, summary) return summary +def run_command( + task: Task, output_dir: Path, env: Mapping[str, str], stop: threading.Event, grace: float +) -> tuple[int | None, dict[str, Any]]: + """Run ``python -m . `` in the first workspace; (exit code, summary). + + (None, {}) when ``stop`` interrupted it: the command did not finish and a resume runs it again. + """ + module = task_module(task.command, env.get("AX_RUNNER_TASK_PACKAGE") or TASK_PACKAGE) + cwd = task.workspaces[0].path if task.workspaces else None + child = Child([sys.executable, "-m", module, *task.command[1:]], cwd, env) + code = child.wait(stop, grace) + if code is None: + return None, {} + log.info("task %s exited with %d", task.metadata.name, code) + return code, summary_after(code, output_dir, task.command[0], list(child.tail)) + + # --- artifact delivery ----------------------------------------------------------------------------------------- @@ -445,16 +518,18 @@ def deliver(url: str, payload: bytes, headers: Mapping[str, str], *, attempts: i raise DeliveryError(f"{last} after {attempts} attempts to {url}") -def _deliver_output(task: Task, output_dir: Path, environ: Mapping[str, str]) -> None: +def _deliver_output(task: Task, output_dir: Path, environ: Mapping[str, str]) -> bool: + """POST the output directory; False when there is nowhere to deliver; :class:`DeliveryError` when it fails.""" url = environ.get("OUSAST_ARTIFACT_URL") if not url: - return + return False 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) + return True # --- entrypoint -------------------------------------------------------------------------------------------------- @@ -465,6 +540,16 @@ class _Session: task: Task workspaces: dict[str, Workspace] output_dir: Path + state_dir: Path + + @property + def marker(self) -> Path: + """``/.done.json``: the command finished; ``delivered`` says whether the receiver has it.""" + return self.state_dir / f"{self.task.metadata.name}.done.json" + + @property + def prepared(self) -> Path: + return self.state_dir / f"{self.task.metadata.name}.workspaces.done" def _prepare(environ: MutableMapping[str, str]) -> _Session: @@ -481,49 +566,126 @@ def _prepare(environ: MutableMapping[str, str]) -> _Session: 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) + state_dir = Path(environ.get("OUSAST_STATE_DIR") or STATE_DIR) + try: + output_dir.mkdir(parents=True, exist_ok=True) + state_dir.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise RunnerError(f"cannot create the output or state directory: {exc}") from exc + return _Session(task, workspaces, output_dir, state_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) +def _exit_of(marker: Mapping[str, Any]) -> int: + return EXIT_BY_STATUS.get(str(marker.get("status")), EXIT_BY_STATUS["failed"]) + + +def _finish(session: _Session, env: Mapping[str, str], exit_code: int | None, summary: Mapping[str, Any]) -> dict[str, Any]: + """Deliver the output directory and write the completion marker; a failed delivery reports ``failed``. + + The marker keeps the task's own summary so a later boot can put it back and retry the delivery alone. + """ + marker: dict[str, Any] = {"exit": exit_code, "status": str(summary.get("status")), "delivered": False, "summary": dict(summary)} + try: + marker["delivered"] = _deliver_output(session.task, session.output_dir, env) + except DeliveryError as exc: + log.error("artifact delivery failed: %s", exc) + marker.update(status="failed", reason=f"artifact delivery failed: {exc}") + write_summary(session.output_dir, _failed(marker["reason"], summary)) + session.marker.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n", encoding="utf-8") + log.info("task %s: status %s, delivered %s", session.task.metadata.name, marker["status"], marker["delivered"]) + return marker + + +def _run_once(env: MutableMapping[str, str], ready: threading.Event, stop: threading.Event) -> int: + """Materialise, run the command once, deliver; on a resumed volume, only what the marker says is left.""" session: _Session | None = None try: session = _prepare(env) - cache_dir = env.get("OUSAST_CASE_CACHE") - pins = load_pins(env.get("OUSAST_GIT_PINS", "")) - materialise(session.task, session.workspaces, Path(cache_dir) if cache_dir else None, pins) + if session.prepared.exists(): + log.info("workspaces were prepared on an earlier boot; keeping them") + else: + cache_dir = env.get("OUSAST_CASE_CACHE") + materialise(session.task, session.workspaces, Path(cache_dir) if cache_dir else None, load_pins(env.get("OUSAST_GIT_PINS", ""))) + session.prepared.touch() ready.set() + done = _read_marker(session.marker) + if done is not None and done.get("delivered"): + log.info("task %s already ran and was delivered (%s); not running it again", session.task.metadata.name, done.get("status")) + return _exit_of(done) + if done is not None: + log.info("task %s already ran; retrying the delivery only", session.task.metadata.name) + stored = done.get("summary") + kept: dict[str, Any] = stored if isinstance(stored, dict) else read_summary(session.output_dir) or {} + write_summary(session.output_dir, kept) + return _exit_of(_finish(session, env, done.get("exit"), kept)) + if stop.is_set(): + return 0 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"]) + code, summary = run_command(session.task, session.output_dir, env, stop, float(env.get("OUSAST_TERM_GRACE") or TERM_GRACE)) + if code is None: + log.info("task %s was stopped before its command finished; a resume runs it again", session.task.metadata.name) + return 0 + return _exit_of(_finish(session, env, code, summary)) 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"] + if session is None: + return EXIT_BY_STATUS["failed"] + summary = _failed(f"runner failed: {exc}", read_summary(session.output_dir)) + write_summary(session.output_dir, summary) + return _exit_of(_finish(session, env, None, summary)) + + +def _read_marker(path: Path) -> dict[str, Any] | 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 main(argv: Sequence[str] | None = None, environ: MutableMapping[str, str] | None = None) -> int: + """The ``ax-task-runner`` entrypoint (PID 1): run the command once, deliver, then serve until SIGTERM. + + Returns 0 after SIGTERM. ``AX_RUNNER_EXIT_AFTER_COMMAND=1`` (unit tests only; the image never sets it) returns + right after delivery with the task's 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, stop = threading.Event(), threading.Event() + previous = _on_sigterm(stop) + server: HealthServer | None = None + if env.get("AX_RUNNER_HTTP", "1") != "0": + metadata = {"task": env.get("AX_TASK_YAML", ""), "workspaces": env.get("AX_WORKSPACES_YAML", "")} + server = start_health_server(int(env.get("AX_RUNNER_PORT", str(DEFAULT_PORT))), ready, metadata) + env["AX_RUNNER_BOUND_PORT"] = str(server.port) + env["AX_METADATA_URL"] = f"http://127.0.0.1:{server.port}" + log.info("health and metadata server on port %d", server.port) + try: + code = _run_once(env, ready, stop) + if stop.is_set(): + return 0 + if env.get("AX_RUNNER_EXIT_AFTER_COMMAND") == "1": + return code + log.info("the command is finished (exit %d by status); serving until SIGTERM", code) + while not stop.wait(1.0): + pass + log.info("SIGTERM: shutting down") + return 0 finally: + if previous is not None: + signal.signal(signal.SIGTERM, previous) if server is not None: server.stop() +def _on_sigterm(stop: threading.Event) -> Any: + """Route SIGTERM (ax's stop and suspend) to ``stop``; the previous handler, or None off the main thread.""" + if threading.current_thread() is not threading.main_thread(): + return None + return signal.signal(signal.SIGTERM, lambda signum, frame: stop.set()) + + 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 index a407706..ee26d09 100644 --- a/tests/plane_stub_task.py +++ b/tests/plane_stub_task.py @@ -1,34 +1,57 @@ """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``). +The tests copy it into a throwaway package (``AX_RUNNER_TASK_PACKAGE``) so the runner starts it the way it starts +a real task, ``python -m .stub_task``. ``STUB_MODE`` picks the outcome (``done``, ``failed``, +``unfinished``, ``crash``, ``nosummary``, ``exit``, ``sleep``); ``STUB_COUNTER`` names a file that gets one line +per run. """ from __future__ import annotations import json import os +import signal +import subprocess +import sys +import time import urllib.request from pathlib import Path +def _get(url: str) -> tuple[int, str]: + with urllib.request.urlopen(url, timeout=5) as response: + return int(response.status), response.read().decode("utf-8") + + 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") + counter = os.environ.get("STUB_COUNTER") + if counter: + with open(counter, "a", encoding="utf-8") as handle: + handle.write(f"{os.getpid()}\n") 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) + readyz = _get(f"http://127.0.0.1:{port}/readyz")[0] if port else None + metadata = os.environ.get("AX_METADATA_URL") 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, + "cwd": os.getcwd(), + "pid": os.getpid(), + "pgid": os.getpgid(0), + "metadata_url": metadata, + "metadata_task": _get(f"{metadata}/metadata/v1alpha1/ax/task")[1] if metadata else None, } (out / "facts.json").write_text(json.dumps(record, indent=2, sort_keys=True), encoding="utf-8") + if mode == "sleep": # ignore SIGTERM so only the runner's SIGKILL ends the group; the grandchild inherits it + signal.signal(signal.SIGTERM, signal.SIG_IGN) + grandchild = subprocess.Popen(["sleep", "60"]) + (out / "pids.json").write_text(json.dumps([os.getpid(), grandchild.pid]), encoding="utf-8") + time.sleep(60) if mode == "crash": raise RuntimeError("stub exploded") if mode == "exit": @@ -37,3 +60,7 @@ def main(argv: list[str]) -> None: 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") + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/tests/test_plane_ax_live.py b/tests/test_plane_ax_live.py index 6fc368e..636cbcc 100644 --- a/tests/test_plane_ax_live.py +++ b/tests/test_plane_ax_live.py @@ -8,12 +8,13 @@ passes, so the host suite never depends on cluster state. It applies the digest- from __future__ import annotations import os +import subprocess import time from pathlib import Path import pytest -from openultrasast.plane.reconciler import Ax, doctor +from openultrasast.plane.reconciler import TRANSIENT, Ax, doctor pytestmark = pytest.mark.ax SMOKE = Path.home() / ".cache/ousast/ax-src/smoke-task.yaml" @@ -32,16 +33,34 @@ def live_ax() -> Ax: return Ax() -def test_smoke_task_completes_on_the_cluster(live_ax: Ax) -> None: +def resume_until_accepted(ax: Ax, name: str, deadline: float) -> None: + """ax creates a Task Suspended; the first resume of a new image may time out while its snapshot is built.""" + error: str | None = "not tried" + while error is not None and time.monotonic() < deadline: + error = ax.resume(name) + if error is not None and not TRANSIENT.search(error): + raise AssertionError(f"ax resume task {name} failed: {error}") + if error is not None: + time.sleep(10) + assert error is None, f"ax resume task {name} never went through: {error}" + + +def test_smoke_task_runs_and_its_runner_survives_suspend_and_resume(live_ax: Ax) -> None: + """ax has no Completed phase: the smoke task must reach Running after resume, and its runner must stay up after + the command so that a suspend and a second resume still go through (an exiting runner made resumes time out).""" + name = "ousast-smoke" + live_ax.delete(name) live_ax.apply(SMOKE) - deadline = time.monotonic() + 600 - phase = "Pending" + deadline = time.monotonic() + 900 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) + resume_until_accepted(live_ax, name, deadline) + phase, message = live_ax.phase(name) + assert phase == "Running", f"after resume: {phase} ({message})" + time.sleep(30) # the model-free repo-facts command is finished by now + suspended = subprocess.run(["ax", "suspend", "task", name], capture_output=True, text=True, check=False) + assert suspended.returncode == 0, suspended.stderr + resume_until_accepted(live_ax, name, deadline) + phase, message = live_ax.phase(name) + assert phase == "Running", f"after the second resume: {phase} ({message})" finally: - live_ax.delete("ousast-smoke") - assert phase.lower() in ("completed", "succeeded"), f"smoke task ended in phase {phase!r}" + live_ax.delete(name) diff --git a/tests/test_plane_reconciler.py b/tests/test_plane_reconciler.py index 78a1da7..7229bdc 100644 --- a/tests/test_plane_reconciler.py +++ b/tests/test_plane_reconciler.py @@ -1,9 +1,11 @@ """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. +The fake is a Python script the tests write into ``tmp_path`` that follows ax's real task lifecycle (ax source: +internal/controller/reconciler.go, internal/server/server.go): ``apply`` creates the task ``Suspended``; ``resume`` +fails with DeadlineExceeded the first time (ax then reports ``Failed``, as while Agent Substrate builds a new +image's golden snapshot) and succeeds after; ``get task`` answers ``Running`` forever -- ax has no Completed phase +-- and on the scripted poll it POSTs a scripted tar to the ``OUSAST_ARTIFACT_URL`` in the applied Task's env, or +turns ``Failed`` with a condition message, or vanishes. Every outcome comes from a scripted delivery. """ from __future__ import annotations @@ -60,7 +62,8 @@ def non_ax_fields(value: object, schema: dict | None, path: str) -> list[str]: FAKE_AX = '''#!{python} -"""Fake ax CLI for the reconciler tests: argv log, scripted phases, scripted deliveries.""" +"""Fake ax CLI with ax's real task lifecycle: apply -> Suspended; resume (the first ones fail DeadlineExceeded and +leave the task Failed, as ax's server does) -> Running; never Completed; the scripted delivery after N polls.""" import io, json, sys, tarfile, time, urllib.error, urllib.request from pathlib import Path import yaml @@ -70,42 +73,73 @@ 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) +record = HERE / "applied" / ((argv[2] if len(argv) > 2 else "-") + ".json") +rec = json.loads(record.read_text()) if record.exists() else None +script = SCRIPT.get(rec["env"]["OUSAST_TASK"], {{}}) if rec else {{}} + + +def gone(): + print(f'Error: getting task "{{argv[2]}}": rpc error: code = NotFound desc = task not found', file=sys.stderr) + sys.exit(1) + + 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}})) + state = {{"docs": docs, "env": env, "polls": 0, "resumes": 0, "phase": "Suspended", "message": "Task is suspended"}} + (HERE / "applied" / (doc["metadata"]["name"] + ".json")).write_text(json.dumps(state)) 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) +elif argv[:2] == ["resume", "task"]: + if rec is None or rec.get("gone"): + gone() + rec["resumes"] += 1 + if rec["resumes"] <= script.get("resume_failures", 1): + rec.update(phase="Failed", message="resuming actor: context deadline exceeded") + record.write_text(json.dumps(rec)) + print("Error: resuming task: rpc error: code = DeadlineExceeded desc = context deadline exceeded", 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") + rec.update(phase="Running", message="Task is running and its workspace is ready") record.write_text(json.dumps(rec)) - print(yaml.safe_dump({{"apiVersion": "ax.io/v1alpha1", "kind": "Task", "metadata": {{"name": argv[2]}}, "status": {{"phase": phase}}}})) + print(f"task.ax.io/{{argv[2]}} resumed") +elif argv[:2] == ["get", "task"]: + if rec is None or rec.get("gone"): + gone() + if rec["phase"] == "Running": + rec["polls"] += 1 + env, task = rec["env"], rec["env"]["OUSAST_TASK"] + if rec["phase"] == "Running" and rec["polls"] == script.get("polls", 1): + if "phase" in script: + rec.update(phase=script["phase"], message=script.get("message", "")) + if script.get("vanish"): + rec["gone"] = True + if script.get("deliver", True) and "phase" not in script and not script.get("vanish"): + 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-Run": env["OUSAST_RUN"]}} + headers["X-Ousast-Task"] = script.get("as_task", task) + req = urllib.request.Request(env["OUSAST_ARTIFACT_URL"], data=buf.getvalue(), method="POST", headers=headers) + 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)) + if rec.get("gone"): + gone() + condition = {{"type": "Ready", "status": "True" if rec["phase"] == "Running" else "False", "message": rec["message"]}} + status = {{"phase": rec["phase"], "conditions": [condition]}} + print(yaml.safe_dump({{"apiVersion": "ax.io/v1alpha1", "kind": "Task", "metadata": {{"name": argv[2]}}, "status": status}})) elif argv[:2] == ["delete", "task"]: + if rec is not None: + rec["gone"] = True + record.write_text(json.dumps(rec)) print("deleted") ''' @@ -222,7 +256,7 @@ class Fake: 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"]: + elif verb in ("delete", "resume") and words[:2] == [verb, "task"]: out.append((stamp, words[2])) elif verb in ("delivered", "rejected") and words[0] == verb: out.append((stamp, words[1])) @@ -234,6 +268,7 @@ 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") + monkeypatch.setenv("OUSAST_TASK_TIMEOUT", "20") # a test that never delivers fails in seconds, not in 2 hours fake = Fake(tmp_path) fake.script(SCRIPT) return fake @@ -343,18 +378,57 @@ def test_failed_task_halts_the_run(fake: Fake, tmp_path: Path) -> None: 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: +def test_resume_retries_deadline_exceeded_and_completion_is_the_delivery(fake: Fake, tmp_path: Path) -> None: + """ax creates a Task Suspended; the first resume of a new image times out; a later one runs it (live evidence).""" + assert reconciler.run(write_run(tmp_path, "life", " - name: repo-facts\n task: repo-facts\n"), ax=str(fake.ax)) == "done" + verbs = [w[0] for _, w in fake.log() if w[0] in ("apply", "resume", "delivered", "delete")] + assert verbs == ["apply", "resume", "resume", "delivered", "delete"], "resumed twice, delivered, then deleted" + resumes = [stamp for stamp, _ in fake.events("resume")] + (delivered, _), (deleted, _) = fake.events("delivered")[0], fake.events("delete")[0] + assert resumes[1] < delivered < deleted, "the fake delivers only once a resume went through" + assert state_of("life")["tasks"]["repo-facts"]["status"] == "done", "done without any Completed phase from ax" + + +def test_resume_gives_up_after_the_resume_timeout_with_the_last_error(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OUSAST_RESUME_TIMEOUT", "0.3") + fake.script({**SCRIPT, "repo-facts": {**SCRIPT["repo-facts"], "resume_failures": 1000}}) + assert reconciler.run(write_run(tmp_path, "stuck"), ax=str(fake.ax)) == "failed" + reason = state_of("stuck")["tasks"]["repo-facts"]["reason"] + assert "OUSAST_RESUME_TIMEOUT" in reason and "DeadlineExceeded" in reason + assert len(fake.events("resume")) >= 2, "retried with backoff before giving up" + assert [n for _, n in fake.events("delete")] == ["stuck-repo-facts"] + + +def test_ax_failed_phase_before_delivery_fails_with_the_condition_message(fake: Fake, tmp_path: Path) -> None: script = json.loads(json.dumps(SCRIPT)) - script["repo-facts"]["phase"] = "Failed" + script["repo-facts"].update(phase="Failed", message="ActorResumeFailed: worker lost") 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 state_of("phase")["tasks"]["repo-facts"]["reason"] == "ax phase Failed: ActorResumeFailed: worker lost" + assert statuses("phase") == {"repo-facts": "failed", "verify": "pending", "agree": "pending"} + assert [n for _, n in fake.events("delete")] == ["phase-repo-facts"], "ax delete follows a failed task" + + +def test_a_task_that_vanishes_before_delivery_is_failed(fake: Fake, tmp_path: Path) -> None: + fake.script({**SCRIPT, "repo-facts": {**SCRIPT["repo-facts"], "vanish": True}}) + assert reconciler.run(write_run(tmp_path, "vanish"), ax=str(fake.ax)) == "failed" + assert "disappeared from ax" in state_of("vanish")["tasks"]["repo-facts"]["reason"] + + +def test_no_delivery_within_the_task_timeout_is_failed(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OUSAST_TASK_TIMEOUT", "0.5") + fake.script({**SCRIPT, "repo-facts": {**SCRIPT["repo-facts"], "deliver": False}}) + started = time.monotonic() 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" + assert time.monotonic() - started >= 0.5 + reason = state_of("silent")["tasks"]["repo-facts"]["reason"] + assert "OUSAST_TASK_TIMEOUT" in reason and "Running" in reason, "Running forever is not completion" + assert [n for _, n in fake.events("delete")] == ["silent-repo-facts"], "ax delete follows every task" + + +def test_outcome_of_needs_a_delivered_summary() -> None: + assert reconciler.outcome_of(None) == ("failed", "no summary.json delivered") + assert reconciler.outcome_of({"status": "done", "units_done": 1, "units_total": 1}) == ("done", "") def test_summary_done_requires_every_unit(fake: Fake, tmp_path: Path) -> None: @@ -393,7 +467,8 @@ def test_receiver_rejects_traversal_unknown_tasks_and_wrong_types(tmp_path: Path receiver.server_close() -def test_delivery_for_a_task_that_is_not_running_is_rejected(fake: Fake, tmp_path: Path) -> None: +def test_delivery_for_a_task_that_is_not_running_is_rejected(fake: Fake, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OUSAST_TASK_TIMEOUT", "1") # the rejected delivery is no delivery: the task times out script = json.loads(json.dumps(SCRIPT)) script["repo-facts"]["as_task"] = "agree" # the runner claims to be a task that has not started fake.script(script) @@ -544,12 +619,12 @@ def test_elapsed_polling_is_paced(fake: Fake, tmp_path: Path, monkeypatch: pytes def test_doctor_treats_an_empty_namespace_as_not_ready(monkeypatch: pytest.MonkeyPatch) -> None: """kubectl's "No resources found" must not read as one ready pod (a plausible-zero instrument failure).""" - from openultrasast.plane import reconciler + from openultrasast.plane import doctor - monkeypatch.setattr(reconciler, "_sh", lambda *a: (True, "No resources found in ax-system namespace.")) - ok, text = reconciler._pods_ready("ax-system", "kind-ousast") + monkeypatch.setattr(doctor, "_sh", lambda *a: (True, "No resources found in ax-system namespace.")) + ok, text = doctor._pods_ready("ax-system", "kind-ousast") assert not ok and "no pods" in text - monkeypatch.setattr(reconciler, "_sh", lambda *a: (True, "a-1 1/1 Running 0 1m\nb-2 0/1 Completed 0 1m")) - assert reconciler._pods_ready("ns", "ctx") == (True, "2 pods ready") - monkeypatch.setattr(reconciler, "_sh", lambda *a: (True, "a-1 0/1 Running 0 1m")) - assert reconciler._pods_ready("ns", "ctx")[0] is False + monkeypatch.setattr(doctor, "_sh", lambda *a: (True, "a-1 1/1 Running 0 1m\nb-2 0/1 Completed 0 1m")) + assert doctor._pods_ready("ns", "ctx") == (True, "2 pods ready") + monkeypatch.setattr(doctor, "_sh", lambda *a: (True, "a-1 0/1 Running 0 1m")) + assert doctor._pods_ready("ns", "ctx")[0] is False diff --git a/tests/test_plane_runner.py b/tests/test_plane_runner.py index 4f9b49f..0e2b015 100644 --- a/tests/test_plane_runner.py +++ b/tests/test_plane_runner.py @@ -2,14 +2,17 @@ from __future__ import annotations -import importlib.util import io import json import os +import shutil +import signal +import socket import subprocess import sys import tarfile import threading +import time import urllib.error import urllib.request from collections.abc import Iterator @@ -30,36 +33,38 @@ from openultrasast.plane.runner import ( ) STUB = Path(__file__).parent / "plane_stub_task.py" -STUB_MODULE = f"{runner.TASK_PACKAGE}.stub_task" +STUB_PACKAGE = "ousast_stub_tasks" +SRC = Path(runner.__file__).resolve().parents[2] # the ``src`` this test imports; the children must import it too -@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(scope="session") +def stub_root(tmp_path_factory: pytest.TempPathFactory) -> Path: + """A throwaway package holding the stub, so the runner starts it as ``python -m ousast_stub_tasks.stub_task``.""" + root = tmp_path_factory.mktemp("stubs") + (root / STUB_PACKAGE).mkdir() + (root / STUB_PACKAGE / "__init__.py").write_text("", encoding="utf-8") + shutil.copy(STUB, root / STUB_PACKAGE / "stub_task.py") + return root @pytest.fixture -def env(monkeypatch: pytest.MonkeyPatch) -> Iterator[dict[str, str]]: +def env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, stub_root: Path) -> 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. + would not undo what the runner exports; the snapshot does. ``AX_RUNNER_EXIT_AFTER_COMMAND`` makes ``main()`` + return after delivery instead of serving until SIGTERM, as the image 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("AX_RUNNER_EXIT_AFTER_COMMAND", "1") + monkeypatch.setenv("AX_RUNNER_TASK_PACKAGE", STUB_PACKAGE) + monkeypatch.setenv("OUSAST_STATE_DIR", str(tmp_path / "state")) monkeypatch.setenv("OUSAST_DELIVERY_BACKOFF", "0") + monkeypatch.setenv("PYTHONPATH", os.pathsep.join(filter(None, [str(stub_root), str(SRC), os.environ.get("PYTHONPATH")]))) try: yield os.environ finally: @@ -156,8 +161,10 @@ def test_readyz_is_503_until_workspaces_are_ready_then_200() -> None: assert get(f"{base}/healthz") == 200 assert get(f"{base}/readyz") == 503 assert get(f"{base}/other") == 404 + assert get(f"{base}/readyz?check=workspace") == 503, "ax's controller probes with a query string" ready.set() assert get(f"{base}/readyz") == 200 + assert get(f"{base}/readyz?check=workspace") == 200 finally: server.stop() @@ -297,6 +304,128 @@ def test_an_unknown_task_module_is_failed(env: dict[str, str], monkeypatch: pyte assert "no_such_task" in json.loads((output / "summary.json").read_text())["reason"] +# --- ax's lifecycle: a child process, stay up, SIGTERM, completion marker (docs/runner.md, docs/sandbox.md) ----------------- + + +def test_the_command_is_a_child_in_its_own_group_in_the_first_workspace( + 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["pid"] != os.getpid(), "the task is a child process, not an import" + assert facts["pgid"] == facts["pid"] != os.getpgid(0), "the child leads its own process group" + assert facts["cwd"] == str(tmp_path / "ws"), "the first workspace is the working directory" + assert facts["metadata_url"] == f"http://127.0.0.1:{os.environ['AX_RUNNER_BOUND_PORT']}" + assert "name: stub" in facts["metadata_task"], "AX_METADATA_URL serves the Task YAML" + + +def free_port() -> int: + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +def alive(pid: int) -> bool: + """True while ``pid`` exists and is not a zombie (a killed orphan may wait for its reaper).""" + try: + return Path(f"/proc/{pid}/stat").read_text().split(")")[-1].split()[0] != "Z" + except (FileNotFoundError, ProcessLookupError, IndexError): + return False + + +def wait_for(predicate: object, timeout: float = 30.0) -> None: + deadline = time.monotonic() + timeout + while not predicate(): # type: ignore[operator] + assert time.monotonic() < deadline, "timed out" + time.sleep(0.05) + + +def spawn_runner(tmp_path: Path, port: int, **extra: str) -> subprocess.Popen[bytes]: + """``python -m openultrasast.plane.runner`` as the image runs it: HTTP on, no exit-after-command escape.""" + child_env = {k: v for k, v in os.environ.items() if k != "AX_RUNNER_EXIT_AFTER_COMMAND"} + child_env.update(AX_TASK_YAML=task_yaml(tmp_path / "out", tmp_path / "ws"), AX_WORKSPACES_YAML=files_workspace_yaml()) + child_env.update(AX_RUNNER_HTTP="1", AX_RUNNER_PORT=str(port), **extra) + log = (tmp_path / "runner.log").open("wb") + return subprocess.Popen([sys.executable, "-m", "openultrasast.plane.runner"], env=child_env, stdout=log, stderr=log) + + +def test_the_runner_keeps_serving_after_the_command_and_exits_0_on_sigterm(env: dict[str, str], receiver: Receiver, tmp_path: Path) -> None: + port = free_port() + proc = spawn_runner(tmp_path, port, OUSAST_ARTIFACT_URL=receiver.url) + marker = tmp_path / "state" / "stub.done.json" + try: + wait_for(marker.exists) + time.sleep(0.5) + assert proc.poll() is None, "PID 1 must outlive the command (ax: 'Container guest is stopped' otherwise)" + base = f"http://127.0.0.1:{port}" + assert get(f"{base}/healthz") == 200 and get(f"{base}/readyz?check=workspace") == 200 + assert get(f"{base}/metadata/v1alpha1/ax/workspaces") == 200 + assert len(receiver.posts) == 1 + assert json.loads(marker.read_text()) | {"summary": None} == {"exit": 0, "status": "done", "delivered": True, "summary": None} + proc.send_signal(signal.SIGTERM) + assert proc.wait(timeout=10) == 0 + finally: + proc.kill() + proc.wait() + + +def test_sigterm_goes_to_the_commands_group_then_sigkill_after_the_grace(env: dict[str, str], receiver: Receiver, tmp_path: Path) -> None: + port = free_port() + proc = spawn_runner(tmp_path, port, OUSAST_ARTIFACT_URL=receiver.url, STUB_MODE="sleep", OUSAST_TERM_GRACE="1") + pids_file = tmp_path / "out" / "pids.json" + try: + wait_for(pids_file.exists) + pids = json.loads(pids_file.read_text()) + assert all(alive(pid) for pid in pids) + started = time.monotonic() + proc.send_signal(signal.SIGTERM) + assert proc.wait(timeout=15) == 0 + assert time.monotonic() - started >= 0.9, "the command ignored SIGTERM, so the runner waited out the grace" + wait_for(lambda: not any(alive(pid) for pid in pids), timeout=5) + finally: + proc.kill() + proc.wait() + assert not (tmp_path / "state" / "stub.done.json").exists(), "an interrupted command is not complete" + assert receiver.posts == [], "nothing is delivered for an interrupted command" + + +def test_the_completion_marker_prevents_a_rerun_on_resume( + env: dict[str, str], receiver: Receiver, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + counter = tmp_path / "runs.txt" + code, _ = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=receiver.url, STUB_COUNTER=str(counter)) + assert code == 0 and len(counter.read_text().splitlines()) == 1 and len(receiver.posts) == 1 + (tmp_path / "ws" / "candidates.json").write_text("changed by the task\n") + code, _ = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=receiver.url, STUB_COUNTER=str(counter)) + assert code == 0 + assert len(counter.read_text().splitlines()) == 1, "a resumed actor must not repeat billed work" + assert len(receiver.posts) == 1, "a delivered result is not delivered again" + assert (tmp_path / "ws" / "candidates.json").read_text() == "changed by the task\n", "workspaces are prepared once" + + +def test_a_failed_delivery_is_retried_alone_on_the_next_boot( + env: dict[str, str], receiver: Receiver, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + counter = tmp_path / "runs.txt" + broken = Receiver(500) + try: + code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=broken.url, STUB_COUNTER=str(counter)) + finally: + broken.stop() + assert code == 2 + marker = json.loads((tmp_path / "state" / "stub.done.json").read_text()) + assert marker["delivered"] is False and marker["status"] == "failed" and marker["exit"] == 0 + code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_ARTIFACT_URL=receiver.url, STUB_COUNTER=str(counter)) + assert code == 0 + assert len(counter.read_text().splitlines()) == 1, "only the delivery is retried" + with tarfile.open(fileobj=io.BytesIO(receiver.posts[0][1]), mode="r:") as archive: + summary = archive.extractfile("summary.json") + assert summary is not None and json.loads(summary.read())["status"] == "done", "the task's own summary" + marker = json.loads((tmp_path / "state" / "stub.done.json").read_text()) + assert marker["delivered"] is True and marker["status"] == "done" + + # --- git workspaces from the case cache and by shallow clone ----------------------------------------------------------------