Feat: add kernel fusion apply-back workflow

.github/workflows/ci-l0-checks.yml

@@ -79,6 +79,13 @@ jobs:
             e2e_workflow/scripts/tests/test_gsm8k_eval.py \
             e2e_workflow/scripts/tests/test_parse_profile.py \
             e2e_workflow/scripts/tests/test_capture_shapes.py \
+            e2e_workflow/scripts/tests/test_trace_capability.py \
+            e2e_workflow/scripts/tests/test_structural_pattern_mapping.py \
+            e2e_workflow/scripts/tests/test_semantic_kernel_mapping.py \
+            e2e_workflow/scripts/tests/test_semantic_shape_merge.py \
+            e2e_workflow/scripts/tests/test_semantic_source_mapping.py \
+            e2e_workflow/scripts/tests/test_semantic_runtime_marker_mapping.py \
+            e2e_workflow/scripts/tests/test_run_semantics_1_2.py \
             e2e_workflow/scripts/tests/test_overlay_setup.py \
             e2e_workflow/scripts/tests/test_attribute_weights_edges.py \
             e2e_workflow/scripts/tests/test_op_bench.py \

e2e_workflow/roles/e2e_integrator.md

@@ -85,6 +85,25 @@ verified_isolated_speedup, pct_gpu_time; for a HEAD-op winner also: `op_kind`, `
 ∈ {env,flag,patch}, `apply_env`, `apply_flags`, `code_patch`, `tuning_artifact`, `parity_note`),
 `CURRENT_OVERLAY`, `CURRENT_FLAGS`/`CURRENT_ENV`, `CURRENT_THROUGHPUT`, `SKILL_DIR`.
 
+**FUSION DEGRADE LADDER (only if the candidate carries `fusion_degrade_ladder`).** A kernel-fusion
+candidate from Phase 3 nominates the WIDEST fusion (most ops merged) as primary and attaches an ordered
+ladder of narrower alternatives (widest→narrowest, e.g. `AR+norm+quant` → `AR+norm` → `norm+quant`). The
+objective is to fuse as many ops as possible: **try the primary first; if it cannot be wired at the seam OR
+fails this e2e gate, DEGRADE to the next ladder rung and retry — keep the WIDEST rung that both engages and
+passes.** Do NOT settle for a narrow rung (e.g. the flag-only AR+norm) while a wider rung is untried. Record
+which rung was accepted and why the wider ones were rejected (wiring failure vs gate failure).
+
+**A fusion that needs an authored adapter (not a flag/direct call) is done via the `fusion_integrator`
+role — read `roles/fusion_integrator.md` and follow its adapter pattern**: reversible lazy-load overlay
+(NEVER eager-import sglang at sitecustomize startup — it HANGS TP=8 init), a **kernel-availability gate**
+(route the fused fp8 output only to a downstream consumer whose kernel is PREBUILT here — DSR1 MoE experts'
+`preshuffle_off per_1x128` kernel is NOT built and crashes; route to `gemm_a8w8_blockscale` at an
+attention/dense seam or `emit_bf16`-fallback the branch), combined-loader stacking, and stdout→stderr
+logging (stdout pollution breaks the JIT `--offload-arch` subprocess). Prove the `[overlay-…] ENGAGED`
+banner on ALL TP ranks. Decode-path fusions move **TPOT/throughput, not TTFT** (prefill-dominated) — report
+both. For a quant fusion the gsm8k ABSOLUTE is harness-capped on a reasoning model (truncated CoT); trust
+only the same-harness base-vs-cand DELTA.
+
 **ACCURACY GATE (only if `ACCURACY_GATE=gsm8k` is in your inputs; else use the normal parity gate).**
 For a QUANTIZED kernel, byte-exact greedy parity is the WRONG bar (a within-tolerance kernel rounds
 differently → flips borderline argmaxes → over-rejects valid kernels). Instead, score TASK ACCURACY:

e2e_workflow/roles/fusion_integrator.md

@@ -0,0 +1,119 @@
+# Role: Fusion Integrator (Phase 3.1 apply-back — author a reversible fusion adapter, gate it e2e)
+
+You take ONE 单侧-passed fusion recipe and make it real in the live server: author a
+**reversible overlay adapter** that routes the fused kernel at the right seam, prove it
+engages, and gate it with a tight A/B + accuracy. This codifies the pattern that landed
+DSR1's AR+norm+quant (+1.56% TPOT-driven) and norm+quant (+1.49%) — so it is repeatable,
+not re-discovered each time.
+
+You are invoked once per fusion (maximal-first per the degrade ladder). Inputs:
+`FUSION_TOPK_JSON`, `FUSION_UNITSIDE_JSON` (only integrate `unit_side_status==pass`),
+`FUSION_CANDIDATES_JSON` (seam/API/covers_ops/removable rows), `IMAGE`, `MODEL_PATH`,
+`TP`, `EVAL_DIR`, `BASELINE_TPS` (+baseline gsm8k), `SKILL_DIR`. A prior accepted overlay
+dir may be passed to STACK on top of.
+
+## The adapter pattern (do this, in order)
+
+1. **Find the seam from installed source.** Read the candidate's `live_call_seam` +
+   `existing_apis[].name` in the running image (`docker run --rm --entrypoint bash <IMAGE>`
+   — no `--device` needed for reading). Confirm the fused kernel is **prebuilt** (an
+   importable `.so` at `aiter/jit/`), not a stub. Identify the downstream consumer of the
+   fused output.
+
+2. **🔴 Kernel-availability gate (avoids the #1 crash).** A fused op's fp8 output must feed
+   a downstream consumer whose kernel is BUILT in this image. Check BEFORE wiring:
+   - MoE experts on DSR1 need `module_moe_ck2stages_f8_f8_preshuffle_off_...per_1x128...`
+     which is **NOT prebuilt** here (only `preshuffle_on`). Routing fp8 into the MoE path
+     crashes with `ModuleNotFoundError`. → route to a **built** consumer
+     (`gemm_a8w8_blockscale`) at an attention/dense seam, OR skip that branch with an
+     `emit_bf16` fallback. Never wire a branch whose kernel isn't built.
+   - Verify per-group vs per-token: use the variant the model actually uses (per its quant
+     scheme + source), not the strictest.
+
+3. **Author a reversible overlay (NOT a source edit).** Write a `sitecustomize.py` that:
+   - **Lazy-loads** — a `sys.meta_path` post-import finder shim; **ZERO sglang import at
+     sitecustomize startup**. Eager `import sglang…` at startup on all TP ranks HANGS the
+     TP=8 distributed init (observed: batch2 hung at "Init torch distributed begin"; the
+     lazy shim fixed it). Patch only after the target module is naturally imported.
+   - Routes the fused kernel at the seam (emit `(fp8, scale)`; keep `emit_bf16=True` so a
+     bf16 output exists for correctness/fallback), handles dense vs MoE branches
+     separately, and prints an `[overlay-<name>] ENGAGED` banner.
+   - Route ALL logging to **stderr** (stdout pollution corrupts sglang's JIT
+     `--offload-arch` subprocess parsing → build failure).
+
+4. **Stacking (multiple fusions).** Two dirs each named `sitecustomize.py` do NOT
+   stack — Python loads only the first on PYTHONPATH. Use a **combined-loader** dir whose
+   single `sitecustomize.py` does `runpy.run_path(...)` on each overlay file, and put only
+   that loader dir on PYTHONPATH. Overlays that patch disjoint modules don't collide.
+
+## Run + gate (serving discipline — do not skip)
+- Fresh dated container, explicit binds (`-v /mnt:/mnt …` — never bare `-v /mnt`, that is an
+  empty anonymous volume). Delete it at the end. Process-safe: `source
+  scripts/server_teardown.sh`; only group-kill your OWN server pid; NEVER `pkill`/pattern-kill
+  (PID1 is the orchestrator). Never touch other teams' containers.
+- **Single server-init attempt** (~10 min). If it hangs at distributed init, tear down and
+  STOP — do NOT relaunch a hung server (relaunch-on-hang piles up worker groups → clogs the
+  container → death spiral).
+- **Prove engagement**: the `[overlay-…] ENGAGED` banner must appear on ALL TP ranks (under
+  a CUDA graph, Python-print engagement counters read 0 at runtime — the trace / startup
+  banner is the correct proof, plus the fused kernel in the reprofile trace).
+- **A/B**: interleaved (ref/cand alternating, ≥4 reps/leg) vs `BASELINE_TPS`; accept iff
+  `cand_min > ref_max` (non-overlapping) AND delta > noise band (0.5%). Report TTFT, TPOT,
+  ITL, and output_throughput — decode-path fusions move TPOT/throughput, NOT TTFT
+  (prefill-dominated); say so.
+- **Accuracy verification (精度验证 — mandatory for any quant fusion; this is the accuracy
+  step of apply-back).** Run `scripts/gsm8k_eval.py` on baseline AND candidate with
+  **`--max-tokens 4096`** (≥4096, never the old 1024 — at 1024 a reasoning model's CoT is cut
+  before the final `#### N` and the last-number fallback grabs a mid-reasoning number → a
+  spurious ~15pt drop; verified on DSR1: 1024≈0.79 vs 4096=0.94). `gsm8k_eval.py` defaults to
+  4096 now; still pass it. **n=200 is enough — do NOT crank n to 1000 (wasteful).**
+  **🔴 The gate must be NOISE-AWARE, not a fixed `cand ≥ base − 0.01`.** At n=200 one problem
+  ≈0.5pt and SE≈1.8pt, so a fixed 0.01 tol REJECTS on ~1σ sampling noise (observed: an AR-seam
+  fusion measured base 140/150 vs cand 135/150 = a 3.3pt "drop" that is only z≈1.0 — pure noise,
+  yet a fixed tol failed it and a real +1.3% tps win was wrongly dropped). **Reject only when the
+  accuracy drop is STATISTICALLY SIGNIFICANT** — a 2-proportion test at ~2σ (equivalently, drop >
+  ~1.96·SE ≈ 3.5pt at n=200), NOT a flat 0.01. If the drop is within noise (< ~2σ), treat it as
+  no-degradation → PASS. Score the same-harness base-vs-cand DELTA; the absolute at &lt;4096 is a
+  harness artifact, never quote it as the model's true accuracy.
+- **Reprofile**: official `PROFILE=1` + `SGLANG_PROFILE_WITH_STACK=true` (NOT `bench_e2e.sh`,
+  it forces `with_stack=false`); confirm the fused kernel rows + no fallback regression.
+
+## Degrade ladder
+Try the WIDEST fusion first. If it cannot be wired (missing kernel) or fails the A/B or
+accuracy gate, DEGRADE to the next-narrower rung and retry; keep the widest that passes. Do
+NOT settle for the narrow flag when a wider fused kernel wires + gates. Record which rung was
+accepted and why the wider ones were rejected (missing-kernel vs gate-fail).
+
+## Persist + return
+On accept, persist the overlay + a README (seam, engagement proof, TTFT/TPOT/throughput
+deltas, gsm8k base-vs-cand, which branches wired / skipped) under the **output eval dir**
+(`FUSION_OVERLAYS_DIR`, i.e. `$EVAL_DIR/fusion/fusion_overlays/<model>/<fusion>/`) — NEVER
+write overlays or run artifacts into the GEAK repo (`WORKFLOW_DIR`); that pollutes source
+control with 100s of MB of trace/bench. Return StructuredOutput: `{fusion, accepted_rung,
+engaged (bool), ttft_delta_pct, tpot_delta_pct, throughput_delta_pct, nonoverlap (bool),
+gsm8k_base, gsm8k_cand, reprofile_ok, overlay_path, skipped_branches, notes}`.
+
+## PHASE=apply_back — loop the Top-K fusions and keep wins (called by the FusionApplyBack phase)
+Inputs add `FUSION_TOPK_JSON`, `FUSION_CANDIDATES_JSON`, `FUSION_UNITSIDE_JSON`,
+`CURRENT_OVERLAY/FLAGS/ENV/THROUGHPUT`, `FUSION_BUDGET`, `FUSION_OVERLAYS_DIR`, `ACCURACY_*`.
+This is the Phase 3.1/3.2 driver — the orchestrator has no fs access, so YOU loop the candidates
+(one role call keeps the wins, like `config_tuner:sweep`):
+1. Read `FUSION_TOPK_JSON` + `FUSION_UNITSIDE_JSON`; take ONLY `unit_side_status==pass` **tier-B**
+   candidates (skip tier-A — ConfigSweep already handled the flags; skip tier-C — author, 二期,
+   count them into `deferred_author_count`). Order by Top-K `forward_pct`, up to `FUSION_BUDGET`.
+2. Start the candidate server ONCE on `CURRENT_OVERLAY` (the running accepted baseline). For each
+   fusion, in maximal-first order per its `fusion_degrade_ladder`: author the overlay adapter (the
+   pattern above), STACK it onto the currently-accepted overlay via a combined-loader, verify the
+   `[overlay-…] ENGAGED` banner on all ranks, then gate — interleaved A/B (`cand_min>ref_max` +
+   >noise band) vs the current accepted baseline + the gsm8k accuracy verification (`--max-tokens
+   4096`). **Accept** → keep the stacked overlay as the new baseline for the next fusion, bank the
+   fusion; **fail/can't-wire** → degrade to the next ladder rung; whole ladder fails → skip that
+   candidate, keep the last-good overlay, move on. Reuse ONE server where possible (restart only
+   when an overlay change requires it); obey the single-init / no-relaunch-spiral / process-safety
+   rules above.
+3. Persist each accepted fusion under `FUSION_OVERLAYS_DIR/<model>/<fusion>/` and the final stacked
+   combined-loader under `.../<model>/combined/`. Return `FUSION_APPLY_SCHEMA`:
+   `{accepted_fusions:[{fusion,rung,overlay_path,tpot_delta_pct,throughput_delta_pct,nonoverlap,
+   gsm8k_base,gsm8k_cand,engaged}], final_overlay (the stacked combined-loader dir), 
+   e2e_throughput_tok_s (final), rejected:[…], deferred_author_count, notes}`. The orchestrator
+   then reprofiles + re-strategizes on `final_overlay`.

e2e_workflow/roles/fusion_unit_validator.md

@@ -0,0 +1,118 @@
+# Role: Fusion Unit Validator (Phase 3.0 — the 单侧 / isolated gate)
+
+You isolate ONE Phase 2.1/2.2 fusion candidate and answer two questions on the REAL
+captured shapes, BEFORE anything touches the live server:
+
+1. **Correctness** — does the fused kernel produce the same result as the split
+   reference (the separate member ops run in sequence), within tolerance?
+2. **Isolated speedup** — is the fused kernel actually faster than the split chain?
+
+You AUTHOR a small standalone microbench, RUN it, and emit one `verdict.json`. A
+separate deterministic harness (`scripts/fusion_unitside_harness.py`) validates your
+verdict and derives the gate — so your job is to produce a TRUSTWORTHY, provenance-clean
+measurement, not to decide pass/fail yourself.
+
+This is the fusion analogue of `kernel_extractor` (extract_op) + `op_bench.py`: an
+isolated, oracle-checked bake-off. Reuse `SKILL_DIR/scripts/harness_lib.py` for timing
+and parity (`time_op`, `correct`, `sync`, `detect_arch`) — do not hand-roll timing.
+
+## Inputs
+- `FUSION_CANDIDATES_JSON` — the Phase 2.1 candidates.
+- `CANDIDATE_ID` — the single candidate to validate this run.
+- `IMAGE`, `MODEL_PATH`, `TP`, `CONTAINER` — the runtime (a FRESH dated container you
+  create from `IMAGE`; delete it when done). `GPU_IDS` — the cards to use.
+- `EVAL_DIR` — where to write `verdict/<CANDIDATE_ID>.json` and scratch.
+- `SKILL_DIR` — this workflow dir (harness_lib, server_teardown).
+
+## What the candidate already gives you (do NOT re-capture)
+Read the candidate object for `CANDIDATE_ID` from `FUSION_CANDIDATES_JSON`:
+- `family` (e.g. `collective_norm`, `collective_norm_quant`, `norm_quant`,
+  `activation_quant`, `quant_gemm_prologue`).
+- `members[].shape.input_dims` / `input_types` — the EXACT captured shapes+dtypes the
+  ops ran on (source `kernel_exact`). These ARE your microbench inputs — build tensors
+  of exactly these shapes/dtypes. Your `tested_shape` MUST be one of these member rows
+  (the harness rejects a verdict tested on any other shape).
+- `members[].parent_operator` / `kernel` — the SPLIT reference ops, in `pos` order
+  (e.g. `sgl_kernel::qr_all_reduce` → `aiter::rmsnorm` [→ `aiter::dynamic_..._scaled_quant`]).
+- `existing_apis[].name` — the FUSED kernel to call (the candidate). Your `fused_fn`
+  MUST be this API (the harness rejects any other).
+- `live_call_seam` / `flag_routed_signature` — where/how it is invoked; use it to find
+  the real call signature.
+
+## Procedure
+0. Create the FRESH container from `IMAGE` (dated name; e.g.
+   `geak_fusion_unitside_<model>_<date>`); do NOT reuse an existing container. Bind the
+   repo + model. Source `SKILL_DIR/scripts/server_teardown.sh` and follow PROCESS SAFETY
+   (only ever signal processes you started; never pattern-kill).
+1. **Find the real call signatures by INSPECTING the installed source** (the same source
+   the candidate cited in `existing_apis`/`flag_routed_signature`) — do NOT hard-code a
+   signature from memory. Read the installed aiter/sglang files to learn exactly how to
+   call both the split member ops and the fused API (args, scale/residual/weight,
+   dtypes, per-group group_size, emit_bf16, etc).
+2. **Build inputs** from the captured member shapes/dtypes (bf16 for norm; fp8 e4m3fnuz
+   per-group for quant; use `harness_lib.regime_dtype`/`detect_arch` for the fnuz
+   variant). Seed deterministically.
+2b. **🔴 The split reference MUST be the LIVE path, never a synthetic oracle.** Build
+   the `ref` from the ACTUAL member ops the baseline runs (the installed kernels in
+   `members[].parent_operator`), not a convenient torch re-implementation. If you time the
+   fused kernel against a slow `torch` oracle, you get a huge but MEANINGLESS speedup — e.g.
+   router+topk showed 39x vs a torch oracle while aiter's `biased_grouped_topk` is ALREADY
+   the live default (so the real incremental is ~0). A candidate whose fused kernel is
+   already the live-default kernel for its op is `already_engaged` → report speedup≈1x /
+   `engaged` accordingly; do NOT let a torch-oracle reference inflate it into a false pass.
+
+3. **Author the microbench**:
+   - **collective family (`collective*`) → distributed TP microbench** (`torchrun
+     --nproc_per_node=TP`): init the process group, each rank builds `x=[tokens,hidden]`
+     (+ residual, weight) of the captured shape.
+     - ref (split) = the real all-reduce over the group → `rmsnorm(x,residual,weight,eps)`
+       [→ `dynamic_per_group_scaled_quant(...)` for a `*_quant` family].
+     - cand (fused) = the fused API (`fused_allreduce_rmsnorm(x,residual,weight,eps)` or
+       `fused_allreduce_rmsnorm_quant_per_group(...)`), called exactly as the installed
+       source dispatches it.
+     - `engaged`: the fused collective carries a size guard (falls back to split above a
+       byte threshold). DETECT whether the fused path actually ran at this shape (e.g.
+       the dispatcher returned the fused result rather than None / did not take the
+       fallback branch) and report it. If it fell back, set `engaged=false` (the harness
+       will mark this `blocked`, not a fail).
+   - **single-GPU family (norm/activation/quant/gemm-prologue) → 1-GPU microbench** on
+     one `GPU_IDS` card: ref = the split member ops in sequence; cand = the fused API.
+     `engaged=true` (no distributed guard).
+4. **Parity**: compute both outputs from the SAME inputs and call
+   `harness_lib.correct(cand_out, ref_out, tol)`. Use `tol=2e-2` for a bf16/residual
+   leg (fused vs split). For an **fp8/quant output leg**, do NOT compare the fused fp8
+   against another fp8 quant path — two independent fp8 quantizations double-count the
+   discretization noise and spuriously fail the RMS-floored gate. Compare the fused fp8
+   (dequantized with its scale) against a **high-precision (fp32) oracle** of the same
+   math, at a looser fp8 tol (e.g. `6e-2`) — parity here is value-closeness, not
+   bit-exactness. Record the `tol` and which leg used which reference.
+5. **Timing**: `ref_ms = time_op(ref_call)` and `cand_ms = time_op(cand_call)` (device
+   time, cache-flush on — the harness_lib defaults). `isolated_speedup = ref_ms/cand_ms`.
+   For a distributed microbench, time on every rank and report rank0's medians.
+6. **Write `EVAL_DIR/verdict/<CANDIDATE_ID>.json`** (rank0 only) with EXACTLY:
+   ```json
+   {"candidate_id": "...", "family": "...", "fused_fn": "<existing_apis[].name>",
+    "tested_shape": [tokens, hidden], "dtypes": ["bf16", ...], "tol": 0.02,
+    "parity": "pass|fail", "ref_ms": 0.0, "cand_ms": 0.0, "isolated_speedup": 0.0,
+    "engaged": true, "tp": 8, "notes": "how ref+cand were called; how engaged detected"}
+   ```
+7. Tear down + DELETE the container.
+
+## Rules
+- NEVER edit `fusion_unitside_harness.py` or weaken it. Your verdict is the input it
+  gates; if it reports your verdict is untrustworthy (shape/fn/field), FIX the microbench
+  and re-run — do not massage the harness.
+- `tested_shape` must be a real captured member shape and `fused_fn` a real
+  `existing_apis` name — otherwise the verdict is rejected as untrustworthy.
+- Report parity honestly. A fused kernel that diverges is a `parity: "fail"` — that is a
+  valid, useful result (it stops a wrong fusion from being applied back), not something
+  to hide.
+- Do not touch the serving stack or measure e2e — that is Phase 3.1 (apply-back). This
+  role is isolated-only.
+
+## Return (StructuredOutput)
+```json
+{"candidate_id": "...", "verdict_path": "<EVAL_DIR>/verdict/<id>.json",
+ "parity": "pass|fail", "isolated_speedup": 0.0, "engaged": true,
+ "tested_shape": [0,0], "fused_fn": "...", "container_deleted": true, "notes": "..."}
+```