"""Deterministic reality-checkpoint runner (ASSEMBLY-WP-0010). 1. Window check against checkpoints/schedule.md (authoritative dates). 2. Fetch the indicators of this kind from checkpoints/sources.json with web-fetcher (provenance: URL, retrieval time, payload hash). 3. Write checkpoints/data/YYYY-MM-DD-.jsonl (raw observations) and checkpoints/YYYY/YYYY-MM-DD-.md (record). 4. Optional assessment via llm-connect (LLM_CONNECT_URL). The model only sees fetched observations; any number in its text that is not among them makes the assessment rejected (no invented values). Usage: python3 checkpoints/run_checkpoint.py --kind light [--date YYYY-MM-DD] [--no-llm] Requires web-fetcher importable (pip install, or WEB_FETCHER_PATH=/path/to/web-fetcher). Exit codes: 0 ok, 3 out of window, 1 failure. """ from __future__ import annotations import argparse import json import os import re import sys import urllib.request from datetime import date, datetime from pathlib import Path from zoneinfo import ZoneInfo HERE = Path(__file__).resolve().parent REPO = HERE.parent if os.environ.get("WEB_FETCHER_PATH"): sys.path.insert(0, os.environ["WEB_FETCHER_PATH"]) WINDOW_RE = re.compile(r"(\d{4}-\d{2}-\d{2})\s*[–-]\s*(\d{4}-\d{2}-\d{2})|from\s+(\d{4}-\d{2}-\d{2})") def berlin_today() -> date: return datetime.now(ZoneInfo("Europe/Berlin")).date() def active_phase(day: date, kind: str, schedule: str) -> str | None: """Return the phase name whose window for `kind` contains `day`, else None.""" for line in schedule.splitlines(): cells = [c.strip().strip("`") for c in line.strip().strip("|").split("|")] if len(cells) < 3 or cells[1] != kind: continue m = WINDOW_RE.search(cells[2]) if not m: continue if m.group(3): if day >= date.fromisoformat(m.group(3)): return cells[0] elif date.fromisoformat(m.group(1)) <= day <= date.fromisoformat(m.group(2)): return cells[0] return None def fetch(kind: str) -> list[dict]: from web_fetcher.registry import run_spec specs = json.loads((HERE / "sources.json").read_text())["specs"] out = [] for spec in specs: if kind == "light" and spec.get("kind") != "light": continue for obs in run_spec(spec): d = obs.to_dict() d.update(indicator=spec.get("indicator"), jurisdiction=spec.get("jurisdiction"), label=spec.get("label"), source=d.get("source") or spec.get("source")) out.append(d) return out def model_version() -> str: readme = (REPO / "models" / "README.md").read_text() return "GADM-0.4" if "GADM-0.4" in readme else "unknown" def _numbers(text: str) -> set[str]: return {n.rstrip(".") for n in re.findall(r"\d+(?:[.,]\d+)?", text)} def assessment(observations: list[dict]) -> tuple[str | None, str | None]: url = os.environ.get("LLM_CONNECT_URL", "").rstrip("/") if not url: return None, "no LLM_CONNECT_URL; assessment skipped" ok = [o for o in observations if o["status"] == "ok"] facts = "\n".join(f"- {o['indicator']} {o['jurisdiction']} {o.get('labels') or ''}: {o['label']} = " f"{o['value']} {o.get('unit') or ''} (period {o['period']})" for o in ok) prompt = ( "You assess AI adoption observations for a research checkpoint. Use ONLY the facts below.\n" "Rules: (a) every number you write must appear literally in the facts; (b) do not compute " "differences, ratios, growth rates or percentage points; (c) use '-' bullets, never numbered " "lists; (d) no forecasts; (e) indicators from different surveys (e.g. US BTOS: all employer firms, " "two-week window; Eurostat: enterprises with 10+ employees, annual) are not directly comparable - " "say so instead of ranking them against each other.\n" "Write two to four short bullets on what the observations say about AI adoption per " "jurisdiction (US, EU, DE) and which contrasts stand out (describe them in words).\n\nFacts:\n" + facts) body = json.dumps({"prompt": prompt, "config": {"temperature": 0.1, "max_tokens": 500}}).encode() try: req = urllib.request.Request(url + "/execute", data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=180) as resp: text = (json.loads(resp.read()).get("content") or "").strip() except Exception as exc: # network / server errors must not fail the checkpoint return None, f"assessment failed: {exc}"[:200] allowed = _numbers(facts) extra = sorted(n for n in _numbers(text) if n not in allowed and n.replace(",", ".") not in allowed) if extra: if os.environ.get("CHECKPOINT_DEBUG"): print("REJECTED ASSESSMENT:\n" + text, file=sys.stderr) return None, f"assessment rejected: numbers not in fetched data: {extra[:8]}" return text, None def registered() -> dict[tuple[str, str], list[str]]: """(indicator, jurisdiction) -> prediction ids, from the headings in predictions.md.""" out: dict[tuple[str, str], list[str]] = {} path = HERE / "predictions.md" if not path.exists(): return out for m in re.finditer(r"^### (P-\d{4}-\d{3}) — (\w+) [^(\n]*\(([^)]+)\)", path.read_text(), re.M): for jur in (j.strip() for j in m.group(3).split(",")): out.setdefault((m.group(2), jur), []).append(m.group(1)) return out GEO_ALIAS = {"EU27_2020": "EU27", "EU": "EU27"} def band_position(preds: dict, indicator: str, jur: str) -> str: ids = preds.get((indicator, jur)) or preds.get((indicator, GEO_ALIAS.get(jur, jur)), []) return f"see {', '.join(ids)} (judged at review)" if ids else "no prediction registered" def render(day: date, kind: str, phase: str, obs: list[dict], assess: str | None, assess_note: str | None, data_rel: str) -> str: rows, missing = [], [] preds = registered() for o in obs: if o["status"] == "ok": geo = (o.get("labels") or {}).get("geo") jur = geo or o["jurisdiction"] pub = (o.get("published") or "")[:10] or "—" rows.append(f"| {o['indicator']} {o['label']} | {jur} | {o['value']} | {o.get('unit') or ''} | " f"{o['period']} | {pub} | [source]({o['url']}) | {band_position(preds, o['indicator'], jur)} |") else: missing.append(f"- {o['indicator']} {o['label']} ({o['jurisdiction']}): {o['status']}" f" — {o.get('note') or ''} Source: {o.get('source') or '—'}") lines = [f"# Checkpoint {day.isoformat()} ({kind})", "", f"- Model version: {model_version()}", f"- Phase: {phase}", "- Executor: checkpoints/run_checkpoint.py (web-fetcher; llm-connect assessment)", f"- Raw observations with provenance: `{data_rel}`", "", "## Observations", "", "| Indicator | Jurisdiction | Value | Unit | Reference period | Published | Source | Band position |", "|---|---|---|---|---|---|---|---|", *rows, "", "## Assessment", ""] lines += [assess] if assess else [f"_No automated assessment: {assess_note}_"] lines += ["", "## Not found / manual", "", *(missing or ["- none"]), "", "## Follow-ups", "", "- Band positions against registered predictions (`predictions.md`) are judged at review" if preds else "- No predictions registered yet; band positions follow once the register has entries.", ""] return "\n".join(lines) def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser() ap.add_argument("--kind", choices=["light", "full"], required=True) ap.add_argument("--date", default=None) ap.add_argument("--no-llm", action="store_true") args = ap.parse_args(argv) day = date.fromisoformat(args.date) if args.date else berlin_today() phase = active_phase(day, args.kind, (HERE / "schedule.md").read_text()) if phase is None: print(json.dumps({"status": "out-of-window", "date": day.isoformat(), "kind": args.kind})) return 3 obs = fetch(args.kind) data_rel = f"checkpoints/data/{day.isoformat()}-{args.kind}.jsonl" (REPO / data_rel).parent.mkdir(parents=True, exist_ok=True) (REPO / data_rel).write_text("\n".join(json.dumps(o, ensure_ascii=False, sort_keys=True) for o in obs) + "\n") assess, note = (None, "disabled (--no-llm)") if args.no_llm else assessment(obs) rec_rel = f"checkpoints/{day.year}/{day.isoformat()}-{args.kind}.md" (REPO / rec_rel).parent.mkdir(parents=True, exist_ok=True) (REPO / rec_rel).write_text(render(day, args.kind, phase, obs, assess, note, data_rel)) summary = {"status": "ok", "date": day.isoformat(), "kind": args.kind, "phase": phase, "path": rec_rel, "data": data_rel, "ok": sum(o["status"] == "ok" for o in obs), "not_found": sum(o["status"] == "not_found" for o in obs), "error": sum(o["status"] == "error" for o in obs), "assessment": bool(assess)} print(json.dumps(summary)) return 0 if __name__ == "__main__": raise SystemExit(main())