diff --git a/src/repo_manager/cli.py b/src/repo_manager/cli.py index 8782c68..afc32b2 100644 --- a/src/repo_manager/cli.py +++ b/src/repo_manager/cli.py @@ -28,6 +28,60 @@ def _json_object(raw: str) -> dict: return value +def _freshness_command(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: + from repo_manager import freshness_receipts as fr + + if args.freshness_command == "receipts": + path = Path(args.receipts_path).expanduser() if args.receipts_path else None + rows = fr.load_receipts(path, repo_slug=args.repo) + if args.latest: + rows = list(fr.index_receipts(rows).last.values()) + print(json.dumps(rows, indent=2)) + return 0 + if args.freshness_command != "run": + parser.print_help() + return 2 + + from repo_manager.freshness_runner import HubNotPrimary, RunConfig, run + from repo_manager.runtime import PublisherSettings + + try: + token = PublisherSettings.from_env().forgejo_token + except ValueError as exc: + print(json.dumps({"ok": False, "status": "config_error", "error": str(exc)})) + return 2 + cfg = RunConfig( + limit=args.limit, + max_seconds=args.max_seconds, + dry_run=args.dry_run, + api_base=args.api_base, + expected_instance_label=args.expected_instance_label, + forge_base=args.forge_base, + forge_owner=args.forge_owner, + forge_token=token, + receipts_path=Path(args.receipts_path).expanduser() if args.receipts_path else None, + definition_id=args.definition_id, + definition_version=args.definition_version, + fire_id=args.fire_id, + ) + try: + result = run(cfg) + except HubNotPrimary as exc: + print(json.dumps({"ok": False, "status": "refused", "reason": "hub_not_primary", + "hub": exc.identity}, indent=2)) + return 3 + except ValueError as exc: + print(json.dumps({"ok": False, "status": "invalid_arguments", "error": str(exc)})) + return 2 + except Exception as exc: # noqa: BLE001 - hub/forge unreachable before any attempt + print(json.dumps({"ok": False, "status": "unavailable", "error": str(exc)[:300]})) + return 4 + failed = result["counts"].get("failed", 0) + result["ok"] = not failed and not result["aborted"] + print(json.dumps(result, indent=2)) + return 0 if result["ok"] else 1 + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="rmgr", @@ -191,6 +245,32 @@ def main(argv: list[str] | None = None) -> int: p_sync.add_argument("--acknowledge-retirements", action="store_true") p_sync.add_argument("--no-queue", action="store_true") + p_fresh = sub.add_parser("freshness", help="Bounded Forge-sourced freshness runner") + fresh_sub = p_fresh.add_subparsers(dest="freshness_command") + p_fresh_run = fresh_sub.add_parser( + "run", help="Reconcile the N stalest repositories from their Forge HEAD commit" + ) + p_fresh_run.add_argument("--limit", type=int, required=True) + p_fresh_run.add_argument("--max-seconds", type=float, required=True) + p_fresh_run.add_argument("--dry-run", action="store_true", help="Emit the plan only") + p_fresh_run.add_argument( + "--api-base", default=os.environ.get("STATE_HUB_API_BASE", "http://127.0.0.1:8000") + ) + p_fresh_run.add_argument("--expected-instance-label", default="railiance01") + p_fresh_run.add_argument( + "--forge-base", + default=os.environ.get("REPO_MANAGER_FORGEJO_BASE_URL", "https://forgejo.coulomb.social"), + ) + p_fresh_run.add_argument("--forge-owner", default="coulomb") + p_fresh_run.add_argument("--receipts-path", default=None) + p_fresh_run.add_argument("--definition-id", default=None) + p_fresh_run.add_argument("--definition-version", default=None) + p_fresh_run.add_argument("--fire-id", default=None) + p_fresh_rcpt = fresh_sub.add_parser("receipts", help="List durable freshness receipts") + p_fresh_rcpt.add_argument("--repo", default=None) + p_fresh_rcpt.add_argument("--receipts-path", default=None) + p_fresh_rcpt.add_argument("--latest", action="store_true", help="Latest receipt per repo") + p_wp = sub.add_parser("workplan", help="Governed file-backed workplan mutations") wp_sub = p_wp.add_subparsers(dest="workplan_command") p_wp_create = wp_sub.add_parser("create", help="Create a workplan file") @@ -779,6 +859,9 @@ def main(argv: list[str] | None = None) -> int: print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 + if args.command == "freshness": + return _freshness_command(args, p_fresh) + if args.command == "workplan": from repo_manager.commands.workplan import mutate_workplan diff --git a/src/repo_manager/freshness_receipts.py b/src/repo_manager/freshness_receipts.py new file mode 100644 index 0000000..ba76285 --- /dev/null +++ b/src/repo_manager/freshness_receipts.py @@ -0,0 +1,154 @@ +"""Durable freshness receipts (RMGR-WP-0020-T01). + +Contract ``repo-manager.freshness-receipt.v1``: one immutable JSON object per +repository attempt of the bounded freshness runner. It records the repository, +the Forge commit the runner read, the expected/derived/projected commits bound by +State Hub's ``state-hub.repository-projection-reconcile.v1`` receipt, the +outcome, the reason, the runner revision, the Activity Core definition and fire, +timings, and the verified primary-hub identity. + +Store: an append-only JSON Lines file in the Repo Manager evidence sink, next to +the mutation meter (``~/.repo-manager/mutation-meter.jsonl``) and the +idempotency store. This is repo-manager's existing persistence pattern. The +State Hub DB is deliberately not used, because the hub is the system under +measurement and is a read model. The file needs no new service, it is readable +with ``jq``/``rmgr freshness receipts``, and appends are serialised with an +advisory lock and fsynced. A deployed runner must mount this path on a +persistent volume (T04). + +Path resolution: ``RM_FRESHNESS_RECEIPTS_PATH`` env, then +``freshness_receipts_path`` in the dual-run config file, then the default. +""" + +from __future__ import annotations + +import fcntl +import json +import os +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +from repo_manager.dual_run import _load_file_config + +SCHEMA = "repo-manager.freshness-receipt.v1" +DEFAULT_PATH = "~/.repo-manager/freshness-receipts.jsonl" + +# ``noop`` covers both "hub reported nothing to change" and "Forge HEAD already +# equals the last successful receipt" (then ``reconciled`` is False). +# ``queued`` and ``skipped`` are reserved by the contract; the bounded runner +# never queues and does not write receipts for repositories it did not attempt. +OUTCOMES = ("applied", "noop", "queued", "failed", "refused", "skipped") +SUCCESS_OUTCOMES = frozenset({"applied", "noop"}) + + +@dataclass +class HubIdentity: + api_base: str + instance_role: str | None + instance_label: str | None + expected_instance_label: str | None + verified_primary: bool + + +@dataclass +class FreshnessReceipt: + receipt_id: str + repo_slug: str + outcome: str + reason: str | None + forge_commit: str | None + forge_ref: str | None + expected_commit: str | None + derived_commit: str | None + projected_commit: str | None + reconciled: bool + http_status: int | None + hub: HubIdentity + runner_revision: str + definition_id: str | None + definition_version: str | None + fire_id: str + started_at: str + finished_at: str + elapsed_ms: float + detail: dict[str, Any] = field(default_factory=dict) + schema: str = SCHEMA + + def __post_init__(self) -> None: + if self.outcome not in OUTCOMES: + raise ValueError(f"invalid freshness outcome: {self.outcome!r}") + if not self.hub.verified_primary: + raise ValueError("a freshness receipt requires a verified primary hub") + + @property + def succeeded(self) -> bool: + return self.outcome in SUCCESS_OUTCOMES + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def receipts_path() -> Path: + raw = os.environ.get("RM_FRESHNESS_RECEIPTS_PATH") + if raw: + return Path(raw).expanduser() + cfg = _load_file_config() + if cfg.get("freshness_receipts_path"): + return Path(str(cfg["freshness_receipts_path"])).expanduser() + return Path(DEFAULT_PATH).expanduser() + + +def append_receipt(receipt: FreshnessReceipt, path: Path | None = None) -> Path: + """Append one receipt durably (locked, flushed and fsynced).""" + path = path or receipts_path() + path.parent.mkdir(parents=True, exist_ok=True) + line = json.dumps(receipt.to_dict(), sort_keys=True, default=str) + "\n" + with path.open("a", encoding="utf-8") as handle: + fcntl.flock(handle, fcntl.LOCK_EX) + try: + handle.write(line) + handle.flush() + os.fsync(handle.fileno()) + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + return path + + +def load_receipts(path: Path | None = None, repo_slug: str | None = None) -> list[dict[str, Any]]: + """Read receipts in append order; malformed or foreign lines are ignored.""" + path = path or receipts_path() + if not path.is_file(): + return [] + rows: list[dict[str, Any]] = [] + with path.open(encoding="utf-8") as handle: + for raw in handle: + try: + row = json.loads(raw) + except json.JSONDecodeError: + continue + if not isinstance(row, dict) or row.get("schema") != SCHEMA: + continue + if repo_slug and row.get("repo_slug") != repo_slug: + continue + rows.append(row) + return rows + + +@dataclass +class ReceiptIndex: + last: dict[str, dict[str, Any]] = field(default_factory=dict) + last_success: dict[str, dict[str, Any]] = field(default_factory=dict) + + +def index_receipts(rows: list[dict[str, Any]]) -> ReceiptIndex: + """Latest receipt and latest successful receipt per repository.""" + index = ReceiptIndex() + for row in rows: + slug = row.get("repo_slug") + if not slug: + continue + index.last[slug] = row + if row.get("outcome") in SUCCESS_OUTCOMES and row.get("forge_commit"): + index.last_success[slug] = row + return index diff --git a/src/repo_manager/freshness_runner.py b/src/repo_manager/freshness_runner.py new file mode 100644 index 0000000..0527bb0 --- /dev/null +++ b/src/repo_manager/freshness_runner.py @@ -0,0 +1,447 @@ +"""Bounded Forge-sourced freshness runner (RMGR-WP-0020-T02). + +``rmgr freshness run --limit N --max-seconds S [--dry-run]`` + +1. Verify that the answering State Hub is the primary (``instance_role`` and + ``instance_label``). Otherwise refuse the whole run and write no receipts. +2. Rank active repositories stalest first. Repositories without a receipt come + first, ordered by ``backing_synced_at`` (missing first). Then come the rest, + ordered by the time of their latest receipt of any outcome, so that a + permanently refused hole cannot starve the rotation. +3. For at most N repositories, read the Forge default-branch HEAD over the + Forgejo HTTP API. Skip with ``noop`` (no hub request) when it equals the + commit of the last successful receipt. +4. Otherwise ask State Hub to derive the projection from that immutable Forge + commit (``POST /repos/{slug}/work-record-projection/reconcile``, + STATE-WP-0086). There is no checkout, no ``--all --remote`` and no + ``/consistency/sweep/remote-all``. ``acknowledge_retirements`` is always + false, so a hub refusal is recorded as ``refused`` and never forced. +5. Verify the hub receipt like ``projection_sync``: schema, primary, instance + label, and expected == derived == projected == Forge commit. +6. Persist one ``repo-manager.freshness-receipt.v1`` per attempted + repository. Stop starting work at ``max_seconds``. Each request timeout is + clipped to the remaining budget. +""" + +from __future__ import annotations + +import subprocess +import time +import uuid +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, wait +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import httpx + +from repo_manager import __version__ +from repo_manager.freshness_receipts import ( + FreshnessReceipt, + HubIdentity, + ReceiptIndex, + append_receipt, + index_receipts, + load_receipts, + receipts_path, +) +from repo_manager.time import utc_now_text + +RUN_SCHEMA = "repo-manager.freshness-run.v1" +RECONCILE_SCHEMA = "state-hub.repository-projection-reconcile.v1" +DEFAULT_FORGE_BASE = "https://forgejo.coulomb.social" +DEFAULT_FORGE_OWNER = "coulomb" +# Never start a reconcile with less budget than this; the request would only +# time out with an unknown server-side outcome. +MIN_RECONCILE_SECONDS = 5.0 +# Share of max_seconds (capped at 60 s) for the pre-receipt backing_synced_at fallback. +SNAPSHOT_BUDGET_FRACTION = 0.25 + + +class HubNotPrimary(RuntimeError): + def __init__(self, identity: dict[str, Any]): + super().__init__("answering State Hub is not the verified primary") + self.identity = identity + + +@dataclass +class RunConfig: + limit: int + max_seconds: float + dry_run: bool = False + api_base: str = "http://127.0.0.1:8000" + expected_instance_label: str = "railiance01" + forge_base: str = DEFAULT_FORGE_BASE + forge_owner: str = DEFAULT_FORGE_OWNER + forge_token: str | None = None + receipts_path: Path | None = None + runner_revision: str | None = None + definition_id: str | None = None + definition_version: str | None = None + fire_id: str | None = None + snapshot_workers: int = 8 + + +@dataclass +class Candidate: + slug: str + has_receipt: bool + rank_ts: str | None + last_success_commit: str | None = None + detail: dict[str, Any] = field(default_factory=dict) + + +def runner_revision() -> str: + """Source revision of this runner: git HEAD of the package checkout or version.""" + root = Path(__file__).resolve().parents[2] + try: + proc = subprocess.run( + ["git", "-C", str(root), "rev-parse", "HEAD"], + capture_output=True, text=True, timeout=5, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + proc = None + if proc is not None and proc.returncode == 0 and proc.stdout.strip(): + return proc.stdout.strip() + return f"repo-manager=={__version__}" + + +def rank_candidates( + active_slugs: list[str], + index: ReceiptIndex, + backing_synced_at: dict[str, str | None], +) -> list[Candidate]: + """Pure stalest-first ranking (unit-tested).""" + never: list[Candidate] = [] + seen: list[Candidate] = [] + for slug in active_slugs: + last = index.last.get(slug) + success = index.last_success.get(slug) + commit = success.get("forge_commit") if success else None + if last is None: + never.append(Candidate(slug, False, backing_synced_at.get(slug), commit)) + else: + seen.append(Candidate(slug, True, last.get("finished_at"), commit)) + # ISO-8601 UTC strings sort chronologically; None (never synced) sorts first. + never.sort(key=lambda c: (c.rank_ts is not None, c.rank_ts or "", c.slug)) + seen.sort(key=lambda c: (c.rank_ts or "", c.slug)) + return never + seen + + +class _Budget: + def __init__(self, max_seconds: float, clock: Callable[[], float]): + self.clock = clock + self.deadline = clock() + max_seconds + + def remaining(self) -> float: + return self.deadline - self.clock() + + +def _identity(health: dict[str, Any], cfg: RunConfig) -> HubIdentity: + role = health.get("instance_role") + label = health.get("instance_label") + return HubIdentity( + api_base=cfg.api_base.rstrip("/"), + instance_role=role, + instance_label=label, + expected_instance_label=cfg.expected_instance_label, + verified_primary=role == "primary" and label == cfg.expected_instance_label, + ) + + +def _forge_head(forge: httpx.Client, owner: str, slug: str, timeout: float) -> dict[str, Any]: + try: + repo = forge.get(f"/api/v1/repos/{owner}/{slug}", timeout=timeout) + repo.raise_for_status() + branch = repo.json().get("default_branch") or "main" + resp = forge.get(f"/api/v1/repos/{owner}/{slug}/branches/{branch}", timeout=timeout) + resp.raise_for_status() + commit = (resp.json().get("commit") or {}).get("id") + except httpx.HTTPStatusError as exc: + return {"commit": None, "ref": None, + "error": f"forge HTTP {exc.response.status_code}"} + except (httpx.HTTPError, ValueError) as exc: + return {"commit": None, "ref": None, "error": f"forge unavailable: {exc}"[:300]} + if not isinstance(commit, str) or len(commit) != 40: + return {"commit": None, "ref": branch, "error": "forge returned no full commit id"} + return {"commit": commit.lower(), "ref": branch, "error": None} + + +def _backing_synced(hub: httpx.Client, slugs: list[str], workers: int, + budget: _Budget) -> dict[str, str | None]: + """Fallback ranking key, bounded to a fraction of the run budget. + + Repositories whose snapshot fails or does not answer in time rank as never + synced (first), which is the conservative choice. + """ + if not slugs: + return {} + phase = max(min(budget.remaining() * SNAPSHOT_BUDGET_FRACTION, 60.0), 0.5) + request_timeout = min(20.0, phase) + + def one(slug: str) -> str | None: + try: + resp = hub.get(f"/repos/{slug}/work-record-projection/snapshot", + timeout=request_timeout) + resp.raise_for_status() + stamps = [w.get("backing_synced_at") for w in resp.json().get("workplans", [])] + except (httpx.HTTPError, ValueError, AttributeError): + return None + stamps = [s for s in stamps if s] + return max(stamps) if stamps else None + + pool = ThreadPoolExecutor(max_workers=max(1, workers)) + futures = {pool.submit(one, slug): slug for slug in slugs} + done, _ = wait(futures, timeout=phase) + # Do not wait for stragglers; in-flight requests end within request_timeout. + pool.shutdown(wait=False, cancel_futures=True) + return {slug: (f.result() if f in done else None) for f, slug in futures.items()} + + +def _verify_reconcile(receipt: Any, identity: HubIdentity, slug: str, + commit: str) -> str | None: + """Return why a hub response is not a valid exact-commit primary receipt.""" + if not isinstance(receipt, dict) or receipt.get("schema") != RECONCILE_SCHEMA: + return "unexpected receipt schema" + if receipt.get("instance_role") != "primary": + return "receipt not issued by a primary hub" + if receipt.get("instance_label") != identity.instance_label: + return "receipt instance label differs from the verified primary" + outcome = receipt.get("outcome") + if not isinstance(outcome, dict): + return "receipt has no outcome" + if not (receipt.get("expected_commit") == commit + and receipt.get("derived_commit") == commit + and outcome.get("commit") == commit): + return "expected/derived/projected commit does not equal the Forge commit" + if outcome.get("repo_slug") != slug: + return "receipt binds a different repository" + if outcome.get("status") not in {"applied", "noop", "refused"}: + return f"unexpected outcome status {outcome.get('status')!r}" + return None + + +def _http_error_detail(exc: httpx.HTTPStatusError) -> Any: + try: + body = exc.response.json() + except ValueError: + return exc.response.text[:300] + return body.get("detail", body) if isinstance(body, dict) else body + + +def _reconcile(hub: httpx.Client, identity: HubIdentity, slug: str, commit: str, + timeout: float) -> dict[str, Any]: + """One immutable-commit reconcile. Returns receipt fields (no store write).""" + path = f"/repos/{slug}/work-record-projection/reconcile" + headers = { + "Idempotency-Key": f"rmgr-freshness:v1:{slug}:{commit}", + "X-StateHub-Source-Agent": "repo-manager", + } + try: + resp = hub.post(path, json={"expected_commit": commit, "acknowledge_retirements": False}, + headers=headers, timeout=timeout) + resp.raise_for_status() + body = resp.json() + except httpx.TimeoutException: + return {"outcome": "failed", "reason": "hub_timeout_outcome_unknown", + "http_status": None} + except httpx.HTTPStatusError as exc: + status = exc.response.status_code + detail = _http_error_detail(exc) + fields: dict[str, Any] = {"outcome": "failed", "http_status": status, + "detail": {"hub_error": detail}} + if status == 409 and isinstance(detail, dict) and "derived_commit" in detail: + fields["reason"] = "forge_moved_during_run" + fields["derived_commit"] = detail.get("derived_commit") + elif status == 409 and isinstance(detail, dict) and "instance_role" in detail: + fields["reason"] = "hub_not_primary" + fields["abort"] = True + fields["no_receipt"] = True + elif status == 424: + fields["reason"] = "forge_unreadable_by_hub" + elif status == 502: + fields["reason"] = "hub_derivation_failed" + else: + fields["reason"] = f"hub_http_{status}" + return fields + except (httpx.HTTPError, ValueError) as exc: + return {"outcome": "failed", "reason": "hub_unavailable", "http_status": None, + "detail": {"error": str(exc)[:300]}, "abort": True} + + invalid = _verify_reconcile(body, identity, slug, commit) + if invalid: + # Kill criterion: never persist an unverified primary receipt as success. + identity_fault = "primary" in invalid or "instance label" in invalid + return {"outcome": "failed", "reason": f"invalid_receipt: {invalid}", + "http_status": resp.status_code, "abort": True, + "no_receipt": identity_fault, + "derived_commit": body.get("derived_commit") if isinstance(body, dict) else None} + outcome = body["outcome"] + status = outcome["status"] + fields = { + "outcome": status, + "reason": None, + "http_status": resp.status_code, + "expected_commit": body["expected_commit"], + "derived_commit": body["derived_commit"], + "projected_commit": outcome["commit"], + "detail": {"counts": outcome.get("counts"), "notes": outcome.get("notes")}, + } + if status == "refused": + refused = outcome.get("refused") or [] + fields["reason"] = "hub_refused" + fields["detail"]["refused"] = refused[:20] + return fields + + +def run( + cfg: RunConfig, + *, + hub_transport: httpx.BaseTransport | None = None, + forge_transport: httpx.BaseTransport | None = None, + clock: Callable[[], float] = time.monotonic, +) -> dict[str, Any]: + if cfg.limit < 1: + raise ValueError("--limit must be at least 1") + if cfg.max_seconds <= 0: + raise ValueError("--max-seconds must be positive") + budget = _Budget(cfg.max_seconds, clock) + started_at = utc_now_text() + fire_id = cfg.fire_id or str(uuid.uuid4()) + revision = cfg.runner_revision or runner_revision() + store = cfg.receipts_path or receipts_path() + forge_headers = {"Authorization": f"token {cfg.forge_token}"} if cfg.forge_token else {} + + summary: dict[str, Any] = { + "schema": RUN_SCHEMA, + "fire_id": fire_id, + "dry_run": cfg.dry_run, + "limit": cfg.limit, + "max_seconds": cfg.max_seconds, + "runner_revision": revision, + "definition_id": cfg.definition_id, + "definition_version": cfg.definition_version, + "receipts_path": str(store), + "started_at": started_at, + } + + with httpx.Client(base_url=cfg.api_base.rstrip("/"), follow_redirects=True, + timeout=httpx.Timeout(30.0, connect=5.0), + transport=hub_transport) as hub, \ + httpx.Client(base_url=cfg.forge_base.rstrip("/"), follow_redirects=True, + timeout=httpx.Timeout(15.0, connect=5.0), headers=forge_headers, + transport=forge_transport) as forge: + health = hub.get("/state/health") + health.raise_for_status() + identity = _identity(health.json(), cfg) + summary["hub"] = vars(identity) + if not identity.verified_primary: + raise HubNotPrimary(vars(identity)) + + repos = hub.get("/repos/", params={"limit": 1000}) + repos.raise_for_status() + rows = {r["slug"]: r for r in repos.json() + if isinstance(r, dict) and r.get("status") == "active" and r.get("slug")} + active = sorted(rows) + index = index_receipts(load_receipts(store)) + never = [s for s in active if s not in index.last] + synced = _backing_synced(hub, never, cfg.snapshot_workers, budget) + # A slow or failing snapshot falls back to the repository row's own + # last_state_synced_at, which costs no extra request. + fallback_used = 0 + for slug in never: + if synced.get(slug) is None and rows[slug].get("last_state_synced_at"): + synced[slug] = rows[slug]["last_state_synced_at"] + fallback_used += 1 + ranked = rank_candidates(active, index, synced) + selected = ranked[: cfg.limit] + summary["active_repositories"] = len(active) + summary["never_received"] = len(never) + summary["rank_fallback_last_state_synced_at"] = fallback_used + + results: list[dict[str, Any]] = [] + deferred: list[str] = [] + aborted: str | None = None + for position, cand in enumerate(selected): + if aborted or budget.remaining() <= 0: + deferred.extend(c.slug for c in selected[position:]) + break + t0 = clock() + attempt_started = utc_now_text() + head = _forge_head(forge, cfg.forge_owner, cand.slug, + timeout=min(15.0, max(budget.remaining(), 0.5))) + row: dict[str, Any] = { + "repo_slug": cand.slug, + "rank_key": {"has_receipt": cand.has_receipt, "ts": cand.rank_ts}, + "forge_commit": head["commit"], + "forge_ref": head["ref"], + "last_success_commit": cand.last_success_commit, + } + fields: dict[str, Any] + if head["commit"] is None: + fields = {"outcome": "failed", "reason": "forge_head_unresolved", + "http_status": None, "detail": {"forge_error": head["error"]}} + row["action"] = "fail" + elif head["commit"] == cand.last_success_commit: + fields = {"outcome": "noop", "reason": "forge_head_matches_last_success", + "http_status": None} + row["action"] = "skip_noop" + elif cfg.dry_run: + row["action"] = "reconcile" + results.append(row) + continue + elif budget.remaining() < MIN_RECONCILE_SECONDS: + deferred.extend(c.slug for c in selected[position:]) + break + else: + row["action"] = "reconcile" + fields = _reconcile(hub, identity, cand.slug, head["commit"], + timeout=min(120.0, budget.remaining())) + row.update({k: v for k, v in fields.items() if k not in {"abort", "no_receipt"}}) + if fields.get("abort"): + aborted = fields["reason"] + # Never persist a receipt whose primary identity the hub did not confirm. + if fields.get("no_receipt"): + row["receipt_withheld"] = True + elif not cfg.dry_run: + receipt = FreshnessReceipt( + receipt_id=str(uuid.uuid4()), + repo_slug=cand.slug, + outcome=fields["outcome"], + reason=fields.get("reason"), + forge_commit=head["commit"], + forge_ref=head["ref"], + expected_commit=fields.get("expected_commit"), + derived_commit=fields.get("derived_commit"), + projected_commit=fields.get("projected_commit"), + reconciled=row["action"] == "reconcile", + http_status=fields.get("http_status"), + hub=identity, + runner_revision=revision, + definition_id=cfg.definition_id, + definition_version=cfg.definition_version, + fire_id=fire_id, + started_at=attempt_started, + finished_at=utc_now_text(), + elapsed_ms=round((clock() - t0) * 1000, 1), + detail=fields.get("detail") or {}, + ) + append_receipt(receipt, store) + row["receipt_id"] = receipt.receipt_id + results.append(row) + + counts: dict[str, int] = {} + for row in results: + key = row.get("outcome") or f"plan:{row.get('action')}" + counts[key] = counts.get(key, 0) + 1 + summary.update({ + "selected": [c.slug for c in selected], + "results": results, + "deferred": deferred, + "aborted": aborted, + "counts": counts, + "finished_at": utc_now_text(), + "elapsed_seconds": round(cfg.max_seconds - budget.remaining(), 2), + }) + return summary diff --git a/tests/test_freshness_runner.py b/tests/test_freshness_runner.py new file mode 100644 index 0000000..a6a04f9 --- /dev/null +++ b/tests/test_freshness_runner.py @@ -0,0 +1,380 @@ +"""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 diff --git a/workplans/RMGR-WP-0020-bounded-freshness-runner.md b/workplans/RMGR-WP-0020-bounded-freshness-runner.md index 98a84e5..e254317 100644 --- a/workplans/RMGR-WP-0020-bounded-freshness-runner.md +++ b/workplans/RMGR-WP-0020-bounded-freshness-runner.md @@ -4,7 +4,7 @@ type: workplan title: "Bounded Forge-sourced freshness runner via Activity Core" domain: infotech repo: repo-manager -status: proposed +status: active owner: claude-repo-manager created: "2026-10-04" updated: "2026-10-04" @@ -14,6 +14,12 @@ depends_on: [] state_hub_workstream_id: "cda2296b-9fe0-5e9e-b630-7e143a3c2796" --- +## Status note + +2026-10-04: the founder approved the T01–T03 design, so the status is now `active`. +Approval covers source implementation only. Deployment, schedules and any +non-dry-run fire stay gated behind T04/T05. + ## Goal Close the freshness half of STATE-WP-0094-T05 with a supported, bounded and @@ -39,7 +45,7 @@ Non-goals and constraints: ```task id: RMGR-WP-0020-T01 -status: todo +status: done priority: high state_hub_task_id: "065ef2ad-82ff-5d02-80e1-717e4d1819af" ``` @@ -53,11 +59,21 @@ read-only-queryable home: the Activity Core run artifact / ops-runs store, or th Repo Manager evidence sink. Do not use the State Hub DB. This closes the "last successful / failed receipt" gap. +Implementation note (2026-10-04): `src/repo_manager/freshness_receipts.py`. +The store is an append-only JSON Lines file in the Repo Manager evidence sink, +`~/.repo-manager/freshness-receipts.jsonl` (override with +`RM_FRESHNESS_RECEIPTS_PATH` or the `freshness_receipts_path` key in the +dual-run config). It is the same persistence pattern as the mutation meter and +the idempotency store, so it needs no new service. Appends are flock-serialised +and fsynced. The store is queryable with `rmgr freshness receipts [--repo] +[--latest]`. A receipt cannot be constructed without a verified primary hub. A +deployed runner must mount this path on a persistent volume (T04). + ## Bounded Forge-sourced selector and runner ```task id: RMGR-WP-0020-T02 -status: todo +status: done priority: high state_hub_task_id: "8333b4a7-9d12-57a1-ba9d-d447a0ba26b5" ``` @@ -76,6 +92,19 @@ Add `rmgr freshness run --limit N --max-seconds S [--dry-run]`: `--dry-run` emits the plan only. Tests must cover the ranking, the noop and the refusals (RMGR-WP-0014/0019 holes stay `refused`, never forced). +Implementation note (2026-10-04): `src/repo_manager/freshness_runner.py`, CLI +`rmgr freshness run --limit N --max-seconds S [--dry-run]`. Optional flags: +`--api-base`, `--expected-instance-label`, `--forge-base`, `--forge-owner`, +`--receipts-path`, `--definition-id`, `--definition-version`, `--fire-id`. +The Forge HEAD is read over the Forgejo HTTP API (`forgejo.coulomb.social/coulomb/`, +the same source the hub derives from). It does not use the workstation SSH alias +in `remote_url`. Repositories that already have receipts rank by their latest +receipt of any outcome, so refused holes cannot starve the rotation. The +`backing_synced_at` fallback is capped at 25 % of the budget (at most 60 s). Snapshots that do not answer in time fall back to the `/repos/` row's `last_state_synced_at`. +Exit codes: 0 ok, 1 failed or aborted attempts, 2 bad arguments, 3 hub not +verified primary (no receipts written), 4 hub unreachable. +Tests: `tests/test_freshness_runner.py`. + ## Prepared ActivityDefinition (draft, disabled) ```task