feat(registrar): verify full empty-projection rebuild

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

src/repo_manager/cli.py

@@ -63,12 +63,18 @@ def main(argv: list[str] | None = None) -> int:
         help="Confirm that --api-base is the authoritative hub",
     )
     p_registrar.add_argument("--push", action="store_true", help="Push the registrar commit")
-    p_registrar.add_argument(
+    registrar_mode = p_registrar.add_mutually_exclusive_group()
+    registrar_mode.add_argument(
         "--repair-workplan",
         default=None,
         metavar="ID",
         help="Rebuild and verify one already-identified workplan projection",
     )
+    registrar_mode.add_argument(
+        "--bootstrap-empty-projection",
+        action="store_true",
+        help="Rebuild all authoritative UUIDs after proving the repo projection is empty",
+    )
 
     p_cmd = sub.add_parser(
         "update-task-status",
@@ -419,6 +425,7 @@ def main(argv: list[str] | None = None) -> int:
             confirm_primary=args.confirm_primary,
             push=args.push,
             repair_workplan=args.repair_workplan,
+            bootstrap_empty_projection=args.bootstrap_empty_projection,
         )
         print(json.dumps(result.to_dict(), indent=2))
         return 0 if result.status in {"applied", "noop"} else 1

src/repo_manager/commands/registrar_reconcile.py

@@ -169,6 +169,83 @@ def _projection_exists(api_base: str, projection_id: str) -> tuple[dict[str, Any
     return payload, None
 
 
+def _check_empty_repo_projection(
+    api_base: str, repo_slug: str
+) -> tuple[dict[str, Any], str | None]:
+    try:
+        repo_response = httpx.get(f"{api_base.rstrip('/')}/repos/{repo_slug}", timeout=10.0)
+        repo_response.raise_for_status()
+        repo_payload = repo_response.json()
+        response = httpx.get(
+            f"{api_base.rstrip('/')}/workplans/",
+            params={"repo_id": repo_payload["id"]},
+            timeout=10.0,
+        )
+        response.raise_for_status()
+        rows = response.json()
+    except (httpx.HTTPError, KeyError, ValueError) as exc:
+        return {}, f"empty projection preflight failed: {exc}"
+    if not isinstance(rows, list):
+        return {}, "empty projection preflight returned an invalid workplan collection"
+    evidence = {"repo_id": repo_payload["id"], "workplan_count": len(rows)}
+    if rows:
+        return evidence, "target repository projection is not empty"
+    return evidence, None
+
+
+def _authoritative_projection_ids(repo: Path) -> tuple[set[str], set[str], str | None]:
+    workplan_ids: set[str] = set()
+    task_ids: set[str] = set()
+    for path in sorted((repo / "workplans").glob("*.md")):
+        parsed = parse_workplan_file(path, repo_root=repo)
+        if parsed.frontmatter.get("type") != "workplan" or not parsed.id:
+            continue
+        if not parsed.state_hub_workstream_id:
+            return set(), set(), f"{parsed.id} has no authoritative projection UUID"
+        workplan_ids.add(parsed.state_hub_workstream_id)
+        for task in parsed.tasks:
+            if task.id and not task.state_hub_task_id:
+                return set(), set(), f"{task.id} has no authoritative projection UUID"
+            if task.state_hub_task_id:
+                task_ids.add(task.state_hub_task_id)
+    if not workplan_ids:
+        return set(), set(), "repository has no root workplans to rebuild"
+    return workplan_ids, task_ids, None
+
+
+def _verify_full_projection(
+    api_base: str, workplan_ids: set[str], task_ids: set[str]
+) -> tuple[dict[str, Any], str | None]:
+    missing_workplans: list[str] = []
+    missing_tasks: list[str] = []
+    try:
+        for projection_id in sorted(workplan_ids):
+            response = httpx.get(
+                f"{api_base.rstrip('/')}/workplans/{projection_id}", timeout=10.0
+            )
+            if response.status_code == 404:
+                missing_workplans.append(projection_id)
+            else:
+                response.raise_for_status()
+        for projection_id in sorted(task_ids):
+            response = httpx.get(f"{api_base.rstrip('/')}/tasks/{projection_id}", timeout=10.0)
+            if response.status_code == 404:
+                missing_tasks.append(projection_id)
+            else:
+                response.raise_for_status()
+    except httpx.HTTPError as exc:
+        return {}, f"full projection verification failed: {exc}"
+    evidence = {
+        "expected_workplans": len(workplan_ids),
+        "expected_tasks": len(task_ids),
+        "missing_workplans": missing_workplans,
+        "missing_tasks": missing_tasks,
+    }
+    if missing_workplans or missing_tasks:
+        return evidence, "full projection rebuild is incomplete"
+    return evidence, None
+
+
 def _run_statehub(command: list[str], *, env: dict[str, str]) -> subprocess.CompletedProcess[str]:
     return subprocess.run(
         command,
@@ -204,6 +281,7 @@ def registrar_reconcile(
     confirm_primary: bool = False,
     push: bool = False,
     repair_workplan: str | None = None,
+    bootstrap_empty_projection: bool = False,
 ) -> RegistrarResult:
     """Register missing workplan/task UUIDs through one scoped child process."""
     cid = str(uuid.uuid4())
@@ -217,6 +295,16 @@ def registrar_reconcile(
     }
 
     repair_projection_id = None
+    if repair_workplan and bootstrap_empty_projection:
+        return RegistrarResult(
+            "rejected",
+            evidence,
+            {
+                "code": "registrar_mode_conflict",
+                "message": "repair-workplan and bootstrap-empty-projection are mutually exclusive",
+            },
+            cid,
+        )
     if repair_workplan:
         repair_projection_id = _workplan_projection_id(repo, repair_workplan)
         evidence["repair_workplan"] = repair_workplan
@@ -242,7 +330,7 @@ def registrar_reconcile(
             },
             cid,
         )
-    if not any(before.values()) and not repair_projection_id:
+    if not any(before.values()) and not repair_projection_id and not bootstrap_empty_projection:
         evidence["missing_after"] = before
         return RegistrarResult("noop", evidence, None, cid)
 
@@ -269,6 +357,31 @@ def registrar_reconcile(
             cid,
         )
 
+    bootstrap_workplans: set[str] = set()
+    bootstrap_tasks: set[str] = set()
+    if bootstrap_empty_projection:
+        bootstrap_before, bootstrap_error = _check_empty_repo_projection(api_base, repo.name)
+        evidence["bootstrap_projection_before"] = bootstrap_before
+        if bootstrap_error:
+            return RegistrarResult(
+                "rejected",
+                evidence,
+                {"code": "bootstrap_precondition_failed", "message": bootstrap_error},
+                cid,
+            )
+        bootstrap_workplans, bootstrap_tasks, source_error = _authoritative_projection_ids(repo)
+        evidence["bootstrap_source"] = {
+            "workplans": len(bootstrap_workplans),
+            "tasks": len(bootstrap_tasks),
+        }
+        if source_error:
+            return RegistrarResult(
+                "rejected",
+                evidence,
+                {"code": "bootstrap_source_invalid", "message": source_error},
+                cid,
+            )
+
     LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
     with LOCK_PATH.open("a+", encoding="utf-8") as lock:
         try:
@@ -290,6 +403,8 @@ def registrar_reconcile(
             "--api-base",
             api_base.rstrip("/"),
         ]
+        if bootstrap_empty_projection:
+            command.append("--bootstrap-empty-projection")
         completed = _run_statehub(command, env=child_env)
 
     evidence["statehub_exit_code"] = completed.returncode
@@ -306,8 +421,21 @@ def registrar_reconcile(
             evidence["repair_projection_error"] = projection_error
         repair_verified = projection_error is None
 
+    bootstrap_verified = False
+    if bootstrap_empty_projection:
+        projection, projection_error = _verify_full_projection(
+            api_base,
+            bootstrap_workplans,
+            bootstrap_tasks,
+        )
+        evidence["bootstrap_projection"] = projection
+        evidence["bootstrap_projection_verified"] = projection_error is None
+        if projection_error:
+            evidence["bootstrap_projection_error"] = projection_error
+        bootstrap_verified = projection_error is None
+
     accepted_exit_codes = {0, 2}
-    if repair_verified:
+    if repair_verified or bootstrap_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)
@@ -324,7 +452,7 @@ def registrar_reconcile(
 
     changed = _git(repo, "status", "--porcelain")
     paths = [line[3:] for line in changed.stdout.splitlines() if len(line) > 3]
-    if repair_verified and paths:
+    if (repair_verified or bootstrap_verified) and paths:
         restored, restore_error = _restore_generated_brief(repo, paths)
         evidence["restored_generated_paths"] = restored
         if restore_error:

tests/test_registrar_reconcile.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 import subprocess
 from pathlib import Path
 
+import pytest
+
 from repo_manager.commands import registrar_reconcile as rr
 
 
@@ -185,3 +187,94 @@ def test_repairs_an_already_identified_workplan_projection(tmp_path: Path, monke
     assert result.evidence["repair_projection_id"] == "11111111-1111-4111-8111-111111111111"
     assert result.evidence["restored_generated_paths"] == [".custodian-brief.md"]
     assert brief.read_text(encoding="utf-8") == "authoritative local brief\n"
+
+
+def test_bootstraps_and_verifies_a_completely_empty_projection(tmp_path: Path, monkeypatch) -> None:
+    repo = _fixture(tmp_path)
+    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")
+    _git(repo, "add", ".")
+    _git(repo, "commit", "-m", "add authoritative identifiers")
+    _git(repo, "push")
+
+    monkeypatch.setattr(
+        rr,
+        "_check_primary",
+        lambda _api: ({"status": "ok", "db": "connected"}, None),
+    )
+    monkeypatch.setattr(
+        rr,
+        "_check_empty_repo_projection",
+        lambda _api, _slug: ({"repo_id": "repo-1", "workplan_count": 0}, None),
+    )
+    monkeypatch.setattr(
+        rr,
+        "_verify_full_projection",
+        lambda _api, workplans, tasks: (
+            {
+                "expected_workplans": len(workplans),
+                "expected_tasks": len(tasks),
+                "missing_workplans": [],
+                "missing_tasks": [],
+            },
+            None,
+        ),
+    )
+
+    def fake_run(command, *, env):
+        assert command[-1] == "--bootstrap-empty-projection"
+        assert env["STATEHUB_REGISTRAR"] == "1"
+        return subprocess.CompletedProcess(command, 2, "bootstrap warnings", "")
+
+    monkeypatch.setattr(rr, "_run_statehub", fake_run)
+
+    result = rr.registrar_reconcile(
+        repo,
+        statehub_bin="statehub",
+        confirm_primary=True,
+        bootstrap_empty_projection=True,
+    )
+
+    assert result.status == "applied"
+    assert result.evidence["bootstrap_source"] == {"workplans": 1, "tasks": 1}
+    assert result.evidence["bootstrap_projection_verified"] is True
+
+
+def test_empty_projection_bootstrap_refuses_existing_rows(tmp_path: Path, monkeypatch) -> None:
+    repo = _fixture(tmp_path)
+    monkeypatch.setattr(
+        rr,
+        "_check_primary",
+        lambda _api: ({"status": "ok", "db": "connected"}, None),
+    )
+    monkeypatch.setattr(
+        rr,
+        "_check_empty_repo_projection",
+        lambda _api, _slug: (
+            {"repo_id": "repo-1", "workplan_count": 1},
+            "target repository projection is not empty",
+        ),
+    )
+    monkeypatch.setattr(
+        rr,
+        "_run_statehub",
+        lambda *_args, **_kwargs: pytest.fail("non-empty bootstrap must not run State Hub"),
+    )
+
+    result = rr.registrar_reconcile(
+        repo,
+        confirm_primary=True,
+        bootstrap_empty_projection=True,
+    )
+
+    assert result.status == "rejected"
+    assert result.error and result.error["code"] == "bootstrap_precondition_failed"