From 697dd49390bd7911daaabb056ab79f817a0c92ac Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 09:43:55 +0200 Subject: [PATCH] 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 | 1 + api/schemas/workplan.py | 3 +- scripts/consistency_check.py | 147 ++++++++++++++---- tests/test_consistency_check.py | 81 +++++++++- tests/test_routers_core.py | 26 ++++ .../STATE-WP-0079-retirement-strangler.md | 19 +++ 6 files changed, 245 insertions(+), 32 deletions(-) diff --git a/api/schemas/task.py b/api/schemas/task.py index ef1c608..2e4526d 100644 --- a/api/schemas/task.py +++ b/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 diff --git a/api/schemas/workplan.py b/api/schemas/workplan.py index 90be8b8..4258e5e 100644 --- a/api/schemas/workplan.py +++ b/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] = [] \ No newline at end of file + blocked_reasons: list[dict] = [] diff --git a/scripts/consistency_check.py b/scripts/consistency_check.py index ed60999..563f69c 100644 --- a/scripts/consistency_check.py +++ b/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]}…" ) diff --git a/tests/test_consistency_check.py b/tests/test_consistency_check.py index e32de95..5d3b5fd 100644 --- a/tests/test_consistency_check.py +++ b/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): diff --git a/tests/test_routers_core.py b/tests/test_routers_core.py index c1601f6..7f5592a 100644 --- a/tests/test_routers_core.py +++ b/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) diff --git a/workplans/STATE-WP-0079-retirement-strangler.md b/workplans/STATE-WP-0079-retirement-strangler.md index f3fe8e0..73a22d5 100644 --- a/workplans/STATE-WP-0079-retirement-strangler.md +++ b/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: