fix(registrar): verify scoped success amid drift

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d

src/repo_manager/commands/registrar_reconcile.py

@@ -287,6 +287,58 @@ def _verify_full_projection(
     return evidence, None
 
 
+def _requested_projection_ids(
+    repo: Path,
+    requested: dict[str, list[str]],
+) -> tuple[set[str], set[str], dict[str, set[str]], str | None]:
+    """Resolve only the identifiers requested by this registrar invocation."""
+    wanted_workplans = set(requested["workplans"])
+    wanted_tasks = set(requested["tasks"])
+    wanted_records = {
+        "intake": set(requested["intakes"]),
+        "decision": set(requested["decisions"]),
+    }
+    resolved_workplans: dict[str, str] = {}
+    resolved_tasks: dict[str, str] = {}
+    resolved_records: dict[str, dict[str, str]] = {"intake": {}, "decision": {}}
+
+    for path in sorted((repo / "workplans").glob("*.md")):
+        parsed = parse_workplan_file(path, repo_root=repo)
+        if parsed.id in wanted_workplans and parsed.state_hub_workstream_id:
+            resolved_workplans[parsed.id] = parsed.state_hub_workstream_id
+        for task in parsed.tasks:
+            if task.id in wanted_tasks and task.state_hub_task_id:
+                resolved_tasks[task.id] = task.state_hub_task_id
+
+    for path in iter_record_files(repo):
+        for record in parse_record_file(path, repo_root=repo):
+            if record.kind in wanted_records and record.id in wanted_records[record.kind] and record.uuid:
+                resolved_records[record.kind][record.id] = record.uuid
+
+    unresolved = sorted(
+        (wanted_workplans - resolved_workplans.keys())
+        | (wanted_tasks - resolved_tasks.keys())
+        | (wanted_records["intake"] - resolved_records["intake"].keys())
+        | (wanted_records["decision"] - resolved_records["decision"].keys())
+    )
+    record_ids = {
+        kind: set(records.values()) for kind, records in resolved_records.items()
+    }
+    if unresolved:
+        return (
+            set(resolved_workplans.values()),
+            set(resolved_tasks.values()),
+            record_ids,
+            f"requested records have no assigned projection UUID: {', '.join(unresolved)}",
+        )
+    return (
+        set(resolved_workplans.values()),
+        set(resolved_tasks.values()),
+        record_ids,
+        None,
+    )
+
+
 def _run_statehub(command: list[str], *, env: dict[str, str]) -> subprocess.CompletedProcess[str]:
     return subprocess.run(
         command,
@@ -484,8 +536,31 @@ def registrar_reconcile(
             evidence["bootstrap_projection_error"] = projection_error
         bootstrap_verified = projection_error is None
 
+    requested_verified = False
+    if completed.returncode == 1 and not any(after.values()):
+        (
+            requested_workplans,
+            requested_tasks,
+            requested_records,
+            requested_source_error,
+        ) = _requested_projection_ids(repo, before)
+        if requested_source_error:
+            evidence["requested_projection_error"] = requested_source_error
+        else:
+            projection, projection_error = _verify_full_projection(
+                api_base,
+                requested_workplans,
+                requested_tasks,
+                requested_records,
+            )
+            evidence["requested_projection"] = projection
+            evidence["requested_projection_verified"] = projection_error is None
+            if projection_error:
+                evidence["requested_projection_error"] = projection_error
+            requested_verified = projection_error is None
+
     accepted_exit_codes = {0, 2}
-    if repair_verified or bootstrap_verified:
+    if repair_verified or bootstrap_verified or requested_verified:
         # A repository-scoped projection repair may coexist with legacy stale
         # references that correctly keep the broader consistency report red.
         accepted_exit_codes.add(1)

tests/test_registrar_reconcile.py

@@ -155,6 +155,61 @@ def test_scopes_registrar_env_and_commits_assigned_ids(tmp_path: Path, monkeypat
     assert subject == "chore(registrar): assign State Hub identifiers"
 
 
+def test_accepts_unrelated_assessment_fail_after_exact_requested_verification(
+    tmp_path: Path, monkeypatch
+) -> None:
+    repo = _fixture(tmp_path)
+    monkeypatch.setattr(
+        rr,
+        "_check_primary",
+        lambda _api: ({"status": "ok", "db": "connected"}, None),
+    )
+
+    def fake_run(command, *, env):
+        workplan = repo / "workplans" / "DEMO-WP-0001.md"
+        text = workplan.read_text(encoding="utf-8")
+        text = text.replace(
+            "status: active\n---",
+            'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
+        )
+        text = text.replace(
+            "priority: high\n```",
+            'priority: high\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```',
+        )
+        workplan.write_text(text, encoding="utf-8")
+        return subprocess.CompletedProcess(command, 1, "unrelated C-03 remains", "")
+
+    monkeypatch.setattr(rr, "_run_statehub", fake_run)
+    monkeypatch.setattr(
+        rr,
+        "_verify_full_projection",
+        lambda _api, workplans, tasks, records: (
+            {
+                "expected_workplans": len(workplans),
+                "expected_tasks": len(tasks),
+                "expected_intakes": len(records["intake"]),
+                "expected_decisions": len(records["decision"]),
+                "missing_workplans": [],
+                "missing_tasks": [],
+                "missing_intakes": [],
+                "missing_decisions": [],
+            },
+            None,
+        ),
+    )
+
+    result = rr.registrar_reconcile(
+        repo,
+        statehub_bin="statehub",
+        confirm_primary=True,
+    )
+
+    assert result.status == "applied"
+    assert result.evidence["requested_projection_verified"] is True
+    assert result.evidence["requested_projection"]["expected_workplans"] == 1
+    assert result.evidence["requested_projection"]["expected_tasks"] == 1
+
+
 def test_repairs_an_already_identified_workplan_projection(tmp_path: Path, monkeypatch) -> None:
     repo = _fixture(tmp_path)
     workplan = repo / "workplans" / "DEMO-WP-0001.md"