feat: qwen35 share expert and deepep combine overlap

[HongminTan]

No description provided.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[LLLLKKKK]

AI Code Review - PR #1343

Status: BLOCKING

Summary: P0/0 · P1/2 · P2/7 · P3/7

Reviewed: commit 40f6439c08a9 · 2026-08-28 12:29 UTC+8

Blocking Issues

P1

• 默认全量生效的 EP 融合归约与副流重叠缺少真实多卡数值覆盖 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:104
  • 建议：按 impl/rocm/test/test_generic_moe_allreduce.py 的既有形态补一条 2 rank 用例（routers/test/BUILD 已有 GPU_COUNT: "2" 脚手架）：对同一输入分别跑 legacy（_finalize_post_tp_gather + all_reduce(shared)，两次集合）与新路径（_finalize_row_scatter + 单次 all_reduce），断言集合次数 2→1 且输出在容差内一致。若受限于多卡资源，请补一条同时满足 DeepEP low-latency + shared expert + ffn_tp_size>1 + ep_size>1 的 smoke case，并在 PR description 中点明 case 名称，使默认切换具备可追溯的数值防线。
• finalize() 新增的 row-scatter 分派分支与能力声明在 CI 中零覆盖 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:364
  • 建议：补一条 self.assertTrue(_make_router().supports_row_scatter_finalize) 对齐同目录先例，防止该能力被误删后 use_ep_unified_allreduce 静默退回旧双集合路径而测试全绿；并用现有 stub self + patch _normal_finalize 覆盖 finalize() 两条分支（纯 CPU 即可，参考 pure_tp_router_skip_allreduce_test.py:88 直接调 finalize 的写法）：断言 (a) 给 row_scatter_target 时返回对象就是传入缓冲区且未发生 all_gather，(b) 不给时走 _finalize_post_tp_gather，(c) 两条路径都把 self._handle 复位为 None。

Non-blocking Suggestions

P2

• 归约合并与跨流重叠没有运行时回滚开关，且无法分别关闭 @ rtp_llm/models_py/model_desc/generic_moe.py:165
  • 建议：补一个 env 或 MoeConfig 开关：使 supports_row_scatter_finalize 返回 False 即可整体退回已验证的 use_ep_shared_allreduce 路径；再拆一个更细的开关（例如强制 shared_expert_stream=None），使跨流重叠可单独关闭而保留一次归约。默认可继续开启，并在 PR description 中记录默认值与回滚方式。
• 层未校验 router 返回的即为自己提供的缓冲区，共享专家分支可能被静默丢弃 @ rtp_llm/models_py/model_desc/generic_moe.py:336
  • 建议：在 all_reduce 之前补一行零成本自校验，例如 assert experts_output is row_scatter_target, "row-scatter router must return the caller's buffer"；这样任何未来 router 的实现偏差都会在首次前向立刻暴露，而不是退化为静默的数值错误。
• supports_row_scatter_finalize 契约未写明必须 join row_scatter_ready，且缺少统一 join helper @ rtp_llm/models_py/modules/factory/fused_moe/defs/fused_moe.py:135
  • 建议：在 defs/fused_moe.py 增加共享 helper（例如 join_row_scatter_producer(extra_finalize_args, target)）内部完成 torch.cuda.current_stream(target.device).wait_event(...)，并要求所有 row-scatter 实现在写入缓冲区前第一步调用它；同时把「必须先 join 再触碰缓冲区，即使本 rank 切片为空」与现有两条要求并列写进 docstring，并说明为何 wait 放在 combine 之后而非 FusedMoe 侧（那样会丧失重叠收益，这个取舍值得留档）。若希望协议不可误用，可把 target 与 event 合并成一个 NamedTuple 传递，使「拿到 target 就必然看到 event」在类型层面成立。
• capture 用例中「多图复用同一 stream/event」的断言恒为真，给出虚假覆盖信号 @ rtp_llm/models_py/model_desc/test/generic_moe_allreduce_test.py:434
  • 建议：改为对被测对象的实际产出取证：在循环内把 _ensure_shared_expert_stream(self.device) 的返回值（而非 stub 属性）加入 forked_on 并断言 len(_SHARED_EXPERT_STREAMS) == 1，或让 _fork_and_join 返回 ready 后收集其 id，或对 torch.cuda.Stream 构造计数。若该不变量已由 test_one_stream_per_device_is_reused 充分覆盖，则直接删除这条不可失败的断言并相应改名，只保留能真正失败的多图 capture + replay 数值校验。
• 副流的真实装配与端到端 fork→join 接线在所有用例中零覆盖 @ rtp_llm/models_py/model_desc/test/generic_moe_allreduce_test.py:366
  • 建议：给 _make_layer 增加 device 参数以支持 CUDA 权重，并补一条 GPU 用例：断言 layer.shared_expert_stream is _ensure_shared_expert_stream(self.device)、layer.shared_expert_ready 是真实 torch.cuda.Event、且 fused_moe.call_args.kwargs["row_scatter_ready"] is layer.shared_expert_ready；再以真实 DenseMLP 作为 shared_expert，与「强制 shared_expert_stream=None 的串行结果」逐元素比对，覆盖 eager 与 capture/replay 两种模式。
• 副流不可用时重叠静默降级，且实际归约策略仅在 DEBUG 日志部分可见 @ rtp_llm/models_py/model_desc/generic_moe.py:189
  • 建议：把 shared_expert_stream is not None 与 use_ep_shared_allreduce 并入现有日志字段；当 use_ep_unified_allreduce 为真却拿不到副流时打一条 logger.warning 显式说明原因（权重不在 CUDA 设备上）；或在模型构建完成后以 INFO 统计一次启用重叠的 MoE 层数，让配置生效情况在默认日志级别下可验证，而不是只能靠 profiling 反推。
• 测试替身用硬编码与无 spec mock，削弱了三个生产边界 @ rtp_llm/models_py/model_desc/test/generic_moe_allreduce_test.py:69
  • 建议：补一条不 patch MoEConfigAdapter 的窄测试，用真实 parallelism_config（令 get_attn_tp_size() 与 get_ffn_tp_size() 返回不同值）断言 FusedMoeDataRouter.tp_collective_size 取的是 attn 视图；或在 fixture 中由 get_attn_tp_size() 派生 tp_collective_size，使该前提在测试内显式成立而非被硬编码。对集合通信与 DenseMLP 两个边界改用 autospec=True / create_autospec(DenseMLP, instance=True)，让签名不匹配在单测阶段即报错。

P3

• row-scatter 的形状/dtype 契约仅由 assert 保障，且负向覆盖不完整 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:314
  • 建议：把这三条契约从 assert 改为显式 raise ValueError（保留当前 f-string 诊断信息），或在 FusedMoe.forward 的入参校验块中前移 row_scatter_target 的 shape/dtype/device 检查，与已有两条 ValueError 保持同一层级；同时把该负向用例扩展为 subTest 矩阵，分别构造 hidden 维不一致与 dtype 不一致的输入并断言各自文案。
• finalize 中 assert 失败会残留 router handle，后续请求报无关错误 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:319
  • 建议：把 self._handle = None 放进 try/finally（或在 finalize 入口把句柄取到局部变量后立即清空实例字段），使无论走哪条分支、是否抛异常，router 都回到可复用状态，assert 的失败信息也能如实反映首次故障点；若认为该复位超出本 PR 范围，请在 generic_moe.py 的 except 注释中明确记录「router 内部状态未回滚」这一已知限制。
• 契约测试用 object() 冒充 torch.cuda.Event，与类型声明不符 @ rtp_llm/models_py/modules/factory/fused_moe/defs/test/fused_moe_allreduce_contract_test.py:156
  • 建议：torch.cuda.Event() 的底层 event 是惰性创建的，无 CUDA 设备也可构造，建议直接改用真实 Event 以对齐类型声明；或至少改成 Mock(spec=torch.cuda.Event)，使任何非 Event 接口调用立刻失败，同时 assertIs 身份断言写法完全不变。
• row-scatter 契约测试缺两个边界组合，且两个契约合并在同一方法内 @ rtp_llm/models_py/modules/factory/fused_moe/defs/test/fused_moe_allreduce_contract_test.py:142
  • 建议：拆成两个测试方法或用 subTest 参数化 (target, ready, expect_keys)，补齐「target 有、ready 无」组合（断言含 ROW_SCATTER_TARGET_ARG 且不含 ROW_SCATTER_READY_ARG）；若保留单方法写法，至少在第二次调用前 router.finalize.reset_mock()。并为互斥组合补一条用例，或先在 FusedMoe.forward 中对该组合抛 ValueError 再补测。
• finalize 仅校验位置参数个数，未固定参数顺序 @ rtp_llm/models_py/modules/factory/fused_moe/defs/test/fused_moe_allreduce_contract_test.py:20
  • 建议：改用 create_autospec(FusedMoeDataRouter, instance=True) 让位置/关键字绑定由 spec 校验，并在提取时按参数名读取（绑定后取 extra_finalize_args），替代按下标手工提取；这样参数顺序漂移在单测阶段即暴露，也无需在测试内重复维护「五个位置参数」这一弱约定。
• _SHARED_EXPERT_STREAMS 缺少清理入口，测试直接清空生产全局 @ rtp_llm/models_py/model_desc/generic_moe.py:39
  • 建议：提供显式清理函数（例如 clear_shared_expert_streams()）供拆卸路径与测试调用；测试侧改用 patch.dict(..., {}, clear=True) 或 addCleanup 还原快照，避免全局状态跨用例泄漏。若确认 layer 构造只发生在单线程加载阶段，请在注释中写明该前提，说明为何不需要加锁。
• 三个归约策略布尔量互相纠缠且依赖赋值顺序 @ rtp_llm/models_py/model_desc/generic_moe.py:172
  • 建议：至少把顺序依赖显式化（例如先算出 ep_shared_eligible 再由它派生两个 flag），使赋值顺序不再隐含正确性；若后续还要新增策略，可考虑把归约策略收敛为一个枚举（如 EP_UNIFIED / TP_UNIFIED / EP_SHARED_ONLY / LOCAL_ONLY）在 __init__ 一次性计算、forward 单点分派，注释里的策略表可直接对应枚举值。

Checklist Findings (14 fail / 44 total)

General Principles Checklist

• [6.1] Architecture — 分层边界：新概念在正确层级，不泄漏内部 → issue ``_SHARED_EXPERT_STREAMS 缺少清理入口，测试直接清空生产全局
  _`_SHARED_EXPERT_STREAMS` 是模块级可变全局，`_ensure_shared_expert_stream` 的 get/create/set 之间无锁，也没有任何清理入口：拆卸路径不会清空，模型重建或多模型共存时会静默复用旧 stream，stream 在进程生命周期内不释放。测试只能自己在 `generic_moe_allreduce_test.py:363` 的 `setUp` 里直接 `SHARED_EXPERT_STREAMS.clear()` 且无 `tearDown` 还原——这既耦合了私有全局，又与该全局注释声明的「每设备一条、必须在首次 captured forward 前存在」不变量相悖，为同 target 内其它用例引入隐式顺序依赖（当前其它用例均用 CPU 权重、无该全局读者，实际影响有限）。
• [6.1] Architecture — 可观测性：日志/指标/超时可操作、非噪声 → issue 副流不可用时重叠静默降级，且实际归约策略仅在 DEBUG 日志部分可见
_ensure_shared_expert_stream 对非 CUDA device 直接返回 None（第 45 行），此时 shared_expert_ready 也不创建，_shared_expert_partial 退回同流串行——本 PR 的全部重叠收益消失，却不产生任何日志或告警。若某条权重加载路径在 __init__ 时权重仍在 CPU/meta 上（随后才 .to(cuda)），use_ep_unified_allreduce 仍为 True 而重叠被永久静默关闭。紧随的 logger.debug（第 193-203 行）只打印 use_unified_tp_allreduce 与 use_ep_unified_allreduce，既不含 use_ep_shared_allreduce，也不含副流是否真正取到，且整段被 isEnabledFor(logging.DEBUG) 包住，线上 INFO 级别无法区分「重叠生效」与「重叠静默失效」。
• [6.1] Architecture — 回滚路径：风险行为存在运维回滚手段 → issue 归约合并与跨流重叠没有运行时回滚开关，且无法分别关闭
use_ep_unified_allreduce 的五个合取项全部来自并行度与 router 能力，全仓搜索确认无任何 env 或 MoeConfig 字段可关闭。一旦线上出现精度漂移、流并发或 SM 争抢导致的 TPOT 回退，唯一手段是回滚镜像。更关键的是「合并 all-reduce」（纯代数等价、风险低）与「跨流重叠」（引入并发、风险明显更高）被绑定在同一个谓词上：出问题时无法只关掉后者而保留前者的收益。
• [6.1] Architecture — 状态不变量：创建/更新/失败/重试/回滚路径有效 → issue ``_SHARED_EXPERT_STREAMS 缺少清理入口，测试直接清空生产全局
  _`_SHARED_EXPERT_STREAMS` 是模块级可变全局，`_ensure_shared_expert_stream` 的 get/create/set 之间无锁，也没有任何清理入口：拆卸路径不会清空，模型重建或多模型共存时会静默复用旧 stream，stream 在进程生命周期内不释放。测试只能自己在 `generic_moe_allreduce_test.py:363` 的 `setUp` 里直接 `SHARED_EXPERT_STREAMS.clear()` 且无 `tearDown` 还原——这既耦合了私有全局，又与该全局注释声明的「每设备一条、必须在首次 captured forward 前存在」不变量相悖，为同 target 内其它用例引入隐式顺序依赖（当前其它用例均用 CPU 权重、无该全局读者，实际影响有限）。
• [6.1] Architecture — 错误语义：fail-fast/retry/fallback/silent 行为显式 → issue finalize 中 assert 失败会残留 router handle，后续请求报无关错误
_finalize_row_scatter 新增的三处 assert（dim / shape+dtype / combined_x.size(0) == slice_size）均在 self._handle = None（第 376 行）之前执行。任一触发后 _handle 仍持有上一次 dispatch 的句柄，下一次请求会撞上 prepare 里的 assert self._handle is None（第 223 行），报出与真实原因完全无关的错误，router 实例事实上永久不可用——一次可恢复的单请求失败被放大成引擎级故障。该闩锁是既有行为（旧 _finalize_post_tp_gather 也有 assert 在复位之前），但本 PR 扩大了失败面；generic_moe.py:327 新增的 except 只补做 stream join、未复位 router 状态，注释却容易让读者误以为失败路径已完整。
• [6.1] Software Engineering — LSP：子类/重写保持基类契约 → issue ``supports_row_scatter_finalize契约未写明必须 joinrow_scatter_ready`，且缺少统一 join helper`
  该 docstring（第 136-146 行）详细要求「累加而非覆盖」「只写本 rank 拥有的行」，却完全没提 `row_scatter_ready`。副流 fork 由 `generic_moe.py:315` 打开，唯一成功路径的 join 却内联在另一模块的 `deepep_low_latency_router.py:311-312`；`fused_moe.py:344-347` 只在事件非空时透传，从不校验是否被消费，router 侧又用 `.get()` 宽松读取。第二个实现者若忽略该键，eager 模式下是静默跨流数据竞争（读到未完成的 shared partial），只有捕获模式才以 fork 未闭合暴露。对照同类能力 `skip_tp_allreduce` 在 `fused_moe.py:80` 提供了共享判定函数 `should_skip_tp_allreduce`，本次 join 逻辑却无共享入口。
• [6.1] Software Engineering — OCP：本地扩展点优先于修改中心逻辑 → issue 三个归约策略布尔量互相纠缠且依赖赋值顺序
use_ep_shared_allreduce 的定义里包含 not self.use_ep_unified_allreduce，因此必须在后者之后赋值，这个顺序约束只靠代码位置隐式保证；forward 中四条分支（ep_unified / unified_tp / ep_shared / fallback）的互斥性也只能靠人工阅读第 140-164 行的 25 行 ASCII 注释块确认。再加入第四种策略时，这张布尔网的组合正确性会难以维持。
• [6.1] Software Engineering — SRP：模块/类职责单一 → issue 三个归约策略布尔量互相纠缠且依赖赋值顺序
use_ep_shared_allreduce 的定义里包含 not self.use_ep_unified_allreduce，因此必须在后者之后赋值，这个顺序约束只靠代码位置隐式保证；forward 中四条分支（ep_unified / unified_tp / ep_shared / fallback）的互斥性也只能靠人工阅读第 140-164 行的 25 行 ASCII 注释块确认。再加入第四种策略时，这张布尔网的组合正确性会难以维持。
• [6.1] Tests — 分布式/跨平台变更有对应覆盖 → issue 副流的真实装配与端到端 fork→join 接线在所有用例中零覆盖
_make_layer 的 W.moe_w1/moe_w2 恒为 CPU 张量（第 59-60 行），故 generic_moe.py:190 的 _ensure_shared_expert_stream(self.w1.device) 必返回 None、shared_expert_ready 亦为 None，即使在 H20 上亦如此——第 306 行正是断言这一点。唯一触达副流的 SharedExpertSideStreamCaptureTest._stub_layer 用 SimpleNamespace 冒充 layer、shared_expert 是 lambda x, skip_allreduce: x * 3.0，join 也由测试体自行调用而非经 router。于是「构造期按 w1.device 建流建 event → forward 透传 → router 在 _finalize_row_scatter join」这条生产链路，以及真实量化 DenseMLP GEMM 与 routed 链路并发的数值等价性，均无任何用例执行
• [6.1] Tests — 新逻辑有聚焦单测 + 相关集成/smoke 测试 → issue ``finalize 仅校验位置参数个数，未固定参数顺序
  _`extract_extra_finalize_args` 通过 `len(args) != 5` 手工校验个数后取 `args[4]`。生产调用为 `router.finalize(combine_payload, expert_topk_weights, expert_topk_ids, apply_router_weight_on_input, finalize_args)`（`fused_moe.py:349-355`）。由于 `router` 是 `Mock(spec=FusedMoeDataRouter)`（spec 只校验属性存在性、不校验调用签名），若 `forward` 误把 `topk_weights` 与 `topk_ids`、或 `apply_router_weight_on_input` 的位置互换，个数仍为 5，本文件全部断言依旧通过，而所有真实 router 都会拿到错位参数。
• [6.1] Tests — 边界 case 覆盖（空、单元素、最大值） → issue row-scatter 契约测试缺两个边界组合，且两个契约合并在同一方法内
FusedMoeRowScatterTargetTest 只覆盖「两者都不传」与「target 与 ready 都传」。但生产上 GenericMoeLayer 在没有副流时恒产生「只传 target、不传 ready」（generic_moe_allreduce_test.py:306 正是这一形态），router 会走 .get(ROW_SCATTER_READY_ARG) 为 None 的免 join 分支，契约层未固化。另外 skip_tp_allreduce=True 与 row_scatter_target 属语义互斥组合，forward 既无校验也无用例，仅靠基类两个能力属性默认 False 而暂不可达。同时第 142-166 行在一个方法里先后调用两次 forward、依赖 call_args 只反映最后一次来断言，两个方向共享同一失败信号，与同文件第 72 行已用 subTest 的风格也不一致。

RTP-LLM Checklist

• [I] 代码质量 — 同一功能用统一工具函数 → issue ``supports_row_scatter_finalize契约未写明必须 joinrow_scatter_ready`，且缺少统一 join helper`
  该 docstring（第 136-146 行）详细要求「累加而非覆盖」「只写本 rank 拥有的行」，却完全没提 `row_scatter_ready`。副流 fork 由 `generic_moe.py:315` 打开，唯一成功路径的 join 却内联在另一模块的 `deepep_low_latency_router.py:311-312`；`fused_moe.py:344-347` 只在事件非空时透传，从不校验是否被消费，router 侧又用 `.get()` 宽松读取。第二个实现者若忽略该键，eager 模式下是静默跨流数据竞争（读到未完成的 shared partial），只有捕获模式才以 fork 未闭合暴露。对照同类能力 `skip_tp_allreduce` 在 `fused_moe.py:80` 提供了共享判定函数 `should_skip_tp_allreduce`，本次 join 逻辑却无共享入口。

Python Static-First Checklist

• [P.G] 测试规范 — mock/fake/stub 不得替代本次声称覆盖的生产边界 → issue ``finalize 仅校验位置参数个数，未固定参数顺序
  _`extract_extra_finalize_args` 通过 `len(args) != 5` 手工校验个数后取 `args[4]`。生产调用为 `router.finalize(combine_payload, expert_topk_weights, expert_topk_ids, apply_router_weight_on_input, finalize_args)`（`fused_moe.py:349-355`）。由于 `router` 是 `Mock(spec=FusedMoeDataRouter)`（spec 只校验属性存在性、不校验调用签名），若 `forward` 误把 `topk_weights` 与 `topk_ids`、或 `apply_router_weight_on_input` 的位置互换，个数仍为 5，本文件全部断言依旧通过，而所有真实 router 都会拿到错位参数。
• [P.G] 测试规范 — 数据驱动测试用 pytest.mark.parametrize → issue row-scatter 契约测试缺两个边界组合，且两个契约合并在同一方法内
FusedMoeRowScatterTargetTest 只覆盖「两者都不传」与「target 与 ready 都传」。但生产上 GenericMoeLayer 在没有副流时恒产生「只传 target、不传 ready」（generic_moe_allreduce_test.py:306 正是这一形态），router 会走 .get(ROW_SCATTER_READY_ARG) 为 None 的免 join 分支，契约层未固化。另外 skip_tp_allreduce=True 与 row_scatter_target 属语义互斥组合，forward 既无校验也无用例，仅靠基类两个能力属性默认 False 而暂不可达。同时第 142-166 行在一个方法里先后调用两次 forward、依赖 call_args 只反映最后一次来断言，两个方向共享同一失败信号，与同文件第 72 行已用 subTest 的风格也不一致。

Strengths

• 归约代数被严格证明而非口头声明：deepep_low_latency_row_scatter_test.py:113 以 all_gather(routed) + all_reduce(shared) 为基线对拍融合结果，token 数覆盖 1/3/5/8/9（整除、ragged 尾部、尾部 rank 完全无 token），并用 assertEqual(routed_full.shape[0], original_num_tokens) 顺带证明切片确实构成划分；target 取非零的 gate * shared_partials[rank]，能同时抓住「add_ 退化为覆盖」与「写入他 rank 行」两类缺陷。
• _tp_token_slice（deepep_low_latency_router.py:107）把切片算术从 _prepare_pre_tp_slice 抽为单一实现，使「dispatch 切法」与「scatter 回写行」不可能漂移，并成为该不变量的唯一来源；ragged 尾部因此不再需要旧路径的 torch.empty 等长补齐（:283-288）。
• _finalize_row_scatter 把 wait_event 放在所有 assert 与写入之前并注释理由（调用方归约全部行，空切片 rank 同样必须 join）；deepep_low_latency_row_scatter_test.py:83 在 patch 的 wait_event 回调里断言 buffer 仍全零，是真正会失败的时序断言而非调用计数，且覆盖了尾部空 rank。
• 跨流内存生命周期双向到位：hidden_states.record_stream(stream) 防止上层解引用后 allocator 复用仍被副流读取的块，shared_partial.record_stream(current) 防止副流分配块在主流仍在写时被回收，注释说明了各自原因。
• SharedExpertSideStreamCaptureTest 选择真实 torch.cuda.CUDAGraph 而非 mock 验证 fork 闭合（未闭合只在 capture end 报错），并覆盖「多 batch size 共享同一 graph pool」这一解码实际形态，replay 前 hidden.fill_ 改值以区分真实重算与残留缓冲；_CUDA_GRAPH_READY 显式排除 HIP 并注明理由，使 generic_moe_allreduce_test_rocm 目标不会误跑 CUDA-only 分支。
• FusedMoe.forward 对能力不匹配与参数不自洽在 router.prepare 之前 fail-fast，三条 ValueError 用例均额外断言 prepare/execute 未被调用；forward 又在 routed 链路抛异常时补做 wait_event，generic_moe_allreduce_test.py:309 覆盖了这条最易漏掉的失败路径。
• test_layer_only_passes_keywords_fused_moe_accepts（generic_moe_allreduce_test.py:329）用 inspect.signature(FusedMoe.forward) 反查 kwargs，精准堵住「MagicMock 接受任意关键字导致 layer 多传参数直到线上才炸」的盲区，四条归约分支全覆盖。
• 归约策略决策表以结构化注释集中写在 generic_moe.py:140-164，显式区分「partial 可合并」与「已完整不可合并」，把本次改动最易出错的前提（已完整分支参与合并归约会被乘 tp_size）留档，后续维护者不必反推。

[LLLLKKKK]

AI Code Review - PR #1343 (non-blocking suggestions)

14 条 P2/P3 建议，不阻塞合并。阻塞判定与完整摘要见上一条 review。

[LLLLKKKK]

AI Code Review - PR #1343

Status: LGTM

Summary: P0/0 · P1/0 · P2/8 · P3/13

Reviewed: commit afdeb0846375 · 2026-08-31 16:08 UTC+8

lgtm ready to ci

Non-blocking Suggestions

P2

• extra_finalize_args 可绕过 row_scatter 的两道能力门禁 @ rtp_llm/models_py/modules/factory/fused_moe/defs/fused_moe.py:344
  • 建议：按 skip_tp_allreduce 对齐：在 finalize_args.update() 中无条件写入这两个键（为 None 时显式 pop），使函数参数成为唯一真源；并在 FusedMoeRowScatterTargetTest 补两条与 test_unsupported_router_cannot_be_bypassed_by_finalize_key 对称的负向用例，断言注入的 row_scatter_target / row_scatter_ready 不会泄漏到 router.finalize。
• supports_row_scatter_finalize 契约未写明必须 join row_scatter_ready，也未写调用方 buffer 义务 @ rtp_llm/models_py/modules/factory/fused_moe/defs/fused_moe.py:135
  • 建议：把「触碰 buffer 前必须 wait_event(row_scatter_ready)，即使本 rank 切片为空」写成与「累加不覆写」同级的硬性要求，并给 FinalizeArgs.row_scatter_ready 加注释；更稳的做法是把 join 上移到 FusedMoe.forward 调用 router.finalize 之前（此时 dispatch 与 expert 计算均已入队，重叠效果不变），让契约由框架强制。同时把调用方义务补进 docstring，并在 FusedMoe.forward 现有校验块加上 is_contiguous() 与与 hidden_states 的存储别名 fail-fast 检查。
• 层未校验 router 返回的即为自己提供的缓冲区，共享专家分支可能被静默丢弃 @ rtp_llm/models_py/model_desc/generic_moe.py:336
  • 建议：在 FusedMoe.forward 中当 row_scatter_target is not None 时增加 assert output is row_scatter_target，把契约违约从静默错误变为 fail-fast；并把 _make_fused_moe 的 row-scatter 分支改为合规假实现（对传入 target 执行 add_ 后返回 target），补一条 assertIs(output, target)，使契约测试不再编码违约行为。
• 归约合并与跨流重叠没有运行时回滚开关，且无法分别关闭 @ rtp_llm/models_py/model_desc/generic_moe.py:165
  • 建议：参照仓内已有风格增加默认开启、可关闭的开关（env 或 MoeConfig 字段）参与谓词，关闭时自然回落 use_ep_shared_allreduce；建议把「row-scatter 融合归约」与「共享专家侧流 overlap」拆成两个独立开关，因为二者失效模式（数值/形状 vs 并发/SM 争用）完全不同。
• 真实 DeepEP combine 未覆盖非整除切片与空切片 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/test/deepep_low_latency_router_test.py:371
  • 建议：在 _run_deepep_low_latency_router_test 中为 _check_row_scatter_matches_gather 增加至少两组 token 数：一个不被 tp_size 整除（如 num_token_per_rank - 1），一个小于 tp_size 使尾部 rank 切片为空，让硬断言与 slice_size == 0 短路在真实 combine kernel 下被验证；若 DeepEP 对 0 行 dispatch 另有约束，也应在该测试中显式暴露。
• capture 用例中「多图复用同一 stream/event」的断言恒为真 @ rtp_llm/models_py/model_desc/test/generic_moe_allreduce_test.py:434
  • 建议：让断言观察真实来源：把 _stub_layer() 移入循环，使每轮真正经过 _ensure_shared_expert_stream，再断言两轮取到的是同一 stream 对象；或在循环内断言 _ensure_shared_expert_stream(self.device) is stub.shared_expert_stream。若确认 stub 模式下无法有效验证，宁可删除该恒真断言与其失败消息，避免留下虚假覆盖信号。
• 侧流的真实装配与端到端 fork→join 接线在所有用例中零覆盖 @ rtp_llm/models_py/model_desc/test/generic_moe_allreduce_test.py:306
  • 建议：给 _make_layer 加 device 参数把 W.moe_w1/W.moe_w2 放到 cuda，在 _CUDA_GRAPH_READY 保护下新增一条用真实 GenericMoeLayer 的用例：断言 layer.shared_expert_stream is _ensure_shared_expert_stream(device)、layer.shared_expert_ready is not None，并在一次 forward 后断言 fused_moe.call_args.kwargs["row_scatter_ready"] is layer.shared_expert_ready，闭合 __init__ → forward → router 的 event 传递链；保留现有 CPU 用例继续覆盖 None 分支。该 target 已配置 H20 与 MI308X（gpu_count: 1），无需新增 CI 资源。
• 侧流不可用时重叠静默降级，且实际归约策略仅在 DEBUG 日志部分可见 @ rtp_llm/models_py/model_desc/generic_moe.py:189
  • 建议：把 self.shared_expert_stream is not None 加入 :194 的日志字段，并把最终生效策略从 DEBUG 提升为启动期一次性 INFO；若 use_ep_unified_allreduce 为真而侧流拿不到，logger.warning 一次并说明降级原因（设备类型 / CUDA 不可用），使「策略已启用但重叠未生效」这一状态可被观测。

P3

• 四种归约策略靠三个互相纠缠且依赖赋值顺序的布尔标志编码 @ rtp_llm/models_py/model_desc/generic_moe.py:172
  • 建议：把策略收敛为一个枚举或小策略对象，携带「共享分支是否 skip 自身 all-reduce / routed 分支是否接收 scatter target / 集合通信次数」三项属性；forward 只做一次分派，表格注释退化为枚举成员文档，后续新增策略变为新增枚举成员而非修改中心逻辑。
• _finalize_post_tp_gather 未复用同批引入的切片工具 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:278
  • 建议：让 _finalize_post_tp_gather 直接调用 _tp_token_slice，或抽出共享的 _tp_token_shard_size(num_tokens)，使均分宽度与落位区间只有一个定义点。
• _ensure_shared_expert_stream 与 dsv4 同名实现近乎逐行重复 @ rtp_llm/models_py/model_desc/generic_moe.py:39
  • 建议：把 per-device 侧流缓存抽到共享位置（例如 models_py/utils/ 下的 side-stream helper），由 dsv4 与 generic_moe 共同复用，并顺带提供测试可用的清理入口，避免测试直接清空生产模块全局态。
• 捕获期无条件调用 record_stream，可能抬高 graph pool 占用 @ rtp_llm/models_py/model_desc/generic_moe.py:256
  • 建议：评估在捕获期跳过 record_stream（捕获期生命周期由 graph 私有内存池保证，不依赖 allocator 的 stream_uses）；若确认必须保留，请在注释中写明理由，并给出捕获前后 graph pool 占用的对比数据。
• all_reduce 的 inplace 用法在两条统一归约路径间不一致 @ rtp_llm/models_py/model_desc/generic_moe.py:365
  • 建议：统一两条路径的调用形式（pure-TP 分支也传 inplace=True），省掉 symm_mem 快路径上的一次 [T, H] 分配；若有意保持差异，请在注释中说明原因。
• 真机测试统计的 all_reduce 次数是自证断言 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/test/deepep_low_latency_router_test.py:208
  • 建议：把 collective_torch.all_reduce 一并以 wraps 方式 patch 计数，让「Group.TP 集合通信从 2 次降到 1 次」成为可失败的断言；若不便 patch，则把 docstring 与断言收敛为「row scatter 路径不再触发 all_gather」，从比较字典中移除 all_reduce 项。
• row-scatter 的形状/dtype 契约仅由 assert 保障，且负向覆盖不完整 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:314
  • 建议：在 DeepEpLowLatencyRowScatterTest 补两个小用例：传入 hidden 维不同的 target、传入 dtype 不同（如 bf16 target 配 fp32 combine 输出）的 target，各自用 assertRaises(AssertionError) 钉住 fail-fast 行为。
• finalize 中 assert 失败会残留 router handle，后续请求报无关错误 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/deepep_low_latency_router.py:319
  • 建议：用 try/finally 包住 finalize 的后处理，把 self._handle = None 放进 finally（或在 combine 之后立即复位），保证断言失败也把 router 状态复位，使首次抛出的异常即为真实根因。
• 契约测试用 object() 冒充 torch.cuda.Event，且两个场景挤在同一方法 @ rtp_llm/models_py/modules/factory/fused_moe/defs/test/fused_moe_allreduce_contract_test.py:156
  • 建议：拆成独立用例或用 subTest 区分场景，并补一条「target 有值、ready 为 None」的用例断言 finalize 字典中不含 ready 键；ready 改用 Mock(spec=torch.cuda.Event)，在保持平台中立的同时保留类型契约。
• finalize 断言仅校验位置参数个数，未固定参数顺序 @ rtp_llm/models_py/modules/factory/fused_moe/defs/test/fused_moe_allreduce_contract_test.py:20
  • 建议：在 helper 中一并断言前四个位置参数的身份（args[0].fused_expert_output、args[1] is topk_weights、args[2] is topk_ids、args[3] is False），或改用 create_autospec / Mock(spec=FusedMoeDataRouter.finalize) 之类可校验绑定的替身。
• setUp 清空生产模块全局缓存且不还原 @ rtp_llm/models_py/model_desc/test/generic_moe_allreduce_test.py:363
  • 建议：改为作用域受限写法：self.enterContext(patch.dict(_SHARED_EXPERT_STREAMS, {}, clear=True))，或先快照再 self.addCleanup 还原；更根本的做法是由生产模块提供正规的清理入口，避免测试对生产全局态留下不可逆改动与顺序依赖。
• 测试在循环体内定义闭包并捕获循环变量 @ rtp_llm/models_py/modules/factory/fused_moe/impl/cuda/routers/test/deepep_low_latency_row_scatter_test.py:132
  • 建议：改为默认参数绑定（def wait_event(seen, target=target, waited=waited):），或把单轮断言逻辑抽成接收 target 的模块级 helper，让绑定关系显式化。
• 侧流未设优先级，且 PR 缺少热路径实测数据 @ rtp_llm/models_py/model_desc/generic_moe.py:50
  • 建议：在 PR 描述中补上目标机型（EP+TP 配置）下 small-batch decode 与 prefill 大 batch 的端到端延迟前后对比；若观测到 routed 链被拖慢，考虑用 torch.cuda.Stream(device=..., priority=...) 把侧流降为低优先级，或引入类似 dsv4 的 token 阈值保护。

Checklist Findings (18 fail / 44 total)

General Principles Checklist

• [6.1] Architecture — 分层边界：新概念在正确层级，不泄漏内部 → issue setUp 清空生产模块全局缓存且不还原
setUp 直接 _SHARED_EXPERT_STREAMS.clear()（:363）修改被测模块的全局缓存，且无 tearDown / addCleanup 还原。该表是「每设备一条侧流」这一 capture 前提的唯一持有者（generic_moe.py:36-52）。当前文件其余用例都用 CPU 权重、不写入该表，故影响尚未显现；但一旦按本报告建议新增构造真实 CUDA layer 的用例，清表会让 _ensure_shared_expert_stream 为同一设备再发一条流，破坏本类 test_one_stream_per_device_is_reused 断言的不变量，并使先前已 capture 的 fork 边指向无人 join 的流。
• [6.1] Architecture — 可观测性：日志/指标/超时可操作、非噪声 → issue 侧流不可用时重叠静默降级，且实际归约策略仅在 DEBUG 日志部分可见
_ensure_shared_expert_stream 在 device.type != "cuda" 或 CUDA 不可用时返回 None（:45-46），此时 shared_expert_stream / shared_expert_ready 保持 None，_shared_expert_partial 退化为串行内联执行（:241-248）——功能仍正确但重叠收益全部丢失。紧随其后的日志（:193-203）只打印两个策略标志与并行度，不含侧流是否真的建立，且整段决策只在 DEBUG 级可见。若 self.w1 在构造期尚未落到 CUDA（延迟加载 / 权重先在 CPU），重叠会整层失效而日志看起来一切正常，性能回退极难定位。
• [6.1] Architecture — 回滚路径：风险行为存在运维回滚手段 → issue 归约合并与跨流重叠没有运行时回滚开关，且无法分别关闭
use_ep_unified_allreduce 完全由静态拓扑条件（:165-171）与 DeepEpLowLatencyRouter.supports_row_scatter_finalize 硬编码 return True（router:104-105）推导；全仓检索确认无任何 env 或 moe_config 开关参与。凡「DeepEP LL router + ep_size>1 + ffn_tp_size>1 + ffn_tp==attn_tp + 有共享专家」的线上配置升级后会一次性同时切换三项行为：归约拓扑与 bf16 累加顺序改变、引入跨流并发、CUDA graph 内新增 fork/join；同时 use_ep_shared_allreduce 因 :176 被挤掉，旧路径不再可达。仓内同类实现 dsv4/moe/shared_expert.py:27,471-474 用 env 选择模式（默认 sequential）并带 token 阈值，本路径两者皆无，出问题只能回滚镜像。
• [6.1] Architecture — 状态不变量：创建/更新/失败/重试/回滚路径有效 → issue setUp 清空生产模块全局缓存且不还原
setUp 直接 _SHARED_EXPERT_STREAMS.clear()（:363）修改被测模块的全局缓存，且无 tearDown / addCleanup 还原。该表是「每设备一条侧流」这一 capture 前提的唯一持有者（generic_moe.py:36-52）。当前文件其余用例都用 CPU 权重、不写入该表，故影响尚未显现；但一旦按本报告建议新增构造真实 CUDA layer 的用例，清表会让 _ensure_shared_expert_stream 为同一设备再发一条流，破坏本类 test_one_stream_per_device_is_reused 断言的不变量，并使先前已 capture 的 fork 边指向无人 join 的流。
• [6.1] Architecture — 错误语义：fail-fast/retry/fallback/silent 行为显式 → issue finalize 中 assert 失败会残留 router handle，后续请求报无关错误
self._handle = None（:376）位于两条 finalize 分支之后。若 _finalize_row_scatter 的四条 assert（:314-322）任一触发（例如 combine 返回行数与切片不符），异常在 :367 向上抛出而 self._handle 仍保留上次 dispatch 的值；下一次请求进入 prepare 时会撞上 assert self._handle is None（:223），报出与真实故障完全无关的错误信息，掩盖首发根因。gather 路径同样存在此结构，但本次新增的硬断言把触发面扩大了，combined_x.size(0) == slice_size 正是最可能在 ragged token 下触发的一条。
• [6.1] Quality — PR description 说明动机与设计 → issue 侧流未设优先级，且 PR 缺少热路径实测数据
torch.cuda.Stream(device=torch.device("cuda", index)) 用默认优先级创建。改动后共享专家的两次 GEMM 与 routed 链（DeepEP dispatch/combine + grouped GEMM）并发争抢 SM，而 routed 链正是 decode 的关键路径。改动动机是 overlap 提速，但 diff 与新增用例只验证正确性，无任何 decode 端到端延迟前后对比；prefill 大 batch 或 small-batch decode 下共享专家抢占 SM 反而拖慢 routed 链完全可能，仅从代码无法判断净收益方向。对比 dsv4/moe/shared_expert.py:471-474 带 token 阈值（默认 4096）在大 batch 下退回串行，本路径无对应保护。
• [6.1] Software Engineering — DRY：重复非平凡逻辑被抽取或显式复用 → issue _ensure_shared_expert_stream 与 dsv4 同名实现近乎逐行重复
modules/dsv4/moe/shared_expert.py:23,43-53 已存在 _SHARED_EXPERT_STREAM_CACHE 与同名函数 _ensure_shared_expert_stream(device)，函数体与新增的 _SHARED_EXPERT_STREAMS / _ensure_shared_expert_stream（generic_moe.py:39-52）近乎逐行相同（同样的非 cuda 早退、同样的 device.index or current_device() 归一化、同样按 index 缓存）。两处同名不同模块的全局缓存让「本进程每设备到底有几条共享专家侧流」难以推断，也让「每设备一条侧流」这一 capture 前提失去单一权威持有者。需说明：新实现仅在 __init__ 调用（:190），不存在捕获期懒建流路径，故不涉及正确性。
• [6.1] Software Engineering — KISS/YAGNI：无投机性抽象 → issue 四种归约策略靠三个互相纠缠且依赖赋值顺序的布尔标志编码
use_ep_unified_allreduce、use_ep_shared_allreduce、use_unified_tp_allreduce 三个布尔量加一个隐式 fallback 共同编码四种互斥策略，且彼此耦合并依赖赋值顺序（use_ep_shared_allreduce 必须显式写 and not self.use_ep_unified_allreduce，:176）。forward 需先用 early return 处理第一种（:311-336），再用 if/elif/else 处理其余三种（:357-381）。__init__ 中那段 25 行 ASCII 表格注释（:140-164）本身就是编码已不自解释的证据；再加第五种策略仍需同时改中心谓词块与 forward，本次已是第二次这样做。
• [6.1] Software Engineering — LSP：子类/重写保持基类契约 → issue 层未校验 router 返回的即为自己提供的缓冲区，共享专家分支可能被静默丢弃
forward 直接对 fused_moe 的返回值做 all_reduce(..., inplace=True)（:336），隐含假设它正是自己传入且已含 gated shared partial 的 buffer；而 FusedMoe.forward 只断言 output.shape == hidden_states.shape（fused_moe.py:357-359），不校验 output is row_scatter_target。若某 router 声明支持该能力却在 finalize 中另分配输出，shared 分支贡献被静默丢弃且不报错。契约测试恰好编码了这种不合规行为：_make_fused_moe 令 router.finalize.return_value = hidden_states.clone()（contract_test:46），而 test_forward_passes_row_scatter_target_only_when_supplied 只断言键透传、不断言返回同一性，故该文件全绿。
• [6.1] Software Engineering — OCP：本地扩展点优先于修改中心逻辑 → issue 四种归约策略靠三个互相纠缠且依赖赋值顺序的布尔标志编码
use_ep_unified_allreduce、use_ep_shared_allreduce、use_unified_tp_allreduce 三个布尔量加一个隐式 fallback 共同编码四种互斥策略，且彼此耦合并依赖赋值顺序（use_ep_shared_allreduce 必须显式写 and not self.use_ep_unified_allreduce，:176）。forward 需先用 early return 处理第一种（:311-336），再用 if/elif/else 处理其余三种（:357-381）。__init__ 中那段 25 行 ASCII 表格注释（:140-164）本身就是编码已不自解释的证据；再加第五种策略仍需同时改中心谓词块与 forward，本次已是第二次这样做。
• [6.1] Software Engineering — SRP：模块/类职责单一 → issue 四种归约策略靠三个互相纠缠且依赖赋值顺序的布尔标志编码
use_ep_unified_allreduce、use_ep_shared_allreduce、use_unified_tp_allreduce 三个布尔量加一个隐式 fallback 共同编码四种互斥策略，且彼此耦合并依赖赋值顺序（use_ep_shared_allreduce 必须显式写 and not self.use_ep_unified_allreduce，:176）。forward 需先用 early return 处理第一种（:311-336），再用 if/elif/else 处理其余三种（:357-381）。__init__ 中那段 25 行 ASCII 表格注释（:140-164）本身就是编码已不自解释的证据；再加第五种策略仍需同时改中心谓词块与 forward，本次已是第二次这样做。
• [6.1] Tests — 分布式/跨平台变更有对应覆盖 → issue 侧流的真实装配与端到端 fork→join 接线在所有用例中零覆盖
_make_layer 的权重恒为 CPU 张量（:58-61），__init__ 依 self.w1.device 取流（generic_moe.py:190）而 _ensure_shared_expert_stream 对非 cuda 返回 None（:45-46），故 use_ep_unified_allreduce 分支下 shared_expert_stream / shared_expert_ready 恒为 None，:306 反向钉住 assertIsNone(row_scatter_ready)；SharedExpertSideStreamCaptureTest 又用 SimpleNamespace（:370-377）调未绑定方法，绕过 __init__ 与 forward。于是两条连接无任何断言：CUDA 权重下 __init__ 确实建出 stream/event；forward 确实把非空 event 交给 fused_moe。删除 kwarg 会因 :306 下标读取而 KeyError，但硬编码 `
• [6.1] Tests — 新逻辑有聚焦单测 + 相关集成/smoke 测试 → issue finalize 断言仅校验位置参数个数，未固定参数顺序
_extract_extra_finalize_args 只检查 len(args) != 5 并返回 args[4]（:20-26）。若 FusedMoe.forward 中 router.finalize(...) 的实参顺序被调换（如 topk_weights 与 topk_ids 互换），位置参数个数不变，本文件全部用例仍通过，而生产会把权重当成 id 传给 combine kernel。由于 router 用 Mock(spec=FusedMoeDataRouter)（:34），spec 只限制属性名、不校验子方法实参绑定，这层保护完全落在该 helper 上。
• [6.1] Tests — 边界 case 覆盖（空、单元素、最大值） → issue 契约测试用 object() 冒充 torch.cuda.Event，且两个场景挤在同一方法
  :156 与 :177 用 object() 充当 row_scatter_ready，而 FinalizeArgs.row_scatter_ready 与 FusedMoe.forward 均声明其为 torch.cuda.Event（fused_moe.py:42、267），测试因此没有钉住类型契约。另外 test_forward_passes_row_scatter_target_only_when_supplied（:142-166）把「未提供 target」与「同时提供 target 与 ready」两个场景串在同一方法且未用 subTest，第一段断言失败会掩盖第二段；「target 有值但 ready 为 None」这一实际生产分支（fused_moe.py:346 的 false 分支）在两个测试文件中均未被断言。

RTP-LLM Checklist

• [I] 代码质量 — 同一功能用统一工具函数 → issue all_reduce 的 inplace 用法在两条统一归约路径间不一致
  新路径用 all_reduce(experts_output, group=Group.TP, inplace=True)（:336），而语义完全对称的 use_unified_tp_allreduce 路径仍是 all_reduce(experts_output, group=Group.TP)（:365）。collective_torch.all_reduce 的 inplace 为 keyword-only 且只影响 symm_mem 快路径（collective_torch.py:694,716）：非 inplace 时 symm_mem_comm.all_reduce(tensor, out=None) 会额外 torch.empty_like 一份 [T, H]（symm_mem.py:167-168）。而 pure-TP 路径的 _merge_shared_expert_output 本就已就地写入 experts_output，加 inplace=True 同样安全。

Python Static-First Checklist

• [P.F] 语言陷阱 — 循环中 lambda late binding → issue 测试在循环体内定义闭包并捕获循环变量
test_joins_the_producer_before_touching_the_buffer 在 for tp_rank in (0, TP_SIZE - 1) 循环体内定义 wait_event(seen)（:132-136），闭包捕获每轮重新绑定的 target 与 waited。当前写法因闭包只在本轮同步调用而行为正确，但这是 late-binding 类缺陷的典型形状：后续若把调用挪出循环（例如收集起来统一断言）会静默指向最后一轮的 target。
• [P.G] 测试规范 — mock/fake/stub 不得替代本次声称覆盖的生产边界 → issue finalize 断言仅校验位置参数个数，未固定参数顺序
_extract_extra_finalize_args 只检查 len(args) != 5 并返回 args[4]（:20-26）。若 FusedMoe.forward 中 router.finalize(...) 的实参顺序被调换（如 topk_weights 与 topk_ids 互换），位置参数个数不变，本文件全部用例仍通过，而生产会把权重当成 id 传给 combine kernel。由于 router 用 Mock(spec=FusedMoeDataRouter)（:34），spec 只限制属性名、不校验子方法实参绑定，这层保护完全落在该 helper 上。
• [P.G] 测试规范 — 数据驱动测试用 pytest.mark.parametrize → issue 契约测试用 object() 冒充 torch.cuda.Event，且两个场景挤在同一方法
  :156 与 :177 用 object() 充当 row_scatter_ready，而 FinalizeArgs.row_scatter_ready 与 FusedMoe.forward 均声明其为 torch.cuda.Event（fused_moe.py:42、267），测试因此没有钉住类型契约。另外 test_forward_passes_row_scatter_target_only_when_supplied（:142-166）把「未提供 target」与「同时提供 target 与 ready」两个场景串在同一方法且未用 subTest，第一段断言失败会掩盖第二段；「target 有值但 ready 为 None」这一实际生产分支（fused_moe.py:346 的 false 分支）在两个测试文件中均未被断言。

Strengths

• __init__ 的归约策略对照表注释（generic_moe.py:140-164）把「partial 才能合并、already-complete 分支不能参与」这条非显然不变量画成表格并标注 Group.TP 集合通信次数，读者无需反推谓词；use_ep_shared_allreduce 上显式的 and not self.use_ep_unified_allreduce（:176）保证互斥，并被 test_ep_unified_displaces_the_separate_shared_reduce 正面钉住。
• 跨流内存生命周期两个方向都覆盖且带解释：hidden_states.record_stream(side_stream) 保护主流分配块在侧流读取期不被回收，shared_partial.record_stream(current) 保护侧流分配块在主流读写期不被回收（generic_moe.py:250-262）——这是同类改动最常漏掉的一环。
• 侧流上只放共享专家（skip_allreduce=True），所有集合通信仍留主流，规避了「NCCL 在 capture 期跨多流」这一已知雷区；routed 链抛异常时补做 join（:327-335），并有 test_a_failed_routed_chain_joins_the_shared_branch 覆盖。
• _finalize_row_scatter 把 join 放在 DeepEP combine 之后、写 buffer 之前，既拿到最大重叠又保证可见性，并显式处理「本 rank 切片为空但仍须 join」的尾部 rank（router:311-324）。
• assert combined_x.size(0) == slice_size（router:319）把「combine 返回的正是 dispatch 切片、不带 padding」这条隐含假设显式钉死（替代旧路径的 padding 兜底），并有 test_rejects_a_combine_output_that_is_not_the_dispatched_slice 反向验证。
• deepep_low_latency_row_scatter_test.py:152 不用任何 mock，以 1/3/5/8/9 个 token 覆盖整除、ragged 尾部、尾部 rank 空切片，并用 torch.cat(routed_slices).shape[0] == original_num_tokens 顺带证明切片无重叠无遗漏——这正是整个优化成立的代数前提。
• 契约校验前置：FusedMoe.forward:270-288 在 router.prepare / experts.execute 之前就对「router 不支持却传 target」「传 ready 但无 target」抛 ValueError，对应用例还断言两者未被调用，避免半执行状态。
• SharedExpertSideStreamCaptureTest 用真实 torch.cuda.graph 验证 fork 在 capture 结束前已闭合、replay 会重算（未闭合的 fork 只在 capture 末尾报错，纯 mock 无法发现），并覆盖 decode 多 batch size 共享同一 graph pool 的实际形态。
• test_layer_only_passes_keywords_fused_moe_accepts 用 inspect.signature(FusedMoe.forward) 反查 MagicMock 吞掉的多余 kwargs；_extract_extra_finalize_args 要求 router.finalize 收到恰好 5 个位置参数——两处都是对 mock 失真的主动补偿。
• _tp_token_slice、_combine_input、_make_fused_moe 三处提取同时消除生产侧与测试侧的重复分片/反量化逻辑；新增 py_test 目标的 tags/exec_properties/deps 与同目录 pure_tp_router_skip_allreduce_test 完全一致，未引入新的 CI 平台约定。

LGTM:阻断已解除(rtpcli 自动清除旧红标)