fix(worker): unify lifecycle into one authoritative, terminal-close() machine

docs/callable-identity-registration.md

@@ -45,11 +45,12 @@ dispatch. For chip targets, the local implementation also prewarms executable
 state before first use when startup or dynamic register control reaches the
 child before dispatch.
 
-The current chip child loop preserves the historical TASK_READY lazy-prepare
-safety net: if a digest is already registered in the child identity table but
-the explicit prewarm step was missed, the child prepares its private slot in
-the dispatch path and logs a warning. This is not a public cid compatibility
-path and it does not fetch missing callable bytes from the parent.
+The chip child loop requires a prepared slot at dispatch: a TASK_READY only
+consumes a slot already staged via `_CTRL_PREPARE` (dynamic register) or via the
+startup snapshot (initial callables, prepared before INIT_READY). Reaching
+TASK_READY for an unprepared slot is a control-flow error and fails loudly
+rather than lazily preparing in the dispatch path. The child never fetches
+missing callable bytes from the parent.
 
 ### Chip Executable Prewarm
 

docs/remote-l3-worker-design.md

@@ -204,10 +204,10 @@ Use a two-process remote model:
    single-threaded pipe. This handoff is not the remote transport protocol.
 4. The session runner reads the manifest before starting transport threads and
    constructs `Worker(level=3)`.
-5. The runner then performs an explicit prestart step equivalent to
-   `inner_worker.init()` plus `_start_hierarchical()` for the inner Worker:
-   allocate local mailboxes, fork local chip/sub children, register local
-   endpoints with the inner C++ Worker, and start the inner Scheduler and
+5. The runner then performs an explicit prestart step, `inner_worker.init()`
+   for the inner Worker — `init()` is the single startup point: it allocates
+   local mailboxes, forks local chip/sub children, registers local endpoints
+   with the inner C++ Worker, and starts the inner Scheduler and
    `WorkerThread`s.
 6. Only after this local L3 child tree is established does the session runner
    bring up sockets, RDMA queue pairs, health threads, or UB doorbells for task

python/bindings/task_interface.cpp

@@ -1361,6 +1361,13 @@ NB_MODULE(_task_interface, m) {
             },
             nb::arg("host_lib_path"), nb::arg("aicpu_path"), nb::arg("aicore_path"), nb::arg("dispatcher_path"),
             nb::arg("device_id"), nb::arg("prewarm_config") = nb::none(),
+            // Release the GIL for the (potentially long) native device attach so
+            // another Python thread can run during it — e.g. a concurrent close()
+            // observing INITIALIZING and failing fast (a GIL held for the whole
+            // attach would block it until init returned). This does not make
+            // ChipWorker cross-thread-safe: init/finalize still run on the owning
+            // thread (enforced by Worker.close()).
+            nb::call_guard<nb::gil_scoped_release>(),
             "Bind the runtime library and attach to device_id. When prewarm_config is "
             "given, its ring sizing is built + cached inside init (fork-constant, no "
             "cross-process control command). A no-op for runtimes without a prebuilt arena."

python/bindings/worker_bind.h

@@ -413,7 +413,11 @@ inline void bind_worker(nb::module_ &m) {
             "Register a REMOTE_L3 endpoint after the session reports HELLO READY."
         )
 
-        .def("init", &Worker::init, "Start the Scheduler thread.")
+        // Release the GIL while starting the Scheduler thread so another Python
+        // thread can run during it — e.g. a concurrent close() observing
+        // INITIALIZING and failing fast. init/close remain same-thread-only
+        // (enforced by Worker.close()).
+        .def("init", &Worker::init, nb::call_guard<nb::gil_scoped_release>(), "Start the Scheduler thread.")
         .def("close", &Worker::close, "Stop the Scheduler thread.")
 
         .def(

tests/st/a2a3/tensormap_and_ringbuffer/dynamic_register/test_dynamic_register.py

@@ -265,7 +265,10 @@ def test_prepare_capacity_overflow_post_start(st_platform, st_device_ids):
     worker = Worker(
         level=3,
         device_ids=[int(st_device_ids[0])],
-        num_sub_workers=0,
+        # A sub worker gives the LOCAL_PYTHON fillers below a valid resolver, so
+        # they exercise the shared capacity ceiling as eligible registrations
+        # (not inert ones) alongside the chip callable.
+        num_sub_workers=1,
         platform=st_platform,
         runtime=_RUNTIME,
     )

tests/ut/py/test_callable_identity.py

@@ -1418,8 +1418,7 @@ def remote_free(self, *args):
     worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
     importer_worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19074", platform="a2a3sim"))
     worker._worker = fake
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
 
     imported = worker.remote_import(exported, worker=importer_worker_id)
     worker.remote_free(owner)
@@ -1470,8 +1469,7 @@ def remote_import(self, *args):
     fake = FailingRemoteWorker()
     worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
     worker._worker = fake  # type: ignore[assignment]
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
 
     with pytest.raises(RuntimeError, match="import failed"):
         worker.remote_import(exported, worker=worker_id)
@@ -1522,8 +1520,7 @@ def fail_from_imported_mapping(**kwargs):
     fake = FakeRemoteWorker()
     worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
     worker._worker = fake  # type: ignore[assignment]
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
     monkeypatch.setattr(RemoteBufferHandle, "_from_imported_mapping", staticmethod(fail_from_imported_mapping))
 
     with pytest.raises(RuntimeError, match="handle build failed"):
@@ -1589,8 +1586,7 @@ def remote_free(self, *args):
     owner._mark_released()
     worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
     worker._worker = FailingRemoteWorker()  # type: ignore[assignment]
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
     worker._pending_remote_buffer_frees = [owner]
 
     worker._flush_pending_remote_frees()
@@ -1963,10 +1959,8 @@ def close(self):
     worker = Worker(level=4, num_sub_workers=0)
     worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
     fake = FakeRemoteCWorker()
-    worker._initialized = True
     worker._worker = fake  # type: ignore[assignment]
-    worker._hierarchical_started = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
     try:
         handle = worker.remote_malloc(worker=worker_id, nbytes=4)
         assert handle.worker_id == worker_id
@@ -2009,8 +2003,7 @@ def remote_prepare_register(self, *args):
 
     worker = Worker(level=4, num_sub_workers=0)
     worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
     worker._worker = FailingPrepareWorker()  # type: ignore[assignment]
     digest = hashid_to_digest(_REMOTE_NOOP_ORCH_HASHID)
 
@@ -2038,8 +2031,7 @@ def remote_abort_register(self, worker_id, *args):
     fake = FailingCommitWorker()
     worker = Worker(level=4, num_sub_workers=0)
     worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
     worker._worker = fake  # type: ignore[assignment]
     digest = hashid_to_digest(_REMOTE_NOOP_ORCH_HASHID)
 
@@ -2058,8 +2050,7 @@ def remote_unregister(self, *args):
     worker = Worker(level=4, num_sub_workers=0)
     worker_id = worker.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
     handle = worker.register(RemoteCallable(_REMOTE_NOOP_ORCH_TARGET), workers=[worker_id])
-    worker._initialized = True
-    worker._hierarchical_start_state = "started"
+    worker._lifecycle = worker_mod._Lifecycle.READY
     worker._worker = FailingUnregisterWorker()  # type: ignore[assignment]
 
     worker.unregister(handle)

tests/ut/py/test_worker/test_host_buffer_registration.py

@@ -122,8 +122,7 @@ def broadcast_control_all(self, worker_type, sub_cmd, payload, digest, *, timeou
             return []
 
     worker = Worker(level=3)
-    worker._initialized = True
-    worker._hierarchical_started = True
+    worker._lifecycle = worker_mod._Lifecycle.READY
     worker._chip_shms = [None]
     worker._worker = FakeWorker()
     worker._start_hierarchical = lambda: None
@@ -260,8 +259,9 @@ def test_sub_only_l3_allows_create_host_buffer(self):
             w.close()
 
     def test_childless_l3_rejects_create_host_buffer(self):
+        # No callable registered: a childless L3 with a pre-registered callable
+        # is rejected earlier, at init() (see TestEligibleTargetPrecheck).
         w = Worker(level=3, num_sub_workers=0)
-        w.register(lambda args: None)
         w.init()
         try:
             with pytest.raises(RuntimeError, match="at least one forked chip or sub child"):

tests/ut/py/test_worker/test_host_worker.py

@@ -269,13 +269,13 @@ def test_prepare_waits_for_first_startup_then_uses_post_start_path(self):
         hw.init()
         try:
             with hw._hierarchical_start_cv:
-                hw._hierarchical_start_state = "starting"
+                hw._lifecycle = worker_mod._Lifecycle.INITIALIZING
 
             observed = {}
 
             def fake_post_start_register(target):
                 observed["target"] = target
-                observed["state"] = hw._hierarchical_start_state
+                observed["state"] = hw._lifecycle
                 observed["hierarchical_started"] = hw._hierarchical_started
                 return 7
 
@@ -301,15 +301,14 @@ def do_register():
             t.start()
             assert wait_entered.wait(timeout=2.0)
             with hw._hierarchical_start_cv:
-                hw._hierarchical_started = True
-                hw._hierarchical_start_state = "started"
+                hw._lifecycle = worker_mod._Lifecycle.READY
                 hw._hierarchical_start_cv.notify_all()
             t.join(timeout=2.0)
 
             assert not t.is_alive()
             assert errors == []
             assert result == [7]
-            assert observed["state"] == "started"
+            assert observed["state"] is worker_mod._Lifecycle.READY
             assert observed["hierarchical_started"] is True
         finally:
             if "original_wait" in locals():
@@ -322,15 +321,15 @@ def test_unregister_waits_for_first_startup_then_uses_post_start_path(self):
         hw.init()
         try:
             with hw._hierarchical_start_cv:
-                hw._hierarchical_start_state = "starting"
+                hw._lifecycle = worker_mod._Lifecycle.INITIALIZING
 
             observed = {}
 
             def fake_broadcast_py_control(worker_types, sub_cmd, *, digest=None, payload=None, strict=False):
                 observed["worker_types"] = worker_types
                 observed["sub_cmd"] = sub_cmd
                 observed["digest"] = digest
-                observed["state"] = hw._hierarchical_start_state
+                observed["state"] = hw._lifecycle
                 observed["hierarchical_started"] = hw._hierarchical_started
                 return []
 
@@ -357,16 +356,15 @@ def do_unregister():
             assert handle.digest in hw._identity_registry
 
             with hw._hierarchical_start_cv:
-                hw._hierarchical_started = True
-                hw._hierarchical_start_state = "started"
+                hw._lifecycle = worker_mod._Lifecycle.READY
                 hw._hierarchical_start_cv.notify_all()
             t.join(timeout=2.0)
 
             assert not t.is_alive()
             assert errors == []
             assert observed["sub_cmd"] == _CTRL_PY_UNREGISTER
             assert observed["digest"] == handle.digest
-            assert observed["state"] == "started"
+            assert observed["state"] is worker_mod._Lifecycle.READY
             assert observed["hierarchical_started"] is True
             assert handle.digest not in hw._identity_registry
         finally:
@@ -382,7 +380,7 @@ def test_register_during_initializing_waits_then_takes_post_start_path(self):
         hw.init()
         try:
             with hw._hierarchical_start_cv:
-                hw._hierarchical_start_state = "starting"
+                hw._lifecycle = worker_mod._Lifecycle.INITIALIZING
 
             errors: list[BaseException] = []
             result: list[object] = []
@@ -408,8 +406,7 @@ def do_register():
             assert len(hw._identity_registry) == 0
 
             with hw._hierarchical_start_cv:
-                hw._hierarchical_started = True
-                hw._hierarchical_start_state = "started"
+                hw._lifecycle = worker_mod._Lifecycle.READY
                 hw._hierarchical_start_cv.notify_all()
             t.join(timeout=2.0)
 
@@ -440,8 +437,9 @@ def test_prepare_chip_callable_at_cid_overflow_raises(self):
         # cid budget is enforced under the new dynamic-prepare path too:
         # pre-fill registry with lambdas pre-init, init, then attempt one
         # post-init ChipCallable prepare and observe the existing
-        # MAX_REGISTERED_CALLABLE_IDS RuntimeError.
-        hw = Worker(level=3, num_sub_workers=0)
+        # MAX_REGISTERED_CALLABLE_IDS RuntimeError. A sub child gives the
+        # pre-registered python callables an eligible dispatch target.
+        hw = Worker(level=3, num_sub_workers=1)
         try:
             for i in range(MAX_REGISTERED_CALLABLE_IDS):
                 hw.register(_unique_py_callable(i))
@@ -485,8 +483,7 @@ def test_unregister_chip_callable_after_init_no_chips_succeeds(self):
 
     def test_prepare_chip_callable_broadcast_runs_without_registry_lock(self):
         hw = Worker(level=3, num_sub_workers=0)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
         callable_obj = ChipCallable.build(signature=[], func_name="x", binary=b"\x00", children=[])
         observed = {}
 
@@ -508,8 +505,7 @@ def test_register_child_chip_broadcast_runs_without_registry_lock(self):
         from simpler.worker import _build_callable_registration  # noqa: PLC0415
 
         hw = Worker(level=3, num_sub_workers=0)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
         callable_obj = ChipCallable.build(signature=[], func_name="x", binary=b"\x00", children=[])
         digest = _build_callable_registration(hw, callable_obj).digest
         observed = {}
@@ -637,8 +633,7 @@ def test_python_unregister_child_failure_warns_pops_and_allows_reuse(self, capsy
 
         hw = Worker(level=3, num_sub_workers=1)
         handle = hw.register(lambda args: None)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
         calls = []
 
         class FakeWorker:
@@ -671,8 +666,7 @@ def test_pending_unregister_cid_is_not_reused_until_broadcast_returns(self):
 
         hw = Worker(level=3, num_sub_workers=1)
         handle = hw.register(lambda args: None)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
 
         broadcast_started = threading.Event()
         release_broadcast = threading.Event()
@@ -720,8 +714,7 @@ def target(args):
 
         hw = Worker(level=3, num_sub_workers=1)
         handle = hw.register(target)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
 
         unregister_started = threading.Event()
         release_unregister = threading.Event()
@@ -769,8 +762,7 @@ def target(args):
         hw = Worker(level=3, num_sub_workers=1)
         first = hw.register(target)
         second = hw.register(target)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
 
         unregister_started = threading.Event()
         release_unregister = threading.Event()
@@ -1224,8 +1216,7 @@ def broadcast_register_all(self, blob_ptr, blob_size, digest):
                 return [_FakeControlResult("NEXT_LEVEL", 0, True)]
 
         hw = Worker(level=3, num_sub_workers=1)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
         hw._worker = FakeWorker()
         callable_obj = ChipCallable.build(signature=[], func_name="x", binary=b"\x00", children=[])
 
@@ -1260,8 +1251,7 @@ def control_digest_only(self, worker_type, worker_id, sub_cmd, digest, timeout_s
                 return _FakeControlResult("NEXT_LEVEL", worker_id, True)
 
         hw = Worker(level=3, num_sub_workers=1)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
         hw._worker = FakeWorker()
         callable_obj = ChipCallable.build(signature=[], func_name="x", binary=b"\x00", children=[])
 
@@ -1292,8 +1282,7 @@ def broadcast_unregister_all(self, digest):
                 return ["cleanup failed"]
 
         hw = Worker(level=3, num_sub_workers=1)
-        hw._initialized = True
-        hw._hierarchical_started = True
+        hw._lifecycle = worker_mod._Lifecycle.READY
         hw._worker = FakeWorker()
         callable_obj = ChipCallable.build(signature=[], func_name="x", binary=b"\x00", children=[])
 

tests/ut/py/test_worker/test_l3_l2_message_queue.py

@@ -243,8 +243,7 @@ def _make_orchestrator() -> tuple[Orchestrator, Worker, SharedMemory, _FakeClien
         (l3_l2_orch_comm, "_l3_host_mapped_counter_wait", l3_l2_orch_comm._l3_host_mapped_counter_wait),
         (l3_l2_orch_comm, "_l3_host_mapped_region_close", l3_l2_orch_comm._l3_host_mapped_region_close),
     ]
-    worker._initialized = True
-    worker._hierarchical_started = True
+    worker._lifecycle = worker_module._Lifecycle.READY
     worker._worker = _FakeCWorker()
     worker._chip_shms = [shm]
     worker._l3_l2_test_fake_client = fake_client

tests/ut/py/test_worker/test_l3_l2_orch_comm.py

@@ -157,8 +157,7 @@ def _make_started_sim_worker() -> tuple[Worker, SharedMemory, _FakeDirectCWorker
     assert shm.buf is not None
     _mailbox_store_i32(_buffer_field_addr(shm.buf, _OFF_STATE), _IDLE)
     fake_c_worker = _FakeDirectCWorker()
-    worker._initialized = True
-    worker._hierarchical_started = True
+    worker._lifecycle = worker_module._Lifecycle.READY
     worker._worker = fake_c_worker
     worker._chip_shms = [shm]
     return worker, shm, fake_c_worker
@@ -173,8 +172,7 @@ def _make_started_onboard_worker(platform: str = "a2a3") -> tuple[Worker, Shared
         access_profile=int(l3_l2_orch_comm.L3L2RegionAccessProfile.ONBOARD_VMM),
         device_id=2,
     )
-    worker._initialized = True
-    worker._hierarchical_started = True
+    worker._lifecycle = worker_module._Lifecycle.READY
     worker._worker = fake_c_worker
     worker._chip_shms = [shm]
     return worker, shm, fake_c_worker