diff --git a/gitm/agents/policy.py b/gitm/agents/policy.py index 1d796c4..403bfa1 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,21 @@ def select_interventions( top_n: int = 5, *, 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. + + ``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 +83,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, fingerprint=fingerprint) + 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/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..586f01d 100644 --- a/gitm/optimizer/history.py +++ b/gitm/optimizer/history.py @@ -16,12 +16,14 @@ from __future__ import annotations import json +import math from dataclasses import dataclass, field from pathlib import Path from typing import Any __all__ = [ "LeverRecord", + "runs_with_results", "History", "load_history", "record_for", @@ -43,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. @@ -89,6 +97,46 @@ 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``. + + ``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. @@ -102,12 +150,28 @@ def _verdict(result: dict[str, Any]) -> str: return "win" if result.get("significant") else "inconclusive" -def load_history(runs_dir: str | Path, *, gpu_sku: str | None = None) -> History: +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, + 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``. """ @@ -117,7 +181,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(): @@ -150,23 +214,41 @@ 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 + 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") - if gpu_sku is not None and sku != 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 - 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 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]) - 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, @@ -175,19 +257,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"], @@ -204,15 +285,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: @@ -263,9 +349,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 "" @@ -273,6 +361,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/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..bb3b751 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,11 @@ 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. ``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). workload_runner: WorkloadRunner | None = None @@ -406,6 +412,9 @@ def _ar_target_residual(ar_run: AutoresearchRun, fallback: float = 0.0) -> float 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") + # 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() @@ -715,8 +724,25 @@ 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=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, + 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, + "filtered": prior_runs.filtered, + "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, + fingerprint=qual.fingerprint) (run_dir / "ranked_candidates.json").write_text( json.dumps( [ diff --git a/tests/test_cli_history_prompt.py b/tests/test_cli_history_prompt.py new file mode 100644 index 0000000..5332cb8 --- /dev/null +++ b/tests/test_cli_history_prompt.py @@ -0,0 +1,123 @@ +"""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.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): + """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 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(_args(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_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(_args(tmp_path, use_history=False)) is False + assert _resolve_use_history(_args(tmp_path, use_history=True)) is True + + +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 + + from gitm.api import optimize + + assert "use_history" in inspect.signature(optimize).parameters + assert "input(" not in inspect.getsource(optimize) diff --git a/tests/test_history.py b/tests/test_history.py index 13007c9..0e459be 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,153 @@ 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 + + +# --------------------------------------------------------------------------- # +# 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" diff --git a/tests/test_policy_history.py b/tests/test_policy_history.py new file mode 100644 index 0000000..0ad7359 --- /dev/null +++ b/tests/test_policy_history.py @@ -0,0 +1,186 @@ +"""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" +FP = "kimi-k2.5" + + +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, fp=FP) -> LeverRecord: + return LeverRecord( + 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.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) + + +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, + 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} + + 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 + + +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)