Fix: preflight remote L3 timeouts before any startup resource

python/simpler/remote_l3_worker.py

@@ -143,6 +143,10 @@ def _reap_session_runner(proc: subprocess.Popen[Any]) -> None:
 
 def _start_session(manifest: dict[str, Any]) -> dict[str, Any]:
     _validate_manifest(manifest)
+    # Both numeric timeouts are validated before any spawn resource (ready pipe,
+    # manifest tempfile, runner Popen) exists: the runner is never launched only
+    # to die on an invalid session_timeout_s or startup_remaining_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)

python/simpler/worker.py

@@ -3836,6 +3836,14 @@ def _init_hierarchical(self) -> None:
         if self.level >= 4 and device_ids:
             raise RuntimeError("Worker level >= 4 must use add_worker(); device_ids are only supported on L3 Workers")
 
+        # Only a worker that carries remote workers has a remote session to
+        # time out. Its remote_session_timeout_s is validated here, before any
+        # startup resource (mailbox shm, pre-fork _Worker mmap, child fork,
+        # daemon socket) exists, so an invalid value fails without a
+        # partially-built subtree to roll back.
+        if self._remote_worker_specs:
+            self._remote_session_timeout_s()
+
         # 1. Allocate sub-worker mailboxes (unified layout, MAILBOX_SIZE each).
         for _ in range(n_sub):
             shm = SharedMemory(create=True, size=MAILBOX_SIZE)
@@ -3896,6 +3904,8 @@ def _activate_remote_sessions(self, deadline: float) -> None:
         failure propagates to init()'s single rollback, which closes every
         session recorded in ``self._remote_sessions``.
         """
+        if not self._remote_worker_specs:
+            return
         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()

tests/ut/py/test_worker/test_remote_startup_budget.py

@@ -28,6 +28,7 @@
 import pytest
 import simpler.remote_l3_session as session_mod
 import simpler.remote_l3_worker as daemon_mod
+import simpler.worker as worker_mod
 from simpler.worker import RemoteWorkerSpec, Worker
 
 
@@ -99,6 +100,119 @@ def test_startup_remaining_used_when_present(self, mod):
         assert mod._startup_remaining_s({"startup_remaining_s": 8.0, "session_timeout_s": 30.0}) == 8.0
 
 
+class TestParentPreflightBeforeResources:
+    """An invalid remote_session_timeout_s fails before the parent builds any
+    startup resource — no mailbox shm, no pre-fork _Worker, ever constructed."""
+
+    @pytest.mark.parametrize("bad", [0.0, -1.0, float("inf"), float("nan")])
+    def test_invalid_timeout_creates_no_startup_resources(self, monkeypatch, bad):
+        shm_ctor = MagicMock(side_effect=AssertionError("SharedMemory constructed before timeout preflight"))
+        worker_ctor = MagicMock(side_effect=AssertionError("_Worker constructed before timeout preflight"))
+        monkeypatch.setattr(worker_mod, "SharedMemory", shm_ctor)
+        monkeypatch.setattr(worker_mod, "_Worker", worker_ctor)
+
+        w = Worker(level=4, num_sub_workers=1, remote_session_timeout_s=bad)
+        w.add_remote_worker(_spec())
+        try:
+            with pytest.raises(ValueError, match="remote_session_timeout_s"):
+                w._init_hierarchical()
+            assert shm_ctor.call_count == 0
+            assert worker_ctor.call_count == 0
+        finally:
+            w.close()
+
+    def test_no_remotes_does_not_validate_remote_timeout(self, monkeypatch):
+        # The preflight is gated on actual remote presence: a worker with no
+        # remote workers never validates remote_session_timeout_s (only a
+        # level >= 4 parent can carry remotes), so an unused/invalid value does
+        # not fail an otherwise-valid tree.
+        monkeypatch.setattr(worker_mod, "_Worker", MagicMock())
+        preflight = MagicMock(side_effect=AssertionError("remote timeout validated without any remote worker"))
+        monkeypatch.setattr(Worker, "_remote_session_timeout_s", preflight)
+
+        w = Worker(level=4, num_sub_workers=0, remote_session_timeout_s=-1.0)
+        try:
+            w._init_hierarchical()
+            assert preflight.call_count == 0
+        finally:
+            w._worker = None
+            w.close()
+
+    def test_activate_remote_sessions_no_remotes_is_noop(self, monkeypatch):
+        # With no remotes, activation returns before validating the timeout or
+        # touching the C++ Worker — the empty tree is a clean no-op.
+        preflight = MagicMock(side_effect=AssertionError("remote timeout validated with no remotes to activate"))
+        monkeypatch.setattr(Worker, "_remote_session_timeout_s", preflight)
+
+        w = Worker(level=4, num_sub_workers=0)
+        mock_worker = MagicMock()
+        w._worker = mock_worker
+        try:
+            w._activate_remote_sessions(time.monotonic() + 5.0)
+            assert preflight.call_count == 0
+            assert mock_worker.add_remote_l3_socket.call_count == 0
+            assert w._remote_sessions == []
+        finally:
+            w._worker = None
+            w.close()
+
+
+class TestDaemonPreflightBeforeSpawn:
+    """The daemon validates both numeric timeouts before any spawn resource —
+    ready pipe, manifest tempfile, runner Popen — is created."""
+
+    @staticmethod
+    def _manifest(**overrides) -> dict:
+        manifest = {
+            "session_id": 1,
+            "worker_id": 0,
+            "parent_worker_level": 4,
+            "remote_worker_level": 3,
+            "platform": "a2a3sim",
+            "transport": "sim",
+            "session_timeout_s": 30.0,
+            "startup_remaining_s": 10.0,
+        }
+        manifest.update(overrides)
+        return manifest
+
+    @pytest.mark.parametrize(
+        ("manifest_kwargs", "match"),
+        [
+            ({"session_timeout_s": 0.0}, "session_timeout_s"),
+            ({"session_timeout_s": -1.0}, "session_timeout_s"),
+            ({"session_timeout_s": float("inf")}, "session_timeout_s"),
+            ({"session_timeout_s": float("nan")}, "session_timeout_s"),
+            ({"startup_remaining_s": 0.0}, "startup_remaining_s"),
+            ({"startup_remaining_s": -1.0}, "startup_remaining_s"),
+            ({"startup_remaining_s": float("inf")}, "startup_remaining_s"),
+            ({"startup_remaining_s": float("nan")}, "startup_remaining_s"),
+        ],
+    )
+    def test_invalid_timeout_spawns_nothing(self, monkeypatch, manifest_kwargs, match):
+        pipe_mock = MagicMock(side_effect=AssertionError("os.pipe before timeout preflight"))
+        tempfile_mock = MagicMock(side_effect=AssertionError("NamedTemporaryFile before timeout preflight"))
+        popen_mock = MagicMock(side_effect=AssertionError("Popen before timeout preflight"))
+        monkeypatch.setattr(daemon_mod.os, "pipe", pipe_mock)
+        monkeypatch.setattr(daemon_mod.tempfile, "NamedTemporaryFile", tempfile_mock)
+        monkeypatch.setattr(daemon_mod.subprocess, "Popen", popen_mock)
+
+        with pytest.raises(ValueError, match=match):
+            daemon_mod._start_session(self._manifest(**manifest_kwargs))
+        assert pipe_mock.call_count == 0
+        assert tempfile_mock.call_count == 0
+        assert popen_mock.call_count == 0
+
+    def test_valid_session_timeout_but_invalid_startup_remaining_spawns_nothing(self, monkeypatch):
+        # A valid session_timeout_s must not shield an invalid startup_remaining_s
+        # from the pre-spawn gate.
+        popen_mock = MagicMock(side_effect=AssertionError("Popen before timeout preflight"))
+        monkeypatch.setattr(daemon_mod.subprocess, "Popen", popen_mock)
+        with pytest.raises(ValueError, match="startup_remaining_s"):
+            daemon_mod._start_session(self._manifest(session_timeout_s=30.0, startup_remaining_s=-1.0))
+        assert popen_mock.call_count == 0
+
+
 class TestActivationAfterFork:
     def test_init_hierarchical_does_not_open_sessions(self, monkeypatch):
         """Opening a session (which starts the remote subtree) is deferred out of