feat: profiling_drift vitals leg — pinned-value drift from autolens_profiling

heart/checks/profiling_drift.py

@@ -0,0 +1,142 @@
+"""heart/checks/profiling_drift.py — scan autolens_profiling result JSONs
+for pinned-value drift flags.
+
+autolens_profiling's runtime cells compare each run's log-likelihood /
+log-evidence against pinned baseline values and **record** the outcome in
+every result JSON instead of crashing (autolens_profiling#54):
+
+- ``pinned_expected`` — the baseline value, or ``null`` when the instrument
+  has no pin;
+- ``pinned_drift`` — list of ``{label, expected, got, rel_diff, rtol}``
+  records; an empty list means every compared value matched.
+
+A non-empty ``pinned_drift`` is a health finding either way — a library
+regression or a stale pin — and the profiling baselines are non-comparable
+until it is resolved (boundary rule:
+``autolens_profiling/results/notes/design_lock_in.md``). Heart observes and
+reports; adjudication belongs to autolens_workspace_test / the library repos.
+
+Inputs:
+- autolens_profiling/results/**/*.json (read-only; repo absent → not observed)
+
+Output:
+- ~/.pyauto-heart/profiling_drift.json with the latest scan summary.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+HEART_STATE_DIR = Path(
+    os.environ.get("HEART_STATE_DIR")
+    or Path.home() / ".pyauto-heart"
+)
+HEART_HOME = Path(__file__).resolve().parents[2]
+PYAUTO_ROOT = Path(__file__).resolve().parents[3] if Path(__file__).resolve().parents[3].name == "PyAutoLabs" else Path.home() / "Code" / "PyAutoLabs"
+RESULTS_ROOT = PYAUTO_ROOT / "autolens_profiling" / "results"
+
+
+def scan_results(results_root: Path) -> dict[str, Any]:
+    """Scan every result JSON under results_root for pinned-drift flags."""
+    files_scanned = 0
+    pinned_checked = 0
+    findings: list[dict[str, Any]] = []
+
+    for json_path in sorted(results_root.rglob("*.json")):
+        try:
+            data = json.loads(json_path.read_text())
+        except (json.JSONDecodeError, OSError):
+            continue
+        if not isinstance(data, dict) or "pinned_drift" not in data:
+            continue
+        files_scanned += 1
+        if data.get("pinned_expected") is not None:
+            pinned_checked += 1
+        drift = data.get("pinned_drift") or []
+        if not drift:
+            continue
+        findings.append({
+            "path": str(json_path.relative_to(results_root)),
+            "instrument": data.get("instrument"),
+            "autolens_version": data.get("autolens_version"),
+            "drift": [
+                {
+                    "label": d.get("label"),
+                    "expected": d.get("expected"),
+                    "got": d.get("got"),
+                    "rel_diff": d.get("rel_diff"),
+                }
+                for d in drift
+                if isinstance(d, dict)
+            ],
+        })
+
+    return {
+        "results_root": str(results_root),
+        "observed": True,
+        "files_scanned": files_scanned,
+        "pinned_checked": pinned_checked,
+        "drift_count": len(findings),
+        "findings": findings,
+    }
+
+
+def run(results_root: Path | None = None) -> dict[str, Any]:
+    """Scan and persist the summary; repo absent → observed: False."""
+    results_root = results_root or RESULTS_ROOT
+
+    if not results_root.is_dir():
+        summary: dict[str, Any] = {
+            "results_root": str(results_root),
+            "observed": False,
+            "files_scanned": 0,
+            "pinned_checked": 0,
+            "drift_count": 0,
+            "findings": [],
+        }
+    else:
+        summary = scan_results(results_root)
+
+    sys.path.insert(0, str(HEART_HOME))
+    from heart import state
+
+    state.atomic_write_json(HEART_STATE_DIR / "profiling_drift.json", summary)
+    return summary
+
+
+def main(argv: list[str]) -> int:
+    results_root = Path(argv[1]) if len(argv) > 1 else RESULTS_ROOT
+    summary = run(results_root)
+
+    from heart_color import c_ok, c_warn, c_info, c_meta, glyph_ok, glyph_warn
+
+    n_drift = summary["drift_count"]
+    n_scanned = summary["files_scanned"]
+    n_pinned = summary["pinned_checked"]
+    if not summary["observed"]:
+        print(
+            f"{glyph_warn()} {c_info('profiling_drift')} "
+            f"{c_meta('skipped (autolens_profiling/results not found)')}"
+        )
+    elif n_drift:
+        print(
+            f"{glyph_warn()} {c_info('profiling_drift')} "
+            f"{c_warn(f'{n_drift} drifted result(s)')} "
+            f"{c_meta(f'({n_scanned} scanned)')}"
+        )
+    else:
+        print(
+            f"{glyph_ok()} {c_info('profiling_drift')} "
+            f"{c_ok(f'{n_scanned} results clean')} "
+            f"{c_meta(f'({n_pinned} with pins)')}"
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    sys.exit(main(sys.argv))

heart/dashboard.py

@@ -56,6 +56,7 @@
     "repo_state",
     "worktree_drift",
     "script_timing",
+    "profiling_drift",
     "test_run",
     "version_skew",
 )
@@ -356,6 +357,31 @@ def build_board(
             ]
             sections.append(Section("script_timing", "Script timing", st, summary, details))
 
+    # Profiling pinned-value drift -------------------------------------------
+    if "profiling_drift" in unobserved:
+        sections.append(_unobs_section("profiling_drift", "Profiling drift"))
+    else:
+        drift = snapshot.get("profiling_drift") or {}
+        if drift:
+            n = _as_int(drift.get("drift_count"))
+            scanned = _as_int(drift.get("files_scanned"))
+            if not drift.get("observed"):
+                st, summary = WARN, "autolens_profiling/results not found"
+                details = []
+            elif n:
+                st, summary = WARN, f"{n} result(s) drifted from pinned baseline"
+                details = [
+                    f"✗ {f.get('path')}  "
+                    f"[{', '.join(str(d.get('label')) for d in (f.get('drift') or []))}]"
+                    for f in (drift.get("findings") or [])[:5]
+                ]
+            else:
+                st, summary = OK, f"{scanned} results clean"
+                details = []
+            sections.append(
+                Section("profiling_drift", "Profiling drift", st, summary, details)
+            )
+
     # Test run ---------------------------------------------------------------
     if "test_run" in unobserved:
         sections.append(_unobs_section("test_run", "Test run"))

heart/readiness.py

@@ -82,6 +82,7 @@
     "test_unknown": (10, 10),
     "timing_red": (15, 15),
     "timing_yellow": (8, 8),
+    "profiling_drift": (10, 30),
     "open_pr": (5, 15),
     "parked": (5, 15),
     "skew_behind": (8, 24),
@@ -424,6 +425,27 @@ def hit(key: str, n: int = 1) -> None:
         yellow.append(f"{_as_int(timing.get('yellow_count'))} slow script(s)")
         hit("timing_yellow")
 
+    # --- profiling pinned-value drift (YELLOW) ---
+    # autolens_profiling result JSONs record drift against pinned
+    # likelihood/evidence baselines instead of crashing (profiling records
+    # and flags; adjudication is autolens_workspace_test's remit). Any
+    # drifted result means the profiling baselines are non-comparable
+    # until resolved — library regression or stale pin, either way a
+    # human decision.
+    drift = snapshot.get("profiling_drift", {}) or {}
+    drift_findings = drift.get("findings") or []
+    if drift_findings:
+        for f in drift_findings[:5]:
+            yellow.append(
+                f"profiling drift: {f.get('path')} "
+                f"[{', '.join(str(d.get('label')) for d in (f.get('drift') or []))}]"
+            )
+            hit("profiling_drift")
+        if len(drift_findings) > 5:
+            yellow.append(
+                f"profiling drift: +{len(drift_findings) - 5} more drifted result(s)"
+            )
+
     # --- open PRs across all repos (YELLOW) ---
     for name, body in sorted(repos.items()):
         if not isinstance(body, dict):

heart/state.py

@@ -74,6 +74,7 @@ def aggregate() -> dict[str, Any]:
         "repos": repos,
         "worktree_drift": _read_json_or_default(HEART_STATE_DIR / "worktree_drift.json", {}),
         "script_timing": _read_json_or_default(HEART_STATE_DIR / "script_timing.json", {}),
+        "profiling_drift": _read_json_or_default(HEART_STATE_DIR / "profiling_drift.json", {}),
         "test_run": _read_json_or_default(HEART_STATE_DIR / "test_run.json", {}),
         "version_skew": _read_json_or_default(HEART_STATE_DIR / "version_skew.json", {}),
         "manifest_drift": _read_json_or_default(HEART_STATE_DIR / "manifest_drift.json", {}),

heart/tick.sh

@@ -27,6 +27,13 @@ else
   heart_log INFO "$(c_meta 'script_timing: skipped (no PyAutoBuild/test_results/latest)')"
 fi
 
+# Python: profiling pinned-value drift. Reads autolens_profiling result JSONs.
+if [ -d "$PYAUTO_ROOT/autolens_profiling/results" ] 2>/dev/null || [ -d "$HEART_HOME/../autolens_profiling/results" ]; then
+  PYTHONPATH="$HEART_HOME" python3 -m heart.checks.profiling_drift || heart_log WARN "$(c_warn 'profiling_drift failed')"
+else
+  heart_log INFO "$(c_meta 'profiling_drift: skipped (no autolens_profiling/results)')"
+fi
+
 # Python: workspace-validation verdict. Runs ALWAYS — it is server-first (reads
 # the cloud run conclusion via MCP-supplied file or `gh`), so it must run even
 # when there is no local report.json (the mobile case the old guard broke).

tests/test_profiling_drift.py

@@ -0,0 +1,153 @@
+"""tests/test_profiling_drift.py — pinned-drift scan + readiness leg."""
+
+from __future__ import annotations
+
+import importlib
+import json
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def drift_mod(tmp_path, monkeypatch):
+    """Redirect HEART_STATE_DIR to a tmp dir and reload the check module."""
+    monkeypatch.setenv("HEART_STATE_DIR", str(tmp_path))
+    import heart.state as state_mod
+
+    importlib.reload(state_mod)
+    import heart.checks.profiling_drift as pd
+
+    importlib.reload(pd)
+    pd.HEART_STATE_DIR = tmp_path
+    return tmp_path, pd
+
+
+def _write_result(
+    root: Path,
+    rel: str,
+    *,
+    pinned_expected=None,
+    pinned_drift=None,
+    instrument="hst",
+    version="2026.7.6.649",
+    include_fields=True,
+) -> Path:
+    p = root / rel
+    p.parent.mkdir(parents=True, exist_ok=True)
+    payload = {"instrument": instrument, "autolens_version": version}
+    if include_fields:
+        payload["pinned_expected"] = pinned_expected
+        payload["pinned_drift"] = pinned_drift if pinned_drift is not None else []
+    p.write_text(json.dumps(payload))
+    return p
+
+
+def test_clean_results_no_findings(drift_mod, tmp_path):
+    state_dir, pd = drift_mod
+    root = tmp_path / "results"
+    _write_result(root, "runtime/imaging/mge/local_cpu_fp64.json", pinned_expected=27379.4)
+    _write_result(root, "runtime/imaging/pixelization/local_cpu_fp64.json")  # no pin
+
+    summary = pd.run(root)
+    assert summary["observed"] is True
+    assert summary["files_scanned"] == 2
+    assert summary["pinned_checked"] == 1
+    assert summary["drift_count"] == 0
+    assert summary["findings"] == []
+    # Persisted for the snapshot.
+    assert json.loads((state_dir / "profiling_drift.json").read_text()) == summary
+
+
+def test_drifted_result_is_a_finding(drift_mod, tmp_path):
+    _, pd = drift_mod
+    root = tmp_path / "results"
+    _write_result(
+        root,
+        "runtime/imaging/mge/local_cpu_fp64.json",
+        pinned_expected=27379.4,
+        pinned_drift=[
+            {"label": "eager", "expected": 27379.4, "got": 7170.9, "rel_diff": 0.738, "rtol": 1e-4}
+        ],
+    )
+
+    summary = pd.run(root)
+    assert summary["drift_count"] == 1
+    (finding,) = summary["findings"]
+    assert finding["path"] == "runtime/imaging/mge/local_cpu_fp64.json"
+    assert finding["instrument"] == "hst"
+    assert finding["drift"][0]["label"] == "eager"
+
+
+def test_json_without_drift_fields_ignored(drift_mod, tmp_path):
+    """Old artifacts (comparison.json, probe JSONs) carry no pinned fields."""
+    _, pd = drift_mod
+    root = tmp_path / "results"
+    _write_result(root, "runtime/imaging/mge/comparison.json", include_fields=False)
+    (root / "runtime/imaging/mge/broken.json").write_text("{not json")
+
+    summary = pd.run(root)
+    assert summary["files_scanned"] == 0
+    assert summary["drift_count"] == 0
+
+
+def test_absent_repo_not_observed(drift_mod, tmp_path):
+    _, pd = drift_mod
+    summary = pd.run(tmp_path / "nonexistent")
+    assert summary["observed"] is False
+    assert summary["drift_count"] == 0
+
+
+def test_readiness_yellow_on_drift():
+    from heart import readiness
+
+    snap = {
+        "ts": "2026-07-08T12:00:00+00:00",
+        "repos": {},
+        "profiling_drift": {
+            "observed": True,
+            "files_scanned": 9,
+            "pinned_checked": 3,
+            "drift_count": 1,
+            "findings": [
+                {
+                    "path": "runtime/imaging/mge/local_cpu_fp64.json",
+                    "instrument": "hst",
+                    "drift": [{"label": "eager", "expected": 1.0, "got": 2.0}],
+                }
+            ],
+        },
+    }
+    v = readiness.compute(snap, libraries=[])
+    assert v["verdict"] in ("yellow", "red")
+    assert any("profiling drift" in r for r in v["yellow_reasons"])
+
+
+def test_readiness_caps_drift_reasons_at_five():
+    from heart import readiness
+
+    findings = [
+        {"path": f"runtime/imaging/mge/cfg{i}.json", "drift": [{"label": "eager"}]}
+        for i in range(8)
+    ]
+    snap = {
+        "ts": "2026-07-08T12:00:00+00:00",
+        "repos": {},
+        "profiling_drift": {"observed": True, "drift_count": 8, "findings": findings},
+    }
+    v = readiness.compute(snap, libraries=[])
+    drift_reasons = [r for r in v["yellow_reasons"] if "profiling drift" in r]
+    assert len(drift_reasons) == 6  # 5 findings + the "+3 more" line
+    assert any("+3 more" in r for r in drift_reasons)
+
+
+def test_readiness_clean_drift_adds_nothing():
+    from heart import readiness
+
+    snap = {
+        "ts": "2026-07-08T12:00:00+00:00",
+        "repos": {},
+        "profiling_drift": {"observed": True, "drift_count": 0, "findings": []},
+    }
+    v = readiness.compute(snap, libraries=[])
+    assert not any("profiling drift" in r for r in v["yellow_reasons"])