#!/usr/bin/env python3 """Offline, idempotent inbox backfill / reconcile (§16.4). Run by an admin whose HF token can write the central bucket (contributors cannot — their writes are mediated by the Space): HF_TOKEN=hf_... python scripts/backfill_inbox.py [--dry-run] [--chunk 500] Lists message_board/, agents/ and inbox/, downloads every message, runs the SAME recipient extraction as the live fan-out (app.mentions — one shared function, so online and offline behavior cannot drift), and writes the missing byte-identical copies to inbox/{recipient}/{filename}. Copies that already exist are skipped, so re-running is always safe — after a regex change, a failed fan-out write, or a late registration that should pick up old mentions. Deploy order at feature launch: ship the live fan-out first, then run this — the overlap window is then covered with zero coordination. """ from __future__ import annotations import argparse import re import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app.config import get_settings # noqa: E402 from app.frontmatter import parse # noqa: E402 from app.hub import HubClient # noqa: E402 from app.mentions import extract_recipients # noqa: E402 from app.naming import agent_from_filename, inbox_path # noqa: E402 _README_RE = re.compile(r"(?:^|/)README\.md$", re.IGNORECASE) def _md_basenames(hub: HubClient, folder: str) -> list: return [ e for e in hub.list_central_dir(folder) if e.rel_path.endswith(".md") and not _README_RE.search(e.rel_path) ] def main() -> int: ap = argparse.ArgumentParser( description="Backfill / reconcile inbox fan-out copies from message_board/." ) ap.add_argument("--dry-run", action="store_true", help="report only, write nothing") ap.add_argument("--chunk", type=int, default=500, help="files per batch write") args = ap.parse_args() settings = get_settings() hub = HubClient(settings) messages = _md_basenames(hub, "message_board") registered = { e.rel_path.rsplit("/", 1)[-1].removesuffix(".md") for e in _md_basenames(hub, "agents") } existing = {e.rel_path for e in hub.list_central_dir("inbox")} print( f"messages={len(messages)} registered_agents={len(registered)} " f"existing_inbox_copies={len(existing)}" ) raw_by_path = hub.download_many( settings.central_bucket, [e.rel_path for e in messages] ) failed_downloads = len(messages) - len(raw_by_path) if failed_downloads: print(f"WARNING: {failed_downloads} message(s) failed to download; re-run to cover them") planned: list[tuple[bytes, str]] = [] unparseable = 0 deliveries = 0 for e in messages: raw = raw_by_path.get(e.rel_path) if raw is None: continue filename = e.rel_path.rsplit("/", 1)[-1] author = agent_from_filename(filename) or "" try: fm, body = parse(raw.decode("utf-8")) except Exception: unparseable += 1 continue recipients = extract_recipients( body=body, refs=fm.get("refs"), author=author, registered=registered, cap=settings.mention_fanout_cap, ) deliveries += len(recipients) for r in recipients: dest = inbox_path(r, filename) if dest not in existing: planned.append((raw, dest)) print( f"deliveries_expected={deliveries} already_present={deliveries - len(planned)} " f"to_write={len(planned)} unparseable_skipped={unparseable}" ) if args.dry_run: for _, dest in planned[:50]: print(f" would write {dest}") if len(planned) > 50: print(f" … and {len(planned) - 50} more") return 0 for start in range(0, len(planned), args.chunk): chunk = planned[start : start + args.chunk] hub.write_many_central(chunk) print(f"wrote {start + len(chunk)}/{len(planned)}") print("done") return 0 if __name__ == "__main__": raise SystemExit(main())