Refine test cases

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

@@ -1,6 +1,6 @@
 ---
 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, or DBO scenarios.
+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.
 ---
 
 # Run AFD E2E Tests
@@ -15,9 +15,8 @@ tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py:
 - afd-graph
 - afd-graph-dbo
 
-Each scenario evaluates the first 7 GSM8K samples with 2 Attention ranks and 1
-FFN rank. Do not use removed markers, categories, TP, or 2A2F workflows. This
-skill does not cover unit tests or source edits.
+Each default scenario evaluates the first 7 GSM8K samples with 2 Attention
+ranks and 1 FFN rank. `afd-graph-dbo-2a2f` is a separate opt-in case.
 
 ## Workflow
 
@@ -80,11 +79,25 @@ From the repository root, stream output in the foreground:
 
 ~~~bash
 python -m pytest -q -s \
-  tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py
+  "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]"
 ~~~
 
 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:
+
+~~~bash
+export AFD_E2E_DEVICES=0,1,2,3
+python -m pytest -q -s \
+  "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo-2a2f]"
+~~~
+
+The first two devices run Attention DP2/TP1. The last two run FFN
+DP2/TP1/EP2.
+
 On cancellation, forward SIGTERM and allow over 90 seconds for cleanup.
 
 ### 5. Report
@@ -97,7 +110,7 @@ actionable error, and cleanup status. Any skip is a gate failure.
 | Variable | Backend | Required |
 |---|---|---|
 | AFD_E2E_BACKEND | both | yes: gpu or npu |
-| AFD_E2E_DEVICES | both | yes: exactly 3 unique IDs |
+| AFD_E2E_DEVICES | both | yes: 3 default or 4 for 2A2F |
 | AFD_GPU_E2E_MODEL | GPU | yes |
 | AFD_GPU_E2E_VLLM_BIN | GPU | no; defaults to vllm |
 | AFD_NPU_E2E_MODEL | NPU | yes |

docs/design/module/e2e_testing.md

@@ -24,6 +24,7 @@ validation_paths:
   - "tests/unit/test_e2e_runner.py"
   - "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"
 upstream_refs:
   - "vLLM 0.26.0 serving and shutdown interfaces"
   - "lm-evaluation-harness GSM8K task and local-completions API"
@@ -50,7 +51,7 @@ operator tests, benchmarks, and performance tests are outside this scope.
 
 ```text
 tests/e2e/
-├── models/<model>/test_<model>.py  # Test entry and case list
+├── models/<model>/test_*.py  # Test entries and case lists
 ├── accuracy/<task>.py       # Accuracy tool and result parsing
 ├── runner.py                # Service startup, evaluation, and cleanup
 └── process_utils.py         # Process-group termination and reaping
@@ -72,7 +73,8 @@ cleanup. Production code does not depend on the E2E harness.
   incomplete results, failed validation, or failed cleanup. They **MUST NOT**
   use `skip`, `xfail`, or success-on-empty behavior.
 - `E2E-INV-005` — The harness **MUST** check service liveness before and after
-  evaluation, evaluator exit status, sample count, `NaN`, and accuracy.
+  evaluation. Accuracy cases **MUST** also check evaluator exit status, sample
+  count, `NaN`, and accuracy.
 - `E2E-INV-006` — Child processes **MUST** use owned process groups.
   Cancellation **MUST** send `SIGTERM`, use a bounded grace period, then reap
   every leader. A harness `SIGKILL` escalation **MUST** fail the case.
@@ -97,6 +99,13 @@ for merge validation. Put slower coverage in a scheduled job.
 Prefer graph coverage. Keep one eager smoke test unless a feature cannot run in
 graph mode.
 
+`afd-eager-async-cam` is a separate NPU-only smoke test. It uses four devices
+for Attention DP1/TP2 and FFN DP2/TP1/EP2. It is not part of the PR gate above.
+
+`afd-graph-dbo-2a2f` is a separate GPU/NPU accuracy case. It uses four devices
+for Attention DP2/TP1 and FFN DP2/TP1/EP2 and runs GSM8K-7. The default gate
+selects only the four default cases by pytest node ID.
+
 ## Accuracy gate
 
 | Setting | PR | Weekly |

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.
-They run four scenarios:
+The default gate runs four scenarios:
 
 - `baseline-graph`
 - `afd-eager`
@@ -47,11 +47,25 @@ Then run:
 
 ```bash
 python -m pytest -q -s \
-  tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py
+  "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]"
 ```
 
 Success means 4 passed and 0 skipped.
 
+### Graph + DBO 2A2F
+
+This separate GPU/NPU case uses four devices: the first two for Attention
+DP=2/TP=1 and the last two for FFN DP=2/TP=1/EP=2. It runs GSM8K-7.
+
+```bash
+export AFD_E2E_DEVICES=0,1,2,3
+python -m pytest -q -s \
+  "tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py::test_deepseek_v2_lite[afd-graph-dbo-2a2f]"
+```
+
 For the weekly full GSM8K test, run only `afd-graph-dbo`:
 
 ```bash
@@ -76,3 +90,21 @@ devices 0,1,2, and HF_HOME /data/huggingface.
 For either backend, provide the model path, three device IDs, and `HF_HOME`.
 The skill checks prerequisites, runs the same four tests, and reports failures
 and process cleanup.
+
+## NPU async CAM smoke test
+
+This separate test uses four NPUs: the first two for Attention TP=2, and the
+last two for FFN DP=2/TP=1/EP=2. It sends one prompt and requests 32 tokens. It
+does not run GSM8K.
+
+```bash
+export AFD_E2E_BACKEND=npu
+export AFD_E2E_DEVICES=0,1,2,3
+export AFD_NPU_E2E_MODEL=/path/to/DeepSeek-V2-Lite
+python -m pytest -q -s \
+  tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py
+```
+
+The CAM/CANN runtime and custom operators must already be installed. Missing
+model configuration or a device list other than four unique IDs fails the
+test.

tests/e2e/models/deepseek_v2_lite/test_async_cam_npu.py

@@ -0,0 +1,130 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the AFD plugin project
+"""NPU smoke test for DeepSeek-V2-Lite with async CAM."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.models.deepseek_v2_lite.test_deepseek_v2_lite import (
+    _devices,
+    _required_env,
+    _run_runner,
+)
+from tests.e2e.runner import (
+    ASYNC_CAM_ATTENTION_RANKS,
+    ASYNC_CAM_FFN_RANKS,
+    ASYNC_CAM_SCENARIO,
+)
+
+CAM_VENDOR_PATH = Path("/usr/local/Ascend/cann-9.0.1/opp/vendors/CAM")
+CAM_OP_API_PATH = CAM_VENDOR_PATH / "op_api"
+CAM_OP_API_LIB_PATH = CAM_OP_API_PATH / "lib"
+CAM_HCCL_BUFFER_SIZE = "4096"
+CAM_MAX_NUM_SEQUENCES = "8"
+CAM_MAX_BATCHED_TOKENS = "8000"
+CAM_MEMORY_UTILIZATION = "0.75"
+
+
+def _prepend_env_paths(env: dict[str, str], name: str, *paths: Path) -> None:
+    existing = [path for path in env.get(name, "").split(os.pathsep) if path]
+    env[name] = os.pathsep.join(
+        dict.fromkeys([*(str(path) for path in paths), *existing]),
+    )
+
+
+def _async_cam_env() -> dict[str, str]:
+    env = os.environ.copy()
+    env.setdefault("VLLM_USE_V1", "1")
+    env["HCCL_BUFFSIZE"] = CAM_HCCL_BUFFER_SIZE
+    env.setdefault("PYTORCH_NPU_ALLOC_CONF", "expandable_segments:True")
+    env.setdefault("ASCEND_LAUNCH_BLOCKING", "1")
+    env.setdefault("VLLM_ASCEND_ENABLE_CONTEXT_PARALLEL", "1")
+    env.setdefault("VLLM_ASCEND_ENABLE_FLASHCOMM1", "1")
+    _prepend_env_paths(
+        env,
+        "LD_LIBRARY_PATH",
+        CAM_OP_API_PATH,
+        CAM_OP_API_LIB_PATH,
+    )
+    _prepend_env_paths(env, "ASCEND_CUSTOM_OPP_PATH", CAM_VENDOR_PATH)
+    return env
+
+
+def _connector_extra_config(model: str) -> str:
+    dynamic_quant = int(
+        os.environ.get(
+            "AFD_NPU_ASYNC_CAM_E2E_DYNAMIC_QUANT",
+            "1" if (Path(model) / "quant_model_description.json").is_file() else "0",
+        ),
+    )
+    return json.dumps({"dynamicQuant": dynamic_quant}, separators=(",", ":"))
+
+
+def build_runner_command() -> list[str]:
+    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")
+    common_arguments = (
+        "--trust-remote-code",
+        "--max-num-seqs",
+        CAM_MAX_NUM_SEQUENCES,
+        "--max-num-batched-tokens",
+        CAM_MAX_BATCHED_TOKENS,
+        "--gpu-memory-utilization",
+        CAM_MEMORY_UTILIZATION,
+        "--no-enable-prefix-caching",
+    )
+    command = [
+        sys.executable,
+        "-m",
+        "tests.e2e.runner",
+        "--model",
+        model,
+        "--vllm-bin",
+        os.environ.get("AFD_NPU_E2E_VLLM_BIN", "vllm"),
+        "--device-backend",
+        "npu",
+        "--attention-devices",
+        ",".join(devices[:ASYNC_CAM_ATTENTION_RANKS]),
+        "--ffn-devices",
+        ",".join(devices[ASYNC_CAM_ATTENTION_RANKS:]),
+        "--scenario",
+        ASYNC_CAM_SCENARIO,
+        "--served-model-name-prefix",
+        "cam-async",
+        "--afd-connector-extra-config",
+        _connector_extra_config(model),
+        "--api-port-base",
+        os.environ.get("AFD_NPU_ASYNC_CAM_E2E_API_PORT", "19080"),
+        "--afd-port",
+        os.environ.get("AFD_NPU_ASYNC_CAM_E2E_AFD_PORT", "6453"),
+        "--startup-timeout",
+        os.environ.get("AFD_NPU_E2E_STARTUP_TIMEOUT", "900"),
+        *(f"--common-vllm-arg={argument}" for argument in common_arguments),
+    ]
+    max_model_len = os.environ.get("AFD_NPU_ASYNC_CAM_E2E_MAX_MODEL_LEN")
+    if max_model_len:
+        command.extend(
+            [
+                "--common-vllm-arg=--max-model-len",
+                f"--common-vllm-arg={max_model_len}",
+            ],
+        )
+    return command
+
+
+@pytest.mark.npu
+@pytest.mark.e2e
+@pytest.mark.slow
+def test_deepseek_v2_lite_async_cam() -> None:
+    _run_runner(build_runner_command(), env=_async_cam_env())

tests/e2e/models/deepseek_v2_lite/test_deepseek_v2_lite.py

@@ -19,6 +19,7 @@
     "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
@@ -51,7 +52,8 @@ def build_runner_command(scenario: str, gsm8k_output_path: Path) -> list[str]:
     else:
         raise RuntimeError("AFD_E2E_BACKEND must be 'gpu' or 'npu'")
 
-    devices = _devices("AFD_E2E_DEVICES", 3)
+    expected_device_count = 4 if scenario == "afd-graph-dbo-2a2f" else 3
+    devices = _devices("AFD_E2E_DEVICES", expected_device_count)
     attention_devices = devices[:2]
     ffn_devices = devices[2:]
     if scenario == "baseline-graph":
@@ -83,7 +85,7 @@ def build_runner_command(scenario: str, gsm8k_output_path: Path) -> list[str]:
     return command
 
 
-def _run_runner(command: list[str]) -> None:
+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
@@ -113,6 +115,7 @@ def forward_cancellation(signum, _frame) -> None:
         process = subprocess.Popen(
             command,
             cwd=REPO_ROOT,
+            env=env,
             start_new_session=True,
         )
         forward_received_signal()