coulomb/repo-manager
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bounded Forge-sourced freshness runner and durable receipts (RMGR-WP-0020-T01/T02)

Browse source 
T01: repo-manager.freshness-receipt.v1 contract and an append-only JSONL
store in the Repo Manager evidence sink (~/.repo-manager/freshness-receipts.jsonl,
RM_FRESHNESS_RECEIPTS_PATH). It is flock-serialised and fsynced, and a receipt
cannot be built without a verified primary hub. The State Hub DB is not used.

T02: `rmgr freshness run --limit N --max-seconds S [--dry-run]`. Repositories
are ranked stalest first. A repository is skipped as noop when its Forgejo
HEAD equals the last successful receipt. Otherwise the runner reconciles from
the immutable Forge commit via POST /repos/{slug}/work-record-projection/reconcile,
never from a checkout or a sweep. The hub receipt is verified (primary, label,
expected == derived == projected), refusals are never forced
(acknowledge_retirements=false), and a hard time cap applies. The run
refuses to write receipts unless the hub is the verified primary.

Workplan status: active (founder approved the design 2026-10-04).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 692292@bnt-lap001
Assistant-Session: 89c7d8e8-e1b3-47eb-9411-a0ab227fca2f
This commit is contained in:
tegwick 2026-10-04 13:15:03 +02:00
parent 271da90fb4
commit ad1ee3e112
5 changed files with 1096 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

83
src/repo_manager/cli.py
View file

@ -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

154
src/repo_manager/freshness_receipts.py Normal file
View file

@ -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

447
src/repo_manager/freshness_runner.py Normal file
View file

@ -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

380
tests/test_freshness_runner.py Normal file
View file

@ -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/<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

35
workplans/RMGR-WP-0020-bounded-freshness-runner.md
View file

@ -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/<slug>`,
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