feat(consistency): rebuild deterministic projection IDs

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

api/schemas/task.py

@@ -19,6 +19,7 @@ class TaskStatusMixin(BaseModel):
 
 
 class TaskCreate(TaskStatusMixin, WorkplanIdCreateMixin):
+    id: uuid.UUID | None = None
     title: str
     description: str | None = None
     status: TaskStatus = TaskStatus.todo

api/schemas/workplan.py

@@ -29,6 +29,7 @@ class WorkplanStatusMixin(BaseModel):
 
 
 class WorkplanCreate(WorkplanStatusMixin):
+    id: uuid.UUID | None = None
     repo_id: uuid.UUID
     topic_id: uuid.UUID | None = None
     slug: str
@@ -121,4 +122,4 @@ class WorkplanWithDeps(WorkplanWithTaskCounts):
     """WorkplanWithTaskCounts enriched with dependency graph edges."""
     depends_on: list[WorkplanDepStub] = []
     blocks: list[WorkplanDepStub] = []
-    blocked_reasons: list[dict] = []
+    blocked_reasons: list[dict] = []

scripts/consistency_check.py

@@ -79,6 +79,7 @@ import socket
 import subprocess
 import sys
 import time
+import uuid
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from collections import Counter
@@ -118,6 +119,12 @@ from api.task_status import (  # noqa: E402
     normalize_task_status,
 )
 
+_WORK_RECORD_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
+
+
+def _derived_work_record_uuid(record_id: str) -> str:
+    return str(uuid.uuid5(_WORK_RECORD_NAMESPACE_UUID, f"helixforge\n{record_id}"))
+
 try:
     import yaml as _yaml
     _HAS_YAML = True
@@ -1320,14 +1327,40 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
 
         ws = _api_get(api_base, f"/workplans/{ws_id}")
         if ws is None:
-            # C-03: stale workstream reference
-            report.add(
-                severity="FAIL", check_id="C-03",
-                message=f"state_hub_workstream_id {ws_id[:8]}… not found in DB (stale reference)",
-                file_path=fname,
-                db_id=ws_id,
-                fixable=False,
-            )
+            wp_id = str(meta.get("id", "")).strip()
+            if wp_id and ws_id == _derived_work_record_uuid(wp_id):
+                # A deterministic file identifier missing from a replaceable
+                # projection is registration work, not a stale reference.
+                report.add(
+                    severity="WARN",
+                    check_id="C-06",
+                    message=(
+                        f"Derived workplan {ws_id[:8]}… is absent from this projection"
+                    ),
+                    file_path=fname,
+                    db_id=ws_id,
+                    fixable=True,
+                    _fix_context={
+                        "wp_file": str(wp_file),
+                        "meta": meta,
+                        "body": body,
+                        "repo_id": repo_id,
+                        "domain": file_domain,
+                        "repo_market_domain": repo_market_domain,
+                        "repo_slug": repo_slug,
+                        "desired_ws_id": ws_id,
+                    },
+                )
+            else:
+                # A non-derived missing identifier may be genuine stale state;
+                # preserve the conservative manual-review behavior.
+                report.add(
+                    severity="FAIL", check_id="C-03",
+                    message=f"state_hub_workstream_id {ws_id[:8]}… not found in DB (stale reference)",
+                    file_path=fname,
+                    db_id=ws_id,
+                    fixable=False,
+                )
             continue
 
         # C-09: repo mismatch — file is here but DB says different repo
@@ -1609,13 +1642,32 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
                 file_task_sh_ids.add(t_sh_id)
                 db_task = _api_get(api_base, f"/tasks/{t_sh_id}")
                 if db_task is None:
-                    report.add(
-                        severity="FAIL", check_id="C-03",
-                        message=f"state_hub_task_id {t_sh_id[:8]}… not found in DB",
-                        file_path=f"{fname}#{t_id}",
-                        db_id=t_sh_id,
-                        fixable=False,
-                    )
+                    if t_id and t_sh_id == _derived_work_record_uuid(t_id):
+                        report.add(
+                            severity="WARN",
+                            check_id="C-11",
+                            message=f"Derived task '{t_id}' is absent from this projection",
+                            file_path=f"{fname}#{t_id}",
+                            db_id=t_sh_id,
+                            fixable=True,
+                            _fix_context={
+                                "ws_id": ws_id,
+                                "ws_status": ws.get("status", ""),
+                                "task": task,
+                                "wp_file": str(wp_file),
+                                "meta": meta,
+                                "body": body,
+                                "desired_task_id": t_sh_id,
+                            },
+                        )
+                    else:
+                        report.add(
+                            severity="FAIL", check_id="C-03",
+                            message=f"state_hub_task_id {t_sh_id[:8]}… not found in DB",
+                            file_path=f"{fname}#{t_id}",
+                            db_id=t_sh_id,
+                            fixable=False,
+                        )
                     continue
                 # C-10 / C-15: task status drift. ADR-001: the file originates
                 # work. Same-rank wait/progress is not "DB ahead" — writing
@@ -2850,6 +2902,9 @@ def fix_repo(
                 repo_id_val = ctx["repo_id"]
                 body = ctx.get("body", "")
                 wp_id = str(meta.get("id", "")).strip()
+                desired_ws_id = str(
+                    ctx.get("desired_ws_id") or _derived_work_record_uuid(wp_id)
+                )
                 title = str(meta.get("title", "")).strip()
                 status = str(meta.get("status", "active")).strip()
                 status = normalise_workstream_status(status)
@@ -2900,16 +2955,23 @@ def fix_repo(
                             (w for w in existing if w.get("repo_id") == repo_id_val),
                             None,
                         )
-                        if existing_same_repo and existing_same_repo.get("title") == (title or wp_id):
+                        if (
+                            existing_same_repo
+                            and existing_same_repo.get("title") == (title or wp_id)
+                            and existing_same_repo.get("id") == desired_ws_id
+                        ):
                             ws_data = existing_same_repo
                             report.fixes_applied.append(
                                 f"C-06 reusing existing workstream {ws_data['id'][:8]}... for {wp_id}"
                             )
                             break
-                        last_error = f"slug {slug!r} already belongs to another workstream"
+                        last_error = (
+                            f"slug {slug!r} already belongs to a different workplan identity"
+                        )
                         continue
 
                     ws_data = _api_post(api_base, "/workplans", {
+                        "id": desired_ws_id,
                         "topic_id": topic_id,
                         "repo_id": repo_id_val,
                         "slug": slug,
@@ -2932,7 +2994,13 @@ def fix_repo(
                     continue
 
                 new_ws_id = ws_data["id"]
-                _add_frontmatter_field(wp_file, "state_hub_workstream_id", new_ws_id)
+                if new_ws_id != desired_ws_id:
+                    report.fixes_applied.append(
+                        f"C-06 FAIL {wp_id}: projection returned unexpected UUID {new_ws_id}"
+                    )
+                    continue
+                if not meta.get("state_hub_workstream_id"):
+                    _add_frontmatter_field(wp_file, "state_hub_workstream_id", new_ws_id)
                 report.fixes_applied.append(
                     f"C-06 fixed: created workstream {new_ws_id[:8]}… "
                     f"for {wp_id}, wrote ID to {wp_file.name}"
@@ -2950,7 +3018,14 @@ def fix_repo(
                     t_priority = str(task.get("priority", "medium")).strip()
                     if t_priority not in VALID_TASK_PRIORITIES:
                         t_priority = "medium"
+                    raw_task_id = task.get("state_hub_task_id")
+                    desired_task_id = (
+                        str(raw_task_id).strip().strip('"')
+                        if raw_task_id not in (None, "", "~", "null", "None", "none")
+                        else _derived_work_record_uuid(t_id)
+                    )
                     t_data = _api_post(api_base, "/tasks", {
+                        "id": desired_task_id,
                         "workplan_id": new_ws_id,
                         "title": str(task.get("title", t_id)).strip() or t_id,
                         "description": task.get("description") or None,
@@ -2960,11 +3035,17 @@ def fix_repo(
                     })
                     if t_data and "_error" not in t_data:
                         t_db_id = t_data["id"]
-                        injected = _inject_task_id_into_block(
-                            wp_file, "state_hub_task_id", t_db_id, t_id
-                        )
-                        if not injected:
-                            _inject_task_id_frontmatter_list(wp_file, t_db_id, t_id)
+                        if t_db_id != desired_task_id:
+                            report.fixes_applied.append(
+                                f"  ! task {t_id} returned unexpected UUID {t_db_id}"
+                            )
+                            continue
+                        if not raw_task_id:
+                            injected = _inject_task_id_into_block(
+                                wp_file, "state_hub_task_id", t_db_id, t_id
+                            )
+                            if not injected:
+                                _inject_task_id_frontmatter_list(wp_file, t_db_id, t_id)
                         report.fixes_applied.append(f"  + task {t_id} → {t_db_id[:8]}…")
                     elif t_data:
                         report.fixes_applied.append(
@@ -3115,7 +3196,11 @@ def fix_repo(
                     t_priority = str(task.get("priority", "medium")).strip()
                     if t_priority not in VALID_TASK_PRIORITIES:
                         t_priority = "medium"
+                    desired_task_id = str(
+                        ctx.get("desired_task_id") or _derived_work_record_uuid(t_id)
+                    )
                     t_data = _api_post(api_base, "/tasks", {
+                        "id": desired_task_id,
                         "workplan_id": ws_id,
                         "title": str(task.get("title", t_id)).strip() or t_id,
                         "description": task.get("description") or None,
@@ -3125,11 +3210,17 @@ def fix_repo(
                     })
                     if t_data:
                         t_db_id = t_data["id"]
-                        injected = _inject_task_id_into_block(
-                            wp_file, "state_hub_task_id", t_db_id, t_id
-                        )
-                        if not injected:
-                            _inject_task_id_frontmatter_list(wp_file, t_db_id, t_id)
+                        if t_db_id != desired_task_id:
+                            report.fixes_applied.append(
+                                f"C-11 FAIL: task '{t_id}' returned unexpected UUID {t_db_id}"
+                            )
+                            continue
+                        if not task.get("state_hub_task_id"):
+                            injected = _inject_task_id_into_block(
+                                wp_file, "state_hub_task_id", t_db_id, t_id
+                            )
+                            if not injected:
+                                _inject_task_id_frontmatter_list(wp_file, t_db_id, t_id)
                         report.fixes_applied.append(
                             f"C-11 fixed: task '{t_id}' → {t_db_id[:8]}…"
                         )

tests/test_consistency_check.py

@@ -36,6 +36,7 @@ from consistency_check import (
     _check_scope_freshness,
     _check_repo_manager_conformance,
     _detect_behind_remote,
+    _derived_work_record_uuid,
     _git_pull,
     _patch_frontmatter_field,
     _patch_task_status_in_file,
@@ -1271,6 +1272,78 @@ class TestC20DependencyDetection:
 
 
 class TestC06WorkstreamCreation:
+    def test_fix_repo_rebuilds_missing_projection_from_derived_file_ids(
+        self, tmp_path, monkeypatch
+    ):
+        repo = tmp_path / "repo"
+        workplans = repo / "workplans"
+        workplans.mkdir(parents=True)
+        workplan_id = _derived_work_record_uuid("STATE-WP-0001")
+        task_id = _derived_work_record_uuid("STATE-WP-0001-T01")
+        wp = workplans / "STATE-WP-0001-demo.md"
+        original = (
+            "---\n"
+            "id: STATE-WP-0001\n"
+            "type: workplan\n"
+            "title: Demo Workplan\n"
+            "domain: financials\n"
+            "repo: demo-repo\n"
+            "status: ready\n"
+            f'state_hub_workstream_id: "{workplan_id}"\n'
+            "---\n\n"
+            "## Implement Demo\n\n"
+            "```task\n"
+            "id: STATE-WP-0001-T01\n"
+            "status: todo\n"
+            "priority: high\n"
+            f'state_hub_task_id: "{task_id}"\n'
+            "```\n"
+        )
+        wp.write_text(original, encoding="utf-8")
+        created = []
+
+        def fake_get(_api_base, path, params=None, **_kwargs):
+            if path == "/repos/demo-repo":
+                import socket
+
+                return {
+                    "id": "repo-1",
+                    "slug": "demo-repo",
+                    "local_path": str(repo),
+                    "host_paths": {socket.gethostname(): str(repo)},
+                    "domain_slug": "financials",
+                }
+            if path == "/topics":
+                return [{"id": "topic-1", "domain_slug": "financials"}]
+            if path == f"/workplans/{workplan_id}":
+                return None
+            if path == "/workplans" and params and "slug" in params:
+                return []
+            if path == "/workplans" and params:
+                return []
+            return []
+
+        def fake_post(_api_base, path, body):
+            created.append((path, body))
+            return body
+
+        monkeypatch.setattr("consistency_check._api_get", fake_get)
+        monkeypatch.setattr("consistency_check._api_post", fake_post)
+        monkeypatch.setattr("consistency_check._api_patch", lambda *args, **kwargs: {"ok": True})
+        monkeypatch.setattr("consistency_check._detect_behind_remote", lambda _repo_path: False)
+        monkeypatch.setattr("consistency_check._detect_ahead_of_remote", lambda _repo_path: 0)
+        monkeypatch.setattr("consistency_check._write_custodian_brief", lambda *args, **kwargs: False)
+        monkeypatch.setattr("consistency_check._git_push", lambda _repo_path: (True, "pushed"))
+        monkeypatch.setenv("STATEHUB_REGISTRAR", "1")
+
+        report = fix_repo("http://unused", "demo-repo")
+
+        assert created[0][1]["id"] == workplan_id
+        assert created[1][1]["id"] == task_id
+        assert created[1][1]["workplan_id"] == workplan_id
+        assert wp.read_text(encoding="utf-8") == original
+        assert any("C-06 fixed" in fix for fix in report.fixes_applied)
+
     def test_fix_repo_uses_repo_qualified_slug_when_base_slug_is_taken(self, tmp_path, monkeypatch):
         repo = tmp_path / "repo"
         workplans = repo / "workplans"
@@ -1342,10 +1415,12 @@ class TestC06WorkstreamCreation:
         report = fix_repo("http://unused", "demo-repo")
 
         assert created_workstreams[0]["slug"] == "demo-repo-state-wp-0001"
-        assert created_tasks[0]["workplan_id"] == "new-ws"
+        expected_workplan_id = created_workstreams[0]["id"]
+        expected_task_id = created_tasks[0]["id"]
+        assert created_tasks[0]["workplan_id"] == expected_workplan_id
         patched = wp.read_text(encoding="utf-8")
-        assert 'state_hub_workstream_id: "new-ws"' in patched
-        assert 'state_hub_task_id: "new-task"' in patched
+        assert f'state_hub_workstream_id: "{expected_workplan_id}"' in patched
+        assert f'state_hub_task_id: "{expected_task_id}"' in patched
         assert any("C-06 fixed" in fix for fix in report.fixes_applied)
 
     def test_fix_repo_skips_c06_mint_when_not_registrar(self, tmp_path, monkeypatch):

tests/test_routers_core.py

@@ -7,6 +7,8 @@ the FastAPI ASGI app.
 """
 from __future__ import annotations
 
+import uuid
+
 import pytest
 
 
@@ -139,6 +141,30 @@ class TestTopics:
 # ---------------------------------------------------------------------------
 
 class TestWorkstreams:
+    async def test_create_preserves_authoritative_projection_ids(self, client):
+        await _create_domain(client)
+        repo = await _create_repo(client, slug="projection-id-repo")
+        workplan_id = str(uuid.uuid4())
+        workplan = await _create_workplan(
+            client,
+            repo["id"],
+            slug="projection-id-wp",
+            id=workplan_id,
+        )
+        assert workplan["id"] == workplan_id
+
+        task_id = str(uuid.uuid4())
+        response = await client.post(
+            "/tasks/",
+            json={
+                "id": task_id,
+                "workplan_id": workplan_id,
+                "title": "File-backed task",
+            },
+        )
+        assert response.status_code == 201, response.text
+        assert response.json()["id"] == task_id
+
     async def test_create_and_list_by_topic(self, client):
         await _create_domain(client)
         topic = await _create_topic(client)

workplans/STATE-WP-0079-retirement-strangler.md

@@ -196,6 +196,25 @@ The remaining gate is an operator-approved pilot ordering the central database
 transaction and authoritative file rewrite/reconciliation as one recoverable
 cutover unit.
 
+**Repo Manager pilot executed (2026-08-22):** the workstation projection was
+the only instance holding `RMGR-WP-0005`; the production/registrar projection
+had no matching row. A verified 4.4 MiB preapply PostgreSQL dump was retained at
+`/tmp/state-hub-rmgr-wp-0005-pilot-preapply.dump`. The live schema advanced to
+`b8d4f0a2c6e1`, then six workplan/task primary keys migrated in one transaction.
+All 12 child-task links, 12 workplan progress links, one decision link, and
+three task progress links followed. Six aliases are `applied`; all old rows are
+absent and all new rows present. Repo Manager committed the matching file
+rewrite in `5de754a`; two consistency runs produced a fresh, non-stale index and
+the new workplan API lookup returns 200 while the old lookup returns 404.
+
+The absent registrar row exposed the final rebuild gap before remote ingestion:
+create schemas did not accept authoritative UUIDs and consistency classified a
+missing derived row as an unfixable stale reference. The projection rebuild path
+now accepts file UUIDs, derives IDs when a live file has none, and registers a
+missing derived workplan/task without rewriting the authoritative identifier.
+Focused rebuild/API tests and the full State Hub suite pass (`624 passed`).
+Deploy and ingest the pilot on the registrar before widening the migration.
+
 **A2a executed (2026-08-20) — first live cutover slice.** Dual-run is on for the
 pilot repo: