diff --git a/CHANGELOG.md b/CHANGELOG.md index 962a4b8..918accf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to the claude-plugins project will be documented in this fil The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). Entries are listed newest-first; each plugin section is treated as released when merged to `main`. +### code-review v3.2.0 + +#### Added +- New `run-prefix` helper subcommand: a resumable, in-process runner for the deterministic review prefix (stages `setup` through `cache-check`). It reads `run_plan.json` and walks the stages in one process — resolving each stage's placeholder tokens from prior-stage artifacts, redirecting stdout per stage, and honoring `on_failure` policies and validation gates — instead of one orchestrator turn per stage. It pauses only at genuine decision points (the hygiene-only early exit, a signal-extraction or coverage-critic singleton that needs an agent, or the route/partition boundary), emitting a status JSON that tells the orchestrator what to do next, and resumes from a given stage on re-invocation. A failed `continue_with_coverage_gap` stage emits a canonical `agent-failure` system finding so the gap is auditable. Documented as the `run-prefix` result contract in `SCHEMA.md`. +- Subprocess A/B parity oracle for the prefix: the golden-fixture harness now walks each fixture two ways — one subprocess per stage (reproducing the current per-stage orchestrator walk) versus the new `run-prefix` runner — and asserts byte-identical normalized artifacts through `cache-check` across all seven fixtures, plus a pause-sequence check pinning the resumable segment boundaries. Contract tests cover token resolution, resumable dependency reconstruction, singleton detection, `on_failure` handling, and the runner's error/boundary returns. +- `run-prefix` failure diagnostics are attributed per stage: each in-process stage's stderr is captured and folded into the returned status message (and, for a `continue_with_coverage_gap` stage, into the emitted `agent-failure` finding's explanation), and an unexpected stage crash logs its full traceback — so a batched-runner failure stays diagnosable without reproducing it, now that one process spans many stages. + ### code-review v3.1.1 #### Added diff --git a/plugins/code-review/.claude-plugin/plugin.json b/plugins/code-review/.claude-plugin/plugin.json index f4e0b0e..3c771b0 100644 --- a/plugins/code-review/.claude-plugin/plugin.json +++ b/plugins/code-review/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "code-review", "description": "Code review plugin", - "version": "3.1.1", + "version": "3.2.0", "author": { "name": "ClosedLoop", "email": "support@closedloop.ai" diff --git a/plugins/code-review/README.md b/plugins/code-review/README.md index e895bc6..75f2d78 100644 --- a/plugins/code-review/README.md +++ b/plugins/code-review/README.md @@ -37,7 +37,7 @@ plugins/code-review/ python/golden_fixture_harness.py Golden fixture harness: replays canonical inputs through helper subcommands and diffs against expected envelopes (PLN-719 Phase 8) python/test_golden_fixtures.py Pytest driver that runs every fixture under tools/python/fixtures/ python/fixtures// Per-fixture directory (config.yaml + inputs/ + expected/); 3 full scenarios + 6 README-stubs for future coverage - python/prefix_golden_harness.py Prefix golden harness: walks the deterministic prefix (setup→spawn-spec) in-process against real git fixtures and snapshots every intermediate artifact (PLN-1229 Phase 0) + python/prefix_golden_harness.py Prefix golden harness + subprocess A/B parity oracle: walks the deterministic prefix against real git fixtures — in-process for golden snapshots, and per-stage-subprocess vs `run-prefix` for byte-equal parity (PLN-1229 Phase 0/1) python/test_prefix_golden.py Pytest driver for the prefix harness: determinism oracle + golden diff across the prefix_fixtures/ matrix python/prefix_fixtures// Per-fixture directory (expected/ golden snapshots); 7 branch scenarios (standard, fast-path, hygiene-only, empty-diff, cache-hit, since-last-review, coverage-critic) ``` @@ -231,6 +231,7 @@ The helper script is a multi-subcommand Python CLI. The orchestrator invokes it | `finalize-result` | Consolidates validated findings + coverage state + verdict into the canonical `review_result.json` envelope; deep-merges `/telemetry.json` into the canonical `telemetry` block and populates `telemetry.cache_hit_rate["bha"]` from `cache_result.json` (PLN-719 Phase 7/9) | | `arbitrate-budget` | Applies the canonical reviewer cap policy; emits coverage gaps for required reviewers that overflow (PLN-719) | | `prepare-run` | Emits a declarative `run_plan.json` describing the 30-stage pipeline (PLN-719) | +| `run-prefix` | Runs the deterministic prefix (setup→cache-check) in one process, resolving tokens and honoring gates/`on_failure`; pauses at the hygiene-only exit, a singleton needing an agent, or the route/partition boundary, emitting a status JSON and resuming from a given stage (PLN-1229) | ## GitHub CI Mode diff --git a/plugins/code-review/SCHEMA.md b/plugins/code-review/SCHEMA.md index fd900f5..135200b 100644 --- a/plugins/code-review/SCHEMA.md +++ b/plugins/code-review/SCHEMA.md @@ -493,6 +493,56 @@ Stages from plans 01/03/05/06 are present in `run_plan.json` but marked --- +## 7b. `run-prefix` result contract (PLN-1229) + +`run-prefix` runs the deterministic prefix (stages 01→`cache_check`) in ONE +process instead of one orchestrator turn per stage. It reads `run_plan.json` + +`setup.json` from `--cr-dir`, walks from `--resume-from` (default: the first +plan stage), and stops at the next genuine decision point — emitting a status +JSON (to stdout, or `--output `) that tells the orchestrator what to do +next. The runner is **resumable**: after handling a pause the orchestrator +re-invokes `run-prefix --resume-from `. Because each segment is a +fresh process, the `depends_on` `completed` set is reconstructed from artifacts +on disk (a prior stage counts as done iff its literal `expected_outputs` exist). + +**Result fields:** + +| Field | Type | Meaning | +| --------------- | --------------- | ----------------------------------------------------------------------- | +| `next_action` | string (enum) | The pause reason — authoritative (read this, not the exit code). | +| `resume_stage` | string \| null | The stage id to pass as `--resume-from` on the next invocation. | +| `singleton` | string \| null | `"extract_signals"` \| `"coverage_critic"` when `needs_singleton`. | +| `failed_stage` | string \| null | The aborting stage id when `next_action == "error"`. | +| `ran_stages` | string[] | Stage ids executed (or `continue`-failed) this segment, in order. | +| `message` | string \| null | Short diagnostic on `error`, else null. | + +**`next_action` values:** + +| Value | Fires at | Orchestrator does next | +| ------------------- | ------------------------------------- | ---------------------------------------------------------------------------- | +| `needs_singleton` | `stage_11` / `stage_15` `needs_agent` | Spawn the `singleton` agent, write its output, re-invoke from `resume_stage`. | +| `hygiene_exit` | Gate A (`hygiene_only` after hygiene) | Present hygiene findings and stop (no verdict/footer). | +| `ready_for_route` | reaching `stage_17_partition` | Run Gate B (`route`) + partition + the rest of the walk. | +| `error` | a stage aborted / a gate failed | Fall back to the per-stage walk from `failed_stage`; partials are preserved. | + +`ready_for_route` distinguishes its two cases by `resume_stage`: a non-null +`resume_stage` (`stage_17_partition`) is the normal boundary — run Gate B + +partition from there. A **null** `resume_stage` means the walk reached the end +of the plan without a partition stage (e.g. a depth tier that filters partition +out); there is nothing left to route, so the orchestrator skips Gate B and +partition and proceeds directly to the reviewer fleet. + +The exit code is `0` for every well-formed result (including `error`) — the +`next_action` field is the contract. Route + partition (Segment 3) fold into the +runner in Phase 2, at which point `ready_for_route` becomes `ready_for_reviewers` +(carrying `fast_path` + `cache_status_message`). The `on_failure` policy of each +stage is honored exactly as the Walker Contract prescribes: `abort` → `error`; +`continue` → proceed; `continue_with_coverage_gap` → proceed after writing an +`agent-failure` system finding to `agent_-failed.json` (collected by +`collect-findings`). + +--- + ## 8. Determinism tiers (PLN-719 Section 8) | Tier | Definition | Required reviewers may depend? | diff --git a/plugins/code-review/tools/python/code_review_helpers.py b/plugins/code-review/tools/python/code_review_helpers.py index dd8062d..29d4087 100644 --- a/plugins/code-review/tools/python/code_review_helpers.py +++ b/plugins/code-review/tools/python/code_review_helpers.py @@ -17,6 +17,7 @@ import copy import functools import hashlib +import io import json import os import random @@ -25,9 +26,10 @@ import stat import subprocess import sys +import traceback import uuid from collections.abc import Generator -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any @@ -9995,12 +9997,12 @@ def _validate_stages_config(config: dict[str, Any]) -> None: f"stages.json {sid}: missing required ``min_depth`` field; " f"tag as one of {sorted(_VALID_MIN_DEPTHS)}", ) - for field in ("min_depth", "max_depth"): - if field in stage: - val = stage[field] + for tier_field in ("min_depth", "max_depth"): + if tier_field in stage: + val = stage[tier_field] if val not in _VALID_MIN_DEPTHS: raise ValueError( - f"stages.json {sid}.{field}: invalid tier {val!r}; " + f"stages.json {sid}.{tier_field}: invalid tier {val!r}; " f"must be one of {sorted(_VALID_MIN_DEPTHS)}", ) lo = stage["min_depth"] @@ -10311,6 +10313,586 @@ def cmd_evaluate_gate(args: argparse.Namespace) -> int: return 0 +# --------------------------------------------------------------------------- +# run-prefix — resumable in-process segment runner (PLN-1229 Phase 1) +# --------------------------------------------------------------------------- +# +# ``cmd_run_prefix`` collapses the orchestrator's stage-by-stage walk of the +# deterministic prefix (stages 01→cache_check) into ONE process: it reads the +# already-emitted ``run_plan.json``, resolves each stage's ```` +# tokens from prior-stage artifacts (the ``{...}`` templates were already +# resolved by ``prepare-run``), dispatches the ``cmd_*`` in-process with the +# stage's stdout redirect, and honors ``on_failure`` + validation gates — +# exactly as ``start.md``'s Walker Contract does, but without a model turn per +# stage. It stops (and hands control back to the orchestrator) only at genuine +# decision points: +# +# * Gate A — ``hygiene_only`` early exit after ``stage_12_hygiene``. +# * Singleton — a PLN-725 prepare stage (``stage_11`` / ``stage_15``) whose +# manifest is ``needs_agent`` → pause so the orchestrator can +# spawn the one synchronous agent, then resume at the sibling +# consolidate stage. +# * Gate B — reaching ``stage_17_partition``. Phase 1 stops here and +# returns ``ready_for_route``; folding route + partition into +# the runner (→ ``ready_for_reviewers``) is Phase 2. +# +# The runner is RESUMABLE: after handling a pause the orchestrator re-invokes +# ``run-prefix --resume-from ``. Because each segment is a fresh process, +# the ``completed`` set (which drives ``depends_on`` skipping) is reconstructed +# from artifacts on disk, not carried in memory. +# +# NOTE ON DUPLICATION: ``prefix_golden_harness.py`` contains a SEPARATE, +# test-side reimplementation of this walk (in-process, with canned agent +# stubs). That duplication is deliberate — the subprocess A/B parity oracle +# runs the harness walk (A) against this runner (B) and asserts byte-identical +# artifacts, so a bug shared between the two implementations cannot hide. The +# canonical stage/gate TABLES (``stages.json`` / ``_build_validation_gates``) +# ARE shared; only the WRAPPER logic is independently implemented on each side. + +# Stage ids the runner special-cases. Named so a rename in stages.json is one +# edit here rather than a scatter of string literals. +_RP_SETUP_STAGE = "stage_01_setup" +_RP_AUTO_INCREMENTAL_STAGE = "stage_07_auto_incremental" +_RP_HYGIENE_STAGE = "stage_12_hygiene" +_RP_PARTITION_STAGE = "stage_17_partition" # Phase-1 terminal boundary + +# The two PLN-725 singleton *prepare* stages: the manifest each writes and the +# path (within that manifest JSON) to the ``status`` field the walker reads to +# decide ``needs_agent`` vs ``cache_hit``/``skipped``. +_RP_SINGLETONS: dict[str, dict[str, Any]] = { + "stage_11_extract_signals": { + "name": "extract_signals", + "manifest": "extract_signals_manifest.json", + "status_path": (), # top-level "status" + }, + "stage_15_coverage_critic": { + "name": "coverage_critic", + "manifest": _COVERAGE_STATE_FILENAME, + "status_path": ("critic",), # coverage.json → critic → status + }, +} + +# Every angle-bracket token the prefix stage args reference. Explicit (vs +# regex-scanning) so an unrecognized token in a future stage fails loudly. +_RP_TOKENS = ( + "", "", "", "", "", + "", "", "", "", + "", "", "", "", "", +) + + +@dataclass +class _RunPrefixContext: + """State threaded through one run-prefix invocation for token resolution.""" + + cr_dir: Path + plugin_root: str + model_id: str + start_time: str + global_cache: str + #: Runtime token overrides keyed by literal token string. Currently only + #: ```` is overridden (by stage_07 auto-incremental narrowing). + overrides: dict[str, str] = field(default_factory=dict) + + +def _rp_read_json(path: Path) -> dict[str, Any]: + """Read a JSON object, degrading a missing/malformed file to ``{}``. + + Mirrors the walker's "pass an empty string for a not-yet-produced artifact" + tolerance — a token whose source file does not exist resolves to "". + """ + try: + with path.open() as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except (OSError, json.JSONDecodeError): + return {} + + +def _rp_default_plugin_root() -> Path: + """```` derived from this file's location. + + ``code_review_helpers.py`` lives at ``/tools/python/`` so the + plugin root is two parents up. Used when ``--plugin-root`` is not supplied. + """ + return Path(__file__).resolve().parent.parent.parent + + +def _rp_resolve_token(token: str, ctx: _RunPrefixContext) -> str: + """Resolve one ```` per the start.md Walker Contract token table. + + Constants (````/````/````/ + ````) come from the context; artifact-derived tokens are read + lazily from ``cr_dir``. A missing source file degrades to "" exactly as the + walker does. + """ + if token in ctx.overrides: + return ctx.overrides[token] + if token == "": + return ctx.plugin_root + if token == "": + return ctx.model_id + if token == "": + return ctx.start_time + if token == "": + return ctx.global_cache + + cr = ctx.cr_dir + scope = _rp_read_json(cr / "scope.json") + if token == "": + return str(scope.get("diff_scope", "")) + if token == "": + return str(scope.get("base_ref", "")) + if token == "": + return str(scope.get("diff_tip", "")) + if token == "": + return str(scope.get("scope_kind", "")) + if token == "": + return str(scope.get("review_root", "")) + if token == "": + review_branch = str(scope.get("review_branch", "")) + base_ref = str(scope.get("base_ref", "")) + if not review_branch and not base_ref: + return "" + return f"{review_branch}:{base_ref}" + if token == "": + return str(_rp_read_json(cr / "cache_config.json").get("cache_dir", "")) + if token == "": + return str(_rp_read_json(cr / "hashes.json").get("prompt_hash", "")) + if token == "": + return str(_rp_read_json(cr / "hashes.json").get("context_key", "")) + if token == "": + return str(_rp_read_json(cr / "intent.json").get("intent", "")) + raise KeyError(f"unresolved run-prefix token: {token}") + + +def _rp_resolve_args(raw_args: list[str], ctx: _RunPrefixContext) -> list[str]: + """Substitute every ```` occurrence within each arg string.""" + out: list[str] = [] + for arg in raw_args: + resolved = arg + for token in _RP_TOKENS: + if token in resolved: + resolved = resolved.replace(token, _rp_resolve_token(token, ctx)) + out.append(resolved) + return out + + +def _rp_outputs_present(stage: dict[str, Any]) -> bool: + """True when every literal ``expected_outputs`` path for ``stage`` exists. + + Glob patterns (``agent_*.json``, ``patches_p.txt``) and any unresolved + ```` residue are treated as satisfied — those are enforced by the + fleet-stage gates downstream, not the deterministic prefix. Paths in + ``run_plan.json`` are already ``{cr_dir}``-absolute (prepare-run resolved + them), so a bare existence check is correct. + """ + for out in stage.get("expected_outputs", []) or []: + if "*" in out or "<" in out: + continue + if not Path(out).is_file(): + return False + return True + + +def _rp_reconstruct_completed( + stages: list[dict[str, Any]], resume_from: str, +) -> set[str]: + """Rebuild the ``completed`` set for stages BEFORE ``resume_from``. + + Each segment is a fresh process, so ``depends_on`` skipping can't rely on an + in-memory set. A prior stage counts as completed iff its literal + ``expected_outputs`` are all present on disk — which is exactly the + invariant a successful run leaves behind, and correctly EXCLUDES a prior + ``continue`` stage that failed (its dependents then skip, matching the + in-process walk). ``stage_01_setup`` is always completed: it ran in the + orchestrator's stage 0 and never re-runs in a segment. + """ + completed: set[str] = set() + if not resume_from: + return completed + for stage in stages: + sid = stage["id"] + if sid == resume_from: + break + if sid == _RP_SETUP_STAGE: + completed.add(sid) + continue + if not stage.get("enabled", True): + continue + if _rp_outputs_present(stage): + completed.add(sid) + return completed + + +def _rp_seed_scope_override(ctx: _RunPrefixContext) -> None: + """Reflect a prior stage_07 auto-incremental scope narrowing from disk. + + On resume the narrowing already happened in an earlier segment, so seed the + ```` override from ``auto_incremental.json`` if present. Harmless + on a fresh run (the file doesn't exist yet); the same override is (re)applied + inline right after stage_07 runs via ``_rp_apply_post_stage_overrides``. + """ + override = _rp_read_json(ctx.cr_dir / "auto_incremental.json").get("diff_scope") + if override: + ctx.overrides[""] = str(override) + + +def _rp_apply_post_stage_overrides(stage_id: str, ctx: _RunPrefixContext) -> None: + """Apply the cross-stage ```` override the walker does in prose. + + Per start.md's stage_07 note, when ``auto_incremental.json.diff_scope`` is + non-null the walker narrows the cached ```` token for every + subsequent stage (parse-diff runs after auto-incremental in array order). + """ + if stage_id == _RP_AUTO_INCREMENTAL_STAGE: + override = _rp_read_json(ctx.cr_dir / "auto_incremental.json").get("diff_scope") + if override: + ctx.overrides[""] = str(override) + + +def _rp_dispatch(func: Any, ns: argparse.Namespace, stdout_path: Path | None) -> int: + """Call ``func(ns)`` with stdout redirected per the stage's ``stdout`` field. + + When ``stdout_path`` is set the helper's stdout IS the artifact (production's + ``> file`` redirect); otherwise the helper writes its own file and prints a + summary we discard. Returns the ``cmd_*`` int return code. + """ + if stdout_path is not None: + with stdout_path.open("w") as fh, contextlib.redirect_stdout(fh): + rc = func(ns) + else: + with contextlib.redirect_stdout(io.StringIO()): + rc = func(ns) + return int(rc or 0) + + +def _emit_prefix_stage_failure_finding( + cr_dir: Path, stage_id: str, diagnostic: str | None = None, +) -> None: + """Write ``agent_-failed.json`` with a canonical agent-failure finding. + + Mirrors start.md Walker Contract step 5 (``continue_with_coverage_gap``): a + prefix stage that failed to run must surface as an operator-visible, + collectable finding — ``collect-findings`` globs ``agent_*.json`` — rather + than vanish silently. ``diagnostic`` is the caller's captured failure detail + (rc / exception / the stage's own stderr tail); it is folded into the + finding's explanation so the gap is debuggable without reproducing the + failure. Fail-open on write error (telemetry is observational). + """ + explanation = ( + f"The deterministic prefix stage {stage_id!r} exited non-zero or " + "produced no output. Its on_failure policy is " + "continue_with_coverage_gap, so the pipeline proceeds but the gap is " + "recorded here for auditability." + ) + if diagnostic: + explanation += f" Diagnostic: {diagnostic}" + finding = { + "reviewer": "foundation", + "source": "foundation", + "finding_scope": "system", + "system_marker": "agent-failure", + "category": "Coverage", + "severity": "MEDIUM", + "file": None, + "line": None, + "issue": f"Prefix stage {stage_id} failed; continuing with coverage gap.", + "explanation": explanation, + "recommendation": ( + "Re-run the review once the underlying issue is resolved. Common " + "causes: malformed diff input, a git error, or a taxonomy mismatch " + "after a recent edit." + ), + "confidence": 1.0, + "emitted_at": datetime.now(timezone.utc).isoformat(), + } + try: + target = cr_dir / f"agent_{stage_id}-failed.json" + with open(target, "w") as f: + json.dump({"findings": [finding]}, f, indent=2) + except OSError: + pass + + +def _execute_stage_inprocess( + stage: dict[str, Any], + ctx: _RunPrefixContext, + parser: argparse.ArgumentParser, + completed: set[str], +) -> tuple[str, str | None]: + """Run one helper stage in-process, honoring stdout + ``on_failure``. + + Mirrors start.md Walker Contract steps 1-5 for ``kind == "helper"`` stages. + Returns ``(status, message)`` where ``status`` is one of: + + * ``"ran"`` — succeeded; ``stage`` added to ``completed``. + * ``"skipped"`` — disabled, unmet ``depends_on``, or the already-run + setup stage. Not added to ``completed`` (except + setup, whose completion the orchestrator owns). + * ``"failed_continue"``— failed under ``on_failure: continue`` / + ``continue_with_coverage_gap``; NOT added to + ``completed`` so dependents skip. + * ``"failed_abort"`` — failed under ``on_failure: abort``; caller stops. + + ``message`` carries a short diagnostic on failure (else ``None``). + """ + stage_id = stage["id"] + + if not stage.get("enabled", True): + return "skipped", None + if any(dep not in completed for dep in stage.get("depends_on", []) or []): + return "skipped", None + if stage_id == _RP_SETUP_STAGE: + # Setup ran in the orchestrator's stage 0; never re-run it (a second + # run would regenerate the non-deterministic start_time). Mark completed + # so dependents resolve. + completed.add(stage_id) + return "ran", None + + kind = stage.get("kind") + if kind != "helper": + return "failed_abort", ( + f"prefix must be helper-only; got kind={kind!r} for {stage_id!r}" + ) + + on_failure = stage.get("on_failure", "abort") + rc = 1 + message: str | None = None + # Capture the stage's stderr so a failing stage's own diagnostic (the + # cmd_* handlers report via ``print(..., file=sys.stderr)``) can be attributed + # to THIS stage in the batched model — one process now runs many stages, so + # bare process stderr no longer maps to an orchestrator turn. It is written + # back to the real stderr below, so live visibility is unchanged. + err_buf = io.StringIO() + try: + with contextlib.redirect_stderr(err_buf): + resolved = _rp_resolve_args(stage.get("args", []) or [], ctx) + ns = parser.parse_args([stage["subcommand"], *resolved]) + stdout_target = stage.get("stdout") + rc = _rp_dispatch( + ns.func, ns, Path(stdout_target) if stdout_target else None, + ) + except SystemExit as exc: # argparse rejected the resolved args + message = f"argparse rejected args for {stage_id}: {exc}" + rc = 1 + except Exception as exc: # noqa: BLE001 — a crash is a stage failure; on_failure decides + message = f"{type(exc).__name__} in {stage_id}: {exc}" + rc = 1 + # Preserve the full traceback on the real stderr — the short ``message`` + # alone can't be debugged without reproducing the crash. + traceback.print_exc() + + stage_stderr = err_buf.getvalue() + if stage_stderr: + sys.stderr.write(stage_stderr) + + outputs_ok = message is None and _rp_outputs_present(stage) + if rc != 0 or not outputs_ok: + if message is None: + message = f"stage {stage_id} failed (rc={rc}, outputs_ok={outputs_ok})" + stderr_tail = stage_stderr.strip().splitlines() + if stderr_tail: + message = f"{message}; stderr: {stderr_tail[-1][:300]}" + if on_failure == "abort": + return "failed_abort", message + if on_failure == "continue_with_coverage_gap": + _emit_prefix_stage_failure_finding(ctx.cr_dir, stage_id, message) + return "failed_continue", message + + _rp_apply_post_stage_overrides(stage_id, ctx) + completed.add(stage_id) + return "ran", None + + +def _rp_build_parser() -> argparse.ArgumentParser: + """Build the same subcommand parser ``main()`` uses. + + Reusing ``_register_subparsers`` guarantees each stage's Namespace is built + exactly as production builds it — the alternative (hand-rolling a Namespace + per subcommand) would silently drift from cli.json. + """ + parser = argparse.ArgumentParser(add_help=False) + subparsers = parser.add_subparsers(dest="command", required=True) + _register_subparsers(subparsers) + return parser + + +def _rp_next_stage_id(stages: list[dict[str, Any]], stage_id: str) -> str | None: + """Return the id of the stage AFTER ``stage_id`` in plan array order.""" + for i, stage in enumerate(stages): + if stage["id"] == stage_id and i + 1 < len(stages): + return str(stages[i + 1]["id"]) + return None + + +def _rp_singleton_needs_agent(ctx: _RunPrefixContext, stage_id: str) -> bool: + """True when a just-run singleton prepare stage's manifest is ``needs_agent``. + + ``cache_hit`` / ``skipped`` (or a missing/malformed manifest) → no agent, so + the walk continues and the sibling consolidate stage no-ops. + """ + spec = _RP_SINGLETONS[stage_id] + node: Any = _rp_read_json(ctx.cr_dir / str(spec["manifest"])) + for key in spec["status_path"]: + node = node.get(key, {}) if isinstance(node, dict) else {} + status = node.get("status", "") if isinstance(node, dict) else "" + return str(status) == "needs_agent" + + +def _rp_evaluate_gates( + run_plan: dict[str, Any], cr_dir: Path, stage_id: str, +) -> tuple[bool, str | None]: + """Evaluate any validation gate anchored on ``stage_id`` (Walker step 7). + + Reads the depth-filtered ``validation_gates`` array from ``run_plan.json`` + (authoritative for this invocation) and applies each matching gate's + ``on_failure_action`` via the canonical ``evaluate_validation_gate`` + enforcer. Returns ``(ok, message)``; ``ok`` is False only when a gate fails + with ``on_failure_action: "abort"``. ``emit_coverage_gap`` gate failures + record a coverage-gap finding and continue. + """ + for gate in run_plan.get("validation_gates", []) or []: + if gate.get("after_stage") != stage_id: + continue + passed, reason = evaluate_validation_gate(gate) + if passed: + continue + action = gate.get("on_failure_action", "abort") + if action == "abort": + return False, f"gate {gate.get('gate', '?')!r} after {stage_id}: {reason}" + if action == "emit_coverage_gap": + _emit_prefix_stage_failure_finding(cr_dir, stage_id) + return True, None + + +def cmd_run_prefix(args: argparse.Namespace) -> int: + """Run the deterministic prefix in-process until the next pause point. + + PLN-1229 Phase 1. Consumes ``run_plan.json`` + ``setup.json`` from + ``--cr-dir`` (both already written by the orchestrator's stage 0) and walks + stages from ``--resume-from`` (default: the first plan stage) to the next + genuine decision point, emitting a small status JSON telling the orchestrator + what to do next. See the module comment above for the pause-point contract + and SCHEMA.md §"run-prefix result" for the emitted fields. + + Result ``next_action`` values (authoritative — read the JSON, not the exit + code, which is always 0 for a well-formed run): + + * ``needs_singleton`` — spawn the ``singleton`` agent, write its output, + then re-invoke ``--resume-from ``. + * ``hygiene_exit`` — Gate A: present hygiene findings and stop. + * ``ready_for_route`` — reached ``stage_17_partition``; run Gate B (route) + + the rest of the walk. (Phase 2 folds route + + partition in and returns ``ready_for_reviewers``.) + * ``error`` — a stage aborted; ``failed_stage`` is set. The + orchestrator falls back to the per-stage walk from + there. Partial artifacts are preserved. + """ + cr_dir = Path(args.cr_dir) + run_plan = _rp_read_json(cr_dir / "run_plan.json") + stages: list[dict[str, Any]] = list(run_plan.get("stages", []) or []) + flags = run_plan.get("flags", {}) or {} + setup = _rp_read_json(cr_dir / "setup.json") + + ctx = _RunPrefixContext( + cr_dir=cr_dir, + plugin_root=(getattr(args, "plugin_root", "") or str(_rp_default_plugin_root())), + model_id=(getattr(args, "model_id", "") or "opus"), + start_time=str(setup.get("start_time", "")), + global_cache=str(setup.get("global_cache", "0")), + ) + _rp_seed_scope_override(ctx) + + resume_from = getattr(args, "resume_from", "") or "" + completed = _rp_reconstruct_completed(stages, resume_from) + parser = _rp_build_parser() + hygiene_only = str(flags.get("hygiene_only", "")).lower() in ("true", "1") + + def emit(result: dict[str, Any]) -> int: + result.setdefault("ran_stages", ran) + text = json.dumps(result, indent=2) + output = getattr(args, "output", None) + if output: + Path(output).write_text(text + "\n") + else: + print(text) + return 0 + + ran: list[str] = [] + started = not resume_from + for stage in stages: + sid = stage["id"] + if not started: + if sid == resume_from: + started = True + else: + continue + # Phase-1 terminal boundary: Gate B (route) + partition are still the + # orchestrator's job. Stop BEFORE executing partition. + if sid == _RP_PARTITION_STAGE: + return emit({ + "next_action": "ready_for_route", + "resume_stage": sid, + "singleton": None, + "failed_stage": None, + "message": None, + }) + + status, message = _execute_stage_inprocess(stage, ctx, parser, completed) + if status == "failed_abort": + return emit({ + "next_action": "error", + "resume_stage": sid, + "singleton": None, + "failed_stage": sid, + "message": message, + }) + if status in ("ran", "failed_continue"): + ran.append(sid) + + gate_ok, gate_msg = _rp_evaluate_gates(run_plan, cr_dir, sid) + if not gate_ok: + return emit({ + "next_action": "error", + "resume_stage": sid, + "singleton": None, + "failed_stage": sid, + "message": gate_msg, + }) + + # Gate A — hygiene-only early exit after the hygiene stage. + if sid == _RP_HYGIENE_STAGE and hygiene_only: + return emit({ + "next_action": "hygiene_exit", + "resume_stage": None, + "singleton": None, + "failed_stage": None, + "message": None, + }) + + # PLN-725 singleton pause — a needs_agent prepare stage yields to the + # orchestrator to spawn one synchronous agent, then resume at the sibling. + if status == "ran" and sid in _RP_SINGLETONS and _rp_singleton_needs_agent(ctx, sid): + return emit({ + "next_action": "needs_singleton", + "resume_stage": _rp_next_stage_id(stages, sid), + "singleton": _RP_SINGLETONS[sid]["name"], + "failed_stage": None, + "message": None, + }) + + # Walked to the end without reaching the partition stage (e.g. a depth tier + # that filters partition out). No route needed; signal completion. + return emit({ + "next_action": "ready_for_route", + "resume_stage": None, + "singleton": None, + "failed_stage": None, + "message": None, + }) + + def cmd_prepare_run(args: argparse.Namespace) -> int: """Emit ``run_plan.json`` describing the full review pipeline. diff --git a/plugins/code-review/tools/python/config/cli.json b/plugins/code-review/tools/python/config/cli.json index 03c2ac5..bea9e93 100644 --- a/plugins/code-review/tools/python/config/cli.json +++ b/plugins/code-review/tools/python/config/cli.json @@ -1612,6 +1612,47 @@ } ] }, + { + "name": "run-prefix", + "help": "Run the deterministic prefix in-process until the next pause point (Gate A / singleton / route); emits a status JSON. PLN-1229.", + "func": "cmd_run_prefix", + "args": [ + { + "flags": [ + "--cr-dir" + ], + "required": true, + "help": "CR_DIR for the review session (must already contain run_plan.json and setup.json)." + }, + { + "flags": [ + "--resume-from" + ], + "default": "", + "help": "Stage id to resume the walk from (default: first plan stage). Pass the sibling consolidate stage after handling a needs_singleton pause." + }, + { + "flags": [ + "--plugin-root" + ], + "default": "", + "help": "Value for the token (default: derived from the helpers file location)." + }, + { + "flags": [ + "--model-id" + ], + "default": "opus", + "help": "Value for the token (BHA's default reviewer model)." + }, + { + "flags": [ + "--output" + ], + "help": "Write the status JSON to this path instead of stdout." + } + ] + }, { "name": "prep-assets", "help": "Copy prompt assets from plugin to CR_DIR", diff --git a/plugins/code-review/tools/python/fixtures/run_plan_snapshots/cli_parser_resolved.json b/plugins/code-review/tools/python/fixtures/run_plan_snapshots/cli_parser_resolved.json index 04c91e7..99a800f 100644 --- a/plugins/code-review/tools/python/fixtures/run_plan_snapshots/cli_parser_resolved.json +++ b/plugins/code-review/tools/python/fixtures/run_plan_snapshots/cli_parser_resolved.json @@ -1735,6 +1735,48 @@ } ] }, + { + "name": "run-prefix", + "func": "cmd_run_prefix", + "args": [ + { + "flags": [ + "--cr-dir" + ], + "dest": "cr_dir", + "default": null, + "required": true + }, + { + "flags": [ + "--resume-from" + ], + "dest": "resume_from", + "default": "" + }, + { + "flags": [ + "--plugin-root" + ], + "dest": "plugin_root", + "default": "" + }, + { + "flags": [ + "--model-id" + ], + "dest": "model_id", + "default": "opus" + }, + { + "flags": [ + "--output" + ], + "dest": "output", + "default": null + } + ] + }, { "name": "prep-assets", "func": "cmd_prep_assets", diff --git a/plugins/code-review/tools/python/prefix_golden_harness.py b/plugins/code-review/tools/python/prefix_golden_harness.py index e3f9256..f6af581 100644 --- a/plugins/code-review/tools/python/prefix_golden_harness.py +++ b/plugins/code-review/tools/python/prefix_golden_harness.py @@ -59,6 +59,7 @@ import json import os import subprocess +import sys from collections.abc import Callable, Iterator from dataclasses import dataclass, field from datetime import datetime, timezone @@ -829,6 +830,25 @@ class PrefixRun: all_artifacts: list[str] +def _make_prefix_context(cr_dir: Path, run_plan: dict[str, Any]) -> PrefixContext: + """Build the walk context, seeding the four constant ```` overrides. + + Shared by the in-process (golden) and subprocess (A-side parity) drivers so + the seeded-constant contract lives in exactly one place. + """ + setup = run_plan["_setup"] + return PrefixContext( + cr_dir=cr_dir, + flags=run_plan.get("flags", {}), + overrides={ + "": str(PLUGIN_ROOT), + "": "opus", + "": str(setup.get("start_time", 0)), + "": str(setup.get("global_cache", "0")), + }, + ) + + def cache_dir_for(home: Path, repo_name: str = "fixture_repo") -> Path: """The cache dir ``finalize-cache`` resolves for a local branch review. @@ -870,29 +890,296 @@ def run_prefix_fixture( base_ref_override=fixture.base_ref_override, pr_number=fixture.pr_number, ) - setup = run_plan["_setup"] - ctx = PrefixContext( - cr_dir=cr_dir, - flags=run_plan.get("flags", {}), - overrides={ - "": str(PLUGIN_ROOT), - "": "opus", - "": str(setup.get("start_time", 0)), - "": str(setup.get("global_cache", "0")), - }, + ctx = _make_prefix_context(cr_dir, run_plan) + results = walk_prefix( + run_plan, ctx, stop_before=stop_before, + singleton_stubs=_fixture_singleton_stubs(fixture), ) - stubs: dict[str, dict[str, Any]] = { - "stage_11_extract_signals": default_extract_signals_stub(), - } - if fixture.coverage_critic_stub is not None: - stubs["stage_15_coverage_critic"] = fixture.coverage_critic_stub - results = walk_prefix(run_plan, ctx, stop_before=stop_before, singleton_stubs=stubs) snapshots = collect_snapshots(cr_dir, repo, home) all_artifacts = sorted(p.name for p in cr_dir.iterdir() if p.is_file()) return PrefixRun(results=results, snapshots=snapshots, all_artifacts=all_artifacts) +# --------------------------------------------------------------------------- +# Subprocess A/B parity oracle (PLN-1229 Phase 1) +# --------------------------------------------------------------------------- +# +# The refactor guarantee is a parity test: run the deterministic prefix TWO ways +# on the same fixture and assert byte-identical (normalized) artifacts. +# +# A-side — the subprocess-per-stage walk below, faithfully reproducing what +# ``start.md`` does today: one ``python3 code_review_helpers.py +# `` per stage, real ``> file`` redirects, token +# resolution + gates done by this (test-side) driver. +# B-side — one ``python3 code_review_helpers.py run-prefix`` per segment, +# i.e. the production ``cmd_run_prefix`` under test. +# +# A and B share the canonical stage/gate TABLES but implement the walk WRAPPER +# independently (A here, B in code_review_helpers), so a wrapper bug shared by +# both cannot hide. Both stop at the Phase-1 boundary (before stage_17_partition +# — route + partition land in Phase 2). Determinism across the two independent +# stage-0 setups is provided by the same normalization the golden test relies on. + +# The production helpers CLI both sides shell out to. +HELPERS_PATH = Path(code_review_helpers.__file__).resolve() + +# The A-side stops before the partition stage: Gate B (route) + partition are +# still the orchestrator's job in Phase 1, so run-prefix (B) never runs them. +PHASE1_STOP_STAGE = _PARTITION_STAGE_ID + + +def _make_fake_gh(bin_dir: Path) -> Path: + """Write a ``gh`` stub that always exits non-zero and return its bin dir. + + ``hermetic_prefix_env`` neutralizes ``_detect_open_pr`` in-process, but a + subprocess stage would call the real ``gh``. Shadowing ``gh`` with a failing + stub on ``PATH`` makes the subprocess take the identical deterministic no-PR + branch (``gh pr view`` → CalledProcessError → ``None``) regardless of whether + the host has ``gh`` installed and authenticated. + """ + bin_dir.mkdir(parents=True, exist_ok=True) + gh = bin_dir / "gh" + gh.write_text("#!/bin/sh\nexit 1\n") + gh.chmod(0o755) + return bin_dir + + +def _subprocess_env(bin_dir: Path) -> dict[str, str]: + """Inherit the hermetic env (HOME / CR_GLOBAL_CACHE / git isolation) and + prepend the fake-``gh`` bin dir so subprocess stages resolve it first.""" + env = dict(os.environ) + env["PATH"] = f"{bin_dir}{os.pathsep}{env.get('PATH', '')}" + return env + + +def _run_stage_subprocess( + subcommand: str, + resolved_args: list[str], + repo: Path, + stdout_path: Path | None, + env: dict[str, str], +) -> int: + """Run one helper stage as a real subprocess, honoring the ``> file`` redirect.""" + cmd = [sys.executable, str(HELPERS_PATH), subcommand, *resolved_args] + if stdout_path is not None: + with stdout_path.open("w") as fh: + proc = subprocess.run( + cmd, cwd=str(repo), env=env, stdout=fh, + stderr=subprocess.PIPE, text=True, + ) + else: + proc = subprocess.run( + cmd, cwd=str(repo), env=env, stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, text=True, + ) + return proc.returncode + + +def _execute_stage_subprocess( + stage: dict[str, Any], + ctx: PrefixContext, + repo: Path, + env: dict[str, str], + completed: set[str], +) -> StageResult: + """Subprocess twin of ``_execute_stage`` — identical control flow, real exec. + + No partition augmentation: the A-side stops before ``stage_17_partition``, + so that branch never runs in Phase 1. + """ + stage_id = stage["id"] + if not stage.get("enabled", True): + return StageResult(stage_id, "skipped") + if any(dep not in completed for dep in stage.get("depends_on", [])): + return StageResult(stage_id, "skipped") + if stage_id == _SETUP_STAGE_ID: + completed.add(stage_id) + return StageResult(stage_id, "ran") + + kind = stage.get("kind") + if kind != "helper": + raise AssertionError( + f"prefix must be helper-only; got kind={kind!r} for {stage_id!r}", + ) + + resolved = _resolve_args(stage.get("args", []), ctx) + stdout_target = stage.get("stdout") + rc = _run_stage_subprocess( + stage["subcommand"], resolved, repo, + Path(stdout_target) if stdout_target else None, env, + ) + outputs_ok = _expected_outputs_present(stage) + if rc != 0 or not outputs_ok: + on_failure = stage.get("on_failure", "abort") + if on_failure == "abort": + raise AssertionError( + f"stage {stage_id!r} failed (rc={rc}, outputs_ok={outputs_ok}) " + f"with on_failure=abort", + ) + return StageResult(stage_id, "failed_continue", rc) + + _apply_post_stage_overrides(stage_id, ctx) + completed.add(stage_id) + return StageResult(stage_id, "ran", rc) + + +def subprocess_walk_prefix( + run_plan: dict[str, Any], + ctx: PrefixContext, + repo: Path, + env: dict[str, str], + *, + stop_before: str = PHASE1_STOP_STAGE, + singleton_stubs: dict[str, dict[str, Any]] | None = None, +) -> list[StageResult]: + """A-side walk: subprocess per stage, Gate A + singleton dispatch, no route.""" + stubs = singleton_stubs or {} + completed: set[str] = set() + results: list[StageResult] = [] + for stage in run_plan["stages"]: + stage_id = stage["id"] + if stage_id == stop_before: + break + results.append(_execute_stage_subprocess(stage, ctx, repo, env, completed)) + if stage_id in _SINGLETON_AGENT_OUTPUT: + _singleton_dispatch(stage_id, ctx, stubs) + if stage_id == _HYGIENE_STAGE_ID and ctx.flags.get("hygiene_only"): + break # Gate A + return results + + +def _fixture_singleton_stubs(fixture: PrefixFixture) -> dict[str, dict[str, Any]]: + """The canned singleton stubs a fixture drives (extract-signals always; + coverage-critic only when the fixture plants a reviewer roster).""" + stubs: dict[str, dict[str, Any]] = { + "stage_11_extract_signals": default_extract_signals_stub(), + } + if fixture.coverage_critic_stub is not None: + stubs["stage_15_coverage_critic"] = fixture.coverage_critic_stub + return stubs + + +def run_prefix_fixture_subprocess( + tmp_root: Path, fixture: PrefixFixture, +) -> dict[str, str]: + """A-side driver: build the fixture, walk it via subprocess-per-stage, snapshot. + + Stage 0 (setup + prepare-run) runs in-process — it is unchanged shared code + both sides call; the refactor under test is the WALK, which runs as real + subprocesses here. + """ + repo = tmp_root / "fixture_repo" + cr_dir = tmp_root / "cr" + home = tmp_root / "home" + bin_dir = _make_fake_gh(tmp_root / "bin") + + build_fixture_repo(repo, fixture.repo) + home.mkdir(parents=True, exist_ok=True) + + with hermetic_prefix_env(repo, home): + # Snapshot env INSIDE the hermetic context so the subprocess inherits the + # redirected HOME / CR_GLOBAL_CACHE (else the cache resolves to the real + # ~/.claude and pollutes it — and makes the run non-deterministic). + env = _subprocess_env(bin_dir) + if fixture.pre_seed is not None: + fixture.pre_seed(home, repo, cr_dir) + run_plan = run_setup_and_prepare( + cr_dir, + depth=fixture.depth, + scope_args=fixture.scope_args, + hygiene_only=fixture.hygiene_only, + since_last_review=fixture.since_last_review, + full_review=fixture.full_review, + base_ref_override=fixture.base_ref_override, + pr_number=fixture.pr_number, + ) + ctx = _make_prefix_context(cr_dir, run_plan) + subprocess_walk_prefix( + run_plan, ctx, repo, env, + singleton_stubs=_fixture_singleton_stubs(fixture), + ) + return collect_snapshots(cr_dir, repo, home) + + +def _invoke_run_prefix_subprocess( + cr_dir: Path, repo: Path, env: dict[str, str], resume_from: str, +) -> dict[str, Any]: + """Invoke ``code_review_helpers.py run-prefix`` as a subprocess; return status JSON.""" + cmd = [ + sys.executable, str(HELPERS_PATH), "run-prefix", + "--cr-dir", str(cr_dir), + "--plugin-root", str(PLUGIN_ROOT), + "--model-id", "opus", + ] + if resume_from: + cmd += ["--resume-from", resume_from] + proc = subprocess.run( + cmd, cwd=str(repo), env=env, capture_output=True, text=True, + ) + if proc.returncode != 0: + raise AssertionError( + f"run-prefix exited {proc.returncode}: {proc.stderr.strip()}", + ) + return json.loads(proc.stdout) + + +def run_prefix_fixture_via_runner( + tmp_root: Path, fixture: PrefixFixture, +) -> tuple[dict[str, str], list[dict[str, Any]]]: + """B-side driver: drive ``run-prefix`` segment-by-segment, snapshot. + + Returns ``(snapshots, statuses)`` — the second element is the ordered list of + status JSONs the runner emitted (one per segment), so a test can assert the + pause sequence as well as artifact parity. + """ + repo = tmp_root / "fixture_repo" + cr_dir = tmp_root / "cr" + home = tmp_root / "home" + bin_dir = _make_fake_gh(tmp_root / "bin") + stubs = _fixture_singleton_stubs(fixture) + + build_fixture_repo(repo, fixture.repo) + home.mkdir(parents=True, exist_ok=True) + + statuses: list[dict[str, Any]] = [] + with hermetic_prefix_env(repo, home): + # Snapshot env INSIDE the hermetic context (see run_prefix_fixture_subprocess). + env = _subprocess_env(bin_dir) + if fixture.pre_seed is not None: + fixture.pre_seed(home, repo, cr_dir) + run_setup_and_prepare( + cr_dir, + depth=fixture.depth, + scope_args=fixture.scope_args, + hygiene_only=fixture.hygiene_only, + since_last_review=fixture.since_last_review, + full_review=fixture.full_review, + base_ref_override=fixture.base_ref_override, + pr_number=fixture.pr_number, + ) + resume_from = "" + # Bounded loop: at most Gate A + 2 singletons + terminal = 4 segments. + for _ in range(8): + status = _invoke_run_prefix_subprocess(cr_dir, repo, env, resume_from) + statuses.append(status) + if status["next_action"] != "needs_singleton": + break + singleton = status["singleton"] + stage_id = ( + "stage_11_extract_signals" + if singleton == "extract_signals" + else "stage_15_coverage_critic" + ) + stub = stubs[stage_id] + (cr_dir / _SINGLETON_AGENT_OUTPUT[stage_id]).write_text( + json.dumps(stub, indent=2), + ) + resume_from = status["resume_stage"] + snapshots = collect_snapshots(cr_dir, repo, home) + return snapshots, statuses + + # --------------------------------------------------------------------------- # Fixture library # --------------------------------------------------------------------------- diff --git a/plugins/code-review/tools/python/test_code_review_helpers.py b/plugins/code-review/tools/python/test_code_review_helpers.py index 7987c36..da88aba 100644 --- a/plugins/code-review/tools/python/test_code_review_helpers.py +++ b/plugins/code-review/tools/python/test_code_review_helpers.py @@ -19314,7 +19314,7 @@ def test_resolved_parser_spec_matches_snapshot(self) -> None: Catches: cli.json type/default/choices/action edits, $$ constant misroutes (the original false positive was --max-files = 20 vs BUDGET_TOTAL_CAP_DEFAULT = 20), missing required flags, mutex routing - drift, and func name drift across all 44 subparsers. + drift, and func name drift across all 46 subparsers. """ expected = json.loads( (self._snapshot_dir() / "cli_parser_resolved.json").read_text(), @@ -22289,3 +22289,504 @@ def test_prep_assets_stdout_reports_impact_path( assert out["impact_analyzer_prompt"].endswith( "impact_analyzer_prompt.txt", ) + + +# --------------------------------------------------------------------------- +# PLN-1229 Phase 1 / P0-C — run-prefix contract tests +# --------------------------------------------------------------------------- +# +# These characterize the behaviors run-prefix newly owns in Python (they lived +# as start.md prose before the refactor): angle-bracket token resolution, the +# resumable ``completed``-set reconstruction, singleton needs_agent detection, +# on_failure abort-vs-continue (incl. continue_with_coverage_gap finding +# emission), and cmd_run_prefix's error / boundary returns. The full happy-path +# walk is covered end-to-end by the subprocess A/B parity oracle in +# test_prefix_golden.py; these pin the branches that oracle's deterministic +# fixtures never exercise (aborts, degraded stages, empty inputs). + + +def _rp_ctx(cr_dir: Path, **over: Any) -> Any: + from code_review_helpers import _RunPrefixContext + + kwargs: dict[str, Any] = { + "cr_dir": cr_dir, + "plugin_root": "/PLUGIN", + "model_id": "opus", + "start_time": "1700000000", + "global_cache": "0", + } + kwargs.update(over) + return _RunPrefixContext(**kwargs) + + +class TestRunPrefixTokenResolution: + """``_rp_resolve_token`` / ``_rp_resolve_args`` — the walker token table.""" + + def test_constant_tokens_come_from_context(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_token + + ctx = _rp_ctx(tmp_path) + assert _rp_resolve_token("", ctx) == "/PLUGIN" + assert _rp_resolve_token("", ctx) == "opus" + assert _rp_resolve_token("", ctx) == "1700000000" + assert _rp_resolve_token("", ctx) == "0" + + def test_scope_derived_tokens(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_token + + (tmp_path / "scope.json").write_text(json.dumps({ + "diff_scope": "main...HEAD", + "base_ref": "main", + "diff_tip": "HEAD", + "scope_kind": "branch", + "review_root": "", + "review_branch": "feature", + })) + ctx = _rp_ctx(tmp_path) + assert _rp_resolve_token("", ctx) == "main...HEAD" + assert _rp_resolve_token("", ctx) == "main" + assert _rp_resolve_token("", ctx) == "HEAD" + assert _rp_resolve_token("", ctx) == "branch" + # STATE_KEY composes review_branch:base_ref. + assert _rp_resolve_token("", ctx) == "feature:main" + + def test_artifact_tokens_read_their_source_files(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_token + + (tmp_path / "cache_config.json").write_text(json.dumps({"cache_dir": "/c/dir"})) + (tmp_path / "hashes.json").write_text(json.dumps({ + "prompt_hash": "ph123", "context_key": "ck456", + })) + (tmp_path / "intent.json").write_text(json.dumps({"intent": "fix"})) + ctx = _rp_ctx(tmp_path) + assert _rp_resolve_token("", ctx) == "/c/dir" + assert _rp_resolve_token("", ctx) == "ph123" + assert _rp_resolve_token("", ctx) == "ck456" + assert _rp_resolve_token("", ctx) == "fix" + + def test_missing_source_file_degrades_to_empty(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_token + + ctx = _rp_ctx(tmp_path) # no scope/cache/hashes/intent files written + assert _rp_resolve_token("", ctx) == "" + assert _rp_resolve_token("", ctx) == "" + assert _rp_resolve_token("", ctx) == "" + # STATE_KEY with neither branch nor base ref is empty, not ":". + assert _rp_resolve_token("", ctx) == "" + + def test_override_takes_precedence_over_scope(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_token + + (tmp_path / "scope.json").write_text(json.dumps({"diff_scope": "orig"})) + ctx = _rp_ctx(tmp_path, overrides={"": "narrowed"}) + assert _rp_resolve_token("", ctx) == "narrowed" + + def test_unknown_token_raises(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_token + + with pytest.raises(KeyError): + _rp_resolve_token("", _rp_ctx(tmp_path)) + + def test_resolve_args_substitutes_every_occurrence(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_resolve_args + + ctx = _rp_ctx(tmp_path) + out = _rp_resolve_args( + ["--model", "", "--path=/x", "--plain"], ctx, + ) + assert out == ["--model", "opus", "--path=/PLUGIN/x", "--plain"] + + +class TestRunPrefixReconstructCompleted: + """``_rp_reconstruct_completed`` rebuilds the depends_on set across resume.""" + + def _stage(self, sid: str, outputs: list[str], **over: Any) -> dict[str, Any]: + stage: dict[str, Any] = { + "id": sid, "kind": "helper", "enabled": True, + "expected_outputs": outputs, + } + stage.update(over) + return stage + + def test_empty_when_no_resume(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_reconstruct_completed + + stages = [self._stage("stage_a", [])] + assert _rp_reconstruct_completed(stages, "") == set() + + def test_setup_always_completed_even_without_outputs(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_reconstruct_completed + + stages = [ + self._stage("stage_01_setup", []), + self._stage("stage_b", []), + ] + completed = _rp_reconstruct_completed(stages, "stage_b") + assert "stage_01_setup" in completed + + def test_present_outputs_marks_completed_absent_excluded(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_reconstruct_completed + + present = tmp_path / "present.json" + present.write_text("{}") + absent = tmp_path / "absent.json" + stages = [ + self._stage("stage_ok", [str(present)]), + self._stage("stage_fail", [str(absent)]), # continue-stage that failed + self._stage("stage_target", []), + ] + completed = _rp_reconstruct_completed(stages, "stage_target") + assert "stage_ok" in completed + assert "stage_fail" not in completed # missing output → dependents skip + + def test_disabled_stage_excluded(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_reconstruct_completed + + stages = [ + self._stage("stage_disabled", [], enabled=False), + self._stage("stage_target", []), + ] + assert "stage_disabled" not in _rp_reconstruct_completed(stages, "stage_target") + + def test_stops_at_resume_from(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_reconstruct_completed + + present = tmp_path / "p.json" + present.write_text("{}") + stages = [ + self._stage("stage_before", [str(present)]), + self._stage("stage_target", [str(present)]), + self._stage("stage_after", [str(present)]), + ] + completed = _rp_reconstruct_completed(stages, "stage_target") + # Only stages strictly before resume_from are reconstructed. + assert completed == {"stage_before"} + + +class TestRunPrefixNextStageAndSingleton: + def test_next_stage_id(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_next_stage_id + + stages = [{"id": "a"}, {"id": "b"}, {"id": "c"}] + assert _rp_next_stage_id(stages, "a") == "b" + assert _rp_next_stage_id(stages, "b") == "c" + assert _rp_next_stage_id(stages, "c") is None # last stage + assert _rp_next_stage_id(stages, "missing") is None + + def test_extract_signals_needs_agent_top_level_status(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_singleton_needs_agent + + ctx = _rp_ctx(tmp_path) + (tmp_path / "extract_signals_manifest.json").write_text( + json.dumps({"status": "needs_agent"}), + ) + assert _rp_singleton_needs_agent(ctx, "stage_11_extract_signals") is True + + def test_extract_signals_cache_hit_no_agent(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_singleton_needs_agent + + ctx = _rp_ctx(tmp_path) + (tmp_path / "extract_signals_manifest.json").write_text( + json.dumps({"status": "cache_hit"}), + ) + assert _rp_singleton_needs_agent(ctx, "stage_11_extract_signals") is False + + def test_coverage_critic_needs_agent_nested_status(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_singleton_needs_agent + + ctx = _rp_ctx(tmp_path) + (tmp_path / "coverage.json").write_text( + json.dumps({"critic": {"status": "needs_agent"}}), + ) + assert _rp_singleton_needs_agent(ctx, "stage_15_coverage_critic") is True + + def test_missing_manifest_no_agent(self, tmp_path: Path) -> None: + from code_review_helpers import _rp_singleton_needs_agent + + ctx = _rp_ctx(tmp_path) # no manifest on disk + assert _rp_singleton_needs_agent(ctx, "stage_11_extract_signals") is False + assert _rp_singleton_needs_agent(ctx, "stage_15_coverage_critic") is False + + +def _fake_stage_parser(func: Any) -> argparse.ArgumentParser: + """A one-subcommand parser (``fake``) whose handler the test controls.""" + parser = argparse.ArgumentParser(add_help=False) + sub = parser.add_subparsers(dest="command", required=True) + fake = sub.add_parser("fake") + fake.add_argument("--out", default="") + fake.set_defaults(func=func) + return parser + + +def _fake_func(rc: int = 0, writes: str | None = None) -> Any: + def _run(ns: argparse.Namespace) -> int: + if writes: + Path(writes).write_text("{}") + return rc + return _run + + +class TestExecuteStageInprocess: + """``_execute_stage_inprocess`` — steps 1-5 of the Walker Contract.""" + + def _stage(self, out_path: Path, **over: Any) -> dict[str, Any]: + stage: dict[str, Any] = { + "id": "stage_x", "kind": "helper", "subcommand": "fake", + "enabled": True, "depends_on": [], + "args": ["--out", str(out_path)], + "expected_outputs": [str(out_path)], + "on_failure": "abort", + } + stage.update(over) + return stage + + def test_success_marks_completed(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + parser = _fake_stage_parser(_fake_func(rc=0, writes=str(out))) + completed: set[str] = set() + status, msg = _execute_stage_inprocess( + self._stage(out), _rp_ctx(tmp_path), parser, completed, + ) + assert status == "ran" + assert msg is None + assert "stage_x" in completed + assert out.exists() + + def test_abort_on_nonzero_rc(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + parser = _fake_stage_parser(_fake_func(rc=1, writes=str(out))) + completed: set[str] = set() + status, msg = _execute_stage_inprocess( + self._stage(out), _rp_ctx(tmp_path), parser, completed, + ) + assert status == "failed_abort" + assert msg is not None + assert "stage_x" not in completed + + def test_abort_on_missing_output(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "never_written.json" + parser = _fake_stage_parser(_fake_func(rc=0, writes=None)) # rc 0 but no file + status, _msg = _execute_stage_inprocess( + self._stage(out), _rp_ctx(tmp_path), parser, set(), + ) + assert status == "failed_abort" + + def test_continue_does_not_emit_finding(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + parser = _fake_stage_parser(_fake_func(rc=1)) + status, _msg = _execute_stage_inprocess( + self._stage(out, on_failure="continue"), _rp_ctx(tmp_path), parser, set(), + ) + assert status == "failed_continue" + assert not list(tmp_path.glob("agent_*-failed.json")) + + def test_continue_with_coverage_gap_emits_agent_failure_finding( + self, tmp_path: Path, + ) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + parser = _fake_stage_parser(_fake_func(rc=1)) + status, _msg = _execute_stage_inprocess( + self._stage(out, id="stage_11_extract_signals", + on_failure="continue_with_coverage_gap"), + _rp_ctx(tmp_path), parser, set(), + ) + assert status == "failed_continue" + failed = tmp_path / "agent_stage_11_extract_signals-failed.json" + assert failed.exists() + payload = json.loads(failed.read_text()) + marker = payload["findings"][0]["system_marker"] + assert marker == "agent-failure" + assert payload["findings"][0]["finding_scope"] == "system" + + def test_coverage_gap_finding_threads_stage_stderr_diagnostic( + self, tmp_path: Path, + ) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + + def _fail_with_stderr(_ns: argparse.Namespace) -> int: + print("Error: taxonomy mismatch after edit", file=sys.stderr) + return 1 + + parser = _fake_stage_parser(_fail_with_stderr) + status, message = _execute_stage_inprocess( + self._stage(out, id="stage_14_resolve_coverage", + on_failure="continue_with_coverage_gap"), + _rp_ctx(tmp_path), parser, set(), + ) + assert status == "failed_continue" + # The stage's own stderr is attributed to the stage in the message ... + assert "taxonomy mismatch after edit" in (message or "") + # ... and folded into the emitted finding's explanation (not discarded). + payload = json.loads( + (tmp_path / "agent_stage_14_resolve_coverage-failed.json").read_text(), + ) + assert "taxonomy mismatch after edit" in payload["findings"][0]["explanation"] + + def test_skip_when_disabled(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + parser = _fake_stage_parser(_fake_func(rc=0, writes=str(out))) + completed: set[str] = set() + status, _msg = _execute_stage_inprocess( + self._stage(out, enabled=False), _rp_ctx(tmp_path), parser, completed, + ) + assert status == "skipped" + assert "stage_x" not in completed + assert not out.exists() # func never ran + + def test_skip_when_depends_on_unmet(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "o.json" + parser = _fake_stage_parser(_fake_func(rc=0, writes=str(out))) + status, _msg = _execute_stage_inprocess( + self._stage(out, depends_on=["missing_dep"]), + _rp_ctx(tmp_path), parser, set(), + ) + assert status == "skipped" + assert not out.exists() + + def test_setup_stage_is_marked_completed_not_rerun(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + def _boom(_ns: argparse.Namespace) -> int: + raise AssertionError("setup must not re-run inside run-prefix") + + parser = _fake_stage_parser(_boom) + completed: set[str] = set() + status, _msg = _execute_stage_inprocess( + {"id": "stage_01_setup", "kind": "helper", "subcommand": "fake", + "enabled": True, "depends_on": [], "args": [], "expected_outputs": []}, + _rp_ctx(tmp_path), parser, completed, + ) + assert status == "ran" + assert "stage_01_setup" in completed + + def test_non_helper_kind_aborts(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + parser = _fake_stage_parser(_fake_func()) + status, msg = _execute_stage_inprocess( + {"id": "stage_fleet", "kind": "agent_fleet", "enabled": True, + "depends_on": [], "args": [], "expected_outputs": []}, + _rp_ctx(tmp_path), parser, set(), + ) + assert status == "failed_abort" + assert "helper-only" in (msg or "") + + def test_auto_incremental_override_applied_on_success(self, tmp_path: Path) -> None: + from code_review_helpers import _execute_stage_inprocess + + out = tmp_path / "auto_incremental.json" + + def _write_scope(ns: argparse.Namespace) -> int: + out.write_text(json.dumps({"diff_scope": "abc123...HEAD"})) + return 0 + + parser = _fake_stage_parser(_write_scope) + ctx = _rp_ctx(tmp_path) + status, _msg = _execute_stage_inprocess( + self._stage(out, id="stage_07_auto_incremental"), ctx, parser, set(), + ) + assert status == "ran" + # The narrowed scope becomes the token for later stages. + assert ctx.overrides.get("") == "abc123...HEAD" + + +class TestCmdRunPrefixReturns: + """cmd_run_prefix's status-JSON returns for the paths fixtures don't isolate.""" + + def _write_plan( + self, cr_dir: Path, stages: list[dict[str, Any]], *, + flags: dict[str, Any] | None = None, + ) -> None: + cr_dir.mkdir(parents=True, exist_ok=True) + (cr_dir / "setup.json").write_text(json.dumps({ + "start_time": "1700000000", "global_cache": "0", + })) + (cr_dir / "run_plan.json").write_text(json.dumps({ + "flags": flags or {}, "stages": stages, "validation_gates": [], + })) + + def _ns(self, cr_dir: Path, **over: Any) -> argparse.Namespace: + kwargs: dict[str, Any] = { + "cr_dir": str(cr_dir), "resume_from": "", "plugin_root": "", + "model_id": "opus", "output": None, + } + kwargs.update(over) + return argparse.Namespace(**kwargs) + + def _run(self, cr_dir: Path, capsys: Any, **over: Any) -> dict[str, Any]: + from code_review_helpers import cmd_run_prefix + + rc = cmd_run_prefix(self._ns(cr_dir, **over)) + assert rc == 0 + return json.loads(capsys.readouterr().out) + + def test_error_on_aborting_stage( + self, tmp_path: Path, capsys: pytest.CaptureFixture[str], + ) -> None: + # A non-helper stage after setup aborts the walk deterministically. + self._write_plan(tmp_path, [ + {"id": "stage_01_setup", "kind": "helper", "subcommand": "setup", + "enabled": True, "depends_on": [], "expected_outputs": []}, + {"id": "stage_bad", "kind": "agent_fleet", "enabled": True, + "depends_on": [], "args": [], "expected_outputs": []}, + ]) + result = self._run(tmp_path, capsys) + assert result["next_action"] == "error" + assert result["failed_stage"] == "stage_bad" + assert result["message"] + + def test_ready_for_route_at_partition_boundary( + self, tmp_path: Path, capsys: pytest.CaptureFixture[str], + ) -> None: + # Reaching the partition stage stops the walk BEFORE running it (Phase 1). + self._write_plan(tmp_path, [ + {"id": "stage_01_setup", "kind": "helper", "subcommand": "setup", + "enabled": True, "depends_on": [], "expected_outputs": []}, + {"id": "stage_17_partition", "kind": "helper", "subcommand": "partition", + "enabled": True, "depends_on": [], "args": [], "expected_outputs": []}, + ]) + result = self._run(tmp_path, capsys) + assert result["next_action"] == "ready_for_route" + assert result["resume_stage"] == "stage_17_partition" + assert result["failed_stage"] is None + + def test_ready_for_route_at_end_when_no_partition( + self, tmp_path: Path, capsys: pytest.CaptureFixture[str], + ) -> None: + # A plan that ends before any partition stage completes cleanly. + self._write_plan(tmp_path, [ + {"id": "stage_01_setup", "kind": "helper", "subcommand": "setup", + "enabled": True, "depends_on": [], "expected_outputs": []}, + ]) + result = self._run(tmp_path, capsys) + assert result["next_action"] == "ready_for_route" + assert result["resume_stage"] is None + + def test_output_flag_writes_status_to_file(self, tmp_path: Path) -> None: + from code_review_helpers import cmd_run_prefix + + self._write_plan(tmp_path, [ + {"id": "stage_01_setup", "kind": "helper", "subcommand": "setup", + "enabled": True, "depends_on": [], "expected_outputs": []}, + ]) + out_path = tmp_path / "status.json" + rc = cmd_run_prefix(self._ns(tmp_path, output=str(out_path))) + assert rc == 0 + result = json.loads(out_path.read_text()) + assert result["next_action"] == "ready_for_route" diff --git a/plugins/code-review/tools/python/test_prefix_golden.py b/plugins/code-review/tools/python/test_prefix_golden.py index 25e9be8..d0d41c8 100644 --- a/plugins/code-review/tools/python/test_prefix_golden.py +++ b/plugins/code-review/tools/python/test_prefix_golden.py @@ -33,6 +33,8 @@ fast_path_fixture, hygiene_only_fixture, run_prefix_fixture, + run_prefix_fixture_subprocess, + run_prefix_fixture_via_runner, since_last_review_fixture, standard_fixture, ) @@ -244,6 +246,91 @@ def test_prefix_matches_golden(name: str, tmp_path: Path, update_golden: bool) - assert not diffs, f"[{name}] prefix artifact drift:\n" + "\n\n".join(diffs) +# --------------------------------------------------------------------------- +# Subprocess A/B parity oracle (PLN-1229 Phase 1) +# --------------------------------------------------------------------------- +# +# The refactor guarantee: run the deterministic prefix two ways and assert the +# artifacts are byte-identical (modulo review_id / timestamps / abs paths, which +# normalization scrubs). A-side is the subprocess-per-stage walk (what start.md +# does today); B-side is production ``run-prefix``. Both stop at the Phase-1 +# boundary (before stage_17_partition — route + partition are Phase 2), so the +# compared artifact set runs 01→cache_check. A and B implement the walk WRAPPER +# independently, so a shared wrapper bug cannot hide. + +# The pause sequence each fixture drives run-prefix through, as +# ``[:]`` per emitted segment. Pins the resumable +# 3-segment contract at the integration level (the two singletons almost always +# fire; hygiene-only is the one-segment Gate A exit). +_EXPECTED_SEGMENTS: dict[str, list[str]] = { + "golden_prefix_standard": ["needs_singleton:extract_signals", "ready_for_route"], + "golden_prefix_fast_path": ["needs_singleton:extract_signals", "ready_for_route"], + "golden_prefix_hygiene_only": ["hygiene_exit"], + "golden_prefix_empty_diff": ["needs_singleton:extract_signals", "ready_for_route"], + "golden_prefix_cache_hit": ["needs_singleton:extract_signals", "ready_for_route"], + "golden_prefix_since_last_review": [ + "needs_singleton:extract_signals", "ready_for_route", + ], + "golden_prefix_coverage_critic": [ + "needs_singleton:extract_signals", + "needs_singleton:coverage_critic", + "ready_for_route", + ], +} + + +def _segment_labels(statuses: list[dict[str, object]]) -> list[str]: + labels: list[str] = [] + for status in statuses: + action = str(status["next_action"]) + singleton = status.get("singleton") + labels.append(f"{action}:{singleton}" if singleton else action) + return labels + + +@pytest.mark.parametrize("name", _ALL_FIXTURES) +def test_run_prefix_matches_subprocess_walk(name: str, tmp_path: Path) -> None: + """``run-prefix`` (B) produces the same artifacts as the per-stage walk (A).""" + factory = _FIXTURE_FACTORIES[name] + a_snaps = run_prefix_fixture_subprocess(tmp_path / "a", factory()) + b_snaps, statuses = run_prefix_fixture_via_runner(tmp_path / "b", factory()) + + assert set(a_snaps) == set(b_snaps), ( + f"[{name}] artifact-set drift between per-stage walk and run-prefix:\n" + f" only in walk: {sorted(set(a_snaps) - set(b_snaps))}\n" + f" only in run-prefix:{sorted(set(b_snaps) - set(a_snaps))}" + ) + diffs: list[str] = [] + for artifact in sorted(a_snaps): + if a_snaps[artifact] != b_snaps[artifact]: + diffs.append( + "\n".join( + difflib.unified_diff( + a_snaps[artifact].splitlines(), + b_snaps[artifact].splitlines(), + fromfile=f"per-stage-walk/{artifact}", + tofile=f"run-prefix/{artifact}", + lineterm="", + ) + ) + ) + assert not diffs, f"[{name}] run-prefix artifact drift:\n" + "\n\n".join(diffs) + + +@pytest.mark.parametrize("name", _ALL_FIXTURES) +def test_run_prefix_pause_sequence(name: str, tmp_path: Path) -> None: + """``run-prefix`` pauses at exactly the expected decision points, in order.""" + _snaps, statuses = run_prefix_fixture_via_runner(tmp_path, _FIXTURE_FACTORIES[name]()) + assert _segment_labels(statuses) == _EXPECTED_SEGMENTS[name] + # The final segment resolves the pipeline (no dangling needs_singleton). + assert statuses[-1]["next_action"] in ("ready_for_route", "hygiene_exit") + assert statuses[-1]["failed_stage"] is None + # Every needs_singleton names the sibling consolidate stage to resume at. + for status in statuses: + if status["next_action"] == "needs_singleton": + assert str(status["resume_stage"]).endswith("_consolidate") + + # --------------------------------------------------------------------------- # Coverage guards # ---------------------------------------------------------------------------