feat: support pipelined orchestration and scheduling across tokens and requests

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 |

docs/worker-manager.md

@@ -14,6 +14,12 @@ drives a shared-memory mailbox consumed by a forked Python child. The child
 runs the real worker (a `ChipWorker` for NEXT_LEVEL, a Python callable for
 SUB) in its own address space.
 
+`WorkerEndpointCaps::max_in_flight` defines the dispatch credits of one
+endpoint. `WorkerThread` owns that many executor threads and reports itself
+available while at least one credit remains. Onboard a2a3 HostGraph chip
+mailboxes advertise two credits; TRB, simulation, SUB, nested Worker, and
+remote endpoints advertise one.
+
 The remote L3 design keeps this local fork/shm path behind
 `LocalMailboxEndpoint` and reserves the same `WorkerEndpoint` boundary for a
 framed `RemoteL3Endpoint` for cross-host NEXT_LEVEL children. A remote endpoint
@@ -79,8 +85,8 @@ the public worker id from the local worker vector index.
 - **Directed NEXT_LEVEL lookup**: `get_worker_by_id` resolves the exact stable
   target selected by the user; the Scheduler never asks the manager to choose
   another NEXT_LEVEL worker
-- **SUB-only idle selection**: `pick_idle_sub_excluding` chooses an idle SUB
-  worker not already used by the same SUB group
+- **SUB-only credit selection**: `pick_idle_sub_excluding` chooses a SUB worker
+  with available endpoint credit that is not already used by the same SUB group
 
 Callable and remote-buffer eligibility is validated against the exact target
 during Orchestrator submission. It is not scheduling metadata and is not
@@ -112,19 +118,21 @@ public:
 private:
     Ring *ring_;                       // reads slot state via ring->slot_state(id)
     std::unique_ptr<WorkerEndpoint> endpoint_;
-    std::thread thread_;
+    std::vector<std::thread> threads_;
     std::queue<WorkerDispatch> queue_;
     std::mutex mu_;
     std::condition_variable cv_;
+    uint32_t capacity_;
+    std::atomic<uint32_t> available_;
 
     void loop();
     WorkerCompletion dispatch_process(WorkerDispatch d);
 };
 ```
 
-The WorkerThread's `std::thread` pumps the internal queue and calls
-`endpoint->run(...)` once per dispatch. `LocalMailboxEndpoint::run` drives the
-shm handshake — one mailbox round trip per dispatch. The forked child loop
+The WorkerThread executor threads pump one shared queue and call
+`endpoint->run(...)` once per reserved credit. `LocalMailboxEndpoint::run`
+drives one mailbox frame round trip per dispatch. The forked child loop
 that consumes the mailbox lives in Python (`_chip_process_loop` /
 `_sub_worker_loop` in `python/simpler/worker.py`); the parent does not fork
 children.
@@ -143,7 +151,13 @@ group sub-index.
 
 ## 3. Dispatch via shm mailbox
 
-Each `LocalMailboxEndpoint` drives a `MAILBOX_SIZE`-byte `MAP_SHARED` region.
+Each `LocalMailboxEndpoint` drives a `MAILBOX_SIZE`-byte `MAP_SHARED` region
+containing two fixed-size task frames. Slot generation identifies frame reuse;
+the protocol magic/version rejects mismatched parent and child layouts. An
+onboard a2a3 HostGraph child owns two long-lived task threads. Each thread has
+its own `ChipWorker`/`DeviceRunner`, Runtime storage, streams, kernel arguments,
+diagnostic collectors, and private device arena. This isolates Host run state
+instead of making `DeviceRunnerBase`'s mutable launch state concurrent.
 The Python facade forks one child per mailbox **before**
 `WorkerManager::start()` (so the parent has only the Python main thread when
 fork runs, avoiding the classical "fork in a multi-threaded process" hazard)
@@ -154,37 +168,42 @@ and the child polls the mailbox for the lifetime of the worker.
 ```cpp
 WorkerCompletion LocalMailboxEndpoint::run(Ring *ring, WorkerDispatch d) {
     TaskSlotState &s = *ring->slot_state(d.task_slot);
-    char *m = static_cast<char *>(mailbox_);
+    size_t slot = acquire_task_slot();
+    char *m = static_cast<char *>(mailbox_) + slot * MAILBOX_TASK_SLOT_SIZE;
 
-    // Write task data: reserved callable field, config, digest prefix, then
+    // Write task data: generation, protocol, config, digest prefix, then
     // length-prefixed TaskArgs blob. Tags are stripped; only
     // [digest][T][S][tensors][scalars] crosses the fork boundary.
-    uint64_t reserved = 0;
-    memcpy(m + MAILBOX_OFF_CALLABLE, &reserved, sizeof(reserved));
+    uint64_t generation = next_generation(slot);
+    memcpy(m + MAILBOX_OFF_GENERATION, &generation, sizeof(generation));
+    write_protocol(m, MAILBOX_PROTOCOL_MAGIC_VERSION);
     memcpy(m + MAILBOX_OFF_CONFIG, &s.config, sizeof(CallConfig));
     memcpy(m + MAILBOX_OFF_TASK_CALLABLE_HASH, s.callable.digest.data(), 32);
     const TaskArgs &args = s.is_group() ? s.task_args_list[d.group_index] : s.task_args;
     write_blob(m + MAILBOX_OFF_TASK_ARGS_BLOB, args);
 
     // Signal child
-    write_state(mailbox_, MailboxState::TASK_READY);
+    write_state(m, MailboxState::TASK_READY);
 
     // Poll for completion
-    while (read_state(mailbox_) != MailboxState::TASK_DONE)
+    while (read_state(m) != MailboxState::TASK_DONE)
         std::this_thread::sleep_for(std::chrono::microseconds(50));
 
-    int err = read_error(mailbox_);
-    write_state(mailbox_, MailboxState::IDLE);
+    check_generation(m, generation);
+    int err = read_error(m);
+    write_state(m, MailboxState::IDLE);
+    release_task_slot(slot);
     return err == 0
         ? WorkerCompletion{d.task_slot, d.group_index, EndpointOutcome::SUCCESS, ""}
         : WorkerCompletion{d.task_slot, d.group_index, EndpointOutcome::TASK_FAILURE,
-                           read_error_msg(mailbox_)};
+                           read_error_msg(m)};
 }
 ```
 
 Parent-side cost per dispatch:
 
-- One reserved `uint64`, one `CallConfig`, one 32-byte digest, and one
+- One generation `uint64`, one protocol word, one `CallConfig`, one 32-byte
+  digest, and one
   TaskArgs blob
 - One signal (`write_state`)
 - Poll loop with `sleep_for(50us)` (not busy-wait)
@@ -205,31 +224,38 @@ The child inherits the parent's full address space at fork time, so:
 - The Python callable registry is COW-visible
 - Tensor data in `torch.share_memory_()` regions is fully shared (MAP_SHARED)
 
+Control commands remain exclusive with task frames. Request-session resources
+that require control operations, such as L3-L2 message regions, must therefore
+be created before the first device task is published. The two-flight hardware
+test creates both token queues before request A submits its task; request B then
+reuses its existing queue without draining A.
+
 ### 3.3 Mailbox layout
 
 ```text
 offset 0:                         int32   state
 offset 4:                         int32   error
-offset 8:                         uint64  reserved task callable field
+offset 8:                         uint64  task generation
                                           or control sub-command
 offset 16:                        CallConfig config
 MAILBOX_OFF_TASK_CALLABLE_HASH:   uint8[32] callable digest
 MAILBOX_OFF_TASK_ARGS_BLOB:       bytes [int32 T][int32 S]
                                         [Tensor x T][uint64_t x S]
-tail:                             fixed-size NUL-terminated error message
+MAILBOX_OFF_PROTOCOL:             uint64  protocol magic/version
+frame tail:                       fixed-size NUL-terminated error message
 ```
 
-The current mailbox size is the C++ `MAILBOX_SIZE` constant exported through
-the nanobind module; Python derives its offsets from the same binding where
-possible so the two sides cannot drift silently.
+`MAILBOX_SIZE` is two `MAILBOX_TASK_SLOT_SIZE` frames. The constants are
+exported through the nanobind module; Python derives its protocol, offsets,
+and allocation size from that binding so the two sides cannot drift silently.
 
 ### 3.4 Shutdown
 
 `WorkerManager::shutdown_children()` writes `SHUTDOWN` to every registered
 endpoint; for `LocalMailboxEndpoint` this writes `SHUTDOWN` to the mailbox.
 Each child loop sees it on its next poll and exits. The Python facade owns the
 child PIDs and calls `waitpid()` after writing `SHUTDOWN` to its own mailbox
-copy. The parent's `WorkerThread::stop()` only joins the C++ dispatcher thread
+copy. The parent's `WorkerThread::stop()` only joins the C++ dispatcher threads
 — it does not own the child process.
 
 ---