diff --git a/docs/optimization-workflow-unification.md b/docs/optimization-workflow-unification.md index 18f00606..2e88ec5a 100644 --- a/docs/optimization-workflow-unification.md +++ b/docs/optimization-workflow-unification.md @@ -176,6 +176,8 @@ inferrsi campaign start --directory \ 项目适配器先检查基线和候选质量，再做 warmup 和交替顺序的基线/候选测量；每轮和 final 使用相同协议。每步骤启动新进程，因此跨进程 warmup 只能预热可持久缓存；设备/进程内 warmup 必须在测量脚本内完成并说明。性能取声明指标的中位数，当前没有通用服务生命周期管理器或统计显著性判定。 +assessment 另附只读的 `paired_statistics`：同一次重复内的第 k 个基线与第 k 个候选样本按顺序配对，计算逐对加速比的中位数，以及按重复重采样的 95% percentile bootstrap 区间（记录 resamples 与 seed），并给出区间与冻结门槛的关系（`above_gate`、`straddles_gate`、`below_gate`；门槛没有有限加速比时为 `gate_unreachable`；少于 5 次重复为 `insufficient_pairs`；样本或派生值无效为 `unavailable`）。`agrees_with_point_gate` 标出中位数点估计的判定与配对区间是否一致。该字段只描述留存样本，不改变 verdict、eligible 或选优，`statistical_significance` 仍为 `not_established`。区间是名义 95%，覆盖率未经校准：合成对照中 3 对时约 75%，因此少于 5 对不报告区间。区间假设各次重复独立，不能消除同一对内共同的设备争用，也不构成环境资格。合成对照可用 `python tools/measurement_simulation.py` 复算。 + 结果协议： - 测量脚本正常结束并写 JSON，提供指定字段的有限正数。 @@ -195,6 +197,36 @@ inferrsi dashboard --input --output 知识检索和 profiling 按当前问题需要进行，不要求每轮重查或每轮 profile。知识关闭时受管 `knowledge` 操作拒绝执行；这不是对宿主所有文件访问的隔离。 +### 用 profile 解释改进（可选） + +端到端时间会随时钟、温度和共享负载整体漂移，单凭数值很难区分真实改进与波动。测量脚本可以在结果 JSON 中额外输出同一步骤的分区域耗时和工作量计数，单位由脚本自定并在各步骤保持一致： + +```json +{"metric": 8.1, + "breakdown": {"attention": 4.0, "norm": 0.1, "other": 4.0}, + "counters": {"kernel_launches": 20, "dram_bytes": 8.0e8}} +``` + +每轮可以用 `--mechanism-file` 声明机制和可核对的预期特征： + +```json +{"mechanism": "把 norm 融合进 attention 的 epilogue，少一次 HBM 往返", + "expect": [{"region": "norm", "direction": "decrease"}, + {"region": "attention", "direction": "unchanged"}, + {"counter": "kernel_launches", "direction": "decrease"}]} +``` + +assessment 的 `profile_evidence` 按重复配对比较每个区域和计数，`raw` 保留原始比值。声明为 `unchanged` 的区域是本轮的假设，不是结论：其他区域另外给出除以同一步骤参照区域总和后的 `normalized` 比较，`share_based_speedup` 用参照区域在步骤总时间中的占比推算整体加速。只有当某种变化让同一步骤的所有区域按同一比例缩放时，它才会被抵消；若各区域受影响程度不同（例如降频对计算受限和访存受限的区域影响不同），抵消就不成立。参照区域本身始终只做原始比较，不会互相归一化，因此整体漂移较大时它可能显示为变化，使解释判为矛盾。计数一类的工作量指标基本不随时钟和温度变化，可以说明波动范围内的小幅改进是否可重复。方向只由配对区间判断（区间整体低于或高于 1），仓库内没有设置数值容差；基线为 0 的计数没有比值，记为 `undefined`。`profile_evidence` 只描述数据，不判断机制文字本身。 + +`explanation.status` 取值：`explained`（每条预期都被数据确认，且至少一条是变化）、`unsupported`（预期的变化没有在数据中出现）、`contradicted`（数据与预期方向相反，或声明不变的项出现明显变化）、`unverifiable`（缺少对应区域或计数、比值无定义，或少于 5 对）、`not_declared`（本轮未声明）。机制声明在执行前绑定：判定使用执行前保存的声明，并核对其摘要与 attempt 目录中的 `mechanism.json`；副本被改动时按完整性错误阻断 campaign。执行后被中断、经 `recover` 结算的轮次无法证明执行前绑定的是哪份声明，因此按未声明处理（记录 `mechanism_unverified_after_recovery`），在 `--require-explanation` 下不会被采纳。final 沿用被选中轮次绑定的机制重新核对。 + +两个开关在 `campaign start` 时冻结，默认都不改变原有判定： + +- `--require-explanation`：只有 `explained` 的候选可以被采纳。`unsupported`、`contradicted` 记为 FAIL，`unverifiable`、`not_declared` 记为 INCONCLUSIVE；两类都不阻塞后续轮次。 +- `--decision paired`：用 `paired_statistics` 判定，要求至少 5 次重复。区间整体高于门槛为 PASS；跨过门槛为 FAIL，`reason` 为 `within_noise`，搜索继续；统计不可用时为 INCONCLUSIVE。默认 `point` 保持中位数点估计规则。 + +提交未修改的候选可以作为 A/A 对照，`paired_statistics` 的区间即给出该环境的噪声范围。 + ## 实际保留什么、保护什么 `campaign.json` 关联计划，`plan.json` 固定协议、预算和候选格式，`state.json` 是唯一可写控制状态。`summary.json` 是可重建摘要，dashboard 从同一记录生成展示；不额外写一份 `rounds.jsonl`。原有知识交付和优化日志格式继续复用。 diff --git a/src/inferrsi/arguments.py b/src/inferrsi/arguments.py index e943f2ec..00dcbc6d 100644 --- a/src/inferrsi/arguments.py +++ b/src/inferrsi/arguments.py @@ -67,6 +67,12 @@ def _optimization_actions(stages, *, project=False): start.add_argument("--min-improvement-pct", type=float, required=True) start.add_argument("--repetitions", type=int, required=True) start.add_argument("--warmups", type=int, required=True) + start.add_argument("--decision", choices=("point", "paired"), default="point", + help="point: independent medians (default); paired: the paired " + "interval must clear the gate, within-noise rounds fail") + start.add_argument("--require-explanation", action="store_true", + help="Adopt a candidate only when its declared mechanism is " + "confirmed by measured breakdown/counters") start.add_argument("--round-mode", choices=("exact", "up_to"), default="up_to") else: start.add_argument("task") @@ -89,6 +95,8 @@ def _optimization_actions(stages, *, project=False): if name == "round": stage.add_argument("--candidate", type=Path, required=True) stage.add_argument("--hypothesis", required=True) + stage.add_argument("--mechanism-file", type=Path, + help="JSON mechanism with expected region/counter directions") if name == "final": stage.add_argument("--reason", help="Why the caller stopped an up-to-budget search") if name in ("round", "final"): diff --git a/src/inferrsi/harness/measurement.py b/src/inferrsi/harness/measurement.py new file mode 100644 index 00000000..f4020263 --- /dev/null +++ b/src/inferrsi/harness/measurement.py @@ -0,0 +1,303 @@ +"""Paired, report-only statistics for alternating baseline/candidate measurements. + +The project protocol alternates AB/BA order inside each repetition, so the k-th +baseline and k-th candidate sample share one repetition. Their per-pair ratio +cancels drift shared within a repetition, which independent per-side medians +retain. This module only describes retained samples: it never changes a verdict, +eligibility or gate, and any value it cannot compute is reported as unavailable. + +The estimand is the median per-repetition speedup. The interval is a nominal 95% +percentile bootstrap over repetitions with a recorded seed. Its coverage is not +calibrated in general: with few repetitions it collapses towards the observed pair +range and under-covers (about 75% at three pairs in tools/measurement_simulation.py), +so no interval is reported below MIN_PAIRS. Repetitions are assumed independent; +pairing cannot +remove contention shared by both sides of one pair and does not qualify the +environment. +""" + +import math +import random +from statistics import median + +METHOD = "paired_median_speedup_percentile_bootstrap" +CONFIDENCE = 0.95 +MIN_PAIRS = 5 +RESAMPLES = 10_000 +SEED = 0 + + +def gate_speedup(min_improvement_pct: float, direction: str) -> float: + """Express the frozen improvement gate as the equivalent speedup ratio. + + A minimize gate of 100% or more has no finite speedup and returns ``inf``. + """ + fraction = min_improvement_pct / 100 + if direction == "maximize": + return 1 + fraction + if direction == "minimize": + return math.inf if fraction >= 1 else 1 / (1 - fraction) + raise ValueError("Metric direction must be explicit") + + +def paired_speedups(baseline: list, candidate: list, direction: str) -> list[float]: + if len(baseline) != len(candidate): + raise ValueError("Paired statistics need one candidate sample per baseline sample") + if direction not in ("minimize", "maximize"): + raise ValueError("Metric direction must be explicit") + values = [*baseline, *candidate] + if any(type(v) not in (int, float) or not math.isfinite(v) or v <= 0 for v in values): + raise ValueError("Paired statistics need finite positive samples") + speedups = [b / c if direction == "minimize" else c / b + for b, c in zip(baseline, candidate)] + if any(not math.isfinite(s) or s <= 0 for s in speedups): + raise ValueError("A paired ratio overflowed or underflowed") + return speedups + + +def relative_mad(values: list) -> float | None: + """Median absolute deviation divided by the median, a robust spread indicator.""" + if not values: + return None + center = median(values) + return median(abs(v - center) for v in values) / center + + +def _percentile(ordered: list, q: float) -> float: + position = q * (len(ordered) - 1) + low = math.floor(position) + high = min(low + 1, len(ordered) - 1) + return ordered[low] + (ordered[high] - ordered[low]) * (position - low) + + +def bootstrap_median_interval(values: list, *, resamples: int = RESAMPLES, + seed: int = SEED) -> list[float]: + if type(resamples) is not int or resamples < 1 or not values: + raise ValueError("Bootstrap needs a positive resample count and at least one value") + rng = random.Random(seed) + medians = sorted(median(rng.choices(values, k=len(values))) for _ in range(resamples)) + tail = (1 - CONFIDENCE) / 2 + return [_percentile(medians, tail), _percentile(medians, 1 - tail)] + + +def _finite(value) -> bool: + if isinstance(value, bool) or value is None or isinstance(value, str): + return True + if isinstance(value, (int, float)): + return math.isfinite(value) + if isinstance(value, (list, tuple)): + return all(_finite(item) for item in value) + if isinstance(value, dict): + return all(_finite(item) for item in value.values()) + return False + + +def _describe(baseline, candidate, *, direction, min_improvement_pct, point_gate_passed, + resamples, seed) -> dict: + speedups = paired_speedups(baseline, candidate, direction) + gate = gate_speedup(min_improvement_pct, direction) + report = { + "status": "available", "pairs": len(speedups), "pair_speedups": speedups, + "median_speedup": median(speedups) if speedups else None, + "geomean_speedup": (math.exp(sum(map(math.log, speedups)) / len(speedups)) + if speedups else None), + "gate_speedup": gate if math.isfinite(gate) else None, + "baseline_relative_mad": relative_mad(baseline), + "candidate_relative_mad": relative_mad(candidate), "interval": None, + } + if len(speedups) < MIN_PAIRS: + return {**report, "status": "insufficient_pairs", "gate_relation": "insufficient_pairs", + "agrees_with_point_gate": None, + "reason": f"At least {MIN_PAIRS} repetitions are needed for an interval"} + low, high = bootstrap_median_interval(speedups, resamples=resamples, seed=seed) + if not math.isfinite(gate): + relation = "gate_unreachable" + elif low > gate: + relation = "above_gate" + elif high < gate: + relation = "below_gate" + else: + relation = "straddles_gate" + return {**report, "interval": [low, high], "gate_relation": relation, + "excludes_no_change": low > 1 or high < 1, + "agrees_with_point_gate": point_gate_passed == (relation == "above_gate")} + + +def paired_statistics(baseline: list, candidate: list, *, direction: str, + min_improvement_pct: float, point_gate_passed: bool, + resamples: int = RESAMPLES, seed: int = SEED) -> dict: + """Describe paired samples against the frozen gate without changing any verdict. + + Every returned number is finite so the record stays strict-JSON serializable; + anything that cannot be computed is reported with ``status="unavailable"``. + """ + header = {"method": METHOD, "report_only": True, "confidence": CONFIDENCE, + "coverage": "nominal_uncalibrated", + "pairing": "kth baseline with kth candidate; AB/BA order alternates by repetition", + "resamples": resamples, "seed": seed} + try: + report = _describe(baseline, candidate, direction=direction, + min_improvement_pct=min_improvement_pct, + point_gate_passed=point_gate_passed, + resamples=resamples, seed=seed) + except (ArithmeticError, TypeError, ValueError) as error: + return {**header, "status": "unavailable", "reason": str(error), "interval": None} + if not _finite(report): + return {**header, "status": "unavailable", + "reason": "A derived statistic is not finite", "interval": None} + return {**header, **report} + + +# Profile evidence and declared mechanisms --------------------------------------- +# +# A measurement script may add per-step ``breakdown`` (region durations in one +# unit) and ``counters`` (work metrics such as kernel launches or DRAM bytes). +# Work metrics are largely independent of clock and thermal state, and region +# shares within one step cancel device-wide fluctuation, so they can explain a +# gain that end-to-end time alone cannot separate from noise. Directions are +# decided only from the paired data; no numeric tolerance is built in. + +DIRECTIONS = ("decrease", "increase", "unchanged") + + +def _numeric_map(value) -> dict | None: + if not isinstance(value, dict) or not value: + return None + if not all(isinstance(k, str) and k and type(v) in (int, float) and math.isfinite(v) + and v >= 0 for k, v in value.items()): + return None + return value + + +def _paired_change(baseline: list, candidate: list, *, resamples: int, seed: int) -> dict: + """Paired ratio direction; a zero baseline makes the ratio undefined, never decisive.""" + if len(baseline) < MIN_PAIRS: + return {"ratio_median": None, "interval": None, "observed": "insufficient_pairs"} + if any(b <= 0 for b in baseline): + return {"ratio_median": None, "interval": None, "observed": "undefined", + "note": "A zero baseline value has no paired ratio"} + ratios = [c / b for b, c in zip(baseline, candidate)] + low, high = bootstrap_median_interval(ratios, resamples=resamples, seed=seed) + observed = "decrease" if high < 1 else "increase" if low > 1 else "no_clear_change" + return {"ratio_median": median(ratios), "interval": [low, high], "observed": observed} + + +def _shares(values: list, totals: list) -> float | None: + shares = [v / t for v, t in zip(values, totals) if t > 0] + return median(shares) if len(shares) == len(values) else None + + +def profile_evidence(baseline_profiles: list, candidate_profiles: list, *, + reference_regions=(), resamples: int = RESAMPLES, + seed: int = SEED) -> dict: + """Compare per-repetition region durations and work counters between paired steps. + + Every region and counter is compared raw. ``reference_regions`` are regions the + round declares unchanged; they are an assumption, not a conclusion. Other regions + are additionally compared after dividing by the reference total of the same step, + which cancels a change that scales every region of that step by the same factor. + References themselves are only ever checked raw, against each other never. + """ + if len(baseline_profiles) != len(candidate_profiles) or not baseline_profiles: + return {"status": "unavailable", "reason": "Profiles must be paired per repetition"} + sides = (baseline_profiles, candidate_profiles) + result = {"status": "available", "pairs": len(baseline_profiles), "regions": {}, + "counters": {}, "reference_regions": [], "share_based_speedup": None} + for kind in ("breakdown", "counters"): + maps = [[_numeric_map((p or {}).get(kind)) for p in side] for side in sides] + if not all(m is not None for side in maps for m in side): + continue + shared = sorted(set.intersection(*(set(m) for side in maps for m in side))) + target = result["regions" if kind == "breakdown" else "counters"] + reference = [r for r in dict.fromkeys(reference_regions) if r in shared] if ( + kind == "breakdown") else [] + clock = [[sum(m[r] for r in reference) for m in side] for side in maps] + usable_clock = bool(reference) and min(min(side) for side in clock) > 0 + totals = [[sum(m.values()) for m in side] for side in maps] + for name in shared: + b = [m[name] for m in maps[0]] + c = [m[name] for m in maps[1]] + raw = _paired_change(b, c, resamples=resamples, seed=seed) + entry = {"baseline_median": median(b), "candidate_median": median(c), + "raw": raw, "observed": raw["observed"], "basis": "raw"} + if usable_clock and name not in reference: + normalized = _paired_change([v / k for v, k in zip(b, clock[0])], + [v / k for v, k in zip(c, clock[1])], + resamples=resamples, seed=seed) + entry.update(normalized=normalized, observed=normalized["observed"], + basis="normalized_to_reference_regions") + if kind == "breakdown": + entry["baseline_share"] = _shares(b, totals[0]) + entry["candidate_share"] = _shares(c, totals[1]) + target[name] = entry + if kind == "breakdown": + result["reference_regions"] = reference + if usable_clock and min(min(side) for side in totals) > 0: + ratios = [(kc / tc) / (kb / tb) for kb, tb, kc, tc in zip( + clock[0], totals[0], clock[1], totals[1])] + report = {"median": median(ratios), "interval": None} + if len(ratios) >= MIN_PAIRS: + report["interval"] = bootstrap_median_interval( + ratios, resamples=resamples, seed=seed) + result["share_based_speedup"] = report + if not result["regions"] and not result["counters"]: + return {**result, "status": "absent", + "reason": "No breakdown or counters were present in every measured step"} + return result if _finite(result) else {"status": "unavailable", + "reason": "A derived profile value is not finite"} + + +def validate_mechanism(document) -> dict: + """Validate a declared mechanism: a statement plus checkable expected signatures.""" + if not isinstance(document, dict) or set(document) - {"mechanism", "expect"}: + raise ValueError("A mechanism declares only 'mechanism' and 'expect'") + text = document.get("mechanism") + expect = document.get("expect") + if not isinstance(text, str) or not text.strip(): + raise ValueError("Describe the mechanism that should cause the change") + if not isinstance(expect, list) or not expect: + raise ValueError("List at least one expected region or counter signature") + for item in expect: + keys = set(item) if isinstance(item, dict) else set() + kinds = keys & {"region", "counter"} + if (len(kinds) != 1 or keys - {"region", "counter", "direction"} + or not isinstance(item[next(iter(kinds))], str) + or not item[next(iter(kinds))].strip() + or item.get("direction") not in DIRECTIONS): + raise ValueError("Each expectation names one region or counter and a direction " + "of decrease, increase or unchanged") + if all(item["direction"] == "unchanged" for item in expect): + raise ValueError("At least one expectation must predict a change") + names = [("region" if "region" in item else "counter", + item.get("region") or item.get("counter")) for item in expect] + if len(names) != len(set(names)): + raise ValueError("Declare each region or counter once") + return document + + +def explain(mechanism: dict | None, evidence: dict) -> dict: + """Check a declared mechanism against observed profile evidence, without judging prose.""" + if mechanism is None: + return {"status": "not_declared"} + checks, missing = [], False + for item in mechanism["expect"]: + kind = "region" if "region" in item else "counter" + table = evidence.get("regions" if kind == "region" else "counters") or {} + observed = table.get(item[kind], {}).get("observed") + if observed in (None, "insufficient_pairs", "undefined"): + missing = True + checks.append({"kind": kind, "name": item[kind], "expected": item["direction"], + "observed": observed or "missing"}) + report = {"mechanism": mechanism["mechanism"], "checks": checks} + if any(c["expected"] in ("decrease", "increase") and c["observed"] in ("decrease", "increase") + and c["observed"] != c["expected"] for c in checks) or any( + c["expected"] == "unchanged" and c["observed"] in ("decrease", "increase") + for c in checks): + return {**report, "status": "contradicted"} + if missing or evidence.get("status") != "available": + return {**report, "status": "unverifiable"} + if all(c["observed"] == c["expected"] or (c["expected"] == "unchanged" + and c["observed"] == "no_clear_change") + for c in checks): + return {**report, "status": "explained"} + return {**report, "status": "unsupported"} diff --git a/src/inferrsi/harness/optimization.py b/src/inferrsi/harness/optimization.py index bbd43922..6d98aa8e 100644 --- a/src/inferrsi/harness/optimization.py +++ b/src/inferrsi/harness/optimization.py @@ -221,6 +221,15 @@ def _settle(root, plan, state, pending, execution, assess): _load(root) if candidates.identity(directory / plan["candidate_name"]) != pending["candidate_sha256"]: raise ValueError("Submitted candidate changed during evaluation") + bound = pending.get("mechanism_sha256") + if bound is not None and not ( + digest(pending.get("mechanism")) == bound + and (directory / "mechanism.json").is_file() + and not (directory / "mechanism.json").is_symlink() + and digest(json.loads((directory / "mechanism.json").read_text(encoding="utf-8"))) + == bound + ): + raise ValueError("Declared mechanism changed during evaluation") except (OSError, ValueError, TypeError, KeyError) as error: state["integrity_error"] = str(error) try: @@ -228,7 +237,11 @@ def _settle(root, plan, state, pending, execution, assess): raise ValueError(state["integrity_error"]) if execution.get("coordinator_error"): raise ValueError(execution["coordinator_error"]) - assessment = assess(root, directory, plan, execution) + # Pass the declaration held in memory since before execution; the candidate + # can write files during execution, including state.json and attempt copies. + assessment = (assess(root, directory, plan, execution, mechanism=pending["mechanism"]) + if pending.get("mechanism") is not None + else assess(root, directory, plan, execution)) if not isinstance(assessment, dict) or assessment.get("status") not in ( "completed", "candidate_failure", "infrastructure_failure" ): @@ -287,7 +300,8 @@ def _settle(root, plan, state, pending, execution, assess): def evaluate_round(root: Path, *, candidate: Path | None, hypothesis: str, - usage: Path | None, execute, assess, final=False, reason=None) -> dict: + usage: Path | None, execute, assess, final=False, reason=None, + mechanism: dict | None = None) -> dict: with _locked(root): plan, state = _load(root) _active(state, reserve=0 if final else plan["final_reserve_seconds"]) @@ -315,10 +329,19 @@ def evaluate_round(root: Path, *, candidate: Path | None, hypothesis: str, (directory / "agent-usage.json").write_bytes(usage_bytes) fingerprint = candidates.retain(source, directory / plan["candidate_name"], plan.get("candidate_includes")) + if final and state["best"]["kind"] == "candidate": + # The independent final rechecks the mechanism bound to the selected round. + mechanism = next((r.get("mechanism") for r in state["rounds"] + if r["candidate_id"] == state["best"]["candidate_id"]), None) + mechanism_sha256 = None + if mechanism is not None: + atomic_write_json(directory / "mechanism.json", mechanism) + mechanism_sha256 = digest(mechanism) pending = {"kind": "final" if final else "round", "path": relative, "candidate_id": state["best"].get("candidate_id", "baseline") if final else f"candidate-{count + 1:03d}", "hypothesis": hypothesis, "observations": observations, + "mechanism": mechanism, "mechanism_sha256": mechanism_sha256, "candidate_sha256": fingerprint, "contexts": list(state["contexts"]), "started_at": time.time()} state["pending"] = pending @@ -366,4 +389,11 @@ def recover(root: Path, assess) -> dict: if not path.is_file(): raise ValueError("Execution record unavailable, inspect cleanup before recovery") execution = json.loads(path.read_text(encoding="utf-8")) + if pending.get("mechanism") is not None or pending.get("mechanism_sha256") is not None: + # The candidate ran with write access to every file that could hold the + # declaration after submission, so an interrupted attempt cannot prove which + # declaration was bound. Settle it without one: the explanation is reported as + # not declared and, under require_explanation, the round is not adopted. + pending = {**pending, "mechanism": None, "mechanism_sha256": None, + "mechanism_unverified_after_recovery": True} return _settle(root, plan, state, pending, execution, assess) diff --git a/src/inferrsi/harness/optimization_session.py b/src/inferrsi/harness/optimization_session.py index 9e7975c2..a2ce3852 100644 --- a/src/inferrsi/harness/optimization_session.py +++ b/src/inferrsi/harness/optimization_session.py @@ -4,6 +4,7 @@ projects and all subsequent operations use Harness and the retained inputs. """ +import json import os import runpy import shlex @@ -44,13 +45,13 @@ def _session_execute(root, directory, plan, deadline): None, environment=environment) -def _session_assess(root, directory, plan, execution): +def _session_assess(root, directory, plan, execution, mechanism=None): from inferrsi.harness.task_protocol import protocol if plan["task_source"] == "project": from inferrsi.harness.project import assess - return assess(directory, plan["evaluation_settings"], execution) + return assess(directory, plan["evaluation_settings"], execution, mechanism=mechanism) return protocol(root / "frozen").assess(directory, plan["evaluation_settings"], execution) @@ -115,7 +116,7 @@ def _runtime_identity(): "infra/processes.py", "infra/artifacts.py", "harness/scenarios/external.py", "harness/atomic_json.py", "harness/usage.py", "harness/agent_sessions.py", "harness/task_protocol.py", - "harness/candidates.py", "harness/project.py", + "harness/candidates.py", "harness/project.py", "harness/measurement.py", "harness/records/knowledge_capture.py", "apps/campaign.py", "knowledge/context_projection.py") return {name: sha256(root / name) for name in names} @@ -154,7 +155,8 @@ def session_command(args, *, task_source: Path | None = None, knowledge_reader=N identity(repo, args.include) # Reject invalid inputs before preparation writes. specification = {key: getattr(args, key) for key in ( "metric_key", "metric_unit", "quality_key", "scope", "direction", - "min_improvement_pct", "repetitions", "warmups")} + "min_improvement_pct", "repetitions", "warmups", "decision", + "require_explanation")} specification.update(measure_command=shlex.split(args.measure_command), quality_command=shlex.split(args.quality_command)) settings = prepare_project(root / "frozen", args.protocol.resolve(), @@ -197,7 +199,6 @@ def session_command(args, *, task_source: Path | None = None, knowledge_reader=N "Report summary.json and its curve. Import actual usage exports when available." ) return result - import json plan = json.loads((root / "plan.json").read_text(encoding="utf-8")) if plan["runtime_identity"] != _runtime_identity(): @@ -215,11 +216,19 @@ def session_command(args, *, task_source: Path | None = None, knowledge_reader=N if args.session_action == "recover": return optimization.recover(root, _session_assess) final = args.session_action == "final" + mechanism = None + if not final and getattr(args, "mechanism_file", None) is not None: + if plan.get("task_source") != "project": + raise ValueError("Mechanism checks apply to project campaign rounds only") + from inferrsi.harness.measurement import validate_mechanism + + mechanism = validate_mechanism( + json.loads(Path(args.mechanism_file).read_text(encoding="utf-8"))) return optimization.evaluate_round( root, candidate=None if final else args.candidate, hypothesis="Independent final of the selected candidate" if final else args.hypothesis, usage=args.usage_export, execute=_session_execute, assess=_session_assess, final=final, - reason=getattr(args, "reason", None), + reason=getattr(args, "reason", None), mechanism=mechanism, ) diff --git a/src/inferrsi/harness/project.py b/src/inferrsi/harness/project.py index fa0da190..2ccf148b 100644 --- a/src/inferrsi/harness/project.py +++ b/src/inferrsi/harness/project.py @@ -11,6 +11,9 @@ from statistics import median from ..infra.artifacts import sha256 +from .measurement import MIN_PAIRS, explain, paired_statistics, profile_evidence, validate_mechanism + +DECISIONS = ("point", "paired") def validate(settings: dict) -> None: @@ -34,6 +37,12 @@ def validate(settings: dict) -> None: value = settings["min_improvement_pct"] if type(value) not in (int, float) or not math.isfinite(value) or value < 0: raise ValueError("An explicit finite nonnegative improvement gate is required") + if settings.get("decision", "point") not in DECISIONS: + raise ValueError("Decision must be point or paired") + if settings.get("decision") == "paired" and settings["repetitions"] < MIN_PAIRS: + raise ValueError(f"A paired decision needs at least {MIN_PAIRS} repetitions") + if type(settings.get("require_explanation", False)) is not bool: + raise ValueError("require_explanation must be a boolean") def prepare(destination: Path, protocol: Path, instructions: Path, settings: dict) -> dict: @@ -116,10 +125,16 @@ def inspect_step(directory: Path, settings: dict, definition: dict, execution: d type(value) not in (int, float) or not math.isfinite(value) or value <= 0 ): return {**failure, "reason": "invalid_metric", "evidence_files": evidence} - return {"status": "completed", "value": value, "evidence_files": evidence} + profile = {key: document[key] for key in ("breakdown", "counters") if key in document} + return {"status": "completed", "value": value, "profile": profile, + "evidence_files": evidence} + -def assess(directory: Path, settings: dict, execution: dict) -> dict: + +def assess(directory: Path, settings: dict, execution: dict, *, + mechanism: dict | None = None) -> dict: + """Judge one attempt; ``mechanism`` is the declaration bound before execution.""" evidence = {} def result(status, reason, **fields): @@ -134,6 +149,7 @@ def result(status, reason, **fields): or any(c.get("verified") is not True for c in checks)): return result("infrastructure_failure", "incomplete_execution") samples = {"baseline": [], "candidate": []} + profiles = {"baseline": [], "candidate": []} candidate_quality = None for definition, record in zip(expected, actual): if record.get("id") != definition["id"]: @@ -149,6 +165,7 @@ def result(status, reason, **fields): candidate_quality = True if definition["phase"] == "measurement": samples[definition["side"]].append(checked["value"]) + profiles[definition["side"]].append(checked["profile"]) if execution.get("status") != "completed" or len(actual) != len(expected): return result("infrastructure_failure", "incomplete_execution") before, after = median(samples["baseline"]), median(samples["candidate"]) @@ -158,15 +175,53 @@ def result(status, reason, **fields): speedup = before / after if settings["direction"] == "minimize" else after / before if not all(math.isfinite(value) for value in (improvement, speedup)): return result("infrastructure_failure", "nonfinite_derived_metric") - accepted = improvement > 0 and improvement >= settings["min_improvement_pct"] + point_accepted = improvement > 0 and improvement >= settings["min_improvement_pct"] + paired = paired_statistics(samples["baseline"], samples["candidate"], + direction=settings["direction"], + min_improvement_pct=settings["min_improvement_pct"], + point_gate_passed=point_accepted) + decision = settings.get("decision", "point") + verdict, reason = (("PASS", "fixed_checks_satisfied") if point_accepted + else ("FAIL", "performance_gate_not_met")) + if decision == "paired": + relation = paired.get("gate_relation") + if paired.get("status") != "available": + verdict, reason = "INCONCLUSIVE", "paired_statistics_unavailable" + elif relation == "above_gate": + verdict, reason = "PASS", "paired_interval_above_gate" + elif relation == "straddles_gate": + verdict, reason = "FAIL", "within_noise" + else: + verdict, reason = "FAIL", "performance_gate_not_met" + try: + mechanism = None if mechanism is None else validate_mechanism(mechanism) + mechanism_error = None + except ValueError as error: + mechanism, mechanism_error = None, str(error) + reference = [item["region"] for item in (mechanism or {}).get("expect", []) + if "region" in item and item["direction"] == "unchanged"] + observed = profile_evidence(profiles["baseline"], profiles["candidate"], + reference_regions=reference) + explanation = explain(mechanism, observed) + if mechanism_error: + explanation = {"status": "unverifiable", "reason": mechanism_error} + if settings.get("require_explanation", False) and verdict == "PASS": + if explanation["status"] in ("contradicted", "unsupported"): + verdict, reason = "FAIL", "explanation_" + explanation["status"] + elif explanation["status"] != "explained": + verdict, reason = "INCONCLUSIVE", "explanation_" + explanation["status"] + accepted = verdict == "PASS" return result( - "completed", "fixed_checks_satisfied" if accepted else "performance_gate_not_met", - verdict="PASS" if accepted else "FAIL", score=after, baseline_score=before, + "completed", reason, + verdict=verdict, score=after, baseline_score=before, eligible=accepted, quality_passed=True, improvement_pct=improvement, speedup=speedup, metrics={"name": settings["metric_key"], "unit": settings["metric_unit"], "direction": settings["direction"], "baseline": before, "candidate": after, "samples": samples}, scope=settings["scope"], statistical_significance="not_established", + paired_statistics=paired, decision=decision, + require_explanation=settings.get("require_explanation", False), + profile_evidence=observed, explanation=explanation, quality_contract="Only the supplied frozen quality script and its boolean result", environment_qualification="Caller-managed; no automatic hardware isolation claim", ) diff --git a/tests/test_harness_measurement.py b/tests/test_harness_measurement.py new file mode 100644 index 00000000..2bce671c --- /dev/null +++ b/tests/test_harness_measurement.py @@ -0,0 +1,292 @@ +"""Paired report-only statistics; every sample below is a synthetic software control.""" + +import json +import math +import random +from statistics import median + +import pytest + +from inferrsi.harness import optimization_session +from inferrsi.harness.measurement import ( + bootstrap_median_interval, + explain, + gate_speedup, + paired_speedups, + paired_statistics, + profile_evidence, + validate_mechanism, +) + +BASE = [10.0, 10.1, 9.9, 10.05, 9.95] + + +def report(baseline, candidate, *, direction="minimize", gate_pct=1.0, passed=True, **kwargs): + return paired_statistics(baseline, candidate, direction=direction, + min_improvement_pct=gate_pct, point_gate_passed=passed, **kwargs) + + +def point_gate(baseline, candidate, gate_pct): + """The current project.assess rule: independent medians, minimize direction.""" + improvement = (median(baseline) - median(candidate)) / median(baseline) * 100 + return improvement > 0 and improvement >= gate_pct + + +def test_gate_speedup_matches_improvement_pct_definition(): + assert gate_speedup(10, "maximize") == pytest.approx(1.10) + # Minimize: a 10% lower latency is a 1/0.9 speedup, not 1.10. + assert gate_speedup(10, "minimize") == pytest.approx(1 / 0.9) + assert gate_speedup(0, "minimize") == 1 + assert gate_speedup(100, "minimize") == math.inf + with pytest.raises(ValueError): + gate_speedup(1, "sideways") + + +def test_pair_direction_and_input_validation(): + assert paired_speedups([10, 20], [5, 10], "minimize") == [2.0, 2.0] + assert paired_speedups([10, 20], [5, 10], "maximize") == [0.5, 0.5] + for baseline, candidate in (([1, 2], [1]), ([1, 0], [1, 1]), ([1, math.nan], [1, 1])): + with pytest.raises(ValueError): + paired_speedups(baseline, candidate, "minimize") + + +@pytest.mark.parametrize("baseline,candidate", [ + ([10, 10, 10], [9, 9]), # lost pair + ([10, 0, 10], [9, 9, 9]), # zero sample + ([10, math.nan, 10], [9, 9, 9]), # non-finite sample +]) +def test_invalid_samples_report_unavailable_instead_of_raising(baseline, candidate): + result = report(baseline, candidate) + assert result["status"] == "unavailable" and result["interval"] is None + assert result["report_only"] is True + + +@pytest.mark.parametrize("baseline,candidate", [ + ([1e-300, 10, 10], [1e300, 5, 5]), # paired ratio underflows to zero + ([1e300, 10, 10], [1e-300, 5, 5]), # paired ratio overflows to infinity +]) +def test_ratio_overflow_reports_unavailable_and_stays_strict_json(baseline, candidate): + result = report(baseline, candidate) + assert result["status"] == "unavailable" + json.dumps(result, allow_nan=False) + + +def test_unreachable_minimize_gate_is_serializable_not_infinite(): + result = report(BASE, [5.0, 5.1, 4.9, 5.02, 4.98], gate_pct=100, passed=False) + assert result["gate_speedup"] is None and result["gate_relation"] == "gate_unreachable" + assert result["agrees_with_point_gate"] is True + json.dumps(result, allow_nan=False) + + +@pytest.mark.parametrize("pairs", [2, 4]) +def test_too_few_repetitions_report_no_interval(pairs): + # Below five pairs the simulated coverage of the nominal 95% interval is ~75%. + result = report(BASE[:pairs], [8.0, 8.1, 7.9, 8.05][:pairs]) + assert result["status"] == "insufficient_pairs" and result["pairs"] == pairs + assert result["interval"] is None and result["agrees_with_point_gate"] is None + + +def test_interval_is_deterministic_for_the_recorded_seed(): + values = [1.02, 0.99, 1.05, 1.01, 0.97, 1.03] + assert bootstrap_median_interval(values, seed=7) == bootstrap_median_interval(values, seed=7) + result = report([10.0] * 6, [10 / v for v in values]) + assert result["seed"] == 0 and result["resamples"] == 10_000 + + +def test_consistent_real_gain_is_above_gate(): + baseline = [10.0, 10.1, 9.9, 10.05, 9.95] + candidate = [8.0, 8.1, 7.9, 8.02, 7.98] + result = report(baseline, candidate, gate_pct=5) + low, high = result["interval"] + assert low > result["gate_speedup"] and low <= result["median_speedup"] <= high + assert result["gate_relation"] == "above_gate" and result["excludes_no_change"] is True + assert result["agrees_with_point_gate"] is True + + +def test_aa_noise_straddles_gate_even_when_point_estimate_clears_it(): + # A/A control: identical code, noisy alternation. Independent medians differ by + # more than the 1% gate, so the current point gate passes; the pairs do not. + baseline = [10.0, 10.6, 9.6, 10.4, 9.9] + candidate = [9.7, 10.2, 9.8, 10.1, 9.75] + assert point_gate(baseline, candidate, 1) + result = report(baseline, candidate, gate_pct=1, passed=True) + assert result["gate_relation"] == "straddles_gate" + assert result["excludes_no_change"] is False + assert result["agrees_with_point_gate"] is False + + +def test_pairing_cancels_shared_drift(): + # Both sides slow down 2% per repetition; the true speedup is 1.25 in every pair. + drift = [1.0 + 0.02 * k for k in range(6)] + result = report([10.0 * d for d in drift], [8.0 * d for d in drift], gate_pct=10) + assert result["pair_speedups"] == pytest.approx([1.25] * 6) + assert result["interval"] == pytest.approx([1.25, 1.25]) + assert result["baseline_relative_mad"] > 0.02 + + +def test_alternating_order_penalty_does_not_create_a_gain(): + # A/A with a 5% penalty for whichever side runs second in a repetition. + baseline, candidate = [], [] + for k in range(8): + first, second = 10.0, 10.5 + baseline.append(first if k % 2 == 0 else second) + candidate.append(second if k % 2 == 0 else first) + result = report(baseline, candidate, gate_pct=1, passed=False) + assert result["gate_relation"] != "above_gate" + + +def test_clear_regression_is_below_gate(): + result = report(BASE, [12.0, 12.1, 11.9, 12.02, 11.98], gate_pct=1, passed=False) + assert result["gate_relation"] == "below_gate" + assert result["agrees_with_point_gate"] is True + + +@pytest.mark.parametrize("noise", ["lognormal", "heavy_tail"]) +def test_aa_simulation_paired_false_gains_stay_rare_and_below_point_gate(noise): + """Ablation: independent medians vs paired interval on simulated A/A campaigns. + + Each campaign has 10 repetitions with a shared random-walk drift, per-sample + 3% noise and, for heavy_tail, a 10% chance of a 1.3x stall. Fixed settings: + 200 campaigns, gate 1%, seed 20261115. + """ + rng = random.Random(20261115) + point_false = paired_false = 0 + campaigns = 200 + for index in range(campaigns): + level, baseline, candidate = 10.0, [], [] + for _ in range(10): + level *= math.exp(rng.gauss(0, 0.02)) + + def sample(): + value = level * math.exp(rng.gauss(0, 0.03)) + if noise == "heavy_tail" and rng.random() < 0.1: + value *= 1.3 + return value + + baseline.append(sample()) + candidate.append(sample()) + point_false += point_gate(baseline, candidate, 1) + result = report(baseline, candidate, gate_pct=1, resamples=1000, seed=index) + paired_false += result["gate_relation"] == "above_gate" + assert paired_false / campaigns <= 0.05 + assert paired_false < point_false + + +def test_measurement_module_is_part_of_runtime_identity(): + assert "harness/measurement.py" in optimization_session._runtime_identity() + + +@pytest.mark.parametrize("resamples", [0, -1, 1.5]) +def test_invalid_resample_count_reports_unavailable(resamples): + result = report(BASE, [8.0, 8.1, 7.9, 8.02, 7.98], resamples=resamples) + assert result["status"] == "unavailable" and result["interval"] is None + + +def profiles(rows, jitter=None): + """Per-repetition profiles; jitter scales every region of a step together.""" + jitter = jitter or [1.0] * len(rows) + return [{"breakdown": {k: v * g for k, v in r["breakdown"].items()}, + "counters": r.get("counters", {"launches": 10})} for r, g in zip(rows, jitter)] + + +STEP_B = {"breakdown": {"gemm": 6.0, "norm": 2.0}, "counters": {"launches": 12, "bytes": 900}} +STEP_C = {"breakdown": {"gemm": 6.0, "norm": 1.0}, "counters": {"launches": 8, "bytes": 900}} +JITTER_B = [1.0, 1.08, 0.93, 1.05, 0.97] +JITTER_C = [1.06, 0.94, 1.02, 0.95, 1.09] + + +def test_reference_regions_cancel_device_wide_jitter(): + evidence = profile_evidence(profiles([STEP_B] * 5, JITTER_B), + profiles([STEP_C] * 5, JITTER_C), reference_regions=["gemm"]) + assert evidence["status"] == "available" and evidence["reference_regions"] == ["gemm"] + norm = evidence["regions"]["norm"] + assert norm["basis"] == "normalized_to_reference_regions" + assert norm["normalized"]["ratio_median"] == pytest.approx(0.5) + assert norm["observed"] == "decrease" + assert evidence["share_based_speedup"]["median"] == pytest.approx(8 / 7) + assert evidence["share_based_speedup"]["interval"] == pytest.approx([8 / 7, 8 / 7]) + assert evidence["counters"]["launches"]["observed"] == "decrease" + assert evidence["counters"]["bytes"]["observed"] == "no_clear_change" + # Without a reference the same region is only compared raw, jitter included. + raw = profile_evidence(profiles([STEP_B] * 5, JITTER_B), profiles([STEP_C] * 5, JITTER_C)) + assert raw["regions"]["norm"]["basis"] == "raw" and raw["share_based_speedup"] is None + + +def test_profile_evidence_absent_or_unpaired(): + assert profile_evidence([{}] * 5, [{}] * 5)["status"] == "absent" + assert profile_evidence([STEP_B], [STEP_C, STEP_C])["status"] == "unavailable" + few = profile_evidence(profiles([STEP_B] * 2), profiles([STEP_C] * 2)) + assert few["counters"]["launches"]["observed"] == "insufficient_pairs" + + +MECHANISM = {"mechanism": "Fuse the norm into the GEMM epilogue", + "expect": [{"region": "norm", "direction": "decrease"}, + {"region": "gemm", "direction": "unchanged"}, + {"counter": "launches", "direction": "decrease"}]} + + +@pytest.mark.parametrize("expect,status", [ + (MECHANISM["expect"], "explained"), + ([{"region": "norm", "direction": "increase"}], "contradicted"), + ([{"counter": "bytes", "direction": "decrease"}], "unsupported"), + ([{"counter": "l2_hits", "direction": "increase"}], "unverifiable"), + ([{"counter": "launches", "direction": "decrease"}, + {"counter": "bytes", "direction": "unchanged"}], "explained"), +]) +def test_explain_classifies_declared_mechanisms(expect, status): + mechanism = validate_mechanism({"mechanism": "m", "expect": expect}) + reference = [e["region"] for e in expect if "region" in e and e["direction"] == "unchanged"] + evidence = profile_evidence(profiles([STEP_B] * 5, JITTER_B), + profiles([STEP_C] * 5, JITTER_C), reference_regions=reference) + assert explain(mechanism, evidence)["status"] == status + assert explain(None, evidence) == {"status": "not_declared"} + + +@pytest.mark.parametrize("document", [ + {"mechanism": "", "expect": [{"region": "a", "direction": "decrease"}]}, + {"mechanism": "m", "expect": []}, + {"mechanism": "m", "expect": [{"region": "a", "direction": "faster"}]}, + {"mechanism": "m", "expect": [{"region": "a", "counter": "b", "direction": "decrease"}]}, + {"mechanism": "m", "expect": [{"region": "a", "direction": "unchanged"}]}, + {"mechanism": "m", "expect": [{"region": "a", "direction": "decrease"}], "extra": 1}, +]) +def test_invalid_mechanisms_are_rejected(document): + with pytest.raises(ValueError): + validate_mechanism(document) + + +def test_zero_baseline_counter_is_undefined_not_decisive(): + one = profile_evidence([{"counters": {"new_work": 0}}], [{"counters": {"new_work": 1}}]) + assert one["counters"]["new_work"]["observed"] == "insufficient_pairs" + mixed = profile_evidence([{"counters": {"w": v}} for v in (0, 100, 100, 100, 100)], + [{"counters": {"w": v}} for v in (1, 50, 50, 50, 50)]) + assert mixed["counters"]["w"]["observed"] == "undefined" + mechanism = validate_mechanism({"mechanism": "m", "expect": [ + {"counter": "w", "direction": "increase"}]}) + assert explain(mechanism, mixed)["status"] == "unverifiable" + + +def test_all_zero_candidate_breakdown_is_reported_not_raised(): + evidence = profile_evidence([{"breakdown": {"only": 2.0}}] * 5, + [{"breakdown": {"only": 0.0}}] * 5) + assert evidence["status"] == "available" + assert evidence["regions"]["only"]["candidate_share"] is None + assert evidence["regions"]["only"]["observed"] == "decrease" + + +def test_references_are_checked_raw_never_against_each_other(): + base = [{"breakdown": {"r1": 2.0, "r2": 2.0, "t": 2.0}}] * 5 + cand = [{"breakdown": {"r1": 1.0, "r2": 1.0, "t": 0.5}}] * 5 + evidence = profile_evidence(base, cand, reference_regions=["r1", "r2"]) + assert evidence["regions"]["r1"]["basis"] == "raw" + assert evidence["regions"]["r1"]["observed"] == "decrease" + mechanism = validate_mechanism({"mechanism": "m", "expect": [ + {"region": "r1", "direction": "unchanged"}, {"region": "r2", "direction": "unchanged"}, + {"region": "t", "direction": "decrease"}]}) + assert explain(mechanism, evidence)["status"] == "contradicted" + + +def test_duplicate_expectations_are_rejected(): + with pytest.raises(ValueError): + validate_mechanism({"mechanism": "m", "expect": [ + {"region": "a", "direction": "unchanged"}, {"region": "a", "direction": "decrease"}]}) diff --git a/tests/test_optimization_session.py b/tests/test_optimization_session.py index b1a052e8..c375e8d2 100644 --- a/tests/test_optimization_session.py +++ b/tests/test_optimization_session.py @@ -315,3 +315,14 @@ def test_runtime_identity_covers_migrated_process_domain_and_not_removed_executo assert "infra/executors/local.py" not in identity for name in identity: assert (root / name).is_file(), name + + +def test_mechanism_checks_are_rejected_for_packaged_task_sessions(session, tmp_path): + mechanism = tmp_path / "mechanism.json" + mechanism.write_text(json.dumps({"mechanism": "m", "expect": [ + {"region": "a", "direction": "decrease"}]}), encoding="utf-8") + with pytest.raises(SystemExit): + main(["benchmarks", "session", "round", "--session", str(session), + "--candidate", str(session / "work/candidate.py"), "--hypothesis", "h", + "--mechanism-file", str(mechanism)]) + assert not (session / "rounds").exists() or not any((session / "rounds").iterdir()) diff --git a/tests/test_project_explanation.py b/tests/test_project_explanation.py new file mode 100644 index 00000000..fbd4624c --- /dev/null +++ b/tests/test_project_explanation.py @@ -0,0 +1,258 @@ +"""Decision and mechanism switches through real campaign commands; values are synthetic.""" + +import json + +import pytest + +from inferrsi.cli import main +from inferrsi.harness.optimization import read_records + +# Device-wide factor applied to every timed value of one measurement call, cycling per +# call. It scales all regions together, as clocks or temperature would. +JITTER = [1.0, 1.06, 0.95, 1.04, 0.97, 1.05, 0.96, 1.03, 0.98, 1.02, 0.94, 1.07] + +BASELINE = {"metric": 10.0, "breakdown": {"attention": 4.0, "norm": 2.0, "other": 4.0}, + "counters": {"kernel_launches": 30, "dram_bytes": 1000}} + + +@pytest.fixture +def project(tmp_path): + repo = tmp_path / "project" + repo.mkdir() + (repo / "engine.py").write_text("def compute(x):\n return x * 2\n", encoding="utf-8") + (repo / "config.json").write_text(json.dumps(BASELINE), encoding="utf-8") + protocol = tmp_path / "protocol" + protocol.mkdir() + calls = tmp_path / "calls.txt" + (protocol / "measure.py").write_text( + "import json, sys\nfrom pathlib import Path\n" + f"JITTER = {JITTER!r}\nCALLS = Path({str(calls)!r})\n" + "k = int(CALLS.read_text()) if CALLS.exists() else 0\n" + "CALLS.write_text(str(k + 1))\n" + "g = JITTER[k % len(JITTER)]\n" + "c = json.loads((Path(sys.argv[1]) / 'config.json').read_text())\n" + "out = {'metric': c['metric'] * g,\n" + " 'breakdown': {n: v * g for n, v in c['breakdown'].items()},\n" + " 'counters': c['counters']}\n" + "Path(sys.argv[2]).write_text(json.dumps(out))\n", + encoding="utf-8", + ) + (protocol / "quality.py").write_text( + "import json, runpy, sys\nfrom pathlib import Path\n" + "compute = runpy.run_path(str(Path(sys.argv[1])/'engine.py'))['compute']\n" + "Path(sys.argv[2]).write_text(json.dumps({'correct':compute(16)==32}))\n", + encoding="utf-8", + ) + instructions = tmp_path / "task.md" + instructions.write_text("Synthetic fixture; preserve compute outputs.", encoding="utf-8") + return repo, protocol, instructions, tmp_path / "campaign" + + +def start(project, capsys, *extra, repetitions=5): + repo, protocol, instructions, root = project + code = main([ + "campaign", "start", "--directory", str(root), "--repo", str(repo), + "--include", "engine.py", "--include", "config.json", "--protocol", str(protocol), + "--instructions", str(instructions), "--measure-command", + '"{python}" "{protocol}/measure.py" "{candidate}" "{result}"', + "--quality-command", '"{python}" "{protocol}/quality.py" "{candidate}" "{result}"', + "--metric-key", "metric", "--metric-unit", "ms", "--direction", "minimize", + "--quality-key", "correct", "--scope", "synthetic-software-control", + "--repetitions", str(repetitions), "--warmups", "1", "--min-improvement-pct", "1", + "--rounds", "4", "--round-mode", "up_to", "--max-minutes", "5", + "--evaluation-timeout-seconds", "20", *extra, + ]) + capsys.readouterr() + return code + + +def submit(project, capsys, config, mechanism=None): + repo, _, _, root = project + (repo / "config.json").write_text(json.dumps(config), encoding="utf-8") + args = ["campaign", "round", "--directory", str(root), "--candidate", str(repo), + "--hypothesis", "Synthetic candidate"] + if mechanism is not None: + path = root.parent / "mechanism.json" + path.write_text(json.dumps(mechanism), encoding="utf-8") + args += ["--mechanism-file", str(path)] + assert main(args) == 0 + return json.loads(capsys.readouterr().out)["latest_observation"]["assessment"] + + +FUSED = {"metric": 8.0, "breakdown": {"attention": 4.0, "norm": 0.0, "other": 4.0}, + "counters": {"kernel_launches": 20, "dram_bytes": 800}} +FUSION = {"mechanism": "Fuse the norm into the attention epilogue, removing one HBM pass", + "expect": [{"region": "norm", "direction": "decrease"}, + {"region": "attention", "direction": "unchanged"}, + {"counter": "kernel_launches", "direction": "decrease"}, + {"counter": "dram_bytes", "direction": "decrease"}]} + + +def test_confirmed_mechanism_is_adopted_and_rechecked_at_final(project, capsys): + assert start(project, capsys, "--require-explanation") == 0 + assessment = submit(project, capsys, FUSED, FUSION) + assert assessment["verdict"] == "PASS" and assessment["eligible"] is True + assert assessment["explanation"]["status"] == "explained" + evidence = assessment["profile_evidence"] + assert evidence["counters"]["dram_bytes"]["raw"]["ratio_median"] == pytest.approx(0.8) + # Device-wide jitter scales every region of a step. Raw cross-run ratios carry it, + # while ratios to the declared-unchanged region and its share of the step cancel it. + assert evidence["reference_regions"] == ["attention"] + norm = evidence["regions"]["norm"] + assert norm["basis"] == "normalized_to_reference_regions" and norm["observed"] == "decrease" + other = evidence["regions"]["other"]["normalized"] + assert other["ratio_median"] == pytest.approx(1.0) and other["interval"] == pytest.approx( + [1.0, 1.0]) + # The declared reference itself is checked raw, so device jitter keeps it unclear. + assert evidence["regions"]["attention"]["basis"] == "raw" + share = evidence["share_based_speedup"] + assert share["median"] == pytest.approx(10 / 8) and share["interval"] == pytest.approx( + [10 / 8, 10 / 8]) + attention = evidence["regions"]["attention"] # raw medians alone carry the jitter + assert attention["baseline_median"] != pytest.approx(attention["candidate_median"]) + assert main(["campaign", "final", "--directory", str(project[3]), + "--reason", "Synthetic search done"]) == 0 + capsys.readouterr() + _, state, _ = read_records(project[3]) + assert state["final"]["assessment"]["explanation"]["status"] == "explained" + assert state["final"]["assessment"]["verdict"] == "PASS" + + +@pytest.mark.parametrize("mechanism,status,verdict", [ + ({"mechanism": "Claims the norm got slower", "expect": [ + {"region": "norm", "direction": "increase"}]}, "contradicted", "FAIL"), + ({"mechanism": "Claims attention got faster", "expect": [ + {"region": "attention", "direction": "decrease"}]}, "unsupported", "FAIL"), + ({"mechanism": "Names a counter that was never measured", "expect": [ + {"counter": "l2_hit_rate", "direction": "increase"}]}, "unverifiable", "INCONCLUSIVE"), + (None, "not_declared", "INCONCLUSIVE"), +]) +def test_unexplained_gain_is_not_adopted_and_search_continues( + project, capsys, mechanism, status, verdict, +): + assert start(project, capsys, "--require-explanation") == 0 + assessment = submit(project, capsys, FUSED, mechanism) + assert assessment["explanation"]["status"] == status + assert assessment["verdict"] == verdict and assessment["eligible"] is False + _, state, _ = read_records(project[3]) + assert state["best"]["kind"] == "baseline" and state["state"] == "active" + assessment = submit(project, capsys, FUSED, FUSION) + assert assessment["verdict"] == "PASS" + + +def test_explanation_is_report_only_without_the_switch(project, capsys): + assert start(project, capsys) == 0 + contradicted = {"mechanism": "Wrong story", "expect": [ + {"counter": "dram_bytes", "direction": "increase"}]} + assessment = submit(project, capsys, FUSED, contradicted) + assert assessment["verdict"] == "PASS" and assessment["eligible"] is True + assert assessment["explanation"]["status"] == "contradicted" + assert assessment["require_explanation"] is False + + +def test_paired_decision_fails_within_noise_without_blocking(project, capsys): + assert start(project, capsys, "--decision", "paired") == 0 + nominal = {**BASELINE, "metric": 9.85} # 1.5% nominal gain inside ±6% device jitter + assessment = submit(project, capsys, nominal) + assert assessment["paired_statistics"]["gate_relation"] == "straddles_gate" + assert assessment["verdict"] == "FAIL" and assessment["reason"] == "within_noise" + _, state, _ = read_records(project[3]) + assert state["state"] == "active" and state["best"]["kind"] == "baseline" + assessment = submit(project, capsys, FUSED) + assert assessment["verdict"] == "PASS" + assert assessment["reason"] == "paired_interval_above_gate" + + +def test_paired_decision_requires_enough_repetitions(project, capsys): + with pytest.raises(SystemExit): + start(project, capsys, "--decision", "paired", repetitions=3) + + +def test_invalid_mechanism_is_rejected_before_execution(project, capsys): + assert start(project, capsys, "--require-explanation") == 0 + with pytest.raises(SystemExit): + submit(project, capsys, FUSED, {"mechanism": "x", "expect": [ + {"region": "norm", "direction": "faster"}]}) + _, state, _ = read_records(project[3]) + assert state["rounds"] == [] + + +@pytest.mark.parametrize("target", ["state", "copy"]) +def test_rewriting_the_bound_declaration_is_detected(project, capsys, target): + """A candidate that rewrites state.json or the attempt copy cannot change the verdict.""" + rewrite_state = ( + " state = parent / 'state.json'\n" + " if state.is_file():\n" + " data = json.loads(state.read_text())\n" + " if data.get('pending') and data['pending'].get('mechanism'):\n" + " data['pending']['mechanism'] = FORGED\n" + " state.write_text(json.dumps(data))\n") + rewrite_copy = ( + " copy = parent / 'mechanism.json'\n" + " if copy.is_file():\n" + " copy.write_text(json.dumps(FORGED))\n") + (project[0] / "engine.py").write_text( + "import json\nfrom pathlib import Path\n" + "FORGED = {'mechanism': 'forged', 'expect': [\n" + " {'counter': 'dram_bytes', 'direction': 'decrease'}]}\n" + "for parent in Path.cwd().parents:\n" + + (rewrite_state if target == "state" else rewrite_copy) + + "def compute(x):\n return x * 2\n", encoding="utf-8") + assert start(project, capsys, "--require-explanation") == 0 + wrong = {"mechanism": "Claims the norm got slower", "expect": [ + {"region": "norm", "direction": "increase"}]} + repo, _, _, root = project + (repo / "config.json").write_text(json.dumps(FUSED), encoding="utf-8") + path = root.parent / "mechanism.json" + path.write_text(json.dumps(wrong), encoding="utf-8") + main(["campaign", "round", "--directory", str(root), "--candidate", str(repo), + "--hypothesis", "Synthetic candidate", "--mechanism-file", str(path)]) + view = json.loads(capsys.readouterr().out) + state = json.loads((root / "state.json").read_text()) + assert state["best"]["kind"] == "baseline" + if target == "copy": + # The sealed copy no longer matches its digest: an integrity failure blocks the + # campaign and nothing is adopted. + assert view["state"] == "blocked" + assert "mechanism changed" in view["integrity_error"] + else: + # state.json is rewritten wholesale at settle from the in-memory record, so the + # forged pending entry is discarded and the original declaration is judged. + assert view["state"] == "active" + record = state["rounds"][-1] + assert record["mechanism"]["mechanism"] == "Claims the norm got slower" + assert record["assessment"]["explanation"]["status"] == "contradicted" + + +def test_recovered_attempt_does_not_trust_a_rewritable_declaration(project, capsys, monkeypatch): + """After an interruption the bound declaration is unprovable, so it is not adopted.""" + from inferrsi.harness import optimization_session + + original = optimization_session._session_execute + + def forge_then_crash(root, directory, plan, deadline): + original(root, directory, plan, deadline) + forged = {"mechanism": "forged", "expect": [ + {"counter": "dram_bytes", "direction": "decrease"}]} + from inferrsi.infra.artifacts import digest + state = json.loads((root / "state.json").read_text()) + state["pending"].update(mechanism=forged, mechanism_sha256=digest(forged)) + (root / "state.json").write_text(json.dumps(state)) + (directory / "mechanism.json").write_text(json.dumps(forged)) + raise KeyboardInterrupt("coordinator interrupted after the candidate ran") + + assert start(project, capsys, "--require-explanation") == 0 + monkeypatch.setattr(optimization_session, "_session_execute", forge_then_crash) + wrong = {"mechanism": "Claims the norm got slower", "expect": [ + {"region": "norm", "direction": "increase"}]} + with pytest.raises(KeyboardInterrupt): + submit(project, capsys, FUSED, wrong) + monkeypatch.setattr(optimization_session, "_session_execute", original) + assert main(["campaign", "recover", "--directory", str(project[3])]) == 0 + capsys.readouterr() + _, state, _ = read_records(project[3]) + record = state["rounds"][-1] + assert record["mechanism_unverified_after_recovery"] is True + assert record["assessment"]["explanation"]["status"] == "not_declared" + assert record["assessment"]["eligible"] is False + assert state["best"]["kind"] == "baseline" and state["state"] == "active" diff --git a/tests/test_project_optimization.py b/tests/test_project_optimization.py index 83df6636..6e42ef83 100644 --- a/tests/test_project_optimization.py +++ b/tests/test_project_optimization.py @@ -423,3 +423,63 @@ def test_final_reserve_blocks_search_but_allows_final(project, capsys, monkeypat assert main(["campaign", "final", "--directory", str(project[3]), "--reason", "Search time exhausted"]) == 0 capsys.readouterr() + + +@pytest.mark.parametrize("direction,value", [("maximize", 20), ("minimize", 5)]) +def test_assessment_reports_paired_statistics_without_changing_the_verdict( + project, capsys, direction, value, +): + start(project, capsys, direction=direction, extra=("--repetitions", "5")) + code, result = submit(project, capsys, value) + assert code == 0 and result["best"]["kind"] == "candidate" + assessment = result["latest_observation"]["assessment"] + assert assessment["verdict"] == "PASS" and assessment["speedup"] == 2 + assert assessment["statistical_significance"] == "not_established" + paired = assessment["paired_statistics"] + assert paired["report_only"] is True and paired["pairs"] == 5 + assert paired["status"] == "available" + assert paired["interval"] == pytest.approx([2.0, 2.0]) + assert paired["gate_relation"] == "above_gate" + assert paired["agrees_with_point_gate"] is True + assert main(["campaign", "final", "--directory", str(project[3]), + "--reason", "Synthetic search done"]) == 0 + capsys.readouterr() + _, state, summary = read_records(project[3]) + final = state["final"]["assessment"] + assert final["verdict"] == "PASS" and summary["optimization_speedup"] == 2 + assert final["paired_statistics"]["gate_relation"] == "above_gate" + + +def test_unreachable_gate_keeps_the_original_fail_and_strict_json_records(project, capsys): + # A minimize gate of 100% has no finite speedup; the report must not block the FAIL. + start(project, capsys, direction="minimize", + extra=("--repetitions", "5", "--min-improvement-pct", "100")) + code, result = submit(project, capsys, 5) + assert code == 0 and result["best"]["kind"] == "baseline" + assessment = result["latest_observation"]["assessment"] + assert assessment["verdict"] == "FAIL" and assessment["eligible"] is False + paired = assessment["paired_statistics"] + assert paired["gate_speedup"] is None and paired["gate_relation"] == "gate_unreachable" + json.dumps(read_records(project[3])[1], allow_nan=False) + + +def test_records_without_paired_statistics_remain_readable(project, capsys, monkeypatch): + """Assessments retained before this field existed still load in records and dashboard.""" + from inferrsi.harness import project as project_protocol + + original = project_protocol.assess + + def legacy_assess(*args, **kwargs): + result = original(*args, **kwargs) + result.pop("paired_statistics", None) + return result + + monkeypatch.setattr(project_protocol, "assess", legacy_assess) + start(project, capsys) + code, result = submit(project, capsys, 20) + assert code == 0 and "paired_statistics" not in result["latest_observation"]["assessment"] + _, state, _ = read_records(project[3]) + assert state["best"]["kind"] == "candidate" + snapshot = dashboard_snapshot(project[3]) + assert snapshot["data"]["candidates"][0]["verdict"] == "PASS" + assert "Synthetic candidate" in render_dashboard(snapshot) diff --git a/tools/measurement_simulation.py b/tools/measurement_simulation.py new file mode 100644 index 00000000..d7759116 --- /dev/null +++ b/tools/measurement_simulation.py @@ -0,0 +1,120 @@ +"""Reproduce the synthetic ablation behind harness paired statistics. + +Simulated campaigns (software only, not device measurements) compare four rules +on the same samples, so the contribution of pairing and of the interval can be +separated: + +* point_rule current project.assess: ratio of independent medians >= gate +* paired_point median of per-repetition speedups >= gate, no interval +* unpaired_interval percentile bootstrap of the ratio of independently resampled + medians; lower bound > gate +* paired_interval harness.measurement.paired_statistics: above_gate + +The data seed generates samples; the paired interval uses the production +bootstrap configuration (RESAMPLES, SEED) unless overridden. Coverage is the +fraction of campaigns whose paired bootstrap interval contains the true median +speedup (computed even below MIN_PAIRS, where the harness reports no interval), +which equals the configured true speedup under this symmetric noise model. + + python tools/measurement_simulation.py # production bootstrap config + python tools/measurement_simulation.py --resamples 1000 --campaigns 100 # quick +""" + +import argparse +import json +import math +import random +from statistics import median + +from inferrsi.harness.measurement import ( + MIN_PAIRS, + RESAMPLES, + SEED, + bootstrap_median_interval, + gate_speedup, + paired_statistics, +) + +REPETITIONS = (3, 5, 10, 20) +TRUE_SPEEDUPS = (1.0, 1.01, 1.03, 1.05, 1.10) +NOISES = ("lognormal", "heavy_tail") + + +def campaign(rng, repetitions, true_speedup, noise): + level, baseline, candidate = 10.0, [], [] + for _ in range(repetitions): + level *= math.exp(rng.gauss(0, 0.02)) # drift shared by both sides of a pair + + def sample(scale): + value = level * scale * math.exp(rng.gauss(0, 0.03)) + if noise == "heavy_tail" and rng.random() < 0.1: + value *= 1.3 + return value + + baseline.append(sample(1.0)) + candidate.append(sample(1 / true_speedup)) + return baseline, candidate + + +def unpaired_lower_bound(baseline, candidate, rng, resamples): + ratios = sorted( + median(rng.choices(baseline, k=len(baseline))) + / median(rng.choices(candidate, k=len(candidate))) + for _ in range(resamples) + ) + return ratios[int(0.025 * (resamples - 1))] + + +def rates(*, repetitions, true_speedup, noise, gate_pct, campaigns, data_seed, resamples, + bootstrap_seed): + rng = random.Random(f"{data_seed}-{noise}-{repetitions}-{true_speedup}") + unpaired_rng = random.Random(f"unpaired-{data_seed}") + gate = gate_speedup(gate_pct, "minimize") + counts = dict.fromkeys(("point_rule", "paired_point", "unpaired_interval", + "paired_interval", "covered"), 0) + widths = [] + for _ in range(campaigns): + baseline, candidate = campaign(rng, repetitions, true_speedup, noise) + point = median(baseline) / median(candidate) + counts["point_rule"] += point >= gate and point > 1 + pairs = [b / c for b, c in zip(baseline, candidate)] + counts["paired_point"] += median(pairs) >= gate + counts["unpaired_interval"] += ( + unpaired_lower_bound(baseline, candidate, unpaired_rng, resamples) > gate + ) + result = paired_statistics(baseline, candidate, direction="minimize", + min_improvement_pct=gate_pct, point_gate_passed=True, + resamples=resamples, seed=bootstrap_seed) + # Coverage is computed for every repetition count, including those below + # MIN_PAIRS where the harness deliberately reports no interval. + low, high = bootstrap_median_interval(pairs, resamples=resamples, seed=bootstrap_seed) + counts["covered"] += low <= true_speedup <= high + widths.append(high / low) + counts["paired_interval"] += result.get("gate_relation") == "above_gate" + row = {"noise": noise, "repetitions": repetitions, "true_speedup": true_speedup, + "harness_reports_interval": repetitions >= MIN_PAIRS} + for name, count in counts.items(): + p = count / campaigns + row[name] = {"rate": round(p, 4), + "monte_carlo_se": round(math.sqrt(p * (1 - p) / campaigns), 4)} + row["median_interval_width_ratio"] = round(median(widths), 4) if widths else None + return row + + +def main(): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--campaigns", type=int, default=200) + parser.add_argument("--gate-pct", type=float, default=1.0) + parser.add_argument("--data-seed", type=int, default=20261115) + parser.add_argument("--resamples", type=int, default=RESAMPLES) + parser.add_argument("--bootstrap-seed", type=int, default=SEED) + args = parser.parse_args() + rows = [rates(repetitions=r, true_speedup=t, noise=n, gate_pct=args.gate_pct, + campaigns=args.campaigns, data_seed=args.data_seed, + resamples=args.resamples, bootstrap_seed=args.bootstrap_seed) + for n in NOISES for r in REPETITIONS for t in TRUE_SPEEDUPS] + print(json.dumps({"settings": vars(args), "rows": rows}, indent=2)) + + +if __name__ == "__main__": + main()