freeze: --measurement filter for a per-measurement frozen label (6d)

AGENTS.md

@@ -195,6 +195,7 @@ bin/pyauto-pulse check --offline  # the same over the committed receipts + snaps
 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 freeze --instance lens --label day_zero_gradient --measurement gradient --commit <sha>  # one measurement's stamps
 bin/pyauto-pulse fetch [--instance K]  # ingest + receipt only
 ```
 

REFERENCE.md

@@ -275,8 +275,11 @@ 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
+with_decision), `decision_ids` and `records_sha256`. `--measurement <m>` keeps
+only the stamped records of that measurement and writes `measurement` into the
+snapshot, which `check` then enforces: `--label day_zero_gradient --measurement
+gradient` freezes the phase-6 gradient stamps beside the runtime `day_zero`. 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

pulse/cli.py

@@ -282,6 +282,7 @@ def cmd_freeze(args) -> int:
             Path(args.out),
             ingest.utc_now(),
             local=sources.get(args.instance),
+            measurement=args.measurement,
         )
     except (freeze.FreezeError, ingest.FetchError) as exc:
         print(f"FAIL freeze: {exc}")
@@ -358,6 +359,11 @@ def common(p, reading=True):
     p.add_argument(
         "--commit", required=True, metavar="SHA", help="the producer commit to freeze (40-hex)"
     )
+    p.add_argument(
+        "--measurement",
+        metavar="M",
+        help="keep only the stamped records of this measurement (e.g. gradient)",
+    )
     p.add_argument(
         "--from",
         dest="sources",

pulse/freeze.py

@@ -8,6 +8,11 @@
 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.
 
+``--measurement <m>`` keeps only the stamped records of that measurement and records it in the
+snapshot (``measurement``), so a later family of stamps gets its own frozen label without
+touching an earlier one: ``freeze --label day_zero_gradient --measurement gradient`` freezes the
+phase-6 gradient records while ``day_zero`` keeps the runtime baseline (autolens_profiling#512).
+
 ``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
@@ -70,24 +75,42 @@ def _counts(records: list[dict]) -> dict:
     }
 
 
-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``."""
+def build(
+    inst,
+    doc: dict,
+    raw: bytes,
+    commit: str,
+    label: str,
+    frozen_at: str,
+    root=ORGAN_ROOT,
+    measurement: str | None = None,
+):
+    """The frozen snapshot document for one validated v3 catalogue at ``commit``.
+
+    With ``measurement``, only the stamped records of that measurement are kept.
+    """
     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]
+    stamped = [
+        r
+        for r in doc["records"]
+        if r.get("day_zero") is not None
+        and (measurement is None or r.get("measurement") == measurement)
+    ]
     tags = sorted({r["day_zero"] for r in stamped})
     if len(tags) != 1:
+        scope = f" {measurement}" if measurement else ""
         raise FreezeError(
-            f"{inst.instance}@{commit[:12]}: a frozen snapshot needs exactly one day_zero tag, "
-            f"found {tags or 'none'}"
+            f"{inst.instance}@{commit[:12]}: a frozen snapshot needs exactly one day_zero tag "
+            f"on its{scope} records, found {tags or 'none'}"
         )
     records = sorted(stamped, key=lambda r: r["id"])
-    return {
+    snap = {
         "schema": SCHEMA,
         "version": VERSION,
         "label": label,
@@ -111,6 +134,9 @@ def build(inst, doc: dict, raw: bytes, commit: str, label: str, frozen_at: str,
         "planned_cells": doc.get("planned_cells") or [],
         "records": records,
     }
+    if measurement is not None:
+        snap["measurement"] = measurement
+    return snap
 
 
 def write(out_dir: Path, snap: dict) -> Path:
@@ -124,7 +150,14 @@ def write(out_dir: Path, snap: dict) -> Path:
 
 
 def freeze(
-    inst, label: str, commit: str, out_dir: Path, frozen_at: str, local=None, root=ORGAN_ROOT
+    inst,
+    label: str,
+    commit: str,
+    out_dir: Path,
+    frozen_at: str,
+    local=None,
+    root=ORGAN_ROOT,
+    measurement: str | None = None,
 ):
     """Fetch ``inst``'s summary at ``commit`` (or read ``local``), build and write the snapshot."""
     if local is not None:
@@ -140,7 +173,7 @@ def freeze(
             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))
+    return write(out_dir, build(inst, doc, raw, commit, label, frozen_at, root, measurement))
 
 
 def verify_one(path: Path, known_decisions: set[str]) -> list[str]:
@@ -164,6 +197,13 @@ def verify_one(path: Path, known_decisions: set[str]) -> list[str]:
     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]}")
+    measurement = snap.get("measurement")
+    if measurement is not None:
+        other = [r.get("id") for r in records if r.get("measurement") != measurement]
+        if other:
+            problems.append(
+                f"{where}: {len(other)} record(s) not measurement {measurement}, e.g. {other[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")

tests/test_freeze.py

@@ -27,10 +27,12 @@ def root(tmp_path):
     return root
 
 
-def _freeze(root, doc=None, label="day_zero"):
+def _freeze(root, doc=None, label="day_zero", measurement=None):
     doc = doc or fixture_doc("setup_summary_v3.json")
     raw = json.dumps(doc).encode()
-    snap = freeze.build(LENS, doc, raw, SHA_A, label, "2026-11-01T00:00:00Z", root)
+    snap = freeze.build(
+        LENS, doc, raw, SHA_A, label, "2026-11-01T00:00:00Z", root, measurement=measurement
+    )
     return freeze.write(root, snap), snap
 
 
@@ -91,6 +93,45 @@ def test_freeze_refuses_a_catalogue_with_no_or_two_day_zero_tags(root):
         _freeze(root, two)
 
 
+def _two_measurements():
+    """The fixture with its breakdown record also stamped, on its own tag."""
+    doc = fixture_doc("setup_summary_v3.json")
+    doc["records"][2]["day_zero"] = "2026.10.25.1"
+    assert doc["records"][2]["measurement"] == "breakdown"
+    return doc
+
+
+def test_freeze_measurement_keeps_only_that_measurement_under_its_own_label(root):
+    """#512: a gradient-style family freezes beside the runtime baseline, never into it."""
+    doc = _two_measurements()
+    path, snap = _freeze(root, doc, label="day_zero_breakdown", measurement="breakdown")
+    assert path == root / "snapshots" / "day_zero_breakdown" / "lens.json"
+    assert snap["measurement"] == "breakdown" and snap["tag"] == "2026.10.25.1"
+    assert [r["id"] for r in snap["records"]] == ["imaging-delaunay-hst-a100-breakdown"]
+    _path, runtime = _freeze(root, doc, measurement="runtime")
+    assert [r["id"] for r in runtime["records"]] == ["imaging-delaunay-hst-a100-runtime"]
+    ok, problems = freeze.verify(root, root)
+    assert problems == [] and len(ok) == 2
+    # Without --measurement the two families are two tags: refused, as before.
+    with pytest.raises(freeze.FreezeError, match="exactly one day_zero tag"):
+        _freeze(root, doc, label="day_zero_all")
+
+
+def test_freeze_measurement_refuses_when_none_of_it_is_stamped(root):
+    with pytest.raises(freeze.FreezeError, match="on its gradient records, found none"):
+        _freeze(root, measurement="gradient")
+
+
+def test_verify_catches_a_record_of_another_measurement(root):
+    path, _snap = _freeze(root, _two_measurements(), "day_zero_breakdown", "breakdown")
+    snap = json.loads(path.read_text())
+    snap["records"][0]["measurement"] = "runtime"
+    snap["records_sha256"] = freeze._digest(snap["records"])
+    path.write_text(json.dumps(snap))
+    _ok, problems = freeze.verify(root, root)
+    assert any("not measurement breakdown" in p for p in problems)
+
+
 def test_freeze_refuses_an_invalid_catalogue(root):
     _decisions(root, ["other"])
     with pytest.raises(freeze.FreezeError, match="invalid"):
@@ -135,3 +176,9 @@ def test_cli_freeze_fetches_the_commit_and_refuses_a_second_write(
     assert snap["commit"] == SHA_A and snap["summary_path"] == "dashboard/summary.json"
     assert cli.main(args) == 1
     assert "never rewritten" in capsys.readouterr().out
+    # --measurement freezes one family under its own label; none of it stamped is refused.
+    assert cli.main(args + ["--label", "day_zero_runtime", "--measurement", "runtime"]) == 0
+    snap = json.loads((out / "snapshots" / "day_zero_runtime" / "lens.json").read_text())
+    assert snap["measurement"] == "runtime"
+    assert cli.main(args + ["--label", "day_zero_gradient", "--measurement", "gradient"]) == 1
+    assert "on its gradient records, found none" in capsys.readouterr().out