fix(verification): restore manual checks and make plan resume idempotent

docs/verification.md

@@ -308,6 +308,21 @@ stops again without a retry loop. A moved HEAD, changed criteria, unbound comman
 hash or unrecoverable plan takes ordinary reconciliation instead. This does not
 add a CLI start point or change `START_ORDER`/`CYCLE_STARTS`.
 
+A manual check, whose command is `manual:<name>` (letters, digits, `.`, `_` and
+`-`, at most 64 characters after the prefix), survives resume. Restoring binds
+that identity from its journal proposal and nothing more: the command stays
+non-executable, is never a verify prefix, and the check stays `unverified` until
+evidence for the current HEAD says otherwise, so it still caps the conclusion.
+Any other command needs today's policy to authorize it, or the plan is not
+restored.
+
+Resume is idempotent. The configured test command is recognized in a stored plan
+by its plain command or by its command digest, so an unrestored plan, whose
+commands are `sha256:` fingerprints, does not gain a second
+`configured-test-<digest>` check. Consecutive resumes of the same change request
+keep the same check identities, results and obligations, and a frozen identity is
+never rewritten.
+
 Hotfix requires `hotfix-reproduction` and `hotfix-regression`, both `required`,
 with distinct focused commands proposed by the implementer. The configured
 general suite satisfies neither identity. A missing proposal stops with

scripts/cycle_plan.py

@@ -9,20 +9,22 @@
 
 if TYPE_CHECKING:
     from .cycle_journal import events as journal_events
-    from .cycle_history import HistoryError, restore_plan
+    from .cycle_history import HistoryError, command_digest, restore_plan
     from .execution_plan import ContractError, ExecutionPlan, ExecutionState, Obligation, Resolution, StagePolicy
     from .execution_profiles import ExecutionSettings, PlanSignals, build_plan, invocation_settings, resolve_execution
     from .security_gate import GateRule, evaluate
     from .stage_permissions import StagePolicy as CommandPolicy, _is_verification_command
-    from .verification_plan import CheckResult, VerificationError, VerificationPlan, configured_check
+    from .verification_plan import (CheckResult, VerificationError, VerificationPlan,
+                                    configured_check, is_manual_command)
 else:
     from cycle_journal import events as journal_events
-    from cycle_history import HistoryError, restore_plan
+    from cycle_history import HistoryError, command_digest, restore_plan
     from execution_plan import ContractError, ExecutionPlan, ExecutionState, Obligation, Resolution, StagePolicy
     from execution_profiles import ExecutionSettings, PlanSignals, build_plan, invocation_settings, resolve_execution
     from security_gate import GateRule, evaluate
     from stage_permissions import StagePolicy as CommandPolicy, _is_verification_command
-    from verification_plan import CheckResult, VerificationError, VerificationPlan, configured_check
+    from verification_plan import (CheckResult, VerificationError, VerificationPlan,
+                                   configured_check, is_manual_command)
 
 
 HOTFIX_IDS = ("hotfix-reproduction", "hotfix-regression")
@@ -220,7 +222,13 @@ def prepare_plan(self: Any):
                 self.state.verification_plan = replace(self.state.verification_plan, criteria=criteria)
         if self.start_from != "implement" and self.state.verification_plan is not None and self.allowed_test_command:
             existing = self.state.verification_plan.checks
-            if not any(check.command == self.allowed_test_command and check.required for check in existing):
+            # An unrestored plan keeps commands as fingerprints. Only the
+            # runtime's own configured-test identities may stand in by digest,
+            # so a proposed check cannot suppress the configured obligation.
+            fingerprint = command_digest(self.allowed_test_command)
+            if not any(check.required and (check.command == self.allowed_test_command
+                    or (check.command == fingerprint and check.id.startswith("configured-test")))
+                    for check in existing):
                 spec = configured_check(self.allowed_test_command, "runtime", "runtime")
                 if any(check.id == spec.id for check in existing):
                     spec = replace(spec, id="configured-test-" + hashlib.sha256(
@@ -241,8 +249,10 @@ def authorized_bindings(self: Any) -> tuple[str, ...]:
         for row in journal_events(self.telemetry.path, self.repo_id, task_id=self.task_id,
                                   kinds=("stage_result",)):
             command = row.get("command")
-            if (row.get("component") == "verification_command"
-                    and isinstance(command, str) and _is_verification_command(command, policy)):
+            if (row.get("component") == "verification_command" and isinstance(command, str)
+                    and (_is_verification_command(command, policy) or is_manual_command(command))):
+                # A bound manual command only restores an identity; it stays
+                # inert because `_is_verification_command` still rejects it.
                 commands.append(command)
         return tuple(dict.fromkeys(commands))
 
@@ -418,6 +428,9 @@ def verify_pending(self: Any):
         plan = self.state.verification_plan
         pending = [check for check in plan.checks if check.required
                    and plan.current(check.id, self.plan_head) == "unverified"] if plan else []
+        # A manual check has no command to run: it stays unverified and the
+        # barrier reports it. It never becomes a verify prefix.
+        pending = [check for check in pending if not is_manual_command(check.command)]
         if not pending:
             return self.verification_barrier()
         source = self.cwd

scripts/verification_plan.py

@@ -17,6 +17,14 @@
 TESTS_VERSION = 2
 CONCLUSIONS = ("failed", "not_verified", "verified_with_reservations", "verified")
 LEVELS = {"static", "test", "boundary"}
+# A manual check names evidence a person or a host supplies. It is an identity,
+# not a command: nothing may execute it, and restoring it never verifies it.
+MANUAL_COMMAND = re.compile(r"manual:[A-Za-z0-9][A-Za-z0-9._-]{0,63}")
+
+
+def is_manual_command(command: Any) -> bool:
+    """Whether a check command is the fixed `manual:<name>` grammar."""
+    return isinstance(command, str) and MANUAL_COMMAND.fullmatch(command) is not None
 
 
 class VerificationError(ValueError):

tests/test_run_cycle.py

@@ -363,6 +363,150 @@ def test_journal_command_proposals_require_current_authorization(self):
         self.assertNotIn(unsafe, candidates)
         self.assertNotIn(withdrawn, candidates)
 
+    def test_manual_proposals_bind_without_gaining_execution_authority(self):
+        from cycle_plan import PlanRuntime
+        from cycle_journal import Journal
+        from stage_permissions import StagePolicy, _is_verification_command
+        from types import SimpleNamespace
+        journal = Journal(self.store.path, "owner/api", "API-7", "cycle-manual")
+        accepted = ("manual:orca-live", "manual:A1.b_c")
+        refused = ("manual:", "manual:a b", "manual:x;echo secret", "manual:$(id)",
+                   "manual:orca-live\nmanual:two", "Manual:orca-live", "manual:" + "a" * 65)
+        for command in (*accepted, *refused):
+            journal.record("stage_result", component="verification_command", command=command)
+        session = SimpleNamespace(telemetry=self.store, repo_id="owner/api", task_id="API-7",
+                                  allowed_test_command="python3 -m unittest",
+                                  allowed_verify_commands=())
+        candidates = PlanRuntime.authorized_bindings(session)
+        for command in accepted:
+            self.assertIn(command, candidates)
+            self.assertFalse(_is_verification_command(command, StagePolicy(
+                role="verify", test_command=session.allowed_test_command,
+                verify_prefixes=session.allowed_verify_commands)))
+        for command in refused:
+            self.assertNotIn(command, candidates)
+
+    def manual_plan_fixture(self, *, proposal: bool):
+        """A persisted revision shaped like the blocked one: manual and collided ids."""
+        import cycle_history as ch
+        import verification_plan as vp
+        from cycle_journal import Journal
+        test = "python3 -m unittest discover -s tests"
+        digest = hashlib.sha256(test.encode()).hexdigest()[:16]
+        criteria = vp.Criteria("github", "b" * 64, "evaluable", ("works", "documented"))
+        checks = (
+            vp.Check("configured-test", test, "codex", "cheap_coder", "project", True,
+                     provenance="project", reason="configured test command"),
+            vp.Check("focused", "python3 -m unittest tests.focused", "codex", "cheap_coder",
+                     "project", True, criteria=("AC-1",)),
+            vp.Check("manual-orca-live", "manual:orca-live", "codex", "cheap_coder",
+                     "boundary", True, criteria=("AC-2",), level="boundary"),
+            vp.Check("configured-test-" + digest, test, "codex", "cheap_coder", "project", True,
+                     provenance="runtime", reason="runtime"))
+        verification = vp.VerificationPlan(criteria, checks)
+        verification = verification.record(vp.CheckResult("focused", "a" * 40, "pass", ({
+            "command": "python3 -m unittest tests.focused", "exit_code": 0, "executed": 3},)))
+        verification = verification.record(vp.CheckResult(
+            "manual-orca-live", "a" * 40, "unverified", cause="unverified"))
+        plan = replace(self.plan(start="rereview"), plan_id="plan-manual", head_sha="a" * 40,
+                       verification=verification)
+        ch.CycleHistoryStore(self.store.path).save_plan(
+            plan, cycle_id="cycle-before", change_request_id="4", provenance="stopped",
+            stop_reason="stage_not_completed")
+        journal = Journal(self.store.path, "owner/api", "API-7", "cycle-before")
+        for command in (test, "python3 -m unittest tests.focused", *(
+                ("manual:orca-live",) if proposal else ())):
+            journal.record("stage_result", component="verification_command", command=command)
+        return criteria, test
+
+    def resume_once(self, criteria, test):
+        import cycle_history as ch
+        report = self.run_cycle(
+            Talker("codex"), Talker("claude", block("BLOCKED")), start_from="rereview",
+            change_request_id="4", execution_plan=self.plan(start="rereview"),
+            allowed_test_command=test, allowed_verify_commands=("python3 -m unittest",),
+            criteria_reader=lambda *_args, **_kw: criteria,
+            work_item_provider="github", work_item_repository="owner/api")
+        record = ch.CycleHistoryStore(self.store.path).latest_plan("owner/api", "API-7", "4")
+        checks = {check["id"]: check for check in record["payload"]["plan"]["verification"]["checks"]}
+        results = [(item["id"], item["head_sha"], item["result"]) for item in
+                   record["payload"]["plan"]["verification"]["results"]]
+        return report, checks, results
+
+    def test_manual_check_survives_three_resumes_with_stable_identities(self):
+        for proposal in (True, False):
+            with self.subTest(proposal=proposal):
+                self.setUp()
+                criteria, test = self.manual_plan_fixture(proposal=proposal)
+                digest = hashlib.sha256(test.encode()).hexdigest()[:16]
+                seen = []
+                for _ in range(3):
+                    report, checks, results = self.resume_once(criteria, test)
+                    self.assertNotIn("frozen", report.stopped_because or "")
+                    self.assertNotIn("identity", report.stopped_because or "")
+                    seen.append((tuple(checks), results,
+                                 {key: (value["required"], value["criteria"])
+                                  for key, value in checks.items()}))
+                self.assertEqual(seen[0][0], seen[1][0])
+                self.assertEqual(seen[0][2], seen[1][2])
+                self.assertEqual(seen[1][0], seen[2][0])
+                self.assertEqual(seen[1][2], seen[2][2])
+                if proposal:
+                    # Bound: nothing is rebuilt, so the history is untouched.
+                    self.assertEqual(seen[0][1], seen[1][1])
+                    self.assertEqual(seen[1][1], seen[2][1])
+                self.assertEqual(set(seen[0][0]), {"configured-test", "focused",
+                                                   "manual-orca-live", "configured-test-" + digest})
+                self.assertIn(("focused", "a" * 40, "pass"), seen[0][1])
+                self.assertIn(("manual-orca-live", "a" * 40, "unverified"), seen[0][1])
+                self.assertEqual(seen[0][2]["manual-orca-live"], (True, ["AC-2"]))
+
+    def test_restored_manual_check_stays_unverified_and_caps_the_conclusion(self):
+        import cycle_history as ch
+        from stage_permissions import StagePolicy, _is_verification_command
+        criteria, test = self.manual_plan_fixture(proposal=True)
+        record = ch.CycleHistoryStore(self.store.path).latest_plan("owner/api", "API-7", "4")
+        commands = (test, "python3 -m unittest tests.focused", "manual:orca-live")
+        restored = ch.restore_plan(record, commands).verification
+        manual = next(check for check in restored.checks if check.id == "manual-orca-live")
+        self.assertEqual(manual.command, "manual:orca-live")
+        self.assertEqual(restored.current("manual-orca-live", "a" * 40), "unverified")
+        self.assertEqual(restored.cap("verified", "a" * 40), "not_verified")
+        self.assertFalse(_is_verification_command("manual:orca-live", StagePolicy(
+            role="verify", test_command=test, verify_prefixes=("python3 -m unittest",))))
+
+    def test_an_unauthorized_non_manual_command_still_fails_restore(self):
+        import cycle_history as ch
+        criteria, test = self.manual_plan_fixture(proposal=True)
+        record = ch.CycleHistoryStore(self.store.path).latest_plan("owner/api", "API-7", "4")
+        with self.assertRaises(ch.HistoryError):
+            ch.restore_plan(record, (test, "manual:orca-live"))
+
+    def test_a_proposed_check_cannot_stand_in_for_the_configured_test_by_digest(self):
+        import cycle_history as ch
+        import verification_plan as vp
+        test = "python3 -m unittest discover -s tests"
+        criteria = vp.Criteria("github", "b" * 64, "evaluable", ("works", "documented"))
+        # Unbound manual proposal: the plan is not restored and keeps fingerprints.
+        verification = vp.VerificationPlan(criteria, (
+            vp.Check("proposed", test, "codex", "cheap_coder", "project", True, criteria=("AC-1",)),
+            vp.Check("manual-orca-live", "manual:orca-live", "codex", "cheap_coder",
+                     "boundary", True, criteria=("AC-2",), level="boundary")))
+        plan = replace(self.plan(start="rereview"), plan_id="plan-alias", head_sha="a" * 40,
+                       verification=verification)
+        ch.CycleHistoryStore(self.store.path).save_plan(
+            plan, cycle_id="cycle-alias", change_request_id="4", provenance="stopped",
+            stop_reason="stage_not_completed")
+        _report, checks, _results = self.resume_once(criteria, test)
+        self.assertIn("configured-test", checks)
+        self.assertIn("proposed", checks)
+
+    def test_manual_command_grammar_is_fixed(self):
+        import verification_plan as vp
+        self.assertTrue(vp.is_manual_command("manual:orca-live"))
+        for command in ("python3 -m unittest", "manual:", "manual:a b", None, 3):
+            self.assertFalse(vp.is_manual_command(command))
+
     def test_new_sensitive_diff_promotes_a_previously_omitted_audit(self):
         from cycle_plan import PlanRuntime
         import execution_plan as ep