Skip to content

Publish the throughput pair an e2e delta is a ratio of #427

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zihaoanllm merged 2 commits into from
Aug 21, 2026
Merged

Publish the throughput pair an e2e delta is a ratio of #427

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
670d698
Publish the throughput pair an e2e delta is a ratio of
iraj465 Aug 20, 2026
cd3ffe2
Merge branch 'main' into fix/publish-e2e-denominator
zihaoanllm Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 23 additions & 11 deletions e2e_workflow/e2e_workflow.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1054,6 +1054,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,
Expand Down Expand Up @@ -1733,7 +1745,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 {
Expand All @@ -1759,7 +1771,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 {
Expand Down Expand Up @@ -2058,7 +2070,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 {
Expand Down Expand Up @@ -2086,7 +2098,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 {
Expand Down Expand Up @@ -2300,7 +2312,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 {
Expand All @@ -2323,7 +2335,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 {
Expand All @@ -2350,7 +2362,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 {
Expand Down Expand Up @@ -2500,7 +2512,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 || '' });
Expand All @@ -2520,7 +2532,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}` });
Expand Down Expand Up @@ -2662,9 +2674,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;
Expand Down
10 changes: 10 additions & 0 deletions interface/run_e2e.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 ""
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 ""),
Expand Down
63 changes: 63 additions & 0 deletions interface/test_run_e2e_dispatch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Loading