diff --git a/api/services/forge_projection.py b/api/services/forge_projection.py index 757b284..4ae3313 100644 --- a/api/services/forge_projection.py +++ b/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 diff --git a/tests/test_forge_projection.py b/tests/test_forge_projection.py index 9b0eb1a..103b4b4 100644 --- a/tests/test_forge_projection.py +++ b/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 diff --git a/workplans/STATE-WP-0083-forge-derived-projection-reset.md b/workplans/STATE-WP-0083-forge-derived-projection-reset.md index 2117213..ee90ea8 100644 --- a/workplans/STATE-WP-0083-forge-derived-projection-reset.md +++ b/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