smoke_record

  1#!/usr/bin/env python3
  2"""smoke_record — one `marola --summarize` run, as the JSON the map shows (MIP-0008 §5.4/§5.5).
  3
  4    scripts/smoke_record.py record --stdout out.txt --exit-code 0 --out-dir site-data/smoke \
  5        --run-id 123 --run-url https://github.com/…/actions/runs/123 \
  6        --model llama3.2:1b --image ghcr.io/<owner>/marola:jvm
  7    scripts/smoke_record.py --self-test        # parses scripts/fixtures/smoke-stdout-*.txt (in `just quality`)
  8
  9Reads the CLI's own output — the `origin ->` line, the ranked list (`Report.line`), the
 10`Draft summary:` and `Reviewer (score N/100, verdict: v):` lines — and writes three files under
 11`--out-dir`: `runs/<run-id>.json` (everything, stdout included), `latest.json` (the same, what
 12app.js renders) and `history.json` (the last 50 runs, newest first, one compact entry each).
 13The deterministic list is the source of truth; the summary is model text and is labelled so in
 14the panel. A run is `ok` only when the pipeline found beaches, the model produced a draft and
 15the reviewer answered — anything else is recorded with its `errors`, so a broken run shows up
 16as broken instead of as an old panel. Standard library only.
 17"""
 18
 19import argparse
 20import datetime as dt
 21import json
 22import re
 23import sys
 24import tempfile
 25from pathlib import Path
 26
 27SCHEMA = 1
 28HISTORY_CAP = 50
 29STDOUT_CAP = 20_000
 30
 31ORIGIN = re.compile(r"^origin -> lat=(-?[\d.]+), lon=(-?[\d.]+) \(radius (\d+)km, source: (.*)\)$")
 32RANKED = re.compile(r"^\s*(\d+)\. \[\s*(\d+)/100\] (.+?)\s+\((\d+\.\d)km\)\s+(.+?)\s+\|")
 33DRAFT = re.compile(r"^Draft summary: (.*)$")
 34REVIEW = re.compile(r"^Reviewer \(score (\d+)/100, verdict: (\w+)\): (.*)$")
 35WATER = re.compile(r"^water quality -> (.*)$")
 36ERRORS = (
 37    "(LLM call failed",
 38    "(reviewer call failed",
 39    "(--summarize needs",
 40    "(nothing to summarize",
 41    "No beaches found nearby",
 42)
 43
 44
 45def parse(stdout: str) -> dict:
 46    """The facts in one run's stdout; `None` where a line never appeared."""
 47    origin = None
 48    water = None
 49    ranked = []
 50    draft = None
 51    review = None
 52    errors = []
 53    for raw in stdout.splitlines():
 54        line = raw.rstrip()
 55        if m := ORIGIN.match(line):
 56            origin = {
 57                "lat": float(m.group(1)),
 58                "lon": float(m.group(2)),
 59                "radius_km": int(m.group(3)),
 60                "source": m.group(4),
 61            }
 62        elif m := WATER.match(line):
 63            water = m.group(1)
 64        elif m := RANKED.match(line):
 65            ranked.append(
 66                {
 67                    "rank": int(m.group(1)),
 68                    "score": int(m.group(2)),
 69                    "name": m.group(3),
 70                    "distance_km": float(m.group(4)),
 71                    "when": m.group(5),
 72                }
 73            )
 74        elif m := DRAFT.match(line):
 75            draft = m.group(1)
 76        elif m := REVIEW.match(line):
 77            review = {"score": int(m.group(1)), "verdict": m.group(2), "summary": m.group(3)}
 78        elif line.startswith(ERRORS):
 79            errors.append(line)
 80    return {
 81        "origin": origin,
 82        "water": water,
 83        "ranked": ranked[:10],
 84        "top_pick": ranked[0] if ranked else None,
 85        "draft": draft,
 86        "review": review,
 87        "errors": errors,
 88    }
 89
 90
 91def record(stdout: str, exit_code: int, meta: dict) -> dict:
 92    facts = parse(stdout)
 93    ok = exit_code == 0 and facts["top_pick"] is not None and facts["review"] is not None
 94    origin = facts.pop("origin") or {}
 95    return {
 96        "schema": SCHEMA,
 97        **meta,
 98        "lat": origin.get("lat"),
 99        "lon": origin.get("lon"),
100        "radius_km": origin.get("radius_km"),
101        "origin_source": origin.get("source"),
102        "exit_code": exit_code,
103        "ok": ok,
104        **facts,
105        "stdout": stdout[-STDOUT_CAP:],
106    }
107
108
109def compact(run: dict) -> dict:
110    top = run.get("top_pick") or {}
111    review = run.get("review") or {}
112    return {
113        "run_id": run["run_id"],
114        "run_url": run["run_url"],
115        "when": run["when"],
116        "lat": run["lat"],
117        "lon": run["lon"],
118        "model": run["model"],
119        "ok": run["ok"],
120        "top_pick": {"name": top.get("name"), "score": top.get("score")} if top else None,
121        "review": {"score": review.get("score"), "verdict": review.get("verdict")}
122        if review
123        else None,
124    }
125
126
127def write(out_dir: Path, run: dict) -> list[Path]:
128    runs = out_dir / "runs"
129    runs.mkdir(parents=True, exist_ok=True)
130    history_path = out_dir / "history.json"
131    history = {"schema": SCHEMA, "runs": []}
132    if history_path.exists():
133        try:
134            history = json.loads(history_path.read_text())
135        except json.JSONDecodeError:
136            pass  # a corrupt history is rebuilt from this run on
137    entries = [e for e in history.get("runs", []) if e.get("run_id") != run["run_id"]]
138    history = {"schema": SCHEMA, "runs": ([compact(run)] + entries)[:HISTORY_CAP]}
139    written = []
140    for path, payload in (
141        (runs / f"{run['run_id']}.json", run),
142        (out_dir / "latest.json", run),
143        (history_path, history),
144    ):
145        path.write_text(json.dumps(payload, ensure_ascii=False, indent=1) + "\n")
146        written.append(path)
147    return written
148
149
150def self_test() -> int:
151    fixtures = sorted(Path(__file__).parent.glob("fixtures/smoke-stdout-*.txt"))
152    assert fixtures, "no scripts/fixtures/smoke-stdout-*.txt"
153    good = fixtures[-1].read_text()
154    meta = {
155        "run_id": "1",
156        "run_url": "https://example.test/runs/1",
157        "when": "2026-09-05T17:40:00Z",
158        "model": "llama3.2:1b",
159        "image": "ghcr.io/marola-dev/marola:jvm",
160    }
161    r = record(good, 0, meta)
162    assert r["ok"], r["errors"]
163    assert r["lat"] == -27.6733 and r["lon"] == -48.47 and r["radius_km"] == 15, (
164        r["lat"],
165        r["lon"],
166    )
167    assert r["origin_source"] == "--lat/--lon flags"
168    assert r["water"] == "IMA/SC"
169    assert r["top_pick"]["name"] == "Praia da Joaquina", r["top_pick"]
170    assert r["top_pick"]["score"] == 55 and r["top_pick"]["distance_km"] == 4.6
171    assert r["top_pick"]["when"] == "Sun 6 Sep, 10:00"
172    assert len(r["ranked"]) == 6 and r["ranked"][5]["name"] == "Praia do Campeche"
173    assert r["draft"].startswith("Still looks good for a swim")
174    assert r["review"] == {
175        "score": 60,
176        "verdict": "revise",
177        "summary": "Still looks good for a swim in calm conditions and lower wind speeds.",
178    }
179    assert r["errors"] == []
180
181    # The pipeline ran but the model call failed: numbers, no summary, not ok.
182    llm_down = (
183        good.split("Draft summary:")[0] + "(LLM call failed, showing numbers above only: x)\n"
184    )
185    r2 = record(llm_down, 0, meta)
186    assert not r2["ok"] and r2["draft"] is None and r2["review"] is None
187    assert r2["top_pick"]["name"] == "Praia da Joaquina" and len(r2["errors"]) == 1
188
189    # Nothing nearby (or a crash): no top pick, not ok, the exit code kept.
190    r3 = record(
191        "marola :: best hour\nNo beaches found nearby, or no forecast data for tomorrow yet.\n",
192        1,
193        meta,
194    )
195    assert not r3["ok"] and r3["top_pick"] is None and r3["exit_code"] == 1 and r3["errors"]
196
197    with tempfile.TemporaryDirectory() as tmp:
198        out = Path(tmp)
199        for i in range(HISTORY_CAP + 2):
200            write(
201                out,
202                record(
203                    good, 0, {**meta, "run_id": str(i), "when": f"2026-09-05T{i % 24:02d}:00:00Z"}
204                ),
205            )
206        history = json.loads((out / "history.json").read_text())
207        assert len(history["runs"]) == HISTORY_CAP, len(history["runs"])
208        assert history["runs"][0]["run_id"] == str(HISTORY_CAP + 1), "newest first"
209        latest = json.loads((out / "latest.json").read_text())
210        assert latest["run_id"] == str(HISTORY_CAP + 1) and latest["ok"]
211        assert (out / "runs" / "0.json").exists()
212        # Re-recording the same run id replaces its history entry instead of duplicating it.
213        write(out, record(good, 0, {**meta, "run_id": str(HISTORY_CAP + 1)}))
214        history = json.loads((out / "history.json").read_text())
215        assert [e["run_id"] for e in history["runs"]].count(str(HISTORY_CAP + 1)) == 1
216    print(f"smoke_record self-test: ok ({fixtures[-1].name})")
217    return 0
218
219
220def main(argv: list[str]) -> int:
221    ap = argparse.ArgumentParser(
222        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
223    )
224    ap.add_argument("--self-test", action="store_true")
225    sub = ap.add_subparsers(dest="cmd")
226    rec = sub.add_parser("record")
227    rec.add_argument("--stdout", required=True, type=Path)
228    rec.add_argument("--exit-code", required=True, type=int)
229    rec.add_argument("--out-dir", required=True, type=Path)
230    rec.add_argument("--run-id", required=True)
231    rec.add_argument("--run-url", required=True)
232    rec.add_argument("--model", required=True)
233    rec.add_argument("--image", required=True)
234    rec.add_argument("--when", default=dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ"))
235    args = ap.parse_args(argv)
236    if args.self_test:
237        return self_test()
238    if args.cmd != "record":
239        ap.print_help()
240        return 2
241    meta = {
242        "run_id": args.run_id,
243        "run_url": args.run_url,
244        "when": args.when,
245        "model": args.model,
246        "image": args.image,
247    }
248    run = record(args.stdout.read_text(errors="replace"), args.exit_code, meta)
249    for path in write(args.out_dir, run):
250        print(path)
251    top = run["top_pick"] or {}
252    review = run["review"] or {}
253    print(
254        f"ok={run['ok']} top={top.get('name')} ({top.get('score')}) "
255        f"review={review.get('score')}/{review.get('verdict')} errors={run['errors']}"
256    )
257    return 0
258
259
260if __name__ == "__main__":
261    sys.exit(main(sys.argv[1:]))