feat(freeze): frozen day_zero snapshot of the lens baseline

AGENTS.md

@@ -110,6 +110,7 @@ The design has **two layers** (`profiling_inference_organs.md`, PyAutoBrain#444)
 | The read contract as this organ implements it | `pulse/summary.py` + `REFERENCE.md` | the dev flow (reader first, producer second on a version bump) |
 | Ingest receipts — the resolved commit, fetch time, outcome and errors per instance | `receipts/<instance>.json` | `bin/pyauto-pulse board` / `fetch` / `check` |
 | Last-good snapshots — the summary as captured, with its commit and fetch time | `snapshots/<instance>.json` | the same, on a good read only |
+| Frozen snapshots — a producer's stamped baseline (`day_zero`), captured once at one commit with a manifest; never rewritten | `snapshots/<label>/<instance>.json` | `bin/pyauto-pulse freeze` (by a human-approved task); verified by `check` |
 | The board | `dashboard.md`, `dashboard.html`, `state.json`, `badge.json` | `bin/pyauto-pulse board` (and `dashboard_refresh.yml`) |
 
 Repository location and GitHub identity are **not** owned here: a registry row
@@ -193,6 +194,7 @@ bin/pyauto-pulse check            # registry + body map, ingest at one commit, r
 bin/pyauto-pulse check --offline  # the same over the committed receipts + snapshots (CI)
 bin/pyauto-pulse board            # ingest, then render dashboard.md/.html, badge.json, state.json
 bin/pyauto-pulse census           # one line per instance from the committed snapshots
+bin/pyauto-pulse freeze --instance lens --label day_zero --commit <sha>  # frozen baseline, once
 bin/pyauto-pulse fetch [--instance K]  # ingest + receipt only
 ```
 

REFERENCE.md

@@ -262,7 +262,25 @@ threads}` and the matching `setup_id`, `host_class`, `model_variant` (e.g.
 - `superseded_by` is `null` unless `status: superseded`, when it names another
   record in the same document.
 - `day_zero`, `release_pin`, `decision_id`: string or `null`. `measured_at`:
-  ISO-8601 UTC.
+  ISO-8601 UTC. A non-null `day_zero` (the producer's frozen-baseline tag,
+  lifecycle report §3.6) is allowed only on a `current` record
+  (autolens_profiling#499; additive, no version bump: no capture before it
+  stamped a superseded row).
+
+**Frozen snapshots** (`snapshots/<label>/<instance>.json`, schema
+`pulse-frozen-snapshot` v1; `pulse/freeze.py`). `bin/pyauto-pulse freeze
+--instance lens --label day_zero --commit <sha>` reads the instance's v3
+catalogue at that commit, validates it (decision ids resolved), and keeps the
+records the producer stamped `day_zero` (exactly one tag, or the freeze is
+refused) with the catalogue's `planned_cells` and a manifest: `commit`, `tag`,
+`frozen_at`, `source` (schema, generated_at, producer_revision, sha256 of the
+full catalogue, its record count), `counts` (records, cells, by_class,
+with_decision), `decision_ids` and `records_sha256`. A frozen snapshot is never
+rewritten: a new baseline is a new label. `pyauto-pulse check` verifies each
+offline: the digest still matches its records, every record is `current` on
+the tag, and every `decision_id` resolves. Pulse selects nothing: the producer's
+stamp decides what is day zero. The rolling `snapshots/<instance>.json` is
+unrelated and is overwritten on every good read.
 - `figures`: list of `{kind, path, sha256}` with safe repo-relative paths and
   64-hex digests (may be empty).
 

pulse/catalogue.py

@@ -607,6 +607,12 @@ def _record_v3(check, item, ids):
             "class optimal requires a decision_id (a Pulse designation decision)",
         )
     _nullable_text(check, item, "day_zero", where)
+    if item.get("day_zero") is not None:
+        check.require(
+            item.get("status") == "current",
+            f"{where}.day_zero",
+            "only a current record carries a day_zero stamp",
+        )
     _nullable_text(check, item, "release_pin", where)
     check.require(
         summary.parse_utc(item.get("measured_at")) is not None,

pulse/cli.py

@@ -1,4 +1,4 @@
-"""``pyauto-pulse``: the organ's command line (check, board, census, fetch)."""
+"""``pyauto-pulse``: the organ's command line (check, board, census, fetch, freeze)."""
 
 from __future__ import annotations
 
@@ -9,7 +9,7 @@
 import sys
 from pathlib import Path
 
-from pulse import ORGAN_ROOT, board, campaigns, decisions, ingest, registry
+from pulse import ORGAN_ROOT, board, campaigns, decisions, freeze, ingest, registry
 from pulse import summary as summary_mod
 
 
@@ -255,10 +255,46 @@ def cmd_check(args) -> int:
         else:
             print("ok   dashboard: current with every receipt")
     problems += _check_state(out / "state.json")
+    try:
+        frozen_ok, frozen_problems = freeze.verify(out)
+    except ValueError as exc:
+        frozen_ok, frozen_problems = [], [f"frozen snapshots: {exc}"]
+    for line in frozen_ok:
+        print(f"ok   frozen {line}")
+    for line in frozen_problems:
+        print(f"FAIL frozen {line}")
+    problems += frozen_problems
     print("check: " + ("FAIL" if problems else "OK"))
     return 1 if problems else 0
 
 
+def cmd_freeze(args) -> int:
+    instances, sources = _load(args)
+    match = [i for i in instances if i.instance == args.instance]
+    if not match:
+        print(f"pyauto-pulse: no instance {args.instance!r}", file=sys.stderr)
+        return 1
+    try:
+        path = freeze.freeze(
+            match[0],
+            args.label,
+            args.commit,
+            Path(args.out),
+            ingest.utc_now(),
+            local=sources.get(args.instance),
+        )
+    except (freeze.FreezeError, ingest.FetchError) as exc:
+        print(f"FAIL freeze: {exc}")
+        return 1
+    snap = json.loads(path.read_text())
+    counts = snap["counts"]
+    print(
+        f"ok   freeze: {path.relative_to(Path(args.out))} — tag {snap['tag']}, "
+        f"{counts['records']} records over {counts['cells']} cells at {snap['commit'][:12]}"
+    )
+    return 0
+
+
 def main(argv=None) -> int:
     parser = argparse.ArgumentParser(
         prog="pyauto-pulse",
@@ -311,6 +347,25 @@ def common(p, reading=True):
     common(p, reading=False)
     p.add_argument("--instance", metavar="K", help="only this instance")
     p.set_defaults(func=cmd_fetch)
+    p = sub.add_parser(
+        "freeze",
+        help="write the frozen snapshot snapshots/<label>/<instance>.json from one commit's "
+        "day_zero-stamped records (never overwrites)",
+    )
+    common(p, reading=False)
+    p.add_argument("--instance", metavar="K", required=True, help="the instance to freeze")
+    p.add_argument("--label", default="day_zero", help="the snapshot label (default day_zero)")
+    p.add_argument(
+        "--commit", required=True, metavar="SHA", help="the producer commit to freeze (40-hex)"
+    )
+    p.add_argument(
+        "--from",
+        dest="sources",
+        action="append",
+        metavar="[INSTANCE=]PATH",
+        help="read the summary from this local file instead of fetching it at --commit",
+    )
+    p.set_defaults(func=cmd_freeze)
 
     args = parser.parse_args(argv)
     try:

pulse/freeze.py

@@ -0,0 +1,202 @@
+"""Frozen snapshots: a producer's day-zero baseline, captured once and never rewritten.
+
+``pyauto-pulse freeze --instance lens --label day_zero --commit <sha>`` reads the instance's
+``profiling-summary@3`` catalogue at one commit, validates it (decision ids resolved against
+this organ's ``decisions.yaml``), keeps the records the producer stamped ``day_zero: <tag>``
+(lifecycle report §3.6; autolens_profiling#499) and writes
+``snapshots/<label>/<instance>.json``: those records, the catalogue's ``planned_cells`` and a
+manifest (source commit, tag, counts, the sha256 of the full catalogue and of the frozen
+records). It refuses to overwrite a frozen snapshot: a new baseline is a new label.
+
+``pyauto-pulse check`` verifies every frozen snapshot offline (``verify``): its manifest
+digest still matches its records, every record is ``current`` and carries the snapshot's tag,
+and every ``decision_id`` resolves. Pulse selects nothing here: the producer's stamp decides
+which records are day zero; Pulse only keeps the capture.
+
+The rolling last-good snapshot (``snapshots/<instance>.json``, ``ingest.py``) is a different
+thing: it is overwritten on every good read. A frozen snapshot lives one folder down.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from pathlib import Path
+
+from pulse import ORGAN_ROOT, decisions, ingest
+
+SCHEMA = "pulse-frozen-snapshot"
+VERSION = 1
+_LABEL = re.compile(r"^[a-z0-9]+(?:_[a-z0-9]+)*$")
+_SHA = re.compile(r"^[0-9a-f]{40}$")
+
+
+class FreezeError(ValueError):
+    """A freeze that must not be written, or a frozen snapshot that fails verification."""
+
+
+def frozen_path(out_dir: Path, label: str, instance: str) -> Path:
+    return Path(out_dir) / "snapshots" / label / f"{instance}.json"
+
+
+def _digest(value) -> str:
+    return hashlib.sha256(
+        json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode()
+    ).hexdigest()
+
+
+def _counts(records: list[dict]) -> dict:
+    cells = {
+        (
+            r["likelihood"]["dataset_type"],
+            r["likelihood"]["variant"],
+            r["instrument"],
+            r["setup_id"],
+            r["host_class"],
+            r.get("model_variant", "default"),
+            r.get("regime"),
+        )
+        for r in records
+    }
+    by_class = {}
+    for r in records:
+        by_class[r["class"]] = by_class.get(r["class"], 0) + 1
+    return {
+        "records": len(records),
+        "cells": len(cells),
+        "by_class": dict(sorted(by_class.items())),
+        "with_decision": sum(1 for r in records if r.get("decision_id")),
+    }
+
+
+def build(inst, doc: dict, raw: bytes, commit: str, label: str, frozen_at: str, root=ORGAN_ROOT):
+    """The frozen snapshot document for one validated v3 catalogue at ``commit``."""
+    if not _LABEL.match(label):
+        raise FreezeError(f"label {label!r} must be lower_snake_case")
+    if not _SHA.match(commit):
+        raise FreezeError(f"commit {commit!r} must be a 40-hex SHA")
+    outcome, errors = ingest.classify(doc, ("profiling-summary", 3), root)
+    if outcome != "ok":
+        raise FreezeError(f"{inst.instance}@{commit[:12]} is {outcome}: " + "; ".join(errors[:5]))
+    stamped = [r for r in doc["records"] if r.get("day_zero") is not None]
+    tags = sorted({r["day_zero"] for r in stamped})
+    if len(tags) != 1:
+        raise FreezeError(
+            f"{inst.instance}@{commit[:12]}: a frozen snapshot needs exactly one day_zero tag, "
+            f"found {tags or 'none'}"
+        )
+    records = sorted(stamped, key=lambda r: r["id"])
+    return {
+        "schema": SCHEMA,
+        "version": VERSION,
+        "label": label,
+        "instance": inst.instance,
+        "repo": inst.repo,
+        "github": inst.github,
+        "summary_path": inst.summary_path,
+        "commit": commit,
+        "frozen_at": frozen_at,
+        "tag": tags[0],
+        "source": {
+            "schema": f"{doc['schema']}@{doc['version']}",
+            "generated_at": doc.get("generated_at"),
+            "producer_revision": doc.get("producer_revision"),
+            "sha256": hashlib.sha256(raw).hexdigest(),
+            "record_count": len(doc["records"]),
+        },
+        "counts": _counts(records),
+        "decision_ids": sorted({r["decision_id"] for r in records if r.get("decision_id")}),
+        "records_sha256": _digest(records),
+        "planned_cells": doc.get("planned_cells") or [],
+        "records": records,
+    }
+
+
+def write(out_dir: Path, snap: dict) -> Path:
+    """Write a frozen snapshot once; an existing one is never replaced."""
+    path = frozen_path(out_dir, snap["label"], snap["instance"])
+    if path.exists():
+        raise FreezeError(f"{path} exists: a frozen snapshot is never rewritten (use a new label)")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(snap, indent=1, sort_keys=True, ensure_ascii=False) + "\n")
+    return path
+
+
+def freeze(
+    inst, label: str, commit: str, out_dir: Path, frozen_at: str, local=None, root=ORGAN_ROOT
+):
+    """Fetch ``inst``'s summary at ``commit`` (or read ``local``), build and write the snapshot."""
+    if local is not None:
+        raw = Path(local).read_bytes()
+    else:
+        raw = ingest.fetch_at(inst.github, commit, inst.summary_path)
+    try:
+        doc = json.loads(raw)
+    except ValueError as exc:
+        raise FreezeError(f"{inst.instance}@{commit[:12]}: unreadable summary ({exc})") from exc
+    if frozen_path(out_dir, label, inst.instance).exists():
+        raise FreezeError(
+            f"{frozen_path(out_dir, label, inst.instance)} exists: a frozen snapshot is never "
+            "rewritten (use a new label)"
+        )
+    return write(out_dir, build(inst, doc, raw, commit, label, frozen_at, root))
+
+
+def verify_one(path: Path, known_decisions: set[str]) -> list[str]:
+    """Problems with one frozen snapshot (empty when it is intact)."""
+    try:
+        snap = json.loads(Path(path).read_text())
+    except (OSError, ValueError) as exc:
+        return [f"{path}: unreadable ({exc})"]
+    where = f"{path.parent.name}/{path.name}"
+    if snap.get("schema") != SCHEMA or snap.get("version") != VERSION:
+        return [f"{where}: not a {SCHEMA}@{VERSION} document"]
+    problems = []
+    if snap.get("label") != path.parent.name or snap.get("instance") != path.stem:
+        problems.append(f"{where}: label/instance do not match its path")
+    records = snap.get("records") or []
+    if snap.get("records_sha256") != _digest(records):
+        problems.append(f"{where}: records changed since the freeze (records_sha256 mismatch)")
+    if snap.get("counts") != _counts(records):
+        problems.append(f"{where}: counts do not match its records")
+    tag = snap.get("tag")
+    off = [r.get("id") for r in records if r.get("day_zero") != tag or r.get("status") != "current"]
+    if off:
+        problems.append(f"{where}: {len(off)} record(s) not current on tag {tag}, e.g. {off[0]}")
+    wanted = {r["decision_id"] for r in records if r.get("decision_id")}
+    if sorted(wanted) != snap.get("decision_ids"):
+        problems.append(f"{where}: decision_ids do not match its records")
+    missing = sorted(wanted - known_decisions)
+    if missing:
+        problems.append(f"{where}: decision_id(s) not in decisions.yaml: {', '.join(missing[:5])}")
+    if not records:
+        problems.append(f"{where}: holds no records")
+    return problems
+
+
+def frozen_snapshots(out_dir: Path) -> list[Path]:
+    """Every ``snapshots/<label>/<instance>.json`` (the rolling ones sit one level up)."""
+    return sorted(p for p in (Path(out_dir) / "snapshots").glob("*/*.json"))
+
+
+def verify(out_dir: Path, root=ORGAN_ROOT) -> tuple[list[str], list[str]]:
+    """(ok lines, problems) for every frozen snapshot under ``out_dir``."""
+    paths = frozen_snapshots(out_dir)
+    if not paths:
+        return [], []
+    known = decisions.ids(root)
+    ok, problems = [], []
+    for path in paths:
+        found = verify_one(path, known)
+        if found:
+            problems += found
+            continue
+        snap = json.loads(path.read_text())
+        counts = snap["counts"]
+        ok.append(
+            f"{snap['label']}/{snap['instance']}: tag {snap['tag']}, {counts['records']} records "
+            f"over {counts['cells']} cells at {snap['commit'][:12]}, "
+            f"{len(snap['decision_ids'])} decision(s) resolved"
+        )
+    return ok, problems