Add measured collective profile gate for Qwen235

.research/experiments/qwen235-tp8-collective-profile-ablation-20260719.md

@@ -0,0 +1,64 @@
+# 实验 EXP-SIMFID-Q235-CC-TP8：真实 TP4/TP8 collective profile 消融
+
+> **状态：** review 通过，执行中
+>
+> 本 card 是 SHA、command、config、log 等 provenance 的唯一归宿；本轮只重跑 simulator，不重跑已经冻结的 48 个真机 trial。
+
+## Claim 与决策
+
+- **Parent claim：** Qwen235 Fixed-PD 的 30%+ TPOT/E2E selection regret，是否主要由 TP8 collective profile 缺失及 TP4 profile 与真实 serving backend 不匹配造成。
+- **目的：** 支持或反驳 mechanism hypothesis；不是用同一 workload 的 E2E calibration 修正 simulator。
+- **Competing hypotheses：**
+  - H1：collective profile coverage/backend mismatch 是排序反转的必要主因。换成与真机 serving 一致的 TP4/TP8 实测 profile 后，Frontier 的 Fixed-PD TPOT winner 从 TP8 翻到 TP4，mean/p90 TPOT 与 E2E selection regret 降到 10% 以内。
+  - H2：collective mismatch 只解释部分误差。换 profile 后 TP8 仍是 Frontier winner，Fixed-PD TPOT/E2E regret 仍超过 10%；下一主因应定位 decode batch/state-conditioned MoE composition。
+- **事前预测：** 当前 Frontier 在 Fixed-PD 上预测 TP4/TP8 mean TPOT 为 87.77/61.59 ms，TP8 有 26.19 ms 优势；真实 TP4/TP8 为 21.04/27.99 ms。若新的 TP4/TP8 collective profile 使这个 26.19 ms 的 simulator margin 反转，则支持 H1；若不能，则支持 H2。
+- **判定规则：** 只以 frozen simulator rerun 的 winner 与真实 frozen surface 计算 selection regret。绝对 latency ratio 作为 secondary metric，不用它替代 selection verdict。
+
+## Setup
+
+- **自变量：**
+  - A0：当前 `measured-allreduce.csv`；TP4 是 Qwen30 hidden=2048 的旧实测，且 profiler 只检查 FlashInfer 可用、没有证明每个 payload 的实际 dispatch；TP8 无行并静默 analytical fallback。
+  - A1：Qwen235 serving-matched piecewise collective profile；TP4/TP8 都在 dash0 H20、vLLM 0.20.0 commit `88d34c640...` 上实测。Frozen server logs 证明真机同时使用 `disable_custom_all_reduce=true` 与 FlashInfer-TRTLLM `allreduce_rms` fusion；profile 对 fusion-eligible payload 测同一 FlashInfer communicator，对阈值外 payload 测真实 PyNCCL/symmetric fallback。
+- **控制变量：** Frontier commit、Qwen235 operator profiles、runtime contract、四类 frozen traces、候选配置、MNS/MBT、prefix policy、real results 与分析脚本全部不变。
+- **System context：** Qwen3-235B-A22B-FP8；vLLM 0.20.0+cu129；dash0 8×H20；`{TP4/EP1, TP8/EP8} × MNS{64,128}`；MBT=8192；Frontier piecewise graph path。
+- **Workload 或 trace：** 重跑四类 simulator surface：Fixed-PD 4096→256 @ 0.2 req/s/GPU、Fixed-PO 4096→1、Trace-PD、Trace-PO；每 cell 沿用原 129-request trace。Fixed-PD 是 primary，另外三类检查 profile 替换是否引入新的 selection regression。
+- **Profile protocol：** payload 覆盖所有 Qwen235 decode graph buckets（1--256，含真实 capture sizes）、fusion 阈值两侧 `{63,64,65}` / `{255,256,257}`，以及 512--8192 prefill sizes；每个 TP 与 payload 先 warmup，再保留 3×20 个 per-rank CUDA-event samples。raw JSON 记录实际 backend dispatch、fusion byte limit、world size、dtype、payload bytes、GPU/runtime/commit 与 source hashes。TP4/TP8 使用相同 payload grid；不把 microbenchmark 直接当作 E2E 结论。
+- **Profile contract：** H20/SM90 上 vLLM 0.20 的 fusion limit 是 TP4 2 MiB、TP8 0.5 MiB，即 Q235 BF16 hidden=4096 时分别为 256/64 tokens。simulator runner 启动前解析 CSV，要求所选 configs 的每个 `TP>1` 都有有限、正值的 measured rows；缺覆盖立即失败。结果 manifest 写入 CSV SHA-256、TP coverage、row counts 与 piecewise backend 集合。决策实验禁止 analytical fallback。
+- **Baselines：** A0 current Frontier、A1 measured-profile Frontier、frozen real hardware surface。
+- **Metrics：** profile latency median/p90 与跨 rank spread；simulated mean/p90 TTFT/TPOT/E2E；winner、selection regret、tau-b（可定义时）；每个 TP 的 measured-profile hit/fallback counters。
+
+## 预期产物与 review
+
+- **预期数据：** TP4/TP8 raw collective JSON；materialized Frontier CSV + manifest；四类 A1 simulator surface；A0/A1/real comparison JSON/Markdown；profile cost ledger。
+- **Figure prototype：** `../../runs/frontier-fidelity-envelope-v1/qwen235-collective-ablation-mock.png`；左图固定真实与 A0 TPOT，并为 A1 留待测 series；右图明确“winner flip→0% regret / unchanged→33% regret”的判定。它回答 profile 修复是否足以改变配置选择。
+- **人工 review：** 通过（2026-07-19，用户明确要求“推进实验”）。
+- **Review 意见：** 保留 frozen real surface，只补真实 TP4/TP8 profile 后重跑 simulator；每个 simulator 实验必须使用真实 profile，缺失 coverage 或运行时 analytical fallback 立即失败。
+
+## Benchmark design audit（experiment-design-review）
+
+| Crime | Verdict | Severity | Evidence | Fix / gate |
+|---|---|---|---|---|
+| 用 microbenchmark 代替 E2E | PASS | — | collective profile 只作为自变量；结论来自完整 simulator surface 对 frozen real surface 的 selection regret | 保留 A0/A1/real 三方结果 |
+| calibration set 等于 evaluation set | PASS | — | A1 只测 collective operator，不使用 real E2E latency 拟合参数 | 禁止 E2E scale/calibration |
+| selective benchmarking | PASS | — | primary Fixed-PD 外，同时重跑另外三类 workload | 报告所有 16 个 simulator cells |
+| 缺失平台/版本 | PASS | — | raw/manifest 绑定 H20、vLLM commit、model、backend 与 hashes | 任一 provenance 缺失则 profile 不可采纳 |
+| 缺失方差 | NEEDS EVIDENCE | Major | 尚未执行 profile repeats | raw artifact 必须保留 per-rank repeated samples并报告 spread |
+| profile 覆盖静默降级 | FAIL（A0） | Blocking | TP8 无 measured rows，Frontier 使用 analytical fallback | A1 runner fail-fast；fallback count 必须为 0 |
+| backend/fusion 阈值未对齐 | FAIL（A0） | Blocking | 真机日志启用 FlashInfer `allreduce_rms`；vLLM 源码规定 H20 TP4/TP8 fusion limit 为 2/0.5 MiB，旧 CSV 未记录这条 piecewise contract | A1 在阈值两侧实测并记录每行 dispatch |
+
+**总体建议：** 已批准执行；coverage gate、backend match 与 provenance gate 任一不通过则 Block。
+
+## 复现信息
+
+- **Code：** AITuner branch `feature/sim`；本 card 创建时 HEAD `f4a75aa8e400ead4eb6d305178192e85940c6de6`，后续运行 commit 待填。vLLM source commit `88d34c6409e9fb3c7b8ca0c04756f061d2099eb1`。
+- **Environment：** dash0 8×H20；`/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1`；model `/home/admin/cpfs/wjh/models/Qwen/Qwen3-235B-A22B-FP8`。
+- **产物路径：** 待 review 后冻结；不得覆盖旧 campaign `/home/admin/cpfs/wjh/aituner/qwen235-v020-fourcase-20260719-r1`。
+- **已知 deviation：** `--disable-custom-all-reduce` 只关闭 vLLM custom AR，不关闭编译器的 FlashInfer `allreduce_rms` fusion。真机 TP4/TP8 日志都显示自动选择 `trtllm` workspace；旧 TP4 profiler 没有记录/执行真实 fusion-limit piecewise dispatch。A1 因此必须同时重测 TP4 与 TP8，不能只追加 TP8 行。
+
+## 结果
+
+- **观察事实：** 待运行。
+- **异常：** 待运行。
+- **含义：** 待运行。
+- **Claim update：** unchanged
+- **下一步：** 依次完成 collective profiling、profile materialization、CPU simulator rerun 与 analysis。

.research/ongoing.md

@@ -0,0 +1,72 @@
+# AITuner 研究当前状态
+
+> 2026-07-17｜写给未参与项目的读者，可直接作为 presentation 讲稿。历史过程与复现信息见 `../runs/*/` 各 experiment card、`../docs/` 各 campaign 文档。
+>
+> **2026-07-19 update：** Qwen235 四类 workload matrix 已完成；Fixed-PD 出现 33.0--37.2% TPOT 与 30.7--34.6% E2E selection regret。当前发现 Q235 simulator 输入的 collective CSV 缺 TP8，并且旧 TP4 profile 未绑定真实 `allreduce_rms` fusion-limit dispatch。active ablation 见 [`experiments/qwen235-tp8-collective-profile-ablation-20260719.md`](experiments/qwen235-tp8-collective-profile-ablation-20260719.md)。
+
+## 一眼看懂
+
+- **Topic / problem：** LLM serving 的自动、低成本配置调优（AITuner）。当前主线问题：用 simulator 给部署配置（并行度、批量上限等）排序，什么时候可信？需要补多少真机证据？算上这些成本还划算吗？
+- **Central claim：** simulator 要能帮助配置调优，必须先满足 scheduler transition 的 liveness/coverage，再满足「配置相关残差小于真机 decision margin」；前者决定 capacity 是否有定义，后者决定排序是否正确。(ID: C0)
+- **当前结论：** 早先 35 个 trace stall 不是 Frontier scheduler liveness failure：adapter 为不满 16-token 的 prefix block 错误生成了 cache identity，Frontier 又没有 fail-fast。修正为完整 block、使用真实 graph buckets/KV blocks 和 `piecewise`/`KERNEL_ONLY` profile 后，Qwen30 Trace-PD 的全部 12 个 cell 完成 129/129 request，Frontier 对 TTFT/TPOT/E2E 的 6 个 argmin 均与三次 fresh-server 真机一致；但绝对 latency 仍高估 4--511×。这只证明一个 MoE Trace-PD surface 的 selection fidelity，不能外推到 prefill-only、fixed workload 或 235B。
+- **最大 uncertainty / risk：** 这个正确选择是否依赖 Trace-PD 的 queue/margin 偶然性；Fixed-PD、Trace-P、Fixed-P 与大型 FP8 MoE 是否仍保留正确 config ranking。Q235 还缺少同栈 profile/runtime contract，不能复用旧 vLLM 0.10.2 数据。
+- **下一项 critical action：** 完成 Qwen30 其余三项 fixed/trace × PD/P surface。Fixed case 已先启动无请求 vLLM runtime-state preflight；Trace-P 的 graph-aligned Frontier 12-cell CPU surface 正在运行。Q235 先做 vLLM 0.20 TP4/TP8 compatibility gate，随后才允许新 profile。
+- **停止条件：** T1 出 verdict 且成本账本建立后：pass 且摊销论证成立 → 转向「sim 剪枝 + 真机终选」的 hybrid 机制设计；fail → 转入失败机制归因；两条路都无 insight 增量 → 收敛写作。
+
+## 核心概念
+
+- **Frontier：** 本项目使用的 simulator，属 Vidur 系（直接使用 vidur backend），加自研 FP8/MoE/EP/decode-profile 兼容补丁。
+- **Regret：** 按 simulator 排序选配置，相对真机最优配置的性能损失百分比（以每 GPU capacity 计）。primary metric：排序选对则 regret=0。
+- **τ-b（Kendall tau-b）：** simulator 排序与真机排序的秩相关，1 = 完全一致，−1 = 完全反序，0 = 无关；tie-aware。
+- **Decision margin：** 真机上头部配置之间的性能差距，即 simulator 误差的容忍带。
+- **Action-differential residual：** simulator 误差中随配置（action）不同而不同的部分。Why needed：所有配置统一偏移不影响排序，只有差异化残差才可能穿过 margin 改变选择——这解释了「绝对误差 33%」与「排序全对」为何可以同时成立。
+- **Capacity bracket：** 真机 anchor 为候选配置的 capacity 划出的上下界；「bracket 不反转」指未测的负载点不可能推翻 top 选择。
+- **Decision-valid coverage：** simulator 能从初始状态推进到所有请求完成，并为 config×workload cell 产生合法 SLO metric 的比例。若 reachable nonterminal state 没有 enabled transition/future event，capacity 与 rank 都没有定义，不能把该 cell 当作 infeasible。
+- **Workload realism 阶梯：** prefill-only（无 decode）→ fixed-shape mixed（固定输入输出长度的混合负载）→ trace-faithful mixed（生产 trace 忠实回放）。fidelity 结论不能向更高一级外推。
+
+## Claim 层级
+
+- **Central claim：** 见「一眼看懂」。(ID: C0)
+  - **Subclaim：** zero-shot 排序失败是真实现象。(ID: C1；supported)
+    - 30B 纯 profile 驱动的 regret 为 25.63%（τ-b=0）；另一 throughput-proxy 评测口径下为 30.46%。Boundary：均发生在 capacity-point + SLO-gated selection——恰是 Vidur 论文自己声明预测误差会爆炸、评测刻意回避的 regime（见 claim map）。
+  - **Subclaim：** 少量结构化的真机证据可以恢复低 regret 排序。(ID: C2)
+    - **Hypothesis（decision-bearing）：** trace-faithful 回放下，同栈 profile + 真机 KV capacity + 兼容补丁、且不做逐案例端到端校准的 Frontier，能满足 gate：regret ≤5% ∧ τ-b ≥0.8 ∧ bracket 不反转。(ID: H1；weakened)
+      - **Supporting：** 235B prefill-only regret=0；235B fixed-shape mixed 的 top set 全中；30B 加 per-TP 校准后 regret 0.76%（但这是外部端到端 scale 给出的上界，不是原生 profile 保真度）。
+      - **Counterevidence：** 修正 prefix trace contract 后的 TP2/MNS16 `none`-graph run 完成但 p50 TPOT 约 96 ms，真机为约 14 ms；然而该比较尚未对齐 real vLLM 的 `FULL_AND_PIECEWISE` graph path。
+      - **下一项 discriminative experiment：** 补齐 `KERNEL_ONLY` graph family并以 `piecewise` 重跑相同 trace；若 full surface 仍错，graph omission 不再是可用解释。
+    - **Hypothesis（机制，active）：** 误差机制是 action-conditioned residual——执行状态的转移（并行拓扑、kernel family、graph mode、batch 组成）使按算子 profile 的组合预测跨配置不可复合；残差大于 margin 时排序失败。(ID: H2；supported)
+      - **Supporting：** 三个 TP 档的端到端校准系数为 0.72/0.47/0.35，残差确实随配置剧烈变化；235B 的批量上限交互预测错误但被 2× margin 容忍；30B prefill-only 在低负载近似对齐、饱和后按 TP 反向放大，最终 τ-b=−1。
+      - **下一步：** 按 collective 通信 → batch 组成 → 调度器逐步轨迹的单变量顺序定位组合性失效点（对应 experiment card 中 A1/A2/A3 消融阶梯）。
+  - **Subclaim：** 成本论证只有在摊销前提下成立。(ID: C3)
+    - **Hypothesis（active）：** 每个 model×硬件×runtime 的一次性对齐成本，摊销到大配置面、频繁重调（引擎版本 churn 的频率证据见 claim map）或禁止在线实验的场景后，低于重复真机调优。(ID: H3；untested——分母已实测，分子未入账)
+      - **下一步：** 建 cost ledger（见「下一步」）。
+
+## 当前 critical experiment
+
+- **Question：** 生产 trace 忠实回放（prefix 打开、原始到达时间与会话结构）下，best-effort Frontier 能否满足 low-regret gate？
+- **为什么现在做：** 这是 H1 的判决实验；所有已完成的机制分解都在人工 workload 上，不能替代这个 verdict。
+- **当前状态：** Trace-PD 的 graph-aligned surface 已通过 selection gate，但绝对 latency 不通过 calibration。现在以不共享结果的 Fixed-PD、Trace-P、Fixed-P surface 检验它的泛化边界。
+- **Result → decision：** 若其它 surface 排序失败，保留 Trace-PD success 为条件化 envelope，并按 fixed/trace/prefill/decode 的差异定位 state composition；若都通过，才扩大到 Q235 或寻找 simulator 已解决范围之外的新问题。
+- **Experiment card：** [`../runs/frontier-fidelity-envelope-v1/experiment-card.md`](../runs/frontier-fidelity-envelope-v1/experiment-card.md)
+
+## Key evidence（最多 3 条）
+
+- **E1（否证「prefill-only 是充分 easy condition」，支持 H2）：** 30B BF16、去掉 decode/prefix/混合 batch 后，真机最优是 TP4（8 vs 7 req/s/GPU），simulator 却把 TP4 排最差（6 vs 8）：top set 无交集，regret 12.5%，τ-b=−1。产物：`../runs/frontier-phase-factorial-v0/results/final/`（dash0，12.07 H20-GPUh）。
+- **E2（margin 可掩盖残差，支持 C0/H2）：** 235B FP8 的 prefill-only 与 fixed-shape mixed 都选对 top set，但 simulator 漏掉并发上限（max-num-seqs）×批 token 上限（max-num-batched-tokens）的交互棋盘格，并把 10/34 个真机 anchor 误判为不可行；真机头部 margin 有 2×，残差被容忍。产物：`../runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_evidence/`、`../runs/frontier-multicase-sufficiency-v1/results/t0-final/`。
+- **E3（selection 与 calibration 分开）：** 用完整 block projection、`piecewise` 和 graph-compatible KERNEL_ONLY profile 后，Qwen30 Trace-PD 的 12/12 sim cells 完成，6 个 mean/p90 latency objective 的 argmin 都与真机一致；但 sim/real latency ratio 仍为 4--511×。产物：[`../runs/frontier-fidelity-envelope-v1/graph-piecewise-experiment-card.md`](../runs/frontier-fidelity-envelope-v1/graph-piecewise-experiment-card.md)。
+
+## 下一步（最多 3 项）
+
+- [ ] **完成 Qwen30 remaining matrix（direct progress）：** Fixed-PD、Trace-P、Fixed-P 分别冻结 Frontier/真机 12-cell surface；P-only 的 TPOT 一律为 N/A。
+- [ ] **Q235 portability gate：** 先验证 vLLM0.20 TP4/TP8 FP8 runtime 和 deadc4a profile provenance，再决定是否允许其 Fixed-P sweep。
+- [ ] **建 cost ledger：** parent H3；完成标准 = 每 case 一行（profiling GPU-h、补丁工时、校准探测、sim CPU-h），与已实测的真机调优成本同表，随每个 case 更新。
+
+## Blocker 或 anomaly
+
+- **下一启动已准备：** dash0 8×H20 当前空闲；graph-compatible attention/linear/MoE/router kernel-only profile 会先在一 GPU smoke，成功后以 3 张 GPU 并行完成 attention shards。完整 replay 仍为 CPU-only。
+- **Anomaly（保留）：** 235B pilot 中 simulator 把 10/34 个 anchor 误判为不可行——false-infeasible 是 H1 的主要威胁模式，T1 分析时须单独报告。
+- **勘误（已固定）：** 统一平台为 dash0-only，早期 dash1 出处是文档错误；fixed-shape pilot 的主 SLO（TPOT 40ms）无判别力，150ms 是事后明示的敏感性分析，不得写成盲选的 primary。
+
+## Related work
+
+- Claim map：[`../docs/simulator-claim-map-20260716.md`](../docs/simulator-claim-map-20260716.md)。核心缺口：capacity-point + SLO-gated selection 的 regret 无人用真机 ground-truth 面验证过；alignment 成本无人与真机调优成本放进同一张表比较。

runs/frontier-fidelity-envelope-v1/analyze_qwen235_v020_campaign.py

@@ -97,14 +97,18 @@ def compare_metric(real: dict, sim: dict, metric: str):
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--campaign-root", type=Path, required=True)
+    parser.add_argument("--real-root", type=Path)
+    parser.add_argument("--sim-root", type=Path)
     parser.add_argument("--json-output", type=Path, required=True)
     parser.add_argument("--markdown-output", type=Path, required=True)
     args = parser.parse_args()
+    real_root = args.real_root or args.campaign_root
+    sim_root = args.sim_root or args.campaign_root
     cases = {}
     lines = ["# Qwen235 vLLM 0.20 Frontier vs real", "", "| case | metric | Frontier winner | real winner | match | regret | pair agreement |", "|---|---|---|---|---:|---:|---:|"]
     for case in CASES:
-        real = real_surface(args.campaign_root, case)
-        sim = sim_surface(args.campaign_root, case)
+        real = real_surface(real_root, case)
+        sim = sim_surface(sim_root, case)
         metrics = ["ttft_mean_ms", "ttft_p90_ms", "e2e_mean_ms", "e2e_p90_ms"]
         if case.endswith("pd"):
             metrics[2:2] = ["tpot_mean_ms", "tpot_p90_ms"]
@@ -118,7 +122,12 @@ def main() -> None:
                 f"{item['pair_direction_agreement']:.1%} |"
             )
         cases[case] = {"real": real, "sim": sim, "comparison": comparisons}
-    payload = {"schema": "qwen235-v020-simulator-real-comparison-v1", "cases": cases}
+    payload = {
+        "schema": "qwen235-v020-simulator-real-comparison-v1",
+        "real_root": str(real_root.resolve()),
+        "sim_root": str(sim_root.resolve()),
+        "cases": cases,
+    }
     args.json_output.parent.mkdir(parents=True, exist_ok=True)
     args.json_output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
     args.markdown_output.write_text("\n".join(lines) + "\n")

runs/frontier-fidelity-envelope-v1/jobs_qwen235_collective_profile.toml

@@ -0,0 +1,45 @@
+version = 1
+
+[[jobs]]
+name = "qwen235-vllm020-serving-allreduce-tp4-20260719-v1"
+gpus = 4
+gpu_model = "H20"
+hosts = ["dash0"]
+command = "cd /home/admin/cpfs/wjh/aituner/aituner/runs/frontier-qwen30-vllm020-profile-v1 && timeout --signal=TERM --kill-after=30s 1020 bash run_allreduce_profile.sh"
+artifacts = ["artifacts/qwen235-serving-allreduce-tp4-20260719-v1"]
+
+[jobs.env]
+XDG_CACHE_HOME = "/tmp/wjh/.cache"
+VLLM_CACHE_ROOT = "/tmp/wjh/.cache/vllm"
+TP = "4"
+NUM_TOKENS = "1 2 4 8 16 24 32 40 48 56 63 64 65 72 80 88 96 104 112 120 128 136 144 152 160 168 176 184 192 200 208 216 224 232 240 248 255 256 257 512 1024 2048 4096 8192"
+HIDDEN_DIM = "4096"
+COLLECTIVE_CONTRACT = "qwen235-serving-projected"
+TRIALS = "3"
+REPEATS = "20"
+MODEL = "/home/admin/cpfs/wjh/models/Qwen/Qwen3-235B-A22B-FP8"
+OUTPUT_ROOT = "/home/admin/cpfs/wjh/aituner/gpu-fleet-fidelity-envelope-v1/artifacts/qwen235-serving-allreduce-tp4-20260719-v1"
+VENV_ROOT = "/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1"
+VLLM_SOURCE = "/home/admin/cpfs/wjh/agentic-kv/third_party/vllm_v20_build"
+
+[[jobs]]
+name = "qwen235-vllm020-serving-allreduce-tp8-20260719-v1"
+gpus = 8
+gpu_model = "H20"
+hosts = ["dash0"]
+command = "cd /home/admin/cpfs/wjh/aituner/aituner/runs/frontier-qwen30-vllm020-profile-v1 && timeout --signal=TERM --kill-after=30s 1020 bash run_allreduce_profile.sh"
+artifacts = ["artifacts/qwen235-serving-allreduce-tp8-20260719-v1"]
+
+[jobs.env]
+XDG_CACHE_HOME = "/tmp/wjh/.cache"
+VLLM_CACHE_ROOT = "/tmp/wjh/.cache/vllm"
+TP = "8"
+NUM_TOKENS = "1 2 4 8 16 24 32 40 48 56 63 64 65 72 80 88 96 104 112 120 128 136 144 152 160 168 176 184 192 200 208 216 224 232 240 248 255 256 257 512 1024 2048 4096 8192"
+HIDDEN_DIM = "4096"
+COLLECTIVE_CONTRACT = "qwen235-serving-projected"
+TRIALS = "3"
+REPEATS = "20"
+MODEL = "/home/admin/cpfs/wjh/models/Qwen/Qwen3-235B-A22B-FP8"
+OUTPUT_ROOT = "/home/admin/cpfs/wjh/aituner/gpu-fleet-fidelity-envelope-v1/artifacts/qwen235-serving-allreduce-tp8-20260719-v1"
+VENV_ROOT = "/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1"
+VLLM_SOURCE = "/home/admin/cpfs/wjh/agentic-kv/third_party/vllm_v20_build"

runs/frontier-fidelity-envelope-v1/materialize_qwen235_v020_allreduce.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+"""Validate Qwen235 serving collectives and materialize Frontier's CC CSV."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+
+VLLM_COMMIT = "88d34c6409e9fb3c7b8ca0c04756f061d2099eb1"
+TOKEN_POINTS = (
+    1, 2, 4, 8, 16, 24, 32, 40, 48, 56, 63, 64, 65, 72, 80, 88, 96,
+    104, 112, 120, 128, 136, 144, 152, 160, 168, 176, 184, 192, 200,
+    208, 216, 224, 232, 240, 248, 255, 256, 257, 512, 1024, 2048, 4096,
+    8192,
+)
+TP_SIZES = (4, 8)
+HIDDEN_DIM = 4096
+ALLOWED_BACKENDS = {
+    "flashinfer_trtllm_fused_projection",
+    "pynccl_symmetric_with_copy",
+    "torch_symmetric_memory",
+    "pynccl",
+}
+FIELDS = (
+    "time_stats.all_reduce.min",
+    "time_stats.all_reduce.max",
+    "time_stats.all_reduce.mean",
+    "time_stats.all_reduce.median",
+    "time_stats.all_reduce.std",
+    "rank",
+    "num_workers",
+    "size",
+    "collective",
+    "devices_per_node",
+    "max_devices_per_node",
+)
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def materialize(
+    input_paths: list[Path], output_path: Path, manifest_path: Path
+) -> dict[str, Any]:
+    if len(input_paths) != 2:
+        raise ValueError("exactly two raw inputs are required for TP4 and TP8")
+    rows = []
+    seen = set()
+    backends_by_tp: dict[int, set[str]] = {}
+    environments = []
+    for path in input_paths:
+        payload = json.loads(path.read_text())
+        if payload.get("schema_version") != "vllm020_allreduce_raw.v2":
+            raise ValueError(f"unexpected raw collective schema in {path}")
+        environment = payload["environment"]
+        if environment.get("vllm_version") != "0.20.0":
+            raise ValueError(f"vLLM version mismatch in {path}")
+        if environment.get("vllm_source_commit") != VLLM_COMMIT:
+            raise ValueError(f"vLLM commit mismatch in {path}")
+        if environment.get("gpu") != "NVIDIA H20":
+            raise ValueError(f"GPU mismatch in {path}: {environment.get('gpu')!r}")
+        if Path(environment.get("model", "")).name != "Qwen3-235B-A22B-FP8":
+            raise ValueError(f"model mismatch in {path}")
+        if environment.get("collective_contract") != "qwen235-serving-projected":
+            raise ValueError(f"collective contract mismatch in {path}")
+        if environment.get("disable_custom_all_reduce") is not True:
+            raise ValueError(f"custom all-reduce was not disabled in {path}")
+        backend_env = environment.get("backend_env", {})
+        if backend_env.get("VLLM_ALLREDUCE_USE_FLASHINFER") != "1":
+            raise ValueError(f"FlashInfer projection was not enabled in {path}")
+        if backend_env.get("VLLM_ALLREDUCE_USE_SYMM_MEM") != "1":
+            raise ValueError(f"symmetric-memory contract mismatch in {path}")
+        environments.append(environment)
+
+        for source in payload["rows"]:
+            tp = int(source["tensor_parallel_size"])
+            tokens = int(source["num_tokens"])
+            key = (tp, tokens)
+            if key in seen:
+                raise ValueError(f"duplicate collective row: {key}")
+            seen.add(key)
+            if tp not in TP_SIZES:
+                raise ValueError(f"unexpected TP: {tp}")
+            if int(source["hidden_dim"]) != HIDDEN_DIM:
+                raise ValueError(f"hidden-dim mismatch for {key}")
+            expected_bytes = tokens * HIDDEN_DIM * 2
+            if int(source["payload_bytes"]) != expected_bytes:
+                raise ValueError(f"payload mismatch for {key}")
+            backend = source["selected_backend"]
+            if backend not in ALLOWED_BACKENDS:
+                raise ValueError(f"unexpected backend for {key}: {backend!r}")
+            fusion_limit = {4: 2 * 1024 * 1024, 8: 512 * 1024}[tp]
+            if int(source["real_fusion_limit_bytes"]) != fusion_limit:
+                raise ValueError(f"fusion limit mismatch for {key}")
+            expected_backend = (
+                "flashinfer_trtllm_fused_projection"
+                if expected_bytes <= fusion_limit
+                else None
+            )
+            if expected_backend is not None and backend != expected_backend:
+                raise ValueError(f"missing fused projection for {key}")
+            if expected_backend is None and backend == "flashinfer_trtllm_fused_projection":
+                raise ValueError(f"fused projection exceeds runtime limit for {key}")
+            backends_by_tp.setdefault(tp, set()).add(backend)
+            median = float(source["critical_path_median_ms"])
+            if not math.isfinite(median) or median <= 0:
+                raise ValueError(f"invalid critical-path median for {key}")
+            if int(source.get("trials", 0)) != 3 or int(
+                source.get("repeats_per_trial", 0)
+            ) != 20:
+                raise ValueError(f"repeat contract mismatch for {key}")
+            samples = source.get("per_trial_rank_samples_ms", [])
+            if (
+                len(samples) != 3
+                or any(len(trial) != tp for trial in samples)
+                or any(
+                    len(rank_samples) != 20
+                    or any(
+                        not math.isfinite(float(value)) or float(value) <= 0
+                        for value in rank_samples
+                    )
+                    for trial in samples
+                    for rank_samples in trial
+                )
+            ):
+                raise ValueError(f"raw sample coverage mismatch for {key}")
+            critical_samples = [
+                max(float(trial[rank][repeat]) for rank in range(tp))
+                for trial in samples
+                for repeat in range(20)
+            ]
+            ordered = sorted(critical_samples)
+            measured_median = (ordered[29] + ordered[30]) / 2
+            if not math.isclose(median, measured_median, rel_tol=1e-6, abs_tol=1e-9):
+                raise ValueError(f"critical-path median/sample mismatch for {key}")
+            rows.append(
+                {
+                    "time_stats.all_reduce.min": median,
+                    "time_stats.all_reduce.max": median,
+                    "time_stats.all_reduce.mean": median,
+                    "time_stats.all_reduce.median": median,
+                    "time_stats.all_reduce.std": 0.0,
+                    "rank": 0,
+                    "num_workers": tp,
+                    "size": expected_bytes,
+                    "collective": "all_reduce",
+                    "devices_per_node": tp,
+                    "max_devices_per_node": 8,
+                }
+            )
+
+    expected = {(tp, tokens) for tp in TP_SIZES for tokens in TOKEN_POINTS}
+    if seen != expected:
+        raise ValueError(
+            f"collective coverage mismatch: missing={expected - seen}, extra={seen - expected}"
+        )
+    if output_path.exists() or manifest_path.exists():
+        raise FileExistsError("refusing to overwrite immutable collective profile")
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open("w", newline="") as output:
+        writer = csv.DictWriter(output, fieldnames=FIELDS, lineterminator="\n")
+        writer.writeheader()
+        writer.writerows(sorted(rows, key=lambda row: (row["num_workers"], row["size"])))
+
+    manifest = {
+        "schema": "qwen235-v020-serving-allreduce-profile-v1",
+        "inputs": {str(path.resolve()): sha256(path) for path in input_paths},
+        "output": str(output_path.resolve()),
+        "output_sha256": sha256(output_path),
+        "vllm_source_commit": VLLM_COMMIT,
+        "hardware": "NVIDIA H20",
+        "model": "Qwen3-235B-A22B-FP8",
+        "dtype": "bfloat16",
+        "hidden_dim": HIDDEN_DIM,
+        "collective_contract": "qwen235-serving-projected",
+        "tp_coverage": list(TP_SIZES),
+        "token_points": list(TOKEN_POINTS),
+        "rows": len(rows),
+        "observed_backends_by_tp": {
+            str(tp): sorted(backends) for tp, backends in sorted(backends_by_tp.items())
+        },
+        "frontier_target": "time_stats.all_reduce.median",
+        "unused_stat_policy": "repeat critical_path_median; std=0",
+        "raw_environments": environments,
+    }
+    manifest_path.parent.mkdir(parents=True, exist_ok=True)
+    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
+    return manifest
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", type=Path, action="append", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--manifest", type=Path, required=True)
+    args = parser.parse_args()
+    manifest = materialize(args.input, args.output, args.manifest)
+    print(json.dumps(manifest, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

runs/frontier-fidelity-envelope-v1/plot_qwen235_collective_ablation_mock.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Render the preregistered Qwen235 collective-profile ablation layout."""
+
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+
+OUT = Path(__file__).with_name("qwen235-collective-ablation-mock.png")
+
+
+def main() -> None:
+    configs = ["TP4 / EP1", "TP8 / EP8"]
+    real_tpot = np.array([21.0442, 27.9942])
+    current_frontier_tpot = np.array([87.7735, 61.5878])
+
+    fig, (ax0, ax1) = plt.subplots(1, 2, figsize=(10.8, 4.4))
+    x = np.arange(len(configs))
+    width = 0.34
+    ax0.bar(
+        x - width / 2,
+        real_tpot,
+        width,
+        color="#4c78a8",
+        label="Frozen real",
+    )
+    ax0.bar(
+        x + width / 2,
+        current_frontier_tpot,
+        width,
+        color="#e45756",
+        label="A0 current Frontier",
+    )
+    for index in x:
+        ax0.text(
+            index,
+            max(real_tpot[index], current_frontier_tpot[index]) + 4,
+            "A1 measured-profile\nresult: TBD",
+            ha="center",
+            va="bottom",
+            fontsize=8,
+            color="#222222",
+        )
+    ax0.set_xticks(x, configs)
+    ax0.set_ylim(0, 112)
+    ax0.set_ylabel("Fixed-PD mean TPOT (ms)")
+    ax0.set_title("A. Does A1 reverse the simulated TP margin?")
+    ax0.legend(frameon=False, fontsize=8)
+    ax0.grid(axis="y", alpha=0.25)
+
+    outcomes = ["A0 current", "A1 flips\nto TP4", "A1 stays\nat TP8"]
+    regret = [33.0, 0.0, 33.0]
+    colors = ["#e45756", "#54a24b", "#f2cf5b"]
+    bars = ax1.bar(outcomes, regret, color=colors, width=0.68)
+    ax1.axhline(10.0, color="#222222", linestyle="--", linewidth=1.2, label="10% gate")
+    ax1.bar_label(bars, labels=["33%", "0%", "33%"], padding=3, fontsize=9)
+    ax1.set_ylim(0, 42)
+    ax1.set_ylabel("Real TPOT selection regret (%)")
+    ax1.set_title("B. Preregistered decision outcomes")
+    ax1.legend(frameon=False, fontsize=8)
+    ax1.grid(axis="y", alpha=0.25)
+
+    fig.suptitle(
+        "SCHEMATIC / preregistered layout — A1 values are not yet measured",
+        fontsize=11,
+    )
+    fig.tight_layout()
+    fig.savefig(OUT, dpi=180, bbox_inches="tight")
+    print(OUT)
+
+
+if __name__ == "__main__":
+    main()

runs/frontier-fidelity-envelope-v1/run_frontier_qwen235_v020_surface.py

@@ -158,6 +158,10 @@ def main() -> None:
         selected = [config for config in GRID if config.name in wanted]
         if {config.name for config in selected} != wanted:
             raise ValueError(f"unknown configs: {wanted - {config.name for config in selected}}")
+    allreduce_profile = Q30.validate_allreduce_profile(
+        args.allreduce_csv, {config.tp for config in selected if config.tp > 1}
+    )
+    allreduce_csv_sha256 = BASE.sha256(args.allreduce_csv)
 
     builder = BASE.load_module(
         "qwen235_frontier_builder", args.replayserve_root / "tools/run_frontier_sweep.py"
@@ -182,7 +186,11 @@ def main() -> None:
             result_path = run_dir / "result.json"
             if args.resume and result_path.is_file():
                 previous = json.loads(result_path.read_text())
-                if previous.get("status") == "completed":
+                if (
+                    previous.get("status") == "completed"
+                    and previous.get("allreduce_csv_sha256")
+                    == allreduce_csv_sha256
+                ):
                     results.append(previous)
                     continue
             run_dir.mkdir(parents=True, exist_ok=True)
@@ -217,8 +225,16 @@ def main() -> None:
                 "trace": Q30.trace_manifest_entry(trace),
                 "elapsed_seconds": time.time() - started,
                 "returncode": completed.returncode,
+                "allreduce_csv_sha256": allreduce_csv_sha256,
             }
-            if completed.returncode == 0:
+            fallback_evidence = Q30.collective_fallback_evidence(run_dir)
+            if completed.returncode == 0 and fallback_evidence:
+                result.update(
+                    status="failed",
+                    failure_class="collective_profile_fallback",
+                    collective_fallback_evidence=fallback_evidence,
+                )
+            elif completed.returncode == 0:
                 metrics = Q30.find_request_metrics(run_dir)
                 result.update(status="completed", metrics=Q30.score(metrics, trace["shapes"]), request_metrics_sha256=BASE.sha256(metrics))
             else:
@@ -233,6 +249,12 @@ def main() -> None:
         "frontier_commit": subprocess.check_output(["git", "-C", str(args.frontier_source), "rev-parse", "HEAD"], text=True).strip(),
         "profiles": {name: BASE.sha256(path) for name, path in paths.items()},
         "runtime_contract_sha256": BASE.sha256(args.runtime_contract),
+        "collective": {
+            "backend": "vidur",
+            "allreduce_csv": str(args.allreduce_csv),
+            "allreduce_csv_sha256": allreduce_csv_sha256,
+            "measured_profile_validation": allreduce_profile,
+        },
         "prefix_caching": args.prefix_caching,
         "results": results,
     }

runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py

@@ -121,6 +121,91 @@ def percentile(values: list[float], fraction: float) -> float | None:
     return ordered[math.ceil(fraction * len(ordered)) - 1]
 
 
+def validate_allreduce_profile(
+    path: Path | None, required_tps: set[int]
+) -> dict[str, Any] | None:
+    """Fail before simulation when a measured collective profile misses a TP."""
+    if not required_tps and path is None:
+        return None
+    if path is None:
+        raise ValueError(
+            f"measured all-reduce profile is required for TP coverage {sorted(required_tps)}"
+        )
+    if not path.is_file():
+        raise FileNotFoundError(path)
+
+    required_fields = {
+        "time_stats.all_reduce.median",
+        "num_workers",
+        "size",
+        "collective",
+    }
+    rows_by_tp: dict[int, int] = {}
+    sizes_by_tp: dict[int, set[int]] = {}
+    seen: set[tuple[int, int]] = set()
+    with path.open(newline="") as handle:
+        reader = csv.DictReader(handle)
+        missing_fields = required_fields - set(reader.fieldnames or ())
+        if missing_fields:
+            raise ValueError(
+                f"all-reduce profile schema missing fields: {sorted(missing_fields)}"
+            )
+        for line_number, row in enumerate(reader, start=2):
+            if row["collective"] != "all_reduce":
+                raise ValueError(
+                    f"unexpected collective at {path}:{line_number}: {row['collective']!r}"
+                )
+            tp = int(row["num_workers"])
+            size = int(row["size"])
+            median = float(row["time_stats.all_reduce.median"])
+            if size <= 0 or not math.isfinite(median) or median <= 0:
+                raise ValueError(
+                    f"invalid measured all-reduce row at {path}:{line_number}"
+                )
+            key = (tp, size)
+            if key in seen:
+                raise ValueError(f"duplicate measured all-reduce row: TP{tp}, size={size}")
+            seen.add(key)
+            rows_by_tp[tp] = rows_by_tp.get(tp, 0) + 1
+            sizes_by_tp.setdefault(tp, set()).add(size)
+
+    missing_tps = required_tps - set(rows_by_tp)
+    if missing_tps:
+        raise ValueError(
+            "measured all-reduce profile coverage missing "
+            f"TPs {sorted(missing_tps)}; available={sorted(rows_by_tp)}; "
+            "analytical fallback is forbidden"
+        )
+    return {
+        "required_tp_coverage": sorted(required_tps),
+        "available_tp_coverage": sorted(rows_by_tp),
+        "rows_by_tp": {str(tp): rows_by_tp[tp] for tp in sorted(rows_by_tp)},
+        "size_range_by_tp": {
+            str(tp): [min(sizes), max(sizes)]
+            for tp, sizes in sorted(sizes_by_tp.items())
+        },
+    }
+
+
+def collective_fallback_evidence(run_dir: Path) -> list[dict[str, Any]]:
+    patterns = (
+        "all-reduce profiling data is empty after filtering",
+        "analytical fallback",
+        "fallback to analytical",
+    )
+    evidence = []
+    for name in ("stdout.log", "stderr.log"):
+        path = run_dir / name
+        for line_number, line in enumerate(
+            path.read_text(errors="replace").splitlines(), start=1
+        ):
+            if any(pattern in line.lower() for pattern in patterns):
+                evidence.append(
+                    {"log": name, "line_number": line_number, "text": line.strip()}
+                )
+    return evidence
+
+
 def manifest_offered_rate(path: Path) -> tuple[float | None, str | None]:
     manifest_path = path.with_name("manifest.json")
     if not manifest_path.is_file():
@@ -415,6 +500,14 @@ def main() -> None:
         selected = [config for config in BASE.GRID if config.name in wanted]
         if {config.name for config in selected} != wanted:
             raise ValueError(f"unknown configs: {wanted - {config.name for config in selected}}")
+    allreduce_profile = None
+    if args.cc_backend == "vidur":
+        allreduce_profile = validate_allreduce_profile(
+            args.allreduce_csv, {config.tp for config in selected if config.tp > 1}
+        )
+    allreduce_csv_sha256 = (
+        BASE.sha256(args.allreduce_csv) if args.allreduce_csv else None
+    )
     paths = BASE.profile_paths(args.profile_root)
     coverage = BASE.validate_profile(paths)
     kernel_paths = None
@@ -477,7 +570,13 @@ def main() -> None:
             run_dir = args.output_root / "runs" / config.name / trace["label"]
             result_path = run_dir / "result.json"
             if args.resume and result_path.is_file():
-                loads.append(json.loads(result_path.read_text()))
+                previous = json.loads(result_path.read_text())
+                if (
+                    previous.get("status") == "completed"
+                    and previous.get("allreduce_csv_sha256")
+                    == allreduce_csv_sha256
+                ):
+                    loads.append(previous)
                     continue
             run_dir.mkdir(parents=True, exist_ok=True)
             command = builder.build_frontier_command(
@@ -520,6 +619,39 @@ def main() -> None:
                     timeout=args.timeout_seconds,
                     check=False,
                 )
+            fallback_evidence = (
+                collective_fallback_evidence(run_dir)
+                if args.cc_backend == "vidur"
+                else []
+            )
+            if completed.returncode == 0 and fallback_evidence:
+                result = {
+                    "status": "frontier_failed",
+                    "failure_kind": "collective_profile_fallback",
+                    "returncode": completed.returncode,
+                    "config": {
+                        "tp": config.tp,
+                        "mns": config.mns,
+                        "name": config.name,
+                    },
+                    "trace_label": trace["label"],
+                    "offered_request_rate": trace["offered_request_rate"],
+                    "offered_request_rate_per_gpu": (
+                        trace["offered_request_rate"] / config.tp
+                    ),
+                    "request_count": trace["requests"],
+                    "elapsed_seconds": time.time() - started,
+                    "trace_sha256": trace["sha256"],
+                    "collective_fallback_evidence": fallback_evidence,
+                    "allreduce_csv_sha256": allreduce_csv_sha256,
+                }
+                BASE.write_json(result_path, result)
+                loads.append(result)
+                if not args.continue_on_failure:
+                    raise RuntimeError(
+                        f"measured collective fallback for {config.name}/{trace['label']}"
+                    )
+                continue
             if completed.returncode != 0:
                 stderr_path = run_dir / "stderr.log"
                 result = {
@@ -542,6 +674,7 @@ def main() -> None:
                     "elapsed_seconds": time.time() - started,
                     "trace_sha256": trace["sha256"],
                     "stderr_sha256": BASE.sha256(stderr_path),
+                    "allreduce_csv_sha256": allreduce_csv_sha256,
                 }
                 BASE.write_json(result_path, result)
                 loads.append(result)
@@ -575,6 +708,7 @@ def main() -> None:
                 "trace_sha256": trace["sha256"],
                 "request_metrics_sha256": BASE.sha256(metrics),
                 "score": score(metrics, trace["shapes"]),
+                "allreduce_csv_sha256": allreduce_csv_sha256,
             }
             BASE.write_json(result_path, result)
             loads.append(result)
@@ -720,7 +854,8 @@ def main() -> None:
         "collective": {
             "backend": args.cc_backend,
             "allreduce_csv": str(args.allreduce_csv) if args.allreduce_csv else None,
-            "allreduce_csv_sha256": BASE.sha256(args.allreduce_csv) if args.allreduce_csv else None,
+            "allreduce_csv_sha256": allreduce_csv_sha256,
+            "measured_profile_validation": allreduce_profile,
         },
         "traces": [trace_manifest_entry(trace) for trace in traces],
         "config_results": config_results,

runs/frontier-fidelity-envelope-v1/run_qwen235_collective_profile_ablation.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+BASE_CAMPAIGN_ROOT="${BASE_CAMPAIGN_ROOT:?BASE_CAMPAIGN_ROOT is required}"
+OUTPUT_ROOT="${OUTPUT_ROOT:?OUTPUT_ROOT is required}"
+ALLREDUCE_CSV="${ALLREDUCE_CSV:?ALLREDUCE_CSV is required}"
+RUNNER_DIR="${RUNNER_DIR:-$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)}"
+PROFILE_ROOT="${PROFILE_ROOT:-/home/admin/cpfs/wjh/aituner/qwen235-v020-profiles-20260719-r1/frozen}"
+FRONTIER_SOURCE="${FRONTIER_SOURCE:-/home/admin/cpfs/wjh/aituner/frontier-q235-v020-5b953f5}"
+REPLAYSERVE_ROOT="${REPLAYSERVE_ROOT:-/home/admin/cpfs/wjh/replayserve}"
+VENV_ROOT="${VENV_ROOT:-/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1}"
+PYTHON_DEPS="${PYTHON_DEPS:-${VENV_ROOT}/lib/python3.12/site-packages}"
+
+mkdir -p "${OUTPUT_ROOT}"/{provenance,sim,analysis}
+exec > >(tee -a "${OUTPUT_ROOT}/controller.log") 2>&1
+echo "Q235_COLLECTIVE_ABLATION_LAUNCH_ECHO host=$(hostname) base=${BASE_CAMPAIGN_ROOT} simulator=Frontier-6e8e0d8 configs={TP4/EP1,TP8/EP8}xMNS{64,128} cases={fixed-pd,fixed-po,trace-pd,trace-po} real_rerun=false measured_collective=${ALLREDUCE_CSV} analytical_fallback=forbidden expected_wall=1-6h output=${OUTPUT_ROOT}"
+date -u +START_UTC=%Y-%m-%dT%H:%M:%SZ
+test "$(git -C "${FRONTIER_SOURCE}" rev-parse HEAD)" = "6e8e0d845bceff11b0b62cb29df3a1a93411fdd4"
+sha256sum "${BASH_SOURCE[0]}" "${ALLREDUCE_CSV}" \
+  "${PROFILE_ROOT}/manifest.json" \
+  "${BASE_CAMPAIGN_ROOT}/provenance/runtime-contract.json" \
+  > "${OUTPUT_ROOT}/provenance/input.sha256"
+git -C "${RUNNER_DIR}" rev-parse HEAD > "${OUTPUT_ROOT}/provenance/aituner.commit"
+git -C "${FRONTIER_SOURCE}" rev-parse HEAD > "${OUTPUT_ROOT}/provenance/frontier.commit"
+
+run_sim() {
+  local case_name="$1" prefix_flag="$2"
+  "${VENV_ROOT}/bin/python" "${RUNNER_DIR}/run_frontier_qwen235_v020_surface.py" \
+    --frontier-source "${FRONTIER_SOURCE}" --replayserve-root "${REPLAYSERVE_ROOT}" \
+    --profile-root "${PROFILE_ROOT}" --python-deps "${PYTHON_DEPS}" \
+    --output-root "${OUTPUT_ROOT}/sim/${case_name}" \
+    --runtime-contract "${BASE_CAMPAIGN_ROOT}/provenance/runtime-contract.json" \
+    --trace-tp "4=${BASE_CAMPAIGN_ROOT}/traces/${case_name}/tp4/public/frontier.csv" \
+    --trace-tp "8=${BASE_CAMPAIGN_ROOT}/traces/${case_name}/tp8/public/frontier.csv" \
+    "${prefix_flag}" --allreduce-csv "${ALLREDUCE_CSV}" --resume
+}
+
+run_sim fixed-pd --no-prefix-caching
+run_sim fixed-po --no-prefix-caching
+run_sim trace-pd --prefix-caching
+run_sim trace-po --prefix-caching
+
+"${VENV_ROOT}/bin/python" "${RUNNER_DIR}/analyze_qwen235_v020_campaign.py" \
+  --campaign-root "${BASE_CAMPAIGN_ROOT}" --sim-root "${OUTPUT_ROOT}" \
+  --json-output "${OUTPUT_ROOT}/analysis/comparison.json" \
+  --markdown-output "${OUTPUT_ROOT}/analysis/comparison.md"
+find "${OUTPUT_ROOT}" -type f ! -path '*/provenance/artifacts.sha256' -print0 \
+  | sort -z | xargs -0 sha256sum > "${OUTPUT_ROOT}/provenance/artifacts.sha256"
+date -u +END_UTC=%Y-%m-%dT%H:%M:%SZ
+echo Q235_COLLECTIVE_ABLATION_COMPLETE

runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py

@@ -26,6 +26,28 @@ def load(name: str):
 
 
 class FidelityEnvelopeTest(unittest.TestCase):
+    @staticmethod
+    def write_allreduce_csv(path: Path, tps: tuple[int, ...]) -> None:
+        fields = [
+            "time_stats.all_reduce.median",
+            "num_workers",
+            "size",
+            "collective",
+        ]
+        with path.open("w", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n")
+            writer.writeheader()
+            for tp in tps:
+                for size in (8192, 65536):
+                    writer.writerow(
+                        {
+                            "time_stats.all_reduce.median": 0.1 * tp,
+                            "num_workers": tp,
+                            "size": size,
+                            "collective": "all_reduce",
+                        }
+                    )
+
     def test_block_identities_are_parent_sensitive_and_prefix_stable(self) -> None:
         module = load("prepare_exact_trace.py")
         prefix = list(range(32))
@@ -86,6 +108,103 @@ class FidelityEnvelopeTest(unittest.TestCase):
                 str(4 + 8192 / 1000),
             )
 
+    def test_allreduce_profile_requires_every_selected_tp(self) -> None:
+        module = load("run_frontier_qwen30_exact_trace_surface.py")
+        with tempfile.TemporaryDirectory() as temporary:
+            profile = Path(temporary) / "allreduce.csv"
+            self.write_allreduce_csv(profile, (4,))
+            with self.assertRaisesRegex(
+                ValueError, "coverage missing TPs \\[8\\].*fallback is forbidden"
+            ):
+                module.validate_allreduce_profile(profile, {4, 8})
+
+            self.write_allreduce_csv(profile, (4, 8))
+            validated = module.validate_allreduce_profile(profile, {4, 8})
+            self.assertEqual(validated["required_tp_coverage"], [4, 8])
+            self.assertEqual(validated["rows_by_tp"], {"4": 2, "8": 2})
+
+    def test_collective_analytical_fallback_is_runtime_failure(self) -> None:
+        module = load("run_frontier_qwen30_exact_trace_surface.py")
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            (root / "stdout.log").write_text(
+                "All-reduce profiling data is empty after filtering\n"
+            )
+            (root / "stderr.log").write_text("unrelated warning\n")
+            evidence = module.collective_fallback_evidence(root)
+            self.assertEqual(len(evidence), 1)
+            self.assertEqual(evidence[0]["log"], "stdout.log")
+
+    def test_materialize_qwen235_allreduce_requires_serving_contract(self) -> None:
+        module = load("materialize_qwen235_v020_allreduce.py")
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            inputs = []
+            for tp in (4, 8):
+                path = root / f"tp{tp}.json"
+                rows = []
+                for tokens in module.TOKEN_POINTS:
+                    fusion_limit = {4: 2 * 1024 * 1024, 8: 512 * 1024}[tp]
+                    payload_bytes = tokens * module.HIDDEN_DIM * 2
+                    median = tp / 10 + tokens / 10000
+                    rows.append(
+                        {
+                            "tensor_parallel_size": tp,
+                            "num_tokens": tokens,
+                            "hidden_dim": module.HIDDEN_DIM,
+                            "payload_bytes": payload_bytes,
+                            "selected_backend": (
+                                "flashinfer_trtllm_fused_projection"
+                                if payload_bytes <= fusion_limit
+                                else "pynccl"
+                            ),
+                            "real_fusion_limit_bytes": fusion_limit,
+                            "trials": 3,
+                            "repeats_per_trial": 20,
+                            "per_trial_rank_samples_ms": [
+                                [[median] * 20 for _ in range(tp)] for _ in range(3)
+                            ],
+                            "critical_path_median_ms": median,
+                        }
+                    )
+                path.write_text(
+                    json.dumps(
+                        {
+                            "schema_version": "vllm020_allreduce_raw.v2",
+                            "environment": {
+                                "vllm_version": "0.20.0",
+                                "vllm_source_commit": module.VLLM_COMMIT,
+                                "gpu": "NVIDIA H20",
+                                "model": "/models/Qwen3-235B-A22B-FP8",
+                                "collective_contract": "qwen235-serving-projected",
+                                "disable_custom_all_reduce": True,
+                                "backend_env": {
+                                    "VLLM_ALLREDUCE_USE_FLASHINFER": "1",
+                                    "VLLM_ALLREDUCE_USE_SYMM_MEM": "1",
+                                },
+                            },
+                            "rows": rows,
+                        }
+                    )
+                )
+                inputs.append(path)
+            output = root / "allreduce.csv"
+            manifest_path = root / "manifest.json"
+            manifest = module.materialize(inputs, output, manifest_path)
+            self.assertEqual(manifest["tp_coverage"], [4, 8])
+            self.assertEqual(manifest["rows"], 2 * len(module.TOKEN_POINTS))
+            self.assertEqual(
+                manifest["observed_backends_by_tp"],
+                {
+                    "4": ["flashinfer_trtllm_fused_projection", "pynccl"],
+                    "8": ["flashinfer_trtllm_fused_projection", "pynccl"],
+                },
+            )
+            with output.open(newline="") as handle:
+                rows = list(csv.DictReader(handle))
+            self.assertEqual(rows[0]["num_workers"], "4")
+            self.assertEqual(rows[-1]["num_workers"], "8")
+
     def test_batch_profile_retains_one_single_request_anchor_per_tp(self) -> None:
         runner = REPO_ROOT / "runs/frontier-phase-factorial-v0/run_frontier_qwen30_prefill_surface.py"
         spec = importlib.util.spec_from_file_location("qwen30_surface_runner", runner)

runs/frontier-qwen30-vllm020-profile-v1/profile_vllm020_allreduce.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Profile vLLM 0.20 TP all-reduce and assert FlashInfer TRTLLM dispatch."""
+"""Profile the vLLM 0.20 TP all-reduce path under a frozen runtime contract."""
 
 from __future__ import annotations
 
@@ -27,8 +27,14 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--output", type=Path, required=True)
     parser.add_argument("--num-tokens", type=int, nargs="+", default=[8])
     parser.add_argument("--hidden-dim", type=int, default=2048)
+    parser.add_argument(
+        "--collective-contract",
+        choices=("flashinfer-trtllm", "qwen235-serving-projected"),
+        default="flashinfer-trtllm",
+    )
     parser.add_argument("--warmup-iters", type=int, default=3)
     parser.add_argument("--repeats", type=int, default=10)
+    parser.add_argument("--trials", type=int, default=1)
     return parser.parse_args()
 
 
@@ -55,10 +61,19 @@ def main() -> None:
     source_head = git_head(args.vllm_source)
     if source_head != VLLM_COMMIT:
         raise SystemExit(f"expected vLLM source {VLLM_COMMIT}, got {source_head}")
+    if args.repeats <= 0 or args.trials <= 0 or args.warmup_iters < 0:
+        raise SystemExit("warmup/repeats/trials must be non-negative/positive")
     if os.getenv("VLLM_ALLREDUCE_USE_FLASHINFER") != "1":
         raise SystemExit("VLLM_ALLREDUCE_USE_FLASHINFER must equal 1")
     if os.getenv("VLLM_FLASHINFER_ALLREDUCE_BACKEND") != "trtllm":
         raise SystemExit("VLLM_FLASHINFER_ALLREDUCE_BACKEND must equal trtllm")
+    expected_symm_mem = (
+        "1" if args.collective_contract == "qwen235-serving-projected" else "0"
+    )
+    if os.getenv("VLLM_ALLREDUCE_USE_SYMM_MEM") != expected_symm_mem:
+        raise SystemExit(
+            f"VLLM_ALLREDUCE_USE_SYMM_MEM must equal {expected_symm_mem}"
+        )
     if "RANK" not in os.environ or "WORLD_SIZE" not in os.environ:
         raise SystemExit("launch with torchrun")
 
@@ -67,6 +82,7 @@ def main() -> None:
         destroy_model_parallel,
         init_distributed_environment,
         initialize_model_parallel,
+        set_custom_all_reduce,
         tensor_model_parallel_all_reduce,
     )
     from vllm.distributed.parallel_state import get_tp_group
@@ -80,11 +96,15 @@ def main() -> None:
     rank = int(os.environ["RANK"])
     local_rank = int(os.environ["LOCAL_RANK"])
     world_size = int(os.environ["WORLD_SIZE"])
-    if world_size not in (2, 4):
-        raise SystemExit(f"expected TP world size 2 or 4, got {world_size}")
+    if world_size not in (2, 4, 8):
+        raise SystemExit(f"expected TP world size 2, 4, or 8, got {world_size}")
     device = torch.device(f"cuda:{local_rank}")
     torch.accelerator.set_device_index(device)
     torch.set_default_device(device)
+    disable_custom_all_reduce = (
+        args.collective_contract == "qwen235-serving-projected"
+    )
+    set_custom_all_reduce(not disable_custom_all_reduce)
     init_distributed_environment()
     model_config = ModelConfig(
         model=str(args.model),
@@ -95,11 +115,80 @@ def main() -> None:
     )
     vllm_config = VllmConfig(
         model_config=model_config,
-        parallel_config=ParallelConfig(tensor_parallel_size=world_size)
+        parallel_config=ParallelConfig(
+            tensor_parallel_size=world_size,
+            disable_custom_all_reduce=disable_custom_all_reduce,
+        )
     )
     with set_current_vllm_config(vllm_config):
         initialize_model_parallel(tensor_model_parallel_size=world_size)
 
+    from vllm.distributed.device_communicators.all_reduce_utils import (
+        should_nccl_symm_mem_allreduce,
+    )
+
+    def resolve_profile_operation(input_tensor: torch.Tensor):
+        communicator = get_tp_group().device_communicator
+        if args.collective_contract == "qwen235-serving-projected":
+            from vllm.compilation.passes.fusion.allreduce_rms_fusion import (
+                FI_ALLREDUCE_FUSION_MAX_SIZE_MB,
+            )
+
+            major, minor = torch.cuda.get_device_capability(device)
+            capability = major * 10 + minor
+            try:
+                fusion_limit_bytes = int(
+                    FI_ALLREDUCE_FUSION_MAX_SIZE_MB[capability][world_size]
+                    * 1024
+                    * 1024
+                )
+            except KeyError as error:
+                raise SystemExit(
+                    f"no FlashInfer fusion limit for SM{capability}, TP{world_size}"
+                ) from error
+            if input_tensor.nbytes <= fusion_limit_bytes:
+                fi_comm = communicator.fi_ar_comm
+                if fi_comm is None or fi_comm.disabled:
+                    raise SystemExit("FlashInfer all-reduce communicator is unavailable")
+                if not fi_comm.should_use_fi_ar(input_tensor):
+                    raise SystemExit("FlashInfer rejected a fusion-eligible payload")
+                return (
+                    "flashinfer_trtllm_fused_projection",
+                    fi_comm.all_reduce,
+                    fusion_limit_bytes,
+                )
+            if (
+                communicator.pynccl_comm is not None
+                and should_nccl_symm_mem_allreduce(world_size, input_tensor)
+            ):
+                return (
+                    "pynccl_symmetric_with_copy",
+                    torch.ops.vllm.all_reduce_symmetric_with_copy,
+                    fusion_limit_bytes,
+                )
+            if (
+                communicator.symm_mem_comm is not None
+                and communicator.symm_mem_comm.should_use_symm_mem(input_tensor)
+            ):
+                return (
+                    "torch_symmetric_memory",
+                    communicator.symm_mem_comm.all_reduce,
+                    fusion_limit_bytes,
+                )
+            if communicator.pynccl_comm is not None:
+                return "pynccl", communicator.pynccl_comm.all_reduce, fusion_limit_bytes
+            return "torch_distributed", tensor_model_parallel_all_reduce, fusion_limit_bytes
+
+        fi_comm = communicator.fi_ar_comm
+        selected_backend = (
+            "flashinfer_trtllm"
+            if fi_comm is not None
+            and not fi_comm.disabled
+            and fi_comm.should_use_fi_ar(input_tensor)
+            else "non_flashinfer_fallback"
+        )
+        return selected_backend, tensor_model_parallel_all_reduce, None
+
     rows: list[dict[str, Any]] = []
     expected_sum = world_size * (world_size + 1) / 2
     try:
@@ -110,8 +199,11 @@ def main() -> None:
                 dtype=torch.bfloat16,
                 device=device,
             )
+            selected_backend, profile_operation, fusion_limit_bytes = (
+                resolve_profile_operation(input_tensor)
+            )
             for _ in range(args.warmup_iters):
-                output = tensor_model_parallel_all_reduce(input_tensor)
+                output = profile_operation(input_tensor)
             torch.accelerator.synchronize()
             torch.testing.assert_close(
                 output,
@@ -120,22 +212,32 @@ def main() -> None:
                 rtol=0.0,
             )
 
-            communicator = get_tp_group().device_communicator
-            fi_comm = communicator.fi_ar_comm
-            if fi_comm is None or fi_comm.disabled:
+            if args.collective_contract == "flashinfer-trtllm":
+                if selected_backend != "flashinfer_trtllm":
                     raise SystemExit(
-                    f"FlashInfer all-reduce was not selected at TP={world_size}, "
-                    f"tokens={num_tokens}"
+                        f"expected FlashInfer TRTLLM, got {selected_backend} at "
+                        f"TP={world_size}, tokens={num_tokens}"
+                    )
+            elif selected_backend not in {
+                "flashinfer_trtllm_fused_projection",
+                "pynccl_symmetric_with_copy",
+                "torch_symmetric_memory",
+                "pynccl",
+            }:
+                raise SystemExit(
+                    f"unexpected Qwen235 serving backend: {selected_backend}"
                 )
-            uses_flashinfer = fi_comm.should_use_fi_ar(input_tensor)
 
+            all_trial_rank_samples: list[list[list[float]]] = []
+            critical_path_samples: list[float] = []
+            for _ in range(args.trials):
                 samples: list[float] = []
                 for _ in range(args.repeats):
                     dist.barrier()
                     start = torch.cuda.Event(enable_timing=True)
                     end = torch.cuda.Event(enable_timing=True)
                     start.record()
-                output = tensor_model_parallel_all_reduce(input_tensor)
+                    output = profile_operation(input_tensor)
                     end.record()
                     torch.accelerator.synchronize()
                     samples.append(float(start.elapsed_time(end)))
@@ -143,7 +245,22 @@ def main() -> None:
                 gathered: list[list[float] | None] = [None] * world_size
                 dist.all_gather_object(gathered, samples)
                 if rank == 0:
-                per_rank = [stats_ms(item) for item in gathered if item is not None]
+                    rank_samples = [item for item in gathered if item is not None]
+                    all_trial_rank_samples.append(rank_samples)
+                    critical_path_samples.extend(
+                        max(per_rank[index] for per_rank in rank_samples)
+                        for index in range(args.repeats)
+                    )
+            if rank == 0:
+                flattened_by_rank = [
+                    [
+                        sample
+                        for trial in all_trial_rank_samples
+                        for sample in trial[rank_index]
+                    ]
+                    for rank_index in range(world_size)
+                ]
+                per_rank = [stats_ms(samples) for samples in flattened_by_rank]
                 row = {
                     "tensor_parallel_size": world_size,
                     "num_tokens": num_tokens,
@@ -153,12 +270,16 @@ def main() -> None:
                     * torch.tensor([], dtype=torch.bfloat16).element_size(),
                     "dtype": "bfloat16",
                     "communicator": "vllm.tensor_model_parallel_all_reduce",
-                    "selected_backend": (
-                        "flashinfer_trtllm" if uses_flashinfer else "nccl_fallback"
-                    ),
+                    "collective_contract": args.collective_contract,
+                    "selected_backend": selected_backend,
+                    "real_fusion_limit_bytes": fusion_limit_bytes,
+                    "trials": args.trials,
+                    "repeats_per_trial": args.repeats,
+                    "per_trial_rank_samples_ms": all_trial_rank_samples,
                     "per_rank_time_ms": per_rank,
-                    "critical_path_median_ms": max(
-                        rank_stats["median"] for rank_stats in per_rank
+                    "critical_path_time_ms": stats_ms(critical_path_samples),
+                    "critical_path_median_ms": statistics.median(
+                        critical_path_samples
                     ),
                 }
                 rows.append(row)
@@ -169,7 +290,11 @@ def main() -> None:
 
     if rank == 0:
         payload = {
-            "schema_version": "qwen30_vllm020_allreduce_raw.v1",
+            "schema_version": (
+                "qwen30_vllm020_allreduce_raw.v1"
+                if args.collective_contract == "flashinfer-trtllm"
+                else "vllm020_allreduce_raw.v2"
+            ),
             "environment": {
                 "vllm_version": vllm.__version__,
                 "vllm_source_commit": source_head,
@@ -178,9 +303,18 @@ def main() -> None:
                 "gpu": torch.cuda.get_device_name(device),
                 "model": str(args.model),
                 "backend_env": {
-                    "VLLM_ALLREDUCE_USE_FLASHINFER": "1",
-                    "VLLM_FLASHINFER_ALLREDUCE_BACKEND": "trtllm",
+                    "VLLM_ALLREDUCE_USE_FLASHINFER": os.getenv(
+                        "VLLM_ALLREDUCE_USE_FLASHINFER", "0"
+                    ),
+                    "VLLM_FLASHINFER_ALLREDUCE_BACKEND": os.getenv(
+                        "VLLM_FLASHINFER_ALLREDUCE_BACKEND"
+                    ),
+                    "VLLM_ALLREDUCE_USE_SYMM_MEM": os.getenv(
+                        "VLLM_ALLREDUCE_USE_SYMM_MEM", "1"
+                    ),
                 },
+                "collective_contract": args.collective_contract,
+                "disable_custom_all_reduce": disable_custom_all_reduce,
             },
             "rows": rows,
         }

runs/frontier-qwen30-vllm020-profile-v1/run_allreduce_profile.sh

@@ -2,10 +2,11 @@
 
 set -euo pipefail
 
-TP="${TP:?TP must be set to 2 or 4}"
+TP="${TP:?TP must be set to 2, 4, or 8}"
 case "${TP}" in
   2) HARD_GPU_CAP="0.40_H20h" ;;
   4) HARD_GPU_CAP="0.80_H20h" ;;
+  8) HARD_GPU_CAP="1.60_H20h" ;;
   *) echo "ERROR: invalid TP=${TP}" >&2; exit 1 ;;
 esac
 OUTPUT_ROOT="${OUTPUT_ROOT:?OUTPUT_ROOT must be set}"
@@ -13,6 +14,10 @@ VENV_ROOT="${VENV_ROOT:-/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1}"
 VLLM_SOURCE="${VLLM_SOURCE:-/home/admin/cpfs/wjh/agentic-kv/third_party/vllm_v20_build}"
 MODEL="${MODEL:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B}"
 NUM_TOKENS="${NUM_TOKENS:-8}"
+HIDDEN_DIM="${HIDDEN_DIM:-2048}"
+COLLECTIVE_CONTRACT="${COLLECTIVE_CONTRACT:-flashinfer-trtllm}"
+TRIALS="${TRIALS:-1}"
+REPEATS="${REPEATS:-10}"
 mkdir -p "${OUTPUT_ROOT}/logs" "${OUTPUT_ROOT}/provenance" "${OUTPUT_ROOT}/raw"
 exec > >(tee -a "${OUTPUT_ROOT}/logs/profile.log") 2>&1
 
@@ -22,9 +27,20 @@ if [[ "${#GPU_IDS[@]}" -ne "${TP}" ]]; then
   exit 1
 fi
 
-export VLLM_ALLREDUCE_USE_FLASHINFER=1
-export VLLM_FLASHINFER_ALLREDUCE_BACKEND=trtllm
-echo "PROFILE_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL} runtime=vLLM-0.20.0+cu129 operator=tensor_model_parallel_all_reduce backend=FlashInfer-TRTLLM tp=${TP} tokens=${NUM_TOKENS} hidden=2048 dtype=BF16 output=${OUTPUT_ROOT} expected_wall=2-6m hard_wall=720s hard_gpu_cap=${HARD_GPU_CAP}"
+case "${COLLECTIVE_CONTRACT}" in
+  flashinfer-trtllm)
+    export VLLM_ALLREDUCE_USE_FLASHINFER=1
+    export VLLM_FLASHINFER_ALLREDUCE_BACKEND=trtllm
+    export VLLM_ALLREDUCE_USE_SYMM_MEM=0
+    ;;
+  qwen235-serving-projected)
+    export VLLM_ALLREDUCE_USE_FLASHINFER=1
+    export VLLM_FLASHINFER_ALLREDUCE_BACKEND=trtllm
+    export VLLM_ALLREDUCE_USE_SYMM_MEM=1
+    ;;
+  *) echo "ERROR: invalid COLLECTIVE_CONTRACT=${COLLECTIVE_CONTRACT}" >&2; exit 1 ;;
+esac
+echo "PROFILE_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL} runtime=vLLM-0.20.0+cu129 operator=tensor_model_parallel_all_reduce contract=${COLLECTIVE_CONTRACT} tp=${TP} tokens=${NUM_TOKENS} hidden=${HIDDEN_DIM} dtype=BF16 trials=${TRIALS} repeats=${REPEATS} output=${OUTPUT_ROOT} expected_wall=2-8m hard_wall=720s hard_gpu_cap=${HARD_GPU_CAP}"
 date -u +"START_UTC=%Y-%m-%dT%H:%M:%SZ"
 nvidia-smi --query-gpu=index,name,driver_version,memory.used,utilization.gpu --format=csv,noheader
 
@@ -46,8 +62,11 @@ timeout --signal=TERM --kill-after=30s 600 \
   --model "${MODEL}" \
   --output "${OUTPUT_ROOT}/raw/allreduce-tp${TP}.json" \
   --num-tokens "${TOKEN_ARGS[@]}" \
+  --hidden-dim "${HIDDEN_DIM}" \
+  --collective-contract "${COLLECTIVE_CONTRACT}" \
   --warmup-iters 3 \
-  --repeats 10
+  --repeats "${REPEATS}" \
+  --trials "${TRIALS}"
 
 test -s "${OUTPUT_ROOT}/raw/allreduce-tp${TP}.json"
 sha256sum "${OUTPUT_ROOT}/raw/allreduce-tp${TP}.json" \