[Jammy2211]

Closes #83. Epic likelihood-lifecycle, phase 3 (day zero). Heart RED override invoked by the human 2026-10-31 (recorded on #83).

What

On 2026-10-31 the human approved all 39 designatable likelihood × hardware-class groups ("Approve all 39 (Recommended)"). They held datacube/delaunay gpu_a100 as provisional ("Hold, keep provisional (Recommended)"): W~ sparse against dense NUFFT is 1.03× at sma, inside the 9 % tolerance.

• decisions/<id>.md ×39. Generated from autolens_profiling dashboard/catalogue.json at c218a428 (PR #496 head, the day-zero rows). Every evidence link is pinned to that commit. Each record gives:
  • the per-instrument table (projected calls/s per node, per call, bound, parity, record id);
  • the separation verdict on the pre-registered axis (human Q1/Q3: per-node throughput; separation beyond the larger p10–p90 spread, else 9 %);
  • the optimal and secondary setups and the rule;
  • basis: projected (Q4 a);
  • revisit conditions.
• decisions.yaml: 39 kind: designation rows. pulse.decisions.load() validates them.
• Test: the placeholder guard test_the_live_index_holds_no_designation_yet becomes test_the_live_index_holds_the_day_zero_designations (39 rows, unique keys, dated 2026-10-31, datacube A100 absent).
• Dashboard: re-rendered (board --offline). pyauto-pulse check --offline OK, 370 tests pass, ruff clean.

The set

• CPU (cpu_many_core):
  • numba (numba_cpu_sparse_fp64, or …_wtilde for interferometer and datacube) for every pixelized likelihood: imaging, group, multi_galaxy, multi_dataset, interferometer, datacube. It separates by 2.17–49.6× per node.
  • JAX CPU where it is the only setup: MGE, point source, cluster, weak.
• A100 (gpu_a100): jax_gpu_dense_fp64 for imaging-like likelihoods, jax_gpu_sparse_fp64_wtilde for interferometer, and jax_gpu_na_fp64 for point source and cluster.

Full derivation: PyAutoMind draft/research/autolens_profiling/likelihood_lifecycle_p3_designation_proposals.md ("Re-derivation 2026-10-31", Appendix B).

Order

The evidence ids are path-hashed, so they are identical on main once autolens_profiling #496 merges. Merge #496 first, or alongside. The producer side comes next and is a separate task: class: optimal + decision_id on the cited rows, and retiring provisional_designations.json.

Generators (reproducibility)

designate.py — the ranking:

"""Re-derive phase-3 designation proposals from a frozen autolens_profiling catalogue.

Axis (pre-registered, report amendment 2026-10-25; human Q1/Q3 2026-10-30): within each hardware
class rank setups by projected calls/s per node; record a designation only where the winner
separates from the runner-up beyond the recorded spread (row p10-p90), else beyond the 9 %
arm-to-arm noise; otherwise keep the provisional label. Q2: gpu_a100 = GPU setups only.
"""
import json, re, sys, collections
from pathlib import Path

root = Path(sys.argv[1]); old_appendix = Path(sys.argv[2])
cat = json.loads((root / "dashboard/catalogue.json").read_text())
HW = {"hpc_a100": "gpu_a100", "hpc_ral_cpu": "cpu_many_core"}
NOISE = 0.09

def spread(path):
    try: row = json.loads((root / path).read_text())
    except Exception: return None
    lo, hi, med = (row.get(f"full_pipeline_single_jit_{k}") for k in ("p10_ms", "p90_ms", "median_ms"))
    if lo and hi and med: return (hi - lo) / med
    return None

cells = collections.defaultdict(lambda: collections.defaultdict(dict))
for r in cat["records"]:
    if r.get("status") != "current" or r["measurement"] != "runtime" or r.get("location") != "cell": continue
    if r.get("model_variant", "default") != "default" or r["host_class"] not in HW: continue
    pnt = r.get("projected_node_throughput") or {}
    if r.get("regime", "standard") != "standard": continue  # off-plan sizes (scans) never designate
    hw = HW[r["host_class"]]
    if hw == "gpu_a100" and r["setup"]["device"] != "gpu": continue  # Q2
    lik = f'{r["likelihood"]["dataset_type"]}/{r["likelihood"]["variant"]}'
    key = (lik, r.get("regime", "standard"), hw)
    cells[key][r["instrument"]][r["setup_id"]] = dict(
        cps=pnt["hosts"][0]["calls_per_s"] if pnt.get("status") == "projected" else None,
        per_call=pnt.get("per_call_s") or r["values"][r["metric"]], id=r["id"], why=pnt.get("reason"),
        spread=spread(r["evidence"]["path"]), cls=r.get("class"))

parity = {(w["path"].rsplit("/", 2)[0]): w for w in (cat["parity_witnesses"] if isinstance(cat["parity_witnesses"], list) else [])}

old = {}
for m in re.finditer(r"- id: (\S+).*?optimal: (\S+)", old_appendix.read_text(), re.S):
    old[m.group(1)] = m.group(2)

def kebab(lik, regime, hw): return f"{lik.replace('/', '-').replace('_', '-')}-{regime}-{hw.replace('_', '-')}"

out = []
for key in sorted(cells):
    lik, regime, hw = key
    insts = cells[key]
    winners, verdicts, detail, ranks = {}, [], [], {}
    for inst, setups in sorted(insts.items()):
        per_node = all(v["cps"] for v in setups.values())
        score = (lambda v: v["cps"]) if per_node else (lambda v: 1 / v["per_call"])
        ranked = sorted(setups.items(), key=lambda kv: -score(kv[1]))
        if not per_node: detail.append(f"{inst}: PER-CALL FALLBACK ({[k for k, v in setups.items() if not v['cps']]} not projected)")
        w, wv = ranked[0]; winners[inst] = w
        if len(ranked) == 1:
            verdicts.append("uncontested"); detail.append(f"{inst}: {w} (only setup)"); continue
        r2, rv = ranked[1]
        ratio = score(wv) / score(rv)
        tol = max([s for s in (wv["spread"], rv["spread"]) if s] or [NOISE])
        sep = ratio > 1 + tol
        verdicts.append("separated" if sep else "close")
        unit = "calls/s/node" if per_node else "1/per-call"
        detail.append(f"{inst}: {w} {score(wv):.3g} vs {r2} {score(rv):.3g} {unit} = {ratio:.2f}x (tol {tol:.0%}) {'SEP' if sep else 'CLOSE'}")
        ranks.setdefault(inst, [k for k, _ in ranked])
    same = len(set(winners.values())) == 1
    if not same: label = "split by instrument"
    elif all(v == "uncontested" for v in verdicts): label = "uncontested"
    elif all(v in ("separated", "uncontested") for v in verdicts): label = "separated"
    else: label = "does not separate"
    opt = next(iter(winners.values())) if same else None
    did = kebab(lik, regime, hw)
    prev = old.get(did)
    change = "" if prev is None else ("same" if prev == opt else f"CHANGED (was {prev})")
    sec = []
    if opt:
        others = collections.Counter(k for rk in ranks.values() for k in rk[1:2])
        sec = [k for k, _ in others.most_common()]
    out.append(dict(id=did, secondary=sec, lik=lik, regime=regime, hw=hw, label=label, optimal=opt, winners=winners,
                    detail=detail, prev=prev, change=change,
                    evidence=[cells[key][i][s]["id"] for i, s in sorted(winners.items())]))
json.dump(out, open(sys.argv[3], "w"), indent=1)
for o in out:
    print(f"{o['id']:52s} {o['label']:20s} {o['optimal'] or '-':32s} prev={o['prev'] or 'n/a'} {o['change']}")

appendix_b.py — entries + _designation validation:

import json, sys, collections, re
from pathlib import Path
sys.path.insert(0, str(Path.home() / "Code/PyAutoLabs/organs/PyAutoPulse"))
from pulse.decisions import _designation

root = Path(sys.argv[1]); props = json.load(open(sys.argv[2])); old = Path(sys.argv[3]).read_text()
cat = json.loads((root / "dashboard/catalogue.json").read_text())
HW = {"hpc_a100": "gpu_a100", "hpc_ral_cpu": "cpu_many_core"}
par = {w["path"].rsplit("/", 1)[0]: w["record_id"] for w in cat["parity_witnesses"]}
ev = collections.defaultdict(list)  # (lik, hw) -> [(inst, setup, runtime id, parity id)]
for r in cat["records"]:
    if r.get("status") != "current" or r["measurement"] != "runtime" or r.get("location") != "cell": continue
    if r.get("regime", "standard") != "standard" or r["host_class"] not in HW: continue
    hw = HW[r["host_class"]]
    if hw == "gpu_a100" and r["setup"]["device"] != "gpu": continue
    lik = f'{r["likelihood"]["dataset_type"]}/{r["likelihood"]["variant"]}'
    cell = r["evidence"]["path"].rsplit("/", 1)[0]
    ev[(lik, hw)].append((r["instrument"], r["setup_id"], r["id"], par.get(cell)))

old_rules = dict(re.findall(r"- id: (\S+)\n(?:.*\n)*?  rule: (.*)", old))
SRC = {"mge": "a parametric (MGE) source", "rectangular": "a rectangular pixelized source",
       "delaunay": "a Delaunay pixelized source", "delaunay_nn": "a Delaunay natural-neighbour source"}
DS = {"imaging": "Any imaging dataset", "interferometer": "Any interferometer dataset",
      "multi_galaxy": "A multi-galaxy imaging dataset", "group": "A group-scale imaging dataset",
      "multi_dataset": "A multi-dataset imaging fit (several datasets fitted together)",
      "datacube": "A datacube (interferometer channels fitted together)"}
HWP = {"gpu_a100": "on an A100 node", "cpu_many_core": "on a many-core CPU node with no GPU"}
UNTESTED = {"interferometer": {"gpu_a100": " Untested: the GPU streaming route (measured up to jvla, 2.5e7 visibilities, on W~ sparse).",
                               "cpu_many_core": " Untested: CPU streaming and the MGE DFT route."}}

def rule(p):
    if p["id"] in old_rules and p["prev"] == p["optimal"]: return old_rules[p["id"]]
    ds, var = p["lik"].split("/")
    hw = HWP[p["hw"]]
    if ds in ("point_source_image", "point_source_source", "cluster"):
        scale = "cluster-scale" if ds == "cluster" else "galaxy-scale"
        plane = "image" if var == "image_plane" else "source"
        s = f"A {scale} point-source dataset fitted in the {plane} plane {hw} uses {p['optimal']}."
    elif ds == "weak":
        s = f"A weak-lensing shear dataset with an isothermal lens {hw} uses {p['optimal']}."
    else:
        s = f"{DS[ds]} with {SRC[var]} {hw} uses {p['optimal']}"
        if p["optimal"].startswith("numba"): s += ", one serial likelihood per core (workers capped by node memory over peak RSS)"
        if p["secondary"]: s += f", falling back to {p['secondary'][0]}"
        s += "."
    return s + UNTESTED.get(p["lik"].split("/")[0], {}).get(p["hw"], "")

lines, bad = [], []
for p in props:
    e = [x for i in sorted(ev[(p["lik"], p["hw"])]) for x in (i[2], i[3]) if x]
    entry = {"id": p["id"], "title": f"Designation: {p['lik']} standard {p['hw']}", "date": "YYYY-MM-DD",
             "url": f"https://github.com/PyAutoLabs/PyAutoPulse/blob/main/decisions/{p['id']}.md",
             "kind": "designation", "likelihood": p["lik"], "regime": "standard", "hardware_class": p["hw"],
             "optimal": p["optimal"] or "HOLD", "secondary": p["secondary"], "evidence": e,
             "rule": rule(p) if p["optimal"] else "HOLD", "supersedes": None}
    if p["optimal"]:
        try: _designation(entry)
        except ValueError as exc: bad.append(str(exc))
    hold = p["label"] not in ("separated", "uncontested")
    lines.append(f"# separation: {p['label']}" + ("  — HOLD: does not separate; do not record as written" if hold else "") +
                 (f"  — CHANGED from 2026-10-29 (was {p['prev']})" if p["prev"] and p["prev"] != p["optimal"] else ""))
    for k, v in entry.items():
        if k == "evidence" or k == "secondary": v = "[" + ", ".join(v) + "]"
        elif v is None: v = "null"
        elif k in ("title", "date", "rule"): v = "'" + str(v).replace("'", "''") + "'"
        lines.append(("- " if k == "id" else "  ") + f"{k}: {v}")
print("\n".join(lines))
print("VALIDATION:", "OK" if not bad else bad, file=sys.stderr)

gen_records.py — records + index rows:

"""Write the approved day-zero designations as Pulse decision records (PyAutoPulse#83)."""
import json, sys, collections
from pathlib import Path
import yaml

prof, pulse, props_path, appendix, sha = Path(sys.argv[1]), Path(sys.argv[2]), sys.argv[3], sys.argv[4], sys.argv[5]
cat = json.loads((prof / "dashboard/catalogue.json").read_text())
props = {p["id"]: p for p in json.load(open(props_path))}
entries = [e for e in yaml.safe_load(open(appendix).read()) if props[e["id"]]["label"] in ("separated", "uncontested")]
held = [e["id"] for e in yaml.safe_load(open(appendix).read()) if props[e["id"]]["label"] not in ("separated", "uncontested")]
assert held == ["datacube-delaunay-standard-gpu-a100"], held
assert len(entries) == 39, len(entries)
HW = {"hpc_a100": "gpu_a100", "hpc_ral_cpu": "cpu_many_core"}
NODE = {"gpu_a100": "RAL gpu node (4 × A100 80GB PCIe; 124 cores, 885 GB)", "cpu_many_core": "RAL ral node (236 cores, 921 GB)"}
par = {w["path"].rsplit("/", 1)[0]: w for w in cat["parity_witnesses"]}
rows = collections.defaultdict(list)
for r in cat["records"]:
    if r.get("status") != "current" or r["measurement"] != "runtime" or r.get("location") != "cell": continue
    if r.get("regime", "standard") != "standard" or r["host_class"] not in HW: continue
    hw = HW[r["host_class"]]
    if hw == "gpu_a100" and r["setup"]["device"] != "gpu": continue
    rows[(f'{r["likelihood"]["dataset_type"]}/{r["likelihood"]["variant"]}', hw)].append(r)
BLOB = f"https://github.com/PyAutoLabs/autolens_profiling/blob/{sha}"
SOURCE = ('AskUserQuestion answer "Approve all 39 (Recommended)" in the ~/Science Claude Code session, '
          "2026-10-31, on the re-derived proposals (PyAutoMind "
          "`draft/research/autolens_profiling/likelihood_lifecycle_p3_designation_proposals.md`, "
          '"Re-derivation 2026-10-31" and Appendix B). Recorded under PyAutoPulse#83.')

def fmt(x, n=3):
    return f"{x:,.{max(n - 1 - (len(str(int(abs(x)))) - 1 if abs(x) >= 1 else 0), 0)}f}" if x >= 100 else f"{x:.{n}g}"

index = []
for e in entries:
    p = props[e["id"]]; lik, hw = e["likelihood"], e["hardware_class"]
    title = e["title"]
    table = ["| instrument | setup | projected calls/s per node | per call (ms) | bound by | parity | record |", "|---|---|---|---|---|---|---|"]
    for r in sorted(rows[(lik, hw)], key=lambda r: (r["instrument"], r["setup_id"])):
        pnt = r.get("projected_node_throughput") or {}
        h = (pnt.get("hosts") or [{}])[0]
        cps = (f"{h['calls_per_s']:,.0f}" if h.get("calls_per_s", 0) >= 1000 else f"{h.get('calls_per_s', 0):.3g}") if pnt.get("status") == "projected" else f"not projected ({pnt.get('reason', 'no peak RSS')})"
        per = (pnt.get("per_call_s") or r["values"][r["metric"]]) * 1e3
        cell = r["evidence"]["path"].rsplit("/", 1)[0]
        w = par.get(cell, {})
        mark = " **(optimal)**" if r["setup_id"] == e["optimal"] else (" (secondary)" if r["setup_id"] in e["secondary"] else "")
        table.append(f"| {r['instrument']} | `{r['setup_id']}`{mark} | {cps} | {(f"{per:,.0f}" if per >= 1000 else f"{per:.4g}")} | {h.get('bound_by', '—')} | {w.get('status', 'none')} | [`{r['id']}`]({BLOB}/{r['evidence']['path']}) |")
    sep = "\n".join(f"- {d}" for d in p["detail"])
    caveats = [
        "Projected throughput is an upper bound: workers × η / per call with η = 1.0 declared, unmeasured until phase 4 (`basis: projected`, Q4 option a). Nautilus `n_batch` and shared memory bandwidth are not modelled.",
        "Day-zero environment: tag 2026.10.25.1 (private checkout), jax/jaxlib 0.11.2 with `--xla_cpu_multi_thread_eigen=false` on CPU legs; numba rows carried (human Q7, 2026-10-30). JAX CPU rows ran about 1.1–1.5× slower than on jax 0.10.2; the flag and the version are not separated.",
    ]
    if hw == "cpu_many_core": caveats.append("RAL CPU rows of 2a and 2d–2g ran on shared, loaded nodes (human ruling 2026-10-30); the 2b/2c RAL CPU rows are the quiet-node reruns (autolens_profiling#494).")
    if p["label"] == "uncontested": caveats.append("Uncontested: only this setup was measured on this hardware class, so the designation records the measured route rather than a contest between routes.")
    if any("FALLBACK" in d for d in p["detail"]): caveats.append("One instrument is ranked per call: its numba row has no peak RSS (sacct sampled no MaxRSS), so it has no node projection.")
    body = f"""# {title}

Date: 2026-10-31
Decision maker: James Nightingale
Decision source: {SOURCE}

## Campaign

Epic `likelihood-lifecycle`, phase 3 (day zero): one designation per likelihood × regime × hardware class, made by the human from measured, parity-checked cells. Evidence: autolens_profiling `dashboard/catalogue.json` at [`{sha[:8]}`]({BLOB}/dashboard/catalogue.json) (PR #496 head, the day-zero rows), on the {NODE[hw]}.

## Evidence

{chr(10).join(table)}

Separation (pre-registered axis, human Q1/Q3 2026-10-30: rank by projected calls/s per node; separate when the winner beats the runner-up by more than the larger recorded p10–p90 spread, else 9 %): **{p['label']}**.

{sep}

## Options considered

{"Only one setup was measured on this hardware class." if p["label"] == "uncontested" else "The setups in the table, ranked on the pre-registered axis; per-call latency is secondary evidence."}

## Decision

- Optimal: `{e['optimal']}`
- Secondary: {", ".join(f"`{s}`" for s in e["secondary"]) or "none"}
- Rule: {e['rule']}
- Basis: projected.

## Implications

autolens_profiling rows cited here are re-declared `class: optimal` with `decision_id: {e['id']}` by the producer (next phase-3 task); `catalogue/provisional_designations.json` is retired in favour of these records. Recording does not change library defaults; phase 5 generates the library rule table from these records.

## Revisit conditions

{chr(10).join(f"- {c}" for c in caveats)}
- Phase 4 measures parallel efficiency; a measured η that reorders the setups supersedes this record.

## History

First designation for this likelihood, regime and hardware class.
"""
    (pulse / "decisions" / f"{e['id']}.md").write_text(body)
    row = {k: e[k] for k in ("id", "title")}
    row.update(date="2026-10-31", url=e["url"])
    for k in ("kind", "likelihood", "regime", "hardware_class", "optimal", "secondary", "evidence", "rule", "supersedes"):
        row[k] = e[k]
    index.append(row)
doc = yaml.safe_load((pulse / "decisions.yaml").read_text())
have = {d["id"] for d in doc["decisions"]}
assert not have & {r["id"] for r in index}
text = (pulse / "decisions.yaml").read_text().rstrip("\n") + "\n"
for r in index:
    text += f"- id: {r['id']}\n  title: '{r['title']}'\n  date: '{r['date']}'\n  url: {r['url']}\n  kind: designation\n"
    text += f"  likelihood: {r['likelihood']}\n  regime: {r['regime']}\n  hardware_class: {r['hardware_class']}\n  optimal: {r['optimal']}\n"
    text += f"  secondary: [{', '.join(r['secondary'])}]\n  evidence: [{', '.join(r['evidence'])}]\n  rule: '{r['rule'].replace(chr(39), chr(39)*2)}'\n  supersedes: null\n"
(pulse / "decisions.yaml").write_text(text)
print(f"wrote {len(index)} records")

🤖 Generated with Claude Code