feat(projection): reconcile a repository's projection against the forge

Implements ADR-012 decision 7 as amended (STATE-WP-0083-T03). Creates what the
forge has and the hub lacks, updates what differs, retires what no longer
derives. It never deletes: hub-native records reference workplans with ON DELETE
RESTRICT, and destroying a progress event to tidy a derived projection would
lose hub-native truth to fix a derived-state problem.

Retirement is refused by default. A record that stops deriving may mean a
deliberately deleted file or a caller pointed at the wrong branch; only the
caller can say which.

Verified against live data and rolled back: whitehat-security applied 5 updates
with no retirements; the-custodian refused, naming the four hub-first records
confirmed by hand to have no backing file.

Tasks of existing workplans are deliberately untouched — hub tasks carry no
canonical identifier, so matching is by title and a renamed heading would
destroy and recreate a record. Tasks are created only alongside a new workplan,
where nothing exists to mis-match. Tracked as T06.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 2583210@bnt-lap001
Assistant-Session: f2bff2d5-e9b2-4338-92ca-10282a927006

api/services/forge_projection.py

@@ -378,3 +378,195 @@ def diff_against_hub(
                      "changed": {"status": {"hub": cur.get("status"), "forge": t.status}}}
                 )
     return d
+
+
+# ---------------------------------------------------------------------------
+# Applying the reset (STATE-WP-0083-T03)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class ResetOutcome:
+    repo_slug: str
+    commit: str
+    status: str                      # applied | refused | noop
+    created: list[str] = field(default_factory=list)
+    updated: list[str] = field(default_factory=list)
+    retired: list[str] = field(default_factory=list)
+    refused: list[dict[str, Any]] = field(default_factory=list)
+    notes: list[str] = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "schema": "state-hub.projection-reset.v1",
+            "repo_slug": self.repo_slug,
+            "commit": self.commit,
+            "status": self.status,
+            "counts": {
+                "created": len(self.created),
+                "updated": len(self.updated),
+                "retired": len(self.retired),
+                "refused": len(self.refused),
+            },
+            "created": self.created,
+            "updated": self.updated,
+            "retired": self.retired,
+            "refused": self.refused,
+            "notes": self.notes,
+        }
+
+
+RETIRE_REASON = "no longer derived from the forge"
+
+
+async def reset_repository_projection(
+    session: Any,
+    repo_slug: str,
+    *,
+    acknowledge_retirements: bool = False,
+    forge_base: str = DEFAULT_FORGE_BASE,
+    derived: DerivedProjection | None = None,
+) -> ResetOutcome:
+    """Reconcile one repository's workplan projection against the forge.
+
+    Creates what the forge has and the hub lacks, updates what differs, and
+    retires what no longer derives. It does not delete: hub-native records
+    reference workplans with `ON DELETE RESTRICT`, and destroying a progress
+    event to tidy a derived projection would lose hub-native truth to fix a
+    derived-state problem (`ADR-012` decision 7 as amended).
+
+    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.
+
+    Scope: workplans, and the tasks of workplans being created. Tasks of
+    *existing* workplans are left alone, because hub task rows carry no
+    canonical identifier and can only be matched by title — renaming a heading
+    would otherwise destroy and recreate its record (`T06`).
+    """
+    from datetime import datetime, timezone
+
+    from sqlalchemy import select
+
+    from api.models.managed_repo import ManagedRepo
+    from api.models.task import Task
+    from api.models.workplan import Workplan
+
+    derived = derived or derive_from_forge(repo_slug, forge_base=forge_base)
+    outcome = ResetOutcome(repo_slug=repo_slug, commit=derived.commit, status="noop")
+
+    repo = (
+        await session.execute(select(ManagedRepo).where(ManagedRepo.slug == repo_slug))
+    ).scalar_one_or_none()
+    if repo is None:
+        outcome.status = "refused"
+        outcome.refused.append({"reason": "repository is not registered", "slug": repo_slug})
+        return outcome
+
+    rows = list(
+        (
+            await session.execute(select(Workplan).where(Workplan.repo_id == repo.id))
+        ).scalars()
+    )
+
+    want = {w.record_id.strip().lower(): w for w in derived.workplans}
+    want_paths = {_path_key(w.relative_path): k for k, w in want.items()}
+
+    matched: dict[str, Any] = {}
+    for row in rows:
+        key = (row.slug or "").strip().lower()
+        if key not in want:
+            bp = row.backing_relative_path
+            if bp and _path_key(bp) in want_paths:
+                key = want_paths[_path_key(bp)]
+            else:
+                cand = [k for k in want if key.startswith(k + "-")]
+                key = cand[0] if len(cand) == 1 else key
+        matched[key] = row
+
+    stale = [
+        r for k, r in matched.items()
+        if k not in want and r.projection_retired_at is None
+    ]
+    if stale and not acknowledge_retirements:
+        outcome.status = "refused"
+        for r in stale:
+            outcome.refused.append(
+                {
+                    "reason": "would be retired; the forge no longer derives it",
+                    "slug": r.slug,
+                    "status": r.status,
+                    "backing_relative_path": r.backing_relative_path,
+                }
+            )
+        outcome.notes.append(
+            "Re-run with acknowledgement to retire these. Nothing was changed."
+        )
+        return outcome
+
+    now = datetime.now(tz=timezone.utc)
+
+    for key, w in want.items():
+        row = matched.get(key)
+        if row is None:
+            row = Workplan(
+                id=uuid.UUID(w.uuid),
+                repo_id=repo.id,
+                topic_id=repo.topic_id,
+                slug=w.record_id.lower(),
+                title=w.title or w.record_id,
+                status=w.status or "proposed",
+                backing_filename=w.relative_path.rsplit("/", 1)[-1],
+                backing_relative_path=w.relative_path,
+                backing_archived=w.archived,
+                derived_from_commit=derived.commit,
+            )
+            session.add(row)
+            await session.flush()
+            for t in w.tasks:
+                # Safe only because nothing exists to mis-match against: this
+                # workplan is new to the hub.
+                session.add(
+                    Task(
+                        id=uuid.UUID(t.uuid),
+                        workplan_id=row.id,
+                        title=t.title or t.record_id,
+                        status=t.status or "todo",
+                        priority=t.priority or "medium",
+                    )
+                )
+            outcome.created.append(w.record_id)
+            continue
+
+        changed = False
+        if w.status and row.status != w.status:
+            row.status = w.status
+            changed = True
+        if row.backing_relative_path != w.relative_path:
+            row.backing_relative_path = w.relative_path
+            row.backing_filename = w.relative_path.rsplit("/", 1)[-1]
+            row.backing_archived = w.archived
+            changed = True
+        if row.projection_retired_at is not None:
+            # It derives again; un-retire rather than leaving a contradiction.
+            row.projection_retired_at = None
+            row.projection_retired_reason = None
+            changed = True
+        if row.derived_from_commit != derived.commit:
+            row.derived_from_commit = derived.commit
+            changed = True
+        if changed:
+            outcome.updated.append(w.record_id)
+
+    for r in stale:
+        r.projection_retired_at = now
+        r.projection_retired_reason = RETIRE_REASON
+        outcome.retired.append(r.slug or str(r.id))
+
+    if outcome.created or outcome.updated or outcome.retired:
+        outcome.status = "applied"
+    outcome.notes.append(
+        "Tasks of existing workplans were not touched; hub tasks carry no "
+        "canonical identifier (STATE-WP-0083-T06)."
+    )
+    return outcome

tests/test_forge_projection.py

@@ -96,3 +96,151 @@ def test_clone_failure_is_reported_not_swallowed(monkeypatch):
     monkeypatch.setattr(fp, "_run_git", boom)
     with pytest.raises(fp.ForgeDeriveError, match="not found"):
         fp.derive_from_forge("nope")
+
+
+class TestReset:
+    """Applying the reset (STATE-WP-0083-T03).
+
+    The properties worth guarding are the refusals, not the happy path. A reset
+    that quietly retires a record someone still needs is worse than one that
+    does nothing.
+    """
+
+    @staticmethod
+    def _derived(*records):
+        wps = []
+        for rid, status, path in records:
+            wps.append(
+                fp.DerivedWorkplan(
+                    record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
+                    status=status, relative_path=path, archived=False, tasks=[],
+                )
+            )
+        return fp.DerivedProjection(repo_slug="demo", commit="c0ffee", workplans=wps)
+
+    @pytest.mark.asyncio
+    async def test_refuses_retirement_unless_acknowledged(self, monkeypatch):
+        """A record that stops deriving may mean a deleted file — or a wrong branch."""
+        session = _FakeSession(
+            repo=_Repo(),
+            rows=[_Row(slug="demo-wp-0001", status="active", path="workplans/a.md")],
+        )
+        out = await fp.reset_repository_projection(
+            session, "demo", derived=self._derived()  # forge has nothing
+        )
+        assert out.status == "refused"
+        assert out.refused and out.refused[0]["slug"] == "demo-wp-0001"
+        assert out.retired == []
+        assert session.committed is False
+
+    @pytest.mark.asyncio
+    async def test_retires_when_acknowledged(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=self._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
+
+    @pytest.mark.asyncio
+    async def test_retirement_is_not_deletion(self):
+        """Hub-native records reference workplans with RESTRICT; the row survives."""
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        session = _FakeSession(repo=_Repo(), rows=[row])
+        await fp.reset_repository_projection(
+            session, "demo", derived=self._derived(), acknowledge_retirements=True
+        )
+        assert row in session.rows
+        assert session.deleted == []
+
+    @pytest.mark.asyncio
+    async def test_unretires_a_record_that_derives_again(self):
+        from datetime import datetime, timezone
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        row.projection_retired_at = datetime.now(tz=timezone.utc)
+        row.projection_retired_reason = fp.RETIRE_REASON
+        session = _FakeSession(repo=_Repo(), rows=[row])
+        out = await fp.reset_repository_projection(
+            session, "demo",
+            derived=self._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
+        )
+        assert out.status == "applied"
+        assert row.projection_retired_at is None
+
+    @pytest.mark.asyncio
+    async def test_unregistered_repository_is_refused_not_created(self):
+        session = _FakeSession(repo=None, rows=[])
+        out = await fp.reset_repository_projection(session, "nope", derived=self._derived())
+        assert out.status == "refused"
+        assert "not registered" in out.refused[0]["reason"]
+
+    @pytest.mark.asyncio
+    async def test_records_the_commit_it_derived_from(self):
+        row = _Row(slug="demo-wp-0001", status="proposed", path="workplans/a.md")
+        session = _FakeSession(repo=_Repo(), rows=[row])
+        await fp.reset_repository_projection(
+            session, "demo",
+            derived=self._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
+        )
+        assert row.derived_from_commit == "c0ffee"
+        assert row.status == "active"
+
+
+class _Repo:
+    def __init__(self):
+        import uuid as _u
+        self.id = _u.uuid4()
+        self.topic_id = None
+
+
+class _Row:
+    def __init__(self, slug, status, path):
+        import uuid as _u
+        self.id = _u.uuid4()
+        self.slug = slug
+        self.status = status
+        self.backing_relative_path = path
+        self.backing_filename = path.rsplit("/", 1)[-1]
+        self.backing_archived = False
+        self.projection_retired_at = None
+        self.projection_retired_reason = None
+        self.derived_from_commit = None
+
+
+class _FakeSession:
+    """Stands in for AsyncSession: enough to prove intent without a database."""
+
+    def __init__(self, repo, rows):
+        self._repo = repo
+        self.rows = list(rows)
+        self.added = []
+        self.deleted = []
+        self.committed = False
+        self._calls = 0
+
+    async def execute(self, *_a, **_k):
+        self._calls += 1
+        repo, rows = self._repo, self.rows
+
+        class R:
+            def scalar_one_or_none(self_inner):
+                return repo
+
+            def scalars(self_inner):
+                return iter(rows)
+
+        return R()
+
+    def add(self, obj):
+        self.added.append(obj)
+
+    def delete(self, obj):
+        self.deleted.append(obj)
+
+    async def flush(self):
+        return None
+
+    async def commit(self):
+        self.committed = True

workplans/STATE-WP-0083-forge-derived-projection-reset.md

@@ -139,7 +139,7 @@ until tasks carry their canonical id, which is `T06`.
 
 ```task
 id: STATE-WP-0083-T03
-status: todo
+status: done
 priority: high
 ```
 
@@ -156,6 +156,30 @@ Acceptance: reset twice produces the same projection; a repository holding
 records the forge lacks is refused with those records named; hub-native record
 counts are unchanged across a reset.
 
+**Done (2026-08-26).** `reset_repository_projection()` in
+`api/services/forge_projection.py`. Verified against live data, rolled back:
+`whitehat-security` applied 5 updates and 0 retirements; `the-custodian`
+**refused**, naming `cust-wp-0023`, `cust-wp-0024`, `state-hub-v0.1` and
+`state-hub-v0.2` — the four records confirmed by hand as genuine hub-first
+records with no file — and changed nothing.
+
+Refusal is the default because a record that stops deriving is ambiguous: the
+file may have been deleted deliberately, or the caller may have pointed at the
+wrong branch. Only the caller can say which, so only the caller may authorise it.
+
+Retirement never deletes, with a test asserting the row survives and
+`session.delete` is never called. A record that derives again is un-retired
+rather than left contradicting the forge.
+
+The first execution of this write path was run against the **cache** database
+rather than central, and rolled back. A write path's first run belongs on the
+discardable copy.
+
+**Scope limit, enforced in the code rather than documented beside it.** Tasks of
+existing workplans are untouched; tasks are created only alongside a *new*
+workplan, where nothing exists to mis-match. The outcome carries that as a note
+so partial convergence cannot be mistaken for full. 678 tests pass.
+
 ## Fleet form as a loop over the repository form
 
 ```task