Fix: bound Remote L3 startup by one root budget, not per-remote timeout

python/simpler/remote_l3_session.py

@@ -20,6 +20,7 @@
 import hashlib
 import importlib
 import json
+import math
 import os
 import signal
 import socket
@@ -163,11 +164,24 @@ def _send_ready(fd: int, payload: dict[str, Any]) -> None:
 
 def _session_timeout_s(manifest: dict[str, Any]) -> float:
     timeout_s = float(manifest.get("session_timeout_s", 30.0))
-    if timeout_s <= 0:
-        raise ValueError("manifest session_timeout_s must be positive")
+    if not (timeout_s > 0 and math.isfinite(timeout_s)):
+        raise ValueError("manifest session_timeout_s must be a positive finite number of seconds")
     return timeout_s
 
 
+def _startup_remaining_s(manifest: dict[str, Any]) -> float:
+    # The parent's remaining slice of the single root startup budget. Absent only
+    # from a pre-P0.3 parent; fall back to the runtime command timeout then. The
+    # fallback is evaluated only when the key is absent, so a valid
+    # startup_remaining_s is not held hostage to an invalid session_timeout_s.
+    if "startup_remaining_s" not in manifest:
+        return _session_timeout_s(manifest)
+    remaining_s = float(manifest["startup_remaining_s"])
+    if not (remaining_s > 0 and math.isfinite(remaining_s)):
+        raise ValueError("manifest startup_remaining_s must be a positive finite number of seconds")
+    return remaining_s
+
+
 def _health_loop(sock: socket.socket, stop: threading.Event, session_id: int, worker_id: int) -> None:
     conn: socket.socket | None = None
     sock.settimeout(0.2)
@@ -911,18 +925,22 @@ def run_session(manifest: dict[str, Any], ready_fd: int) -> int:
     health_thread: threading.Thread | None = None
     try:
         session_timeout_s = _session_timeout_s(manifest)
+        startup_remaining_s = _startup_remaining_s(manifest)
         manifest_dispatch_registry = _install_manifest_dispatcher_registry(manifest)
         # Register the inner L3 callables before init() so they are frozen into
         # the eager startup snapshot and uploaded to the chip children before
         # those children publish INIT_READY. init() is the single, eager startup
         # point: it forks and readies the whole inner L3->L2 tree, so the runner
         # reports ready (below) only once that subtree is up.
         manifest_inner_handles = _install_manifest_inner_registry(manifest, inner_worker)
-        # Bound the inner startup and mark it non-root so its chip/sub children
-        # inherit this runner's process group (see start_new_session in the
-        # daemon) rather than splitting into their own — the daemon reaps the
-        # whole L3->L2 subtree with one killpg on the runner.
-        inner_worker.init(_startup_deadline=time.monotonic() + session_timeout_s)
+        # Bound the inner startup by the parent's remaining startup budget
+        # (rebuilt against this host's own monotonic clock — the parent's
+        # absolute deadline is not comparable across machines), not a fresh full
+        # session_timeout_s. Mark it non-root so its chip/sub children inherit
+        # this runner's process group (see start_new_session in the daemon)
+        # rather than splitting into their own — the daemon reaps the whole
+        # L3->L2 subtree with one killpg on the runner.
+        inner_worker.init(_startup_deadline=time.monotonic() + startup_remaining_s)
 
         listen_host = str(manifest.get("listen_host", "127.0.0.1"))
         command_sock = _bind_listener(listen_host)

python/simpler/remote_l3_worker.py

@@ -13,6 +13,7 @@
 import argparse
 import contextlib
 import json
+import math
 import os
 import select
 import signal
@@ -67,11 +68,25 @@ def _validate_manifest(manifest: dict[str, Any]) -> None:
 
 def _session_timeout_s(manifest: dict[str, Any]) -> float:
     timeout_s = float(manifest.get("session_timeout_s", 30.0))
-    if timeout_s <= 0:
-        raise ValueError("manifest session_timeout_s must be positive")
+    if not (timeout_s > 0 and math.isfinite(timeout_s)):
+        raise ValueError("manifest session_timeout_s must be a positive finite number of seconds")
     return timeout_s
 
 
+def _startup_remaining_s(manifest: dict[str, Any]) -> float:
+    # The parent's remaining slice of the single root startup budget bounds how
+    # long the runner may take to bring up its subtree. Absent only from a
+    # pre-P0.3 parent; fall back to the runtime command timeout then. The
+    # fallback is evaluated only when the key is absent, so a valid
+    # startup_remaining_s is not held hostage to an invalid session_timeout_s.
+    if "startup_remaining_s" not in manifest:
+        return _session_timeout_s(manifest)
+    remaining_s = float(manifest["startup_remaining_s"])
+    if not (remaining_s > 0 and math.isfinite(remaining_s)):
+        raise ValueError("manifest startup_remaining_s must be a positive finite number of seconds")
+    return remaining_s
+
+
 def _read_runner_ready(fd: int, timeout_s: float) -> dict[str, Any]:
     chunks = bytearray()
     deadline = time.monotonic() + timeout_s
@@ -128,7 +143,9 @@ def _reap_session_runner(proc: subprocess.Popen[Any]) -> None:
 
 def _start_session(manifest: dict[str, Any]) -> dict[str, Any]:
     _validate_manifest(manifest)
-    timeout_s = _session_timeout_s(manifest)
+    # The runner must publish ready within the parent's remaining startup budget,
+    # not a fresh full command timeout.
+    timeout_s = _startup_remaining_s(manifest)
     ready_r, ready_w = os.pipe()
     manifest_path = ""
     proc: subprocess.Popen[Any] | None = None

python/simpler/worker.py

@@ -2133,8 +2133,8 @@ def _is_wildcard_session_host(host: str) -> bool:
 
     def _remote_session_timeout_s(self) -> float:
         timeout_s = float(self._config.get("remote_session_timeout_s", 30.0))
-        if timeout_s <= 0:
-            raise ValueError("Worker remote_session_timeout_s must be positive")
+        if not (timeout_s > 0 and math.isfinite(timeout_s)):
+            raise ValueError("Worker remote_session_timeout_s must be a positive finite number of seconds")
         return timeout_s
 
     @staticmethod
@@ -2182,7 +2182,9 @@ def _remote_dispatcher_entries_for_worker(self, worker_id: int) -> list[dict[str
             )
         return entries
 
-    def _build_remote_manifest(self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int) -> dict[str, Any]:
+    def _build_remote_manifest(
+        self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, startup_remaining_s: float
+    ) -> dict[str, Any]:
         daemon_host, _daemon_port = self._parse_remote_endpoint(spec.endpoint)
         listen_host = spec.session_listen_host or ("127.0.0.1" if daemon_host == "localhost" else daemon_host)
         if self._is_wildcard_session_host(listen_host) and not spec.allow_wildcard_session_bind:
@@ -2198,7 +2200,11 @@ def _build_remote_manifest(self, *, spec: RemoteWorkerSpec, worker_id: int, sess
             "num_sub_workers": int(spec.num_sub_workers),
             "heap_ring_size": self._config.get("remote_heap_ring_size", None),
             "transport": spec.transport,
+            # session_timeout_s bounds the runtime command socket; startup_remaining_s
+            # bounds this session's slice of the single root startup budget. They are
+            # distinct: the remote must not spend runtime-command time as startup time.
             "session_timeout_s": self._remote_session_timeout_s(),
+            "startup_remaining_s": float(startup_remaining_s),
             "listen_host": listen_host,
             "connect_host": daemon_host,
             "remote_task_dispatcher": self._remote_dispatcher_entries_for_worker(worker_id),
@@ -2207,10 +2213,12 @@ def _build_remote_manifest(self, *, spec: RemoteWorkerSpec, worker_id: int, sess
         }
 
     def _open_remote_session(
-        self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, timeout_s: float
+        self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, timeout_s: float, startup_remaining_s: float
     ) -> _RemoteSession:
         daemon_host, daemon_port = self._parse_remote_endpoint(spec.endpoint)
-        manifest = self._build_remote_manifest(spec=spec, worker_id=worker_id, session_id=session_id)
+        manifest = self._build_remote_manifest(
+            spec=spec, worker_id=worker_id, session_id=session_id, startup_remaining_s=startup_remaining_s
+        )
         with socket.create_connection((daemon_host, daemon_port), timeout=timeout_s) as sock:
             sock.settimeout(timeout_s)
             self._send_remote_daemon_json(sock, manifest)
@@ -3874,28 +3882,53 @@ def _init_hierarchical(self) -> None:
         else:
             self._worker = _Worker(self.level, int(heap_ring_size))
 
-        # Open remote L3 sessions (the remote runner brings up its own L3->L2
-        # tree and answers the daemon handshake). Opening only exchanges a
-        # transient socket message — it starts no local thread. Registering the
-        # socket into the C++ Worker (which spawns the RemoteL3Endpoint health
-        # thread) is deferred to _start_hierarchical, after the last local fork,
-        # to preserve the fork-before-thread invariant.
-        opened_remote_sessions: list[_RemoteSession] = []
-        try:
-            for worker_id, spec in zip(self._remote_worker_ids, self._remote_worker_specs, strict=True):
-                session_id = uuid.uuid4().int & ((1 << 63) - 1)
-                if session_id == 0:
-                    session_id = 1
-                timeout_s = self._remote_session_timeout_s()
-                session = self._open_remote_session(
-                    spec=spec, worker_id=worker_id, session_id=session_id, timeout_s=timeout_s
-                )
-                opened_remote_sessions.append(session)
-                self._remote_sessions.append(session)
-                opened_remote_sessions.pop()
-        except BaseException:
-            self._close_remote_sessions(opened_remote_sessions)
-            raise
+    def _activate_remote_sessions(self, deadline: float) -> None:
+        """Open and register every remote L3 session within the shared startup budget.
+
+        Called only from _start_hierarchical, after this process's last local
+        fork, so opening a session (which starts the remote subtree) and
+        registering its endpoint (which spawns the health thread) both stay
+        behind every local fork. All remotes draw from the single root startup
+        ``deadline``: each computes the remaining budget at the moment it opens,
+        propagates it as the manifest's ``startup_remaining_s`` so the remote
+        bounds its own subtree by this process's remaining time (measured on the
+        remote's own monotonic clock) instead of a fresh full timeout. Any
+        failure propagates to init()'s single rollback, which closes every
+        session recorded in ``self._remote_sessions``.
+        """
+        session_timeout = self._remote_session_timeout_s()
+        for worker_id, spec in zip(self._remote_worker_ids, self._remote_worker_specs, strict=True):
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise RuntimeError("remote L3 session activation: startup deadline exceeded")
+            session_id = uuid.uuid4().int & ((1 << 63) - 1)
+            if session_id == 0:
+                session_id = 1
+            # The handshake blocks until the remote subtree is READY, so the
+            # socket timeout must cover the startup budget granted below — not
+            # the (shorter) runtime command timeout.
+            session = self._open_remote_session(
+                spec=spec,
+                worker_id=worker_id,
+                session_id=session_id,
+                timeout_s=remaining,
+                startup_remaining_s=remaining,
+            )
+            self._remote_sessions.append(session)
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise RuntimeError("remote L3 endpoint attach: startup deadline exceeded")
+            assert self._worker is not None
+            self._worker.add_remote_l3_socket(
+                session.worker_id,
+                session.session_id,
+                spec.transport,
+                session.command_host,
+                session.command_port,
+                session.health_host,
+                session.health_port,
+                min(session_timeout, remaining),
+            )
 
     def _start_hierarchical(self) -> None:  # noqa: PLR0912 -- three parallel fork loops (sub/chip/next) + bootstrap wait + scheduler register/init; branches track the fork order documented in the body
         """Fork every local child, await the subtree, register endpoints, start the scheduler.
@@ -4065,25 +4098,11 @@ def _setup(inner=inner_worker):
         # failure, exit, or hang aborts startup here.
         self._await_children_ready(self._next_level_shms, self._next_level_pids, "next_level", deadline)
 
-        # Last local fork is done. Now — and only now — register remote L3
-        # endpoints, which spawns the RemoteL3Endpoint health thread; doing it
-        # here keeps every local fork ahead of every C++/Python thread. Each
-        # remote attach consumes the shared startup budget.
-        for session, spec in zip(self._remote_sessions, self._remote_worker_specs, strict=True):
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
-                raise RuntimeError("remote L3 endpoint attach: startup deadline exceeded")
-            assert self._worker is not None
-            self._worker.add_remote_l3_socket(
-                session.worker_id,
-                session.session_id,
-                spec.transport,
-                session.command_host,
-                session.command_port,
-                session.health_host,
-                session.health_port,
-                min(self._remote_session_timeout_s(), remaining),
-            )
+        # Last local fork is done. Now — and only now — open and register remote
+        # L3 sessions: opening starts the remote subtree and registering spawns
+        # the RemoteL3Endpoint health thread, so both must follow every local
+        # fork. Each remote consumes this process's remaining startup budget.
+        self._activate_remote_sessions(deadline)
 
         # _Worker was constructed in _init_hierarchical (pre-fork) so children
         # inherit the HeapRing MAP_SHARED mmap. Register PROCESS-mode workers via

tests/ut/py/test_callable_identity.py

@@ -504,7 +504,7 @@ def close(self):
     def fake_worker_ctor(*args):
         return fake_c_worker
 
-    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s):
+    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
         opened_worker_ids.append(worker_id)
         return worker_mod._RemoteSession(  # noqa: SLF001
             worker_id=worker_id,
@@ -545,6 +545,7 @@ def test_remote_session_manifest_uses_endpoint_host_as_default_bind():
             spec=RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"),
             worker_id=0,
             session_id=1,
+            startup_remaining_s=30.0,
         )
         assert loopback["listen_host"] == "127.0.0.1"
         assert loopback["connect_host"] == "127.0.0.1"
@@ -553,6 +554,7 @@ def test_remote_session_manifest_uses_endpoint_host_as_default_bind():
             spec=RemoteWorkerSpec(endpoint="10.0.0.8:19073", platform="a2a3sim"),
             worker_id=0,
             session_id=1,
+            startup_remaining_s=30.0,
         )
         assert remote["listen_host"] == "10.0.0.8"
         assert remote["connect_host"] == "10.0.0.8"
@@ -569,7 +571,7 @@ def test_remote_session_manifest_requires_wildcard_bind_opt_in():
             session_listen_host="0.0.0.0",
         )
         with pytest.raises(ValueError, match="wildcard session bind"):
-            worker._build_remote_manifest(spec=spec, worker_id=0, session_id=1)
+            worker._build_remote_manifest(spec=spec, worker_id=0, session_id=1, startup_remaining_s=30.0)
 
         opted_in = worker._build_remote_manifest(
             spec=RemoteWorkerSpec(
@@ -580,6 +582,7 @@ def test_remote_session_manifest_requires_wildcard_bind_opt_in():
             ),
             worker_id=0,
             session_id=1,
+            startup_remaining_s=30.0,
         )
         assert opted_in["listen_host"] == "0.0.0.0"
         assert opted_in["connect_host"] == "10.0.0.8"
@@ -1613,7 +1616,7 @@ def fake_worker_ctor(*args):
 
     calls = 0
 
-    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s):
+    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
         nonlocal calls
         calls += 1
         if calls == 2:
@@ -1670,7 +1673,7 @@ def close(self):
     def fake_worker_ctor(*args):
         return fake_c_worker
 
-    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s):
+    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
         return opened_session
 
     def fake_close_remote_session(self, session):

tests/ut/py/test_remote_l3_lifecycle.py

@@ -10,6 +10,7 @@
 import os
 import socket
 import threading
+import time
 from typing import cast
 
 import pytest
@@ -175,6 +176,63 @@ def close(self):
         os.close(ready_r)
 
 
+def test_run_session_bounds_subtree_by_startup_remaining_not_session_timeout(monkeypatch):
+    """The inner subtree deadline comes from the parent's startup_remaining_s
+    (its slice of the single root startup budget), not the runtime command
+    session_timeout_s — the two are deliberately different here."""
+
+    captured = {}
+
+    class FakeWorker:
+        def __init__(self, *args, **kwargs):
+            self.closed = False
+
+        def init(self, *args, _startup_deadline=None, **kwargs):
+            captured["deadline"] = _startup_deadline
+            captured["at"] = time.monotonic()
+
+        def close(self):
+            self.closed = True
+
+    class FakeCommandSock:
+        def getsockname(self):
+            return ("127.0.0.1", 12345)
+
+        def settimeout(self, timeout):
+            pass
+
+        def accept(self):
+            raise socket.timeout("stop after init")
+
+        def close(self):
+            pass
+
+    class FakeHealthSock:
+        def getsockname(self):
+            return ("127.0.0.1", 12346)
+
+        def close(self):
+            pass
+
+    sockets = [FakeCommandSock(), FakeHealthSock()]
+    ready_r, ready_w = os.pipe()
+
+    monkeypatch.setattr(remote_l3_session, "Worker", FakeWorker)
+    monkeypatch.setattr(remote_l3_session, "_install_manifest_dispatcher_registry", lambda manifest: {})
+    monkeypatch.setattr(remote_l3_session, "_install_manifest_inner_registry", lambda manifest, worker: {})
+    monkeypatch.setattr(remote_l3_session, "_bind_listener", lambda host: sockets.pop(0))
+    monkeypatch.setattr(remote_l3_session, "_health_loop", lambda *args: None)
+
+    try:
+        remote_l3_session.run_session(_manifest(session_timeout_s=0.01, startup_remaining_s=50.0), ready_w)
+    finally:
+        os.close(ready_r)
+
+    budget = captured["deadline"] - captured["at"]
+    # ~50s startup budget, not the 0.01s runtime command timeout.
+    assert 40.0 < budget <= 50.0
+
+
 def test_health_loop_closes_active_connection_on_stop():
     stop = threading.Event()