tegwick
510ba0b07c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
401 lines
17 KiB
Python
401 lines
17 KiB
Python
"""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/<slug>[/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
|
|
|
|
|
|
def test_targeted_freshness_covers_only_selected_set(tmp_path):
|
|
hub = FakeHub([("a", "active"), ("b", "active"), ("outside", "active")])
|
|
result = run(_cfg(tmp_path, repo_slugs=("a", "b"), limit=2),
|
|
hub_transport=httpx.MockTransport(hub), forge_transport=forge_transport({"a": A, "b": B}))
|
|
assert set(result["selected"]) == {"a", "b"}
|
|
assert {row[0] for row in hub.posts} == {"a", "b"}
|
|
|
|
|
|
def test_targeted_freshness_refuses_incomplete_limit(tmp_path):
|
|
with pytest.raises(ValueError, match="sufficient limit"):
|
|
run(_cfg(tmp_path, repo_slugs=("a", "b"), limit=1))
|
|
|
|
|
|
def test_targeted_freshness_refuses_missing_repo_before_write(tmp_path):
|
|
hub = FakeHub([("a", "active")])
|
|
with pytest.raises(ValueError, match="missing from active"):
|
|
run(_cfg(tmp_path, repo_slugs=("a", "missing")), hub_transport=httpx.MockTransport(hub),
|
|
forge_transport=forge_transport({"a": A}))
|
|
assert hub.posts == []
|