"""RMGR-WP-0020-T01/T02: durable receipts and the bounded freshness runner. State Hub and Forge are mocked at the HTTP boundary (httpx.MockTransport). """ from __future__ import annotations import json from pathlib import Path import httpx import pytest from repo_manager import freshness_receipts as fr from repo_manager.cli import main as cli_main from repo_manager.freshness_runner import ( HubNotPrimary, RunConfig, rank_candidates, run, ) A = "a" * 40 B = "b" * 40 C = "c" * 40 PRIMARY = {"status": "ok", "instance_role": "primary", "instance_label": "railiance01"} def _hub_receipt(slug: str, commit: str, status: str = "applied", **over): body = { "schema": "state-hub.repository-projection-reconcile.v1", "instance_role": "primary", "instance_label": "railiance01", "expected_commit": commit, "derived_commit": commit, "outcome": {"repo_slug": slug, "commit": commit, "status": status, "counts": {}, "refused": [], "notes": []}, } body.update(over) return body class FakeHub: def __init__(self, repos, *, health=None, snapshots=None, reconcile=None): self.repos = repos self.health = health or PRIMARY self.snapshots = snapshots or {} self.reconcile = reconcile or {} self.posts: list[tuple[str, dict]] = [] self.paths: list[str] = [] def __call__(self, request: httpx.Request) -> httpx.Response: path = request.url.path self.paths.append(f"{request.method} {path}") if path == "/state/health": return httpx.Response(200, json=self.health) if path == "/repos/": return httpx.Response(200, json=[{"slug": s, "status": st} for s, st in self.repos]) if path.endswith("/work-record-projection/snapshot"): slug = path.split("/")[2] ts = self.snapshots.get(slug) return httpx.Response(200, json={"workplans": [{"backing_synced_at": ts}]}) if path.endswith("/work-record-projection/reconcile"): slug = path.split("/")[2] body = json.loads(request.content) self.posts.append((slug, body)) handler = self.reconcile.get(slug) if callable(handler): return handler(slug, body) return httpx.Response(200, json=_hub_receipt(slug, body["expected_commit"])) return httpx.Response(404, json={"detail": "not found"}) def forge_transport(heads: dict[str, str | None]): def handler(request: httpx.Request) -> httpx.Response: parts = request.url.path.split("/") # /api/v1/repos/coulomb/[/branches/main] slug = parts[5] if heads.get(slug) is None: return httpx.Response(404, json={"message": "not found"}) if len(parts) == 6: return httpx.Response(200, json={"default_branch": "main"}) return httpx.Response(200, json={"commit": {"id": heads[slug]}}) return httpx.MockTransport(handler) def _cfg(tmp_path: Path, **over) -> RunConfig: base = {"limit": 10, "max_seconds": 60, "receipts_path": tmp_path / "r.jsonl", "runner_revision": "rev-test", "fire_id": "fire-1", "definition_id": "repo-manager-bounded-freshness", "definition_version": "0.1"} base.update(over) return RunConfig(**base) def _seed(path: Path, slug: str, outcome: str, commit: str, finished_at: str) -> None: fr.append_receipt(fr.FreshnessReceipt( receipt_id=f"seed-{slug}-{finished_at}", repo_slug=slug, outcome=outcome, reason=None, forge_commit=commit, forge_ref="main", expected_commit=commit, derived_commit=commit, projected_commit=commit, reconciled=True, http_status=200, hub=fr.HubIdentity("http://hub", "primary", "railiance01", "railiance01", True), runner_revision="old", definition_id=None, definition_version=None, fire_id="old", started_at=finished_at, finished_at=finished_at, elapsed_ms=1.0, ), path) class Clock: def __init__(self, step: float = 0.0): self.now = 0.0 self.step = step def __call__(self) -> float: self.now += self.step return self.now # --- T01: receipt contract and store ------------------------------------------------- def test_receipt_contract_roundtrip_and_index(tmp_path): path = tmp_path / "r.jsonl" _seed(path, "x", "applied", A, "2026-10-01T00:00:00Z") _seed(path, "x", "refused", B, "2026-10-02T00:00:00Z") path.open("a").write("not json\n{\"schema\": \"other\"}\n") rows = fr.load_receipts(path) assert len(rows) == 2 assert rows[0]["schema"] == "repo-manager.freshness-receipt.v1" assert rows[0]["hub"]["verified_primary"] is True idx = fr.index_receipts(rows) assert idx.last["x"]["outcome"] == "refused" assert idx.last_success["x"]["forge_commit"] == A def test_receipt_rejects_unverified_primary_and_bad_outcome(): hub = fr.HubIdentity("http://hub", "replica", "x", "railiance01", False) kwargs = {"receipt_id": "1", "repo_slug": "x", "outcome": "applied", "reason": None, "forge_commit": A, "forge_ref": "main", "expected_commit": A, "derived_commit": A, "projected_commit": A, "reconciled": True, "http_status": 200, "hub": hub, "runner_revision": "r", "definition_id": None, "definition_version": None, "fire_id": "f", "started_at": "t", "finished_at": "t", "elapsed_ms": 0} with pytest.raises(ValueError, match="verified primary"): fr.FreshnessReceipt(**kwargs) kwargs["hub"] = fr.HubIdentity("http://hub", "primary", "railiance01", "railiance01", True) kwargs["outcome"] = "succeeded" with pytest.raises(ValueError, match="invalid freshness outcome"): fr.FreshnessReceipt(**kwargs) def test_receipts_path_env_override(tmp_path, monkeypatch): monkeypatch.setenv("RM_FRESHNESS_RECEIPTS_PATH", str(tmp_path / "x.jsonl")) assert fr.receipts_path() == tmp_path / "x.jsonl" # --- T02: ranking -------------------------------------------------------------------- def test_rank_never_received_first_then_oldest_receipt(): rows = [ {"repo_slug": "seen-new", "outcome": "applied", "forge_commit": A, "finished_at": "2026-10-04T00:00:00Z"}, {"repo_slug": "seen-old", "outcome": "refused", "forge_commit": B, "finished_at": "2026-10-01T00:00:00Z"}, ] idx = fr.index_receipts(rows) ranked = rank_candidates( ["seen-new", "seen-old", "never-recent", "never-unsynced", "never-old"], idx, {"never-recent": "2026-10-03T00:00:00Z", "never-unsynced": None, "never-old": "2026-09-01T00:00:00Z"}, ) assert [c.slug for c in ranked] == [ "never-unsynced", "never-old", "never-recent", "seen-old", "seen-new", ] assert ranked[3].last_success_commit is None # refused is not a success assert ranked[4].last_success_commit == A # --- T02: runner --------------------------------------------------------------------- def test_run_reconciles_stalest_first_and_persists_receipts(tmp_path): cfg = _cfg(tmp_path, limit=2) hub = FakeHub([("r1", "active"), ("r2", "active"), ("r3", "active"), ("gone", "archived")], snapshots={"r1": "2026-10-04T00:00:00Z", "r2": None, "r3": "2026-09-01T00:00:00Z"}) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": A, "r2": B, "r3": C})) assert out["selected"] == ["r2", "r3"] assert hub.posts == [("r2", {"expected_commit": B, "acknowledge_retirements": False}), ("r3", {"expected_commit": C, "acknowledge_retirements": False})] assert out["counts"] == {"applied": 2} rows = fr.load_receipts(cfg.receipts_path) assert [(r["repo_slug"], r["outcome"], r["forge_commit"], r["projected_commit"]) for r in rows] == [("r2", "applied", B, B), ("r3", "applied", C, C)] r = rows[0] assert r["runner_revision"] == "rev-test" and r["fire_id"] == "fire-1" assert r["definition_id"] == "repo-manager-bounded-freshness" assert r["hub"]["instance_label"] == "railiance01" and r["hub"]["verified_primary"] assert r["reconciled"] is True and r["elapsed_ms"] >= 0 # Never a checkout sweep. assert not any("sweep" in p or "consistency" in p for p in hub.paths) def test_run_skips_noop_when_forge_head_matches_last_success(tmp_path): cfg = _cfg(tmp_path) _seed(cfg.receipts_path, "r1", "applied", A, "2026-10-01T00:00:00Z") hub = FakeHub([("r1", "active")]) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": A})) assert hub.posts == [] assert out["results"][0]["action"] == "skip_noop" last = fr.load_receipts(cfg.receipts_path)[-1] assert last["outcome"] == "noop" and last["reconciled"] is False assert last["reason"] == "forge_head_matches_last_success" def test_run_reconciles_when_forge_moved_past_last_success(tmp_path): cfg = _cfg(tmp_path) _seed(cfg.receipts_path, "r1", "applied", A, "2026-10-01T00:00:00Z") hub = FakeHub([("r1", "active")]) run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": B})) assert hub.posts[0][1]["expected_commit"] == B def test_hub_refusal_is_recorded_and_never_forced(tmp_path): cfg = _cfg(tmp_path) refused = _hub_receipt("hole", A, status="refused") refused["outcome"]["refused"] = [{"reason": "unknown alias"}] hub = FakeHub([("hole", "active")], reconcile={"hole": lambda s, b: httpx.Response(200, json=refused)}) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"hole": A})) assert len(hub.posts) == 1 and hub.posts[0][1]["acknowledge_retirements"] is False last = fr.load_receipts(cfg.receipts_path)[-1] assert last["outcome"] == "refused" and last["reason"] == "hub_refused" assert last["detail"]["refused"] == [{"reason": "unknown alias"}] assert out["counts"] == {"refused": 1} # A refused hole is not a success: the next run reconciles it again, not noop. run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"hole": A})) assert len(hub.posts) == 2 def test_refuses_whole_run_when_hub_is_not_verified_primary(tmp_path): cfg = _cfg(tmp_path) for health in ({"instance_role": "cache", "instance_label": "railiance01"}, {"instance_role": "primary", "instance_label": "workstation"}): hub = FakeHub([("r1", "active")], health=health) with pytest.raises(HubNotPrimary): run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": A})) assert hub.posts == [] assert not cfg.receipts_path.exists() def test_receipt_from_non_primary_is_withheld_and_aborts(tmp_path): cfg = _cfg(tmp_path) bad = _hub_receipt("r1", A, instance_role="replica") hub = FakeHub([("r1", "active"), ("r2", "active")], reconcile={"r1": lambda s, b: httpx.Response(200, json=bad)}) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": A, "r2": B})) assert out["aborted"].startswith("invalid_receipt") assert out["results"][0]["receipt_withheld"] is True assert out["deferred"] == ["r2"] assert fr.load_receipts(cfg.receipts_path) == [] def test_commit_mismatch_receipt_is_failed_not_success(tmp_path): cfg = _cfg(tmp_path) bad = _hub_receipt("r1", A) bad["derived_commit"] = B hub = FakeHub([("r1", "active")], reconcile={"r1": lambda s, b: httpx.Response(200, json=bad)}) run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": A})) last = fr.load_receipts(cfg.receipts_path)[-1] assert last["outcome"] == "failed" and last["reason"].startswith("invalid_receipt") def test_http_failures_map_to_failed_receipts(tmp_path): cfg = _cfg(tmp_path) hub = FakeHub( [("moved", "active"), ("unreadable", "active")], reconcile={ "moved": lambda s, b: httpx.Response(409, json={"detail": { "message": "forge default branch is not at the expected commit", "expected_commit": A, "derived_commit": B}}), "unreadable": lambda s, b: httpx.Response(424, json={"detail": {"message": "x"}}), }, ) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"moved": A, "unreadable": A, "missing": None})) reasons = {r["repo_slug"]: (r["outcome"], r["reason"]) for r in fr.load_receipts(cfg.receipts_path)} assert reasons == {"moved": ("failed", "forge_moved_during_run"), "unreadable": ("failed", "forge_unreadable_by_hub")} assert out["counts"] == {"failed": 2} def test_forge_head_unresolved_is_failed_without_hub_write(tmp_path): cfg = _cfg(tmp_path) hub = FakeHub([("ghost", "active")]) run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({})) assert hub.posts == [] last = fr.load_receipts(cfg.receipts_path)[-1] assert last["outcome"] == "failed" and last["reason"] == "forge_head_unresolved" def test_hub_not_primary_409_mid_run_withholds_receipt(tmp_path): cfg = _cfg(tmp_path) hub = FakeHub([("r1", "active")], reconcile={"r1": lambda s, b: httpx.Response( 409, json={"detail": {"message": "primary required", "instance_role": "cache"}})}) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"r1": A})) assert out["aborted"] == "hub_not_primary" assert fr.load_receipts(cfg.receipts_path) == [] def test_hard_time_cap_defers_remaining_repositories(tmp_path): cfg = _cfg(tmp_path, max_seconds=10) hub = FakeHub([(f"r{i}", "active") for i in range(5)]) clock = Clock(step=1.0) # every clock read advances one second out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({f"r{i}": A for i in range(5)}), clock=clock) assert out["deferred"], "time cap must stop starting new work" assert len(hub.posts) + len(out["deferred"]) <= 5 assert len(fr.load_receipts(cfg.receipts_path)) == len(hub.posts) def test_dry_run_plans_only(tmp_path): cfg = _cfg(tmp_path, dry_run=True) _seed(cfg.receipts_path, "same", "applied", A, "2026-10-01T00:00:00Z") before = cfg.receipts_path.read_text() hub = FakeHub([("same", "active"), ("moved", "active")]) out = run(cfg, hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"same": A, "moved": B})) assert hub.posts == [] assert cfg.receipts_path.read_text() == before assert {r["repo_slug"]: r["action"] for r in out["results"]} == { "moved": "reconcile", "same": "skip_noop"} def test_invalid_bounds_rejected(tmp_path): with pytest.raises(ValueError): run(_cfg(tmp_path, limit=0)) with pytest.raises(ValueError): run(_cfg(tmp_path, max_seconds=0)) def test_cli_receipts_latest(tmp_path, capsys): path = tmp_path / "r.jsonl" _seed(path, "x", "applied", A, "2026-10-01T00:00:00Z") _seed(path, "x", "refused", B, "2026-10-02T00:00:00Z") assert cli_main(["freshness", "receipts", "--receipts-path", str(path), "--latest"]) == 0 rows = json.loads(capsys.readouterr().out) assert [r["outcome"] for r in rows] == ["refused"] def test_cli_run_requires_bounds(capsys): with pytest.raises(SystemExit): cli_main(["freshness", "run", "--dry-run"]) def test_rank_falls_back_to_repo_row_when_snapshot_unavailable(tmp_path): class SlowSnapshotHub(FakeHub): def __call__(self, request): if request.url.path == "/repos/": return httpx.Response(200, json=[ {"slug": "fresh", "status": "active", "last_state_synced_at": "2026-10-04T00:00:00Z"}, {"slug": "stale", "status": "active", "last_state_synced_at": "2026-08-01T00:00:00Z"}, ]) if request.url.path.endswith("/snapshot"): return httpx.Response(504) return super().__call__(request) hub = SlowSnapshotHub([]) out = run(_cfg(tmp_path, limit=1, dry_run=True), hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"fresh": A, "stale": B})) assert out["selected"] == ["stale"] assert out["rank_fallback_last_state_synced_at"] == 2