Guard Forge retirement with an exact expected UUID set

Blanket acknowledgement is unchanged when the UUID list is omitted. A present
expected_retirement_uuids set must equal the stale projections, and the call
retires nothing when that set mismatches, a stale row still has tasks or
dependency edges, or both authorizations are set together.

Assistant: grok
Assistant-Session: 01a10946-16b1-7973-b1dd-71fc58b8fcf0

api/routers/work_record_projection.py

@@ -131,6 +131,7 @@ async def reconcile_repository_work_records(
         session,
         slug,
         acknowledge_retirements=body.acknowledge_retirements,
+        expected_retirement_uuids=body.expected_retirement_uuids,
         derived=derived,
     )
     if outcome.status in {"applied", "noop"} or outcome.released:

api/schemas/work_record_projection.py

@@ -14,6 +14,10 @@ from api.schemas.workplan_dependency import WorkplanDependencyRead
 class RepositoryProjectionReconcile(BaseModel):
     expected_commit: str
     acknowledge_retirements: bool = False
+    # None keeps today's blanket acknowledgement. An empty list is present and
+    # means the caller expects zero retirements. Combined with
+    # acknowledge_retirements, the call is ambiguous and retires nothing.
+    expected_retirement_uuids: list[uuid.UUID] | None = None
 
     @field_validator("expected_commit")
     @classmethod

api/services/forge_projection.py

@@ -1071,6 +1071,81 @@ def _tombstone_slug(slug: str, when: datetime) -> str:
 RETIRE_REASON = "no longer derived from the forge"
 
 
+def _uuid_key(value: Any) -> str:
+    return str(uuid.UUID(str(value)))
+
+
+async def _exact_retirement_blockers(session: Any, stale: list[Any]) -> list[dict[str, Any]]:
+    """Block an exact retirement when any stale row still has tasks or edges.
+
+    Zero means zero. Terminal tasks count, and so does a dependency edge in
+    either direction. The caller retires nothing unless this returns empty.
+    """
+    if not stale:
+        return []
+
+    from sqlalchemy import or_, select
+
+    from api.models.task import Task
+    from api.models.workplan_dependency import WorkplanDependency
+
+    stale_ids = [row.id for row in stale]
+    tasks = list(
+        (
+            await session.execute(select(Task).where(Task.workplan_id.in_(stale_ids)))
+        ).scalars()
+    )
+    edges = list(
+        (
+            await session.execute(
+                select(WorkplanDependency).where(
+                    or_(
+                        WorkplanDependency.from_workplan_id.in_(stale_ids),
+                        WorkplanDependency.to_workplan_id.in_(stale_ids),
+                    )
+                )
+            )
+        ).scalars()
+    )
+    refused: list[dict[str, Any]] = []
+    for row in stale:
+        key = str(row.id)
+        row_tasks = [task for task in tasks if str(task.workplan_id) == key]
+        row_edges = [
+            edge
+            for edge in edges
+            if str(edge.from_workplan_id) == key
+            or (edge.to_workplan_id is not None and str(edge.to_workplan_id) == key)
+        ]
+        if row_tasks:
+            refused.append(
+                {
+                    "reason": "stale projection has tasks",
+                    "uuid": key,
+                    "slug": row.slug,
+                    "task_ids": [str(task.id) for task in row_tasks],
+                }
+            )
+        if row_edges:
+            refused.append(
+                {
+                    "reason": "stale projection has dependency edges",
+                    "uuid": key,
+                    "slug": row.slug,
+                    "dependency_ids": [str(edge.id) for edge in row_edges],
+                }
+            )
+    if (tasks or edges) and not refused:
+        refused.append(
+            {
+                "reason": "stale projection has tasks or dependency edges",
+                "task_ids": [str(task.id) for task in tasks],
+                "dependency_ids": [str(edge.id) for edge in edges],
+            }
+        )
+    return refused
+
+
 def _coerce_task_status(raw: str | None) -> Any:
     from api.models.task import TaskStatus
 
@@ -1202,6 +1277,7 @@ async def reset_repository_projection(
     repo_slug: str,
     *,
     acknowledge_retirements: bool = False,
+    expected_retirement_uuids: list[uuid.UUID] | None = None,
     forge_base: str = DEFAULT_FORGE_BASE,
     derived: DerivedProjection | None = None,
 ) -> ResetOutcome:
@@ -1215,7 +1291,10 @@ async def reset_repository_projection(
 
     Retirement is refused by default. A record that stops deriving may mean the
     file was removed deliberately — or that someone pointed this at the wrong
-    branch. The caller must say which.
+    branch. The caller must say which, either by blanket acknowledgement or by
+    the exact hub UUIDs expected to retire. The two authorizations cannot be
+    combined. Omitting the UUID list keeps blanket acknowledgement unchanged;
+    an empty list expects zero retirements.
 
     Scope: workplans, and their tasks. Tasks of existing workplans are matched
     by ``record_id`` (STATE-WP-0083-T06); rows with no ``record_id`` are left
@@ -1480,6 +1559,24 @@ async def reset_repository_projection(
         r for k, r in matched.items()
         if k not in want and r.projection_retired_at is None
     ] + [r for r in displaced if r.projection_retired_at is None]
+    # Exact UUIDs and blanket acknowledgement are different authorizations.
+    # A call that sets both retires nothing, before any retirement mutation.
+    if expected_retirement_uuids is not None and acknowledge_retirements:
+        outcome.status = "refused"
+        outcome.refused.append(
+            {
+                "reason": "ambiguous authorization",
+                "detail": (
+                    "expected_retirement_uuids and acknowledge_retirements "
+                    "cannot both be set"
+                ),
+            }
+        )
+        outcome.notes.append(
+            "Nothing was retired. Pass either an exact retirement set or "
+            "blanket acknowledgement, not both."
+        )
+        return outcome
     if stale and not derived.retirement_eligible:
         # Acknowledgement cannot authorise this. The caller is confirming that
         # records which stopped deriving should be retired; here nothing has
@@ -1501,7 +1598,37 @@ async def reset_repository_projection(
             "regardless of acknowledgement (STATE-WP-0084-T01)."
         )
         return outcome
-    if stale and not acknowledge_retirements:
+    if expected_retirement_uuids is not None:
+        # `stale` is the unmatched live rows plus the displaced live rows.
+        # A row that still matches, including a historical parent, is not in it.
+        actual = {_uuid_key(row.id) for row in stale}
+        expected = {_uuid_key(value) for value in expected_retirement_uuids}
+        missing = sorted(actual - expected)
+        extra = sorted(expected - actual)
+        if missing or extra:
+            outcome.status = "refused"
+            outcome.refused.append(
+                {
+                    "reason": "expected retirement set does not match stale projections",
+                    "missing": missing,
+                    "extra": extra,
+                }
+            )
+            outcome.notes.append(
+                "Nothing was retired. The expected set must equal the stale "
+                "set; partial retirement is not applied."
+            )
+            return outcome
+        blockers = await _exact_retirement_blockers(session, stale)
+        if blockers:
+            outcome.status = "refused"
+            outcome.refused.extend(blockers)
+            outcome.notes.append(
+                "Nothing was retired. A stale projection with tasks or "
+                "dependency edges cannot be retired."
+            )
+            return outcome
+    elif stale and not acknowledge_retirements:
         outcome.status = "refused"
         for r in stale:
             outcome.refused.append(

tests/test_forge_projection.py

@@ -301,6 +301,8 @@ class _FakeSession:
         slug_clash=None,
         task_rows=None,
         task_collision_rows=None,
+        retirement_tasks=None,
+        retirement_edges=None,
     ):
         self._repo = repo
         self.rows = list(rows)
@@ -308,14 +310,42 @@ class _FakeSession:
         self._slug_clash = list(slug_clash or [])
         self._task_rows = task_rows
         self._task_collision_rows = list(task_collision_rows or [])
+        # Exact-set guard only. None leaves the call sequence used by older tests.
+        self._retirement_tasks = retirement_tasks
+        self._retirement_edges = retirement_edges
+        self._served_retirement_tasks = False
+        self._served_retirement_edges = False
         self.added = []
         self.deleted = []
         self.committed = False
         self._calls = 0
 
-    async def execute(self, *_a, **_k):
+    async def execute(self, statement=None, *_a, **_k):
+        try:
+            text = "" if statement is None else str(statement).lower()
+        except Exception:
+            text = ""
+        # These queries sit ahead of the older call sequence. Answer them
+        # without consuming that sequence, and only the first of each: later
+        # task and edge loads still follow the call counter.
+        if (
+            self._retirement_tasks is not None
+            and not self._served_retirement_tasks
+            and "from tasks" in text
+            and "workplan_id" in text
+        ):
+            self._served_retirement_tasks = True
+            return self._result(self._retirement_tasks)
+        if (
+            self._retirement_edges is not None
+            and not self._served_retirement_edges
+            and "from workplan_dependencies" in text
+        ):
+            self._served_retirement_edges = True
+            return self._result(self._retirement_edges)
+
         self._calls += 1
-        repo, rows = self._repo, self.rows
+        rows = self.rows
         # 1 resolves the repo, 2 loads its workplans, 3 is the identifier
         # lookup, 4 the slug lookup. On the update-only path (no creates)
         # the 3rd call is the existing-task load instead.
@@ -339,6 +369,11 @@ class _FakeSession:
         else:
             payload = []
 
+        return self._result(payload)
+
+    def _result(self, payload):
+        repo = self._repo
+
         class R:
             def scalar_one_or_none(self_inner):
                 return repo
@@ -1353,3 +1388,273 @@ async def test_reset_refuses_duplicate_task_identity_in_forge_projection():
     assert out.status == "refused"
     assert any(item.get("source") == "forge" for item in out.refused)
     assert session.added == []
+
+
+# The preserved Activity parent. A still-deriving row with this id is not stale.
+_HISTORICAL_PARENT = "b4eb45a9-69e3-4ab0-b00c-67a53c3117c5"
+
+
+def _historical_parent_derived():
+    return fp.DerivedProjection(
+        repo_slug="demo",
+        commit="c0ffee",
+        workplans=[
+            fp.DerivedWorkplan(
+                record_id="custodian-WP-0003",
+                uuid=_HISTORICAL_PARENT,
+                title="Preserved parent",
+                status="archived",
+                relative_path="workplans/archived/custodian-wp-0003.md",
+                archived=True,
+                historical_preservation=True,
+                tasks=[],
+            )
+        ],
+    )
+
+
+def _bound_rows():
+    """One still-deriving parent, one path-displaced copy, one unmatched live row."""
+    import uuid as _u
+
+    path = "workplans/archived/custodian-wp-0003.md"
+    parent = _Row(slug="custodian-wp-0003", status="archived", path=path, title="Preserved parent")
+    parent.id = _u.UUID(_HISTORICAL_PARENT)
+    displaced = _Row(slug="empty-duplicate-copy", status="done", path=path, title="Displaced copy")
+    unmatched = _Row(
+        slug="custodian-wp-0099", status="active", path="workplans/other.md", title="Unmatched"
+    )
+    return parent, displaced, unmatched
+
+
+class _Dependency:
+    def __init__(self, from_workplan_id, to_workplan_id):
+        import uuid as _u
+
+        self.id = _u.uuid4()
+        self.from_workplan_id = from_workplan_id
+        self.to_workplan_id = to_workplan_id
+        self.from_task_id = None
+        self.to_task_id = None
+
+
+class TestExactRetirementSet:
+    """Exact UUID authorization for Forge retirement. Nothing is deleted."""
+
+    @staticmethod
+    def _apply(session, derived, **kwargs):
+        return fp.reset_repository_projection(session, "demo", derived=derived, **kwargs)
+
+    @pytest.mark.asyncio
+    async def test_exact_uuid_set_retires_only_those_stale_and_displaced_rows(self):
+        parent, displaced, unmatched = _bound_rows()
+        session = _FakeSession(repo=_Repo(), rows=[parent, displaced, unmatched])
+        out = await self._apply(
+            session,
+            _historical_parent_derived(),
+            expected_retirement_uuids=[displaced.id, unmatched.id],
+        )
+        assert out.status == "applied"
+        assert out.refused == []
+        assert set(out.retired) == {"empty-duplicate-copy", "custodian-wp-0099"}
+        assert displaced.projection_retired_at is not None
+        assert unmatched.projection_retired_at is not None
+        assert displaced.projection_retired_reason == fp.RETIRE_REASON
+        assert unmatched.projection_retired_reason == fp.RETIRE_REASON
+        assert fp.RETIRED_SLUG_MARK in displaced.slug
+        assert fp.RETIRED_SLUG_MARK in unmatched.slug
+        assert parent.projection_retired_at is None
+        assert parent.slug == "custodian-wp-0003"
+        assert parent.derived_from_commit is None
+        assert parent.id.hex == __import__("uuid").UUID(_HISTORICAL_PARENT).hex
+        assert session.deleted == []
+        assert {parent, displaced, unmatched} <= set(session.rows)
+
+    @pytest.mark.asyncio
+    async def test_missing_or_extra_uuid_refuses_and_retires_nothing(self):
+        parent, displaced, unmatched = _bound_rows()
+        derived = _historical_parent_derived()
+
+        missing = _FakeSession(repo=_Repo(), rows=[parent, displaced, unmatched])
+        out = await self._apply(
+            missing, derived, expected_retirement_uuids=[displaced.id]
+        )
+        assert out.status == "refused" and out.retired == []
+        assert out.refused[0]["missing"] == [str(unmatched.id)]
+        assert out.refused[0]["extra"] == []
+        assert displaced.projection_retired_at is None
+        assert unmatched.projection_retired_at is None
+        assert parent.projection_retired_at is None
+
+        extra = _FakeSession(repo=_Repo(), rows=[parent, displaced, unmatched])
+        out = await self._apply(
+            extra,
+            derived,
+            expected_retirement_uuids=[displaced.id, unmatched.id, parent.id],
+        )
+        assert out.status == "refused" and out.retired == []
+        assert out.refused[0]["missing"] == []
+        assert out.refused[0]["extra"] == [str(parent.id)]
+        assert displaced.projection_retired_at is None
+        assert unmatched.projection_retired_at is None
+        assert parent.slug == "custodian-wp-0003"
+
+        # An empty list is present and expects zero retirements.
+        empty = _FakeSession(repo=_Repo(), rows=[parent, displaced, unmatched])
+        out = await self._apply(empty, derived, expected_retirement_uuids=[])
+        assert out.status == "refused" and out.retired == []
+        assert out.refused[0]["extra"] == []
+        assert set(out.refused[0]["missing"]) == {str(displaced.id), str(unmatched.id)}
+        assert displaced.projection_retired_at is None
+        assert extra.deleted == [] and missing.deleted == [] and empty.deleted == []
+
+    @pytest.mark.asyncio
+    async def test_stale_row_with_a_task_refuses(self):
+        """Terminal tasks count. A clean sibling in the same set is not retired."""
+        clean = _Row(slug="demo-wp-0002", status="done", path="workplans/b.md")
+        dirty = _Row(slug="demo-wp-0001", status="done", path="workplans/a.md")
+        task = _TaskRow("DEMO-WP-0001-T01", status="done", workplan_id=dirty.id)
+        session = _FakeSession(
+            repo=_Repo(),
+            rows=[clean, dirty],
+            retirement_tasks=[task],
+            retirement_edges=[],
+        )
+        out = await self._apply(
+            session,
+            TestReset()._derived(),
+            expected_retirement_uuids=[clean.id, dirty.id],
+        )
+        assert out.status == "refused"
+        assert out.retired == []
+        assert clean.projection_retired_at is None
+        assert dirty.projection_retired_at is None
+        assert session.deleted == []
+        assert out.refused[0]["reason"] == "stale projection has tasks"
+        assert out.refused[0]["uuid"] == str(dirty.id)
+        assert task.status == "done"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("direction", ["from", "to"])
+    async def test_stale_row_with_a_dependency_edge_refuses(self, direction):
+        clean = _Row(slug="demo-wp-0002", status="done", path="workplans/b.md")
+        dirty = _Row(slug="demo-wp-0001", status="done", path="workplans/a.md")
+        other = __import__("uuid").uuid4()
+        if direction == "from":
+            edge = _Dependency(dirty.id, other)
+        else:
+            edge = _Dependency(other, dirty.id)
+        session = _FakeSession(
+            repo=_Repo(),
+            rows=[clean, dirty],
+            retirement_tasks=[],
+            retirement_edges=[edge],
+        )
+        out = await self._apply(
+            session,
+            TestReset()._derived(),
+            expected_retirement_uuids=[clean.id, dirty.id],
+        )
+        assert out.status == "refused"
+        assert out.retired == []
+        assert clean.projection_retired_at is None
+        assert dirty.projection_retired_at is None
+        assert session.deleted == []
+        assert out.refused[0]["reason"] == "stale projection has dependency edges"
+        assert out.refused[0]["uuid"] == str(dirty.id)
+        assert str(edge.id) in out.refused[0]["dependency_ids"]
+
+    @pytest.mark.asyncio
+    async def test_boolean_acknowledgement_still_retires_a_simple_stale_row(self):
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        session = _FakeSession(repo=_Repo(), rows=[row])
+        out = await fp.reset_repository_projection(
+            session, "demo", derived=TestReset()._derived(), acknowledge_retirements=True
+        )
+        assert out.status == "applied" and out.retired == ["demo-wp-0001"]
+        assert row.projection_retired_at is not None
+        assert row.projection_retired_reason == fp.RETIRE_REASON
+        assert session.deleted == []
+        assert row in session.rows
+
+    @pytest.mark.asyncio
+    async def test_both_flags_together_refuse(self):
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        session = _FakeSession(repo=_Repo(), rows=[row])
+        out = await fp.reset_repository_projection(
+            session,
+            "demo",
+            derived=TestReset()._derived(),
+            acknowledge_retirements=True,
+            expected_retirement_uuids=[row.id],
+        )
+        assert out.status == "refused"
+        assert out.retired == []
+        assert out.refused[0]["reason"] == "ambiguous authorization"
+        assert row.projection_retired_at is None
+        assert fp.RETIRED_SLUG_MARK not in (row.slug or "")
+        assert session.deleted == []
+
+    @pytest.mark.asyncio
+    async def test_still_deriving_row_is_kept_unless_the_displaced_row_is_named(self):
+        parent, displaced, _unmatched = _bound_rows()
+        derived = _historical_parent_derived()
+
+        unnamed = _FakeSession(repo=_Repo(), rows=[parent, displaced])
+        out = await self._apply(unnamed, derived, expected_retirement_uuids=[parent.id])
+        assert out.status == "refused" and out.retired == []
+        assert str(displaced.id) in out.refused[0]["missing"]
+        assert str(parent.id) in out.refused[0]["extra"]
+        assert parent.projection_retired_at is None
+        assert displaced.projection_retired_at is None
+
+        named = _FakeSession(repo=_Repo(), rows=[parent, displaced])
+        out = await self._apply(
+            named, derived, expected_retirement_uuids=[displaced.id]
+        )
+        assert out.status == "applied"
+        assert out.retired == ["empty-duplicate-copy"]
+        assert displaced.projection_retired_at is not None
+        assert displaced.projection_retired_reason == fp.RETIRE_REASON
+        assert fp.RETIRED_SLUG_MARK in displaced.slug
+        assert parent.projection_retired_at is None
+        assert parent.slug == "custodian-wp-0003"
+        assert parent.derived_from_commit is None
+        assert str(parent.id) == _HISTORICAL_PARENT
+        assert named.deleted == []
+        assert parent in named.rows and displaced in named.rows
+
+    @pytest.mark.asyncio
+    async def test_exact_set_cannot_authorize_an_empty_source(self):
+        derived = fp.DerivedProjection(
+            repo_slug="demo", commit="c0ffee", records_source_present=False
+        )
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        session = _FakeSession(repo=_Repo(), rows=[row])
+        out = await fp.reset_repository_projection(
+            session, "demo", derived=derived, expected_retirement_uuids=[row.id]
+        )
+        assert out.status == "refused"
+        assert out.retired == []
+        assert row.projection_retired_at is None
+        assert out.refused[0]["reason"].startswith("source produced no records")
+
+    def test_schema_distinguishes_omitted_and_empty_expected_set(self):
+        from api.schemas.work_record_projection import RepositoryProjectionReconcile
+        import inspect
+        from api.routers import work_record_projection as route
+
+        omitted = RepositoryProjectionReconcile(expected_commit="a" * 40)
+        assert omitted.expected_retirement_uuids is None
+        assert omitted.acknowledge_retirements is False
+        empty = RepositoryProjectionReconcile(
+            expected_commit="a" * 40, expected_retirement_uuids=[]
+        )
+        assert empty.expected_retirement_uuids == []
+        named = RepositoryProjectionReconcile(
+            expected_commit="b" * 40,
+            expected_retirement_uuids=[_HISTORICAL_PARENT],
+        )
+        assert str(named.expected_retirement_uuids[0]) == _HISTORICAL_PARENT
+        src = inspect.getsource(route.reconcile_repository_work_records)
+        assert "expected_retirement_uuids=body.expected_retirement_uuids" in src

workplans/STATE-WP-0095-historical-source-receiving.md

@@ -7,7 +7,7 @@ repo: state-hub
 status: active
 owner: codex
 created: "2026-10-04"
-updated: "2026-10-04"
+updated: "2026-10-05"
 flavor: implementation
 depends_on: []
 state_hub_workstream_id: "79579572-8120-5d3b-aca5-dd4aff60d058"
@@ -89,6 +89,12 @@ Verify Activity four historical parent UUIDs/64 task UUIDs and Custodian retired
 parent/task unchanged; no historical row creation, movement, deletion,
 relabeling or unretirement. Unrelated stale projections remain separate decisions.
 
+2026-10-05: Source exact-set retirement support now exists. Repository projection
+reconcile accepts `expected_retirement_uuids`; that set must equal the stale
+projections, and any task or dependency edge on a stale row refuses the call.
+Live apply still needs a reviewed rollout. Decision
+91b7c633-a66f-5d43-9f4b-6565aefb9356 remains the disposition gate.
+
 ## Offline evidence, 2026-10-04
 
 Prepared receiver suite: Forge projection regression plus historical preservation,