Separate plan-review verdict policy from application confirmation

doc/architecture/SPEC/app/core.md

@@ -3,8 +3,9 @@
 <!-- verified-against: 2026-09-26 -->
 
 `Copilot` is the headless workflow application. It resolves a TaskSpec,
-enforces plan review and confirmation, executes or resumes a playbook, and
-reserves/executes durable runs. The CLI imports it through `cli.copilot` for
+delegates reviewer verdict policy to `review.plan_gate`, confirms when needed,
+executes or resumes a playbook, and reserves/executes durable runs. The CLI
+imports it through `cli.copilot` for
 compatibility; the application never imports CLI or MCP transports.
 
 The pre-execution gate runs before a normal CLI run directory is created.
@@ -14,5 +15,7 @@ binding; a run ID must stay contained under the configured run root. Existing
 checkpoint keys, run IDs, and blocked exit codes remain stable.
 
 `GateOutcome`, `_mode_review_context`, and `Copilot` retain their import path
-through the compatibility module. Only orchestration belongs here; command
-argument parsing and stdio protocol registration remain transport concerns.
+through the compatibility module. The application passes its reviewer to the
+plan gate, preserving the no-human fail-closed rule. Only orchestration belongs
+here; command argument parsing and stdio protocol registration remain
+transport concerns.

doc/architecture/SPEC/review/plan_gate.md

@@ -0,0 +1,20 @@
+# review/plan_gate.py — plan-review policy
+
+<!-- verified-against: 2026-09-26 -->
+
+This component builds the authoritative mode-aware context for a planned
+playbook and interprets the injected plan reviewer's verdict. Exact reused
+plans that do not require review pass without a reviewer call. A `block`
+verdict always stops; `revise` or `unavailable` stops an unattended `--yes`
+run because no person remains to inspect the plan. Interactive runs surface
+those verdicts and leave final confirmation to `app.core`.
+
+The mode context uses the executor's `when` evaluation and resolved rebase
+mode, so inactive write/push steps are not presented as active. It does not
+execute a plan, print notices, prompt for confirmation, create a run directory,
+or own a transport. It returns a `PlanGateResult` with the decision and notices;
+`app.core` renders them and keeps `_mode_review_context` and
+`_plan_review_gate` compatibility entry points while supplying the reviewer.
+
+Verification: `test_planner_playbooks.py`, `test_curator.py`, and
+`test_phase_b.py` cover verdicts, mode context, and the application gate.

src/infermatrix_copilot/app/core.py

@@ -17,8 +17,6 @@
 from pathlib import Path
 from typing import NamedTuple
 
-import yaml
-
 from .. import idempotency as idem
 from .. import run_status as rs
 from ..config import Settings, TierNotConfiguredError
@@ -31,6 +29,7 @@
 from ..notify import BLOCKED_EXIT, Notifier
 from ..playbooks.store import PlaybookStore, parse_playbook, playbook_to_doc
 from ..push import PushPolicy
+from ..review.plan_gate import _mode_review_context, review_plan_gate
 from ..review.reviewer import run_plan_review
 from ..run_trace import RunTrace
 from ..task_spec import TaskSpec
@@ -57,43 +56,6 @@ def stop(cls, exit_code: int) -> "GateOutcome":
         return cls(proceed=False, exit_code=exit_code)
 
 
-def _mode_review_context(playbook, spec) -> str:
-    """Plan-review context for MODE-AWARE playbooks: the reviewer sees the
-    raw yaml's FULL step list, but the `when:` gates resolve against the
-    ALREADY-RESOLVED mode (`resolve_effective_mode` runs before the review
-    gate) — without this context a report-only plan looks like it runs its
-    write/push steps and gets spuriously blocked. The active-step set is
-    the same mechanical truth the executor computes, never prose."""
-    if not getattr(playbook, "mode_aware", False):
-        return ""
-    mode = str((getattr(spec, "params", None) or {})
-               .get("rebase_mode", "") or "")
-    if not mode:
-        return ""
-    from ..engine.executor import _eval_when
-    from ..rebase_engine.modes import mode_state_flags
-
-    flags = {"task_spec": {}, **mode_state_flags(mode)}
-    active = [s.get("id", s.get("step", "?"))
-              for s in playbook_to_doc(playbook).get("steps", [])
-              if "when" not in s or _eval_when(s["when"], flags)]
-    repo = str(getattr(spec, "repo", "") or "")
-    repo_line = (f"\nTarget repo (authoritative): {repo!r} — bound at "
-                 "runtime from the TaskSpec; the yaml `repos:` list is a "
-                 "planner RECALL FILTER where empty means repo-neutral, "
-                 "never untargeted." if repo else "")
-    return (f"\n\nResolved mode context (authoritative): "
-            f"rebase_mode={mode}. Under this mode the `when:` gates run "
-            f"ONLY these steps: {active}. Every other listed step is "
-            "statically gated OFF for this run — judge the plan for THIS "
-            "mode's step set."
-            + repo_line +
-            "\nWrite/push governance (authoritative): the mode's own "
-            "push/CI steps are governed at runtime by the push-gate "
-            "ruling, guard_push, and the ALLOW_PUSH env double-gate — "
-            "the task tier does not forbid steps this mode activates.")
-
-
 class Copilot:
     """Orchestration core: resolves a TaskSpec to a playbook (reuse > adapt >
     generate), runs it through the plan-review gate + confirmation into the
@@ -165,42 +127,14 @@ def resolve(self, spec: TaskSpec) -> Resolution:
 
     def _plan_review_gate(self, resolution: Resolution, spec: TaskSpec,
                           assume_yes: bool) -> bool:
-        """Inline Plan-Review for adapted/generated plans. LLM verdict shown in
-        the session; block stops. A non-`lgtm` verdict is only ever *surfaced*
-        — the human `[y/N]` is what actually gates it — so when `--yes` removes
-        that human, the same verdict must stop the run instead
-        (`ReviewVerdict.passing`, SPEC C6: only `lgtm` passes)."""
-        if not resolution.requires_review:
-            return True
-        doc = yaml.safe_dump(playbook_to_doc(resolution.playbook), sort_keys=False)
-        task_text = spec.describe() + _mode_review_context(
-            resolution.playbook, spec)
-        verdict = run_plan_review(self.llm, playbook_doc=doc,
-                                  task=task_text,
-                                  model=self.settings.reviewer)
-        if verdict.verdict != "unavailable":
-            print(f"  plan review: {verdict.verdict}"
-                  + (f" — {verdict.critiques}" if verdict.critiques else ""))
-        if verdict.verdict == "block":
-            print("✋ plan blocked by reviewer.")
-            return False
-        if verdict.passing:
-            return True
-        # Everything below is non-passing: `revise`, or `unavailable`. Both used
-        # to return True on the strength of a confirmation that `--yes` had
-        # already deleted, so an unattended run executed an unvetted plan on an
-        # unread verdict. Measured on the release matrix: an unparseable review
-        # let three of four backends run a pr-rebase plan through to its push
-        # gate, while the one backend whose reviewer parsed cleanly BLOCKED the
-        # very same plan.
-        if assume_yes:
-            reason = ("no reviewer LLM" if verdict.verdict == "unavailable"
-                      else f"plan review returned {verdict.verdict}")
-            print(f"✋ {reason} and --yes leaves no human to gate it — blocked.")
-            return False
-        if verdict.verdict == "unavailable":
-            print("  ⚠ no reviewer LLM — your confirmation is the plan-review gate")
-        return True  # revise/unavailable, surfaced to the user before their confirm
+        """Apply the reviewer verdict policy before human confirmation."""
+        result = review_plan_gate(
+            resolution, spec, self.llm, self.settings.reviewer, assume_yes,
+            review_fn=run_plan_review,
+        )
+        for notice in result.notices:
+            print(notice)
+        return result.allowed
 
     def _gate_and_confirm(self, resolution: Resolution, spec: TaskSpec,
                           assume_yes: bool, *, prompt: str = "Proceed?",

src/infermatrix_copilot/review/plan_gate.py

@@ -0,0 +1,85 @@
+"""Plan-review verdict policy and mode-aware context for generated playbooks.
+
+The application supplies its reviewer and retains the human confirmation gate.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import yaml
+
+from ..playbooks.store import playbook_to_doc
+
+
+def _mode_review_context(playbook, spec) -> str:
+    """Plan-review context for MODE-AWARE playbooks: the reviewer sees the
+    raw yaml's FULL step list, but the `when:` gates resolve against the
+    ALREADY-RESOLVED mode (`resolve_effective_mode` runs before the review
+    gate) — without this context a report-only plan looks like it runs its
+    write/push steps and gets spuriously blocked. The active-step set is
+    the same mechanical truth the executor computes, never prose."""
+    if not getattr(playbook, "mode_aware", False):
+        return ""
+    mode = str((getattr(spec, "params", None) or {})
+               .get("rebase_mode", "") or "")
+    if not mode:
+        return ""
+    from ..engine.executor import _eval_when
+    from ..rebase_engine.modes import mode_state_flags
+
+    flags = {"task_spec": {}, **mode_state_flags(mode)}
+    active = [s.get("id", s.get("step", "?"))
+              for s in playbook_to_doc(playbook).get("steps", [])
+              if "when" not in s or _eval_when(s["when"], flags)]
+    repo = str(getattr(spec, "repo", "") or "")
+    repo_line = (f"\nTarget repo (authoritative): {repo!r} — bound at "
+                 "runtime from the TaskSpec; the yaml `repos:` list is a "
+                 "planner RECALL FILTER where empty means repo-neutral, "
+                 "never untargeted." if repo else "")
+    return (f"\n\nResolved mode context (authoritative): "
+            f"rebase_mode={mode}. Under this mode the `when:` gates run "
+            f"ONLY these steps: {active}. Every other listed step is "
+            "statically gated OFF for this run — judge the plan for THIS "
+            "mode's step set."
+            + repo_line +
+            "\nWrite/push governance (authoritative): the mode's own "
+            "push/CI steps are governed at runtime by the push-gate "
+            "ruling, guard_push, and the ALLOW_PUSH env double-gate — "
+            "the task tier does not forbid steps this mode activates.")
+
+
+@dataclass(frozen=True)
+class PlanGateResult:
+    allowed: bool
+    notices: tuple[str, ...] = ()
+
+
+def review_plan_gate(
+    resolution, spec, llm, model, assume_yes, *, review_fn
+) -> PlanGateResult:
+    """Interpret a reviewer verdict without owning confirmation or output."""
+    if not resolution.requires_review:
+        return PlanGateResult(True)
+    doc = yaml.safe_dump(playbook_to_doc(resolution.playbook), sort_keys=False)
+    task_text = spec.describe() + _mode_review_context(resolution.playbook, spec)
+    verdict = review_fn(llm, playbook_doc=doc, task=task_text, model=model)
+    notices = []
+    if verdict.verdict != "unavailable":
+        notices.append(f"  plan review: {verdict.verdict}"
+                       + (f" — {verdict.critiques}" if verdict.critiques else ""))
+    if verdict.verdict == "block":
+        notices.append("✋ plan blocked by reviewer.")
+        return PlanGateResult(False, tuple(notices))
+    if verdict.passing:
+        return PlanGateResult(True, tuple(notices))
+    # A non-passing verdict can only proceed through explicit human review.
+    # --yes removes that confirmation, so revise/unavailable must block.
+    if assume_yes:
+        reason = ("no reviewer LLM" if verdict.verdict == "unavailable"
+                  else f"plan review returned {verdict.verdict}")
+        notices.append(f"✋ {reason} and --yes leaves no human to gate it — blocked.")
+        return PlanGateResult(False, tuple(notices))
+    if verdict.verdict == "unavailable":
+        notices.append("  ⚠ no reviewer LLM — your confirmation is the plan-review gate")
+    return PlanGateResult(True, tuple(notices))