Feat: add launch-acceptance flight fence

docs/orchestrator.md

@@ -55,6 +55,8 @@ public:
     RunId begin_run();
     void close_run_submission(RunId run_id);
     void fail_run_submission(RunId run_id, std::exception_ptr error);
+    void wait_run_accepted(RunId run_id);
+    bool run_accepted(RunId run_id) const;
     void wait_run(RunId run_id);
     bool wait_run_for(RunId run_id, double timeout_seconds);
     bool run_done(RunId run_id) const;
@@ -96,9 +98,12 @@ worker.submit(orchestration, args, config).wait()
 ```
 
 The current L2 backend is synchronous, so L2 `submit()` executes the existing
-blocking path and returns an already-completed handle. L3 asynchronous return
-does not yet imply overlapping runs: a later submit waits for the prior run's
-fence and cleanup before building the next DAG.
+blocking path and returns an already-completed handle. At L3 and above, graph
+callbacks remain serialized, but a later submit waits only for every dispatch
+in the prior run to cross its endpoint acceptance boundary. On A2A3 onboard,
+that boundary is after both device kernels are enqueued and before stream
+synchronization. Endpoints without an earlier signal fall back to completion.
+Each run still owns its completion error, keepalives, and cleanup independently.
 
 Remote L3 submit adds two hidden pieces of metadata: final eligible worker-id
 sets and optional `RemoteTaskArgsSidecar` entries aligned by tensor index.

docs/task-flow.md

@@ -348,6 +348,15 @@ the forked child decodes it. Remote NEXT_LEVEL dispatch through
 `RemoteL3Endpoint` serializes the same logical payload into a framed TASK
 request instead.
 
+Every dispatched group member contributes one run-acceptance obligation. For
+an A2A3 onboard chip endpoint, the child-side native runner writes
+`TASK_ACCEPTED` after its AICore and AICPU kernels are both enqueued; the parent
+observes it without releasing the mailbox. Other endpoint paths satisfy the
+same obligation conservatively when their completion returns. Once submission
+is closed and all obligations are satisfied, the next serialized orchestration
+callback may build its DAG even though the prior run has not reached its
+completion fence.
+
 Local mailbox path:
 
 ```text

docs/worker-manager.md

@@ -46,7 +46,8 @@ public:
     void add_sub       (void *mailbox);
 
     // Lifecycle
-    void start(Ring *ring, OnCompleteFn on_complete);   // starts all WorkerThreads
+    void start(Ring *ring, OnCompleteFn on_complete,
+               OnAcceptFn on_accept);              // starts all WorkerThreads
     void stop();
 
     // Scheduler API
@@ -100,8 +101,9 @@ struct WorkerDispatch {
 
 class WorkerThread {
 public:
-    void start(Ring *ring, WorkerManager *manager,
+    void start(Ring *ring,
                const std::function<void(WorkerCompletion)> &on_complete,
+               const std::function<void(WorkerDispatch)> &on_accept,
                std::unique_ptr<WorkerEndpoint> endpoint);
     void stop();
     void dispatch(WorkerDispatch d);       // slot id + group sub-index
@@ -118,13 +120,14 @@ private:
     std::condition_variable cv_;
 
     void loop();
-    WorkerCompletion dispatch_process(WorkerDispatch d);
+    WorkerCompletion dispatch_process(WorkerDispatch d,
+                                      const std::function<void()> &on_accept);
 };
 ```
 
 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
+`endpoint->run_with_accept(...)` once per dispatch. `LocalMailboxEndpoint`
+drives the shm handshake — one mailbox 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.
@@ -152,7 +155,11 @@ and the child polls the mailbox for the lifetime of the worker.
 ### 3.1 Parent-side dispatch
 
 ```cpp
-WorkerCompletion LocalMailboxEndpoint::run(Ring *ring, WorkerDispatch d) {
+// `run(ring, d)` forwards here with an empty hook; the acceptance-aware
+// overload is what WorkerThread calls.
+WorkerCompletion LocalMailboxEndpoint::run_with_accept(
+    Ring *ring, const WorkerDispatch &d, const std::function<void()> &on_accept
+) {
     TaskSlotState &s = *ring->slot_state(d.task_slot);
     char *m = static_cast<char *>(mailbox_);
 
@@ -175,7 +182,18 @@ WorkerCompletion LocalMailboxEndpoint::run(Ring *ring, WorkerDispatch d) {
     // kChildLivenessPollPeriod steady-clock period rather than an iteration
     // count, which no longer maps to a bounded time at spin speed.
     auto next_liveness_check = steady_clock::now() + kChildLivenessPollPeriod;
-    while (read_state(mailbox_) != MailboxState::TASK_DONE) {
+    bool acceptance_observed = false;
+    while (true) {
+        MailboxState state = read_state(mailbox_);
+        // Launch acceptance is a sticky word, not a state: the child sets it
+        // before TASK_DONE and nothing clears it until the next TASK_READY, so
+        // a task that finishes between two polls cannot lose the ACK. Read
+        // after the state so a TASK_DONE observation implies it is visible.
+        if (!acceptance_observed && read_accepted(mailbox_)) {
+            acceptance_observed = true;
+            if (on_accept) on_accept();
+        }
+        if (state == MailboxState::TASK_DONE) break;
         auto now = steady_clock::now();
         if (now >= next_liveness_check) {
             next_liveness_check = now + kChildLivenessPollPeriod;
@@ -206,6 +224,8 @@ Parent-side cost per dispatch:
   ([codestyle](../.claude/rules/codestyle.md) rule 5). Child liveness is sampled
   on a steady-clock period, so a dead child ends the wait with
   `ENDPOINT_FAILURE` instead of spinning the parent forever
+- A sticky launch-acceptance word, observed before completion and cleared only
+  when the parent publishes the next `TASK_READY`
 - One explicit completion outcome: success, task failure, or endpoint failure
 
 Total ~nanoseconds overhead; the wait is dominated by actual kernel execution.
@@ -216,7 +236,10 @@ The child loop lives in Python — see `_chip_process_loop` and
 `_sub_worker_loop` in `python/simpler/worker.py`. Each child polls
 `MAILBOX_OFF_STATE`, decodes the digest-prefixed args blob on `TASK_READY`,
 resolves the digest to its private local slot/callable, writes back any error,
-and publishes `TASK_DONE`.
+and publishes `TASK_DONE`. An A2A3 onboard chip child also exposes its mailbox
+state word to the native runner, which publishes `TASK_ACCEPTED` after both
+kernel launches succeed. SUB, remote, A5, and failure paths use WorkerThread's
+completion-time acceptance fallback.
 The child inherits the parent's full address space at fork time, so:
 
 - ChipCallable objects (pre-fork allocated) are COW-visible at the same VA

python/bindings/task_interface.cpp

@@ -1433,17 +1433,20 @@ NB_MODULE(_task_interface, m) {
         .def(
             "run_from_blob",
             [](ChipWorker &self, int32_t callable_id, uint64_t args_blob_ptr, size_t blob_capacity,
-               const CallConfig &config) {
+               const CallConfig &config, uint64_t accepted_state_addr, int32_t accepted_value) {
                 // The mailbox region is the on-wire format `write_blob` produced;
                 // `read_blob` is the matching reader that returns a zero-copy
                 // TaskArgsView into the caller-owned bytes. Forwards to the
                 // existing `run(cid, view, config)` path so chip-child
                 // loops never re-implement the tensor/scalar layout in Python
                 // (where it has historically dropped fields like child_memory).
                 TaskArgsView view = read_blob(reinterpret_cast<const uint8_t *>(args_blob_ptr), blob_capacity);
-                self.run(callable_id, view, config);
+                self.run(
+                    callable_id, view, config, reinterpret_cast<volatile int32_t *>(accepted_state_addr), accepted_value
+                );
             },
             nb::arg("callable_id"), nb::arg("args_blob_ptr"), nb::arg("blob_capacity"), nb::arg("config"),
+            nb::arg("accepted_state_addr") = 0, nb::arg("accepted_value") = 0,
             "Launch a callable_id from a raw mailbox-blob pointer + capacity "
             "(used by chip-child mailbox loops to avoid Python-side re-deserialisation "
             "of the per-task tensor/scalar layout). The blob must be in the format "

python/bindings/worker_bind.h

@@ -367,6 +367,12 @@ inline void bind_worker(nb::module_ &m) {
             "_wait_run", &Orchestrator::wait_run, nb::arg("run_id"), nb::call_guard<nb::gil_scoped_release>(),
             "Block until one run is terminal and raise only that run's error."
         )
+        .def(
+            "_wait_run_accepted", &Orchestrator::wait_run_accepted, nb::arg("run_id"),
+            nb::call_guard<nb::gil_scoped_release>(),
+            "Block until every dispatch in one closed run has crossed its endpoint acceptance boundary."
+        )
+        .def("_run_accepted", &Orchestrator::run_accepted, nb::arg("run_id"))
         .def(
             "_wait_run_for", &Orchestrator::wait_run_for, nb::arg("run_id"), nb::arg("timeout_seconds"),
             nb::call_guard<nb::gil_scoped_release>(), "Wait up to timeout_seconds for one run to become terminal."

python/simpler/orchestrator.py

@@ -566,6 +566,12 @@ def _fail_run_submission(self, run_id: int, message: str = "") -> None:
     def _wait_run(self, run_id: int) -> None:
         self._o._wait_run(run_id)
 
+    def _wait_run_accepted(self, run_id: int) -> None:
+        self._o._wait_run_accepted(run_id)
+
+    def _run_accepted(self, run_id: int) -> bool:
+        return bool(self._o._run_accepted(run_id))
+
     def _wait_run_for(self, run_id: int, timeout_seconds: float) -> bool:
         return bool(self._o._wait_run_for(run_id, timeout_seconds))
 

python/simpler/task_interface.py

@@ -1264,7 +1264,7 @@ def host_dlopen_count(self):
 
     @property
     def run_stream_set_create_count(self):
-        """Number of run stream sets the bound runner has created."""
+        """Number of run stream generations the bound runner has created."""
         return self._impl.run_stream_set_create_count
 
     def malloc(self, size):

python/simpler/worker.py

@@ -186,7 +186,13 @@ def my_l4_orch(orch, args, config):
 # MAILBOX_ARGS_CAPACITY mirrors the C++ constexpr in worker_manager.h so the
 # Python reader can bounds-check incoming args blobs. Source-of-truth for the
 # constants on the right is the nanobind binding (cannot drift).
-_MAILBOX_ARGS_CAPACITY = MAILBOX_SIZE - _OFF_TASK_ARGS_BLOB - MAILBOX_ERROR_MSG_SIZE
+# Mirrors MAILBOX_OFF_ACCEPTED / MAILBOX_TASK_ACCEPTED: launch acceptance is a
+# sticky word rather than a MailboxState, because a state carrying it is lost
+# whenever the child reaches TASK_DONE between two parent polls. The parent
+# clears it when it publishes the next TASK_READY.
+_OFF_ACCEPTED = MAILBOX_SIZE - MAILBOX_ERROR_MSG_SIZE - 8
+_TASK_ACCEPTED = 1
+_MAILBOX_ARGS_CAPACITY = MAILBOX_SIZE - _OFF_TASK_ARGS_BLOB - MAILBOX_ERROR_MSG_SIZE - 8
 _OFF_CONTROL_CALLABLE_HASH = _OFF_ARGS + 32
 # MAILBOX_OFF_ERROR_MSG / MAILBOX_ERROR_MSG_SIZE come from the C++
 # nanobind module so the two sides cannot drift.
@@ -1534,7 +1540,14 @@ def handle_task() -> tuple[int, str]:
             # it in Python is N×40B of avoidable work and a permanent
             # opportunity to drop a field.  C++ reinterpret_cast<ChipStorageTaskArgs*>
             # is the source of truth.
-            cw._impl.run_from_blob(cid, mailbox_addr + _OFF_TASK_ARGS_BLOB, _MAILBOX_ARGS_CAPACITY, cfg)
+            cw._impl.run_from_blob(
+                cid,
+                mailbox_addr + _OFF_TASK_ARGS_BLOB,
+                _MAILBOX_ARGS_CAPACITY,
+                cfg,
+                mailbox_addr + _OFF_ACCEPTED,
+                _TASK_ACCEPTED,
+            )
         except Exception as e:  # noqa: BLE001
             code = 1
             msg = _format_exc(f"chip_process dev={device_id}", e)
@@ -1967,6 +1980,8 @@ def __init__(
         self._resources = resources if resources is not None else _RunResources()
         self._cv = threading.Condition()
         self._wait_in_progress = False
+        self._accept_wait_in_progress = False
+        self._launch_accepted = False
         self._terminal = False
         self._error: BaseException | None = None
 
@@ -1979,6 +1994,8 @@ def _completed(cls, worker: Worker) -> RunHandle:
         handle._resources = _RunResources()
         handle._cv = threading.Condition()
         handle._wait_in_progress = False
+        handle._accept_wait_in_progress = False
+        handle._launch_accepted = True
         handle._terminal = True
         handle._error = None
         return handle
@@ -2064,6 +2081,20 @@ def wait(self, timeout: float | None = None) -> None:
                 self._cv.notify_all()
             raise TimeoutError("RunHandle.wait() timed out")
 
+        # An acceptance waiter that already captured this run id must leave the
+        # native wait before finalize releases that id. It is terminal now, so
+        # this is only a short ownership hand-off and does not serialize device
+        # execution with acceptance waiting.
+        try:
+            with self._cv:
+                while self._accept_wait_in_progress:
+                    self._cv.wait(timeout=_RUN_HANDLE_WAIT_RECHECK_S)
+        except BaseException:
+            self._wait_in_progress = False
+            with self._cv:
+                self._cv.notify_all()
+            raise
+
         # Cleanup runs exactly once, on this waiter, and its outcome IS the
         # handle's result: an interruption mid-finalize is cached like any other
         # error rather than lost, so the publication below is unconditional.
@@ -2074,8 +2105,10 @@ def wait(self, timeout: float | None = None) -> None:
         self._error = error
         self._run_id = None
         self._keepalive = None
+        self._launch_accepted = True
         self._terminal = True
         self._wait_in_progress = False
+        self._accept_wait_in_progress = False
         with self._cv:
             self._cv.notify_all()
         if error is not None:
@@ -2093,6 +2126,30 @@ def _wait_for_serialization(self) -> None:
             # The result remains cached for this handle's public wait/result.
             pass
 
+    def _wait_for_acceptance(self) -> None:
+        """Wait until this run's dispatches cross their acceptance boundary."""
+        with self._cv:
+            while not self._terminal and self._accept_wait_in_progress:
+                self._cv.wait(timeout=_RUN_HANDLE_WAIT_RECHECK_S)
+            if self._terminal or self._launch_accepted:
+                return
+            self._accept_wait_in_progress = True
+            run_id = self._run_id
+
+        assert run_id is not None
+        try:
+            self._worker._wait_run_handle_accepted(run_id)
+        except BaseException:
+            self._accept_wait_in_progress = False
+            with self._cv:
+                self._cv.notify_all()
+            raise
+
+        self._launch_accepted = True
+        self._accept_wait_in_progress = False
+        with self._cv:
+            self._cv.notify_all()
+
 
 def _forked_child_main(buf: memoryview, label: str, setup, serve, make_group_leader: bool = False) -> None:
     """Run a forked child to completion, always terminating via ``os._exit``.
@@ -5989,8 +6046,9 @@ def submit(self, callable, args=None, config=None) -> RunHandle:
         ``args``  : TaskArgs (optional)
         ``config``: CallConfig (optional, default-constructed if None)
 
-        Only one live device run is admitted: a later submission waits for the
-        previous handle's fence and cleanup before building its DAG.
+        Graph construction remains serialized. A later submission waits until
+        prior dispatches are accepted; completion and cleanup stay attached to
+        each returned handle.
         """
         with self._operation_lease("submit"):
             return self._submit_locked(callable, args, config)
@@ -6017,13 +6075,12 @@ def _submit_locked(self, callable, args, config) -> RunHandle:
             return RunHandle._completed(self)
 
         with self._submit_mu:
-            # Cleanup is Worker-global while only one live device run is
-            # admitted. Drain prior handles before a new callback can mutate
-            # those resources; errors remain attached only to their origin.
+            # Graph callbacks are serialized, but accepted runs may remain live:
+            # their Python resources are isolated in each RunHandle.
             with self._hierarchical_start_cv:
                 prior_handles = tuple(self._accepted_run_handles)
             for handle in prior_handles:
-                handle._wait_for_serialization()
+                handle._wait_for_acceptance()
             return self._submit_l3_locked(callable, args, cfg)
 
     def _submit_l3_locked(self, callable, args, cfg: CallConfig) -> RunHandle:
@@ -6080,6 +6137,10 @@ def _wait_run_handle(self, run_id: int, timeout: float | None) -> bool:
             return True
         return self._orch._wait_run_for(run_id, timeout)
 
+    def _wait_run_handle_accepted(self, run_id: int) -> None:
+        assert self._orch is not None
+        self._orch._wait_run_accepted(run_id)
+
     def _finalize_run_handle(
         self, handle: RunHandle, run_id: int, native_error: BaseException | None
     ) -> BaseException | None:
@@ -6155,12 +6216,12 @@ def host_dlopen_count(self) -> int:
 
     @property
     def run_stream_set_create_count(self) -> int:
-        """L2 only: number of run stream sets the bound runner has created.
+        """L2 only: number of run stream generations the runner has created.
 
-        A set belongs to a pipeline slot and is reused for every run on that
-        slot, so a worker that has served any number of runs reports 1.
-        Returns 0 on non-L2 workers and on platforms whose runs use the
-        persistent bootstrap stream pair (simulation, a5).
+        AICPU streams belong to pipeline slots. AICore streams are reused only
+        while the loaded AICore image is unchanged, so each code transition
+        advances this count. Returns 0 on non-L2 workers and on platforms whose
+        runs use the persistent bootstrap stream pair (simulation, a5).
         """
         if self.level != 2 or self._chip_worker is None:
             return 0