diff --git a/heart/checks/script_timing.py b/heart/checks/script_timing.py index cc35682..a01c9d9 100644 --- a/heart/checks/script_timing.py +++ b/heart/checks/script_timing.py @@ -8,7 +8,8 @@ State (per Heart instance): - ~/.pyauto-heart/timings/____.json - containing a rolling window of recent durations. + containing a rolling window of recent observations, each an entry + ``{"duration_s": float, "run_id": str, "ts": str}``. Output: - ~/.pyauto-heart/script_timing.json with the latest regression summary. @@ -19,10 +20,45 @@ - red: ratio > red_factor Where ratio = latest_duration / median(rolling_window). + +Why the entries carry provenance +-------------------------------- +The tick runs every few minutes but ``run_logs/latest`` only changes when a +new ``run_all`` finishes. The original implementation appended the durations +it read on *every* tick, so a single run was copied into the window until all +seven slots held the same number: ``median(prior)`` was then that one +observation wearing the clothes of a seven-run median — stable by +construction, and one unlucky run read as a regression. + +The cure is run identity, not seeding rules. Each entry records the ``run_id`` +it came from (the real name of the timestamped run dir behind the ``latest`` +symlink); re-ticking on a run already at the head of the window *replaces* +that entry instead of appending, so ticks are idempotent and a window of +seven means seven distinct runs. Histories written before provenance existed +are read as entries with an empty ``run_id``; one whose values are all +identical is provably that same single-observation artefact, so it collapses +to one entry the first time it is touched. + +Classification then waits for a real baseline: fewer than +``MIN_BASELINE_RUNS`` distinct runs in the prior window counts the script as +*building* rather than green/yellow/red, so a thin baseline never masquerades +as a verdict. + +Why migration is attempted +-------------------------- +The slug is path-derived (see :func:`slug_for`), so moving a script strands +its history under a filename nothing writes to again — and the check then has +nothing to compare against and silently never fires. Every scan therefore +notices history files it did not touch: where such an orphan unambiguously +corresponds to a script with no history (same workspace, same leaf script +name), its history is renamed onto the new slug so the baseline survives the +move. Ambiguous or unmatched orphans are never deleted or guessed at — they +are reported in the summary, loudly, which is what the silent case lacked. """ from __future__ import annotations +import datetime import json import os import statistics @@ -42,6 +78,13 @@ PYAUTO_ROOT = Path(__file__).resolve().parents[3] if Path(__file__).resolve().parents[3].name == "PyAutoLabs" else Path.home() / "Code" / "PyAutoLabs" TEST_RESULTS_LATEST = PYAUTO_ROOT / "PyAutoHands" / "run_logs" / "latest" +# Distinct runs the prior window must hold before a ratio is a verdict rather +# than a coin flip. Below this the script is counted as "building" a baseline. +MIN_BASELINE_RUNS = 3 + +# Cap on the migrated/orphaned lists carried in the summary JSON. +MAX_LISTED = 20 + def load_thresholds() -> tuple[float, float, int]: """Return (yellow_factor, red_factor, baseline_window) from config.""" @@ -62,6 +105,10 @@ def slug_for(workspace: str, directory: str, file_path: str) -> str: Uses the FULL relative file path so scripts in nested subdirs (e.g. ``imaging/modeling.py`` vs ``imaging/features/.../modeling.py``) do not collide on a shared leaf name. + + Moving a script therefore changes its slug; :func:`migrate_orphans` (run + from :func:`run`) reattaches the stranded history where the match is + unambiguous. """ # The autohands run_all writes ``file`` as an absolute path. Strip # everything up to and including "scripts/" so the slug is workspace- @@ -77,25 +124,168 @@ def slug_for(workspace: str, directory: str, file_path: str) -> str: return f"{workspace}__{relative}.json" -def update_history(slug: str, duration: float, window: int) -> list[float]: - """Append duration to slug's rolling history, return the new list.""" - HEART_TIMINGS_DIR.mkdir(parents=True, exist_ok=True) +def _now_iso() -> str: + return datetime.datetime.now(datetime.timezone.utc).isoformat() + + +def _state_module(): + """Import ``heart.state`` lazily (this module is also run standalone).""" + if str(HEART_HOME) not in sys.path: + sys.path.insert(0, str(HEART_HOME)) + from heart import state + + return state + + +def normalize_history(raw: Any) -> list[dict[str, Any]]: + """Coerce a stored history to the entry shape, tolerating legacy floats. + + Legacy histories are bare ``[float, ...]`` lists with no provenance; they + become entries with an empty ``run_id``. A legacy history whose values are + ALL identical is the single-observation artefact this check used to + manufacture (see the module docstring) — it collapses to ONE entry, since + that is the only real observation it ever held. + """ + if not isinstance(raw, list): + return [] + + entries: list[dict[str, Any]] = [] + all_legacy = True + for item in raw: + if isinstance(item, dict): + duration = item.get("duration_s", item.get("duration")) + if duration is None: + continue + try: + duration = float(duration) + except (TypeError, ValueError): + continue + all_legacy = False + entries.append({ + "duration_s": duration, + "run_id": str(item.get("run_id") or ""), + "ts": str(item.get("ts") or ""), + }) + elif isinstance(item, (int, float)) and not isinstance(item, bool): + entries.append({"duration_s": float(item), "run_id": "", "ts": ""}) + # Anything else is unreadable noise; drop it. + + if all_legacy and len(entries) > 1: + durations = {e["duration_s"] for e in entries} + if len(durations) == 1: + entries = entries[:1] + return entries + + +def read_history(slug: str) -> list[dict[str, Any]]: + """Return the normalized history entries stored for ``slug`` (may be []).""" history_path = HEART_TIMINGS_DIR / slug - if history_path.is_file(): - try: - history = json.loads(history_path.read_text()) - except json.JSONDecodeError: - history = [] + if not history_path.is_file(): + return [] + try: + raw = json.loads(history_path.read_text()) + except (json.JSONDecodeError, OSError): + return [] + return normalize_history(raw) + + +def _write_history(slug: str, entries: list[dict[str, Any]]) -> None: + """Atomically persist ``entries`` for ``slug`` (concurrent ticks are real).""" + HEART_TIMINGS_DIR.mkdir(parents=True, exist_ok=True) + _state_module().atomic_write_json(HEART_TIMINGS_DIR / slug, entries) + + +def distinct_run_count(entries: list[dict[str, Any]]) -> int: + """Number of distinct runs ``entries`` represents. + + Entries with an empty ``run_id`` predate provenance; each counts as its + own run, because a legacy history that survived the identical-value + collapse is genuine accumulation across ticks we cannot name. + """ + named: set[str] = set() + unnamed = 0 + for entry in entries: + run_id = entry.get("run_id") or "" + if run_id: + named.add(run_id) + else: + unnamed += 1 + return len(named) + unnamed + + +def update_history( + slug: str, + duration: float, + window: int, + run_id: str = "", + ts: str | None = None, +) -> list[dict[str, Any]]: + """Record ``duration`` for ``slug``, returning the new history entries. + + An observation from the run already at the head of the window REPLACES + that entry rather than appending: the tick fires far more often than + ``run_all`` does, and re-reading one run must not fill the window with + copies of it. Distinct runs append as usual, trimmed to ``window``. + """ + history = read_history(slug) + entry = { + "duration_s": float(duration), + "run_id": str(run_id or ""), + "ts": ts or _now_iso(), + } + + if history and entry["run_id"] and history[-1].get("run_id") == entry["run_id"]: + history[-1] = entry else: - history = [] + history.append(entry) - history.append(duration) # Keep at most `window` most recent. history = history[-window:] - history_path.write_text(json.dumps(history)) + _write_history(slug, history) return history +def run_id_for(results_dir: Path) -> str: + """Identity of the run behind ``results_dir``. + + ``run_logs/latest`` is a symlink onto a timestamped run dir; the real + directory name is the stable identity of the run that produced these + durations, so resolve through the link. Falls back to the raw name when + the path cannot be resolved. + """ + try: + resolved = results_dir.resolve() + name = resolved.name + except OSError: + name = "" + return name or results_dir.name + + +def _slug_identity(slug: str) -> tuple[str, str]: + """Return (workspace, final script token) for a history filename.""" + stem = slug[:-len(".json")] if slug.endswith(".json") else slug + parts = stem.split("__") + workspace = parts[0] if parts else "" + final = parts[-1] if parts else "" + return workspace, final + + +def migration_candidates(slug: str, orphans: list[str]) -> list[str]: + """Orphan history files that could belong to ``slug`` after a move. + + A move keeps the workspace and the script's own name and changes only the + directories between them, so those two tokens are the match. Anything less + specific would silently glue one script's baseline onto another. + """ + workspace, final = _slug_identity(slug) + if not workspace or not final: + return [] + return [ + o for o in orphans + if _slug_identity(o) == (workspace, final) + ] + + def classify(ratio: float, yellow: float, red: float) -> str: if ratio > red: return "red" @@ -136,26 +326,57 @@ def scan_latest_results(results_dir: Path) -> list[dict[str, Any]]: return entries +def _existing_history_files() -> list[str]: + if not HEART_TIMINGS_DIR.is_dir(): + return [] + return sorted(p.name for p in HEART_TIMINGS_DIR.glob("*.json") if p.is_file()) + + def run(results_dir: Path | None = None) -> dict[str, Any]: """Update rolling timings from results_dir; return classification summary.""" results_dir = results_dir or TEST_RESULTS_LATEST yellow_factor, red_factor, window = load_thresholds() + run_id = run_id_for(results_dir) + ts = _now_iso() findings: dict[str, list[dict[str, Any]]] = {"red": [], "yellow": [], "green": []} total = 0 new_scripts = 0 - - for entry in scan_latest_results(results_dir): - slug = slug_for(entry["project"], entry["directory"], entry["file"]) - history = update_history(slug, entry["duration"], window) + building = 0 + migrations: list[dict[str, str]] = [] + + scanned = scan_latest_results(results_dir) + slugs = [slug_for(e["project"], e["directory"], e["file"]) for e in scanned] + touched = set(slugs) + # Anything this scan does not write to is a candidate stranded baseline. + orphans = [name for name in _existing_history_files() if name not in touched] + + for entry, slug in zip(scanned, slugs): + if not read_history(slug): + # No baseline here: a move may have stranded it under another name. + candidates = migration_candidates(slug, orphans) + if len(candidates) == 1: + source = candidates[0] + (HEART_TIMINGS_DIR / source).replace(HEART_TIMINGS_DIR / slug) + orphans.remove(source) + migrations.append({"from": source, "to": slug}) + # Zero or several candidates: never guess — they stay orphaned and + # get reported below. + + history = update_history(slug, entry["duration"], window, run_id=run_id, ts=ts) total += 1 - if len(history) <= 1: + + prior = history[:-1] + if not prior: # First observation, no baseline yet. new_scripts += 1 continue + if distinct_run_count(prior) < MIN_BASELINE_RUNS: + # A baseline too thin to be a verdict. + building += 1 + continue # Compare latest to median of the prior history (exclude current). - prior = history[:-1] - baseline = statistics.median(prior) + baseline = statistics.median(p["duration_s"] for p in prior) if baseline <= 0: continue ratio = entry["duration"] / baseline @@ -172,8 +393,14 @@ def run(results_dir: Path | None = None) -> dict[str, Any]: summary = { "results_dir": str(results_dir), + "run_id": run_id, "total_scripts": total, "new_scripts_no_baseline": new_scripts, + "building_count": building, + "migrated_count": len(migrations), + "orphaned_count": len(orphans), + "migrated": migrations[:MAX_LISTED], + "orphaned": orphans[:MAX_LISTED], "red_count": len(findings["red"]), "yellow_count": len(findings["yellow"]), "green_count": len(findings["green"]), @@ -181,10 +408,7 @@ def run(results_dir: Path | None = None) -> dict[str, Any]: "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 / "script_timing.json", summary) + _state_module().atomic_write_json(HEART_STATE_DIR / "script_timing.json", summary) return summary @@ -204,7 +428,15 @@ def main(argv: list[str]) -> int: else: glyph = glyph_ok() label = c_ok(f"{summary['green_count']} scripts within baseline") - extra = c_meta(f" ({summary['new_scripts_no_baseline']} new, no baseline)") + extra = c_meta( + f" ({summary['new_scripts_no_baseline']} new," + f" {summary['building_count']} building)" + ) + if summary["migrated_count"]: + extra += c_meta(f" · {summary['migrated_count']} migrated") + if summary["orphaned_count"]: + # Loud: a stranded baseline is a check that silently never fires. + extra += " " + c_warn(f"· {summary['orphaned_count']} orphaned baseline(s)") print(f"{glyph} {c_info('script_timing')} {label}{extra}") return 0 diff --git a/tests/test_script_timing.py b/tests/test_script_timing.py index a1a041f..f080f62 100644 --- a/tests/test_script_timing.py +++ b/tests/test_script_timing.py @@ -1,4 +1,14 @@ -"""tests/test_script_timing.py — regression classifier thresholds + idempotence.""" +"""tests/test_script_timing.py — regression classifier thresholds, run-identity +dedup, baseline provenance and orphan migration. + +The dedup tests are the load-bearing ones: the tick re-reads the same +``run_logs/latest`` on every cycle, and the check used to append those same +durations until the window held seven copies of one observation. "Same run +re-ticked" must therefore leave the window's length alone. + +Repo/workspace names here are deliberately fictional (``workspace_a``) — this +suite is organ code under the tenant firewall. +""" from __future__ import annotations @@ -25,8 +35,19 @@ def tmp_state(tmp_path, monkeypatch): return tmp_path, st -def _make_results_dir(root: Path, project: str, directory: str, results: list[dict]) -> Path: - rdir = root / "results" +def _make_results_dir( + root: Path, + project: str, + directory: str, + results: list[dict], + run: str = "run_1", +) -> Path: + """Write a run_all-shaped results dir for one run. + + ``run`` names the directory, which is the run identity the check reads — + two calls with the same ``run`` are the same run observed twice. + """ + rdir = root / "run_logs" / run rdir.mkdir(parents=True, exist_ok=True) safe_dir = directory.replace("/", "__") fname = f"{project}__scripts__{safe_dir}__script.json" @@ -38,12 +59,23 @@ def _make_results_dir(root: Path, project: str, directory: str, results: list[di return rdir +def _observe(tmp_path, st, duration: float, run: str, file: str = "imaging/simulator.py"): + """Run one observation of ``file`` at ``duration`` under run id ``run``.""" + rdir = _make_results_dir(tmp_path, "workspace_a", "imaging", [ + {"file": file, "status": "passed", "duration_seconds": duration}, + ], run=run) + return st.run(rdir) + + +def _only_history(st) -> list[dict]: + files = list(st.HEART_TIMINGS_DIR.glob("*.json")) + assert len(files) == 1, [f.name for f in files] + return json.loads(files[0].read_text()) + + def test_first_observation_has_no_baseline(tmp_state): tmp_path, st = tmp_state - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 50.0}, - ]) - summary = st.run(rdir) + summary = _observe(tmp_path, st, 50.0, "run_1") assert summary["new_scripts_no_baseline"] == 1 assert summary["red_count"] == 0 assert summary["yellow_count"] == 0 @@ -51,49 +83,38 @@ def test_first_observation_has_no_baseline(tmp_state): def test_within_baseline_classified_green(tmp_state): tmp_path, st = tmp_state - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 50.0}, - ]) - # Run twice to populate history; second call has 1-sample baseline = 50, - # ratio = 50/50 = 1.0 → green. - st.run(rdir) - summary = st.run(rdir) + # Three distinct prior runs are the floor for a verdict; the fourth is + # compared against their median (50) → ratio 1.0 → green. + for run in ("run_1", "run_2", "run_3"): + _observe(tmp_path, st, 50.0, run) + summary = _observe(tmp_path, st, 50.0, "run_4") assert summary["red_count"] == 0 assert summary["yellow_count"] == 0 assert summary["green_count"] == 1 + assert summary["building_count"] == 0 def test_above_yellow_factor_classified_yellow(tmp_state): tmp_path, st = tmp_state - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 50.0}, - ]) - st.run(rdir) - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 100.0}, - ]) - summary = st.run(rdir) + for run in ("run_1", "run_2", "run_3"): + _observe(tmp_path, st, 50.0, run) + summary = _observe(tmp_path, st, 100.0, "run_4") assert summary["yellow_count"] == 1 assert summary["red_count"] == 0 def test_above_red_factor_classified_red(tmp_state): tmp_path, st = tmp_state - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 50.0}, - ]) - st.run(rdir) - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 200.0}, - ]) - summary = st.run(rdir) + for run in ("run_1", "run_2", "run_3"): + _observe(tmp_path, st, 50.0, run) + summary = _observe(tmp_path, st, 200.0, "run_4") assert summary["red_count"] == 1 assert summary["yellow_count"] == 0 def test_failed_scripts_excluded_from_baseline(tmp_state): tmp_path, st = tmp_state - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ + rdir = _make_results_dir(tmp_path, "workspace_a", "imaging", [ {"file": "imaging/simulator.py", "status": "failed", "duration_seconds": 50.0}, ]) summary = st.run(rdir) @@ -103,14 +124,185 @@ def test_failed_scripts_excluded_from_baseline(tmp_state): def test_rolling_window_caps_history_length(tmp_state): tmp_path, st = tmp_state - for d in [10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, 17.0, 18.0, 19.0]: - rdir = _make_results_dir(tmp_path, "autolens", "imaging", [ - {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": d}, - ]) - st.run(rdir) + for i, d in enumerate([10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 16.0, 17.0, 18.0, 19.0]): + _observe(tmp_path, st, d, f"run_{i}") # Default window is 7 → history should be trimmed. - history_files = list(st.HEART_TIMINGS_DIR.glob("*.json")) - assert len(history_files) == 1 - history = json.loads(history_files[0].read_text()) + history = _only_history(st) assert len(history) == 7 - assert history[-1] == 19.0 # most recent + assert history[-1]["duration_s"] == 19.0 # most recent + # Ten distinct runs in, seven distinct runs stored — no repetition. + assert len({e["run_id"] for e in history}) == 7 + + +def test_same_run_reticked_does_not_grow_window(tmp_state): + """The defect: re-reading run_logs/latest each tick filled the window.""" + tmp_path, st = tmp_state + _observe(tmp_path, st, 10.0, "run_1") + _observe(tmp_path, st, 20.0, "run_2") + assert len(_only_history(st)) == 2 + + # Same run observed four more times — the window must not grow. + for _ in range(4): + _observe(tmp_path, st, 20.0, "run_2") + history = _only_history(st) + assert len(history) == 2 + assert [e["run_id"] for e in history] == ["run_1", "run_2"] + + +def test_same_run_retick_replaces_the_newest_entry(tmp_state): + tmp_path, st = tmp_state + _observe(tmp_path, st, 10.0, "run_1") + _observe(tmp_path, st, 20.0, "run_2") + # A corrected duration for the same run overwrites, it does not append. + _observe(tmp_path, st, 25.0, "run_2") + history = _only_history(st) + assert [e["duration_s"] for e in history] == [10.0, 25.0] + + +def test_legacy_identical_window_collapses_to_one_entry(tmp_state): + """Seven copies of one number were never seven runs — collapse them.""" + tmp_path, st = tmp_state + slug = st.slug_for("workspace_a", "scripts/imaging", "imaging/simulator.py") + (st.HEART_TIMINGS_DIR / slug).write_text(json.dumps([45.99] * 7)) + + summary = _observe(tmp_path, st, 46.0, "run_1") + history = _only_history(st) + assert [e["duration_s"] for e in history] == [45.99, 46.0] + assert history[0]["run_id"] == "" # legacy entry, provenance unknown + # One prior sample is below the floor — building, not a verdict. + assert summary["building_count"] == 1 + assert summary["green_count"] == 0 + + +def test_legacy_mixed_float_history_still_classifies(tmp_state): + """Real accumulation predating provenance keeps its baseline.""" + tmp_path, st = tmp_state + slug = st.slug_for("workspace_a", "scripts/imaging", "imaging/simulator.py") + (st.HEART_TIMINGS_DIR / slug).write_text(json.dumps([10.0, 12.0, 11.0])) + + summary = _observe(tmp_path, st, 11.0, "run_1") + assert summary["green_count"] == 1 + assert summary["building_count"] == 0 + assert summary["green_count"] + summary["yellow_count"] + summary["red_count"] == 1 + history = _only_history(st) + assert len(history) == 4 + + +def test_baseline_floor_needs_three_distinct_runs(tmp_state): + tmp_path, st = tmp_state + first = _observe(tmp_path, st, 50.0, "run_1") + assert first["new_scripts_no_baseline"] == 1 + assert first["building_count"] == 0 + + second = _observe(tmp_path, st, 50.0, "run_2") + assert second["building_count"] == 1 + assert second["green_count"] == 0 + + third = _observe(tmp_path, st, 50.0, "run_3") + assert third["building_count"] == 1 + assert third["green_count"] == 0 + + fourth = _observe(tmp_path, st, 50.0, "run_4") + assert fourth["building_count"] == 0 + assert fourth["green_count"] == 1 + + +def test_unambiguous_orphan_is_migrated_and_history_continues(tmp_state): + """A moved script keeps its baseline when exactly one orphan matches.""" + tmp_path, st = tmp_state + old_slug = "workspace_a__scripts__jax_grad__lp.json" + (st.HEART_TIMINGS_DIR / old_slug).write_text(json.dumps([ + {"duration_s": 40.0, "run_id": "old_1", "ts": ""}, + {"duration_s": 41.0, "run_id": "old_2", "ts": ""}, + {"duration_s": 39.0, "run_id": "old_3", "ts": ""}, + ])) + + summary = _observe(tmp_path, st, 40.0, "run_1", file="scripts/imaging/jax_grad/lp.py") + new_slug = "workspace_a__scripts__imaging__jax_grad__lp.json" + + assert not (st.HEART_TIMINGS_DIR / old_slug).exists() + assert (st.HEART_TIMINGS_DIR / new_slug).is_file() + assert summary["migrated_count"] == 1 + assert summary["migrated"] == [{"from": old_slug, "to": new_slug}] + assert summary["orphaned_count"] == 0 + # The baseline survived the move, so this run is classified, not "new". + assert summary["green_count"] == 1 + assert summary["new_scripts_no_baseline"] == 0 + history = json.loads((st.HEART_TIMINGS_DIR / new_slug).read_text()) + assert [e["duration_s"] for e in history] == [40.0, 41.0, 39.0, 40.0] + + +def test_ambiguous_orphans_are_not_migrated_and_are_reported(tmp_state): + """Two candidates means we do not know — report, never guess.""" + tmp_path, st = tmp_state + orphans = [ + "workspace_a__scripts__jax_grad__lp.json", + "workspace_a__scripts__point_source__jax_grad__lp.json", + ] + for name in orphans: + (st.HEART_TIMINGS_DIR / name).write_text(json.dumps([40.0, 41.0, 39.0])) + + summary = _observe(tmp_path, st, 40.0, "run_1", file="scripts/imaging/jax_grad/lp.py") + + for name in orphans: + assert (st.HEART_TIMINGS_DIR / name).is_file() + assert summary["migrated_count"] == 0 + assert summary["orphaned_count"] == 2 + assert sorted(summary["orphaned"]) == sorted(orphans) + # No baseline was adopted, so this is a first observation. + assert summary["new_scripts_no_baseline"] == 1 + + +def test_history_entries_carry_run_provenance(tmp_state): + """Entry shape on disk: duration_s + run_id + ts, run_id resolved through + the `latest` symlink to the real run dir.""" + tmp_path, st = tmp_state + rdir = _make_results_dir(tmp_path, "workspace_a", "imaging", [ + {"file": "imaging/simulator.py", "status": "passed", "duration_seconds": 50.0}, + ], run="2026-08-24_120000") + latest = tmp_path / "run_logs" / "latest" + latest.symlink_to(rdir, target_is_directory=True) + + summary = st.run(latest) + history = _only_history(st) + assert len(history) == 1 + entry = history[0] + assert set(entry) == {"duration_s", "run_id", "ts"} + assert isinstance(entry["duration_s"], float) + assert entry["run_id"] == "2026-08-24_120000" + assert entry["ts"] + assert summary["run_id"] == "2026-08-24_120000" + + +def test_history_is_written_atomically(tmp_state): + """House rule: state writes go through heart.state.atomic_write_json.""" + tmp_path, st = tmp_state + import heart.state as state_mod + + written: list[str] = [] + original = state_mod.atomic_write_json + + def spy(path, payload): + written.append(Path(path).name) + original(path, payload) + + state_mod.atomic_write_json = spy + try: + _observe(tmp_path, st, 50.0, "run_1") + finally: + state_mod.atomic_write_json = original + + slug = st.slug_for("workspace_a", "scripts/imaging", "imaging/simulator.py") + assert slug in written + assert "script_timing.json" in written + # No half-written temp files left behind. + assert not list(st.HEART_TIMINGS_DIR.glob("*.tmp")) + + +def test_summary_carries_the_new_counters(tmp_state): + tmp_path, st = tmp_state + summary = _observe(tmp_path, st, 50.0, "run_1") + for key in ("building_count", "migrated_count", "orphaned_count", + "migrated", "orphaned", "new_scripts_no_baseline", + "red_count", "yellow_count", "green_count", "total_scripts"): + assert key in summary