Fix: make SDMA a declared capability, and drop stale ci.yml line refs

.claude/skills/testing/SKILL.md

@@ -13,9 +13,11 @@ description: Testing guide and pre-commit testing strategy for simpler. Use when
 
 ## Running Tests
 
-**Important**: Always read `.github/workflows/ci.yml` first to extract both the
-current `--pto-session-timeout` values **and the `--ignore` sets the onboard
-sweep carries**. PTO-ISA reproducibility comes from the repo-root `pto_isa.pin`.
+**Important**: Always read `.github/workflows/ci.yml` first for the current
+`--pto-session-timeout` values. Quarantines are **not** in the workflow — they
+are markers on the tests, so mirror the sweep with `-m "not sdma"` rather than
+copying a path list. PTO-ISA reproducibility comes from the repo-root
+`pto_isa.pin`.
 
 **CI does not run one flat sweep.** Some tests are quarantined out of the
 general onboard sweep and run in their own step on their own devices, because
@@ -25,7 +27,7 @@ and will report failures that CI never sees:
 
 | Quarantine | Excludes | Runs instead in |
 | ---------- | -------- | --------------- |
-| `SDMA_IGNORE` | `sdma_async_completion_demo`, `prefetch_async_demo` | dedicated "SDMA pytest (a2a3)" step, `--device-num 2` |
+| `@pytest.mark.sdma` | `sdma_async_completion_demo`, `prefetch_async_demo` | dedicated "SDMA pytest (a2a3)" step, `--device-num 2` |
 
 The SDMA demos provision 48 device-only STARS streams, which makes an AICore
 fault take ~306 s to tear down instead of ~0.3 s — so they must not share a
@@ -34,9 +36,10 @@ sweep with the `aicore_op_timeout` fault-injection test
 issue #1425).
 
 When an onboard test fails, **run it alone before calling it a regression**.
-Alone-passes plus sweep-fails is an isolation requirement, not a defect: grep
-`ci.yml` for an `--ignore` naming it. Do not "fix" such a test by reordering it
-earlier — that only moves the pollution onto whatever now runs after it.
+Alone-passes plus sweep-fails is an isolation requirement, not a defect: check
+the test for a quarantine marker such as `@pytest.mark.sdma`. Do not "fix" such
+a test by reordering it earlier — that only moves the pollution onto whatever
+now runs after it.
 
 ### Runtime rebuild decision
 
@@ -68,9 +71,9 @@ ctest --test-dir tests/ut/cpp/build -L "^requires_hardware(_a2a3)?$" --output-on
 pytest examples tests/st --platform a2a3sim \
     --pto-session-timeout <timeout>
 
-# All hardware scene tests — mirror ci.yml: carry its --ignore set, or the
-# quarantined tests fail here and nowhere else (extract it from ci.yml)
-pytest examples tests/st $SDMA_IGNORE --platform a2a3 --device <range> \
+# All hardware scene tests — mirror ci.yml: deselect the quarantined marker, or
+# those tests fail here and nowhere else
+pytest examples tests/st -m "not sdma" --platform a2a3 --device <range> \
     --pto-session-timeout <timeout>
 
 # The quarantined tests, the way CI runs them — alone, on their own devices

.github/workflows/ci.yml

@@ -597,12 +597,16 @@ jobs:
           # — which also runs the aicore_op_timeout fault-injection test — and run
           # them on their own dedicated device(s) in the SDMA step below, so an
           # SDMA fault's slow teardown can never mix with ordinary fault recovery.
-          SDMA_IGNORE="--ignore=examples/a2a3/tensormap_and_ringbuffer/prefetch_async_demo --ignore=examples/a2a3/tensormap_and_ringbuffer/sdma_async_completion_demo"
+          #
+          # Selected by marker, not by path. `pytest --ignore=<moved path>` exits
+          # 0 and says nothing, so a rename used to drop the quarantine silently
+          # and land these back beside the fault-injection test. `-m` travels
+          # with the test instead.
           if [ "$(uname -m)" = "x86_64" ]; then
-            python -m pytest examples tests/st $SDMA_IGNORE --platform a2a3 --device ${DEVICE_RANGE} -v --require-pto-isa --pto-session-timeout 600
+            python -m pytest examples tests/st -m "not sdma" --platform a2a3 --device ${DEVICE_RANGE} -v --require-pto-isa --pto-session-timeout 600
           else
             task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" \
-              --run "python -m pytest examples tests/st $SDMA_IGNORE --platform a2a3 --device \$TASK_DEVICE -v --require-pto-isa --pto-session-timeout 600"
+              --run "python -m pytest examples tests/st -m 'not sdma' --platform a2a3 --device \$TASK_DEVICE -v --require-pto-isa --pto-session-timeout 600"
           fi
 
       # SDMA pytest — every SDMA test runs here on its own dedicated device(s),
@@ -615,7 +619,12 @@ jobs:
         run: |
           source /usr/local/Ascend/cann/set_env.sh
           source .venv/bin/activate
-          SDMA_TESTS="examples/a2a3/tensormap_and_ringbuffer/prefetch_async_demo examples/a2a3/tensormap_and_ringbuffer/sdma_async_completion_demo"
+          # Selected by marker, so this step cannot drift from what the sweep
+          # deselects. Kept as a step of its own until #1425 is fixed: an
+          # AICore fault on a device that has provisioned SDMA still costs
+          # minutes, so these must not share a device with the fault-injection
+          # cases even though the in-session ordering already separates them.
+          SDMA_TESTS="examples tests/st -m sdma"
           if [ "$(uname -m)" = "x86_64" ]; then
             python -m pytest $SDMA_TESTS --platform a2a3 --device ${DEVICE_RANGE} -v --require-pto-isa --pto-session-timeout 600
           else

conftest.py

@@ -407,6 +407,18 @@ def pytest_configure(config):
     config.addinivalue_line("markers", "platforms(list): supported platforms for standalone ST functions")
     config.addinivalue_line("markers", "requires_hardware: test needs Ascend toolchain and real device")
     config.addinivalue_line("markers", "device_count(n): number of NPU devices needed")
+    config.addinivalue_line(
+        "markers",
+        "sdma: the test provisions the PTO-ISA async-SDMA workspace, so its "
+        "Worker is built with enable_sdma=True. Provisioning creates 48 "
+        "device-only STARS streams that sit in the device fault domain, which "
+        "makes a later AICore fault on that device cost minutes instead of "
+        "~0.3 s (#1425). Two consequences follow from the one marker: such a "
+        "test never shares an L2 Worker (the pool key carries the flag) and it "
+        "sorts after every ordinary test, so fault-injection cases run on a "
+        "device that has never provisioned. CI additionally runs them in a step "
+        "of their own via -m sdma until #1425 is fixed",
+    )
     config.addinivalue_line(
         "markers",
         "runtime(name): runtime this standalone test targets; used by runtime-isolation subprocess "
@@ -566,7 +578,13 @@ def pytest_collection_modifyitems(session, config, items):  # noqa: PLR0912
     def sort_key(item):
         cls = getattr(item, "cls", None)
         level = getattr(cls, "_st_level", 0) if cls else 0
-        return (0 if level >= 3 else 1, item.nodeid)
+        # SDMA last, for the same class of reason L3 goes first: provisioning
+        # the workspace leaves 48 STARS streams in the device's fault domain,
+        # so every fault-injection case must have already run on a device that
+        # never provisioned (#1425). Keyed off the marker, not the class, since
+        # the fault-injection tests are plain functions with no _st_level.
+        sdma_last = 1 if item.get_closest_marker("sdma") else 0
+        return (sdma_last, 0 if level >= 3 else 1, item.nodeid)
 
     items.sort(key=sort_key)
 
@@ -1361,9 +1379,17 @@ def st_worker(request, st_platform, device_pool, _l2_worker_pool, _l2_poisoned):
         # st_device_ids) that also draw from it; retaining the id would drain
         # the pool and break any non-st_worker test that runs afterward on the
         # same xdist worker.
-        for (rt, dev_id), existing in _l2_worker_pool.items():
-            if rt == runtime:
-                _register_l2_pool_recycle(request, _l2_worker_pool, (rt, dev_id), _l2_poisoned)
+        # The SDMA capability is part of the Worker's identity, not a per-run
+        # option: an enable_sdma Worker holds 48 STARS streams for its whole
+        # life, so it must never be handed to a test that did not ask for them,
+        # nor a plain Worker to one that did. It therefore takes a slot in the
+        # pool key and gates reuse. Ordering puts every sdma test after the
+        # rest, so the swap happens once, at the end.
+        wants_sdma = request.node.get_closest_marker("sdma") is not None
+
+        for (rt, dev_id, pooled_sdma), existing in _l2_worker_pool.items():
+            if rt == runtime and pooled_sdma == wants_sdma:
+                _register_l2_pool_recycle(request, _l2_worker_pool, (rt, dev_id, pooled_sdma), _l2_poisoned)
                 yield existing
                 return
 
@@ -1372,7 +1398,7 @@ def st_worker(request, st_platform, device_pool, _l2_worker_pool, _l2_poisoned):
             pytest.fail(f"no devices available in --device pool (requested 1, pool has {len(device_pool._available)})")
         dev_id = ids[0]
         device_pool.release(ids)
-        key = (runtime, dev_id)
+        key = (runtime, dev_id, wants_sdma)
 
         # At most one runtime-specific Worker may own a device: finalization
         # resets resources that would invalidate every other Worker on it.
@@ -1387,7 +1413,7 @@ def st_worker(request, st_platform, device_pool, _l2_worker_pool, _l2_poisoned):
 
         from simpler.worker import Worker  # noqa: PLC0415
 
-        w = Worker(level=2, device_id=dev_id, platform=st_platform, runtime=runtime)
+        w = Worker(level=2, device_id=dev_id, platform=st_platform, runtime=runtime, enable_sdma=wants_sdma)
         w._st_device_id = dev_id
         # First rebuild after a poison-and-heal lands here. On arches where the
         # device re-inits cleanly this just works; on a5 the op-timeout poison
@@ -1434,6 +1460,7 @@ def st_worker(request, st_platform, device_pool, _l2_worker_pool, _l2_poisoned):
             num_sub_workers=max_subs,
             platform=st_platform,
             runtime=runtime,
+            enable_sdma=request.node.get_closest_marker("sdma") is not None,
         )
         w._st_device_id = ids[0]  # expose primary device to test_run for profiling snapshots
 

docs/capability-survey.md

@@ -172,7 +172,7 @@ values**: op-execute 45 s, stream-sync 50 s, scheduler 10 s
 env-overridable with ordering validation
 (`resolve_onboard_timeout_config`, `device_runner_base.cpp:66-110`;
 [troubleshooting/local-timeout-defaults.md](troubleshooting/local-timeout-defaults.md)),
-and **CI runs at 2 s / 3 s / 4 s** (`.github/workflows/ci.yml:480-482`). Triage
+and **CI runs at 2 s / 3 s / 4 s** (the `SIMPLER_*_TIMEOUT_*` env block on each self-hosted job in `.github/workflows/ci.yml`). Triage
 a CI timeout against the CI values, not the header constants.
 
 Every failed launch runs a recovery path that is easy to miss:
@@ -217,8 +217,8 @@ values yield `PTO2_ERROR_ASYNC_COMPLETION_INVALID`.
 
 | Engine | a2a3 | a5 | Status |
 | ------ | ---- | -- | ------ |
-| COUNTER (default) | registered | registered | **Shipped** — `async_notify_demo` runs onboard on both arches (`ci.yml:607`, `:884`); `deferred_notify_demo` runs in sim on both (`ci.yml:216`, `:310`). Routed by `@pytest.mark.platforms`, no `skipif` |
-| SDMA | build macro forced ON; runtime opt-in | `option(... OFF)` | a2a3 **Shipped** (dedicated CI step, `ci.yml:628-643`); a5 not built |
+| COUNTER (default) | registered | registered | **Shipped** — `async_notify_demo` runs onboard on both arches and `deferred_notify_demo` runs in sim on both, through the `st-onboard-*` / `st-sim-*` jobs in `ci.yml`. Routed by `@pytest.mark.platforms`, no `skipif` |
+| SDMA | build macro forced ON; runtime opt-in | `option(... OFF)` | a2a3 **Shipped** (the "SDMA pytest (a2a3)" step in `ci.yml`); a5 not built |
 | URMA | absent | full implementation | **Gated** — see below |
 | ROCE, CCU | enum only | enum only | **Name only** |
 
@@ -227,7 +227,7 @@ compiled, but provisioning the 48 STARS streams requires
 `Worker(..., enable_sdma=True)`, default `False`
 (`python/simpler/worker.py:4178`, `:4396`;
 `python/bindings/task_interface.cpp:1367`). It is quarantined from the general
-CI sweep via `SDMA_IGNORE` (`ci.yml:600`) for a measured hazard: with 48
+CI sweep via `@pytest.mark.sdma` for a measured hazard: with 48
 device-only SDMA streams an AICore fault takes ~306 s to tear down versus ~0.3 s
 without, traced to a single 300,000 ms remote TRS event timeout
 ([investigations/2026-07-a2a3-sdma-fault-teardown.md](investigations/2026-07-a2a3-sdma-fault-teardown.md),

examples/a2a3/tensormap_and_ringbuffer/prefetch_async_demo/test_prefetch_async_demo.py

@@ -29,112 +29,61 @@
 
 from __future__ import annotations
 
-import argparse
-import os
-
 import pytest
 import torch
-from simpler.task_interface import (
-    ArgDirection,
-    CallConfig,
-    ChipCallable,
-    ChipStorageTaskArgs,
-    CoreCallable,
-)
-from simpler.worker import Worker
+from simpler.task_interface import ArgDirection as D
 
-from simpler_setup.elf_parser import extract_text_section
-from simpler_setup.kernel_compiler import KernelCompiler
-from simpler_setup.pto_isa import ensure_pto_isa_root
-from simpler_setup.torch_interop import make_tensor_arg
+from simpler_setup import SceneTestCase, TaskArgsBuilder, Tensor, scene_test
 
-HERE = os.path.dirname(os.path.abspath(__file__))
-RUNTIME = "tensormap_and_ringbuffer"
 N = 128
 
 
-def build_chip_callable(platform: str) -> ChipCallable:
-    kc = KernelCompiler(platform=platform)
-    pto_isa_root = ensure_pto_isa_root()
-    include_dirs = kc.get_orchestration_include_dirs(RUNTIME)
-    extra_includes = list(include_dirs) + [str(kc.project_root / "src" / "common")]
-
-    # in (IN), out (OUT) -- the SDMA workspace is injected, not an arg.
-    signature = [ArgDirection.IN, ArgDirection.OUT]
+@pytest.mark.sdma
+@scene_test(level=2, runtime="tensormap_and_ringbuffer")
+class TestPrefetchAsyncDemo(SceneTestCase):
+    """Prefetching a GM region then copying it leaves the data bit-exact."""
 
-    kernel = kc.compile_incore(
-        source_path=os.path.join(HERE, "kernels/aiv/kernel_prefetch_copy.cpp"),
-        core_type="aiv",
-        pto_isa_root=pto_isa_root,
-        extra_include_dirs=extra_includes,
-    )
-    if not platform.endswith("sim"):
-        kernel = extract_text_section(kernel)
-    children = [
-        (
-            0,
-            CoreCallable.build(
-                signature=signature,
-                binary=kernel,
-            ),
-        )
+    CALLABLE = {
+        "orchestration": {
+            "source": "kernels/orchestration/prefetch_async_orch.cpp",
+            "function_name": "prefetch_async_orchestration",
+            # in (IN), out (OUT) — the SDMA workspace is injected into every
+            # kernel's GlobalContext by the enable_sdma Worker, not threaded
+            # through as a user arg.
+            "signature": [D.IN, D.OUT],
+        },
+        "incores": [
+            {
+                "func_id": 0,
+                "source": "kernels/aiv/kernel_prefetch_copy.cpp",
+                "core_type": "aiv",
+                "signature": [D.IN, D.OUT],
+            },
+        ],
+    }
+
+    CASES = [
+        {
+            "name": "prefetch_copy",
+            "platforms": ["a2a3"],
+            "config": {},
+            "params": {},
+        },
     ]
 
-    orch = kc.compile_orchestration(
-        runtime_name=RUNTIME,
-        source_path=os.path.join(HERE, "kernels/orchestration/prefetch_async_orch.cpp"),
-        extra_include_dirs=[str(kc.project_root / "src" / "common")],
-    )
-    return ChipCallable.build(
-        signature=signature,
-        func_name="prefetch_async_orchestration",
-        binary=orch,
-        children=children,
-    )
-
-
-def run(platform: str = "a2a3", device_id: int = 0) -> int:
-    if platform.endswith("sim"):
-        raise ValueError("prefetch_async_demo requires onboard hardware")
-
-    src = torch.arange(N, dtype=torch.float32) / 8.0
-    out = torch.full((N,), -1.0, dtype=torch.float32)
-
-    chip_callable = build_chip_callable(platform)
-    worker = Worker(level=2, platform=platform, runtime=RUNTIME, device_id=device_id, enable_sdma=True)
-    worker.init()
-    try:
-        handle = worker.register(chip_callable)
-        args = ChipStorageTaskArgs()
-        args.add_tensor(make_tensor_arg(src))
-        args.add_tensor(make_tensor_arg(out))
-        worker.run(handle, args, CallConfig())
-    finally:
-        worker.close()
-
-    if not torch.equal(out, src):
-        bad = int((out != src).sum().item())
-        first = int((out != src).nonzero()[0].item())
-        print(
-            f"[ERROR] prefetch_async_demo mismatch count={bad}, first={first}, "
-            f"got={float(out[first])}, expect={float(src[first])}"
+    def generate_args(self, params):
+        return TaskArgsBuilder(
+            Tensor("src", torch.arange(N, dtype=torch.float32) / 8.0),
+            Tensor("out", torch.full((N,), -1.0, dtype=torch.float32)),
         )
-        return 1
-    print("[INFO] prefetch_async_demo: out matches src after injected SDMA prefetch + copy")
-    return 0
 
-
-@pytest.mark.platforms(["a2a3"])
-@pytest.mark.runtime("tensormap_and_ringbuffer")
-@pytest.mark.device_count(1)
-def test_prefetch_async_demo(st_platform, st_device_ids) -> None:
-    """Prefetching a GM region then copying it leaves the data bit-exact."""
-    assert run(platform=st_platform, device_id=int(st_device_ids[0])) == 0
+    def compute_golden(self, args, params):
+        # The prefetch is a pure cache hint that changes no value, so the copy
+        # must reproduce the source bit-for-bit. What the case really proves is
+        # that the injected workspace was real and the event wait completed
+        # rather than hanging.
+        args.out[:] = args.src
 
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--platform", default="a2a3")
-    parser.add_argument("--device", type=int, default=0)
-    cli = parser.parse_args()
-    raise SystemExit(run(platform=cli.platform, device_id=cli.device))
+    SceneTestCase.run_module(__name__)

examples/a2a3/tensormap_and_ringbuffer/sdma_async_completion_demo/test_sdma_async_completion_demo.py

@@ -189,6 +189,7 @@ def orch_fn(orch, _args, cfg):
         worker.close()
 
 
+@pytest.mark.sdma
 @pytest.mark.platforms(["a2a3"])
 @pytest.mark.runtime("tensormap_and_ringbuffer")
 @pytest.mark.device_count(2)

simpler_setup/scene_test.py

@@ -89,6 +89,15 @@ def clear_compile_cache() -> None:
 #
 # Results are unaffected: the same fixture golden-computed at 320 and at 8
 # threads is bit-identical across `out`, `k_cache` and `v_cache`.
+def _class_wants_sdma(cls) -> bool:
+    """True when the SceneTestCase carries ``@pytest.mark.sdma``.
+
+    Read from ``cls.pytestmark`` rather than a pytest item so the standalone
+    ``python test_x.py`` path sees the same declaration the pytest path does.
+    """
+    return any(getattr(m, "name", None) == "sdma" for m in getattr(cls, "pytestmark", ()))
+
+
 _GOLDEN_MAX_THREADS = 8
 
 
@@ -1133,7 +1142,13 @@ def _create_worker(cls, platform, device_id=0):
         """
         from simpler.worker import Worker  # noqa: PLC0415
 
-        w = Worker(level=2, device_id=device_id, platform=platform, runtime=cls._st_runtime)
+        w = Worker(
+            level=2,
+            device_id=device_id,
+            platform=platform,
+            runtime=cls._st_runtime,
+            enable_sdma=_class_wants_sdma(cls),
+        )
         w.init()
         return w
 
@@ -2065,6 +2080,7 @@ def _create_standalone_worker(group, level, args, selected_by_cls):
         num_sub_workers=max_subs,
         platform=args.platform,
         runtime=first_cls._st_runtime,
+        enable_sdma=any(_class_wants_sdma(c) for c in group),
     )
     # Prepare sub callables per-class to avoid name collisions.
     per_class_sub_handles: dict[type, dict] = {}