Feat: add bounded whole-run FIFO admission

docs/task-flow.md

@@ -64,8 +64,9 @@ to C++:
 | `w3.submit_sub(handle, …)` dispatched to a SUB child | `LOCAL_PYTHON` | child resolves digest to a Python callable and calls `fn(args)` |
 
 All three paths share one mailbox wire format: `MAILBOX_OFF_CALLABLE` is
-reserved, and the 32-byte digest prefixes the args blob. The receiving child
-does the digest-to-slot resolve in its own address space.
+reserved, the run's generation-safe pipeline lease follows `CallConfig`, and
+the 32-byte digest prefixes the args blob. The receiving child does the
+digest-to-slot resolve in its own address space.
 
 The proposed remote L3 path keeps the same callable identity contract, but
 sends it in a versioned TASK frame. The remote endpoint resolves the digest
@@ -292,6 +293,39 @@ own AICore stream and retires it on every exit path, and no record of which
 image a stream last ran is load-bearing. The AICPU stream carries no such state
 and stays with its slot.
 
+#### Whole-run FIFO admission
+
+L3 graph callbacks remain synchronous and serialized, but native admission
+allows two live run reservations:
+
+```text
+run N:     EXECUTING
+run N+1:   PREPARED
+run N+2:   blocked in begin_run before its graph callback
+```
+
+`begin_run` acquires a generation-safe lease before invoking the callback.
+The FIFO head may enter `EXECUTING` while its callback is still building, which
+preserves orchestration callbacks that submit device work and wait for L2
+communication before returning. A non-head run remains `BUILDING` or becomes
+`PREPARED` when graph construction closes; it cannot execute until every prior
+run is terminal. The scheduler observes only the ready-queue partition belonging
+to that single active FIFO head. A run's device effects therefore cannot
+interleave with another run even when both graphs contain ready tasks. TensorMap
+keys remain `(run_id, tensor_key)`, so adjacent runs may reuse the same tensor
+address without creating cross-run dependencies.
+
+The terminal transition releases the reservation and lease exactly once,
+wakes a blocked third submission, and activates the next prepared run. Empty
+runs take the same transition immediately. If graph construction fails, every
+unstarted slot is poisoned and consumed, its ready-queue partition is erased,
+and the lease is returned without dispatching device work.
+
+Each direct chip child publishes its runtime contract's `pipeline_depth` in
+the startup mailbox before `INIT_READY`. The parent configures admission to the
+minimum published depth. Backends without a depth-two contract therefore keep
+depth-one serial behavior instead of receiving an invalid slot-1 lease.
+
 This is dormant capacity at this layer. The ordinary synchronous entry point
 continues to use slot 0, and the chip child's mailbox loop passes no lease, so
 every production run is unleased. This contract does not enable a second
@@ -382,6 +416,9 @@ Where the data goes after submit:
    Tags are consumed during the same submit call for dep inference and
    **never carried further**.
 3. `CallConfig` — copied into `slot.config` (parent heap, POD)
+4. `PipelineSlotLease` — copied from the owning run into
+   `slot.pipeline_lease`; local chip mailboxes forward `{slot_id, generation}`
+   to `ChipWorker::run_with_lease`.
 
 For the full submit mechanics (ring alloc, TensorMap lookup/insert, scope ref,
 fanout wiring), see [orchestrator.md](orchestrator.md).
@@ -390,7 +427,8 @@ fanout wiring), see [orchestrator.md](orchestrator.md).
 
 For local endpoints, after the Scheduler resolves the submitted NEXT_LEVEL
 target (or chooses an idle SUB worker), `LocalMailboxEndpoint` encodes
-`(callable digest, CallConfig, TaskArgs)` into the per-worker shm mailbox and
+`(callable digest, CallConfig, PipelineSlotLease, TaskArgs)` into the
+per-worker shm mailbox and
 the forked child decodes it. Remote NEXT_LEVEL dispatch through
 `RemoteL3Endpoint` serializes the same logical payload into a framed TASK
 request instead.

python/bindings/worker_bind.h

@@ -350,7 +350,7 @@ inline void bind_worker(nb::module_ &m) {
         .def("scope_end", &Orchestrator::scope_end, "Close the innermost scope. Non-blocking.")
         .def("_scope_begin", &Orchestrator::scope_begin)
         .def("_scope_end", &Orchestrator::scope_end)
-        .def("_begin_run", &Orchestrator::begin_run)
+        .def("_begin_run", &Orchestrator::begin_run, nb::call_guard<nb::gil_scoped_release>())
         .def("_close_run_submission", &Orchestrator::close_run_submission, nb::arg("run_id"))
         .def(
             "_fail_run_submission",
@@ -410,6 +410,10 @@ inline void bind_worker(nb::module_ &m) {
             nb::arg("worker_id"), nb::arg("mailbox_ptr"), nb::arg("child_pid") = -1,
             "Add a NEXT_LEVEL sub-worker with an explicit worker id."
         )
+        .def(
+            "configure_pipeline_depth", &Worker::configure_pipeline_depth, nb::arg("depth"),
+            "Set run admission depth from the minimum direct-chip runtime capability before init."
+        )
         .def(
             "add_sub_worker",
             [](Worker &self, uint64_t mailbox_ptr, int child_pid) {
@@ -766,6 +770,7 @@ inline void bind_worker(nb::module_ &m) {
     m.attr("MAILBOX_SIZE") = static_cast<int>(MAILBOX_SIZE);
     m.attr("MAILBOX_OFF_ERROR_MSG") = static_cast<int>(MAILBOX_OFF_ERROR_MSG);
     m.attr("MAILBOX_ERROR_MSG_SIZE") = static_cast<int>(MAILBOX_ERROR_MSG_SIZE);
+    m.attr("PTO_PIPELINE_MAX_DEPTH") = static_cast<uint32_t>(PTO_PIPELINE_MAX_DEPTH);
     m.attr("MAX_RING_DEPTH") = static_cast<int32_t>(MAX_RING_DEPTH);
     m.attr("MAX_SCOPE_DEPTH") = static_cast<int32_t>(MAX_SCOPE_DEPTH);
 

python/simpler/worker.py

@@ -82,6 +82,7 @@ def my_l4_orch(orch, args, config):
 import cloudpickle
 from _task_interface import (  # pyright: ignore[reportMissingImports]
     MAX_REGISTERED_CALLABLE_IDS,
+    PTO_PIPELINE_MAX_DEPTH,
     RUNTIME_ENV_RING_COUNT,
     TENSOR_CHILD_MEMORY_OFFSET,
     WorkerType,
@@ -176,10 +177,13 @@ def my_l4_orch(orch, args, config):
 # travels separately via ChipWorker.init(log_level, log_info_v) — not on per-task wire.
 _RUNTIME_ENV_UINT64_FIELD_COUNT = 3 * RUNTIME_ENV_RING_COUNT
 _CFG_FMT = struct.Struct("=iiiiii" + ("Q" * _RUNTIME_ENV_UINT64_FIELD_COUNT) + "1024s")
-# Args region starts after CONFIG, rounded up to 8 bytes so the first
+# The generation-safe pipeline lease follows CONFIG. Args start after the
+# lease, rounded up to 8 bytes so the first
 # Tensor.data (uint64_t at OFF_ARGS+8) is 8-byte aligned, avoiding
 # SIGBUS on strict-alignment platforms (aarch64 atomics, some ARM cores).
-_OFF_ARGS = (_OFF_CONFIG + _CFG_FMT.size + 7) & ~7
+_PIPELINE_LEASE_FMT = struct.Struct("=IIQ")
+_OFF_PIPELINE_LEASE = (_OFF_CONFIG + _CFG_FMT.size + 7) & ~7
+_OFF_ARGS = (_OFF_PIPELINE_LEASE + _PIPELINE_LEASE_FMT.size + 7) & ~7
 assert _OFF_ARGS % 8 == 0, "_OFF_ARGS must be 8-aligned for Tensor.data"
 _OFF_TASK_CALLABLE_HASH = _OFF_ARGS
 _OFF_TASK_ARGS_BLOB = _OFF_TASK_CALLABLE_HASH + CALLABLE_HASH_DIGEST_BYTES
@@ -1558,9 +1562,14 @@ def handle_task() -> tuple[int, str]:
             # No-op when nothing is registered.
             if host_buf_ranges:
                 _rewrite_blob_host_addrs(buf, _OFF_TASK_ARGS_BLOB, host_buf_ranges)
+            pipeline_slot, pipeline_reserved, pipeline_generation = _PIPELINE_LEASE_FMT.unpack_from(
+                buf, _OFF_PIPELINE_LEASE
+            )
+            if pipeline_reserved != 0:
+                raise RuntimeError(f"chip_process dev={device_id}: invalid pipeline lease reserved field")
             # Hand the mailbox bytes straight to C++ (zero-copy zero-decode):
             # the blob layout is what `write_blob` already wrote, so re-parsing
-            # it in Python is N×40B of avoidable work and a permanent
+            # it in Python is N x 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(
@@ -1570,6 +1579,8 @@ def handle_task() -> tuple[int, str]:
                 cfg,
                 mailbox_addr + _OFF_ACCEPTED,
                 _TASK_ACCEPTED,
+                pipeline_slot,
+                pipeline_generation,
             )
         except Exception as e:  # noqa: BLE001
             code = 1
@@ -1769,6 +1780,7 @@ def _chip_process_loop(  # noqa: PLR0913 -- fork-child entry: all context (bins,
     # child to reach _INIT_READY before dispatching the first task, so the
     # per-rank host-side stream sync budget only covers actual op execution
     # rather than absorbing peer-rank init skew.
+    _PIPELINE_LEASE_FMT.pack_into(buf, _OFF_PIPELINE_LEASE, int(cw.pipeline_depth), 0, 0)
     _mailbox_store_i32(state_addr, _INIT_READY)
     sys.stderr.write(f"[chip_process pid={os.getpid()} dev={device_id}] ready\n")
     sys.stderr.flush()
@@ -4491,6 +4503,7 @@ def _start_hierarchical(self) -> None:  # noqa: PLR0912 -- three parallel fork l
         device_ids = self._config.get("device_ids", [])
         n_sub = self._config.get("num_sub_workers", 0)
         deadline = self._startup_deadline
+        direct_chip_pipeline_depth = PTO_PIPELINE_MAX_DEPTH
 
         # Freeze the startup registry snapshot. init() already holds the epoch in
         # the INITIALIZING state, so a concurrent register/unregister is blocked
@@ -4597,6 +4610,14 @@ def _setup():
             # documented in issue #897.  A chip that fails or dies during init
             # raises here rather than spinning forever.
             self._await_children_ready(self._chip_shms, self._chip_pids, "chip", deadline)
+            chip_depths = []
+            for shm in self._chip_shms:
+                buf = shm.buf
+                assert buf is not None
+                chip_depths.append(_PIPELINE_LEASE_FMT.unpack_from(buf, _OFF_PIPELINE_LEASE)[0])
+            if any(depth <= 0 or depth > PTO_PIPELINE_MAX_DEPTH for depth in chip_depths):
+                raise RuntimeError(f"chip worker published invalid pipeline depths: {chip_depths}")
+            direct_chip_pipeline_depth = min(chip_depths)
 
         # Fork next-level Worker children (L4+ with Worker children).
         # Each child process eagerly inits the inner Worker, which forks its own
@@ -4660,6 +4681,7 @@ def _setup(inner=inner_worker):
         # the unified mailbox.
         dw = self._worker
         assert dw is not None
+        dw.configure_pipeline_depth(direct_chip_pipeline_depth)
 
         # Register chip workers as NEXT_LEVEL (L3). The child pid lets the C++
         # endpoint fail a dispatch whose child died instead of spinning on a
@@ -6150,9 +6172,9 @@ def submit(self, callable, args=None, config=None) -> RunHandle:
         ``args``  : TaskArgs (optional)
         ``config``: CallConfig (optional, default-constructed if None)
 
-        Graph construction remains serialized. A later submission waits until
-        prior dispatches are accepted; completion and cleanup stay attached to
-        each returned handle.
+        Graph construction remains serialized. Native admission permits one
+        active and one prepared run; a third submission blocks before invoking
+        its graph callback. Completion and cleanup stay attached to each handle.
         """
         with self._operation_lease("submit"):
             return self._submit_locked(callable, args, config)
@@ -6179,12 +6201,8 @@ def _submit_locked(self, callable, args, config) -> RunHandle:
             return RunHandle._completed(self)
 
         with self._submit_mu:
-            # 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_acceptance()
+            # Graph callbacks stay serialized. The native begin_run admission
+            # boundary owns depth-two backpressure and slot generations.
             return self._submit_l3_locked(callable, args, cfg)
 
     def _submit_l3_locked(self, callable, args, cfg: CallConfig) -> RunHandle: