From 670d698b8089a6e1c7f661898f7bbe726ae18cb5 Mon Sep 17 00:00:00 2001 From: iraj465 Date: Thu, 20 Aug 2026 23:04:09 +0000 Subject: [PATCH] Publish the throughput pair an e2e delta is a ratio of A journey e2e block states `e2e_gain_pct` and nothing else. A percentage alone does not compose: two wins measured against different denominators cannot be added. Every consumer holding only percentages is forced to sum them, which is a unit error, not a gain. The two numbers are already measured. `integrate_result.json` carries `ref_med` and `cand_med` from the same A/B that produced the delta, and 25 of the 26 accepted/stack overlays on disk reproduce their own delta from that pair to within 4.74e-03 pp. GEAK measured the denominator and threw it away. Publish it. Both journey builders now emit `base_tput`/`new_tput` alongside the delta, read off that same A/B, and the workflow carries the pair on each accepted head/kernel record so the live path has it too. A run that genuinely has no pair emits null rather than a fabricated denominator. Replaying all 36 KEEP blocks in /shared_nfs/hyperloom-claw through the consumer's real arithmetic: without the pair 34 of 36 steps fall back to summing local percentages; with it 23 take the composable baseline chain. The remaining 12 are live-path KEEPs whose runs predate this change. Co-Authored-By: Claude Opus 5 --- e2e_workflow/e2e_workflow.js | 34 ++++++++++------ interface/run_e2e.py | 10 +++++ interface/test_run_e2e_dispatch.py | 63 ++++++++++++++++++++++++++++++ 3 files changed, 96 insertions(+), 11 deletions(-) diff --git a/e2e_workflow/e2e_workflow.js b/e2e_workflow/e2e_workflow.js index 3dcf124ec..2c27e6b3a 100644 --- a/e2e_workflow/e2e_workflow.js +++ b/e2e_workflow/e2e_workflow.js @@ -777,6 +777,18 @@ async function extractWithBaseline(role, phase, intro, inputs, opts) { // verdict. gate:'incomplete' or ab_complete:false means a leg is still missing. const abDone = (integ) => !!(integ && integ.gate !== 'incomplete' && integ.ab_complete !== false); +// An e2e delta is a RATIO. These are the two numbers it is a ratio of, taken +// off the same A/B that produced it: ref_med is where the workload stood +// before this win, cand_med is where it stood after. Published together, a +// consumer can restate the win in points of one fixed baseline. Published +// alone, the only thing a consumer can do is add percentages that were each +// measured against a different denominator, and those do not compose. +const e2eFrom = (integ) => ({ + e2e_delta_pct: integ.e2e_delta_pct, + base_tput: integ.ref_med, + new_tput: integ.cand_med, +}); + // Run ONE integrate A/B and GUARANTEE both legs complete. The first call does a // normal apply+gate; if the integrator returns incomplete (ran only ref, hung, // or degraded mid-A/B) we RE-INVOKE it in resume mode to run the MISSING leg, @@ -1456,7 +1468,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { history.ledger.push({ direction: c.uid, isolated_speedup: c.best, e2e_delta_pct: integ.e2e_delta_pct, verdict: 'dead_end', lesson: 'parity fail vs true baseline' }); } else if (integAccepted(integ, c.head.pct_gpu_time, c.best) && integ.e2e_throughput_tok_s > curTput) { curOverlay = integ.accepted_overlay || curOverlay; curTput = integ.e2e_throughput_tok_s; bankedHeads.add(c.head.short_name); - acceptedHeads.push({ short_name: c.head.short_name, op_kind: c.ext.op_kind, backend: c.lang, lane: c.key, kind: 'patch', e2e_delta_pct: integ.e2e_delta_pct, isolated: c.best }); + acceptedHeads.push({ short_name: c.head.short_name, op_kind: c.ext.op_kind, backend: c.lang, lane: c.key, kind: 'patch', ...e2eFrom(integ), isolated: c.best }); log(` [deep] ${c.uid}: ACCEPTED. e2e now ${curTput} tok/s (+${integ.e2e_delta_pct}%); target ${Math.round(BASELINE_TPUT * DEEP_E2E_TARGET)} tok/s.`); history.ledger.push({ direction: c.uid, isolated_speedup: c.best, e2e_delta_pct: integ.e2e_delta_pct, verdict: 'confirmed', lesson: integ.reason || '' }); } else { @@ -1482,7 +1494,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { }); if (dcorr.banked) { curOverlay = dcorr.integ.accepted_overlay || curOverlay; curTput = dcorr.integ.e2e_throughput_tok_s; bankedHeads.add(c.head.short_name); - acceptedHeads.push({ short_name: c.head.short_name, op_kind: c.ext.op_kind, backend: c.lang, lane: c.key, kind: 'patch', e2e_delta_pct: dcorr.integ.e2e_delta_pct, isolated: dcorr.isolated, corrective: true }); + acceptedHeads.push({ short_name: c.head.short_name, op_kind: c.ext.op_kind, backend: c.lang, lane: c.key, kind: 'patch', ...e2eFrom(dcorr.integ), isolated: dcorr.isolated, corrective: true }); log(` [deep] ${c.uid}: ACCEPTED after corrective re-author. e2e now ${curTput} tok/s (+${dcorr.integ.e2e_delta_pct}%).`); history.ledger.push({ direction: c.uid, isolated_speedup: dcorr.isolated, e2e_delta_pct: dcorr.integ.e2e_delta_pct, verdict: 'confirmed_corrective', lesson: `fixed: ${dreason}` }); } else { @@ -1781,7 +1793,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { if (cand.winner_kind === 'env' && cand.apply_env) curEnv = (curEnv ? curEnv + ' ' : '') + cand.apply_env; if (cand.winner_kind === 'flag' && cand.apply_flags) curFlags = (curFlags ? curFlags + ' ' : '') + cand.apply_flags; curTput = integ.e2e_throughput_tok_s; - acceptedHeads.push({ short_name: h.short_name, op_kind: st.ext.op_kind, backend: cand.source, kind: cand.winner_kind, e2e_delta_pct: integ.e2e_delta_pct, isolated: cand.isolated }); + acceptedHeads.push({ short_name: h.short_name, op_kind: st.ext.op_kind, backend: cand.source, kind: cand.winner_kind, ...e2eFrom(integ), isolated: cand.isolated }); log(` ${h.short_name}: ACCEPTED. e2e now ${curTput} tok/s (+${integ.e2e_delta_pct}%).`); history.ledger.push({ direction: h.short_name, isolated_speedup: cand.isolated, e2e_delta_pct: integ.e2e_delta_pct, verdict: 'confirmed', lesson: integ.reason || '' }); } else { @@ -1809,7 +1821,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { : { banked: false }; if (corr.banked) { curOverlay = corr.integ.accepted_overlay || curOverlay; curTput = corr.integ.e2e_throughput_tok_s; - acceptedHeads.push({ short_name: h.short_name, op_kind: st.ext.op_kind, backend: cand.source, kind: 'authored', e2e_delta_pct: corr.integ.e2e_delta_pct, isolated: corr.isolated, corrective: true }); + acceptedHeads.push({ short_name: h.short_name, op_kind: st.ext.op_kind, backend: cand.source, kind: 'authored', ...e2eFrom(corr.integ), isolated: corr.isolated, corrective: true }); log(` ${h.short_name}: ACCEPTED after corrective re-author (${reason}). e2e now ${curTput} tok/s (+${corr.integ.e2e_delta_pct}%).`); history.ledger.push({ direction: h.short_name, isolated_speedup: corr.isolated, e2e_delta_pct: corr.integ.e2e_delta_pct, verdict: 'confirmed_corrective', lesson: `fixed: ${reason}` }); } else { @@ -2023,7 +2035,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { if (cand.winner_kind === 'env' && cand.apply_env) curEnv = (curEnv ? curEnv + ' ' : '') + cand.apply_env; if (cand.winner_kind === 'flag' && cand.apply_flags) curFlags = (curFlags ? curFlags + ' ' : '') + cand.apply_flags; curTput = integ.e2e_throughput_tok_s; - acceptedHeads.push({ short_name: h.short_name, op_kind: ext.op_kind, backend: cand.source, kind: cand.winner_kind, e2e_delta_pct: integ.e2e_delta_pct, isolated: cand.isolated }); + acceptedHeads.push({ short_name: h.short_name, op_kind: ext.op_kind, backend: cand.source, kind: cand.winner_kind, ...e2eFrom(integ), isolated: cand.isolated }); log(` ${h.short_name}: ACCEPTED best candidate=${cand.source} (${(cand.isolated || 0).toFixed(2)}x iso). e2e now ${curTput} tok/s (+${integ.e2e_delta_pct}%).`); history.ledger.push({ direction: h.short_name, isolated_speedup: cand.isolated, e2e_delta_pct: integ.e2e_delta_pct, verdict: 'confirmed', lesson: integ.reason || '' }); } else { @@ -2046,7 +2058,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { : { banked: false }; if (corr.banked) { curOverlay = corr.integ.accepted_overlay || curOverlay; curTput = corr.integ.e2e_throughput_tok_s; - acceptedHeads.push({ short_name: h.short_name, op_kind: ext.op_kind, backend: cand.source, kind: 'authored', e2e_delta_pct: corr.integ.e2e_delta_pct, isolated: corr.isolated, corrective: true }); + acceptedHeads.push({ short_name: h.short_name, op_kind: ext.op_kind, backend: cand.source, kind: 'authored', ...e2eFrom(corr.integ), isolated: corr.isolated, corrective: true }); log(` ${h.short_name}: ACCEPTED after corrective re-author (was crash/incomplete: ${reason}). e2e now ${curTput} tok/s (+${corr.integ.e2e_delta_pct}%).`); history.ledger.push({ direction: h.short_name, isolated_speedup: corr.isolated, e2e_delta_pct: corr.integ.e2e_delta_pct, verdict: 'confirmed_corrective', lesson: `fixed crash: ${reason}` }); } else { @@ -2073,7 +2085,7 @@ if (want('head') && headQueue.length && HEAD_BUDGET > 0) { : { banked: false }; if (corr.banked) { curOverlay = corr.integ.accepted_overlay || curOverlay; curTput = corr.integ.e2e_throughput_tok_s; - acceptedHeads.push({ short_name: h.short_name, op_kind: ext.op_kind, backend: cand.source, kind: 'authored', e2e_delta_pct: corr.integ.e2e_delta_pct, isolated: corr.isolated, corrective: true }); + acceptedHeads.push({ short_name: h.short_name, op_kind: ext.op_kind, backend: cand.source, kind: 'authored', ...e2eFrom(corr.integ), isolated: corr.isolated, corrective: true }); log(` ${h.short_name}: ACCEPTED after corrective re-author. e2e now ${curTput} tok/s (+${corr.integ.e2e_delta_pct}%).`); history.ledger.push({ direction: h.short_name, isolated_speedup: corr.isolated, e2e_delta_pct: corr.integ.e2e_delta_pct, verdict: 'confirmed_corrective', lesson: `fixed: ${reason}` }); } else { @@ -2223,7 +2235,7 @@ while (want('kernel') && !TIME_DEADLINE_HIT && dispatched < BUDGET && (dispatche if (abDone && integAccepted(integ, c.pct_gpu_time, kl.final_geomean) && integ.e2e_throughput_tok_s > curTput) { curOverlay = integ.accepted_overlay || curOverlay; curTput = integ.e2e_throughput_tok_s; - acceptedKernels.push({ short_name: c.short_name, backend: kl.note || '', e2e_delta_pct: integ.e2e_delta_pct, isolated: kl.final_geomean }); + acceptedKernels.push({ short_name: c.short_name, backend: kl.note || '', ...e2eFrom(integ), isolated: kl.final_geomean }); milestoneImproved = true; log(` ${c.short_name}: ACCEPTED. e2e now ${curTput} tok/s (+${integ.e2e_delta_pct}%).`); history.ledger.push({ direction: c.short_name, isolated_speedup: kl.final_geomean, e2e_delta_pct: integ.e2e_delta_pct, verdict: 'confirmed', lesson: integ.reason || '' }); @@ -2243,7 +2255,7 @@ while (want('kernel') && !TIME_DEADLINE_HIT && dispatched < BUDGET && (dispatche : { banked: false }; if (corr.banked) { curOverlay = corr.integ.accepted_overlay || curOverlay; curTput = corr.integ.e2e_throughput_tok_s; - acceptedKernels.push({ short_name: c.short_name, backend: kl.note || '', e2e_delta_pct: corr.integ.e2e_delta_pct, isolated: corr.isolated, corrective: true }); + acceptedKernels.push({ short_name: c.short_name, backend: kl.note || '', ...e2eFrom(corr.integ), isolated: corr.isolated, corrective: true }); milestoneImproved = true; log(` ${c.short_name}: ACCEPTED after corrective re-author (${reason}). e2e now ${curTput} tok/s (+${corr.integ.e2e_delta_pct}%).`); history.ledger.push({ direction: c.short_name, isolated_speedup: corr.isolated, e2e_delta_pct: corr.integ.e2e_delta_pct, verdict: 'confirmed_corrective', lesson: `fixed: ${reason}` }); @@ -2385,9 +2397,9 @@ if (want('final')) { if (p.track === 'head') { if (p.winner_kind === 'env' && p.apply_env) curEnv = (curEnv ? curEnv + ' ' : '') + p.apply_env; if (p.winner_kind === 'flag' && p.apply_flags) curFlags = (curFlags ? curFlags + ' ' : '') + p.apply_flags; - acceptedHeads.push({ short_name: p.short_name, op_kind: p.op_kind, backend: p.backend, kind: p.winner_kind, e2e_delta_pct: integ.e2e_delta_pct, isolated: p.isolated }); + acceptedHeads.push({ short_name: p.short_name, op_kind: p.op_kind, backend: p.backend, kind: p.winner_kind, ...e2eFrom(integ), isolated: p.isolated }); } else { - acceptedKernels.push({ short_name: p.short_name, backend: p.backend || '', e2e_delta_pct: integ.e2e_delta_pct, isolated: p.isolated }); + acceptedKernels.push({ short_name: p.short_name, backend: p.backend || '', ...e2eFrom(integ), isolated: p.isolated }); } curTput = integ.e2e_throughput_tok_s; finalTput = curTput; finalSpeedup = BASELINE_TPUT ? curTput / BASELINE_TPUT : 1.0; diff --git a/interface/run_e2e.py b/interface/run_e2e.py index d2b459fa9..8c68ceac4 100644 --- a/interface/run_e2e.py +++ b/interface/run_e2e.py @@ -3403,6 +3403,12 @@ def _journey_overlay_entry(eval_dir: Path, short: str, ir: dict, wf: dict, gpu_pct = gpu_pct_prof micro = _ir_get(ir, "isolated_speedup", "micro_speedup", "speedup") if ir else None delta = _ir_get(ir, "e2e_delta_pct", "delta_pct") if ir else None + # The delta is a ratio; these are the two numbers it is a ratio of, read + # off the SAME integrate_result. A consumer holding the pair can restate + # the win in points of one fixed baseline. Holding only the percentage it + # can do nothing but add figures measured against different denominators. + base_tput = _ir_get(ir, "ref_med", "ref_median_tok_s") if ir else None + new_tput = _ir_get(ir, "cand_med", "cand_median_tok_s") if ir else None winner_kind = str(_ir_get(ir, "winner_kind") or "").lower() if ir else "" is_config = winner_kind in ("env", "config", "flags") flags = str(_ir_get(ir, "apply_flags") or "") if ir else "" @@ -3463,6 +3469,8 @@ def _journey_overlay_entry(eval_dir: Path, short: str, ir: dict, wf: dict, "kernel_id": kernel_id, "integrated": accepted, "e2e_gain_pct": delta, + "base_tput": base_tput, + "new_tput": new_tput, "validated": True, # an A/B gate ran either way "decision": "KEEP" if accepted else "REJECTED", "patch_path": patch, @@ -3507,6 +3515,8 @@ def _journey_return_entry(eval_dir: str, k: dict, idx: int, wf: dict, }, "e2e": { "kernel_id": kid, "integrated": True, "e2e_gain_pct": k.get("e2e_delta_pct"), + # Same pair as the overlay path, carried through the workflow return. + "base_tput": k.get("base_tput"), "new_tput": k.get("new_tput"), "validated": True, "decision": "KEEP", "patch_path": patch, "target_file": k.get("target_file") or k.get("target_callable"), "extra_server_args": str((wf.get("accepted_config") or {}).get("flags") or ""), diff --git a/interface/test_run_e2e_dispatch.py b/interface/test_run_e2e_dispatch.py index e36c50636..50313a45d 100644 --- a/interface/test_run_e2e_dispatch.py +++ b/interface/test_run_e2e_dispatch.py @@ -2407,5 +2407,68 @@ def boom(eval_dir, wf, normalized): self.assertNotIn("kernel_journey_path", out) +class TestE2EDenominatorIsPublished(_RunE2ECase): + """A journey e2e block must carry the two throughputs its percentage is a + ratio of. + + A percentage alone does not compose: two wins measured against different + denominators cannot be added. The consumer needs the pair to restate each + win in points of one fixed baseline. Both journey builders read the pair + off the same A/B that produced the delta, so the three numbers always + agree. + """ + + def _overlay(self, eval_dir, ir): + d = eval_dir / "overlay" / "cand_my_kernel_fwd" + self.write_json(d / "integrate_result.json", ir) + return d + + def test_overlay_path_publishes_the_pair_and_it_reproduces_the_delta(self): + eval_dir = self.tmp / "e2e_flat" + self._overlay(eval_dir, {"gate": "accepted", "e2e_delta_pct": 25.0, + "ref_med": 400.0, "cand_med": 500.0}) + journey = rx.build_kernel_journey({"eval_dir": str(eval_dir)}, + {"eval_dir": str(eval_dir)}) + e2e = journey["kernels"][0]["e2e"] + self.assertEqual(e2e["base_tput"], 400.0) + self.assertEqual(e2e["new_tput"], 500.0) + self.assertAlmostEqual( + (e2e["new_tput"] / e2e["base_tput"] - 1.0) * 100.0, + e2e["e2e_gain_pct"], places=6) + + def test_overlay_path_reads_the_pair_from_the_nested_shape_too(self): + """The integrator emits either shape; the delta already reads both, and + the pair it is a ratio of must not be lost on the nested one.""" + eval_dir = self.tmp / "e2e_nested" + self._overlay(eval_dir, {"gate": "accepted", + "e2e": {"delta_pct": 25.0, + "ref_median_tok_s": 400.0, + "cand_median_tok_s": 500.0}}) + e2e = rx.build_kernel_journey({"eval_dir": str(eval_dir)}, + {"eval_dir": str(eval_dir)})["kernels"][0]["e2e"] + self.assertEqual((e2e["base_tput"], e2e["new_tput"]), (400.0, 500.0)) + + def test_return_path_carries_the_pair_from_the_workflow_return(self): + eval_dir = self.tmp / "e2e_return" + wf = {"eval_dir": str(eval_dir), + "accepted_kernels": [{"short_name": "my_kernel_fwd", + "e2e_delta_pct": 25.0, + "base_tput": 400.0, "new_tput": 500.0}]} + e2e = rx.build_kernel_journey(wf, {"eval_dir": str(eval_dir)})["kernels"][0]["e2e"] + self.assertEqual((e2e["base_tput"], e2e["new_tput"]), (400.0, 500.0)) + + def test_a_missing_pair_is_null_not_fabricated(self): + """An older workflow return has no throughputs. The block must say so + rather than invent a denominator the A/B never measured.""" + eval_dir = self.tmp / "e2e_nopair" + wf = {"eval_dir": str(eval_dir), + "accepted_kernels": [{"short_name": "my_kernel_fwd", + "e2e_delta_pct": 25.0}]} + e2e = rx.build_kernel_journey(wf, {"eval_dir": str(eval_dir)})["kernels"][0]["e2e"] + self.assertIsNone(e2e["base_tput"]) + self.assertIsNone(e2e["new_tput"]) + self.assertEqual(e2e["e2e_gain_pct"], 25.0) + + if __name__ == "__main__": raise SystemExit(0 if unittest.main(exit=False).result.wasSuccessful() else 1)