Feat: add generation-safe depth-two pipeline slots

docs/task-flow.md

@@ -257,10 +257,57 @@ void ChipWorker::run(int32_t local_slot, TaskArgsView view, const CallConfig &co
 
 One memcpy of a few KB per task; negligible.
 
+#### Pipeline resource leases
+
+A2/A3 host runtimes declare `pipeline_depth = 2` as resource capacity. The
+contract determines the concrete copy count rather than permitting concurrent
+device execution by itself:
+
+| Resource class | Copies | Selection |
+| -------------- | -----: | --------- |
+| `HOST_PER_RUN` | `pipeline_depth` | lease `slot_id` |
+| `DEVICE_SCRATCH` | 1 | slot 0 |
+| `EXEC_HANDLE` | `pipeline_depth` | lease `slot_id`, plus any hardware-generation key |
+
+A run-owned lease is `{slot_id, generation}`. `PipelineSlotPool` mints it and
+is the authority on ownership: releasing the current lease is idempotent, while
+releasing it after the slot has been re-leased is rejected.
+
+`ChipWorker` is downstream of that pool and cannot re-derive ownership — it
+never sees an acquire or a release. It keeps a per-slot high-water mark and
+rejects any generation below it, which stops a *superseded* lease from
+selecting resources the slot's newer owner holds. That is strictly weaker than
+an ownership check: a lease that was released but whose successor has not yet
+reached `ChipWorker` still passes, because nothing has raised the mark. Closing
+that window needs the admission layer to gate dispatch on `pool.owns(lease)`
+before handing work down, which is whole-run admission's job, not this
+layer's.
+
+AICore streams are outside that lease. The instruction cache belongs to the
+cores, every slot publishes its image to the same GM code address, and the
+platform offers no cache invalidation for code replaced there. Creating a
+stream is the only operation known to leave a core free of the previous image's
+instructions; selecting an already-existing one is not. So each run creates its
+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.
+
+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
+mailbox frame, a second device execution, or cross-run publication overlap.
+Carrying a lease across the mailbox, and deciding when slot 1 may be leased at
+all, belong to whole-run admission.
+
+Simulation implements the same depth, so the contract means the same thing on
+both platforms: its runner owns one arena bank and one retained temporary
+buffer per slot, and its single-entry prebuilt-arena cache stays owned by
+bank 0 exactly as onboard's does.
+
 #### TRB temporary buffer
 
 `tensormap_and_ringbuffer` stages ordinary non-child tensor arguments through a
-runner-scoped retained temporary buffer instead of a per-run `device_malloc()` /
+retained temporary buffer owned per pipeline slot, instead of a per-run `device_malloc()` /
 `device_free()` pair. This is always on for TRB — an internal allocation
 optimization with no user-facing switch. It is not serialized in task mailboxes
 and does not change `TaskArgs`, `CallConfig`, child-memory tensors, or public

python/bindings/task_interface.cpp

@@ -1430,23 +1430,51 @@ NB_MODULE(_task_interface, m) {
             "Launch a callable_id from a TaskArgs (used for in-process callers). "
             "Returns None; timing is emitted as `[STRACE]` log markers."
         )
+        .def(
+            "_run_with_pipeline_lease",
+            [](ChipWorker &self, int32_t callable_id, TaskArgs &args, const CallConfig &config, uint32_t slot_id,
+               uint64_t generation) {
+                self.run_with_lease(callable_id, make_view(args), config, PipelineSlotLease{slot_id, 0, generation});
+            },
+            nb::arg("callable_id"), nb::arg("args"), nb::arg("config"), nb::arg("slot_id"), nb::arg("generation"),
+            "Internal generation-safe pipeline-slot launch used by hierarchical admission."
+        )
+        .def(
+            "_run_with_pipeline_lease",
+            [](ChipWorker &self, int32_t callable_id, ChipStorageTaskArgs &args, const CallConfig &config,
+               uint32_t slot_id, uint64_t generation) {
+                self.run_with_lease(callable_id, &args, config, PipelineSlotLease{slot_id, 0, generation});
+            },
+            nb::arg("callable_id"), nb::arg("args"), nb::arg("config"), nb::arg("slot_id"), nb::arg("generation"),
+            "Internal generation-safe pipeline-slot launch for pre-encoded task args."
+        )
         .def(
             "run_from_blob",
             [](ChipWorker &self, int32_t callable_id, uint64_t args_blob_ptr, size_t blob_capacity,
-               const CallConfig &config, uint64_t accepted_state_addr, int32_t accepted_value) {
+               const CallConfig &config, uint64_t accepted_state_addr, int32_t accepted_value, uint32_t pipeline_slot,
+               uint64_t pipeline_generation) {
                 // 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, reinterpret_cast<volatile int32_t *>(accepted_state_addr), accepted_value
-                );
+                if (pipeline_generation == 0) {
+                    self.run(
+                        callable_id, view, config, reinterpret_cast<volatile int32_t *>(accepted_state_addr),
+                        accepted_value
+                    );
+                } else {
+                    self.run_with_lease(
+                        callable_id, view, config, PipelineSlotLease{pipeline_slot, 0, pipeline_generation},
+                        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,
+            nb::arg("accepted_state_addr") = 0, nb::arg("accepted_value") = 0, nb::arg("pipeline_slot") = 0,
+            nb::arg("pipeline_generation") = 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 "
@@ -1464,6 +1492,23 @@ NB_MODULE(_task_interface, m) {
         )
         .def_prop_ro("device_id", &ChipWorker::device_id)
         .def_prop_ro("initialized", &ChipWorker::initialized)
+        .def_prop_ro("pipeline_depth", &ChipWorker::pipeline_depth)
+        .def_prop_ro("runtime_slot_count", &ChipWorker::runtime_slot_count)
+        .def_prop_ro(
+            "runtime_buffer_addrs", &ChipWorker::runtime_buffer_addrs,
+            "Host Runtime staging buffer address of every copy the runtime's "
+            "PipelineContract asked for, in slot order."
+        )
+        .def(
+            "retained_temp_addr", &ChipWorker::retained_temp_addr, nb::arg("slot_id"),
+            "Retained temporary-buffer address held for one pipeline slot, or 0 "
+            "while that slot holds none."
+        )
+        .def(
+            "arena_bank_gm_heap_base", &ChipWorker::arena_bank_gm_heap_base, nb::arg("bank_id"),
+            "Committed GM heap base of one arena bank on the bound runner, or 0 "
+            "when that bank has never been committed."
+        )
         .def_prop_ro(
             "aicpu_dlopen_count", &ChipWorker::aicpu_dlopen_count,
             "Number of distinct callable entries the AICPU has dlopened for on the "
@@ -1479,11 +1524,10 @@ NB_MODULE(_task_interface, m) {
         )
         .def_prop_ro(
             "run_stream_set_create_count", &ChipWorker::run_stream_set_create_count,
-            "Number of run stream sets the bound 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; "
-            "platforms whose runs use the persistent bootstrap pair report 0. "
-            "Tests assert this to verify repeated runs do not rebuild the set."
+            "Number of AICore run streams the bound runner has created. The AICPU "
+            "stream belongs to a pipeline slot for the worker's lifetime, while each "
+            "run creates and retires its own AICore stream, so this advances once per "
+            "run; platforms whose runs use the persistent bootstrap pair report 0."
         )
         .def("malloc", &ChipWorker::malloc, nb::arg("size"))
         .def("free", &ChipWorker::free, nb::arg("ptr"))

python/simpler/task_interface.py

@@ -1320,6 +1320,13 @@ def _run_slot(self, callable_id, args, config=None, **kwargs):
         # Returns None; per-stage timing is emitted as `[STRACE]` log markers.
         self._impl.run(int(callable_id), args, config)
 
+    def _run_slot_with_pipeline_lease(self, callable_id, args, slot_id, generation, config=None, **kwargs):
+        if config is None:
+            config = CallConfig()
+        for k, v in kwargs.items():
+            setattr(config, k, v)
+        self._impl._run_with_pipeline_lease(int(callable_id), args, config, int(slot_id), int(generation))
+
     def _unregister_slot(self, callable_id):
         self._impl.unregister_callable(int(callable_id))
 
@@ -1335,9 +1342,30 @@ def host_dlopen_count(self):
 
     @property
     def run_stream_set_create_count(self):
-        """Number of run stream generations the bound runner has created."""
+        """Number of AICore run streams the bound runner has created."""
         return self._impl.run_stream_set_create_count
 
+    @property
+    def pipeline_depth(self):
+        return self._impl.pipeline_depth
+
+    @property
+    def runtime_slot_count(self):
+        return self._impl.runtime_slot_count
+
+    @property
+    def runtime_buffer_addrs(self):
+        """Address of each host Runtime staging buffer, in slot order."""
+        return list(self._impl.runtime_buffer_addrs)
+
+    def arena_bank_gm_heap_base(self, bank_id):
+        """Committed GM heap base of one arena bank, or 0 when uncommitted."""
+        return int(self._impl.arena_bank_gm_heap_base(int(bank_id)))
+
+    def retained_temp_addr(self, slot_id):
+        """Retained temporary-buffer address for one slot, or 0 when unheld."""
+        return int(self._impl.retained_temp_addr(int(slot_id)))
+
     def malloc(self, size):
         """Allocate memory. Returns a pointer (uint64)."""
         return int(self._impl.malloc(int(size)))

python/simpler/worker.py

@@ -6275,12 +6275,12 @@ def host_dlopen_count(self) -> int:
 
     @property
     def run_stream_set_create_count(self) -> int:
-        """L2 only: number of run stream generations the runner has created.
+        """L2 only: number of AICore run streams the runner has created.
 
-        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).
+        AICPU streams belong to pipeline slots for the worker's lifetime, while
+        each run creates and retires its own AICore stream, so this advances
+        once per run. 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