fix(retirement): close projection and launch contract gaps

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02b7c-1c49-76a0-955a-49e7b3ddfc0d

tests/test_routers_core.py

@@ -1006,7 +1006,7 @@ class TestReconciliationEndpoints:
 
 
 class TestExecutionQueueEndpoints:
-    async def test_execution_semantics_separates_state_hub_and_activity_core(self, client):
+    async def test_execution_semantics_retires_workplan_launch_pickup(self, client):
         r = await client.get("/execution/semantics")
 
         assert r.status_code == 200
@@ -1014,8 +1014,15 @@ class TestExecutionQueueEndpoints:
         assert "queued" in body["execution_states"]
         assert "immediate" in body["launch_modes"]
         assert "parallel" in body["concurrency_modes"]
-        assert any("launch requests" in item for item in body["state_hub_responsibility"])
-        assert any("dispatch" in item for item in body["activity_core_responsibility"])
+        assert body["launch_requests_accepted"] is False
+        assert "authoritative repository file" in body["replacements"][
+            "POST /execution/launch-requests"
+        ]
+        assert any(
+            "reject new workplan launch" in item
+            for item in body["state_hub_responsibility"]
+        )
+        assert any("do not consume" in item for item in body["activity_core_responsibility"])
 
     async def test_execution_intent_update_does_not_change_lifecycle_status(self, client):
         await _create_domain(client)
@@ -1107,7 +1114,7 @@ class TestExecutionQueueEndpoints:
         assert rows[2]["eligible"] is False
         assert rows[2]["blocked_by_workstream_ids"] == [dependency["id"]]
 
-    async def test_launch_request_records_handoff_and_updates_execution_intent(self, client):
+    async def test_launch_request_is_gone_and_does_not_update_execution_intent(self, client):
         await _create_domain(client)
         topic = await _create_topic(client)
         ws = await _create_workstream(client, topic["id"], status="ready")
@@ -1124,21 +1131,18 @@ class TestExecutionQueueEndpoints:
             "notes": "start now",
         })
 
-        assert r.status_code == 201, r.text
-        body = r.json()
-        assert body["workstream_id"] == ws["id"]
-        assert body["launch_mode"] == "immediate"
-        assert body["immediate_pickup"] is True
-        assert body["status"] == "requested"
+        assert r.status_code == 410, r.text
+        assert "no consumer picks up these rows" in r.json()["detail"]
+        assert "repo file queue" in r.headers["x-statehub-replacement"]
 
         r = await client.get(f"/workplans/{ws['id']}")
         updated = r.json()
         assert updated["status"] == "ready"
-        assert updated["execution_state"] == "launching"
-        assert updated["launch_mode"] == "immediate"
+        assert updated["execution_state"] == "manual"
+        assert updated["launch_mode"] == "manual"
 
         r = await client.get(f"/execution/launch-requests?workstream_id={ws['id']}")
-        assert len(r.json()) == 1
+        assert r.json() == []
 
 
 def _fabric_graph_export(generated_at="2026-05-23T12:00:00Z", extra_node=False):

tests/test_work_record_identifier_migration.py

@@ -13,6 +13,7 @@ from api.models.domain import Domain
 from api.models.managed_repo import ManagedRepo
 from api.models.progress_event import ProgressEvent
 from api.models.task import Task
+from api.models.topic import Topic
 from api.models.token_event import TokenEvent
 from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias
 from api.models.workplan import Workplan
@@ -21,6 +22,7 @@ from api.services.work_record_identifier_migration import (
     DERIVATION_NAMESPACE_UUID,
     IdentifierMigrationError,
     apply_repository_identifier_migration,
+    repair_absent_prederivation_projection,
     reverse_repository_identifier_migration,
 )
 
@@ -296,3 +298,233 @@ async def test_repository_migration_is_atomic_when_a_source_is_missing(test_engi
         assert await session.scalar(
             text("SELECT count(*) FROM work_record_identifier_aliases")
         ) == 0
+
+
+async def _seed_empty_repository(factory, repo_slug: str):
+    domain_id = uuid.uuid4()
+    topic_id = uuid.uuid4()
+    repo_id = uuid.uuid4()
+    async with factory() as session:
+        session.add(Domain(id=domain_id, slug="infotech", name="Infotech", status="active"))
+        await session.flush()
+        session.add(
+            Topic(
+                id=topic_id,
+                domain_id=domain_id,
+                slug="infotech",
+                title="Infotech",
+                status="active",
+            )
+        )
+        await session.flush()
+        session.add(
+            ManagedRepo(
+                id=repo_id,
+                domain_id=domain_id,
+                topic_id=topic_id,
+                slug=repo_slug,
+                name="Repair Repository",
+                status="active",
+            )
+        )
+        await session.commit()
+    return repo_id, topic_id
+
+
+def _repair_unit(plan: dict, topic_id: uuid.UUID) -> dict:
+    repository = plan["repositories"][0]
+    replacements = [m for m in repository["mappings"] if m["action"] == "replace"]
+    workplan = next(m for m in replacements if m["kind"] == "workplan")
+    task = next(m for m in replacements if m["kind"] == "task")
+    return {
+        "workplan": {
+            "record_id": workplan["record_id"],
+            "id": workplan["current_uuid"],
+            "topic_id": str(topic_id),
+            "slug": "test-wp-0001",
+            "title": "Migration test",
+            "description": "Exact authoritative projection",
+            "status": "active",
+            "owner": "codex",
+            "planning_priority": "high",
+            "planning_order": 1,
+        },
+        "tasks": [
+            {
+                "record_id": task["record_id"],
+                "id": task["current_uuid"],
+                "title": "Mapped task",
+                "description": "Exact task projection",
+                "status": "progress",
+                "priority": "high",
+                "assignee": "codex",
+            }
+        ],
+    }
+
+
+def _repair_kwargs(plan: dict) -> dict:
+    repository = plan["repositories"][0]
+    return {
+        "expected_plan_sha256": plan["plan_sha256"],
+        "source_revision": repository["planned_head_sha"],
+        "source_fingerprint": repository["source_fingerprint"],
+        "source_clean": True,
+        "source_synchronized": True,
+        "primary_confirmed": True,
+        "projection_identity": "test-projection",
+    }
+
+
+@pytest.mark.asyncio
+async def test_sealed_repair_restores_absent_projection_and_retries_as_noop(test_engine):
+    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    repo_id, topic_id = await _seed_empty_repository(factory, "repair-repo")
+    workplan_old = uuid.uuid4()
+    task_old = uuid.uuid4()
+    plan = _sealed_plan("repair-repo", workplan_old, task_old)
+    unit = _repair_unit(plan, topic_id)
+
+    async with factory() as session:
+        receipt = await repair_absent_prederivation_projection(
+            session, plan, "repair-repo", unit, **_repair_kwargs(plan)
+        )
+    assert receipt.outcome == "repaired"
+    assert receipt.repository_id == str(repo_id)
+    assert receipt.workplan_old_id == str(workplan_old)
+    assert receipt.task_records == (("TEST-WP-0001-T01", str(task_old)),)
+
+    async with factory() as session:
+        assert await session.get(Workplan, workplan_old) is not None
+        assert await session.get(Task, task_old) is not None
+        assert await session.get(Workplan, _derived("TEST-WP-0001")) is None
+        assert await session.get(Task, _derived("TEST-WP-0001-T01")) is None
+
+    async with factory() as session:
+        retry = await repair_absent_prederivation_projection(
+            session, plan, "repair-repo", unit, **_repair_kwargs(plan)
+        )
+    assert retry.outcome == "verified_noop"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("override", "message"),
+    [
+        ({"source_clean": False}, "clean synchronized"),
+        ({"source_synchronized": False}, "clean synchronized"),
+        ({"source_revision": "f" * 40}, "source revision drifted"),
+        ({"source_fingerprint": "e" * 64}, "source fingerprint drifted"),
+        ({"expected_plan_sha256": "0" * 64}, "explicit plan SHA-256"),
+        ({"primary_confirmed": False}, "primary confirmation"),
+    ],
+)
+async def test_sealed_repair_rejects_untrusted_source(test_engine, override, message):
+    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    _repo_id, topic_id = await _seed_empty_repository(factory, "untrusted-repo")
+    workplan_old = uuid.uuid4()
+    task_old = uuid.uuid4()
+    plan = _sealed_plan("untrusted-repo", workplan_old, task_old)
+    kwargs = _repair_kwargs(plan) | override
+
+    async with factory() as session:
+        with pytest.raises(IdentifierMigrationError, match=message):
+            await repair_absent_prederivation_projection(
+                session, plan, "untrusted-repo", _repair_unit(plan, topic_id), **kwargs
+            )
+    async with factory() as session:
+        assert await session.get(Workplan, workplan_old) is None
+        assert await session.get(Task, task_old) is None
+
+
+@pytest.mark.asyncio
+async def test_sealed_repair_rejects_partial_or_derived_projection_presence(test_engine):
+    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    repo_id, topic_id = await _seed_empty_repository(factory, "partial-repo")
+    workplan_old = uuid.uuid4()
+    task_old = uuid.uuid4()
+    plan = _sealed_plan("partial-repo", workplan_old, task_old)
+    unit = _repair_unit(plan, topic_id)
+
+    async with factory() as session:
+        session.add(
+            Workplan(
+                id=workplan_old,
+                repo_id=repo_id,
+                topic_id=topic_id,
+                slug="test-wp-0001",
+                title="Migration test",
+                status="active",
+            )
+        )
+        await session.commit()
+    async with factory() as session:
+        with pytest.raises(IdentifierMigrationError, match="partial old projection"):
+            await repair_absent_prederivation_projection(
+                session, plan, "partial-repo", unit, **_repair_kwargs(plan)
+            )
+
+    async with factory() as session:
+        await session.execute(text("DELETE FROM workplans WHERE id = :id"), {"id": workplan_old})
+        session.add(
+            Workplan(
+                id=_derived("TEST-WP-0001"),
+                repo_id=repo_id,
+                topic_id=topic_id,
+                slug="derived-conflict",
+                title="Derived conflict",
+                status="active",
+            )
+        )
+        await session.commit()
+    async with factory() as session:
+        with pytest.raises(IdentifierMigrationError, match="derived target presence"):
+            await repair_absent_prederivation_projection(
+                session, plan, "partial-repo", unit, **_repair_kwargs(plan)
+            )
+
+
+@pytest.mark.asyncio
+async def test_sealed_repair_http_surface_returns_receipt(client, test_engine):
+    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    repo_id, topic_id = await _seed_empty_repository(factory, "http-repair-repo")
+    workplan_old = uuid.uuid4()
+    task_old = uuid.uuid4()
+    plan = _sealed_plan("http-repair-repo", workplan_old, task_old)
+    response = await client.post(
+        "/identifier-migrations/sealed-projection-repairs",
+        json={
+            "plan": plan,
+            "repo_slug": "http-repair-repo",
+            "unit": _repair_unit(plan, topic_id),
+            **_repair_kwargs(plan),
+        },
+    )
+    assert response.status_code == 200, response.text
+    receipt = response.json()
+    assert receipt["outcome"] == "repaired"
+    assert receipt["repository_id"] == str(repo_id)
+    assert receipt["task_records"] == [["TEST-WP-0001-T01", str(task_old)]]
+    assert (await client.get(f"/workplans/{workplan_old}")).status_code == 200
+    assert (await client.get(f"/tasks/{task_old}")).status_code == 200
+    assert (await client.get(f"/workplans/{_derived('TEST-WP-0001')}")).status_code == 404
+    assert (await client.get(f"/tasks/{_derived('TEST-WP-0001-T01')}")).status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_sealed_repair_http_surface_rejects_missing_confirmation(client, test_engine):
+    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    _repo_id, topic_id = await _seed_empty_repository(factory, "http-reject-repo")
+    plan = _sealed_plan("http-reject-repo", uuid.uuid4(), uuid.uuid4())
+    payload = {
+        "plan": plan,
+        "repo_slug": "http-reject-repo",
+        "unit": _repair_unit(plan, topic_id),
+        **_repair_kwargs(plan),
+    }
+    payload["primary_confirmed"] = False
+    response = await client.post(
+        "/identifier-migrations/sealed-projection-repairs", json=payload
+    )
+    assert response.status_code == 409
+    assert "primary confirmation" in response.json()["detail"]