assembly-economics/checkpoints/run_checkpoint.py

"""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-<kind>.jsonl (raw observations) and
   checkpoints/YYYY/YYYY-MM-DD-<kind>.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. "
        "Do not add any number, date or statistic that is not literally in the facts. "
        "Write 2-4 short bullet points: what the observations say about the pace of AI adoption per "
        "jurisdiction (US, EU, DE), and which differences stand out. No forecasts.\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:
        return None, f"assessment rejected: numbers not in fetched data: {extra[:8]}"
    return text, None


def render(day: date, kind: str, phase: str, obs: list[dict], assess: str | None, assess_note: str | None,
           data_rel: str) -> str:
    rows, missing = [], []
    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']}) | no prediction registered |")
        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", "", "- Predictions are not yet registered (ASSEMBLY-WP-0009-T06); band positions",
              "  are reported 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())