feat: support pipelined orchestration and scheduling across tokens and requests

.github/workflows/ci.yml

@@ -743,23 +743,25 @@ jobs:
       - name: Run Python hardware unit tests (a5)
         run: |
           source .venv/bin/activate
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "python -m pytest tests -m requires_hardware --platform a5 --device ${DEVICE_RANGE} -v"
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          PYTEST_BASETEMP="${RUNNER_TEMP}/pytest-${GITHUB_JOB}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" \
+            --run "python -m pytest tests -m requires_hardware --platform a5 \
+            --device \$TASK_DEVICE --basetemp $PYTEST_BASETEMP -v"
 
       - name: Build and run C++ hardware unit tests (a5)
         run: |
           source .venv/bin/activate
           cmake -B tests/ut/cpp/build -S tests/ut/cpp -DSIMPLER_ENABLE_HARDWARE_TESTS=ON
           cmake --build tests/ut/cpp/build
-          python3 -c "
-          import json, os
-          p = os.environ['DEVICE_RANGE'].split('-'); s, e = p[0], p[-1]  # tolerate a single id (no hyphen)
-          npus = [{'id': str(i), 'slots': 1} for i in range(int(s), int(e)+1)]
-          json.dump({'version': {'major': 1, 'minor': 0}, 'local': [{'npus': npus}]},
-                    open('tests/ut/cpp/build/resources.json', 'w'))
-          "
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "ctest --test-dir tests/ut/cpp/build -L '^requires_hardware(_a5)?\$' --resource-spec-file $PWD/tests/ut/cpp/build/resources.json -j$(nproc) --output-on-failure"
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          RESOURCE_SPEC="$PWD/tests/ut/cpp/build/resources.json"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" \
+            --run "python3 -c 'import json, os; ids=os.environ[\"TASK_DEVICE\"].split(\",\"); \
+            spec={\"version\":{\"major\":1,\"minor\":0},\"local\":[{\"npus\":[{\"id\":i,\"slots\":1} for i in ids]}]}; \
+            json.dump(spec, open(\"$RESOURCE_SPEC\", \"w\"))' && \
+            ctest --test-dir tests/ut/cpp/build -L '^requires_hardware(_a5)?\$' \
+            --resource-spec-file $RESOURCE_SPEC -j$(nproc) --output-on-failure"
 
       - name: Build cann-examples/query (CANN host-API smoke)
         run: |
@@ -813,9 +815,12 @@ jobs:
       - name: Run pytest scene tests (a5)
         run: |
           source .venv/bin/activate
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
-          PYTEST="python -m pytest examples tests/st --platform a5 --device ${DEVICE_RANGE} -v --require-pto-isa"
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "$PYTEST --pto-session-timeout 1200"
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          PYTEST_BASETEMP="${RUNNER_TEMP}/pytest-${GITHUB_JOB}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          PYTEST="python -m pytest examples tests/st --platform a5 --device \$TASK_DEVICE \
+            --basetemp $PYTEST_BASETEMP -v --require-pto-isa"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" \
+            --run "$PYTEST --pto-session-timeout 1200"
 
       # DFX per-feature smokes — hardware mirror of the st-sim-a5 set. The
       # race window in dep_gen only fires on real timing, so this row is
@@ -824,35 +829,43 @@ jobs:
       - name: dep_gen smoke
         run: |
           source .venv/bin/activate
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          PYTEST_BASETEMP="${RUNNER_TEMP}/pytest-${GITHUB_JOB}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
           PYTEST="python -m pytest tests/st/a5/tensormap_and_ringbuffer/dfx/dep_gen/test_dep_gen.py \
-            --platform a5 --device ${DEVICE_RANGE} -p no:xdist --pto-session-timeout 600 \
+            --platform a5 --device \$TASK_DEVICE --basetemp $PYTEST_BASETEMP \
+            -p no:xdist --pto-session-timeout 600 \
             --require-pto-isa --enable-dep-gen"
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "$PYTEST"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" --run "$PYTEST"
 
       - name: l2_swimlane smoke
         run: |
           source .venv/bin/activate
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          PYTEST_BASETEMP="${RUNNER_TEMP}/pytest-${GITHUB_JOB}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
           PYTEST="python -m pytest tests/st/a5/tensormap_and_ringbuffer/dfx/l2_swimlane/ \
-            --platform a5 --device ${DEVICE_RANGE} -p no:xdist --pto-session-timeout 600 \
+            --platform a5 --device \$TASK_DEVICE --basetemp $PYTEST_BASETEMP \
+            -p no:xdist --pto-session-timeout 600 \
             --require-pto-isa --enable-l2-swimlane --enable-dep-gen"
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "$PYTEST"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" --run "$PYTEST"
 
       - name: PMU smoke
         run: |
           source .venv/bin/activate
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          PYTEST_BASETEMP="${RUNNER_TEMP}/pytest-${GITHUB_JOB}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
           PYTEST="python -m pytest tests/st/a5/tensormap_and_ringbuffer/dfx/pmu/test_pmu.py \
-            --platform a5 --device ${DEVICE_RANGE} -p no:xdist --pto-session-timeout 600 \
+            --platform a5 --device \$TASK_DEVICE --basetemp $PYTEST_BASETEMP \
+            -p no:xdist --pto-session-timeout 600 \
             --require-pto-isa --enable-pmu 2"
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "$PYTEST"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" --run "$PYTEST"
 
       - name: args_dump smoke
         run: |
           source .venv/bin/activate
-          DEVICE_LIST=$(python -c "p='${DEVICE_RANGE}'.split('-'); s,e=p[0],p[-1]; print(','.join(str(i) for i in range(int(s),int(e)+1)))")
+          DEVICE_NUM=$(python -c "p='${DEVICE_RANGE}'.split('-'); print(int(p[-1])-int(p[0])+1)")
+          PYTEST_BASETEMP="${RUNNER_TEMP}/pytest-${GITHUB_JOB}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
           PYTEST="python -m pytest tests/st/a5/tensormap_and_ringbuffer/dfx/args_dump/test_args_dump.py \
-            --platform a5 --device ${DEVICE_RANGE} -p no:xdist --pto-session-timeout 600 \
+            --platform a5 --device \$TASK_DEVICE --basetemp $PYTEST_BASETEMP \
+            -p no:xdist --pto-session-timeout 600 \
             --require-pto-isa --dump-args"
-          task-submit --timeout 1800 --max-time 1800 --device "$DEVICE_LIST" --run "$PYTEST"
+          task-submit --timeout 1800 --max-time 1800 --device auto --device-num "$DEVICE_NUM" --run "$PYTEST"

README.md

@@ -73,6 +73,7 @@ export ASCEND_HOME_PATH=/usr/local/Ascend/ascend-toolkit/latest
 | [Orchestrator](docs/orchestrator.md) | DAG submission internals: submit flow, TensorMap, Scope, Ring, task state machine |
 | [Scheduler](docs/scheduler.md) | DAG dispatch internals: wiring/ready/completion queues, dispatch loop |
 | [Worker Manager](docs/worker-manager.md) | Worker pool, WorkerThread, THREAD/PROCESS modes, fork + mailbox mechanics |
+| [Streaming Request Sessions](docs/request-session.md) | Bounded multi-flight runs, per-request token streams, and two-slot HostGraph execution |
 | [Getting Started](docs/getting-started.md) | Setup, prerequisites, build process, configuration |
 | [Developer Guide](docs/developer-guide.md) | Directory structure, role ownership, conventions |
 | [Testing Guide](docs/testing.md) | CI pipeline, test types, writing new tests |

docs/dynamic-linking.md

@@ -179,6 +179,7 @@ uses `pthread_key_t` (POSIX TLS) for per-thread state in framework SOs.
 | `s_orch_thread_idx` | `__thread int` | AICPU SO | Profiling thread index (profiling off by default) |
 | `g_platform_phase_base` | plain global + `extern "C"` setter | AICPU SO | Device-phase buffer base; published by host (onboard kernel.cpp / sim dlsym), read by the `[STRACE]` phase stamps. Per-thread slotting via the affinity pthread-key index, not TLS. |
 | strace `inv`/`depth`/`hid` | `pthread_key_t` (`ThreadState`) | host runtime SO | Per-thread `[STRACE]` host-trace state (was C++ `thread_local`, converted per this rule). |
+| active arena bank | `pthread_key_t` | onboard host platform SO | Per-thread Host O arena selection for overlapping requests. |
 | `execution_context` | `thread_local` | Kernel SO (PTO ISA) | Per-thread execution context (fallback, cached values only) |
 | `NPUMemoryModel::instance` | `thread_local` | Kernel SO (PTO ISA) | Per-core memory model simulation |
 

docs/hierarchical_level_runtime.md

@@ -310,3 +310,6 @@ device allocation algorithm.
 | `src/common/worker/chip_worker.{h,cpp}` | L2 `ChipWorker` (kernel-running leaf, runs in the forked chip child) |
 | `python/bindings/` | nanobind exposure of C++ engine to Python |
 | `python/simpler/worker.py` | Python `Worker` factory + lifecycle wrapper |
+
+For bounded multi-flight request execution and per-request token streams on one
+L3 Worker, see [Streaming Request Sessions](request-session.md).

docs/request-session.md

@@ -0,0 +1,130 @@
+# Streaming Request Sessions
+
+`RequestSession` lets an initialized hierarchical `Worker` keep multiple
+`Worker.run` calls in flight. It also provides a per-request output stream,
+bounded submission, cancellation, and failure isolation.
+
+`max_active_runs` controls the number of dispatcher threads. Each dispatcher
+enters a real `Worker.run` call with its own outer scope. Submission mutates the
+shared orchestrator under one lock, while callbacks may wait for token messages
+independently. Onboard a2a3 HostGraph chip workers expose two mailbox credits,
+so two session dispatchers can execute complete runs concurrently. Other
+endpoint types remain single-credit.
+
+## PR1379 Multi-Flight Architecture
+
+The bounded multi-flight mailbox has these invariants:
+
+- one resident child message thread owns mailbox state transitions;
+- two long-lived run threads execute complete HostGraph runs, each through an
+  isolated `ChipWorker`/`DeviceRunner`;
+- each run thread uses bank 0 of its private runner-owned arena;
+- each run thread owns its Runtime storage and mutable Host run context;
+- Host O may overlap an earlier Device S, while a process-level device gate
+  serializes Device S because the resident AICPU runtime owns process-global
+  executor, affinity, register, and diagnostic state;
+- task completion is correlated by slot generation, so completion order may
+  differ from submission order;
+- control and shutdown operations are exclusive and wait for active task slots
+  to drain;
+- cancellation never interrupts active device work, but it suppresses output
+  and releases the slot after completion;
+- onboard HostGraph endpoints advertise two credits; TRB, simulation, SUB, and
+  remote endpoints initially remain single-credit.
+
+The target request path is:
+
+```text
+parent scheduler -> task slot 0 -> run thread 0 -> runner 0 arena -> completion
+                 -> task slot 1 -> run thread 1 -> runner 1 arena -> completion
+```
+
+The parent-side endpoint publishes a ready slot without waiting for Device S.
+A completion pump reports the matching scheduler completion only after the
+child marks that slot done. This is the non-blocking boundary; changing only
+`submit_next_level` would leave the current synchronous endpoint bottleneck in
+place.
+
+HostApi thread context, scheduler completion latches, the HostGraph epoch/token
+pipeline, runner-owned arenas, and concurrent top-level `Worker.run` batching
+remain.
+There is no cross-request admission lane or split prepare/execute ABI: every
+mailbox task is a self-contained run.
+
+## API
+
+Open one session on an initialized L3-or-higher Worker:
+
+```python
+def request_orch(orch, request, request_id, emitter, config):
+    token = build_and_submit(orch, request, request_id, config)
+    emitter.emit(token, final=token.is_final)
+
+with worker.open_request_session(
+    request_orch, max_pending=8, max_active_runs=2
+) as session:
+    stream_a = session.submit(request_a, config=config, request_id=100)
+    stream_b = session.submit(request_b, config=config, request_id=101)
+
+    forward_concurrently(stream_a, stream_b)
+```
+
+`submit` returns immediately after enqueueing. Request IDs are unsigned 64-bit
+integers and must be unique among live requests in the session. The session
+rejects a full pending queue with `RequestBackpressureError`; `max_pending`
+bounds queued work and does not count requests already owned by dispatchers.
+
+`RequestStream.next(timeout)` returns one item, raises `StopIteration` after a
+successful terminal item, and re-raises the request's terminal exception on
+failure. `RequestEmitter.emit()` does not return until `next()` has received
+that item. This one-item handoff prevents L3 from accumulating several tokens
+before returning them to the user. Consumers must drain the stream;
+`wait(timeout)` alone cannot complete while a token is waiting for delivery.
+A callback exception terminates its request stream, then the dispatcher
+continues with later queued requests.
+
+Only one request session may own a Worker. Its dispatcher threads are the only
+threads allowed to call `Worker.run` until the session closes. Closing a
+session stops new admission, drains every dispatcher, and releases ownership.
+
+## Cancellation
+
+`RequestStream.cancel()` is a soft cancellation:
+
+- a queued request is skipped before `Worker.run`;
+- active device work is drained to preserve arena and scope lifetimes;
+- later output items are suppressed;
+- consumers observe `RequestCancelledError`.
+
+Cancellation does not interrupt an AICPU or AICore operation. Reusing its
+buffers before Device S finishes would allow the next Host O to overwrite live
+state, so active cancellation deliberately favors memory safety over immediate
+device preemption.
+
+Each run thread has a dedicated `ChipWorker`, `DeviceRunner`, runtime buffer,
+and device arena. Both select bank 0 because HostGraph async threads do not
+inherit the caller's thread-local bank selection. Host O can overlap an
+earlier Device S without sharing mutable Host runtime state. Device S remains
+bounded to one active launch per child process. Control operations are still
+exclusive: create token queues and other L3-L2 resources before publishing the
+first task when later requests will reuse them.
+
+The HostGraph token trailer carries `request_id`, sequence, token value, final
+state, and status. L3 decodes it into `HostGraphToken`. A depth-one L2-to-L3
+token queue plus the acknowledged `emit()` handoff forms this flow-control
+chain:
+
+```text
+L2 publish → L3 read → user next() → L3 release → L2 may publish next token
+```
+
+## Implementation
+
+| Path | Role |
+| ---- | ---- |
+| `python/simpler/request_session.py` | Session and protocol |
+| `python/simpler/worker.py` | Worker ownership and two-slot child execution |
+| `src/common/hierarchical/scope.cpp` | Per-run thread-local scope frames |
+| `src/common/hierarchical/worker_manager.cpp` | Credit and mailbox-slot protocol |
+| `src/common/worker/chip_worker.cpp` | Per-run runtime ownership and arena selection |
+| `src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp` | HostGraph epoch/token pipeline |