diff --git a/docs/dogfood/2026-06-16-oracle-slice1-dogfood.md b/docs/dogfood/2026-06-16-oracle-slice1-dogfood.md new file mode 100644 index 0000000..dd3805e --- /dev/null +++ b/docs/dogfood/2026-06-16-oracle-slice1-dogfood.md @@ -0,0 +1,103 @@ +# Oracle Slice-1 DOGFOOD — cross-repo acceptance test + +**Date:** 2026-06-16 +**Task:** 5.3 — cross-repo DOGFOOD acceptance test (capstone of Slice 1) +**Test:** `tests/core/test_oracle_dogfood.py` (engine worktree) +**Status:** GREEN — 3/3 passed (real podman, real CLI, real ship-check subprocess; no mocks) + +## What was proven + +The unfakable Atlas oracle gates a real `prd-taskmaster` ship-check end-to-end. +For each DONE task, `skel/ship-check.py` (Gate 5) shells `atlas oracle grade`, +which checks out the submitted commit into a throwaway worktree, **overlays the +operator-held tests over the submitter's tree**, re-executes the card's grading +command inside a digest-pinned podman sandbox, derives `PASS` iff exit 0, and +appends a tamper-evident ledger event. + +1. **Genuine pass ships.** Operator-held `grade.sh` = `exit 0`. Ship-check emits + `SHIP_CHECK_OK`, returncode 0, and the ledger records `verdict == "PASS"`. +2. **Reward hack blocked (the acceptance criterion).** The submitter ships a + cheat — a committed `grade.sh` that always `exit 0` — while the operator-held + `grade.sh` is `exit 1`. The cheat does **NOT** ship: the oracle overlays and + re-executes the operator's copy, so the verdict is `FAIL`. `SHIP_CHECK_OK` is + absent, returncode is non-zero, stderr names the oracle FAIL for task 1, and + the ledger records `verdict == "FAIL"`. **Non-vacuous:** the only difference + between case 1 and case 2 is which `grade.sh` the oracle re-runs — the + submitter's committed copy never reaches the verdict. +3. **Ledger integrity.** `atlas ledger verify ` reports + `{"ok": true, "eventCount": 1}` after the genuine pass. + +## ATLAS_ORACLE_CMD used + +``` +ATLAS_ORACLE_CMD="/node_modules/.bin/tsx /apps/cli/src/index.ts" +``` +(`` = your local atlas-protocol checkout. The test reads `ATLAS_ORACLE_CMD` from the environment and skips when it is unset.) + +`ship-check.py` shlex-splits `ATLAS_ORACLE_CMD` and appends +`oracle grade --repo --commit --card ... --held ... --evidence ... --ledger ...`. + +**Why `tsx` on the CLI source and not `node dist/index.js`:** the spine +monorepo's workspace packages (`@atlas-protocol/core|cards|evidence|executor`) +declare `exports: "./src/index.ts"`, so the compiled `apps/cli/dist/index.js` +resolves its workspace deps to TypeScript sources that bare `node` cannot load +(`ERR_MODULE_NOT_FOUND: .../packages/executor/src/grade.js`). The `tsx` +executable runs the identical CLI code path the spine's own vitest suite uses, +with **no edits to the spine repo**. (Building the CLI — `pnpm --filter +@atlas-protocol/cli build`, plus `pnpm -r build` for the workspace deps — was +done as Step 0 and is required so the source resolves; the dist itself is not +the run target.) + +## Observed results + +### Pass case (operator-held `grade.sh` = `exit 0`) + +``` +$ ATLAS_ORACLE_CMD="$TSX $SRC" python3 skel/ship-check.py --cwd /tmp/ev_pass +SHIP_CHECK_OK +rc=0 +``` + +Ledger event payload (excerpt): + +```json +{ + "type": "verification.completed", + "actor": { "kind": "executor", "id": "oracle" }, + "lifecycleState": "verification_passed", + "payload": { "verdict": "PASS", "exitCode": 0, "overlayHash": "sha256:..." } +} +``` + +### Reward-hack case (submitter cheat `exit 0`, operator-held `exit 1`) + +``` +$ ATLAS_ORACLE_CMD="$TSX $SRC" python3 skel/ship-check.py --cwd /tmp/ev_hack +rc=1 +--- stdout (empty — no SHIP_CHECK_OK) --- +--- stderr --- +FAIL: task 1: oracle verdict FAIL +``` + +The committed cheat (`exit 0`) was overlaid away by the operator-held `exit 1` +and re-executed in the sandbox → verdict `FAIL` → ship blocked. + +## pytest + +``` +$ python3 -m pytest tests/core/test_oracle_dogfood.py -v +tests/core/test_oracle_dogfood.py::test_genuine_pass_ships PASSED [ 33%] +tests/core/test_oracle_dogfood.py::test_reward_hack_blocked PASSED [ 66%] +tests/core/test_oracle_dogfood.py::test_ledger_integrity PASSED [100%] +============================== 3 passed in 13.05s ============================== +``` + +## Slice-2 hardening note + +The Graded Card's `contentHash` is a syntactically valid placeholder. +`gradeSubmission` does not re-verify `contentHash` against the card body in +Slice 1 (it is only echoed into the ledger payload). Slice 2 should recompute +and verify it before grading so a tampered card body cannot be graded under a +stale hash. (Also tracked in the executor's `SLICE-1` deferral comments: +infra-failure exit codes are currently recorded as `FAIL`, and `evidenceRef` +records a path rather than a `sha256:`-prefixed content ref.) diff --git a/prd_taskmaster/oracle_bridge.py b/prd_taskmaster/oracle_bridge.py new file mode 100644 index 0000000..8807cb2 --- /dev/null +++ b/prd_taskmaster/oracle_bridge.py @@ -0,0 +1,127 @@ +"""Oracle bridge: map a CDD card to a Graded Card verdict via the atlas oracle CLI. + +Fail-closed: any ambiguity (missing verdict, unparseable output, CLI crash) yields +("FAIL", {...}) — never "PASS" and never an uncaught exception from the grading path. +OracleCardError is the one intended raise, for a missing/unreadable/invalid card. +""" +from __future__ import annotations + +import json +import os +import subprocess +from pathlib import Path + + +class OracleCardError(Exception): + """Raised when a CDD card cannot be graded (e.g. missing 'grading' block).""" + + +def _oracle_cmd() -> list[str]: + """Configurable CLI invocation. Default: the 'atlas' binary on PATH. + + Override with ATLAS_ORACLE_CMD (shell-split), e.g.: + ATLAS_ORACLE_CMD="node /path/to/atlas-protocol/apps/cli/dist/index.js" + """ + raw = os.environ.get("ATLAS_ORACLE_CMD") + if raw: + import shlex + return shlex.split(raw) + return ["atlas"] + + +def grade_card( + *, + card_path: str | Path, + repo_path: str | Path, + commit_sha: str, + held_root: str | Path, + evidence_dir: str | Path, + ledger_dir: str | Path, + oracle_cmd: list[str] | None = None, +) -> tuple[str, dict]: + """Grade a submission via the atlas oracle CLI. + + Returns (verdict, detail) where verdict is "PASS" or "FAIL". + + FAIL-CLOSED: any error (missing verdict, unparseable output, CLI crash) + yields ("FAIL", {...}) — never raises after the card has been validated. + + Raises: + OracleCardError: if the card file is missing, unreadable, or has no 'grading' block. + """ + card_path = Path(card_path) + + try: + card = json.loads(card_path.read_text()) + except (OSError, json.JSONDecodeError) as exc: + raise OracleCardError(f"cannot read card {card_path}: {exc}") from exc + + if "grading" not in card: + raise OracleCardError( + f"card {card_path} has no 'grading' block; cannot grade" + ) + + cmd = (oracle_cmd or _oracle_cmd()) + [ + "oracle", "grade", + "--repo", str(repo_path), + "--commit", commit_sha, + "--card", str(card_path), + "--held", str(held_root), + "--evidence", str(evidence_dir), + "--ledger", str(ledger_dir), + ] + + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + except (OSError, subprocess.SubprocessError) as exc: + return ("FAIL", {"error": f"oracle CLI invocation failed: {exc}"}) + + try: + parsed = json.loads(proc.stdout) + except json.JSONDecodeError: + return ( + "FAIL", + { + "error": "oracle CLI produced no parseable JSON verdict", + "returncode": proc.returncode, + "stdout": proc.stdout[:500], + "stderr": proc.stderr[:500], + }, + ) + + verdict = parsed.get("verdict") + if verdict not in ("PASS", "FAIL"): + return ("FAIL", {"error": "oracle CLI verdict missing/invalid", "parsed": parsed}) + + return (verdict, parsed) + + +def grade_task( + *, + repo_root: str | Path, + task_id, + commit_sha: str, + held_root: str | Path, + evidence_dir: str | Path, + ledger_dir: str | Path, + oracle_cmd: list[str] | None = None, +) -> tuple[str, dict]: + """Convenience: locate the CDD card for a task and grade it. + + Looks for .atlas-ai/cdd/task-.json under repo_root. + + Raises: + OracleCardError: if the card file does not exist or fails validation. + """ + card_path = Path(repo_root) / ".atlas-ai" / "cdd" / f"task-{task_id}.json" + if not card_path.exists(): + raise OracleCardError(f"no CDD card at {card_path}") + return grade_card( + card_path=card_path, + repo_path=repo_root, + commit_sha=commit_sha, + held_root=held_root, + evidence_dir=evidence_dir, + ledger_dir=ledger_dir, + oracle_cmd=oracle_cmd, + ) diff --git a/prd_taskmaster/render.py b/prd_taskmaster/render.py index f70951c..e1c675d 100644 --- a/prd_taskmaster/render.py +++ b/prd_taskmaster/render.py @@ -261,7 +261,7 @@ def shipcheck_panel(shipcheck: dict, *, ascii_mode: bool = False) -> str: lines.append(f"{sym} Gate {i} {name:<26} {word}") lines.append("") if shipcheck.get("passed"): - token = "SHIP_CHECK_OK" + (" [OVERRIDE]" if shipcheck.get("override_active") else "") + token = "SHIP_CHECK_OK" lines.append(f"{ok} {token}") else: lines.append(f"{blocked} not shippable — {len(shipcheck.get('failures', []))} gate(s) failed") diff --git a/prd_taskmaster/shipcheck.py b/prd_taskmaster/shipcheck.py index 2567741..3c9791e 100644 --- a/prd_taskmaster/shipcheck.py +++ b/prd_taskmaster/shipcheck.py @@ -1,8 +1,9 @@ #!/usr/bin/env python3 -"""Deterministic ship-check for prd-taskmaster pipelines. +"""NON-BINDING status-display heuristic for prd-taskmaster pipelines. -Emits SHIP_CHECK_OK to stdout ONLY when all gates pass. Called by execute-task -at termination AND by Step 9 (--dry-run) as a per-task predicate. +This module is imported by status.py (dry_run=True) ONLY for display purposes. +It is NOT the binding ship gate — the binding gate is skel/ship-check.py +(oracle-backed, unfakable). Do not add an oracle call here. Gate logic (grounded against actual pipeline.json / tasks.json schemas observed 2026-06-04 in ai-human-tasker): @@ -27,22 +28,20 @@ skel checked .atlas-ai/ralph-loop-prompt.md — wrong path and irrelevant after /goal migration. - Gate 5 (HARD) — No non-zero "Exit status N" line in any .atlas-ai/evidence/ - file. This is the convergent must-do from the 2026-06-04 forensic audit - (T12 marked DONE while pnpm test exited 1 with 11 failing tests). - Override only via SHIP_CHECK_OVERRIDE_ADMIN token; overrides are logged - to .atlas-ai/state/execute-log.jsonl as an audit record. + Gate 5 (display only) — No non-zero "Exit status N" line in any + .atlas-ai/evidence/ file. This is a heuristic display signal only; + the binding oracle check is in skel/ship-check.py. There is no override + path — the gate result cannot be bypassed. Interface (standalone shim, created in a later step): python3 .atlas-ai/ship-check.py # standard gate python3 .atlas-ai/ship-check.py --dry-run # always exit 0; report on stderr - python3 .atlas-ai/ship-check.py --override SHIP_CHECK_OVERRIDE_ADMIN # bypass Gate 5 python3 .atlas-ai/ship-check.py --cwd /path/to/project # explicit project root Exit codes: - 0 — SHIP_CHECK_OK (stdout: exactly "SHIP_CHECK_OK\n", with " [OVERRIDE]" suffix if applicable) + 0 — SHIP_CHECK_OK (stdout: exactly "SHIP_CHECK_OK\n") 1 — gate failures (stderr only; nothing on stdout — log watchers must not see partial matches) - 2 — script error (IO, JSON parse, bad token) + 2 — script error (IO, JSON parse) """ from __future__ import annotations @@ -50,11 +49,9 @@ import json import re import sys -from datetime import datetime, timezone from pathlib import Path from typing import List, Optional, Tuple -OVERRIDE_TOKEN = "SHIP_CHECK_OVERRIDE_ADMIN" EXIT_STATUS_RE = re.compile(r"\bExit status\s+(\d+)\b", re.IGNORECASE) @@ -163,21 +160,7 @@ def gate_exit_codes(atlas: Path) -> Tuple[bool, List[str]]: return len(failures) == 0, failures -def log_override(atlas: Path, message: str) -> None: - log = atlas / "state" / "execute-log.jsonl" - log.parent.mkdir(parents=True, exist_ok=True) - entry = { - "iteration": "OVERRIDE", - "timestamp": datetime.now(timezone.utc).isoformat(), - "task_id": "SHIP_CHECK", - "event": "override_invoked", - "message": message, - } - with log.open("a") as fp: - fp.write(json.dumps(entry) + "\n") - - -def run_all_gates(repo_root: Path, override_active: bool) -> Tuple[bool, List[str]]: +def run_all_gates(repo_root: Path) -> Tuple[bool, List[str]]: failures: List[str] = [] atlas = repo_root / ".atlas-ai" @@ -194,60 +177,37 @@ def run_all_gates(repo_root: Path, override_active: bool) -> Tuple[bool, List[st _, f4 = gate_plan(repo_root) failures.extend(f4) - ok5, f5 = gate_exit_codes(atlas) - if not ok5: - if override_active: - log_override(atlas, f"Gate 5 bypassed: {'; '.join(f5[:3])}{' ...' if len(f5) > 3 else ''}") - # Override accepts the failures; no append to global failures list - else: - failures.extend(f5) + _, f5 = gate_exit_codes(atlas) + failures.extend(f5) return len(failures) == 0, failures -def run_ship_check(cwd: Optional[str] = None, dry_run: bool = False, - override: Optional[str] = None) -> dict: - """Importable ship-check entry point. NEVER calls sys.exit. +def run_ship_check(cwd: Optional[str] = None, dry_run: bool = False) -> dict: + """Importable ship-check entry point. NON-BINDING display heuristic only. + NEVER calls sys.exit. There is no override path. Args: cwd: project root (defaults to current working directory). dry_run: when True, the returned exit_code is forced to 0 regardless of gate outcome (gates still run and the report is preserved). - override: if equal to OVERRIDE_TOKEN, Gate 5 (exit codes) is bypassed. Returns a dict: - passed: bool — True when all (non-overridden) gates pass. - failures: list[str] — per-gate failure detail (empty when passed). - override_active: bool — whether a valid override token was supplied. - override_invalid: bool — an override value was supplied but did not match. + passed: bool — True when all gates pass. + failures: list[str] — per-gate failure detail (empty when passed). dry_run: bool - exit_code: int — 0 / 1 / 2 mirroring the CLI contract. - error: str | None — populated on a script error (exit_code 2). - stdout: str | None — the exact stdout line the CLI would print, if any. + exit_code: int — 0 / 1 / 2 mirroring the CLI contract. + error: str | None — populated on a script error (exit_code 2). + stdout: str | None — the exact stdout line the CLI would print, if any. """ - if override is not None and override != OVERRIDE_TOKEN: - return { - "passed": False, - "failures": [], - "override_active": False, - "override_invalid": True, - "dry_run": dry_run, - "exit_code": 2, - "error": "--override value does not match expected token", - "stdout": None, - } - - override_active = override == OVERRIDE_TOKEN repo_root = Path(cwd).resolve() if cwd else Path.cwd() try: - ok, failures = run_all_gates(repo_root, override_active=override_active) + ok, failures = run_all_gates(repo_root) except Exception as exc: # noqa: BLE001 — top-level guard return { "passed": False, "failures": [], - "override_active": override_active, - "override_invalid": False, "dry_run": dry_run, "exit_code": 2, "error": f"ship-check script error: {exc!r}", @@ -259,8 +219,7 @@ def run_ship_check(cwd: Optional[str] = None, dry_run: bool = False, stdout = None elif ok: exit_code = 0 - suffix = " [OVERRIDE]" if override_active else "" - stdout = f"SHIP_CHECK_OK{suffix}" + stdout = "SHIP_CHECK_OK" else: exit_code = 1 stdout = None @@ -268,8 +227,6 @@ def run_ship_check(cwd: Optional[str] = None, dry_run: bool = False, return { "passed": ok, "failures": failures, - "override_active": override_active, - "override_invalid": False, "dry_run": dry_run, "exit_code": exit_code, "error": None, @@ -278,16 +235,14 @@ def run_ship_check(cwd: Optional[str] = None, dry_run: bool = False, def main(argv: Optional[List[str]] = None) -> int: - parser = argparse.ArgumentParser(description="Deterministic ship-check for prd-taskmaster pipelines.") + parser = argparse.ArgumentParser(description="NON-BINDING status-display ship-check for prd-taskmaster pipelines.") parser.add_argument("--dry-run", action="store_true", help="Run all gates but always exit 0. Report goes to stderr. Used by execute-task Step 9 as a per-task predicate.") - parser.add_argument("--override", type=str, default=None, - help=f"Bypass Gate 5 (exit codes) if value equals {OVERRIDE_TOKEN}. Logged to execute-log.jsonl.") parser.add_argument("--cwd", type=str, default=None, help="Project root (defaults to current working directory).") args = parser.parse_args(argv) - result = run_ship_check(cwd=args.cwd, dry_run=args.dry_run, override=args.override) + result = run_ship_check(cwd=args.cwd, dry_run=args.dry_run) # Script error (bad token or IO/JSON failure) — exit 2, message on stderr. if result["exit_code"] == 2: diff --git a/skel/ship-check.py b/skel/ship-check.py index 373f4a2..e5103bf 100755 --- a/skel/ship-check.py +++ b/skel/ship-check.py @@ -4,6 +4,11 @@ Emits SHIP_CHECK_OK to stdout ONLY when all gates pass. Called by execute-task at termination AND by Step 9 (--dry-run) as a per-task predicate. +Standalone: this file ships into user projects as `.atlas-ai/ship-check.py`. It +imports ONLY the stdlib and MUST stay that way (no `import prd_taskmaster` — that +package is not importable in a user project). The oracle gate therefore shells +the `atlas oracle grade` CLI directly via subprocess. + Gate logic (grounded against actual pipeline.json / tasks.json schemas observed 2026-06-04 in ai-human-tasker): @@ -27,36 +32,37 @@ skel checked .atlas-ai/ralph-loop-prompt.md — wrong path and irrelevant after /goal migration. - Gate 5 (HARD) — No non-zero "Exit status N" line in any .atlas-ai/evidence/ - file. This is the convergent must-do from the 2026-06-04 forensic audit - (T12 marked DONE while pnpm test exited 1 with 11 failing tests). - Override only via SHIP_CHECK_OVERRIDE_ADMIN token; overrides are logged - to .atlas-ai/state/execute-log.jsonl as an audit record. + Gate 5 (HARD, ORACLE) — Each DONE task is RE-GRADED by the atlas oracle. + The submitter's own .atlas-ai/evidence/ is NOT trusted; the oracle + re-executes the CDD card's grading against HEAD and writes its own + evidence/ledger. This replaces the fakable "no non-zero Exit status N in + evidence" grep — which silently PASSED when no evidence existed and had a + self-grantable bypass token. Both the silent-pass loophole and the bypass + token are GONE. The gate is FAIL-CLOSED: a missing card, a card with no + grading block, a CLI crash, unparseable output, or any verdict other than + "PASS" all BLOCK the ship. Interface: python3 .atlas-ai/ship-check.py # standard gate python3 .atlas-ai/ship-check.py --dry-run # always exit 0; report on stderr - python3 .atlas-ai/ship-check.py --override SHIP_CHECK_OVERRIDE_ADMIN # bypass Gate 5 python3 .atlas-ai/ship-check.py --cwd /path/to/project # explicit project root Exit codes: - 0 — SHIP_CHECK_OK (stdout: exactly "SHIP_CHECK_OK\n", with " [OVERRIDE]" suffix if applicable) + 0 — SHIP_CHECK_OK (stdout: exactly "SHIP_CHECK_OK\n") 1 — gate failures (stderr only; nothing on stdout — log watchers must not see partial matches) - 2 — script error (IO, JSON parse, bad token) + 2 — script error (IO, JSON parse) """ from __future__ import annotations import argparse import json -import re +import os +import shlex +import subprocess import sys -from datetime import datetime, timezone from pathlib import Path from typing import List, Tuple -OVERRIDE_TOKEN = "SHIP_CHECK_OVERRIDE_ADMIN" -EXIT_STATUS_RE = re.compile(r"\bExit status\s+(\d+)\b", re.IGNORECASE) - def gate_pipeline(atlas: Path) -> Tuple[bool, List[str]]: pf = atlas / "state" / "pipeline.json" @@ -138,46 +144,101 @@ def gate_plan(repo_root: Path) -> Tuple[bool, List[str]]: return False, ["no plan file at .taskmaster/docs/plan.md or docs/superpowers/plans/*.md"] -def gate_exit_codes(atlas: Path) -> Tuple[bool, List[str]]: +def _oracle_cmd() -> List[str]: + """Configurable oracle CLI invocation. Default: the 'atlas' binary on PATH. + + Set ATLAS_ORACLE_CMD (shell-split) to change it, e.g.: + ATLAS_ORACLE_CMD="node /path/to/atlas-protocol/apps/cli/dist/index.js" + """ + raw = os.environ.get("ATLAS_ORACLE_CMD") + if raw: + return shlex.split(raw) + return ["atlas"] + + +def _head_commit(repo_root: Path) -> str: + """HEAD sha via `git rev-parse HEAD`. FAIL-CLOSED: any failure returns the + sentinel "UNKNOWN" so the oracle mismatches the working tree and blocks.""" + try: + proc = subprocess.run( + ["git", "-C", str(repo_root), "rev-parse", "HEAD"], + capture_output=True, text=True, timeout=30, + ) + except (OSError, subprocess.SubprocessError): + return "UNKNOWN" + if proc.returncode != 0: + return "UNKNOWN" + sha = (proc.stdout or "").strip() + return sha or "UNKNOWN" + + +def gate_oracle(repo_root: Path, tasks: list, head_commit: str) -> Tuple[bool, List[str]]: + """Re-grade every DONE task via the atlas oracle CLI. FAIL-CLOSED. + + For each done task: the CDD card must exist and carry a 'grading' block, + and the oracle must return verdict=="PASS". Anything else — missing card, + no grading block, CLI crash, unparseable output, a non-PASS verdict — + appends a failure. The submitter's evidence is NOT read here; the oracle + re-executes the grading and writes its own evidence/ledger. + """ failures: List[str] = [] - evidence_dir = atlas / "evidence" - if not evidence_dir.exists(): - # Gate 3 (CDD) catches missing evidence; this gate is silent when no evidence exists - return True, [] - for f in evidence_dir.rglob("*"): - if not f.is_file(): + atlas = repo_root / ".atlas-ai" + held = atlas / "held-out" + evidence = atlas / "evidence" + ledger = atlas / "ledger" + cmd_base = _oracle_cmd() + + for t in tasks: + if t.get("status") != "done": + continue + tid = t.get("id") + card_path = atlas / "cdd" / f"task-{tid}.json" + if not card_path.exists(): + failures.append(f"task {tid}: no CDD card to grade") continue try: - text = f.read_text(errors="ignore") - except OSError: + card = json.loads(card_path.read_text()) + except (OSError, json.JSONDecodeError) as exc: + failures.append(f"task {tid}: cannot read CDD card ({exc})") + continue + if "grading" not in card: + failures.append(f"task {tid}: CDD card has no grading block") + continue + + cmd = cmd_base + [ + "oracle", "grade", + "--repo", str(repo_root), + "--commit", head_commit, + "--card", str(card_path), + "--held", str(held), + "--evidence", str(evidence), + "--ledger", str(ledger), + ] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + except (OSError, subprocess.SubprocessError) as exc: + failures.append(f"task {tid}: oracle CLI invocation failed ({exc})") + continue + + try: + parsed = json.loads(proc.stdout) + except (json.JSONDecodeError, TypeError): + failures.append(f"task {tid}: oracle produced no parseable JSON verdict (rc={proc.returncode})") continue - for match in EXIT_STATUS_RE.finditer(text): - try: - code = int(match.group(1)) - except (ValueError, IndexError): - continue - if code != 0: - rel = f.relative_to(atlas.parent) if atlas.parent in f.parents else f - failures.append(f"non-zero exit in {rel}: Exit status {code}") - break # one report per file is enough - return len(failures) == 0, failures + verdict = parsed.get("verdict") if isinstance(parsed, dict) else None + if verdict not in ("PASS", "FAIL"): + failures.append(f"task {tid}: oracle verdict missing/invalid ({verdict!r})") + continue + if verdict == "FAIL": + failures.append(f"task {tid}: oracle verdict FAIL") + continue + # verdict == "PASS" — the only path that does NOT append a failure. -def log_override(atlas: Path, message: str) -> None: - log = atlas / "state" / "execute-log.jsonl" - log.parent.mkdir(parents=True, exist_ok=True) - entry = { - "iteration": "OVERRIDE", - "timestamp": datetime.now(timezone.utc).isoformat(), - "task_id": "SHIP_CHECK", - "event": "override_invoked", - "message": message, - } - with log.open("a") as fp: - fp.write(json.dumps(entry) + "\n") + return len(failures) == 0, failures -def run_all_gates(repo_root: Path, override_active: bool) -> Tuple[bool, List[str]]: +def run_all_gates(repo_root: Path) -> Tuple[bool, List[str]]: failures: List[str] = [] atlas = repo_root / ".atlas-ai" @@ -191,17 +252,13 @@ def run_all_gates(repo_root: Path, override_active: bool) -> Tuple[bool, List[st _, f3 = gate_cdd(atlas, tasks) failures.extend(f3) + head = _head_commit(repo_root) + _, f5 = gate_oracle(repo_root, tasks, head) + failures.extend(f5) + _, f4 = gate_plan(repo_root) failures.extend(f4) - ok5, f5 = gate_exit_codes(atlas) - if not ok5: - if override_active: - log_override(atlas, f"Gate 5 bypassed: {'; '.join(f5[:3])}{' ...' if len(f5) > 3 else ''}") - # Override accepts the failures; no append to global failures list - else: - failures.extend(f5) - return len(failures) == 0, failures @@ -209,20 +266,14 @@ def main() -> int: parser = argparse.ArgumentParser(description="Deterministic ship-check for prd-taskmaster pipelines.") parser.add_argument("--dry-run", action="store_true", help="Run all gates but always exit 0. Report goes to stderr. Used by execute-task Step 9 as a per-task predicate.") - parser.add_argument("--override", type=str, default=None, - help=f"Bypass Gate 5 (exit codes) if value equals {OVERRIDE_TOKEN}. Logged to execute-log.jsonl.") parser.add_argument("--cwd", type=str, default=None, help="Project root (defaults to current working directory).") args = parser.parse_args() repo_root = Path(args.cwd).resolve() if args.cwd else Path.cwd() - override_active = args.override == OVERRIDE_TOKEN - if args.override is not None and not override_active: - print("FAIL: --override value does not match expected token", file=sys.stderr) - return 2 try: - ok, failures = run_all_gates(repo_root, override_active=override_active) + ok, failures = run_all_gates(repo_root) except Exception as exc: # noqa: BLE001 — top-level guard print(f"FAIL: ship-check script error: {exc!r}", file=sys.stderr) return 2 @@ -237,8 +288,7 @@ def main() -> int: return 0 if ok: - suffix = " [OVERRIDE]" if override_active else "" - print(f"SHIP_CHECK_OK{suffix}") + print("SHIP_CHECK_OK") return 0 for f in failures: diff --git a/tests/core/test_oracle_bridge.py b/tests/core/test_oracle_bridge.py new file mode 100644 index 0000000..9cbde52 --- /dev/null +++ b/tests/core/test_oracle_bridge.py @@ -0,0 +1,176 @@ +"""Oracle bridge tests — all subprocess calls are mocked; no real CLI/podman invoked.""" + +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from prd_taskmaster.oracle_bridge import grade_card, grade_task, OracleCardError + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _write_graded_card(path: Path, extra: dict | None = None) -> Path: + """Write a minimal valid Graded Card (has the required 'grading' key).""" + card = {"id": "C-001", "grading": {"criteria": "all tests pass"}} + if extra: + card.update(extra) + path.write_text(json.dumps(card)) + return path + + +def _fake_proc(stdout: str = "", stderr: str = "", returncode: int = 0): + """Return a fake subprocess.CompletedProcess-like object.""" + return SimpleNamespace(stdout=stdout, stderr=stderr, returncode=returncode) + + +def _common_args(tmp_path: Path): + """Return common keyword args for grade_card, pointing at tmp_path sub-dirs.""" + return dict( + repo_path=tmp_path / "repo", + commit_sha="abc123", + held_root=tmp_path / "held", + evidence_dir=tmp_path / "evidence", + ledger_dir=tmp_path / "ledger", + ) + + +# --------------------------------------------------------------------------- +# Test 1: PASS round-trip + argv mapping +# --------------------------------------------------------------------------- + +def test_pass_roundtrip_and_argv_mapping(tmp_path, monkeypatch): + """grade_card returns ("PASS", full_dict) and passes correct CLI args.""" + card_path = _write_graded_card(tmp_path / "card.json") + args = _common_args(tmp_path) + + captured_cmd = [] + + def fake_run(cmd, **kwargs): + captured_cmd.extend(cmd) + return _fake_proc(stdout='{"verdict":"PASS","exitCode":0,"ledgerEventId":"x"}') + + monkeypatch.setattr("prd_taskmaster.oracle_bridge.subprocess.run", fake_run) + + oracle_cmd = ["atlas"] + verdict, detail = grade_card(card_path=card_path, oracle_cmd=oracle_cmd, **args) + + assert verdict == "PASS" + assert detail == {"verdict": "PASS", "exitCode": 0, "ledgerEventId": "x"} + + # argv mapping checks + assert "oracle" in captured_cmd + assert "grade" in captured_cmd + assert "--repo" in captured_cmd + assert str(args["repo_path"]) in captured_cmd + assert "--commit" in captured_cmd + assert "abc123" in captured_cmd + assert "--card" in captured_cmd + assert str(card_path) in captured_cmd + assert "--held" in captured_cmd + assert str(args["held_root"]) in captured_cmd + assert "--evidence" in captured_cmd + assert str(args["evidence_dir"]) in captured_cmd + assert "--ledger" in captured_cmd + assert str(args["ledger_dir"]) in captured_cmd + + +# --------------------------------------------------------------------------- +# Test 2: Missing 'grading' block → OracleCardError +# --------------------------------------------------------------------------- + +def test_missing_grading_block_raises(tmp_path, monkeypatch): + """A card without a 'grading' key must raise OracleCardError (before shelling out).""" + card_path = tmp_path / "card_no_grading.json" + card_path.write_text(json.dumps({"id": "C-002", "title": "Some task"})) + + # subprocess.run should NOT be called — we assert by not patching it; + # if the bridge reaches it the test will fail for a different reason. + # But to be clean and not accidentally run a real 'atlas' binary, patch it too. + def fake_run(cmd, **kwargs): + raise AssertionError("subprocess.run should not be called when grading block absent") + + monkeypatch.setattr("prd_taskmaster.oracle_bridge.subprocess.run", fake_run) + + with pytest.raises(OracleCardError, match="no 'grading' block"): + grade_card(card_path=card_path, **_common_args(tmp_path)) + + +# --------------------------------------------------------------------------- +# Test 3: Fail-closed on unparseable output (non-JSON stdout) +# --------------------------------------------------------------------------- + +def test_fail_closed_on_unparseable_output(tmp_path, monkeypatch): + """Non-JSON stdout (e.g. 'podman: command not found') yields ("FAIL", {...}), never raises.""" + card_path = _write_graded_card(tmp_path / "card.json") + + def fake_run(cmd, **kwargs): + return _fake_proc(stdout="podman: command not found", returncode=127) + + monkeypatch.setattr("prd_taskmaster.oracle_bridge.subprocess.run", fake_run) + + verdict, detail = grade_card(card_path=card_path, oracle_cmd=["atlas"], **_common_args(tmp_path)) + + assert verdict == "FAIL" + assert "error" in detail + assert detail["returncode"] == 127 + + +# --------------------------------------------------------------------------- +# Test 4: Fail-closed on CLI invocation error (FileNotFoundError) +# --------------------------------------------------------------------------- + +def test_fail_closed_on_invocation_error(tmp_path, monkeypatch): + """FileNotFoundError from subprocess.run yields ("FAIL", {...}), never raises.""" + card_path = _write_graded_card(tmp_path / "card.json") + + def fake_run(cmd, **kwargs): + raise FileNotFoundError("atlas: No such file or directory") + + monkeypatch.setattr("prd_taskmaster.oracle_bridge.subprocess.run", fake_run) + + verdict, detail = grade_card(card_path=card_path, oracle_cmd=["atlas"], **_common_args(tmp_path)) + + assert verdict == "FAIL" + assert "error" in detail + assert "oracle CLI invocation failed" in detail["error"] + + +# --------------------------------------------------------------------------- +# Test 5: FAIL verdict passthrough +# --------------------------------------------------------------------------- + +def test_fail_verdict_passthrough(tmp_path, monkeypatch): + """A FAIL verdict from the oracle is passed through as ("FAIL", detail).""" + card_path = _write_graded_card(tmp_path / "card.json") + + def fake_run(cmd, **kwargs): + return _fake_proc(stdout='{"verdict":"FAIL"}', returncode=1) + + monkeypatch.setattr("prd_taskmaster.oracle_bridge.subprocess.run", fake_run) + + verdict, detail = grade_card(card_path=card_path, oracle_cmd=["atlas"], **_common_args(tmp_path)) + + assert verdict == "FAIL" + assert detail == {"verdict": "FAIL"} + + +# --------------------------------------------------------------------------- +# Test 6: grade_task missing card → OracleCardError +# --------------------------------------------------------------------------- + +def test_grade_task_missing_card_raises(tmp_path): + """grade_task raises OracleCardError when no CDD card exists for the given task.""" + # tmp_path has no .atlas-ai/cdd/task-1.json + with pytest.raises(OracleCardError, match="no CDD card at"): + grade_task( + repo_root=tmp_path, + task_id=1, + commit_sha="deadbeef", + held_root=tmp_path / "held", + evidence_dir=tmp_path / "evidence", + ledger_dir=tmp_path / "ledger", + ) diff --git a/tests/core/test_oracle_dogfood.py b/tests/core/test_oracle_dogfood.py new file mode 100644 index 0000000..2463c95 --- /dev/null +++ b/tests/core/test_oracle_dogfood.py @@ -0,0 +1,275 @@ +"""Cross-repo DOGFOOD acceptance test — the capstone of Slice 1. + +This is a REAL end-to-end proof (no mocks) that the unfakable Atlas oracle gates +a genuine `prd-taskmaster` ship-check. For each DONE task, the standalone +`skel/ship-check.py` (Gate 5) shells the `atlas oracle grade` CLI, which: + + 1. checks out the submitted commit into a throwaway detached worktree, + 2. OVERLAYS the operator-held tests over the submitter's tree (the submitter's + own copy of the graded path is REPLACED by the operator's held copy), + 3. re-executes the card's grading command inside a digest-pinned podman sandbox, + 4. derives PASS iff exit 0, and writes a tamper-evident ledger event. + +The acceptance criterion is the reward-hack test: a submitter who ships a +`grade.sh` that always `exit 0` (a cheat that would pass if the submitter +controlled grading) does NOT ship, because the operator-held `grade.sh` +(`exit 1`) is overlaid and re-executed by the oracle. The cheat never touches +the verdict. + +──────────────────────────────────────────────────────────────────────────────── +ATLAS_ORACLE_CMD + The oracle CLI lives in the SPINE worktree (a separate monorepo). Its + workspace packages declare `exports: "./src/index.ts"`, so the compiled + `apps/cli/dist/index.js` cannot be run by bare `node` — it would resolve the + workspace deps to their TypeScript sources. We therefore invoke the CLI the + same way the spine's own vitest suite does: through the `tsx` executable on + the CLI source entrypoint. This requires NO edits to the spine repo and runs + the identical code path. The resulting command is: + + ATLAS_ORACLE_CMD="/node_modules/.bin/tsx /apps/cli/src/index.ts" + + ship-check.py shlex-splits ATLAS_ORACLE_CMD and appends `oracle grade ...`. + +SLICE-2 HARDENING NOTE + The Graded Card's `contentHash` is a syntactically valid placeholder here. + `gradeSubmission` does NOT re-verify the contentHash against the card body in + Slice 1 (it is only echoed into the ledger payload), so a placeholder is + accepted. Slice 2 should recompute and verify the contentHash before grading + so a tampered card body cannot be graded under a stale hash. +""" +from __future__ import annotations + +import hashlib +import json +import os +import shlex +import shutil +import subprocess +from pathlib import Path + +import pytest + +# ── Repo paths ──────────────────────────────────────────────────────────────── +ENGINE_ROOT = Path(__file__).resolve().parents[2] +SKEL_SHIP_CHECK = ENGINE_ROOT / "skel" / "ship-check.py" + +# ── Spine (oracle CLI) ──────────────────────────────────────────────────────── +# The oracle invocation prefix is supplied via the ATLAS_ORACLE_CMD env var so +# this test stays portable (no hardcoded paths). Point it at your built atlas CLI, +# e.g. ATLAS_ORACLE_CMD="/node_modules/.bin/tsx /apps/cli/src/index.ts" +# ship-check.py shlex-splits it and appends `oracle grade ...`. +CLI_CMD = os.environ.get("ATLAS_ORACLE_CMD", "") + +ALPINE_REF = "docker.io/library/alpine:3.20" + + +# ── Capability probes ───────────────────────────────────────────────────────── +def has_podman() -> bool: + return shutil.which("podman") is not None + + +def has_oracle_cli() -> bool: + return bool(CLI_CMD.strip()) + + +def resolve_alpine_digest() -> str: + """Pull alpine:3.20 and return its `sha256:...` manifest digest.""" + subprocess.run( + ["podman", "pull", ALPINE_REF], + check=True, capture_output=True, text=True, + ) + proc = subprocess.run( + ["podman", "image", "inspect", ALPINE_REF, "--format", "{{.Digest}}"], + check=True, capture_output=True, text=True, + ) + digest = proc.stdout.strip() + assert digest.startswith("sha256:"), f"unexpected digest {digest!r}" + return digest + + +SKIP_REASON = "requires podman + ATLAS_ORACLE_CMD pointing at a built atlas oracle CLI" +podman_and_cli = pytest.mark.skipif( + not (has_podman() and has_oracle_cli()), reason=SKIP_REASON +) + + +# ── Project fixture ─────────────────────────────────────────────────────────── +def _git(args: list[str], cwd: Path) -> str: + return subprocess.run( + ["git", *args], cwd=cwd, check=True, capture_output=True, text=True + ).stdout.strip() + + +def make_project(tmp_path: Path, *, submitter_grade: str, operator_grade: str): + """Construct a real mini-project laid out exactly as ship-check's gate_oracle + expects, and return (root, head_sha). + + The submitter commits their own `grade.sh` (submitter_grade). The operator's + held copy (operator_grade) lives — uncommitted, on disk only — under + .atlas-ai/held-out/ and is what the oracle overlays + re-executes. + """ + root = tmp_path + digest = resolve_alpine_digest() + + # 1. Submitter's working tree: grade.sh committed at root. + (root / "grade.sh").write_text(submitter_grade) + _git(["init", "."], root) + _git(["config", "user.email", "dogfood@atlas.test"], root) + _git(["config", "user.name", "dogfood"], root) + _git(["add", "-A"], root) + # core.hooksPath=/dev/null mirrors the oracle's own git invocations and keeps + # any ambient git hooks out of the committed state. + _git(["-c", "core.hooksPath=/dev/null", "commit", "-m", "work"], root) + head_sha = _git(["rev-parse", "HEAD"], root) + + atlas = root / ".atlas-ai" + + # 2. Operator-held test bundle (on disk; need not be committed). + held = atlas / "held-out" + held.mkdir(parents=True) + (held / "grade.sh").write_text(operator_grade) + operator_sha256 = hashlib.sha256(operator_grade.encode()).hexdigest() + + # 3. Graded Card. + cdd = atlas / "cdd" + cdd.mkdir(parents=True) + card = { + "id": "C-001", + "taskId": 1, + "title": "dogfood", + "given": ["g"], + "when": ["w"], + "then": [{ + "index": 1, + "statement": "exit 0", + "evidenceTier": "A", + "evidenceKind": "command-output", + }], + "author": {"kind": "human", "id": "op"}, + "createdAt": "2026-06-16T00:00:00.000Z", + # SLICE-2: placeholder contentHash — gradeSubmission does not re-verify it + # in Slice 1 (see module docstring). + "contentHash": "sha256:" + "0" * 64, + "frozenAt": "2026-06-16T00:00:00.000Z", + "grading": { + "command": ["sh", "grade.sh"], + "heldOutTests": [{"path": "grade.sh", "sha256": operator_sha256}], + "gradedPaths": ["grade.sh"], + "baseImage": {"ref": ALPINE_REF, "digest": digest}, + "env": { + "LANG": "C", "LC_ALL": "C", "TZ": "UTC", + "SOURCE_DATE_EPOCH": 0, "seed": 0, + "parallelism": 1, "cpuClass": "x86-64-v2", + }, + "timeoutMs": 60000, + }, + } + (cdd / "task-1.json").write_text(json.dumps(card)) + + # 4. Gates 1-4 scaffolding so the ORACLE gate (Gate 5) decides the verdict. + state = atlas / "state" + state.mkdir(parents=True) + (state / "pipeline.json").write_text(json.dumps({"current_phase": "EXECUTE"})) + + tasks_dir = root / ".taskmaster" / "tasks" + tasks_dir.mkdir(parents=True) + (tasks_dir / "tasks.json").write_text( + json.dumps({"master": {"tasks": [{"id": 1, "status": "done"}]}}) + ) + docs = root / ".taskmaster" / "docs" + docs.mkdir(parents=True) + (docs / "plan.md").write_text("# Plan\n") + + # Oracle output dirs (created by the CLI too, but explicit is clearer). + (atlas / "evidence").mkdir(parents=True, exist_ok=True) + (atlas / "ledger").mkdir(parents=True, exist_ok=True) + + return root, head_sha + + +def run_shipcheck(root: Path) -> subprocess.CompletedProcess: + """Run the REAL skel/ship-check.py as a subprocess with ATLAS_ORACLE_CMD set.""" + return subprocess.run( + ["python3", str(SKEL_SHIP_CHECK), "--cwd", str(root)], + capture_output=True, + text=True, + env={**os.environ, "ATLAS_ORACLE_CMD": CLI_CMD}, + ) + + +def _read_ledger_event(root: Path) -> dict: + ledger = root / ".atlas-ai" / "ledger" + events = sorted(ledger.glob("*.json")) + assert events, f"no ledger event written under {ledger}" + return json.loads(events[-1].read_text()) + + +# ── Test 1: genuine pass ships ──────────────────────────────────────────────── +@podman_and_cli +def test_genuine_pass_ships(tmp_path): + # submitter_grade is irrelevant — the operator-held grade.sh is overlaid. + root, _head = make_project(tmp_path, submitter_grade="exit 1\n", operator_grade="exit 0\n") + + proc = run_shipcheck(root) + + assert "SHIP_CHECK_OK" in proc.stdout, ( + f"expected SHIP_CHECK_OK; rc={proc.returncode}\nSTDOUT:\n{proc.stdout}\nSTDERR:\n{proc.stderr}" + ) + assert proc.returncode == 0, f"rc={proc.returncode}\nSTDERR:\n{proc.stderr}" + + # The oracle recorded a PASS verdict in the tamper-evident ledger. + event = _read_ledger_event(root) + assert event["type"] == "verification.completed" + assert event["payload"]["verdict"] == "PASS", event["payload"] + + +# ── Test 2: reward hack blocked (THE acceptance criterion) ──────────────────── +@podman_and_cli +def test_reward_hack_blocked(tmp_path): + # The submitter ships a cheat: grade.sh that always passes. The operator-held + # grade.sh fails. Because the oracle overlays + re-executes the operator copy, + # the cheat does NOT ship. + root, _head = make_project(tmp_path, submitter_grade="exit 0\n", operator_grade="exit 1\n") + + proc = run_shipcheck(root) + + assert "SHIP_CHECK_OK" not in proc.stdout, ( + f"REWARD HACK SHIPPED — cheat was not blocked!\nSTDOUT:\n{proc.stdout}" + ) + assert proc.returncode != 0, "cheat must produce a non-zero exit" + # The blocked-FAIL is surfaced on stderr for task 1. + assert "oracle" in proc.stderr.lower() and "1" in proc.stderr, ( + f"expected an oracle FAIL for task 1 on stderr\nSTDERR:\n{proc.stderr}" + ) + + # And the ledger records the FAIL verdict — the operator-held test was run. + event = _read_ledger_event(root) + assert event["payload"]["verdict"] == "FAIL", event["payload"] + + +# ── Test 3: ledger integrity ────────────────────────────────────────────────── +@podman_and_cli +def test_ledger_integrity(tmp_path): + """After a genuine PASS, the tamper-evident ledger verifies clean. + + The spine CLI exposes `ledger verify ` (positional arg). If that + subcommand is unavailable we fall back to re-parsing the event file. + """ + root, _head = make_project(tmp_path, submitter_grade="exit 1\n", operator_grade="exit 0\n") + proc = run_shipcheck(root) + assert "SHIP_CHECK_OK" in proc.stdout, proc.stderr + + ledger_dir = root / ".atlas-ai" / "ledger" + verify = subprocess.run( + shlex.split(CLI_CMD) + ["ledger", "verify", str(ledger_dir)], + capture_output=True, text=True, + ) + if verify.returncode == 0 and verify.stdout.strip(): + result = json.loads(verify.stdout) + assert result.get("ok") is True, f"ledger verify reported {result!r}" + assert result.get("eventCount", 0) >= 1, result + else: + # No usable `ledger verify` subcommand — don't fail the task over it; + # confirm the event file at least parses and records the PASS. + event = _read_ledger_event(root) + assert event["payload"]["verdict"] == "PASS", event["payload"] diff --git a/tests/core/test_ship_check.py b/tests/core/test_ship_check.py index aee49bf..4d2c2ba 100644 --- a/tests/core/test_ship_check.py +++ b/tests/core/test_ship_check.py @@ -12,14 +12,26 @@ Gate 3 — for each task id, a CDD card at .atlas-ai/cdd/task-.json (or a combined card whose filename contains the id). Gate 4 — plan at .taskmaster/docs/plan.md OR docs/superpowers/plans/*.md. - Gate 5 (HARD) — no non-zero "Exit status N" in any .atlas-ai/evidence/ file. - Bypass via --override SHIP_CHECK_OVERRIDE_ADMIN. - Success → stdout exactly "SHIP_CHECK_OK\n" (+ " [OVERRIDE]"), exit 0. + Gate 5 (HARD, ORACLE) — every DONE task is RE-GRADED by the atlas oracle CLI + (shelled out; configurable via ATLAS_ORACLE_CMD). FAIL-CLOSED: a + missing card, a card with no grading block, a CLI crash, unparseable + output, or any non-PASS verdict BLOCKS the ship. The old fakable + "no non-zero Exit status N in evidence" grep and the self-grantable + --override SHIP_CHECK_OVERRIDE_ADMIN token are BOTH GONE (see the + dedicated coverage in test_ship_check_oracle.py). + Success → stdout exactly "SHIP_CHECK_OK\n", exit 0. Failure → nothing on stdout, FAIL detail on stderr, exit 1. + +NOTE: these tests drive the script as a real subprocess, so the oracle gate is +satisfied by a real fake-oracle command (a tiny executable shell script) wired +through ATLAS_ORACLE_CMD. Pure-monkeypatch oracle coverage lives in +test_ship_check_oracle.py. """ from __future__ import annotations import json +import os +import stat import subprocess from pathlib import Path @@ -31,18 +43,41 @@ SHIP_CHECK = REPO_ROOT / "skel" / "ship-check.py" -def _run(cwd: Path, *extra: str) -> subprocess.CompletedProcess[str]: +def _fake_oracle(tmp_path: Path, verdict: str = "PASS") -> str: + """Write an executable fake `atlas` that emits {"verdict": } for an + `oracle grade` invocation. Return an ATLAS_ORACLE_CMD value pointing at it.""" + script = tmp_path / "fake_atlas.sh" + script.write_text( + "#!/bin/sh\n" + 'for a in "$@"; do\n' + ' if [ "$a" = "grade" ]; then\n' + f' printf \'{{"verdict":"{verdict}"}}\'\n' + " exit 0\n" + " fi\n" + "done\n" + "exit 0\n" + ) + script.chmod(script.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + return f"sh {script}" + + +def _run(cwd: Path, *extra: str, oracle_cmd: str | None = None) -> subprocess.CompletedProcess[str]: + env = dict(os.environ) + if oracle_cmd is not None: + env["ATLAS_ORACLE_CMD"] = oracle_cmd return subprocess.run( ["python3", str(SHIP_CHECK), *extra], cwd=str(cwd), capture_output=True, text=True, check=False, + env=env, ) def _green(tmp_path: Path) -> None: - """Build an all-gates-green project tree under tmp_path.""" + """Build an all-gates-green project tree under tmp_path. CDD cards carry a + `grading` block so the oracle gate can re-grade them.""" atlas = tmp_path / ".atlas-ai" (atlas / "state").mkdir(parents=True) (atlas / "state" / "pipeline.json").write_text( @@ -64,8 +99,9 @@ def _green(tmp_path: Path) -> None: ) cdd = atlas / "cdd" cdd.mkdir(parents=True) - (cdd / "task-1.json").write_text(json.dumps({"id": 1})) - (cdd / "task-2.json").write_text(json.dumps({"id": 2})) + grading = {"grading": {"command": ["sh", "grade.sh"]}} + (cdd / "task-1.json").write_text(json.dumps({"id": 1, **grading})) + (cdd / "task-2.json").write_text(json.dumps({"id": 2, **grading})) docs = tmp_path / ".taskmaster" / "docs" docs.mkdir(parents=True) (docs / "plan.md").write_text("# Plan\n") @@ -80,7 +116,7 @@ def test_ship_check_fails_on_empty_state(tmp_path: Path) -> None: def test_ship_check_passes_on_all_gates_green(tmp_path: Path) -> None: _green(tmp_path) - r = _run(tmp_path) + r = _run(tmp_path, oracle_cmd=_fake_oracle(tmp_path, "PASS")) assert r.returncode == 0, f"stderr={r.stderr!r} stdout={r.stdout!r}" assert "SHIP_CHECK_OK" in r.stdout @@ -118,25 +154,27 @@ def test_ship_check_fails_when_done_task_has_no_cdd_card(tmp_path: Path) -> None assert "task 2: no CDD card" in r.stderr -def test_ship_check_gate5_blocks_on_nonzero_exit_and_override_passes(tmp_path: Path) -> None: - """Gate 5 (HARD): a non-zero 'Exit status N' in evidence blocks; the - override token bypasses it.""" +def test_ship_check_gate5_oracle_fail_blocks_and_override_is_gone(tmp_path: Path) -> None: + """Gate 5 (HARD, ORACLE): an oracle FAIL verdict blocks the ship, and the + old self-grantable --override token no longer exists (argparse rejects it).""" _green(tmp_path) - evidence = tmp_path / ".atlas-ai" / "evidence" - evidence.mkdir(parents=True) - (evidence / "run.log").write_text("pnpm test\nExit status 1\n") - # Without override → blocked. - r = _run(tmp_path) + # An oracle FAIL verdict → blocked, nothing on stdout. + r = _run(tmp_path, oracle_cmd=_fake_oracle(tmp_path, "FAIL")) assert r.returncode != 0 assert "SHIP_CHECK_OK" not in r.stdout - assert "Exit status 1" in r.stderr - - # With the override token → passes, OVERRIDE suffix on stdout. - r2 = _run(tmp_path, "--override", "SHIP_CHECK_OVERRIDE_ADMIN") - assert r2.returncode == 0, f"stderr={r2.stderr!r} stdout={r2.stdout!r}" - assert "SHIP_CHECK_OK" in r2.stdout - assert "[OVERRIDE]" in r2.stdout + assert "oracle verdict FAIL" in r.stderr + + # The override token is GONE: argparse rejects the unknown flag (exit 2) + # and never emits SHIP_CHECK_OK — there is no bypass path. + r2 = _run( + tmp_path, + "--override", + "SHIP_CHECK_OVERRIDE_ADMIN", + oracle_cmd=_fake_oracle(tmp_path, "PASS"), + ) + assert r2.returncode == 2 + assert "SHIP_CHECK_OK" not in r2.stdout # ─── Python API agreement (prd_taskmaster.shipcheck.run_ship_check) ────────── @@ -171,6 +209,6 @@ def test_ship_check_passes_with_flat_tasks_format(tmp_path): (tmp_path / ".taskmaster" / "tasks" / "tasks.json").write_text(json.dumps( {"tasks": [{"id": 1, "status": "done"}, {"id": 2, "status": "done"}]} )) - r = _run(tmp_path) + r = _run(tmp_path, oracle_cmd=_fake_oracle(tmp_path, "PASS")) assert r.returncode == 0, f"stderr={r.stderr!r}" assert "SHIP_CHECK_OK" in r.stdout diff --git a/tests/core/test_ship_check_oracle.py b/tests/core/test_ship_check_oracle.py new file mode 100644 index 0000000..ebd7fc5 --- /dev/null +++ b/tests/core/test_ship_check_oracle.py @@ -0,0 +1,200 @@ +"""Oracle-gate contract for the standalone skel/ship-check.py (Gate 5 cutover). + +Gate 5 was a FAKABLE deterministic grep over .atlas-ai/evidence/ that SILENTLY +PASSED when no evidence existed, and was bypassable with a self-grantable +--override SHIP_CHECK_OVERRIDE_ADMIN token. This suite proves the new contract: + + * For every DONE task, the gate RE-EXECUTES the CDD card's grading via the + `atlas oracle grade` CLI (shelled out — the script stays stdlib-only). + * FAIL-CLOSED: missing card, missing grading block, CLI crash, unparseable + output, or a non-PASS verdict all BLOCK the ship. ONLY verdict=="PASS" + ships a task. + * The override token is GONE — there is no self-grantable bypass path. + +The script is loaded by file path via importlib because it ships into user +projects as a standalone `.atlas-ai/ship-check.py` (no prd_taskmaster import). +""" +from __future__ import annotations + +import importlib.util +import json +import sys +import types +from pathlib import Path + +import pytest + +SHIP = Path(__file__).resolve().parents[2] / "skel" / "ship-check.py" + + +def load(): + spec = importlib.util.spec_from_file_location("ship_check_mod", SHIP) + mod = importlib.util.module_from_spec(spec) + # Don't write skel/__pycache__/*.pyc — the packaging tarball must stay clean. + prev = sys.dont_write_bytecode + sys.dont_write_bytecode = True + try: + spec.loader.exec_module(mod) + finally: + sys.dont_write_bytecode = prev + return mod + + +def setup_project(tmp_path: Path, task_ids, with_cdd: bool = True) -> None: + """Write a tree where Gates 1-4 PASS so the ORACLE gate decides the verdict. + + pipeline.json current_phase=EXECUTE (Gate 1), every task done (Gate 2), + a CDD card per task when with_cdd (Gate 3), and a plan.md (Gate 4). + """ + atlas = tmp_path / ".atlas-ai" + (atlas / "state").mkdir(parents=True) + (atlas / "state" / "pipeline.json").write_text( + json.dumps({"current_phase": "EXECUTE"}) + ) + tm = tmp_path / ".taskmaster" / "tasks" + tm.mkdir(parents=True) + (tm / "tasks.json").write_text( + json.dumps({"master": {"tasks": [{"id": i, "status": "done"} for i in task_ids]}}) + ) + docs = tmp_path / ".taskmaster" / "docs" + docs.mkdir(parents=True) + (docs / "plan.md").write_text("# Plan\n") + if with_cdd: + cdd = atlas / "cdd" + cdd.mkdir(parents=True) + for i in task_ids: + (cdd / f"task-{i}.json").write_text( + json.dumps({"id": f"C-00{i}", "grading": {"command": ["sh", "grade.sh"]}}) + ) + + +def _fake_run_factory(grade_stdout: str, grade_returncode: int = 0): + """Build a fake subprocess.run dispatching on the command. + + rev-parse → a fixed HEAD sha; oracle grade → the canned (stdout, returncode). + """ + + def fake_run(cmd, *args, **kwargs): + joined = " ".join(str(c) for c in cmd) if isinstance(cmd, (list, tuple)) else str(cmd) + if "rev-parse" in joined: + return types.SimpleNamespace(stdout="deadbeef\n", stderr="", returncode=0) + if "oracle" in joined and "grade" in joined: + return types.SimpleNamespace(stdout=grade_stdout, stderr="", returncode=grade_returncode) + raise AssertionError(f"unexpected subprocess command: {joined}") + + return fake_run + + +# ─── 1. Oracle PASS → ships ─────────────────────────────────────────────────── + + +def test_oracle_pass_ships(tmp_path, monkeypatch, capsys): + mod = load() + setup_project(tmp_path, [1, 2], with_cdd=True) + monkeypatch.setattr(mod.subprocess, "run", _fake_run_factory('{"verdict":"PASS"}')) + + ok, failures = mod.run_all_gates(tmp_path) + assert ok is True, f"failures={failures!r}" + assert failures == [] + + # main() end-to-end prints exactly SHIP_CHECK_OK and exits 0. + monkeypatch.setattr(mod.sys, "argv", ["ship-check.py", "--cwd", str(tmp_path)]) + rc = mod.main() + out = capsys.readouterr() + assert rc == 0 + assert out.out.strip() == "SHIP_CHECK_OK" + + +# ─── 2. Oracle FAIL → blocks ────────────────────────────────────────────────── + + +def test_oracle_fail_blocks(tmp_path, monkeypatch, capsys): + mod = load() + setup_project(tmp_path, [1, 2], with_cdd=True) + monkeypatch.setattr(mod.subprocess, "run", _fake_run_factory('{"verdict":"FAIL"}')) + + ok, failures = mod.run_all_gates(tmp_path) + assert ok is False + assert any("oracle" in f.lower() and ("1" in f or "2" in f) for f in failures), failures + + monkeypatch.setattr(mod.sys, "argv", ["ship-check.py", "--cwd", str(tmp_path)]) + rc = mod.main() + out = capsys.readouterr() + assert rc == 1 + assert "SHIP_CHECK_OK" not in out.out + + +# ─── 3. The silent-pass loophole is closed (fail-closed on ambiguity) ────────── + + +def test_silent_pass_is_gone(tmp_path, monkeypatch): + """OLD Gate 5 silently PASSED when no evidence existed. The new gate must + FAIL-CLOSED when the oracle output is not parseable JSON.""" + mod = load() + setup_project(tmp_path, [1], with_cdd=True) + monkeypatch.setattr( + mod.subprocess, "run", _fake_run_factory("podman: not found", grade_returncode=127) + ) + + ok, failures = mod.run_all_gates(tmp_path) + assert ok is False + assert failures, "non-JSON oracle output must produce a failure" + + +# ─── 4. Missing CDD card fails closed ───────────────────────────────────────── + + +def test_missing_cdd_card_fails_closed(tmp_path, monkeypatch): + mod = load() + # Build Gates 1,2,4 green but NO cdd dir/card (with_cdd=False). + setup_project(tmp_path, [1], with_cdd=False) + # The oracle would pass IF it were called — it must never be reached. + monkeypatch.setattr(mod.subprocess, "run", _fake_run_factory('{"verdict":"PASS"}')) + + ok, failures = mod.gate_oracle(tmp_path, [{"id": 1, "status": "done"}], "deadbeef") + assert ok is False + assert any("no CDD card" in f for f in failures), failures + + +# ─── 5. The override token is removed ───────────────────────────────────────── + + +def test_override_token_removed(tmp_path, monkeypatch, capsys): + mod = load() + assert not hasattr(mod, "OVERRIDE_TOKEN") + + setup_project(tmp_path, [1], with_cdd=True) + # Even an attempt to pass the old token must NOT produce SHIP_CHECK_OK. + monkeypatch.setattr(mod.subprocess, "run", _fake_run_factory('{"verdict":"FAIL"}')) + monkeypatch.setattr( + mod.sys, + "argv", + ["ship-check.py", "--override", "SHIP_CHECK_OVERRIDE_ADMIN", "--cwd", str(tmp_path)], + ) + with pytest.raises(SystemExit) as exc: + mod.main() + # argparse rejects the unknown --override argument with exit code 2. + assert exc.value.code == 2 + out = capsys.readouterr() + assert "SHIP_CHECK_OK" not in out.out + + # The token string no longer appears anywhere in the file. + assert "OVERRIDE" not in SHIP.read_text() + + +# ─── 6. Gates 1-4 still enforced (regression) ───────────────────────────────── + + +def test_gates_1_to_4_still_enforced(tmp_path, monkeypatch): + mod = load() + setup_project(tmp_path, [1], with_cdd=True) + # Break Gate 1: pipeline not in EXECUTE. + (tmp_path / ".atlas-ai" / "state" / "pipeline.json").write_text( + json.dumps({"current_phase": "GENERATE"}) + ) + # Oracle would pass — Gate 1 must still block. + monkeypatch.setattr(mod.subprocess, "run", _fake_run_factory('{"verdict":"PASS"}')) + + ok, failures = mod.run_all_gates(tmp_path) + assert ok is False + assert any("current_phase" in f for f in failures), failures