diff --git a/CHANGELOG.md b/CHANGELOG.md index 566efa6..f7bd46b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,25 @@ All notable changes to this project are documented here. Format based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added +- **Independent out-of-band re-execution watcher** (`prd_taskmaster/tournament/watcher.py`; + `watcher-run` / `watcher-status` CLI) — the precondition for ever enabling real + (`--enforce-slash`) forfeiture. It re-adjudicates settled tournament submissions from + **primary evidence** (the claimed commit + the CDD card) without trusting the recorded + verdict: it re-runs the oracle gate, re-derives `sha256(diff base..HEAD)` to catch diff-copy + tamper independently of the live collector, and accumulates a concordance ledger over real + *slash decisions*. A `permit_enforce_slash` gate is **fail-closed**: real slashing is permitted + only when every to-be-slashed submission is independently confirmed, there is **no discrepancy + or abstain anywhere in the job**, and the watcher's historical concordance clears a threshold + over a minimum number of prior decisions. Inability to verify (oracle could not run, no + worktree, failed hash recompute) **abstains** — it is never counted as grounds to slash. +- **Engine-enforced shadow-until-permitted** — `run_tournament` now consults the watcher before + any real `--enforce-slash` settle and **downgrades to shadow** unless the watcher permits the + job (fail-closed on any watcher error). Real AtlasCoin is never burned without an independent + positive confirmation; the shadow-slash default path is unchanged. + ## [5.3.0] — 2026-06-17 The "unfakable done" release: a task ships only when its test genuinely re-ran green AND diff --git a/CLAUDE.md b/CLAUDE.md index 52b23ff..ab3424c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -18,11 +18,59 @@ provable?* What makes Atlas a product, not a prompt pack (the four moats): 1. **Cross-vendor arbitrage** — route work across any harness/API by cost and capability. 2. **Engine-enforced unfakable gates** — validate-tasks, ship-check (`SHIP_CHECK_OK`), - structural tests. If a feature could be a skill, it's not a product feature until the - engine enforces it. + the two-axis "done" (oracle + reachability, below), structural tests. If a feature could + be a skill, it's not a product feature until the engine enforces it. 3. **Persistent vendor-neutral task graph** — tasks.json survives across vendors/sessions. 4. **Cross-vendor cost ledger** — telemetry + economy-report prove the savings. +## Unfakable "done" — the two gates (5.3.0+, the trust backbone) + +A task ships only when **both** axes pass — `done = oracle-PASS AND code WIRED/EXEMPT`. +This is the property the whole product (and the marketplace below) rests on; do not weaken it. + +- **Gate 5 — re-execution oracle.** The binding ship gate is `skel/ship-check.py` + (oracle-backed). It re-runs the operator-held tests at the *claimed commit* in a + network-isolated, digest-pinned **podman** sandbox, so a submitter cannot pass by editing + their own logs. `prd_taskmaster/oracle_bridge.py` maps a CDD card → graded verdict via the + spine's `atlas oracle` CLI; it is **fail-closed** (any ambiguity → FAIL, never PASS). The + self-grantable `SHIP_CHECK_OVERRIDE_ADMIN` bypass was removed. + (`prd_taskmaster/shipcheck.py` is a NON-binding display heuristic only — never add an oracle + call there.) +- **Gate 6 — reachability** (`prd_taskmaster/reachability.py`, `reachability-sweep` CLI). A + green test on an orphan module (imported by nothing) is **blocked** and surfaced as + `⚠ scaffolded`. Verdicts: WIRED (a non-test file imports it) / EXEMPT (declared + `cli:|route:|tool:|hook:|plugin:|dynamic:` scheme, accepted on trust in v1) / ORPHAN (fail). + Read-only at runtime. + +## Tournament marketplace (`prd_taskmaster/tournament/`) + +The trustworthy two-gate signal feeds a **settled tournament**: N executors race one job → +every submission is adjudicated through both gates → the winner is paid in **AtlasCoin** → +a UCB reputation store routes the next job to the cheapest proven-capable executor. + +- `tournament-run` / `tournament-status` CLI; flow is `spawn → collect → adjudicate → settle + → reputation` (`cmd.py`), with anti-sybil slots released in a crash-safe `finally`. +- `collect.py` is the security core — **commit-reveal** (`commit_hash = sha256(diff base..HEAD)`) + defeats the diff-copy attack. `antisybil.py` enforces per-job / per-operator economic caps. +- `goose_backend.py` is the cheap-API racer (one OpenRouter model via `goose run`). +- `watcher.py` (`watcher-run` / `watcher-status`) is the **independent out-of-band re-execution + watcher** — the precondition for real slashing. It re-adjudicates settled submissions from + primary evidence (re-runs the oracle, independently re-derives `sha256(diff base..HEAD)` to + catch diff-copy), abstains when it cannot verify (never a false confirm), and **engine-gates** + real forfeiture: `run_tournament` downgrades `--enforce-slash` to shadow unless the fail-closed + `permit_enforce_slash` confirms the whole job behind a concordance track record. +- **Invariants (enforced + tested — keep them):** **shadow-slash by default**; real slashing only + behind the watcher's fail-closed permit (no AtlasCoin burned without an independent positive + confirmation); honest losers are **always refunded**; **AtlasCoin is conserved** (no mint/burn, + even under account aliasing). + +## The spine (`atlas-protocol`, separate repo) + +The gates shell out to the **spine CLI** (`atlas oracle` / `atlas tournament settle` / ledger), +which lives in the `atlas-protocol` repo (branch `dev`) and runs via `tsx` today. Point +`ATLAS_ORACLE_CMD` at it (e.g. `tsx apps/cli/src/index.ts`) for the real-podman e2e gates. +Packaging the spine for production is on the roadmap — see CHANGELOG `## [5.3.0]`. + ## Backend model (v4.1+) TaskMaster is one pluggable backend, not a prerequisite. *"Atlas speaks TaskMaster @@ -40,6 +88,30 @@ natively — but doesn't need it."* - The normative reference is the **"## Backend operations" table in SKILL.md**. When docs and code disagree, fix the docs to match the table. +## Repository layout + +The engine is **stdlib-only Python** in `prd_taskmaster/`, surfaced two ways that share the +same code — keep them in lockstep: +- `script.py` — thin shim over `prd_taskmaster/cli.py:main` (~40 subcommands; `python3 script.py -h`). +- `mcp-server/server.py` — FastMCP server exposing the same ops as MCP tools (`next_task`, + `set_task_status`, `validate_prd`, …). It needs `pip install -r mcp-server/requirements.txt`; + the CLI does not. + +Key modules: `cli.py` (dispatch) · `backend.py` (5-op backend protocol) · `validation.py` + +`tasks.py` (graded PRD + task graph) · `task_state.py` (atomic next/claim/set-status) · +`reachability.py` (Gate 6) · `oracle_bridge.py` + `shipcheck.py` (Gate 5 wiring) · `economy.py` ++ `reputation.py` (cost ledger + UCB routing) · `fleet.py` / `parallel.py` (fleet waves) · +`tournament/` (the settled marketplace; `tournament/watcher.py` is the independent re-execution +watcher that engine-gates real slashing). The binding ship gate is `skel/ship-check.py`; the +skill/slash commands live in `skills/` (atlas, go, generate, execute-task, execute-fleet, …) and +ship inside the package. `SKILL.md` is the normative behavior spec. Tests split into +`tests/core` + `tests/plugin` (stdlib-only), `tests/mcp` (needs the mcp package), and +`tests/parity` (golden-parity harness, run as a module, not via pytest). + +`AGENTS.md` is the harness-neutral twin of this file — the **same operating contract** for any +agent (Codex, Gemini, the next harness). No test enforces their agreement, so when you change an +operating rule here, mirror it in `AGENTS.md` (and vice-versa). + ## Task execution workflow 1. `python3 script.py next-task` — get the next dependency-ready task @@ -51,16 +123,30 @@ natively — but doesn't need it."* ## Commands ```bash -pytest tests/ -q # full suite — must stay green +pytest tests/ -q # full suite (877); the 4 real-podman e2e gates + # auto-skip unless the env below is set +pytest tests/core tests/plugin -q # stdlib-only path (no mcp pkg); tests/mcp needs + # pip install -r mcp-server/requirements.txt first +pytest tests/core/test_validation.py::test_name -q # run a single test (file::test or -k expr) +ATLAS_ORACLE_CMD="/node_modules/.bin/tsx /apps/cli/src/index.ts" \ + pytest tests/core -q -k "e2e or dogfood" # un-skip the 4 real-podman gates (need the spine CLI + podman) python3 script.py engine-preflight # one-call environment + backend probe python3 script.py validate-tasks # gate before any tasks.json write +python3 script.py reachability-sweep # Gate 6 — orphan-module detection +python3 script.py tournament-run # settled marketplace: spawn→…→reputation +python3 script.py watcher-run --job --card --task --base-ref # out-of-band re-adjudication +python3 script.py watcher-status # watcher concordance + whether real slashing is permitted yet python3 script.py economy-report # cost ledger from .atlas-ai/telemetry.jsonl ``` ## Conventions (engine code) - **Stdlib only** in `prd_taskmaster/` and `mcp-server/server.py` — no new dependencies. - CI enforces stdlib-only imports on a bare runner. + CI (`.github/workflows/ci.yml`) AST-scans the package for non-stdlib imports on a bare runner. +- CI also enforces, and these break easily: **all version strings must agree** — + `prd_taskmaster/__init__.py`, `package.json`, `.claude-plugin/plugin.json`, `install.sh`, plus a + matching `CHANGELOG.md` entry (bump them together); and the **native-no-keys** path must still + resolve `backend=native, ai_ops=agent` with no TaskMaster binary and no API keys. - Errors return FR-28-safe dicts; library code never `sys.exit`s. - All `.atlas-ai/*.jsonl` writes go through the shared flock-guarded append (`economy.append_telemetry` pattern). diff --git a/prd_taskmaster/cli.py b/prd_taskmaster/cli.py index e9a8fa0..41a832a 100644 --- a/prd_taskmaster/cli.py +++ b/prd_taskmaster/cli.py @@ -26,7 +26,12 @@ from prd_taskmaster import fleet, parallel, task_state from prd_taskmaster.lib import _detect_taskmaster_method from prd_taskmaster.reachability_cmd import cmd_reachability_sweep -from prd_taskmaster.tournament.cmd import cmd_tournament_run, cmd_tournament_status +from prd_taskmaster.tournament.cmd import ( + cmd_tournament_run, + cmd_tournament_status, + cmd_watcher_run, + cmd_watcher_status, +) def _backend_source() -> str: @@ -419,6 +424,26 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument("--reputation-path", default=None, help="Path to reputation.jsonl (default: .atlas-ai/reputation.jsonl)") p.add_argument("--operators-path", default=None, help="Path to operators.json (default: .atlas-ai/tournament/operators.json)") + # watcher-run — independent out-of-band re-adjudication of a settled job + p = sub.add_parser( + "watcher-run", + help="Re-adjudicate a settled tournament job out-of-band; append concordance; report the real-slash permit", + ) + p.add_argument("--job", required=True, help="Path to the settled job dir (contains submissions.json)") + p.add_argument("--card", required=True, help="Path to the CDD card JSON") + p.add_argument("--task", required=True, help="Task id (e.g. 7 or 1.2)") + p.add_argument("--base-ref", required=True, help="Fork-point git SHA the diff range is measured from") + p.add_argument("--repo-root", default=".", help="Repo containing the racer commits (default: .)") + p.add_argument("--held-root", default=None, help="Held root for the oracle gate (default: .atlas-ai/cdd)") + p.add_argument("--ledger-path", default=None, help="Watcher ledger path (default: .atlas-ai/tournament/watcher.jsonl)") + + # watcher-status — report watcher concordance + real-slash readiness + p = sub.add_parser( + "watcher-status", + help="Show the watcher's historical concordance and whether real slashing is permitted yet", + ) + p.add_argument("--ledger-path", default=None, help="Watcher ledger path (default: .atlas-ai/tournament/watcher.jsonl)") + # status — render progress panels p = sub.add_parser("status", help="Render Atlas progress panels for the current phase") p.add_argument("--phase", default=None, help="Render a specific phase instead of the current one") @@ -476,6 +501,8 @@ def cmd_status(args) -> None: "reachability-sweep": cmd_reachability_sweep, "tournament-run": cmd_tournament_run, "tournament-status": cmd_tournament_status, + "watcher-run": cmd_watcher_run, + "watcher-status": cmd_watcher_status, "economy-report": cmd_economy_report, "context-pack": cmd_context_pack, "feedback-add": cmd_feedback_add, diff --git a/prd_taskmaster/tournament/cmd.py b/prd_taskmaster/tournament/cmd.py index df96331..e7c8393 100644 --- a/prd_taskmaster/tournament/cmd.py +++ b/prd_taskmaster/tournament/cmd.py @@ -33,6 +33,7 @@ import argparse import json +import subprocess import sys from pathlib import Path from typing import Any, Callable, Optional @@ -47,9 +48,28 @@ FakeClock, ) from prd_taskmaster.tournament.adjudicate import adjudicate_job, settle_job +from prd_taskmaster.tournament import watcher as _watcher from prd_taskmaster.reputation import record_tournament, summarize_reputation, _winner_id +def _resolve_repo_root(cwd: "str | None" = None, default: str = ".") -> str: + """Resolve the git repo root for the watcher's re-execution worktrees. + + Falls back to *default* (never raises) when not in a git repo, so the + watcher gate degrades gracefully rather than crashing the settle path. + """ + try: + proc = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], + cwd=cwd, capture_output=True, text=True, timeout=10, + ) + if proc.returncode == 0 and proc.stdout.strip(): + return proc.stdout.strip() + except Exception: # noqa: BLE001 + pass + return default + + # ─── Core orchestration function ───────────────────────────────────────────── @@ -74,6 +94,10 @@ def run_tournament( now: str, window_s: float = 120.0, enforce_slash: bool = False, + repo_root: str = ".", + watcher_ledger_path: "str | Path | None" = None, + _re_adjudicate: "Optional[Callable[..., dict]]" = None, + _permit: "Optional[Callable[..., dict]]" = None, _spawn_fn: "Callable[[Any], dict]" = default_launcher_adapter, _inbox_read: "Callable[..., list[dict]]" = default_inbox_adapter, _dispatch_reveal: "Callable[..., Optional[dict]]" = default_reveal_adapter, @@ -290,8 +314,38 @@ def run_tournament( job_poster=job_poster, ) + # ── Step 5b: Watcher gate on REAL slashing (engine-enforced) ───── + # Shadow-slash only until the independent out-of-band watcher PERMITS + # real forfeiture for THIS job. Fail-closed: a watcher error, a failed + # re-adjudication, or a non-permit downgrades enforce_slash to shadow + # so no AtlasCoin is ever burned without an independent confirmation. + effective_enforce = enforce_slash + if enforce_slash: + _radj = _re_adjudicate or _watcher.re_adjudicate_job + _perm = _permit or _watcher.permit_enforce_slash + w_ledger = ( + Path(watcher_ledger_path) if watcher_ledger_path + else Path(".atlas-ai/tournament/watcher.jsonl") + ) + try: + wrec = _radj( + job_dir=job_dir, repo_root=repo_root, card_path=card_path, + held_root=held_root, task_id=task_id, start_commit=base_ref, + base_ref=base_ref, now=now, ledger_path=w_ledger, + ) + if not wrec.get("ok"): + permit = {"permitted": False, "reason": "watcher re-adjudication failed"} + else: + permit = _perm(wrec, ledger_path=w_ledger, current_job_id=job_id) + except Exception as exc: # noqa: BLE001 — fail-closed → shadow + permit = {"permitted": False, "reason": f"watcher error: {exc}"} + summary["watcher_permit"] = permit + if not permit.get("permitted"): + effective_enforce = False + summary["enforce_slash_downgraded"] = True + # ── Step 6: Settle (FAIL-CLOSED on ok:false) ───────────────────── - settle_env = _settle_fn(job_dir=job_dir, enforce_slash=enforce_slash) + settle_env = _settle_fn(job_dir=job_dir, enforce_slash=effective_enforce) settled_ok = settle_env.get("ok") is True summary["settled_ok"] = settled_ok @@ -415,6 +469,7 @@ def cmd_tournament_run(args: argparse.Namespace) -> None: now=now, window_s=float(getattr(args, "window", 120.0)), enforce_slash=bool(getattr(args, "enforce_slash", False)), + repo_root=_resolve_repo_root(), # Real adapters must be wired by the orchestrator skill; the defaults # raise RuntimeError with guidance if called directly. _spawn_fn=default_launcher_adapter, @@ -426,6 +481,60 @@ def cmd_tournament_run(args: argparse.Namespace) -> None: _emit({"ok": False, "error": str(exc)}) +def cmd_watcher_run(args: argparse.Namespace) -> None: + """CLI handler for ``watcher-run``. + + Re-adjudicates a settled job out-of-band from primary evidence, appends a + concordance row to the watcher ledger, and reports the fail-closed + real-slash permit for the job. FAIL-CLOSED: a missing/unreadable job → ok:false. + """ + job_dir = Path(args.job) + repo_root = getattr(args, "repo_root", None) or "." + card_path = Path(args.card) + held_root = Path(getattr(args, "held_root", None) or ".atlas-ai/cdd") + ledger_path = Path(getattr(args, "ledger_path", None) or ".atlas-ai/tournament/watcher.jsonl") + + import datetime + now = datetime.datetime.now(datetime.timezone.utc).isoformat() + + try: + rec = _watcher.re_adjudicate_job( + job_dir=job_dir, + repo_root=repo_root, + card_path=card_path, + held_root=held_root, + task_id=args.task, + start_commit=args.base_ref, + base_ref=args.base_ref, + now=now, + ledger_path=ledger_path, + ) + except Exception as exc: # noqa: BLE001 + _emit({"ok": False, "error": str(exc)}) + return + + if not rec.get("ok"): + _emit({"ok": False, **rec}) + return + + # Exclude the in-flight job (just appended to the ledger) from its own gate. + permit = _watcher.permit_enforce_slash( + rec, ledger_path=ledger_path, current_job_id=rec.get("job_id") + ) + _emit({"ok": True, **rec, "permit": permit}) + + +def cmd_watcher_status(args: argparse.Namespace) -> None: + """CLI handler for ``watcher-status``. + + Reports the watcher's historical concordance and whether the track record + would clear the real-slash gate (``real_slash_ready``). A view command. + """ + ledger_path = Path(getattr(args, "ledger_path", None) or ".atlas-ai/tournament/watcher.jsonl") + summ = _watcher.concordance_summary(ledger_path) + _emit({"ok": True, **summ}) + + def cmd_tournament_status(args: argparse.Namespace) -> None: """CLI handler for ``tournament-status``. diff --git a/prd_taskmaster/tournament/watcher.py b/prd_taskmaster/tournament/watcher.py new file mode 100644 index 0000000..f0fc4cd --- /dev/null +++ b/prd_taskmaster/tournament/watcher.py @@ -0,0 +1,464 @@ +"""Independent out-of-band re-execution watcher for the tournament marketplace. + +The marketplace runs **shadow-slash only** until an independent watcher exists: +the in-band adjudicator records ``wouldSlash`` but burns no AtlasCoin. This module +is that watcher. It re-adjudicates settled submissions from PRIMARY EVIDENCE — the +claimed commit + the CDD card — *without trusting the recorded verdict*, compares +its independently-derived verdict to the adjudicator's, accumulates a concordance +ledger over real *slash decisions*, and exposes a FAIL-CLOSED gate +(:func:`permit_enforce_slash`) that allows real (``--enforce-slash``) forfeiture +only behind an observed track record. + +Hardened invariants (the catastrophic failure is permitting an UNJUST real slash): +* **ABSTAIN, never confirm-on-inability.** Only POSITIVE independent evidence — a + clean oracle re-run that FAILED, an ORPHAN reachability sweep, or a real diff + hash that DIFFERS from the recorded one — counts as grounds to slash. If the + watcher could not run a gate (oracle errored, no worktree, failed recompute) it + ABSTAINS, which BLOCKS the permit. +* **Whole-job veto.** ANY discrepancy in the job — including a cheating *winner* + the in-band adjudicator passed — blocks the permit. +* **No self-bootstrapping.** Historical concordance excludes the in-flight job and + is measured over real slash *decisions*, not trivial double-PASS winners. +* **Read-only.** Uses the pure (non-card-writing) reachability sweep; writes ONLY + its own ledger; never mutates tournament state, reputation, or AtlasCoin. +""" +from __future__ import annotations + +import json +import shutil +import subprocess +import tempfile +from pathlib import Path +from typing import Any, Callable + +from prd_taskmaster import reachability_cmd as _rc +from prd_taskmaster.economy import append_telemetry +from prd_taskmaster.oracle_bridge import grade_card +from prd_taskmaster.reachability import sweep_task as _sweep_task +from prd_taskmaster.tournament.collect import _compute_diff_hash + +# Real slashing stays gated until the watcher has a track record. High bar by +# design: it must independently uphold the vast majority of a meaningful sample of +# real slash DECISIONS before any AtlasCoin is ever burned. +ACCURACY_THRESHOLD = 0.95 +MIN_OBSERVATIONS = 20 + +_PASS_REACH = {"WIRED", "EXEMPT"} +_DEFAULT_LEDGER = Path(".atlas-ai/tournament/watcher.jsonl") + + +# ── internal helpers ───────────────────────────────────────────────────────── + +def _readonly_sweep(task_id: str, start_commit: str, cwd: "str | None" = None) -> dict: + """Pure reachability sweep — returns a verdict WITHOUT mutating the CDD card. + + Reachability needs tasks.json + the CDD card, which are gitignored and absent + from a fresh worktree checkout; any failure to load them ABSTAINS (it is best- + effort, never card-writing, never a false WIRED). + """ + try: + repo_root = Path(cwd).resolve() if cwd else Path.cwd().resolve() + task = _rc._load_task(repo_root, task_id) + verdict = _sweep_task(repo_root, task, start_commit) + return verdict if isinstance(verdict, dict) else {"verdict": "ABSTAIN"} + except Exception: # noqa: BLE001 — best-effort, read-only, fail to ABSTAIN + return {"verdict": "ABSTAIN"} + + +def _make_worktree(repo_root: str, commit_sha: str) -> "str | None": + """Create a detached worktree at *commit_sha*; None on any git failure.""" + try: + wt = tempfile.mkdtemp(prefix="atlas-watcher-") + proc = subprocess.run( + ["git", "-C", str(repo_root), "worktree", "add", "--detach", wt, commit_sha], + capture_output=True, timeout=120, + ) + if proc.returncode != 0: + shutil.rmtree(wt, ignore_errors=True) + return None + return wt + except Exception: # noqa: BLE001 — fail-closed: no worktree on any error + return None + + +def _drop_worktree(repo_root: str, wt: "str | None") -> None: + if not wt: + return + try: + subprocess.run( + ["git", "-C", str(repo_root), "worktree", "remove", "--force", wt], + capture_output=True, timeout=60, + ) + except Exception: # noqa: BLE001 + pass + shutil.rmtree(wt, ignore_errors=True) + + +def _passes_both(oracle_verdict: str, oracle_error: bool, reach_verdict: str) -> bool: + return (oracle_verdict == "PASS" and not oracle_error) and reach_verdict in _PASS_REACH + + +def _recorded_passes_both(submission: dict) -> bool: + oracle = (submission.get("oracle") or {}).get("verdict") + reach = (submission.get("reachability") or {}).get("verdict") + return oracle == "PASS" and reach in _PASS_REACH + + +# ── public API ─────────────────────────────────────────────────────────────── + +def re_adjudicate_submission( + submission: dict, + *, + worktree: "str | None", + repo_root: str, + card_path: "str | Path", + held_root: "str | Path", + job_dir: "str | Path", + task_id: str, + start_commit: str, + base_ref: str, + oracle_cmd: "list[str] | None" = None, + _grade: Callable[..., Any] = grade_card, + _sweep: Callable[..., Any] = _readonly_sweep, + _hash: Callable[..., str] = _compute_diff_hash, +) -> dict: + """Independently re-adjudicate one submission from primary evidence. + + Returns a verdict dict. ``slash_grounds`` is True ONLY on positive independent + evidence (clean oracle FAIL, ORPHAN sweep, or real differing hash). ``agreement`` + is ``ABSTAIN`` when the watcher could not independently verify (and there is no + tamper), ``CONFIRM`` when its fail/pass conclusion matches the recorded one, and + ``DISCREPANCY`` otherwise. ABSTAIN and DISCREPANCY both block the slash permit. + """ + claimant_id = str((submission.get("claimant") or {}).get("id", "unknown")) + commit_sha = str(submission.get("commitSha", "")) + job_dir = Path(job_dir) + + evidence_dir = job_dir / "watcher-evidence" / claimant_id + ledger_dir = job_dir / "watcher-ledger" + try: + evidence_dir.mkdir(parents=True, exist_ok=True) + ledger_dir.mkdir(parents=True, exist_ok=True) + except Exception: # noqa: BLE001 + pass + + # ── Gate 1: oracle — the workhorse independent re-execution ────────────── + # grade_card is itself fail-closed: it returns ("FAIL", {"error": ...}) for an + # INFRA failure rather than raising. We must NOT read that as a genuine fail. + oracle_error = False + try: + watcher_oracle, detail = _grade( + 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, + ) + if isinstance(detail, dict) and "error" in detail: + oracle_error = True + except Exception: # noqa: BLE001 — gate could not run → abstain, never a PASS or a genuine FAIL + watcher_oracle = "FAIL" + oracle_error = True + + # ── Gate 2: reachability — best-effort, never falls back to repo_root ──── + if not worktree: + # No worktree → we did NOT evaluate the racer's commit; abstain this axis. + watcher_reach = "ABSTAIN" + else: + try: + reach = _sweep(task_id, start_commit, cwd=worktree) + watcher_reach = str(reach.get("verdict", "ERROR")) + except Exception: # noqa: BLE001 + watcher_reach = "ERROR" + + # ── Independent commit-reveal hash re-derivation (anti diff-copy) ──────── + recorded_hash = str(submission.get("commitHash", "")) + recomputed = "" + if worktree: + try: + recomputed = _hash(worktree, base_ref, commit_sha) or "" + except Exception: # noqa: BLE001 + recomputed = "" + hash_recomputed = recomputed != "" + # Tamper requires a REAL recomputed hash that DIFFERS — an empty/failed recompute + # is "unverifiable", NOT positive tamper evidence (it must abstain, not confirm). + tamper = hash_recomputed and (recomputed != recorded_hash) + + watcher_passes_both = _passes_both(str(watcher_oracle), oracle_error, watcher_reach) + recorded_passes_both = _recorded_passes_both(submission) + recorded_fail = not recorded_passes_both + + # POSITIVE independent grounds to slash. + slash_grounds = ( + (str(watcher_oracle) == "FAIL" and not oracle_error) + or watcher_reach == "ORPHAN" + or tamper + ) + # The watcher has an independent verdict iff the oracle genuinely ran. + verified = not oracle_error + + if not verified and not tamper: + agreement = "ABSTAIN" + elif slash_grounds == recorded_fail: + agreement = "CONFIRM" + else: + agreement = "DISCREPANCY" + + return { + "claimant_id": claimant_id, + "commit_sha": commit_sha, + "watcher_oracle": str(watcher_oracle), + "watcher_oracle_error": oracle_error, + "watcher_reachability": watcher_reach, + "watcher_passes_both": watcher_passes_both, + "recorded_oracle": str((submission.get("oracle") or {}).get("verdict", "")), + "recorded_reachability": str((submission.get("reachability") or {}).get("verdict", "")), + "recorded_passes_both": recorded_passes_both, + "recorded_hash": recorded_hash, + "recomputed_hash": recomputed, + "hash_recomputed": hash_recomputed, + "tamper": tamper, + "verified": verified, + "slash_grounds": slash_grounds, + "agreement": agreement, + } + + +def re_adjudicate_job( + *, + job_dir: "str | Path", + repo_root: str, + card_path: "str | Path", + held_root: "str | Path", + task_id: str, + start_commit: str, + base_ref: str, + now: str, + oracle_cmd: "list[str] | None" = None, + ledger_path: "str | Path | None" = None, + _grade: Callable[..., Any] = grade_card, + _sweep: Callable[..., Any] = _readonly_sweep, + _hash: Callable[..., str] = _compute_diff_hash, + _worktree_for: "Callable[[str], str | None] | None" = None, +) -> dict: + """Re-adjudicate every submission in a settled job; append one ledger row. + + FAIL-CLOSED: a missing/unreadable submissions file returns ``{"ok": False, ...}`` + and writes NO ledger row. The ledger row records ``decisions`` (real slash + decisions = recorded-fail submissions) and ``confirmed_slashes`` — the track + record measures slash-detection skill, not trivial winners. + """ + job_dir = Path(job_dir) + ledger = Path(ledger_path) if ledger_path is not None else _DEFAULT_LEDGER + + try: + submissions = json.loads((job_dir / "submissions.json").read_text()) + if not isinstance(submissions, list): + raise ValueError("submissions.json is not a list") + except Exception as exc: # noqa: BLE001 — fail-closed, no ledger row + return {"ok": False, "error": f"cannot read submissions: {exc}", "job_dir": str(job_dir)} + + try: + job_meta = json.loads((job_dir / "job.json").read_text()) + job_id = str(job_meta.get("jobId", job_dir.name)) + except Exception: # noqa: BLE001 + job_id = job_dir.name + + verdicts: list[dict] = [] + for sub in submissions: + commit_sha = str((sub or {}).get("commitSha", "")) + if _worktree_for is not None: + wt = _worktree_for(commit_sha) + owns_wt = False + else: + wt = _make_worktree(repo_root, commit_sha) + owns_wt = True + try: + verdicts.append( + re_adjudicate_submission( + sub, worktree=wt, repo_root=repo_root, card_path=card_path, + held_root=held_root, job_dir=job_dir, task_id=task_id, + start_commit=start_commit, base_ref=base_ref, oracle_cmd=oracle_cmd, + _grade=_grade, _sweep=_sweep, _hash=_hash, + ) + ) + finally: + if owns_wt: + _drop_worktree(repo_root, wt) + + confirms = sum(1 for v in verdicts if v["agreement"] == "CONFIRM") + discrepancies = sum(1 for v in verdicts if v["agreement"] == "DISCREPANCY") + decisions = sum(1 for v in verdicts if not v["recorded_passes_both"]) + confirmed_slashes = sum( + 1 for v in verdicts + if (not v["recorded_passes_both"]) and v["agreement"] == "CONFIRM" and v["slash_grounds"] + ) + + row = { + "job_id": job_id, + "ts": now, + "observations": len(verdicts), + "confirms": confirms, + "discrepancies": discrepancies, + "decisions": decisions, + "confirmed_slashes": confirmed_slashes, + } + ledger.parent.mkdir(parents=True, exist_ok=True) + ref = append_telemetry(row, path=ledger) + + return { + "ok": True, + "job_id": job_id, + "observations": len(verdicts), + "confirms": confirms, + "discrepancies": discrepancies, + "decisions": decisions, + "confirmed_slashes": confirmed_slashes, + "submissions": verdicts, + "ledger_ref": ref, + } + + +def _is_to_be_slashed(verdict: dict, slashed_ids: "set[str] | None") -> bool: + """A submission is to-be-slashed if explicitly named, else if it failed the gates.""" + if slashed_ids is not None: + return verdict.get("claimant_id") in slashed_ids + return not verdict.get("recorded_passes_both", False) + + +def _read_concordance( + ledger_path: "str | Path", *, exclude_job_id: "str | None" = None +) -> "tuple[int, int]": + """Return (total_decisions, total_confirmed_slashes) across PRIOR ledger rows. + + Rows for ``exclude_job_id`` (the in-flight job) are skipped so a job can never + vouch for its own gate. Malformed rows (confirmed > decisions, negatives) are + clamped so they cannot inflate the track record. + """ + p = Path(ledger_path) + if not p.is_file(): + return (0, 0) + decisions = confirmed = 0 + for line in p.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + r = json.loads(line) + except json.JSONDecodeError: + continue + if exclude_job_id is not None and r.get("job_id") == exclude_job_id: + continue + try: + d = max(int(r.get("decisions", 0) or 0), 0) + c = max(int(r.get("confirmed_slashes", 0) or 0), 0) + except (TypeError, ValueError): + continue # skip malformed rows rather than crash callers (e.g. watcher-status) + c = min(c, d) # clamp: confirmed can never exceed decisions + decisions += d + confirmed += c + return (decisions, confirmed) + + +def concordance_summary(ledger_path: "str | Path") -> dict: + """Summarize the watcher's historical track record over real slash decisions.""" + observations, confirmed = _read_concordance(ledger_path) + conc = (confirmed / observations) if observations else 0.0 + conc = min(conc, 1.0) + return { + "observations": observations, + "confirmed_slashes": confirmed, + "concordance": conc, + "min_observations": MIN_OBSERVATIONS, + "accuracy_threshold": ACCURACY_THRESHOLD, + "real_slash_ready": observations >= MIN_OBSERVATIONS and conc >= ACCURACY_THRESHOLD, + } + + +def permit_enforce_slash( + record: dict, + *, + ledger_path: "str | Path", + slashed_ids: "set[str] | None" = None, + current_job_id: "str | None" = None, +) -> dict: + """FAIL-CLOSED gate: may real slashing be enforced for this job? + + ``permitted`` is True only when ALL of: + * at least one to-be-slashed submission, and EVERY one is independently + confirmed (positive grounds + CONFIRM agreement); + * NO discrepancy anywhere in the job (a cheating winner vetoes the whole job); + * NO abstained to-be-slashed submission (could-not-verify never permits); + * historical concordance over PRIOR jobs clears the threshold over enough real + slash decisions. + Any error → ``permitted: False``. + """ + try: + subs = record.get("submissions") + if not isinstance(subs, list): + raise TypeError("record.submissions must be a list") + + job_id = current_job_id if current_job_id is not None else record.get("job_id") + + # Whole-job veto: real coin moves only when the watcher independently + # reproduced the ENTIRE adjudication. Any discrepancy (incl. a cheating + # winner) OR any ABSTAIN (a submission — winner or loser — the watcher + # could not independently verify) blocks the permit. + discrepancies = [ + str(v.get("claimant_id")) for v in subs if v.get("agreement") == "DISCREPANCY" + ] + abstained = [ + str(v.get("claimant_id")) for v in subs if v.get("agreement") == "ABSTAIN" + ] + + to_slash = [v for v in subs if _is_to_be_slashed(v, slashed_ids)] + confirmed = [ + str(v.get("claimant_id")) for v in to_slash + if v.get("slash_grounds") and v.get("agreement") == "CONFIRM" + ] + + observations, confirmed_hist = _read_concordance(ledger_path, exclude_job_id=job_id) + concordance = min((confirmed_hist / observations) if observations else 0.0, 1.0) + track_record_ok = observations >= MIN_OBSERVATIONS and concordance >= ACCURACY_THRESHOLD + + all_slashes_confirmed = bool(to_slash) and len(confirmed) == len(to_slash) + permitted = bool( + track_record_ok and not discrepancies and not abstained and all_slashes_confirmed + ) + + if permitted: + reason = ( + f"permitted: {len(confirmed)} confirmed slash(es), " + f"concordance={concordance:.3f} over {observations} prior decisions" + ) + elif not to_slash: + reason = "blocked: no to-be-slashed submissions to confirm" + elif discrepancies: + reason = f"blocked: {len(discrepancies)} discrepancy(ies) in the job" + elif abstained: + reason = f"blocked: {len(abstained)} submission(s) in the job could not be independently verified (abstain)" + elif not track_record_ok: + reason = ( + f"blocked: thin track record — {observations}/{MIN_OBSERVATIONS} prior " + f"decisions, concordance={concordance:.3f}/{ACCURACY_THRESHOLD}" + ) + else: + reason = "blocked: not all to-be-slashed submissions confirmed" + + return { + "permitted": permitted, + "reason": reason, + "confirmed": confirmed, + "discrepancies": discrepancies, + "abstained": abstained, + "concordance": concordance, + "observations": observations, + } + except Exception as exc: # noqa: BLE001 — fail-closed + return { + "permitted": False, + "reason": f"error: {exc}", + "confirmed": [], + "discrepancies": [], + "abstained": [], + "concordance": 0.0, + "observations": 0, + } diff --git a/tests/core/test_tournament_cmd.py b/tests/core/test_tournament_cmd.py index 49c989b..18e2322 100644 --- a/tests/core/test_tournament_cmd.py +++ b/tests/core/test_tournament_cmd.py @@ -654,3 +654,58 @@ def _capturing_spawn(spec: RacerSpec) -> dict: f"racer {spec.claimant_id!r} prompt does not contain {orch_session!r}:\n" f"{spec.prompt[:300]}" ) + + +# ─── Watcher gate on REAL slashing (engine-enforced shadow-until-permitted) ─── + +def _settle_spy(): + """A settle stub that records the enforce_slash it was called with.""" + calls = [] + def _stub(*, job_dir, enforce_slash=False): + calls.append(enforce_slash) + return _make_settle_ok()(job_dir=job_dir, enforce_slash=enforce_slash) + return _stub, calls + + +def test_watcher_gate_downgrades_to_shadow_when_not_permitted(tmp_path): + """enforce_slash=True but the watcher does NOT permit → settle runs in SHADOW.""" + settle_fn, calls = _settle_spy() + kwargs = _run_args(tmp_path, settle_fn=settle_fn) + kwargs["enforce_slash"] = True + kwargs["_re_adjudicate"] = lambda **kw: {"ok": True, "job_id": JOB_ID, "submissions": []} + kwargs["_permit"] = lambda rec, **kw: {"permitted": False, "reason": "thin track record"} + + summary = run_tournament(**kwargs) + + assert calls == [False], "real slashing must be downgraded to shadow when not permitted" + assert summary["enforce_slash_downgraded"] is True + assert summary["watcher_permit"]["permitted"] is False + + +def test_watcher_gate_allows_real_slash_when_permitted(tmp_path): + """enforce_slash=True and the watcher permits → settle runs with real enforcement.""" + settle_fn, calls = _settle_spy() + kwargs = _run_args(tmp_path, settle_fn=settle_fn) + kwargs["enforce_slash"] = True + kwargs["_re_adjudicate"] = lambda **kw: {"ok": True, "job_id": JOB_ID, "submissions": []} + kwargs["_permit"] = lambda rec, **kw: {"permitted": True, "reason": "ok", "confirmed": ["x"]} + + summary = run_tournament(**kwargs) + + assert calls == [True], "real slashing must proceed when the watcher permits" + assert summary.get("enforce_slash_downgraded") is not True + + +def test_watcher_gate_fail_closed_when_watcher_errors(tmp_path): + """A crashing watcher must FAIL-CLOSED to shadow, never burn coin.""" + settle_fn, calls = _settle_spy() + kwargs = _run_args(tmp_path, settle_fn=settle_fn) + kwargs["enforce_slash"] = True + def _boom(**kw): + raise RuntimeError("watcher exploded") + kwargs["_re_adjudicate"] = _boom + + summary = run_tournament(**kwargs) + + assert calls == [False], "a watcher error must downgrade to shadow (fail-closed)" + assert summary["enforce_slash_downgraded"] is True diff --git a/tests/core/test_tournament_watcher.py b/tests/core/test_tournament_watcher.py new file mode 100644 index 0000000..74d79c7 --- /dev/null +++ b/tests/core/test_tournament_watcher.py @@ -0,0 +1,479 @@ +"""Tests for the independent out-of-band re-execution watcher. + +The watcher re-adjudicates settled tournament submissions from PRIMARY EVIDENCE +(the claimed commit + the CDD card), compares its independently-derived verdict +to the adjudicator's recorded verdict, accumulates a concordance ledger over real +*slash decisions*, and exposes a FAIL-CLOSED gate that permits real +(--enforce-slash) forfeiture only behind an observed track record. + +Hardened contract (post adversarial review): + * ABSTAIN — inability to independently verify (oracle could not run, no worktree, + failed hash recompute) is NEVER counted as grounds to slash. It abstains and + BLOCKS the permit. Only POSITIVE independent evidence confirms a slash. + * Whole-job veto — ANY discrepancy in the job (including a cheating winner the + in-band adjudicator passed) blocks the permit. + * No self-bootstrapping — the historical concordance excludes the in-flight job + and is measured over real slash *decisions*, not trivial double-PASS winners. + * Read-only — the watcher writes ONLY its own ledger. +""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from prd_taskmaster.tournament import watcher + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _submission( + claimant_id: str, + *, + oracle: str = "PASS", + reach: str = "WIRED", + commit_hash: str = "hash-ok", + commit_sha: str = "c0ffee", +) -> dict: + return { + "claimant": {"kind": "executor", "id": claimant_id}, + "commitSha": commit_sha, + "oracle": {"verdict": oracle}, + "reachability": {"verdict": reach}, + "commitHash": commit_hash, + } + + +def _write_job(job_dir: Path, submissions: list[dict], job_id: str = "job-1") -> None: + job_dir.mkdir(parents=True, exist_ok=True) + (job_dir / "submissions.json").write_text(json.dumps(submissions)) + (job_dir / "job.json").write_text(json.dumps({"jobId": job_id})) + + +def _stub_gates(*, oracle="PASS", detail=None, reach="WIRED", recomputed_hash="hash-ok"): + return ( + lambda **kw: (oracle, detail if detail is not None else {}), + lambda task_id, start_commit, cwd=None: {"verdict": reach}, + lambda worktree, base_ref, commit_sha, **kw: recomputed_hash, + ) + + +def _radj(submission, tmp_path, *, worktree=None, **stubs): + grade, sweep, hsh = _stub_gates(**stubs) + return watcher.re_adjudicate_submission( + submission, + worktree=tmp_path if worktree is None else worktree, + repo_root=str(tmp_path), card_path=tmp_path / "card.json", held_root=tmp_path, + job_dir=tmp_path, task_id="7", start_commit="base", base_ref="base", + _grade=grade, _sweep=sweep, _hash=hsh, + ) + + +def _v(claimant_id, *, agreement, slash_grounds, recorded_passes_both): + """Build a per-submission verdict (the subset permit_enforce_slash consumes).""" + return { + "claimant_id": claimant_id, + "agreement": agreement, + "slash_grounds": slash_grounds, + "recorded_passes_both": recorded_passes_both, + } + + +# ── re_adjudicate_submission: classification ───────────────────────────────── + +def test_confirm_when_watcher_agrees_pass(tmp_path): + v = _radj(_submission("ex-a", oracle="PASS", reach="WIRED"), tmp_path, + oracle="PASS", reach="WIRED", recomputed_hash="hash-ok") + assert v["watcher_passes_both"] is True + assert v["slash_grounds"] is False + assert v["verified"] is True + assert v["agreement"] == "CONFIRM" + + +def test_confirm_when_watcher_agrees_fail(tmp_path): + # Adjudicator recorded FAIL; watcher independently also FAILs (clean grade) → confirmed slash. + v = _radj(_submission("ex-b", oracle="FAIL", reach="WIRED"), tmp_path, + oracle="FAIL", reach="WIRED", recomputed_hash="hash-ok") + assert v["slash_grounds"] is True + assert v["recorded_passes_both"] is False + assert v["agreement"] == "CONFIRM" + + +def test_discrepancy_when_watcher_passes_but_recorded_failed(tmp_path): + v = _radj(_submission("ex-c", oracle="FAIL", reach="WIRED"), tmp_path, + oracle="PASS", reach="WIRED", recomputed_hash="hash-ok") + assert v["slash_grounds"] is False + assert v["recorded_passes_both"] is False + assert v["agreement"] == "DISCREPANCY" + + +def test_discrepancy_when_watcher_fails_but_recorded_passed(tmp_path): + # Cheater-caught direction: in-band let it through (recorded PASS), watcher independently FAILs. + v = _radj(_submission("ex-w", oracle="PASS", reach="WIRED"), tmp_path, + oracle="FAIL", reach="WIRED", recomputed_hash="hash-ok") + assert v["slash_grounds"] is True + assert v["recorded_passes_both"] is True + assert v["agreement"] == "DISCREPANCY" + + +def test_reachability_orphan_is_slash_grounds(tmp_path): + v = _radj(_submission("ex-o", oracle="PASS", reach="WIRED"), tmp_path, + oracle="PASS", reach="ORPHAN", recomputed_hash="hash-ok") + assert v["slash_grounds"] is True + assert v["watcher_passes_both"] is False + + +def test_reachability_exempt_passes(tmp_path): + v = _radj(_submission("ex-x", oracle="PASS", reach="EXEMPT"), tmp_path, + oracle="PASS", reach="EXEMPT", recomputed_hash="hash-ok") + assert v["watcher_passes_both"] is True + assert v["slash_grounds"] is False + assert v["agreement"] == "CONFIRM" + + +def test_real_hash_mismatch_is_tamper_discrepancy(tmp_path): + # A real (non-empty) recomputed hash that DIFFERS from recorded → positive tamper. + v = _radj(_submission("ex-d", oracle="PASS", reach="WIRED", commit_hash="hash-ok"), tmp_path, + oracle="PASS", reach="WIRED", recomputed_hash="DIFFERENT-REAL-HASH") + assert v["tamper"] is True + assert v["slash_grounds"] is True + assert v["agreement"] == "DISCREPANCY" # recorded PASS but tamper detected + + +# ── ABSTAIN: inability to verify is NEVER grounds to slash ──────────────────── + +def test_abstain_when_oracle_raises(tmp_path): + def boom(**kw): + raise RuntimeError("oracle exploded") + grade, sweep, hsh = _stub_gates(reach="WIRED", recomputed_hash="hash-ok") + v = watcher.re_adjudicate_submission( + _submission("ex-e", oracle="FAIL", reach="WIRED"), + worktree=str(tmp_path), repo_root=str(tmp_path), card_path=tmp_path / "card.json", + held_root=tmp_path, job_dir=tmp_path, task_id="7", start_commit="base", base_ref="base", + _grade=boom, _sweep=sweep, _hash=hsh, + ) + assert v["watcher_oracle_error"] is True + assert v["verified"] is False + assert v["slash_grounds"] is False # could not verify → NOT grounds to slash + assert v["agreement"] == "ABSTAIN" + + +def test_abstain_when_oracle_detail_has_error(tmp_path): + # grade_card is fail-closed: it can return ("FAIL", {"error": ...}) for an INFRA error, + # which must NOT be read as a genuine independent fail. + v = _radj(_submission("ex-f", oracle="FAIL", reach="WIRED"), tmp_path, + oracle="FAIL", detail={"error": "oracle CLI crashed"}, reach="WIRED", recomputed_hash="hash-ok") + assert v["watcher_oracle_error"] is True + assert v["slash_grounds"] is False + assert v["agreement"] == "ABSTAIN" + + +def test_abstain_when_no_worktree(tmp_path): + # Worktree creation failed: never fall back to repo_root HEAD; reachability/hash abstain. + grade, sweep, hsh = _stub_gates(oracle="PASS", reach="WIRED", recomputed_hash="hash-ok") + v = watcher.re_adjudicate_submission( + _submission("ex-n", oracle="PASS", reach="WIRED"), + worktree=None, repo_root=str(tmp_path), card_path=tmp_path / "card.json", + held_root=tmp_path, job_dir=tmp_path, task_id="7", start_commit="base", base_ref="base", + _grade=grade, _sweep=sweep, _hash=hsh, + ) + assert v["watcher_reachability"] == "ABSTAIN" + assert v["tamper"] is False # cannot claim tamper without a real recompute + assert v["hash_recomputed"] is False + + +def test_sweep_raises_is_abstain_reach(tmp_path): + def boom_sweep(task_id, start_commit, cwd=None): + raise RuntimeError("grep blew up") + grade, _sweep, hsh = _stub_gates(oracle="PASS", recomputed_hash="hash-ok") + v = watcher.re_adjudicate_submission( + _submission("ex-s", oracle="PASS", reach="WIRED"), + worktree=str(tmp_path), repo_root=str(tmp_path), card_path=tmp_path / "card.json", + held_root=tmp_path, job_dir=tmp_path, task_id="7", start_commit="base", base_ref="base", + _grade=grade, _sweep=boom_sweep, _hash=hsh, + ) + assert v["watcher_reachability"] == "ERROR" + assert v["slash_grounds"] is False # ERROR reachability is not ORPHAN → no grounds + + +# ── re_adjudicate_job: ledger + read-only ──────────────────────────────────── + +def test_job_appends_one_ledger_row_with_decisions(tmp_path): + job_dir = tmp_path / "jobs" / "job-1" + _write_job(job_dir, [ + _submission("ex-a", oracle="PASS", reach="WIRED"), + _submission("ex-b", oracle="FAIL", reach="WIRED"), + ], job_id="job-1") + ledger = tmp_path / "watcher.jsonl" + grade, sweep, hsh = _stub_gates(oracle="PASS", reach="WIRED", recomputed_hash="hash-ok") + rec = watcher.re_adjudicate_job( + job_dir=job_dir, repo_root=str(tmp_path), card_path=tmp_path / "card.json", + held_root=tmp_path, task_id="7", start_commit="base", base_ref="base", + now="2026-06-17T00:00:00Z", ledger_path=ledger, + _grade=grade, _sweep=sweep, _hash=hsh, _worktree_for=lambda sha: str(tmp_path), + ) + assert rec["ok"] is True + assert rec["observations"] == 2 + rows = [json.loads(l) for l in ledger.read_text().splitlines() if l.strip()] + assert len(rows) == 1 + assert rows[0]["job_id"] == "job-1" + assert rows[0]["decisions"] == 1 # one recorded-fail (ex-b) → one slash decision + assert "confirmed_slashes" in rows[0] + + +def test_job_missing_submissions_returns_error_no_ledger(tmp_path): + job_dir = tmp_path / "jobs" / "empty" + job_dir.mkdir(parents=True) + ledger = tmp_path / "watcher.jsonl" + rec = watcher.re_adjudicate_job( + job_dir=job_dir, repo_root=str(tmp_path), card_path=tmp_path / "card.json", + held_root=tmp_path, task_id="7", start_commit="base", base_ref="base", + now="2026-06-17T00:00:00Z", ledger_path=ledger, _worktree_for=lambda sha: str(tmp_path), + ) + assert rec["ok"] is False + assert not ledger.exists() + + +def test_job_writes_only_watcher_ledger(tmp_path): + job_dir = tmp_path / "jobs" / "job-1" + _write_job(job_dir, [_submission("ex-a")], job_id="job-1") + ledger = tmp_path / "watcher.jsonl" + reputation = tmp_path / "reputation.jsonl" + telemetry = tmp_path / "telemetry.jsonl" + grade, sweep, hsh = _stub_gates(recomputed_hash="hash-ok") + watcher.re_adjudicate_job( + job_dir=job_dir, repo_root=str(tmp_path), card_path=tmp_path / "card.json", + held_root=tmp_path, task_id="7", start_commit="base", base_ref="base", + now="2026-06-17T00:00:00Z", ledger_path=ledger, + _grade=grade, _sweep=sweep, _hash=hsh, _worktree_for=lambda sha: str(tmp_path), + ) + assert ledger.exists() + assert not reputation.exists() + assert not telemetry.exists() + + +# ── permit_enforce_slash: the fail-closed real-slash gate ──────────────────── + +def _seed_ledger(ledger: Path, *, decisions: int, confirmed: int, job_id: str = "hist") -> None: + ledger.write_text(json.dumps({ + "job_id": job_id, "ts": "2026-06-01T00:00:00Z", + "decisions": decisions, "confirmed_slashes": confirmed, + "discrepancies": 0, "observations": decisions, "confirms": confirmed, + }) + "\n") + + +def test_permit_blocks_below_min_observations(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS - 1, confirmed=watcher.MIN_OBSERVATIONS - 1) + record = {"job_id": "job-1", "submissions": [ + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert out["observations"] < watcher.MIN_OBSERVATIONS + + +def test_permit_blocks_on_discrepancy_among_slashes(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-c", agreement="DISCREPANCY", slash_grounds=False, recorded_passes_both=False), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert "ex-c" in out["discrepancies"] + + +def test_permit_blocks_on_cheating_winner_discrepancy(tmp_path): + # The slashed racer is confirmed, but a WINNER is a discrepancy (watcher fails one the + # in-band adjudicator passed) → the whole job is blocked. + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), # confirmed slash + _v("ex-win", agreement="DISCREPANCY", slash_grounds=True, recorded_passes_both=True), # cheating winner + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert "ex-win" in out["discrepancies"] + + +def test_permit_blocks_on_abstained_slash(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-a", agreement="ABSTAIN", slash_grounds=False, recorded_passes_both=False), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + + +def test_permit_blocks_on_abstained_winner(tmp_path): + # Real coin moves only when the watcher reproduced the WHOLE adjudication — an + # unverifiable (ABSTAIN) winner blocks the slash even though the loser is confirmed. + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), # confirmed slash + _v("ex-win", agreement="ABSTAIN", slash_grounds=False, recorded_passes_both=True), # unverified winner + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert "ex-win" in out["abstained"] + + +def test_permit_blocks_on_empty_to_slash(tmp_path): + # No to-be-slashed submissions → never a blanket green-light. + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-a", agreement="CONFIRM", slash_grounds=False, recorded_passes_both=True), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert "no to-be-slashed" in out["reason"].lower() + + +def test_permit_allows_when_confirmed_and_track_record(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-a", agreement="CONFIRM", slash_grounds=False, recorded_passes_both=True), # clean winner + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), # confirmed slash + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is True + assert out["confirmed"] == ["ex-b"] + assert out["discrepancies"] == [] + + +def test_permit_blocks_mixed_batch_one_discrepancy_vetoes(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), + _v("ex-c", agreement="DISCREPANCY", slash_grounds=False, recorded_passes_both=False), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert "ex-c" in out["discrepancies"] + + +def test_permit_excludes_in_flight_job_from_track_record(tmp_path): + # The current job's own ledger row must NOT count toward its own gate. + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS, job_id="job-1") + record = {"job_id": "job-1", "submissions": [ + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger, current_job_id="job-1") + assert out["permitted"] is False # only prior (excluded) row existed → 0 obs + assert out["observations"] == 0 + + +def test_permit_concordance_boundary(tmp_path): + ledger = tmp_path / "watcher.jsonl" + # exactly threshold → permitted + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, + confirmed=round(watcher.ACCURACY_THRESHOLD * watcher.MIN_OBSERVATIONS)) + record = {"job_id": "job-1", "submissions": [ + _v("ex-b", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), + ]} + assert watcher.permit_enforce_slash(record, ledger_path=ledger)["permitted"] is True + # below threshold → blocked + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, + confirmed=int(0.90 * watcher.MIN_OBSERVATIONS)) + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert out["permitted"] is False + assert "track record" in out["reason"].lower() + + +def test_permit_confirms_tamper_slash(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + record = {"job_id": "job-1", "submissions": [ + _v("ex-t", agreement="CONFIRM", slash_grounds=True, recorded_passes_both=False), + ]} + out = watcher.permit_enforce_slash(record, ledger_path=ledger) + assert "ex-t" in out["confirmed"] + assert out["permitted"] is True + + +def test_permit_fail_closed_on_bad_record(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + out = watcher.permit_enforce_slash({"submissions": "not-a-list"}, ledger_path=ledger) + assert out["permitted"] is False + assert out["reason"] + + +# ── concordance summary + CLI handlers ─────────────────────────────────────── + +def test_concordance_summary_reports_readiness(tmp_path): + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + summ = watcher.concordance_summary(ledger) + assert summ["observations"] == watcher.MIN_OBSERVATIONS + assert summ["concordance"] == 1.0 + assert summ["real_slash_ready"] is True + + +def test_concordance_summary_empty_ledger_not_ready(tmp_path): + summ = watcher.concordance_summary(tmp_path / "missing.jsonl") + assert summ["observations"] == 0 + assert summ["real_slash_ready"] is False + + +def test_concordance_clamps_malformed_rows(tmp_path): + # A malformed row with confirmed > decisions must not inflate concordance past 1.0. + ledger = tmp_path / "watcher.jsonl" + ledger.write_text(json.dumps({"job_id": "bad", "decisions": 5, "confirmed_slashes": 999}) + "\n") + summ = watcher.concordance_summary(ledger) + assert summ["concordance"] <= 1.0 + + +def test_concordance_ignores_nonnumeric_rows(tmp_path): + # A malformed ledger row with a non-numeric field must not crash watcher-status. + ledger = tmp_path / "watcher.jsonl" + ledger.write_text( + json.dumps({"job_id": "bad", "decisions": "oops", "confirmed_slashes": None}) + "\n" + + json.dumps({"job_id": "ok", "decisions": 4, "confirmed_slashes": 4}) + "\n" + ) + summ = watcher.concordance_summary(ledger) # must not raise + assert summ["observations"] == 4 # only the valid row counts + + +def test_resolve_repo_root_falls_back_outside_git(tmp_path): + from prd_taskmaster.tournament.cmd import _resolve_repo_root + # A non-git directory → falls back to the default rather than raising. + assert _resolve_repo_root(cwd=str(tmp_path), default="SENTINEL") == "SENTINEL" + + +def test_cmd_watcher_status_emits_ok_json(tmp_path, capsys): + import argparse + from prd_taskmaster.tournament.cmd import cmd_watcher_status + ledger = tmp_path / "watcher.jsonl" + _seed_ledger(ledger, decisions=watcher.MIN_OBSERVATIONS, confirmed=watcher.MIN_OBSERVATIONS) + with pytest.raises(SystemExit) as exc: + cmd_watcher_status(argparse.Namespace(ledger_path=str(ledger))) + assert exc.value.code == 0 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is True + assert out["real_slash_ready"] is True + + +def test_cmd_watcher_run_missing_job_is_fail_closed(tmp_path, capsys): + import argparse + from prd_taskmaster.tournament.cmd import cmd_watcher_run + ledger = tmp_path / "watcher.jsonl" + args = argparse.Namespace( + job=str(tmp_path / "no-such-job"), repo_root=".", card=str(tmp_path / "card.json"), + task="7", base_ref="base", held_root=str(tmp_path / "cdd"), ledger_path=str(ledger), + ) + with pytest.raises(SystemExit) as exc: + cmd_watcher_run(args) + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert not ledger.exists()