test(e2e): add Qwen3 MoE CUDA coverage

.agents/skills/run-e2e/SKILL.md

@@ -1,14 +1,18 @@
 ---
 name: run-e2e
-description: Use when the user asks to run, validate, or diagnose the AFD plugin's DeepSeek-V2-Lite end-to-end tests on GPU or Ascend NPU hardware, including PR-gate E2E, GSM8K-7 accuracy, graph, eager, DBO, or 2A2F scenarios.
+description: Use when the user asks to run, validate, or diagnose the AFD plugin's DeepSeek-V2-Lite GPU/NPU or Qwen3 MoE GPU end-to-end tests, including PR-gate E2E, GSM8K-7 accuracy, graph, eager, DBO, or 2A2F scenarios.
 ---
 
 # Run AFD E2E Tests
 
 ## Scope
 
-Run the four tests in
-tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py:
+Run one of the model suites:
+
+- `tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py` on GPU or NPU
+- `tests/e2e/models/qwen3_moe/test_qwen3_moe.py` on GPU
+
+Each suite contains four scenarios:
 
 - baseline-graph
 - afd-eager
@@ -30,7 +34,7 @@ If both are available, ask which to use. If neither is available, stop.
 Before starting pytest, confirm:
 
 - AFD_E2E_DEVICES contains the device IDs required by test cases.
-- The backend model variable points to an available model.
+- The selected model variable points to an available model.
 - The selected vllm command runs.
 - pytest, afd_plugin, and lm_eval are importable.
 - HF_HOME points to the Hugging Face cache used for GSM8K.
@@ -52,7 +56,7 @@ python -c 'from datasets import load_dataset; load_dataset("openai/gsm8k", "main
 
 ### 3. Configure the run
 
-For GPU:
+For DeepSeek-V2-Lite on GPU:
 
 ~~~bash
 export AFD_E2E_BACKEND=gpu
@@ -61,7 +65,16 @@ export AFD_GPU_E2E_MODEL=/path/to/DeepSeek-V2-Lite
 # Optional: export AFD_GPU_E2E_VLLM_BIN=/path/to/vllm
 ~~~
 
-For NPU:
+For Qwen3 MoE on GPU:
+
+~~~bash
+export AFD_E2E_BACKEND=gpu
+export AFD_E2E_DEVICES=0,1,2
+export AFD_GPU_E2E_MODEL=/path/to/Qwen3-30B-A3B
+# Optional: export AFD_GPU_E2E_VLLM_BIN=/path/to/vllm
+~~~
+
+For DeepSeek-V2-Lite on NPU:
 
 ~~~bash
 export AFD_E2E_BACKEND=npu
@@ -85,6 +98,13 @@ python -m pytest -q -s \
   "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[baseline-graph]"
 ~~~
 
+For Qwen3 MoE, run:
+
+~~~bash
+python -m pytest -q -s \
+  tests/e2e/models/qwen3_moe/test_qwen3_moe.py
+~~~
+
 Do not add backend markers or run scenarios in parallel; they share devices.
 
 For the opt-in GPU/NPU 2A2F case, set four unique device IDs and run:

docs/design/module/e2e_testing.md

@@ -25,15 +25,17 @@ validation_paths:
   - "tests/unit/test_e2e_process_utils.py"
   - "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py"
   - "tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py"
+  - "tests/e2e/models/qwen3_moe/test_qwen3_moe.py"
 upstream_refs:
   - "vLLM 0.26.0 serving and shutdown interfaces"
   - "lm-evaluation-harness GSM8K task and local-completions API"
   - "pytest parameterized test IDs"
 verified_platform_refs:
   - "CUDA DeepSeek-V2-Lite"
   - "Ascend NPU DeepSeek-V2-Lite"
+  - "CUDA Qwen3 MoE"
 related_issues: []
-last_reviewed: 2026-08-12
+last_reviewed: 2026-08-13
 ---
 
 # E2E testing
@@ -139,5 +141,5 @@ of 1319.
 7. Update `tests/e2e/README.md`. Update the `run-e2e` skill only when commands,
    variables, prerequisites, or the default suite change.
 
-Before adding a second model, extract the shared environment, device, and
-runner-subprocess logic from `test_deepseek_v2_lite.py`.
+Model entrypoints reuse the environment, device, and subprocess logic in
+`tests/e2e/models/model_test_utils.py`.

tests/e2e/README.md

@@ -1,7 +1,7 @@
 # End-to-End Tests
 
-These tests validate DeepSeek-V2-Lite on real GPU or Ascend NPU hardware.
-The default gate runs four scenarios:
+These tests validate DeepSeek-V2-Lite on real GPU or Ascend NPU hardware and
+Qwen3 MoE on real GPU hardware. The default gate runs four scenarios:
 
 - `baseline-graph`
 - `afd-eager`
@@ -27,30 +27,43 @@ export HF_HOME=/path/to/huggingface
 python -c 'from datasets import load_dataset; load_dataset("openai/gsm8k", "main")'
 ```
 
-GPU:
+DeepSeek-V2-Lite on GPU:
 
 ```bash
 export AFD_E2E_BACKEND=gpu
 export AFD_E2E_DEVICES=0,1,2
 export AFD_GPU_E2E_MODEL=/path/to/DeepSeek-V2-Lite
 ```
 
-NPU:
+Qwen3 MoE on GPU:
+
+```bash
+export AFD_E2E_BACKEND=gpu
+export AFD_E2E_DEVICES=0,1,2
+export AFD_GPU_E2E_MODEL=/path/to/Qwen3-30B-A3B
+```
+
+DeepSeek-V2-Lite on NPU:
 
 ```bash
 export AFD_E2E_BACKEND=npu
 export AFD_E2E_DEVICES=0,1,2
 export AFD_NPU_E2E_MODEL=/path/to/DeepSeek-V2-Lite
 ```
 
-Then run:
+Then run the selected model suite:
 
 ```bash
+# DeepSeek-V2-Lite
 python -m pytest -q -s \
   "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-eager]" \
   "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph]" \
   "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo]" \
   "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[baseline-graph]"
+
+# Qwen3 MoE
+python -m pytest -q -s \
+  tests/e2e/models/qwen3_moe/test_qwen3_moe.py
 ```
 
 Success means 4 passed and 0 skipped.
@@ -70,8 +83,13 @@ For the weekly full GSM8K test, run only `afd-graph-dbo`:
 
 ```bash
 export AFD_GSM8K_LIMIT=all
+# DeepSeek-V2-Lite
 python -m pytest -q -s \
   "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo]"
+
+# Qwen3 MoE
+python -m pytest -q -s \
+  "tests/e2e/models/qwen3_moe/test_qwen3_moe.py::test_qwen3_moe[afd-graph-dbo]"
 ```
 
 This evaluates all 1319 GSM8K test samples. Without `AFD_GSM8K_LIMIT`, each
@@ -83,8 +101,8 @@ The repository includes the [`run-e2e`](../../.agents/skills/run-e2e/SKILL.md)
 skill. Open the repository in Codex and ask, for example:
 
 ```text
-Use run-e2e to run the GPU E2E tests with model /models/DeepSeek-V2-Lite,
-devices 0,1,2, and HF_HOME /data/huggingface.
+Use run-e2e to run the Qwen3 MoE GPU E2E tests with model
+/models/Qwen3-30B-A3B, devices 0,1,2, and HF_HOME /data/huggingface.
 ```
 
 For either backend, provide the model path, three device IDs, and `HF_HOME`.

tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py

@@ -11,10 +11,10 @@
 
 import pytest
 
-from tests.e2e.models.deepseek_v2_lite.test_deepseek_v2_lite import (
-    _devices,
-    _required_env,
-    _run_runner,
+from tests.e2e.models.model_test_utils import (
+    parse_devices,
+    required_env,
+    run_runner,
 )
 from tests.e2e.runner import (
     ASYNC_CAM_ATTENTION_RANKS,
@@ -67,13 +67,13 @@ def _connector_extra_config(model: str) -> str:
 
 
 def build_runner_command() -> list[str]:
-    backend = _required_env("AFD_E2E_BACKEND")
+    backend = required_env("AFD_E2E_BACKEND")
     if backend != "npu":
         raise RuntimeError("async CAM E2E requires AFD_E2E_BACKEND=npu")
 
     device_count = ASYNC_CAM_ATTENTION_RANKS + ASYNC_CAM_FFN_RANKS
-    devices = _devices("AFD_E2E_DEVICES", device_count)
-    model = _required_env("AFD_NPU_E2E_MODEL")
+    devices = parse_devices("AFD_E2E_DEVICES", device_count)
+    model = required_env("AFD_NPU_E2E_MODEL")
     common_arguments = (
         "--trust-remote-code",
         "--max-num-seqs",
@@ -127,4 +127,4 @@ def build_runner_command() -> list[str]:
 @pytest.mark.e2e
 @pytest.mark.slow
 def test_deepseek_v2_lite_async_cam() -> None:
-    _run_runner(build_runner_command(), env=_async_cam_env())
+    run_runner(build_runner_command(), env=_async_cam_env())

tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py

@@ -5,55 +5,39 @@
 from __future__ import annotations
 
 import os
-import signal
-import subprocess
 import sys
-from contextlib import suppress
 from pathlib import Path
 
 import pytest
 
-REPO_ROOT = Path(__file__).resolve().parents[4]
+from tests.e2e.models.model_test_utils import (
+    parse_devices,
+    required_env,
+    run_runner,
+)
+
 SCENARIOS = (
     "baseline-graph",
     "afd-eager",
     "afd-graph",
     "afd-graph-dbo",
     "afd-graph-dbo-2a2f",
 )
-# Covers the roughly 64-second nested lm-eval/vLLM cleanup bound with buffer.
-RUNNER_CLEANUP_TIMEOUT_S = 90
-
-
-def _required_env(name: str) -> str:
-    value = os.environ.get(name)
-    if not value:
-        raise RuntimeError(f"{name} must be set")
-    return value
-
-
-def _devices(name: str, expected_count: int) -> list[str]:
-    devices = [item.strip() for item in _required_env(name).split(",") if item.strip()]
-    if len(devices) != expected_count:
-        raise RuntimeError(f"{name} must contain exactly {expected_count} devices")
-    if len(devices) != len(set(devices)):
-        raise RuntimeError(f"{name} devices must be unique")
-    return devices
 
 
 def build_runner_command(scenario: str, gsm8k_output_path: Path) -> list[str]:
-    backend = _required_env("AFD_E2E_BACKEND")
+    backend = required_env("AFD_E2E_BACKEND")
     if backend == "gpu":
-        model = _required_env("AFD_GPU_E2E_MODEL")
+        model = required_env("AFD_GPU_E2E_MODEL")
         vllm_bin = os.environ.get("AFD_GPU_E2E_VLLM_BIN", "vllm")
     elif backend == "npu":
-        model = _required_env("AFD_NPU_E2E_MODEL")
+        model = required_env("AFD_NPU_E2E_MODEL")
         vllm_bin = os.environ.get("AFD_NPU_E2E_VLLM_BIN", "vllm")
     else:
         raise RuntimeError("AFD_E2E_BACKEND must be 'gpu' or 'npu'")
 
     expected_device_count = 4 if scenario == "afd-graph-dbo-2a2f" else 3
-    devices = _devices("AFD_E2E_DEVICES", expected_device_count)
+    devices = parse_devices("AFD_E2E_DEVICES", expected_device_count)
     attention_devices = devices[:2]
     ffn_devices = devices[2:]
     if scenario == "baseline-graph":
@@ -85,61 +69,8 @@ def build_runner_command(scenario: str, gsm8k_output_path: Path) -> list[str]:
     return command
 
 
-def _run_runner(command: list[str], env: dict[str, str] | None = None) -> None:
-    handled_signals = (signal.SIGTERM, signal.SIGINT)
-    previous_handlers = {signum: signal.getsignal(signum) for signum in handled_signals}
-    process: subprocess.Popen | None = None
-    received_signal: int | None = None
-    forwarded = False
-
-    def forward_received_signal() -> None:
-        nonlocal forwarded
-        if process is None or received_signal is None or forwarded:
-            return
-        forwarded = True
-        with suppress(ProcessLookupError):
-            os.killpg(process.pid, received_signal)
-        raise SystemExit(128 + received_signal)
-
-    def forward_cancellation(signum, _frame) -> None:
-        nonlocal received_signal
-        if received_signal is not None:
-            return
-        received_signal = signum
-        forward_received_signal()
-
-    for signum in handled_signals:
-        signal.signal(signum, forward_cancellation)
-
-    try:
-        process = subprocess.Popen(
-            command,
-            cwd=REPO_ROOT,
-            env=env,
-            start_new_session=True,
-        )
-        forward_received_signal()
-        returncode = process.wait()
-        if returncode != 0:
-            raise subprocess.CalledProcessError(returncode, command)
-    finally:
-        try:
-            if process is not None and process.poll() is None:
-                try:
-                    process.wait(timeout=RUNNER_CLEANUP_TIMEOUT_S)
-                except subprocess.TimeoutExpired:
-                    try:
-                        with suppress(ProcessLookupError):
-                            os.killpg(process.pid, signal.SIGKILL)
-                    finally:
-                        process.wait()
-        finally:
-            for signum, previous_handler in previous_handlers.items():
-                signal.signal(signum, previous_handler)
-
-
 @pytest.mark.e2e
 @pytest.mark.parametrize("scenario", SCENARIOS, ids=SCENARIOS)
 def test_deepseek_v2_lite(scenario: str, tmp_path: Path) -> None:
     command = build_runner_command(scenario, tmp_path / scenario)
-    _run_runner(command)
+    run_runner(command)