feat(projection): add the fleet reset as a loop over the repository form

ADR-012 decision 7 requires the fleet form to share the per-repository
implementation: the rarely-run wide operation must be the frequently-run narrow
one, or the wide one is trusted on the strength of never having been exercised.

Failure behaviour is the substance. A refusal does not stop the pass — aborting
on the first refusal means one unresolved repository blocks reconstruction
everywhere, which in practice means permanently. An error does not stop it
either. Each repository gets its own session so one failure cannot roll back
another's work, and only repositories that applied are committed.

Refs STATE-WP-0083-T04

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

@@ -618,3 +618,72 @@ async def reset_repository_projection(
         "canonical identifier (STATE-WP-0083-T06)."
     )
     return outcome
+
+
+# ---------------------------------------------------------------------------
+# Fleet form (STATE-WP-0083-T04)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class FleetResetOutcome:
+    results: dict[str, dict[str, Any]] = field(default_factory=dict)
+    errors: dict[str, str] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        by_status: dict[str, int] = {}
+        for r in self.results.values():
+            by_status[r["status"]] = by_status.get(r["status"], 0) + 1
+        return {
+            "schema": "state-hub.fleet-projection-reset.v1",
+            "repositories": len(self.results) + len(self.errors),
+            "by_status": by_status,
+            "errored": len(self.errors),
+            "totals": {
+                k: sum(r["counts"][k] for r in self.results.values())
+                for k in ("created", "updated", "retired", "refused")
+            },
+            "results": self.results,
+            "errors": self.errors,
+        }
+
+
+async def reset_fleet_projection(
+    session_factory: Any,
+    repo_slugs: list[str],
+    *,
+    acknowledge_retirements: bool = False,
+    forge_base: str = DEFAULT_FORGE_BASE,
+) -> FleetResetOutcome:
+    """Reset every repository, one at a time, sharing the per-repository path.
+
+    The fleet form is a loop over the repository form and nothing else
+    (`ADR-012` decision 7). The rarely-run wide operation must be the frequently
+    run narrow one, or the wide one is trusted on the strength of never having
+    been exercised.
+
+    A repository that refuses or errors is recorded and the pass continues.
+    Aborting on the first refusal would mean one unresolved repository blocks
+    reconstruction everywhere — which in practice means permanently.
+
+    Each repository gets its own session, so one failure cannot roll back
+    another's work or leave a poisoned transaction behind.
+    """
+    outcome = FleetResetOutcome()
+    for slug in repo_slugs:
+        try:
+            async with session_factory() as session:
+                result = await reset_repository_projection(
+                    session,
+                    slug,
+                    acknowledge_retirements=acknowledge_retirements,
+                    forge_base=forge_base,
+                )
+                if result.status == "applied":
+                    await session.commit()
+                else:
+                    await session.rollback()
+                outcome.results[slug] = result.to_dict()
+        except Exception as exc:  # noqa: BLE001 - one repo must not end the pass
+            outcome.errors[slug] = f"{type(exc).__name__}: {exc}"[:300]
+    return outcome

tests/test_forge_projection.py

@@ -249,6 +249,9 @@ class _FakeSession:
     async def commit(self):
         self.committed = True
 
+    async def rollback(self):
+        self.rolled_back = True
+
 
 class TestCollisionRefusal:
     """A colliding identifier must refuse, not raise (STATE-WP-0083-T03).
@@ -299,3 +302,85 @@ class TestCollisionRefusal:
         session = _FakeSession(repo=_Repo(), rows=[], foreign=[])
         out = await fp.reset_repository_projection(session, "demo", derived=derived)
         assert out.status == "applied" and out.created == ["DEMO-WP-0001"]
+
+
+class TestFleetReset:
+    """The fleet form must be a loop over the repository form (T04).
+
+    Its value is entirely in what it does with failure: a wide reset that stops
+    at the first refusal is one nobody can run, because there is always one
+    unresolved repository somewhere.
+    """
+
+    class _Factory:
+        def __init__(self, sessions):
+            self._sessions = list(sessions)
+
+        def __call__(self):
+            session = self._sessions.pop(0)
+            class Ctx:
+                async def __aenter__(self_inner):
+                    return session
+                async def __aexit__(self_inner, *a):
+                    return False
+            return Ctx()
+
+    @staticmethod
+    def _derived(rid):
+        return fp.DerivedProjection(
+            repo_slug="x", commit="c0ffee",
+            workplans=[fp.DerivedWorkplan(
+                record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
+                status="active", relative_path="workplans/a.md", archived=False, tasks=[])])
+
+    @pytest.mark.asyncio
+    async def test_a_refusal_does_not_stop_the_pass(self, monkeypatch):
+        calls = []
+
+        async def fake(session, slug, **kw):
+            calls.append(slug)
+            out = fp.ResetOutcome(repo_slug=slug, commit="c0ffee", status="applied")
+            if slug == "bad":
+                out.status = "refused"
+                out.refused.append({"reason": "would be retired", "slug": "x"})
+            else:
+                out.updated.append("A-WP-0001")
+            return out
+
+        monkeypatch.setattr(fp, "reset_repository_projection", fake)
+        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
+        res = await fp.reset_fleet_projection(self._Factory(s), ["good", "bad", "also-good"])
+        assert calls == ["good", "bad", "also-good"]
+        d = res.to_dict()
+        assert d["by_status"] == {"applied": 2, "refused": 1}
+        assert d["totals"]["updated"] == 2
+
+    @pytest.mark.asyncio
+    async def test_an_error_does_not_stop_the_pass(self, monkeypatch):
+        async def fake(session, slug, **kw):
+            if slug == "boom":
+                raise RuntimeError("clone failed")
+            return fp.ResetOutcome(repo_slug=slug, commit="c", status="noop")
+
+        monkeypatch.setattr(fp, "reset_repository_projection", fake)
+        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
+        res = await fp.reset_fleet_projection(self._Factory(s), ["a", "boom", "b"])
+        d = res.to_dict()
+        assert d["errored"] == 1 and "clone failed" in res.errors["boom"]
+        assert set(res.results) == {"a", "b"}
+
+    @pytest.mark.asyncio
+    async def test_only_applied_repositories_are_committed(self, monkeypatch):
+        async def fake(session, slug, **kw):
+            out = fp.ResetOutcome(repo_slug=slug, commit="c", status="applied")
+            if slug == "refuser":
+                out.status = "refused"
+            else:
+                out.updated.append("A-WP-0001")
+            return out
+
+        monkeypatch.setattr(fp, "reset_repository_projection", fake)
+        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(2)]
+        await fp.reset_fleet_projection(self._Factory(s), ["applier", "refuser"])
+        assert s[0].committed is True
+        assert s[1].committed is False

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

@@ -184,7 +184,7 @@ so partial convergence cannot be mistaken for full. 678 tests pass.
 
 ```task
 id: STATE-WP-0083-T04
-status: todo
+status: progress
 priority: medium
 ```
 
@@ -196,6 +196,24 @@ the pass.
 Acceptance: a fleet dry-run reports per-repository outcomes including refusals,
 and completes despite them.
 
+**Built (2026-08-26).** `reset_fleet_projection()` is a loop over the
+per-repository form and nothing else, so the rarely-run wide operation is the
+frequently-run narrow one rather than a second implementation trusted on the
+strength of never having been exercised.
+
+Its behaviour on failure is the part that matters, and all three are tested:
+
+- **A refusal does not stop the pass.** Aborting on the first refusal means one
+  unresolved repository blocks reconstruction everywhere, which in practice means
+  permanently.
+- **An error does not stop the pass.** A repository that cannot be cloned is
+  recorded and the loop continues.
+- **Each repository gets its own session**, so one failure cannot roll back
+  another's work or leave a poisoned transaction behind. Only repositories that
+  applied are committed.
+
+694 tests pass. Needs deployment before it can run against central.
+
 ## Retire what the reset replaces
 
 ```task