From 6c9fad229731c17df7eee80f87091c331b8986e5 Mon Sep 17 00:00:00 2001 From: Gaurang Date: Sun, 20 Sep 2026 16:36:52 +0530 Subject: [PATCH 1/6] policy: rank a lever from what it measured, where there is a record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The loop has been able to read its own past since #118 and nothing consulted it. `select_interventions` scored every candidate as `coverage x expected_delta_mean` — a hand-authored constant, identical on every run no matter what the last one measured — so a lever that lost 9% last night ranked exactly where it did before anyone ran it. Substitution, not a blend. `expected_delta_mean` is an estimate of the lever's effect; a delta the lever actually recorded on this GPU is a better estimate of that same quantity, so it replaces it and coverage — a property of this trace — is untouched. There is no weighting constant between prior and measurement, because a number like `K = 2` reads as calibrated and is not; `playbook/match.py` refuses to pick such a threshold and says so, and this follows it. Ranking stays a precedence tuple for the reason that module gives: terms answering different questions should not collapse into a scalar where one can quietly outvote another. Four terms now — rejected -> not worth a run -> demoted -> magnitude, then name "Not worth a run" sits above the demotion, and finding that out is what the end-to-end check was for. With the demotion higher, a lever measured at -9% in every run outranked a conflicted lever estimated at +1.4%, because the loser's record did not disagree with itself. That spends the run on a result already in hand. Estimate sign first, then consistency among the levers that might actually help. A conflicted record demotes but never removes. Won twice and lost twice is not a measured neutral: it behaved differently under conditions the record does not capture, which makes it the weaker bet while that holds. The demotion lifts by itself once the record stops disagreeing, because it describes the evidence rather than the lever. No SKU means no substitution. A result measured on an H100 says nothing about an MI355X, and scoring one from the other is the single mistake the record's GPU key exists to prevent — so an unnamed box keeps the prior rather than guessing which record applies. Off by default, on `Policy.use_history` and `LoopConfig.use_history`, matching the two flags already on Policy. Merging changes no ranking; turning it on is its own decision, and the loop writes `history_read.json` beside the other run artifacts so what the ranking saw is on the record. History is passed in rather than read from disk inside the ranker, which keeps selection a pure function of its arguments. Validation: 1266 passed, 1 skipped, exit 0; Ruff clean. Nine new tests. Also driven end to end against a real runs/ directory from generate_demo_runs.py: with the flag off all three levers score an identical +0.0167 from the constant; with it on they separate to +0.1550 measured, +0.0144 measured and demoted, and -0.0300 measured. No run against a live GPU. Co-Authored-By: Claude Opus 5 --- gitm/agents/policy.py | 77 ++++++++++++++++- gitm/optimizer/replay.py | 14 ++- gitm/scheduler/loop.py | 23 ++++- tests/test_policy_history.py | 161 +++++++++++++++++++++++++++++++++++ 4 files changed, 267 insertions(+), 8 deletions(-) create mode 100644 tests/test_policy_history.py diff --git a/gitm/agents/policy.py b/gitm/agents/policy.py index 1d796c4..0075834 100644 --- a/gitm/agents/policy.py +++ b/gitm/agents/policy.py @@ -1,4 +1,10 @@ -"""Selection policy: pre-filter by safety, rank by predicted delta, return top-N.""" +"""Selection policy: pre-filter by safety, rank by predicted delta, return top-N. + +Ranking is a precedence tuple rather than one blended number, for the reason +:mod:`gitm.playbook.match` gives: terms answering different questions should not +be collapsed into a scalar where one can quietly outvote another. Gate first, +then evidence quality, then magnitude, then a deterministic tie-break. +""" from __future__ import annotations @@ -6,6 +12,7 @@ from dataclasses import dataclass from gitm.kernels.spec import InterventionSpec +from gitm.optimizer.history import History, record_for from gitm.optimizer.preconditions import GateContext, applicable from gitm.optimizer.replay import predict_delta from gitm.tracer.schema import Trace @@ -16,6 +23,17 @@ class RankedCandidate: spec: InterventionSpec predicted_delta: float rejected_reason: str | None = None + #: Where ``predicted_delta``'s effect estimate came from: ``"prior"`` for + #: the spec's hand-authored ``expected_delta_mean``, ``"measured"`` for a + #: delta this lever recorded on this GPU. Carried for the same reason + #: ``rejected_reason`` is: a number is worth less without what produced it. + delta_source: str = "prior" + #: Ranked below every undemoted candidate, but never removed. A lever whose + #: record both won and lost has not come out neutral — it behaved differently + #: under conditions the record does not capture, so it is the weaker bet + #: while that holds. The demotion lifts by itself once the record stops + #: disagreeing: it describes the evidence, not the lever. + demoted: bool = False @dataclass @@ -24,6 +42,11 @@ class Policy: require_qualification_commit: bool = False skip_high_risk: bool = False + #: Score a lever from what it measured before, where there is a record for + #: this GPU. Off by default because it changes which experiments run, and + #: that should be a decision someone made rather than one that arrived + #: with an upgrade. + use_history: bool = False def select_interventions( @@ -33,7 +56,20 @@ def select_interventions( top_n: int = 5, *, ctx: GateContext | None = None, + history: History | None = None, + gpu_sku: str | None = None, ) -> list[RankedCandidate]: + """Rank the library for this trace, rejected candidates last. + + ``history`` is passed in rather than read from disk here, so ranking stays a + pure function of what it is given and a caller can rank against a record it + has already filtered. It is consulted only when ``policy.use_history`` is on + *and* ``gpu_sku`` names the box: a result measured on an H100 says nothing + about an MI355X, and scoring one from the other is the mistake the record's + GPU key exists to prevent. No SKU therefore means no substitution, not a + guess at which box the record came from. + """ + use_history = policy.use_history and history is not None and gpu_sku is not None candidates: list[RankedCandidate] = [] for spec in library: @@ -46,10 +82,43 @@ def select_interventions( reason = "policy.skip_high_risk" elif reason is None and (spec.safety.requires_qualification_commit and not policy.require_qualification_commit): reason = "safety.requires_qualification_commit" - delta = predict_delta(trace, spec) if reason is None else 0.0 - candidates.append(RankedCandidate(spec=spec, predicted_delta=delta, rejected_reason=reason)) + record = ( + record_for(history, spec.name, gpu_sku=gpu_sku) + if use_history and reason is None + else None + ) + # A record with no usable delta is still a record: it says the lever was + # tried and how it fared, but carries no number to rank on. The prior + # stands in that case and only the demotion applies. + measured = record.mean_delta if record is not None else None + delta = predict_delta(trace, spec, delta_mean=measured) if reason is None else 0.0 + candidates.append(RankedCandidate( + spec=spec, + predicted_delta=delta, + rejected_reason=reason, + delta_source="measured" if measured is not None else "prior", + demoted=bool(record is not None and record.conflicted), + )) + # Four terms, in this order and for these reasons: + # + # 1. Rejected. The gate's answer is categorical and comes first. + # 2. Not worth a run. A lever whose estimate is zero or negative is not a + # candidate whatever the evidence behind it says, so it sorts below every + # lever that might help. This sits above the demotion because a lever + # measured at -9% every time is a worse bet than one that is merely + # inconsistent, and ranking the known loser higher would spend the run on + # a result already in hand. + # 3. Demoted. Among levers that might help, prefer the one whose record does + # not disagree with itself. + # 4. Magnitude, then name for a deterministic order. candidates.sort( - key=lambda c: (c.rejected_reason is not None, -c.predicted_delta, c.spec.name) + key=lambda c: ( + c.rejected_reason is not None, + c.predicted_delta <= 0.0, + c.demoted, + -c.predicted_delta, + c.spec.name, + ) ) return candidates[:top_n] diff --git a/gitm/optimizer/replay.py b/gitm/optimizer/replay.py index bfd7871..3dd476f 100644 --- a/gitm/optimizer/replay.py +++ b/gitm/optimizer/replay.py @@ -19,12 +19,21 @@ from gitm.tracer.schema import Trace -def predict_delta(trace: Trace, spec: InterventionSpec) -> float: +def predict_delta( + trace: Trace, spec: InterventionSpec, *, delta_mean: float | None = None +) -> float: """Predicted fractional delta in wall-clock time on this trace. v0 model: apply the spec's ``expected_delta_mean`` weighted by the fraction of trace time spent in ops the spec is applicable to. The trace-driven replay engine that replaces this v0 is on the roadmap. + + ``delta_mean`` replaces the spec's estimate of the effect, leaving coverage — + which is a property of *this* trace — untouched. ``expected_delta_mean`` is + hand-authored and identical on every run; a delta this lever actually + measured on this GPU is a better estimate of the same quantity, so it + substitutes rather than being blended in against some weighting constant + nobody has calibrated. """ total_ns = max(trace.duration_ns, 1) applicable_ns = 0 @@ -32,7 +41,8 @@ def predict_delta(trace: Trace, spec: InterventionSpec) -> float: if _applies(spec, k.name): applicable_ns += k.end_ns - k.start_ns coverage = applicable_ns / total_ns - return coverage * spec.expected_delta_mean + mean = spec.expected_delta_mean if delta_mean is None else delta_mean + return coverage * mean def _applies(spec: InterventionSpec, kernel_name: str) -> bool: diff --git a/gitm/scheduler/loop.py b/gitm/scheduler/loop.py index 05d165c..24a6072 100644 --- a/gitm/scheduler/loop.py +++ b/gitm/scheduler/loop.py @@ -38,6 +38,7 @@ from gitm.optimizer.collective_signal import collective_causes, worst_device_comm from gitm.optimizer.deviation import deviation_summary, deviation_trace, write_deviation_jsonl from gitm.optimizer.dr import attribute_dr +from gitm.optimizer.history import load_history from gitm.optimizer.measure import measure_trace, measurement_claims, measurement_summary from gitm.optimizer.monitor import check_invariants, residuals from gitm.optimizer.qualification import qualify @@ -161,6 +162,10 @@ class LoopConfig: target: float = 0.15 scratch: str | None = None top_n_interventions: int = 5 + #: Rank levers from what previous runs measured on this GPU, instead of from + #: the library's hand-authored estimates alone. Off by default: it changes + #: which experiments the run actually spends its budget on. + use_history: bool = False # Optional explicit driver for the embedded/engine path. When unset, the # loop looks up ``workload`` in the workload registry (gitm.workloads). workload_runner: WorkloadRunner | None = None @@ -715,8 +720,22 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]: for s in load_library(workload=workload) for resolved in expand_relative_candidates(s, cfg.engine) ] - policy = Policy(require_qualification_commit=qual.commit, skip_high_risk=not qual.commit) - ranked = select_interventions(trace, library, policy, top_n=cfg.top_n_interventions, ctx=pctx.gate) + policy = Policy(require_qualification_commit=qual.commit, skip_high_risk=not qual.commit, + use_history=cfg.use_history) + # Read once per run, filtered to this box. A lever measured on another GPU is + # not evidence about this one, and load_history counts what it filtered out + # rather than letting a thin record look like a weak lever. + prior_runs = load_history(runs_dir(cfg.scratch), gpu_sku=pctx.sku) if cfg.use_history else None + if prior_runs is not None: + (run_dir / "history_read.json").write_text(json.dumps({ + "runs_read": prior_runs.runs_read, + "filtered": prior_runs.filtered, + "skipped": prior_runs.skipped, + "levers": len(prior_runs.records), + "gpu_sku": pctx.sku, + }, indent=2)) + ranked = select_interventions(trace, library, policy, top_n=cfg.top_n_interventions, + ctx=pctx.gate, history=prior_runs, gpu_sku=pctx.sku) (run_dir / "ranked_candidates.json").write_text( json.dumps( [ diff --git a/tests/test_policy_history.py b/tests/test_policy_history.py new file mode 100644 index 0000000..9d3bf4c --- /dev/null +++ b/tests/test_policy_history.py @@ -0,0 +1,161 @@ +"""Ranking levers from what previous runs measured, rather than from constants alone.""" + +from __future__ import annotations + +from gitm.agents.policy import Policy, select_interventions +from gitm.kernels.spec import Applicability, InterventionSpec, SafetyGate +from gitm.optimizer.history import History, LeverRecord +from gitm.tracer.schema import KernelEvent, Trace + +SKU = "AMD Instinct MI355X" + + +def _trace() -> Trace: + """One kernel per lever's scope, so coverage is equal and only the effect + estimate can move the ranking.""" + events = [ + KernelEvent(name="fused_moe_kernel", start_ns=0, end_ns=500, stream_id=7, + device_id=0, correlation_id=1), + KernelEvent(name="void gemm_kernel", start_ns=500, end_ns=1000, stream_id=7, + device_id=0, correlation_id=2), + ] + return Trace( + workload_id="vllm-decode", fingerprint="fp", run_id="r", device_count=1, + vendor="amd", captured_at_ns=0, duration_ns=1000, events=events, + ) + + +def _spec(name, kernels, *, mean=0.05) -> InterventionSpec: + return InterventionSpec( + name=name, summary="s", knob=name, value=1, + expected_delta_mean=mean, expected_delta_lo=0.0, expected_delta_hi=0.1, + source="t", applies_to_kernels=kernels, + applicability=Applicability(workloads=["vllm-decode"]), + safety=SafetyGate(tier="moderate"), + ) + + +def _record(name, *, mean, wins=1, losses=0, gpu=SKU) -> LeverRecord: + return LeverRecord( + intervention_name=name, gpu_sku=gpu, runs=1, attempts=wins + losses, + wins=wins, losses=losses, inconclusive=0, mean_delta=mean, + best_delta=mean, worst_delta=mean, last_run_id="r1", + ) + + +def _history(*records) -> History: + return History(records={(r.intervention_name, r.gpu_sku): r for r in records}, + runs_read=1) + + +def _ranked(**kw): + lib = [_spec("moe_lever", ["fused_moe_kernel"]), _spec("gemm_lever", ["gemm"])] + return select_interventions(_trace(), lib, kw.pop("policy", Policy()), top_n=5, **kw) + + +def test_history_is_ignored_until_the_policy_asks_for_it(): + """The flag defaults off, so merging this changes no ranking by itself — + turning it on is its own decision.""" + h = _history(_record("moe_lever", mean=-0.30)) + + off = {c.spec.name: c for c in _ranked(history=h, gpu_sku=SKU)} + + assert off["moe_lever"].delta_source == "prior" + assert off["moe_lever"].predicted_delta > 0 # scored from the constant + + +def test_a_measured_delta_replaces_the_hand_authored_estimate(): + """Both levers cover the same share of the trace, so the only thing that can + separate them is the effect estimate. The lever measured at -30% must fall + below the one still scored from its prior.""" + h = _history(_record("moe_lever", mean=-0.30)) + policy = Policy(use_history=True) + + ranked = _ranked(policy=policy, history=h, gpu_sku=SKU) + by_name = {c.spec.name: c for c in ranked} + + assert by_name["moe_lever"].delta_source == "measured" + assert by_name["moe_lever"].predicted_delta < 0 + assert by_name["gemm_lever"].delta_source == "prior" + assert ranked[0].spec.name == "gemm_lever" + + +def test_a_measured_win_outranks_an_unmeasured_lever(): + h = _history(_record("moe_lever", mean=0.49)) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=SKU) + + assert ranked[0].spec.name == "moe_lever" + assert ranked[0].delta_source == "measured" + + +def test_a_conflicted_lever_is_demoted_but_never_removed(): + """Won twice and lost twice is not neutral — it behaved differently under + conditions the record does not capture. It ranks below every clean candidate + and still runs when nothing better is available.""" + h = _history(_record("moe_lever", mean=0.49, wins=2, losses=2)) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=SKU) + by_name = {c.spec.name: c for c in ranked} + + assert by_name["moe_lever"].demoted is True + # demoted despite the larger measured delta, which alone would rank it first + assert by_name["moe_lever"].predicted_delta > by_name["gemm_lever"].predicted_delta + assert ranked[0].spec.name == "gemm_lever" + assert by_name["moe_lever"] in ranked # still a candidate + + +def test_the_demotion_lifts_once_the_record_stops_disagreeing(): + """It describes the evidence, not the lever.""" + settled = _history(_record("moe_lever", mean=0.49, wins=3, losses=0)) + ranked = _ranked(policy=Policy(use_history=True), history=settled, gpu_sku=SKU) + + assert ranked[0].spec.name == "moe_lever" + assert ranked[0].demoted is False + + +def test_a_record_from_another_box_is_not_evidence_about_this_one(): + h = _history(_record("moe_lever", mean=-0.30, gpu="NVIDIA H100 80GB HBM3")) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=SKU) + by_name = {c.spec.name: c for c in ranked} + + assert by_name["moe_lever"].delta_source == "prior" + assert by_name["moe_lever"].predicted_delta > 0 + + +def test_no_sku_means_no_substitution_rather_than_a_guess(): + """An unnamed box is the case the record's GPU key exists to protect against + — scoring off whichever record happened to be there is the mistake.""" + h = _history(_record("moe_lever", mean=-0.30)) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=None) + + assert all(c.delta_source == "prior" for c in ranked) + + +def test_a_record_with_no_usable_delta_keeps_the_prior_and_the_demotion(): + """"Tried, and we have no number" is not "measured at zero" — the record + still says the lever disagreed with itself, but carries nothing to rank on.""" + rec = LeverRecord(intervention_name="moe_lever", gpu_sku=SKU, runs=2, attempts=2, + wins=1, losses=1, inconclusive=0, mean_delta=None, + best_delta=None, worst_delta=None, last_run_id="r1") + ranked = _ranked(policy=Policy(use_history=True), history=_history(rec), gpu_sku=SKU) + by_name = {c.spec.name: c for c in ranked} + + assert by_name["moe_lever"].delta_source == "prior" + assert by_name["moe_lever"].predicted_delta > 0 # the constant still applies + assert by_name["moe_lever"].demoted is True # but the conflict still counts + + +def test_a_known_loser_never_outranks_an_uncertain_candidate(tmp_path=None): + """The demotion orders levers that might help; it does not promote one that + measured negative every time. Ranking a consistent -9% above an inconsistent + +1% would spend the run on a result already in hand.""" + h = _history( + _record("moe_lever", mean=0.02, wins=2, losses=2), # conflicted, demoted + _record("gemm_lever", mean=-0.09, wins=0, losses=3), # a settled loser + ) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=SKU) + by_name = {c.spec.name: c for c in ranked} + + assert by_name["moe_lever"].demoted is True + assert by_name["gemm_lever"].demoted is False + assert by_name["gemm_lever"].predicted_delta < 0 + assert ranked[0].spec.name == "moe_lever" # demoted, but still the better bet From 352c0e159e017850f2c2a317c738e395491dc0d2 Mon Sep 17 00:00:00 2001 From: Gaurang Date: Sun, 20 Sep 2026 17:29:20 +0530 Subject: [PATCH 2/6] loop: ask once whether to rank from the previous runs' results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `use_history` was a flag someone had to know existed, defaulting to off, which meant the record stayed unread unless a caller went looking for it. The run now asks: where previous runs left measured results and there is a terminal to ask at, it says how many and offers to rank from them. Three notes on what the question is, because the obvious reading of it does not match how runs are stored. Nothing can be overwritten. Each run writes into `runs//`, so a new run never lands on an existing `verification.json` and there is no collision to resolve. The choice is only whether this run *reads* the earlier exports. Declining skips them for this run and leaves every one of them on disk — measurements that cost GPU time are not discarded by answering a prompt. No answer means yes. An unattended run — a 24h budget started over ssh, a cron job — must not sit on a prompt forever, and of the two answers using the record is the one that throws nothing away. Without a terminal there is nobody to ask, so it takes the same default immediately rather than waiting out sixty seconds against a pipe that will never reply. The question is put before the capture, not after it, so nobody is answering a prompt that arrived an hour into their run. `LoopConfig.use_history` becomes tri-state: None asks, True and False decide outright and are never second-guessed, which is what keeps scripted and scheduled runs deterministic. Validation: 1275 passed, 1 skipped, exit 0; Ruff clean. Nine new tests driving a real pipe rather than a StringIO, since the prompt waits on select and a stream that cannot be selected would test something the run never does. One of them pins that declining deletes nothing. Co-Authored-By: Claude Opus 5 --- gitm/scheduler/loop.py | 74 ++++++++++++++++++++++--- tests/test_history_prompt.py | 102 +++++++++++++++++++++++++++++++++++ 2 files changed, 170 insertions(+), 6 deletions(-) create mode 100644 tests/test_history_prompt.py diff --git a/gitm/scheduler/loop.py b/gitm/scheduler/loop.py index 24a6072..f472b36 100644 --- a/gitm/scheduler/loop.py +++ b/gitm/scheduler/loop.py @@ -11,6 +11,8 @@ import json import os import re +import select +import sys import time import uuid from dataclasses import asdict, dataclass @@ -38,7 +40,7 @@ from gitm.optimizer.collective_signal import collective_causes, worst_device_comm from gitm.optimizer.deviation import deviation_summary, deviation_trace, write_deviation_jsonl from gitm.optimizer.dr import attribute_dr -from gitm.optimizer.history import load_history +from gitm.optimizer.history import EXPORT_NAME, load_history from gitm.optimizer.measure import measure_trace, measurement_claims, measurement_summary from gitm.optimizer.monitor import check_invariants, residuals from gitm.optimizer.qualification import qualify @@ -163,9 +165,10 @@ class LoopConfig: scratch: str | None = None top_n_interventions: int = 5 #: Rank levers from what previous runs measured on this GPU, instead of from - #: the library's hand-authored estimates alone. Off by default: it changes - #: which experiments the run actually spends its budget on. - use_history: bool = False + #: the library's hand-authored estimates alone. ``None`` asks, where there is + #: a record to ask about and someone to ask; ``True``/``False`` decide it + #: outright and skip the question. + use_history: bool | None = None # Optional explicit driver for the embedded/engine path. When unset, the # loop looks up ``workload`` in the workload registry (gitm.workloads). workload_runner: WorkloadRunner | None = None @@ -408,9 +411,68 @@ def _ar_target_residual(ar_run: AutoresearchRun, fallback: float = 0.0) -> float return _clamp_pct(ar_run.target.residual) if ar_run.target is not None else fallback +def _prior_runs_with_results(scratch: str | None) -> int: + """How many previous runs left a verification export under ``runs/``.""" + d = runs_dir(scratch) + if not d.is_dir(): + return 0 + return sum(1 for p in d.iterdir() if p.is_dir() and (p / EXPORT_NAME).exists()) + + +def _ask_use_history( + n_runs: int, *, timeout_s: float = 60.0, stream: Any = None, tty: bool | None = None +) -> bool: + """Ask whether to score this run from what previous runs measured. + + No answer means yes, for two reasons. An unattended run — a 24h budget + started over ssh, a cron job — must not sit on a prompt forever, and of the + two choices, using the record is the one that discards nothing: declining + only skips it for this run. Nothing is deleted either way, because each run + writes into its own ``runs//`` and never touches another's export. + + Without a terminal there is nobody to ask, so it takes the same default + rather than waiting out the timeout against a pipe that will never answer. + """ + stream = sys.stdin if stream is None else stream + interactive = tty if tty is not None else bool(getattr(stream, "isatty", lambda: False)()) + if not interactive: + return True + + print( + f"\n{n_runs} previous run(s) left measured results." + "\n [Y] rank this run from them [n] ignore them and score from the catalog" + f"\n Nothing is deleted either way. No answer within {timeout_s:.0f}s uses them." + "\n> ", + end="", flush=True, + ) + try: + ready, _, _ = select.select([stream], [], [], timeout_s) + except (OSError, ValueError): # not a selectable stream + return True + if not ready: + print(f"\n no answer in {timeout_s:.0f}s — using previous results.") + return True + answer = (stream.readline() or "").strip().lower() + if answer.startswith("n"): + print(" ignoring previous results for this run; they stay on disk.") + return False + return True + + +def _resolve_use_history(cfg: LoopConfig) -> bool: + """``cfg.use_history`` when it was set, otherwise the answer to the prompt.""" + if cfg.use_history is not None: + return cfg.use_history + n = _prior_runs_with_results(cfg.scratch) + return _ask_use_history(n) if n else False + + def run_loop(cfg: LoopConfig) -> dict[str, Any]: """Execute the 24-hour loop and return ``{summary, report_md, ...}``.""" workload = cfg.workload or (getattr(cfg.engine, "workload_id", None) or "vllm-decode") + # Asked before the capture rather than after it, so nobody is answering a + # prompt that arrived an hour into their run. + use_history = _resolve_use_history(cfg) run_id = uuid.uuid4().hex budget_s = _parse_budget_s(cfg.budget) started_ns = time.time_ns() @@ -721,11 +783,11 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]: for resolved in expand_relative_candidates(s, cfg.engine) ] policy = Policy(require_qualification_commit=qual.commit, skip_high_risk=not qual.commit, - use_history=cfg.use_history) + use_history=use_history) # Read once per run, filtered to this box. A lever measured on another GPU is # not evidence about this one, and load_history counts what it filtered out # rather than letting a thin record look like a weak lever. - prior_runs = load_history(runs_dir(cfg.scratch), gpu_sku=pctx.sku) if cfg.use_history else None + prior_runs = load_history(runs_dir(cfg.scratch), gpu_sku=pctx.sku) if use_history else None if prior_runs is not None: (run_dir / "history_read.json").write_text(json.dumps({ "runs_read": prior_runs.runs_read, diff --git a/tests/test_history_prompt.py b/tests/test_history_prompt.py new file mode 100644 index 0000000..0787114 --- /dev/null +++ b/tests/test_history_prompt.py @@ -0,0 +1,102 @@ +"""Asking, once, whether a run should be scored from what previous runs measured.""" + +from __future__ import annotations + +import json +import os + +from gitm.scheduler.loop import ( + LoopConfig, + _ask_use_history, + _prior_runs_with_results, + _resolve_use_history, +) + + +def _pipe(text: str | None): + """A real pipe, because the prompt waits on ``select`` and a StringIO is not + selectable — a fake stream here would test something the run never does.""" + r, w = os.pipe() + if text is not None: + os.write(w, text.encode()) + os.close(w) if text is not None else None + return os.fdopen(r) + + +def _runs(tmp_path, n, *, with_export=True): + runs = tmp_path / "runs" + runs.mkdir(parents=True, exist_ok=True) + for i in range(n): + d = runs / f"run{i}" + d.mkdir() + if with_export: + (d / "verification.json").write_text(json.dumps({"results": []})) + return tmp_path + + +# --------------------------------------------------------------------------- # +# what there is to ask about # +# --------------------------------------------------------------------------- # +def test_only_runs_that_actually_recorded_something_count(tmp_path): + _runs(tmp_path, 2) + _runs(tmp_path / "other", 3, with_export=False) + + assert _prior_runs_with_results(str(tmp_path)) == 2 + assert _prior_runs_with_results(str(tmp_path / "other")) == 0 + + +def test_no_previous_results_asks_nothing_and_uses_nothing(tmp_path): + """There is no question to put, and no record to rank from.""" + assert _resolve_use_history(LoopConfig(scratch=str(tmp_path))) is False + + +# --------------------------------------------------------------------------- # +# the answer # +# --------------------------------------------------------------------------- # +def test_yes_uses_the_previous_results(): + assert _ask_use_history(3, stream=_pipe("y\n"), tty=True) is True + + +def test_a_bare_enter_uses_them(): + """Capitalised [Y] in the prompt promises this.""" + assert _ask_use_history(3, stream=_pipe("\n"), tty=True) is True + + +def test_no_ignores_them(): + assert _ask_use_history(3, stream=_pipe("n\n"), tty=True) is False + + +def test_silence_uses_them_rather_than_waiting_forever(): + """An unattended run must not sit on a prompt. Of the two answers, using the + record is the one that discards nothing.""" + assert _ask_use_history(3, timeout_s=0.2, stream=_pipe(None), tty=True) is True + + +def test_without_a_terminal_it_does_not_wait_at_all(): + """Nobody is there to answer, so it takes the same default immediately rather + than burning the timeout against a pipe that will never reply.""" + assert _ask_use_history(3, timeout_s=30.0, stream=_pipe(None), tty=False) is True + + +# --------------------------------------------------------------------------- # +# the config still decides when it was told to # +# --------------------------------------------------------------------------- # +def test_an_explicit_setting_is_not_second_guessed(tmp_path): + """A caller that said which way it wants this is never prompted — that is + what keeps scripted and scheduled runs deterministic.""" + _runs(tmp_path, 2) + + assert _resolve_use_history(LoopConfig(scratch=str(tmp_path), use_history=False)) is False + assert _resolve_use_history(LoopConfig(scratch=str(tmp_path), use_history=True)) is True + + +def test_declining_deletes_nothing(tmp_path): + """Answering no skips the record for this run. It does not throw away + measurements that cost GPU time to produce.""" + _runs(tmp_path, 2) + before = sorted(p.name for p in (tmp_path / "runs").iterdir()) + + assert _ask_use_history(2, stream=_pipe("n\n"), tty=True) is False + + assert sorted(p.name for p in (tmp_path / "runs").iterdir()) == before + assert all((tmp_path / "runs" / d / "verification.json").exists() for d in before) From 17c619c3841fd7482b1a556c7d3d29e5ce06142d Mon Sep 17 00:00:00 2001 From: Gaurang Date: Sun, 20 Sep 2026 18:05:27 +0530 Subject: [PATCH 3/6] cli: the question about previous results belongs to `gitm run` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prompt went into `run_loop`, which put it on the wrong side of the library boundary. `gitm.optimize` is the embedded entry point — a vLLM process calls it in-line — and a function that can block on stdin is one such a caller cannot use. Nothing would have appeared on a terminal nobody was watching; the run would simply have stopped for sixty seconds in the middle of someone else's process, once per run, for a question they never asked to be asked. A prompt is a property of being run by a person at a terminal, so it now lives with the other things that are: `gitm run` asks before it calls optimize, and passes the answer down as an argument. The three layers each keep one job. `runs_with_results` in `optimizer/history` counts run folders that left an export, which is cheap enough to call before deciding whether there is anything worth asking about — it stats directories rather than parsing them. `gitm/cli` owns the question and the sixty-second timeout. `run_loop` takes a bool and never asks; unset now reads as off rather than as "go and find out". `--use-history` and `--no-history` decide it without a prompt, for scripted and scheduled runs, and an explicit flag is never re-litigated. Behaviour at the terminal is unchanged from the previous commit: the question is still put before any capture, silence still means use them, no tty still means use them, and declining still deletes nothing. Two of the new tests are structural rather than behavioural. One reads the scheduler's source and fails if `select.select` or a prompt helper ever drifts back into it; the other pins that `optimize` takes the answer as a parameter and contains no `input(`. Both guard the boundary rather than today's arrangement of it. Validation: 1276 passed, 1 skipped, exit 0; Ruff clean. Co-Authored-By: Claude Opus 5 --- gitm/api.py | 8 ++ gitm/cli.py | 74 +++++++++++++++++++ gitm/optimizer/history.py | 13 ++++ gitm/scheduler/loop.py | 72 ++---------------- ...y_prompt.py => test_cli_history_prompt.py} | 67 +++++++++++------ 5 files changed, 146 insertions(+), 88 deletions(-) rename tests/{test_history_prompt.py => test_cli_history_prompt.py} (57%) diff --git a/gitm/api.py b/gitm/api.py index 9a4ddc8..1795865 100644 --- a/gitm/api.py +++ b/gitm/api.py @@ -20,6 +20,7 @@ def optimize( target: float = 0.15, scratch: str | None = None, workload_runner: Callable[[], dict[str, Any]] | None = None, + use_history: bool | None = None, ) -> dict[str, Any]: """Run the autonomous 24-hour optimization loop and return a report. @@ -29,6 +30,12 @@ def optimize( of ``target`` fraction improvement within ``budget`` wall time, or a qualification-gate diagnostic explaining why the floor was not committed. + ``use_history`` ranks candidates from what previous runs measured on this + GPU rather than from the library's hand-authored estimates. It is never + asked for here — this entry point does not touch stdin, so an embedded + caller cannot be blocked by a prompt it did not expect. ``gitm run`` puts + the question to the operator and passes the answer down. + ``workload_runner`` optionally supplies an explicit zero-arg callable that launches the workload's GPU work; it runs inside the capture window. When omitted, the loop resolves ``workload`` against the registry in @@ -41,5 +48,6 @@ def optimize( target=target, scratch=scratch, workload_runner=workload_runner, + use_history=use_history, ) return run_loop(cfg) diff --git a/gitm/cli.py b/gitm/cli.py index 4dd987f..559bbc7 100644 --- a/gitm/cli.py +++ b/gitm/cli.py @@ -6,6 +6,7 @@ import json import sys from pathlib import Path +from typing import Any from gitm.optimizer.deviation import add_deviate_arguments from gitm.planner.registry import add_plan_arguments @@ -84,6 +85,15 @@ def _parser() -> argparse.ArgumentParser: run = sub.add_parser("run", help="Run the autonomous optimization loop.") run.add_argument("--workload", required=True, help="Workload identifier, e.g. vllm-decode.") run.add_argument("--budget", default="24h", help="Wall-clock budget, e.g. 24h.") + hist = run.add_mutually_exclusive_group() + hist.add_argument( + "--use-history", dest="use_history", action="store_true", default=None, + help="Rank levers from what previous runs measured on this GPU, without asking.", + ) + hist.add_argument( + "--no-history", dest="use_history", action="store_false", + help="Ignore previous runs' results and score from the catalog. Deletes nothing.", + ) run.add_argument( "--target", default="15%", @@ -264,6 +274,67 @@ def _parse_target(s: str) -> float: _HFT_WORKLOADS = {"hft", "hft-lob"} +def _ask_use_history(n_runs: int, *, timeout_s: float = 60.0, stream: Any = None, + tty: bool | None = None) -> bool: + """Ask whether this run should be scored from what previous runs measured. + + Lives here, and not in the loop, because a prompt is a property of being run + by a person at a terminal. ``gitm.optimize`` never touches stdin, so an + embedded caller cannot be blocked by a question it did not ask for. + + No answer means yes, for two reasons. An unattended run must not sit on a + prompt forever, and of the two answers using the record is the one that + discards nothing: declining only skips it for this run. Nothing is deleted + either way, since every run writes into its own ``runs//`` and never + touches another run's export. + + Without a terminal there is nobody to ask, so it takes the same default at + once rather than waiting out the timeout against a pipe that will not reply. + """ + import select + + stream = sys.stdin if stream is None else stream + interactive = tty if tty is not None else bool(getattr(stream, "isatty", lambda: False)()) + if not interactive: + return True + + print( + f"\n{n_runs} previous run(s) left measured results." + "\n [Y] rank this run from them [n] ignore them and score from the catalog" + f"\n Nothing is deleted either way. No answer within {timeout_s:.0f}s uses them." + "\n> ", + end="", flush=True, + ) + try: + ready, _, _ = select.select([stream], [], [], timeout_s) + except (OSError, ValueError): # not a selectable stream + return True + if not ready: + print(f"\n no answer in {timeout_s:.0f}s \u2014 using previous results.") + return True + answer = (stream.readline() or "").strip().lower() + if answer.startswith("n"): + print(" ignoring previous results for this run; they stay on disk.") + return False + return True + + +def _resolve_use_history(args: Any) -> bool: + """What ``--use-history`` said, or the answer to the prompt. + + An explicit flag is never second-guessed, which is what keeps scripted and + scheduled runs deterministic. With no flag and no previous results there is + nothing to ask about and nothing to rank from. + """ + if getattr(args, "use_history", None) is not None: + return bool(args.use_history) + from gitm._paths import runs_dir + from gitm.optimizer.history import runs_with_results + + n = runs_with_results(runs_dir(args.scratch)) + return _ask_use_history(n) if n else False + + def _apply_hft_run_flags(args) -> None: """Map the hft-only run flags onto the ``GITM_BENCH_*`` env the workload factory reads. Errors if they're used with a non-hft workload, where they @@ -354,11 +425,14 @@ def main(argv: list[str] | None = None) -> int: from gitm import optimize _apply_hft_run_flags(args) + # Asked here, before the loop starts any capture, so nobody answers a + # prompt that arrived an hour into a 24h run. result = optimize( workload=args.workload, budget=args.budget, target=_parse_target(args.target), scratch=args.scratch, + use_history=_resolve_use_history(args), ) summary = result.get("summary", {}) if args.report is not None: diff --git a/gitm/optimizer/history.py b/gitm/optimizer/history.py index 009fc91..64e7e88 100644 --- a/gitm/optimizer/history.py +++ b/gitm/optimizer/history.py @@ -22,6 +22,7 @@ __all__ = [ "LeverRecord", + "runs_with_results", "History", "load_history", "record_for", @@ -102,6 +103,18 @@ def _verdict(result: dict[str, Any]) -> str: return "win" if result.get("significant") else "inconclusive" +def runs_with_results(runs_dir: str | Path) -> int: + """How many run folders under ``runs_dir`` left a verification export. + + Cheap enough to call before deciding whether there is anything worth asking + the operator about: it stats each directory rather than parsing any of them. + """ + runs_dir = Path(runs_dir) + if not runs_dir.is_dir(): + return 0 + return sum(1 for p in runs_dir.iterdir() if p.is_dir() and (p / EXPORT_NAME).exists()) + + def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History: """Aggregate every readable ``verification.json`` under ``runs_dir``. diff --git a/gitm/scheduler/loop.py b/gitm/scheduler/loop.py index f472b36..c71c5bf 100644 --- a/gitm/scheduler/loop.py +++ b/gitm/scheduler/loop.py @@ -11,8 +11,6 @@ import json import os import re -import select -import sys import time import uuid from dataclasses import asdict, dataclass @@ -40,7 +38,7 @@ from gitm.optimizer.collective_signal import collective_causes, worst_device_comm from gitm.optimizer.deviation import deviation_summary, deviation_trace, write_deviation_jsonl from gitm.optimizer.dr import attribute_dr -from gitm.optimizer.history import EXPORT_NAME, load_history +from gitm.optimizer.history import load_history from gitm.optimizer.measure import measure_trace, measurement_claims, measurement_summary from gitm.optimizer.monitor import check_invariants, residuals from gitm.optimizer.qualification import qualify @@ -165,9 +163,9 @@ class LoopConfig: scratch: str | None = None top_n_interventions: int = 5 #: Rank levers from what previous runs measured on this GPU, instead of from - #: the library's hand-authored estimates alone. ``None`` asks, where there is - #: a record to ask about and someone to ask; ``True``/``False`` decide it - #: outright and skip the question. + #: the library's hand-authored estimates alone. ``None`` means nobody said, + #: which reads as off: the loop never asks. ``gitm run`` puts the question to + #: the operator and passes an explicit answer down. use_history: bool | None = None # Optional explicit driver for the embedded/engine path. When unset, the # loop looks up ``workload`` in the workload registry (gitm.workloads). @@ -411,68 +409,12 @@ def _ar_target_residual(ar_run: AutoresearchRun, fallback: float = 0.0) -> float return _clamp_pct(ar_run.target.residual) if ar_run.target is not None else fallback -def _prior_runs_with_results(scratch: str | None) -> int: - """How many previous runs left a verification export under ``runs/``.""" - d = runs_dir(scratch) - if not d.is_dir(): - return 0 - return sum(1 for p in d.iterdir() if p.is_dir() and (p / EXPORT_NAME).exists()) - - -def _ask_use_history( - n_runs: int, *, timeout_s: float = 60.0, stream: Any = None, tty: bool | None = None -) -> bool: - """Ask whether to score this run from what previous runs measured. - - No answer means yes, for two reasons. An unattended run — a 24h budget - started over ssh, a cron job — must not sit on a prompt forever, and of the - two choices, using the record is the one that discards nothing: declining - only skips it for this run. Nothing is deleted either way, because each run - writes into its own ``runs//`` and never touches another's export. - - Without a terminal there is nobody to ask, so it takes the same default - rather than waiting out the timeout against a pipe that will never answer. - """ - stream = sys.stdin if stream is None else stream - interactive = tty if tty is not None else bool(getattr(stream, "isatty", lambda: False)()) - if not interactive: - return True - - print( - f"\n{n_runs} previous run(s) left measured results." - "\n [Y] rank this run from them [n] ignore them and score from the catalog" - f"\n Nothing is deleted either way. No answer within {timeout_s:.0f}s uses them." - "\n> ", - end="", flush=True, - ) - try: - ready, _, _ = select.select([stream], [], [], timeout_s) - except (OSError, ValueError): # not a selectable stream - return True - if not ready: - print(f"\n no answer in {timeout_s:.0f}s — using previous results.") - return True - answer = (stream.readline() or "").strip().lower() - if answer.startswith("n"): - print(" ignoring previous results for this run; they stay on disk.") - return False - return True - - -def _resolve_use_history(cfg: LoopConfig) -> bool: - """``cfg.use_history`` when it was set, otherwise the answer to the prompt.""" - if cfg.use_history is not None: - return cfg.use_history - n = _prior_runs_with_results(cfg.scratch) - return _ask_use_history(n) if n else False - - def run_loop(cfg: LoopConfig) -> dict[str, Any]: """Execute the 24-hour loop and return ``{summary, report_md, ...}``.""" workload = cfg.workload or (getattr(cfg.engine, "workload_id", None) or "vllm-decode") - # Asked before the capture rather than after it, so nobody is answering a - # prompt that arrived an hour into their run. - use_history = _resolve_use_history(cfg) + # Never prompts. Deciding that is the CLI's job, because a library entry + # point that can block on stdin is one an embedded caller cannot use. + use_history = bool(cfg.use_history) run_id = uuid.uuid4().hex budget_s = _parse_budget_s(cfg.budget) started_ns = time.time_ns() diff --git a/tests/test_history_prompt.py b/tests/test_cli_history_prompt.py similarity index 57% rename from tests/test_history_prompt.py rename to tests/test_cli_history_prompt.py index 0787114..5332cb8 100644 --- a/tests/test_history_prompt.py +++ b/tests/test_cli_history_prompt.py @@ -1,16 +1,19 @@ -"""Asking, once, whether a run should be scored from what previous runs measured.""" +"""Asking, once, whether a run should be scored from what previous runs measured. + +The question lives in the CLI, not the loop: a prompt is a property of being run +by a person at a terminal, and ``gitm.optimize`` must stay usable from a process +that has no stdin to answer with. +""" from __future__ import annotations +import argparse import json import os -from gitm.scheduler.loop import ( - LoopConfig, - _ask_use_history, - _prior_runs_with_results, - _resolve_use_history, -) +from gitm.cli import _ask_use_history, _resolve_use_history +from gitm.optimizer.history import runs_with_results +from gitm.scheduler.loop import LoopConfig def _pipe(text: str | None): @@ -41,13 +44,17 @@ def test_only_runs_that_actually_recorded_something_count(tmp_path): _runs(tmp_path, 2) _runs(tmp_path / "other", 3, with_export=False) - assert _prior_runs_with_results(str(tmp_path)) == 2 - assert _prior_runs_with_results(str(tmp_path / "other")) == 0 + assert runs_with_results(tmp_path / "runs") == 2 + assert runs_with_results(tmp_path / "other" / "runs") == 0 + + +def _args(scratch, use_history=None): + return argparse.Namespace(scratch=str(scratch), use_history=use_history) def test_no_previous_results_asks_nothing_and_uses_nothing(tmp_path): """There is no question to put, and no record to rank from.""" - assert _resolve_use_history(LoopConfig(scratch=str(tmp_path))) is False + assert _resolve_use_history(_args(tmp_path)) is False # --------------------------------------------------------------------------- # @@ -81,22 +88,36 @@ def test_without_a_terminal_it_does_not_wait_at_all(): # --------------------------------------------------------------------------- # # the config still decides when it was told to # # --------------------------------------------------------------------------- # -def test_an_explicit_setting_is_not_second_guessed(tmp_path): - """A caller that said which way it wants this is never prompted — that is - what keeps scripted and scheduled runs deterministic.""" +def test_an_explicit_flag_is_not_second_guessed(tmp_path): + """``--use-history`` / ``--no-history`` are never re-litigated by a prompt — + that is what keeps scripted and scheduled runs deterministic.""" _runs(tmp_path, 2) - assert _resolve_use_history(LoopConfig(scratch=str(tmp_path), use_history=False)) is False - assert _resolve_use_history(LoopConfig(scratch=str(tmp_path), use_history=True)) is True + assert _resolve_use_history(_args(tmp_path, use_history=False)) is False + assert _resolve_use_history(_args(tmp_path, use_history=True)) is True -def test_declining_deletes_nothing(tmp_path): - """Answering no skips the record for this run. It does not throw away - measurements that cost GPU time to produce.""" - _runs(tmp_path, 2) - before = sorted(p.name for p in (tmp_path / "runs").iterdir()) +def test_the_loop_cannot_reach_a_prompt(tmp_path): + """An embedded caller has no stdin to answer with, so the loop must not be + able to ask at all — not merely avoid asking today. This fails if a prompt + ever drifts back into the scheduler.""" + import inspect + + from gitm.scheduler import loop as loop_mod + + src = inspect.getsource(loop_mod) + assert "select.select" not in src + assert not hasattr(loop_mod, "_ask_use_history") + # and with nobody having said either way, history is simply off + assert LoopConfig(scratch=str(tmp_path)).use_history is None + + +def test_optimize_does_not_ask_either(): + """The public entry point takes the answer as an argument. If it grew a + prompt, an embedded vLLM process would hang on a question nobody sees.""" + import inspect - assert _ask_use_history(2, stream=_pipe("n\n"), tty=True) is False + from gitm.api import optimize - assert sorted(p.name for p in (tmp_path / "runs").iterdir()) == before - assert all((tmp_path / "runs" / d / "verification.json").exists() for d in before) + assert "use_history" in inspect.signature(optimize).parameters + assert "input(" not in inspect.getsource(optimize) From 3de1a3548c3cd7c3671b11768ee1dfec8a752445 Mon Sep 17 00:00:00 2001 From: Gaurang Date: Sun, 20 Sep 2026 20:52:33 +0530 Subject: [PATCH 4/6] history: a record belongs to one model as much as to one GPU MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review catches, both live once the record started deciding which experiments run. Records keyed on (lever, gpu_sku) merged results from different models measured on the same box, which a shared scratch directory guarantees: runs/ is per-machine, not per-checkpoint. A +40% on kimi-k2.5 and a -20% on glm-5.2 came back as one record reading a confident +10%, describing neither — and not flagged conflicted either, because both were kept and significant, so both counted as wins. Nothing in the record said it was answering about a model nobody asked about. `fingerprint` is the third key. It already rides in every export's provenance and is exactly the right grain: qualification derives it from kernel mix and shapes, so the same model and config reproduce it and a different checkpoint does not. Asking with the wrong one now returns None, which falls back to the catalog estimate — the same shape as never having measured the lever at all, which is the truth in that case. `load_history` filters on it beside gpu_sku and counts the exclusions as filtered rather than skipped, and the rendered table gains a workload column, without which two rows differing only by model look like one row recorded twice. The reader also trusted its own numbers too far, and this now runs in Phase 3 — after the capture has been paid for. A string `speedup` raised TypeError out of the entire read, taking every sound run with it and aborting the run that had just spent its budget on a trace. A NaN delta passed the isinstance check, survived the mean, and reached json.dumps, which writes the literal NaN into `history_read.json` — not valid JSON, in an artifact whose only purpose is being read by something else. True passed it too, bool being an int subclass, and read as a measured +100%. A value that is not a real finite number is now absent rather than coerced, and the attempt still counts as won or lost: what is missing is the magnitude, not the result. That is the distinction this module already draws everywhere else. Validation: 1283 passed, 1 skipped, exit 0; Ruff clean. Seven new tests, including the two-model average, the string speedup that aborted the read, the NaN that produced invalid JSON, and True reading as +100%. Co-Authored-By: Claude Opus 5 --- gitm/agents/policy.py | 3 +- gitm/optimizer/history.py | 95 +++++++++++++++++++++++++++--------- gitm/scheduler/loop.py | 7 ++- tests/test_history.py | 90 ++++++++++++++++++++++++++++++++-- tests/test_policy_history.py | 33 +++++++++++-- 5 files changed, 195 insertions(+), 33 deletions(-) diff --git a/gitm/agents/policy.py b/gitm/agents/policy.py index 0075834..403bfa1 100644 --- a/gitm/agents/policy.py +++ b/gitm/agents/policy.py @@ -58,6 +58,7 @@ def select_interventions( ctx: GateContext | None = None, history: History | None = None, gpu_sku: str | None = None, + fingerprint: str | None = None, ) -> list[RankedCandidate]: """Rank the library for this trace, rejected candidates last. @@ -83,7 +84,7 @@ def select_interventions( elif reason is None and (spec.safety.requires_qualification_commit and not policy.require_qualification_commit): reason = "safety.requires_qualification_commit" record = ( - record_for(history, spec.name, gpu_sku=gpu_sku) + record_for(history, spec.name, gpu_sku=gpu_sku, fingerprint=fingerprint) if use_history and reason is None else None ) diff --git a/gitm/optimizer/history.py b/gitm/optimizer/history.py index 64e7e88..f913dd3 100644 --- a/gitm/optimizer/history.py +++ b/gitm/optimizer/history.py @@ -16,6 +16,7 @@ from __future__ import annotations import json +import math from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -44,6 +45,12 @@ class LeverRecord: intervention_name: str gpu_sku: str | None + #: The workload fingerprint the result was measured under — model plus + #: config, from :func:`gitm.optimizer.qualification.fingerprint`. Part of the + #: key for the same reason ``gpu_sku`` is: a lever that helps a sparse-MoE + #: checkpoint says nothing about a dense one, and averaging a +40% on one + #: model with a -20% on another reports a confident +10% describing neither. + fingerprint: str | None #: Distinct run folders this lever appears in. Kept apart from ``attempts`` #: because five A/Bs inside one run is far weaker evidence than five across #: five runs, and a single count cannot tell those apart. @@ -90,6 +97,34 @@ def __len__(self) -> int: return len(self.records) +def _finite(value: Any) -> float | None: + """A real, finite number, or ``None``. + + ``bool`` is excluded deliberately: it is an ``int`` subclass, so ``True`` + would otherwise be read as a measured delta of 1.0. + """ + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) if math.isfinite(value) else None + + +def _delta_of(result: dict[str, Any]) -> float | None: + """The signed delta this A/B recorded, or ``None`` when it has none usable. + + A value that is not a real finite number is treated as absent rather than + coerced, and the attempt still counts as won or lost. Two failures this + prevents, both of which land in Phase 3 *after* the capture has been paid + for: a string ``speedup`` raised ``TypeError`` out of the whole read, and a + NaN propagated through the mean into ``json.dumps``, which emits the literal + ``NaN`` \u2014 not valid JSON, in an artifact meant to be machine-read. + """ + delta = _finite(result.get("delta")) + if delta is not None: + return delta + speedup = _finite(result.get("speedup")) + return None if speedup is None else speedup - 1.0 + + def _verdict(result: dict[str, Any]) -> str: """win / loss / inconclusive for one A/B. @@ -115,12 +150,16 @@ def runs_with_results(runs_dir: str | Path) -> int: return sum(1 for p in runs_dir.iterdir() if p.is_dir() and (p / EXPORT_NAME).exists()) -def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History: +def load_history( + runs_dir: str | Path, *, gpu_sku: str | None = None, + fingerprint: str | None = None, +) -> History: """Aggregate every readable ``verification.json`` under ``runs_dir``. - ``gpu_sku`` filters to one GPU: a result measured on an H100 says nothing - about an MI355X, so a caller ranking for one box should not see the other's - record. Runs are ordered by the export's mtime — the export carries a + ``gpu_sku`` filters to one GPU and ``fingerprint`` to one workload: a result + measured on an H100 says nothing about an MI355X, and one measured on a + sparse-MoE checkpoint says nothing about a dense one. A caller ranking for a + particular box and model should see neither of the others' records. Runs are ordered by the export's mtime — the export carries a ``run_id`` but no timestamp, and the run directories are UUIDs, so the file is the only ordering available for ``last_run_id``. """ @@ -130,7 +169,7 @@ def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History if not runs_dir.is_dir(): return History(skipped={str(runs_dir): "runs dir does not exist"}) - exports: list[tuple[float, str, str | None, list[dict[str, Any]]]] = [] + exports: list[tuple[float, str, str | None, str | None, list[dict[str, Any]]]] = [] for d in sorted(p for p in runs_dir.iterdir() if p.is_dir()): path = d / EXPORT_NAME if not path.exists(): @@ -163,23 +202,26 @@ def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History # attempts with nothing saying so, which is what skipped is for. skipped[d.name] = "malformed result entry" continue + prov = doc.get("provenance") if isinstance(doc.get("provenance"), dict) else {} sku = (env or {}).get("gpu_sku") - if gpu_sku is not None and sku != gpu_sku: + fp = prov.get("fingerprint") + if (gpu_sku is not None and sku != gpu_sku) or ( + fingerprint is not None and fp != fingerprint + ): filtered += 1 continue - prov = doc.get("provenance") - run_id = (prov.get("run_id") if isinstance(prov, dict) else None) or d.name - exports.append((path.stat().st_mtime, run_id, sku, results)) + run_id = prov.get("run_id") or d.name + exports.append((path.stat().st_mtime, run_id, sku, fp, results)) exports.sort(key=lambda e: e[0]) - acc: dict[tuple[str, str | None], dict[str, Any]] = {} - for _mtime, run_id, sku, results in exports: + acc: dict[tuple[str, str | None, str | None], dict[str, Any]] = {} + for _mtime, run_id, sku, fp, results in exports: for r in results: name = r.get("intervention_name") if not name: continue - key = (name, sku) + key = (name, sku, fp) a = acc.setdefault( key, {"runs": set(), "attempts": 0, "win": 0, "loss": 0, @@ -188,19 +230,18 @@ def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History a["runs"].add(run_id) a["attempts"] += 1 a[_verdict(r)] += 1 - delta = r.get("delta") - if delta is None and r.get("speedup") is not None: - delta = r["speedup"] - 1.0 - if isinstance(delta, (int | float)): - a["deltas"].append(float(delta)) + delta = _delta_of(r) + if delta is not None: + a["deltas"].append(delta) a["last_run_id"] = run_id records = {} - for (name, sku), a in acc.items(): + for (name, sku, fp), a in acc.items(): deltas = a["deltas"] - records[(name, sku)] = LeverRecord( + records[(name, sku, fp)] = LeverRecord( intervention_name=name, gpu_sku=sku, + fingerprint=fp, runs=len(a["runs"]), attempts=a["attempts"], wins=a["win"], @@ -217,15 +258,20 @@ def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History def record_for( - history: History, intervention_name: str, *, gpu_sku: str | None = None + history: History, intervention_name: str, *, gpu_sku: str | None = None, + fingerprint: str | None = None, ) -> LeverRecord | None: - """The record for one lever, or ``None`` if no run ever tried it. + """The record for one lever on one box under one workload, or ``None``. ``None`` rather than a zeroed record on purpose: "never measured" and "measured, and it did nothing" must not look alike to a caller deciding whether this lever is worth an experiment. They point opposite ways. + + Both ``gpu_sku`` and ``fingerprint`` are part of the key, so asking with the + wrong one returns nothing rather than another context's answer. A caller + that has no fingerprint gets records measured without one, not all of them. """ - return history.records.get((intervention_name, gpu_sku)) + return history.records.get((intervention_name, gpu_sku, fingerprint)) def _fit(value: str, width: int) -> str: @@ -276,9 +322,11 @@ def render_history(history: History, *, top: int = 20) -> str: # look like one box is the exact confusion keying on gpu_sku exists to stop. name_w = _column(shown, lambda r: r.intervention_name, "lever", cap=40) gpu_w = _column(shown, lambda r: r.gpu_sku or "-", "gpu", cap=30) + fp_w = _column(shown, lambda r: r.fingerprint or "-", "workload", cap=24) out.append("") - out.append(f" {'lever':{name_w}s} {'gpu':{gpu_w}s} {'runs':>5s} {'a/b':>4s} " + out.append(f" {'lever':{name_w}s} {'gpu':{gpu_w}s} {'workload':{fp_w}s} " + f"{'runs':>5s} {'a/b':>4s} " f"{'won':>4s} {'lost':>5s} {'incon':>6s} {'mean':>8s} last") for r in shown: flag = " CONFLICTED" if r.conflicted else "" @@ -286,6 +334,7 @@ def render_history(history: History, *, top: int = 20) -> str: out.append( f" {_fit(r.intervention_name, name_w):{name_w}s} " f"{_fit(r.gpu_sku or '-', gpu_w):{gpu_w}s} " + f"{_fit(r.fingerprint or '-', fp_w):{fp_w}s} " f"{r.runs:5d} {r.attempts:4d} {r.wins:4d} {r.losses:5d} {r.inconclusive:6d} " f"{mean:>7s} {(r.last_run_id or '-')[:8]}{flag}" ) diff --git a/gitm/scheduler/loop.py b/gitm/scheduler/loop.py index c71c5bf..bb3b751 100644 --- a/gitm/scheduler/loop.py +++ b/gitm/scheduler/loop.py @@ -729,7 +729,8 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]: # Read once per run, filtered to this box. A lever measured on another GPU is # not evidence about this one, and load_history counts what it filtered out # rather than letting a thin record look like a weak lever. - prior_runs = load_history(runs_dir(cfg.scratch), gpu_sku=pctx.sku) if use_history else None + prior_runs = (load_history(runs_dir(cfg.scratch), gpu_sku=pctx.sku, + fingerprint=qual.fingerprint) if use_history else None) if prior_runs is not None: (run_dir / "history_read.json").write_text(json.dumps({ "runs_read": prior_runs.runs_read, @@ -737,9 +738,11 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]: "skipped": prior_runs.skipped, "levers": len(prior_runs.records), "gpu_sku": pctx.sku, + "fingerprint": qual.fingerprint, }, indent=2)) ranked = select_interventions(trace, library, policy, top_n=cfg.top_n_interventions, - ctx=pctx.gate, history=prior_runs, gpu_sku=pctx.sku) + ctx=pctx.gate, history=prior_runs, gpu_sku=pctx.sku, + fingerprint=qual.fingerprint) (run_dir / "ranked_candidates.json").write_text( json.dumps( [ diff --git a/tests/test_history.py b/tests/test_history.py index 13007c9..519501f 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -223,8 +223,8 @@ def test_reads_an_export_written_by_the_real_writer(tmp_path): ]) h = load_history(tmp_path) - won = record_for(h, "kv_cache_dtype_fp8", gpu_sku="NVIDIA H100 80GB") - lost = record_for(h, "cuda_graphs_enable", gpu_sku="NVIDIA H100 80GB") + won = record_for(h, "kv_cache_dtype_fp8", gpu_sku="NVIDIA H100 80GB", fingerprint="fp") + lost = record_for(h, "cuda_graphs_enable", gpu_sku="NVIDIA H100 80GB", fingerprint="fp") assert h.runs_read == 1 assert (won.wins, won.losses) == (1, 0) @@ -244,7 +244,8 @@ def test_gpu_sku_survives_the_real_writer(tmp_path): h = load_history(tmp_path, gpu_sku="AMD MI355X") assert h.runs_read == 1 and h.filtered == 0 - assert record_for(h, "enable_expert_parallel", gpu_sku="AMD MI355X").wins == 1 + assert record_for(h, "enable_expert_parallel", gpu_sku="AMD MI355X", + fingerprint="fp").wins == 1 def test_a_real_export_with_no_results_is_not_mistaken_for_a_crash(tmp_path): @@ -415,3 +416,86 @@ def test_a_damaged_export_is_never_counted_as_filtered(tmp_path): assert h.filtered == 0 assert h.skipped == {"bad-env": "malformed environment"} + + +def test_two_models_on_one_box_do_not_average_into_one_number(tmp_path): + """A lever that helps a sparse-MoE checkpoint says nothing about a dense one. + Keyed on the GPU alone, a +40% on one model and a -20% on another merged into + a confident +10% describing neither — and not even flagged conflicted, since + both were kept and significant, so both counted as wins.""" + for run_id, fp, delta in (("r1", "kimi-k2.5", 0.40), ("r2", "glm-5.2", -0.20)): + d = tmp_path / run_id + d.mkdir() + (d / "verification.json").write_text(json.dumps({ + "provenance": {"run_id": run_id, "fingerprint": fp}, + "environment": {"gpu_sku": "AMD Instinct MI355X"}, + "results": [_result("kv_cache_dtype_fp8", delta=delta)], + })) + + h = load_history(tmp_path) + + assert len(h.records) == 2 + kimi = record_for(h, "kv_cache_dtype_fp8", gpu_sku="AMD Instinct MI355X", + fingerprint="kimi-k2.5") + glm = record_for(h, "kv_cache_dtype_fp8", gpu_sku="AMD Instinct MI355X", + fingerprint="glm-5.2") + assert abs(kimi.mean_delta - 0.40) < 1e-9 + assert abs(glm.mean_delta - (-0.20)) < 1e-9 + # and asking for a model nobody measured returns nothing, not someone else's + assert record_for(h, "kv_cache_dtype_fp8", gpu_sku="AMD Instinct MI355X", + fingerprint="mimi-v2.5") is None + + +def test_the_workload_filter_is_counted_apart_from_damage(tmp_path): + _run(tmp_path, "keep", [_result("kv_cache_dtype_fp8")]) + d = tmp_path / "other" + d.mkdir() + (d / "verification.json").write_text(json.dumps({ + "provenance": {"run_id": "other", "fingerprint": "glm-5.2"}, + "environment": {"gpu_sku": "NVIDIA H100 80GB"}, + "results": [_result("kv_cache_dtype_fp8")], + })) + + h = load_history(tmp_path, fingerprint="glm-5.2") + + assert h.runs_read == 1 and h.filtered == 1 and not h.skipped + + +def test_a_string_speedup_does_not_abort_the_whole_read(tmp_path): + """This runs in Phase 3, after the capture has been paid for. A malformed + number in one old export must not take the run down with it.""" + _run(tmp_path, "good", [_result("kv_cache_dtype_fp8", delta=0.10)]) + _run(tmp_path, "bad", [{**_result("enforce_eager"), "delta": None, "speedup": "1.1"}]) + + h = load_history(tmp_path) + + assert h.runs_read == 2 + assert record_for(h, "kv_cache_dtype_fp8", gpu_sku="NVIDIA H100 80GB").wins == 1 + # the attempt still counts; only its unusable number is dropped + bad = record_for(h, "enforce_eager", gpu_sku="NVIDIA H100 80GB") + assert bad.attempts == 1 and bad.mean_delta is None + + +def test_a_non_finite_delta_is_absent_rather_than_ranked(tmp_path): + """NaN survives a mean and reaches json.dumps, which emits the literal NaN — + not valid JSON, in an artifact written to be machine-read.""" + _run(tmp_path, "nan", [_result("kv_cache_dtype_fp8", delta=float("nan"))]) + + rec = record_for(load_history(tmp_path), "kv_cache_dtype_fp8", + gpu_sku="NVIDIA H100 80GB") + + assert rec.attempts == 1 + assert rec.mean_delta is None + json.dumps({"mean": rec.mean_delta}) # round-trips as null + + +def test_true_is_not_a_measured_delta_of_one(tmp_path): + """``bool`` is an ``int`` subclass, so a sloppy check reads True as +100%.""" + _run(tmp_path, "b", [{**_result("kv_cache_dtype_fp8"), + "delta": True, "speedup": None}]) + + rec = record_for(load_history(tmp_path), "kv_cache_dtype_fp8", + gpu_sku="NVIDIA H100 80GB") + + assert rec.attempts == 1 + assert rec.mean_delta is None diff --git a/tests/test_policy_history.py b/tests/test_policy_history.py index 9d3bf4c..0ad7359 100644 --- a/tests/test_policy_history.py +++ b/tests/test_policy_history.py @@ -8,6 +8,7 @@ from gitm.tracer.schema import KernelEvent, Trace SKU = "AMD Instinct MI355X" +FP = "kimi-k2.5" def _trace() -> Trace: @@ -35,21 +36,24 @@ def _spec(name, kernels, *, mean=0.05) -> InterventionSpec: ) -def _record(name, *, mean, wins=1, losses=0, gpu=SKU) -> LeverRecord: +def _record(name, *, mean, wins=1, losses=0, gpu=SKU, fp=FP) -> LeverRecord: return LeverRecord( - intervention_name=name, gpu_sku=gpu, runs=1, attempts=wins + losses, + intervention_name=name, gpu_sku=gpu, fingerprint=fp, runs=1, + attempts=wins + losses, wins=wins, losses=losses, inconclusive=0, mean_delta=mean, best_delta=mean, worst_delta=mean, last_run_id="r1", ) def _history(*records) -> History: - return History(records={(r.intervention_name, r.gpu_sku): r for r in records}, + return History(records={(r.intervention_name, r.gpu_sku, r.fingerprint): r + for r in records}, runs_read=1) def _ranked(**kw): lib = [_spec("moe_lever", ["fused_moe_kernel"]), _spec("gemm_lever", ["gemm"])] + kw.setdefault("fingerprint", FP) return select_interventions(_trace(), lib, kw.pop("policy", Policy()), top_n=5, **kw) @@ -134,7 +138,7 @@ def test_a_record_with_no_usable_delta_keeps_the_prior_and_the_demotion(): """"Tried, and we have no number" is not "measured at zero" — the record still says the lever disagreed with itself, but carries nothing to rank on.""" rec = LeverRecord(intervention_name="moe_lever", gpu_sku=SKU, runs=2, attempts=2, - wins=1, losses=1, inconclusive=0, mean_delta=None, + fingerprint=FP, wins=1, losses=1, inconclusive=0, mean_delta=None, best_delta=None, worst_delta=None, last_run_id="r1") ranked = _ranked(policy=Policy(use_history=True), history=_history(rec), gpu_sku=SKU) by_name = {c.spec.name: c for c in ranked} @@ -159,3 +163,24 @@ def test_a_known_loser_never_outranks_an_uncertain_candidate(tmp_path=None): assert by_name["gemm_lever"].demoted is False assert by_name["gemm_lever"].predicted_delta < 0 assert ranked[0].spec.name == "moe_lever" # demoted, but still the better bet + + +def test_another_models_record_is_not_evidence_about_this_one(): + """A shared scratch holds runs from several checkpoints on one box. A lever + measured on a sparse-MoE model says nothing about a dense one.""" + h = _history(_record("moe_lever", mean=-0.30, fp="glm-5.2")) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=SKU) + by_name = {c.spec.name: c for c in ranked} + + assert by_name["moe_lever"].delta_source == "prior" + assert by_name["moe_lever"].predicted_delta > 0 + + +def test_no_fingerprint_means_no_substitution(): + """Same reasoning as an unnamed GPU: without knowing which workload the + record came from, the prior stands rather than a guess.""" + h = _history(_record("moe_lever", mean=-0.30)) + ranked = _ranked(policy=Policy(use_history=True), history=h, gpu_sku=SKU, + fingerprint=None) + + assert all(c.delta_source == "prior" for c in ranked) From 154145bee70fd0d6a70e84675e905dd5aa5a2643 Mon Sep 17 00:00:00 2001 From: Gaurang Date: Sun, 20 Sep 2026 21:00:50 +0530 Subject: [PATCH 5/6] history: the isinstance form ruff 0.12 asks for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI pins ruff 0.12.11 and lints the whole tree with `ruff check .`; my local run was 0.16.8 over `gitm/ tests/`. UP038 wants `int | float` rather than `(int, float)` in isinstance, and the rule was removed in 0.16, so the newer version had nothing to say about it. The code this replaced used the `|` form already — the tuple was mine. Behaviour is identical; only the spelling changes. Co-Authored-By: Claude Opus 5 --- gitm/optimizer/history.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gitm/optimizer/history.py b/gitm/optimizer/history.py index f913dd3..64a1527 100644 --- a/gitm/optimizer/history.py +++ b/gitm/optimizer/history.py @@ -103,7 +103,7 @@ def _finite(value: Any) -> float | None: ``bool`` is excluded deliberately: it is an ``int`` subclass, so ``True`` would otherwise be read as a measured delta of 1.0. """ - if isinstance(value, bool) or not isinstance(value, (int, float)): + if isinstance(value, bool) or not isinstance(value, int | float): return None return float(value) if math.isfinite(value) else None From 5a555484614d5ab9a3176d542c8a743dbb02a31c Mon Sep 17 00:00:00 2001 From: Gaurang Date: Tue, 22 Sep 2026 10:50:39 +0530 Subject: [PATCH 6/6] history: an identity field is a key, so a list there is not a bad value MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Greptile held the PR at 4/5 on this and was right to: the previous fix covered the measured numbers and left the fields that identify them. `gpu_sku`, `fingerprint` and `intervention_name` become part of the record key, and `run_id` goes into a set, so a list or a dict in any of them raised `unhashable type` out of `load_history` entirely. That is one damaged export taking every sound run with it, in Phase 3, after the capture has already been paid for — the failure this module exists to contain rather than cause. An int was worse in a quieter way: hashable, so it loaded fine and broke `render_history` on `len()` instead, one screen further from its cause. The three key fields must be a string or absent, and a run that gets anything else is skipped with a reason naming which field. They are keys with no fallback: a record filed under the wrong box or the wrong model is the mistake the key exists to prevent, so keying it under None would be worse than not reading it. `run_id` stays tolerated, because it is the one identity field with a documented fallback — it becomes the run's directory name, exactly as it already does when provenance is missing, and the run's measurements are kept. Validation: 1286 passed, 1 skipped, exit 0. Ruff 0.12.11 over the whole tree, which is what CI pins and runs — the version I had been checking locally drops UP038 and passed a form CI rejects. Three new tests, all confirmed to fail without the guards. Co-Authored-By: Claude Opus 5 --- gitm/optimizer/history.py | 29 ++++++++++++++++- tests/test_history.py | 67 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 1 deletion(-) diff --git a/gitm/optimizer/history.py b/gitm/optimizer/history.py index 64a1527..586f01d 100644 --- a/gitm/optimizer/history.py +++ b/gitm/optimizer/history.py @@ -97,6 +97,18 @@ def __len__(self) -> int: return len(self.records) +def _named(value: Any) -> bool: + """True when ``value`` is usable as an identity field: a string, or absent. + + These become part of the record key and of the run set, so a list or a dict + here raises ``unhashable type`` out of the whole read — one damaged export + taking every sound run with it, which is the failure this module exists to + contain rather than cause. A number is hashable and would survive the load, + then break the renderer on ``len()`` instead. + """ + return value is None or isinstance(value, str) + + def _finite(value: Any) -> float | None: """A real, finite number, or ``None``. @@ -202,15 +214,30 @@ def load_history( # attempts with nothing saying so, which is what skipped is for. skipped[d.name] = "malformed result entry" continue + if not all(_named(r.get("intervention_name")) for r in results): + skipped[d.name] = "malformed intervention name" + continue prov = doc.get("provenance") if isinstance(doc.get("provenance"), dict) else {} sku = (env or {}).get("gpu_sku") fp = prov.get("fingerprint") + # Both are keys and neither has a fallback: a record filed under the wrong + # box or the wrong model is the mistake the key exists to prevent, so a + # malformed one skips the run rather than keying under None. + if not _named(sku): + skipped[d.name] = "malformed gpu_sku" + continue + if not _named(fp): + skipped[d.name] = "malformed fingerprint" + continue if (gpu_sku is not None and sku != gpu_sku) or ( fingerprint is not None and fp != fingerprint ): filtered += 1 continue - run_id = prov.get("run_id") or d.name + # run_id is the one identity field with a documented fallback, so a + # malformed one costs the run nothing. + raw_run_id = prov.get("run_id") + run_id = raw_run_id if isinstance(raw_run_id, str) and raw_run_id else d.name exports.append((path.stat().st_mtime, run_id, sku, fp, results)) exports.sort(key=lambda e: e[0]) diff --git a/tests/test_history.py b/tests/test_history.py index 519501f..0e459be 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -499,3 +499,70 @@ def test_true_is_not_a_measured_delta_of_one(tmp_path): assert rec.attempts == 1 assert rec.mean_delta is None + + +# --------------------------------------------------------------------------- # +# identity fields: keys and set members, so the wrong type is not a bad value # +# but an unhashable one # +# --------------------------------------------------------------------------- # +def _doc(*, prov, env, results): + return json.dumps({"provenance": prov, "environment": env, "results": results}) + + +_OK_PROV = {"run_id": "ok", "fingerprint": "fp"} +_OK_ENV = {"gpu_sku": "AMD Instinct MI355X"} + + +def test_an_unhashable_identity_field_does_not_abort_the_read(tmp_path): + """gpu_sku, fingerprint and intervention_name become part of the record key; + run_id goes into a set. A list or a dict in any of them raised + ``unhashable type`` out of the whole read — after the capture, taking every + sound run with it. That is the failure this module exists to contain.""" + cases = { + "bad-sku": _doc(prov=_OK_PROV, env={"gpu_sku": ["MI355X"]}, + results=[_result("kv_cache_dtype_fp8")]), + "bad-fp": _doc(prov={"run_id": "x", "fingerprint": {"a": 1}}, env=_OK_ENV, + results=[_result("kv_cache_dtype_fp8")]), + "bad-name": _doc(prov=_OK_PROV, env=_OK_ENV, + results=[{**_result("x"), "intervention_name": ["a"]}]), + } + _run(tmp_path, "good", [_result("kv_cache_dtype_fp8")], + gpu_sku="AMD Instinct MI355X") + (tmp_path / "good" / "verification.json").write_text( + _doc(prov=_OK_PROV, env=_OK_ENV, results=[_result("kv_cache_dtype_fp8")])) + for name, body in cases.items(): + _run(tmp_path, name, None, body=body) + + h = load_history(tmp_path) + + assert h.runs_read == 1 + assert set(h.skipped) == set(cases) + assert record_for(h, "kv_cache_dtype_fp8", gpu_sku="AMD Instinct MI355X", + fingerprint="fp").wins == 1 + + +def test_a_number_where_a_sku_belongs_is_caught_at_load_not_at_render(tmp_path): + """An int is hashable, so it survived the load and broke render_history on + ``len()`` instead — a failure one screen further from its cause.""" + _run(tmp_path, "bad", None, + body=_doc(prov=_OK_PROV, env={"gpu_sku": 42}, + results=[_result("kv_cache_dtype_fp8")])) + + h = load_history(tmp_path) + + assert h.skipped == {"bad": "malformed gpu_sku"} + render_history(h) # renders rather than raising + + +def test_a_malformed_run_id_falls_back_to_the_directory(tmp_path): + """The one identity field with a documented fallback: nothing is lost by + reading the run, so it is not worth skipping its measurements over.""" + _run(tmp_path, "weird", None, + body=_doc(prov={"run_id": {"a": 1}, "fingerprint": "fp"}, env=_OK_ENV, + results=[_result("kv_cache_dtype_fp8")])) + + rec = record_for(load_history(tmp_path), "kv_cache_dtype_fp8", + gpu_sku="AMD Instinct MI355X", fingerprint="fp") + + assert rec.runs == 1 + assert rec.last_run_id == "weird"