feat: import_time observation leg — off-tick import-cost regression tracking (#61)

.gitignore

@@ -8,3 +8,4 @@ __pycache__/
 build/
 dist/
 .coverage
+root.log

heart/checks/import_time.py

@@ -0,0 +1,214 @@
+"""heart/checks/import_time.py — measure PyAuto library import cost in a
+subprocess, track rolling per-package baselines, classify regressions.
+
+OFF-TICK by design. Importing the science stack costs several seconds
+(``import autolens`` ~3.6s warm, more cold), which does not fit the <30s
+watch-loop tick budget (``docs/internals.md`` rule 3). Run this on a slower
+cadence — a daily cron or on demand — and do **not** wire it into
+``heart/tick.sh``:
+
+    python -m heart.checks.import_time
+    HYGIENE_PYTHON=~/venv/PyAuto/bin/python python -m heart.checks.import_time
+
+The measurement runs in a SUBPROCESS (rule 4): Heart's own process never imports
+the science/JAX stack, and the tests inject a fake measurer so the stdlib-only
+suite never touches it either. Advisory only — surfaced on the board but never a
+release gate (import time is not release-blocking).
+
+Inputs:
+- spawns ``<python> -c "import <pkg>"`` per configured package (``HYGIENE_PYTHON``,
+  default ``python3`` — point it at the PyAuto venv to time the science libs).
+
+State (per Heart instance):
+- ~/.pyauto-heart/timings-import/<pkg>.json — a rolling window of recent durations.
+
+Output:
+- ~/.pyauto-heart/import_time.json — the latest regression summary.
+
+Classification (mirrors script_timing):
+- green: ratio <= yellow_factor (default 1.5)
+- yellow: yellow_factor < ratio <= red_factor (default 3.0)
+- red:    ratio > red_factor
+Where ratio = latest_duration / median(rolling_window).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import statistics
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Any, Callable
+
+import yaml
+
+HEART_STATE_DIR = Path(
+    os.environ.get("HEART_STATE_DIR")
+    or Path.home() / ".pyauto-heart"
+)
+HEART_IMPORT_TIMINGS_DIR = HEART_STATE_DIR / "timings-import"
+HEART_HOME = Path(__file__).resolve().parents[2]
+CONFIG_PATH = HEART_HOME / "config" / "repos.yaml"
+
+# autolens pulls the whole stack; the others give a per-package breakdown.
+DEFAULT_PACKAGES = ["autoconf", "autofit", "autoarray", "autogalaxy", "autolens"]
+
+# A measurer maps a package import-name to its import duration in seconds, or
+# None when it cannot be imported (missing / errored / timed out). Injected in
+# tests so the stdlib-only suite never imports the science stack.
+Measurer = Callable[[str], "float | None"]
+
+
+def load_thresholds() -> tuple[float, float, int]:
+    """Return (yellow_factor, red_factor, baseline_window) from config."""
+    if not CONFIG_PATH.is_file():
+        return 1.5, 3.0, 7
+    cfg = yaml.safe_load(CONFIG_PATH.read_text()) or {}
+    t = cfg.get("thresholds", {}).get("import_time", {})
+    return (
+        float(t.get("yellow_factor", 1.5)),
+        float(t.get("red_factor", 3.0)),
+        int(t.get("baseline_window", 7)),
+    )
+
+
+def default_measurer(python: str, timeout: float) -> Measurer:
+    """A measurer that times ``<python> -c "import <pkg>"`` in a subprocess.
+
+    Heart's own process never imports the package — it shells out — so the
+    daemon stays free of the science/JAX stack (internals rule 4).
+    """
+    env = {
+        **os.environ,
+        "NUMBA_CACHE_DIR": os.environ.get("NUMBA_CACHE_DIR", "/tmp/numba_cache"),
+        "MPLCONFIGDIR": os.environ.get("MPLCONFIGDIR", "/tmp/matplotlib"),
+    }
+
+    def measure(pkg: str) -> float | None:
+        start = time.perf_counter()
+        try:
+            proc = subprocess.run(
+                [python, "-c", f"import {pkg}"],
+                capture_output=True, timeout=timeout, env=env,
+            )
+        except (subprocess.TimeoutExpired, FileNotFoundError):
+            return None
+        if proc.returncode != 0:
+            return None
+        return time.perf_counter() - start
+
+    return measure
+
+
+def update_history(pkg: str, duration: float, window: int) -> list[float]:
+    """Append duration to pkg's rolling history, return the new list."""
+    HEART_IMPORT_TIMINGS_DIR.mkdir(parents=True, exist_ok=True)
+    history_path = HEART_IMPORT_TIMINGS_DIR / f"{pkg}.json"
+    if history_path.is_file():
+        try:
+            history = json.loads(history_path.read_text())
+        except json.JSONDecodeError:
+            history = []
+    else:
+        history = []
+    history.append(duration)
+    history = history[-window:]
+    history_path.write_text(json.dumps(history))
+    return history
+
+
+def classify(ratio: float, yellow: float, red: float) -> str:
+    if ratio > red:
+        return "red"
+    if ratio > yellow:
+        return "yellow"
+    return "green"
+
+
+def run(
+    measurer: Measurer | None = None,
+    packages: list[str] | None = None,
+) -> dict[str, Any]:
+    """Measure each package's import cost, update rolling baselines, classify."""
+    yellow_factor, red_factor, window = load_thresholds()
+    packages = packages if packages is not None else DEFAULT_PACKAGES
+    python = os.environ.get("HYGIENE_PYTHON", "python3")
+    timeout = float(os.environ.get("HEART_IMPORT_TIMEOUT", "90"))
+    measurer = measurer or default_measurer(python, timeout)
+
+    findings: dict[str, list[dict[str, Any]]] = {"red": [], "yellow": [], "green": []}
+    unavailable: list[str] = []
+    measured = 0
+    new_packages = 0
+
+    for pkg in packages:
+        duration = measurer(pkg)
+        if duration is None:
+            unavailable.append(pkg)
+            continue
+        measured += 1
+        history = update_history(pkg, duration, window)
+        if len(history) <= 1:
+            new_packages += 1
+            continue
+        prior = history[:-1]
+        baseline = statistics.median(prior)
+        if baseline <= 0:
+            continue
+        ratio = duration / baseline
+        category = classify(ratio, yellow_factor, red_factor)
+        findings[category].append({
+            "package": pkg,
+            "latest_seconds": round(duration, 3),
+            "baseline_seconds": round(baseline, 3),
+            "ratio": round(ratio, 2),
+            "samples": len(prior),
+        })
+
+    summary = {
+        "python": python,
+        "packages_measured": measured,
+        "packages_unavailable": unavailable,
+        "new_packages_no_baseline": new_packages,
+        "red_count": len(findings["red"]),
+        "yellow_count": len(findings["yellow"]),
+        "green_count": len(findings["green"]),
+        "red": sorted(findings["red"], key=lambda x: -x["ratio"]),
+        "yellow": sorted(findings["yellow"], key=lambda x: -x["ratio"]),
+    }
+
+    sys.path.insert(0, str(HEART_HOME))
+    from heart import state
+
+    state.atomic_write_json(HEART_STATE_DIR / "import_time.json", summary)
+    return summary
+
+
+def main(argv: list[str]) -> int:
+    summary = run()
+
+    from heart_color import c_ok, c_warn, c_fail, c_info, c_meta, glyph_ok, glyph_warn, glyph_fail
+
+    if summary["red_count"]:
+        glyph = glyph_fail()
+        label = c_fail(f"{summary['red_count']} red") + " " + c_warn(f"{summary['yellow_count']} yellow")
+    elif summary["yellow_count"]:
+        glyph = glyph_warn()
+        label = c_warn(f"{summary['yellow_count']} slower than baseline")
+    elif not summary["packages_measured"]:
+        glyph = glyph_warn()
+        label = c_warn("no libraries importable (set HYGIENE_PYTHON to the PyAuto venv)")
+    else:
+        glyph = glyph_ok()
+        label = c_ok(f"{summary['green_count']} imports within baseline")
+    extra = c_meta(f" ({summary['new_packages_no_baseline']} new, no baseline)")
+    print(f"{glyph} {c_info('import_time')} {label}{extra}")
+    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",
+    "import_time",
     "profiling_drift",
     "test_run",
     "version_skew",
@@ -373,6 +374,30 @@ def build_board(
             ]
             sections.append(Section("script_timing", "Script timing", st, summary, details))
 
+    # Import timing (advisory; off-tick daily) --------------------------------
+    if "import_time" in unobserved:
+        sections.append(_unobs_section("import_time", "Import timing"))
+    else:
+        imp = snapshot.get("import_time") or {}
+        if imp:
+            r = _as_int(imp.get("red_count"))
+            y = _as_int(imp.get("yellow_count"))
+            g = _as_int(imp.get("green_count"))
+            if r:
+                st, summary = FAIL, f"{r} import regressions (>3× baseline), {y} slow (>1.5×)"
+            elif y:
+                st, summary = WARN, f"{y} imports >1.5× baseline, {g} within"
+            elif not _as_int(imp.get("packages_measured")):
+                st, summary = WARN, "no libraries importable (set HYGIENE_PYTHON)"
+            else:
+                st, summary = OK, f"{g} imports within baseline"
+            details = [
+                f"✗ {e['package']}  "
+                f"{e['latest_seconds']:.2f}s vs {e['baseline_seconds']:.2f}s ({e['ratio']}×)"
+                for e in (imp.get("red") or [])[:5]
+            ]
+            sections.append(Section("import_time", "Import timing", st, summary, details))
+
     # Profiling pinned-value drift -------------------------------------------
     if "profiling_drift" in unobserved:
         sections.append(_unobs_section("profiling_drift", "Profiling drift"))

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", {}),
+        "import_time": _read_json_or_default(HEART_STATE_DIR / "import_time.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", {}),

tests/test_import_time.py

@@ -0,0 +1,88 @@
+"""tests/test_import_time.py — import-cost regression classifier.
+
+Stdlib-only (internals rule 4): the measurer is injected, so the suite never
+imports the science/JAX stack — it exercises the rolling-baseline + classifier
+logic with fixed durations.
+"""
+
+from __future__ import annotations
+
+import importlib
+import json
+
+import pytest
+
+
+@pytest.fixture
+def tmp_state(tmp_path, monkeypatch):
+    """Redirect HEART_STATE_DIR to a tmp dir and reload heart.checks.import_time."""
+    monkeypatch.setenv("HEART_STATE_DIR", str(tmp_path))
+    import heart.state as state_mod
+    importlib.reload(state_mod)
+    import heart.checks.import_time as it
+    importlib.reload(it)
+    it.HEART_STATE_DIR = tmp_path
+    it.HEART_IMPORT_TIMINGS_DIR = tmp_path / "timings-import"
+    it.HEART_IMPORT_TIMINGS_DIR.mkdir(parents=True, exist_ok=True)
+    return tmp_path, it
+
+
+def _fixed(value):
+    """A measurer that always reports `value` seconds (or None)."""
+    return lambda pkg: value
+
+
+def test_first_observation_has_no_baseline(tmp_state):
+    _, it = tmp_state
+    summary = it.run(measurer=_fixed(1.0), packages=["autolens"])
+    assert summary["new_packages_no_baseline"] == 1
+    assert summary["red_count"] == 0 and summary["yellow_count"] == 0
+    assert summary["packages_measured"] == 1
+
+
+def test_within_baseline_classified_green(tmp_state):
+    _, it = tmp_state
+    it.run(measurer=_fixed(1.0), packages=["autolens"])
+    summary = it.run(measurer=_fixed(1.0), packages=["autolens"])
+    assert summary["green_count"] == 1
+    assert summary["red_count"] == 0 and summary["yellow_count"] == 0
+
+
+def test_above_yellow_factor_classified_yellow(tmp_state):
+    _, it = tmp_state
+    it.run(measurer=_fixed(1.0), packages=["autolens"])
+    summary = it.run(measurer=_fixed(2.0), packages=["autolens"])  # ratio 2.0 > 1.5
+    assert summary["yellow_count"] == 1 and summary["red_count"] == 0
+
+
+def test_above_red_factor_classified_red(tmp_state):
+    _, it = tmp_state
+    it.run(measurer=_fixed(1.0), packages=["autolens"])
+    summary = it.run(measurer=_fixed(4.0), packages=["autolens"])  # ratio 4.0 > 3.0
+    assert summary["red_count"] == 1 and summary["yellow_count"] == 0
+
+
+def test_unavailable_package_excluded(tmp_state):
+    _, it = tmp_state
+    summary = it.run(measurer=_fixed(None), packages=["autolens"])
+    assert summary["packages_measured"] == 0
+    assert summary["packages_unavailable"] == ["autolens"]
+    # No history file is written for an unmeasurable package.
+    assert not (it.HEART_IMPORT_TIMINGS_DIR / "autolens.json").is_file()
+
+
+def test_rolling_window_caps_history_length(tmp_state):
+    _, it = tmp_state
+    for d in [1.0, 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7, 1.8, 1.9]:
+        it.run(measurer=_fixed(d), packages=["autolens"])
+    history_files = list(it.HEART_IMPORT_TIMINGS_DIR.glob("*.json"))
+    assert len(history_files) == 1
+    history = json.loads(history_files[0].read_text())
+    assert len(history) == 7  # default window
+    assert history[-1] == 1.9
+
+
+def test_summary_sidecar_written(tmp_state):
+    tmp_path, it = tmp_state
+    it.run(measurer=_fixed(1.0), packages=["autolens"])
+    assert (tmp_path / "import_time.json").is_file()