diff --git a/docs/remote-l3-worker-design.md b/docs/remote-l3-worker-design.md
index c5edc32911..f981e99e7a 100644
--- a/docs/remote-l3-worker-design.md
+++ b/docs/remote-l3-worker-design.md
@@ -346,8 +346,10 @@ from simpler.worker import RemoteCallable, RemoteWorkerSpec, Worker
 
 w4 = Worker(level=4)
 
+# endpoint host must be a numeric IP (or "localhost"); hostnames are rejected
+# because getaddrinfo resolution is unbounded and could pin startup.
 l3 = RemoteWorkerSpec(
-    endpoint="node17:19073",
+    endpoint="10.0.0.17:19073",
     platform="a2a3",
     runtime="tensormap_and_ringbuffer",
     device_ids=list(range(16)),
diff --git a/python/bindings/worker_bind.h b/python/bindings/worker_bind.h
index 2da4007efe..06b834a619 100644
--- a/python/bindings/worker_bind.h
+++ b/python/bindings/worker_bind.h
@@ -402,15 +402,16 @@ inline void bind_worker(nb::module_ &m) {
             "add_remote_l3_socket",
             [](Worker &self, int32_t worker_id, uint64_t session_id, const std::string &transport_name,
                const std::string &host, uint16_t port, const std::string &health_host, uint16_t health_port,
-               double timeout_s) {
+               double attach_timeout_s, double runtime_timeout_s) {
                 nb::gil_scoped_release release;
                 self.add_remote_l3_socket(
-                    worker_id, session_id, transport_name, host, port, health_host, health_port, timeout_s
+                    worker_id, session_id, transport_name, host, port, health_host, health_port, attach_timeout_s,
+                    runtime_timeout_s
                 );
             },
             nb::arg("worker_id"), nb::arg("session_id"), nb::arg("transport_name"), nb::arg("host"), nb::arg("port"),
-            nb::arg("health_host"), nb::arg("health_port"), nb::arg("timeout_s") = 30.0,
-            "Register a REMOTE_L3 endpoint after the session reports HELLO READY."
+            nb::arg("health_host"), nb::arg("health_port"), nb::arg("attach_timeout_s") = 30.0,
+            nb::arg("runtime_timeout_s") = 30.0, "Register a REMOTE_L3 endpoint after the session reports HELLO READY."
         )
 
         // Release the GIL while starting the Scheduler thread so another Python
diff --git a/python/simpler/remote_l3_session.py b/python/simpler/remote_l3_session.py
index 09317eb8a7..aa7748eca5 100644
--- a/python/simpler/remote_l3_session.py
+++ b/python/simpler/remote_l3_session.py
@@ -182,6 +182,19 @@ def _startup_remaining_s(manifest: dict[str, Any]) -> float:
     return remaining_s
 
 
+def _startup_deadline(manifest: dict[str, Any]) -> float:
+    # The daemon and this runner share CLOCK_MONOTONIC (same host), so when the
+    # daemon supplies an absolute deadline the runner uses it directly — its
+    # remaining is measured post-spawn, and its deadline cannot exceed the
+    # daemon's. Only a pre-P0.3 daemon omits it; fall back to the duration then.
+    if "startup_deadline_monotonic" in manifest:
+        deadline = float(manifest["startup_deadline_monotonic"])
+        if not math.isfinite(deadline):
+            raise ValueError("manifest startup_deadline_monotonic must be a finite monotonic timestamp")
+        return deadline
+    return time.monotonic() + _startup_remaining_s(manifest)
+
+
 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)
@@ -924,8 +937,14 @@ def run_session(manifest: dict[str, Any], ready_fd: int) -> int:
     stop_health = threading.Event()
     health_thread: threading.Thread | None = None
     try:
-        session_timeout_s = _session_timeout_s(manifest)
-        startup_remaining_s = _startup_remaining_s(manifest)
+        # Validate the runtime command timeout wire value up front (rejects a
+        # malformed session_timeout_s); the command lane itself idle-waits blocking.
+        _session_timeout_s(manifest)
+        # Establish this runner's single absolute deadline before any startup work
+        # (registry install, inner init) so every stage draws from it. It is the
+        # daemon's shared-clock deadline when supplied (so spawn/import time is
+        # already charged), else derived from the remaining duration.
+        startup_deadline = _startup_deadline(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
@@ -933,14 +952,13 @@ def run_session(manifest: dict[str, Any], ready_fd: int) -> int:
         # 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 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
+        # Bound the inner startup by the runner's single deadline established
+        # above (which already charges registry-install time), 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)
+        inner_worker.init(_startup_deadline=startup_deadline)
 
         listen_host = str(manifest.get("listen_host", "127.0.0.1"))
         command_sock = _bind_listener(listen_host)
@@ -965,8 +983,22 @@ def run_session(manifest: dict[str, Any], ready_fd: int) -> int:
             },
         )
 
-        command_sock.settimeout(session_timeout_s)
+        # Waiting for the parent to attach is still the attach phase, so bound it
+        # by the remaining startup budget — not session_timeout_s. Otherwise a
+        # tiny runtime timeout could drop the parent before it connects, and a
+        # large one could keep the runner alive well past the root deadline if the
+        # parent is lost.
+        attach_remaining = startup_deadline - time.monotonic()
+        if attach_remaining <= 0:
+            raise TimeoutError("remote L3 runner: startup deadline exceeded before parent attach")
+        command_sock.settimeout(attach_remaining)
         conn, _addr = command_sock.accept()
+        # Force the command connection blocking, explicitly: accept() inherits
+        # socket.getdefaulttimeout(), which a user module imported during registry
+        # install could have set. A finite timeout would tear down a healthy but
+        # idle session; the loop instead idle-waits blocking and ends when the
+        # parent closes the command socket (read_frame sees EOF).
+        conn.settimeout(None)
         with conn:
             _run_command_loop(conn, manifest, inner_worker, manifest_inner_handles, manifest_dispatch_registry)
         return 0
diff --git a/python/simpler/remote_l3_worker.py b/python/simpler/remote_l3_worker.py
index a53834d9ed..3561ee95c4 100644
--- a/python/simpler/remote_l3_worker.py
+++ b/python/simpler/remote_l3_worker.py
@@ -87,9 +87,8 @@ def _startup_remaining_s(manifest: dict[str, Any]) -> float:
     return remaining_s
 
 
-def _read_runner_ready(fd: int, timeout_s: float) -> dict[str, Any]:
+def _read_runner_ready(fd: int, deadline: float) -> dict[str, Any]:
     chunks = bytearray()
-    deadline = time.monotonic() + timeout_s
     while True:
         remaining = deadline - time.monotonic()
         if remaining <= 0:
@@ -154,18 +153,30 @@ def _start_session(manifest: dict[str, Any]) -> tuple[dict[str, Any], subprocess
     # 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)
+    # One local absolute deadline for this session's startup: the runner-ready
+    # wait below and the budget handed to the runner both derive from it, so the
+    # daemon's own setup and the runner's init cannot each restart the full slice.
+    deadline = time.monotonic() + _startup_remaining_s(manifest)
     ready_r, ready_w = os.pipe()
     manifest_path = ""
     proc: subprocess.Popen[Any] | None = None
     try:
+        runner_remaining = deadline - time.monotonic()
+        if runner_remaining <= 0:
+            raise TimeoutError("remote L3 session: startup deadline exceeded before runner launch")
+        runner_manifest = dict(manifest)
+        # Hand the runner this daemon's ABSOLUTE monotonic deadline. Daemon and
+        # runner share CLOCK_MONOTONIC (same host, Popen-spawned), so the runner
+        # derives its remaining post-spawn and its deadline equals this one — the
+        # spawn/import/registry time is charged, not restarted from a frozen
+        # duration. startup_remaining_s stays for a pre-P0.3 runner's fallback.
+        runner_manifest["startup_deadline_monotonic"] = deadline
+        runner_manifest["startup_remaining_s"] = runner_remaining
         with tempfile.NamedTemporaryFile(
             "w", encoding="utf-8", prefix="simpler-remote-l3-", suffix=".json", delete=False
         ) as f:
             manifest_path = f.name
-            json.dump(manifest, f, sort_keys=True)
+            json.dump(runner_manifest, f, sort_keys=True)
         proc = subprocess.Popen(
             [
                 sys.executable,
@@ -185,7 +196,7 @@ def _start_session(manifest: dict[str, Any]) -> tuple[dict[str, Any], subprocess
         os.close(ready_w)
         ready_w = -1
         try:
-            ready = _read_runner_ready(ready_r, timeout_s)
+            ready = _read_runner_ready(ready_r, deadline)
         except BaseException:
             _wait_or_kill_runner(proc)
             raise
diff --git a/python/simpler/worker.py b/python/simpler/worker.py
index 4e568a3b67..3b6effbe3d 100644
--- a/python/simpler/worker.py
+++ b/python/simpler/worker.py
@@ -369,6 +369,9 @@ def qualname(self) -> str:
 
 @dataclass(frozen=True)
 class RemoteWorkerSpec:
+    # endpoint is "host:port"; host must be a numeric IP (or "localhost").
+    # Hostnames are rejected at add_remote_worker time — getaddrinfo resolution is
+    # unbounded and uncancellable and would risk pinning startup on a hung DNS.
     endpoint: str
     platform: str
     runtime: str = "tensormap_and_ringbuffer"
@@ -2110,6 +2113,11 @@ def add_remote_worker(self, spec: RemoteWorkerSpec) -> int:
                 raise TypeError("Worker.add_remote_worker: remote L3 workers require a level >= 4 parent")
             if not isinstance(spec, RemoteWorkerSpec):
                 raise TypeError("Worker.add_remote_worker expects a RemoteWorkerSpec")
+            # Validate the endpoint here, before any startup resource exists, so a
+            # non-numeric host fails at registration rather than mid-activation
+            # (which would roll back the whole already-forked tree).
+            host, _port = self._parse_remote_endpoint(spec.endpoint)
+            self._validate_numeric_endpoint_host(host)
             worker_id = self._allocate_next_level_worker_id()
             self._remote_worker_specs.append(spec)
             self._remote_worker_ids.append(worker_id)
@@ -2127,6 +2135,21 @@ def _parse_remote_endpoint(endpoint: str) -> tuple[str, int]:
             raise ValueError(f"RemoteWorkerSpec.endpoint port out of range: {port}")
         return host, port
 
+    @staticmethod
+    def _validate_numeric_endpoint_host(host: str) -> None:
+        # Remote L3 endpoints are numeric-only (or localhost) by contract:
+        # hostname resolution via getaddrinfo is unbounded and uncancellable, so
+        # it is rejected rather than risk pinning startup on a hung resolver.
+        if host == "localhost":
+            return
+        try:
+            socket.getaddrinfo(host, None, flags=socket.AI_NUMERICHOST)
+        except socket.gaierror as exc:
+            raise ValueError(
+                f"RemoteWorkerSpec.endpoint host must be a numeric IP address (hostname resolution is "
+                f"unbounded and unsupported for remote L3); got {host!r}"
+            ) from exc
+
     @staticmethod
     def _is_wildcard_session_host(host: str) -> bool:
         return host in ("0.0.0.0", "::")
@@ -2137,15 +2160,70 @@ def _remote_session_timeout_s(self) -> float:
             raise ValueError("Worker remote_session_timeout_s must be a positive finite number of seconds")
         return timeout_s
 
+    @staticmethod
+    def _remaining_until(deadline: float, what: str) -> float:
+        # A blocking op's slice of the single root startup deadline. Raising here
+        # keeps the timeout local and clear, and avoids settimeout(0.0) — which
+        # would flip the socket to non-blocking and surface BlockingIOError.
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError(f"{what}: startup deadline exceeded")
+        return remaining
+
+    @staticmethod
+    def _resolve_within_deadline(host: str, port: int, deadline: float) -> list[Any]:
+        # Numeric-only, by contract: getaddrinfo is not cancellable, so a hung
+        # NSS/DNS lookup could pin init() in INITIALIZING past the root deadline.
+        # AI_NUMERICHOST performs NO name resolution (it parses a numeric literal
+        # or fails immediately), so this never blocks; a hostname is rejected
+        # outright rather than risk an unbounded stall. "localhost" is accepted as
+        # the loopback literal. The deadline pre-check keeps a spent budget from
+        # even attempting the parse.
+        Worker._remaining_until(deadline, "remote L3 session resolve")
+        lookup = "127.0.0.1" if host == "localhost" else host
+        try:
+            return socket.getaddrinfo(
+                lookup, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST
+            )
+        except socket.gaierror as exc:
+            raise ValueError(
+                f"remote L3 endpoint host must be a numeric IP address (hostname resolution is "
+                f"unbounded and unsupported); got {host!r}"
+            ) from exc
+
+    @staticmethod
+    def _connect_within_deadline(host: str, port: int, deadline: float) -> socket.socket:
+        # Bound name resolution AND every per-address connect attempt by the single
+        # root deadline — mirroring the C++ connect_tcp_socket — so a slow resolver
+        # or a black-holed first address cannot let this stage restart the clock or
+        # outrun the startup budget (unlike socket.create_connection, which grants
+        # a fresh full timeout to every address and never bounds getaddrinfo).
+        infos = Worker._resolve_within_deadline(host, port, deadline)
+        last_exc: BaseException | None = None
+        for family, socktype, proto, _canonname, sockaddr in infos:
+            remaining = Worker._remaining_until(deadline, "remote L3 session connect")
+            sock = socket.socket(family, socktype, proto)
+            try:
+                sock.settimeout(remaining)
+                sock.connect(sockaddr)
+                return sock
+            except OSError as exc:
+                last_exc = exc
+                sock.close()
+        if last_exc is not None:
+            raise last_exc
+        raise OSError(f"remote L3 session connect: no address for {host}:{port}")
+
     @staticmethod
     def _send_remote_daemon_json(sock: socket.socket, payload: dict[str, Any]) -> None:
         data = json.dumps(payload, sort_keys=True).encode("utf-8")
         sock.sendall(struct.pack("<I", len(data)) + data)
 
     @staticmethod
-    def _recv_remote_daemon_json(sock: socket.socket) -> dict[str, Any]:
+    def _recv_remote_daemon_json(sock: socket.socket, deadline: float) -> dict[str, Any]:
         size_data = bytearray()
         while len(size_data) < 4:
+            sock.settimeout(Worker._remaining_until(deadline, "remote daemon reply"))
             chunk = sock.recv(4 - len(size_data))
             if not chunk:
                 raise EOFError("remote daemon closed before reply length")
@@ -2155,6 +2233,7 @@ def _recv_remote_daemon_json(sock: socket.socket) -> dict[str, Any]:
             raise RuntimeError("remote daemon reply exceeds maximum")
         data = bytearray()
         while len(data) < size:
+            sock.settimeout(Worker._remaining_until(deadline, "remote daemon reply"))
             chunk = sock.recv(size - len(data))
             if not chunk:
                 raise EOFError("remote daemon closed before full reply")
@@ -2213,16 +2292,24 @@ def _build_remote_manifest(
         }
 
     def _open_remote_session(
-        self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, timeout_s: float, startup_remaining_s: float
+        self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, deadline: 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, startup_remaining_s=startup_remaining_s
-        )
-        with socket.create_connection((daemon_host, daemon_port), timeout=timeout_s) as sock:
-            sock.settimeout(timeout_s)
+        # Every blocking op (resolve, connect, send, framed recv) derives its
+        # remaining from the single root deadline, so their sum cannot exceed the
+        # root startup budget.
+        with self._connect_within_deadline(daemon_host, daemon_port, deadline) as sock:
+            manifest = self._build_remote_manifest(
+                spec=spec, worker_id=worker_id, session_id=session_id, startup_remaining_s=0.0
+            )
+            # Derive the send budget AFTER building the (registry-iterating)
+            # manifest, right before send, so the socket timeout and the wire
+            # duration reflect what is actually left — not a pre-build sample.
+            startup_remaining_s = self._remaining_until(deadline, "remote L3 session handshake")
+            manifest["startup_remaining_s"] = startup_remaining_s
+            sock.settimeout(startup_remaining_s)
             self._send_remote_daemon_json(sock, manifest)
-            reply = self._recv_remote_daemon_json(sock)
+            reply = self._recv_remote_daemon_json(sock, deadline)
         if not reply.get("ok", False):
             raise RuntimeError(f"remote L3 session startup failed for worker {worker_id}: {reply.get('error')}")
         return _RemoteSession(
@@ -3784,6 +3871,14 @@ def init(self, prewarm_config=None, *, _startup_deadline: float | None = None) -
             # single lifecycle state so no thread ever observes a started
             # hierarchy while the worker is not yet READY.
             with self._hierarchical_start_cv:
+                # Final root-deadline gate, in the same critical section as the
+                # commit: a thread descheduled between startup and here cannot
+                # publish READY past the single root startup deadline. Applied to
+                # every hierarchical worker, not just those with direct remote
+                # sessions — a local child may have remote descendants whose
+                # startup this deadline also bounds.
+                if self.level >= 3 and time.monotonic() >= self._startup_deadline:
+                    raise RuntimeError("hierarchical startup: startup deadline exceeded before READY")
                 self._lifecycle = _Lifecycle.READY
                 self._hierarchical_start_cv.notify_all()
         except BaseException as exc:
@@ -3914,21 +4009,22 @@ def _activate_remote_sessions(self, deadline: float) -> None:
             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.
+            # The handshake blocks until the remote subtree is READY; the whole
+            # open derives its per-op remaining from the shared root deadline.
             session = self._open_remote_session(
                 spec=spec,
                 worker_id=worker_id,
                 session_id=session_id,
-                timeout_s=remaining,
-                startup_remaining_s=remaining,
+                deadline=deadline,
             )
             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
+            # attach_timeout bounds the command/health connect + HELLO read by the
+            # remaining startup budget; runtime_timeout is the full runtime command
+            # budget, never clamped by leftover startup time.
             self._worker.add_remote_l3_socket(
                 session.worker_id,
                 session.session_id,
@@ -3937,8 +4033,13 @@ def _activate_remote_sessions(self, deadline: float) -> None:
                 session.command_port,
                 session.health_host,
                 session.health_port,
-                min(session_timeout, remaining),
+                remaining,
+                session_timeout,
             )
+        # Attach may have consumed the last slice of the budget; a final root
+        # deadline check keeps a just-over-budget attach from committing READY.
+        if time.monotonic() >= deadline:
+            raise RuntimeError("remote L3 activation: startup deadline exceeded after attach")
 
     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.
diff --git a/src/common/hierarchical/remote_endpoint.cpp b/src/common/hierarchical/remote_endpoint.cpp
index 2827faebe3..fb96e3e6f4 100644
--- a/src/common/hierarchical/remote_endpoint.cpp
+++ b/src/common/hierarchical/remote_endpoint.cpp
@@ -175,6 +175,9 @@ void configure_socket_no_sigpipe(int fd, const std::string &label) {
 }
 
 ssize_t send_no_sigpipe(int fd, const uint8_t *data, size_t size) {
+    // The fd is O_NONBLOCK, so a send never blocks: it writes what fits and
+    // returns EAGAIN when the buffer is full, letting write_all re-poll under the
+    // deadline. MSG_NOSIGNAL only suppresses SIGPIPE on a closed peer.
 #if defined(MSG_NOSIGNAL)
     return ::send(fd, data, size, MSG_NOSIGNAL);
 #else
@@ -182,12 +185,6 @@ ssize_t send_no_sigpipe(int fd, const uint8_t *data, size_t size) {
 #endif
 }
 
-void restore_socket_flags(int fd, int flags, const std::string &label) {
-    if (::fcntl(fd, F_SETFL, flags) != 0) {
-        throw std::runtime_error(label + ": fcntl(F_SETFL) failed: " + std::strerror(errno));
-    }
-}
-
 short poll_socket(
     int fd, short events, Deadline deadline, const std::string &timeout_message, const std::string &poll_error_context
 ) {
@@ -253,10 +250,15 @@ void validate_remote_buffer_export_range(
     }
 }
 
-int connect_tcp_socket(const std::string &host, uint16_t port, const std::string &label, double timeout_s) {
-    Deadline deadline =
-        std::chrono::steady_clock::now() +
-        std::chrono::duration_cast<std::chrono::steady_clock::duration>(std::chrono::duration<double>(timeout_s));
+std::chrono::steady_clock::time_point deadline_from_now(double timeout_s) {
+    return std::chrono::steady_clock::now() +
+           std::chrono::duration_cast<std::chrono::steady_clock::duration>(std::chrono::duration<double>(timeout_s));
+}
+
+int connect_tcp_socket(const std::string &host, uint16_t port, const std::string &label, Deadline deadline) {
+    // The caller's absolute deadline is used verbatim for resolution and every
+    // per-address connect — never re-derived as now()+remaining, which would
+    // extend the hard wall if this thread is descheduled before entry.
     std::string port_s = std::to_string(port);
     std::vector<TcpAddress> addresses = resolve_tcp_addresses_with_timeout(host, port_s, label, deadline);
     int fd = -1;
@@ -283,7 +285,10 @@ int connect_tcp_socket(const std::string &host, uint16_t port, const std::string
             }
             int rc = ::connect(candidate, reinterpret_cast<const sockaddr *>(&addr.addr), addr.addrlen);
             if (rc == 0) {
-                restore_socket_flags(candidate, flags, label);
+                // Keep the fd O_NONBLOCK for its whole life: all frame I/O polls
+                // for readiness under a deadline, so recv/send never block (a
+                // blocking send of a large frame to a stalled reader would
+                // outlast the deadline, and MSG_DONTWAIT is not portable).
                 fd = candidate;
                 break;
             }
@@ -294,7 +299,6 @@ int connect_tcp_socket(const std::string &host, uint16_t port, const std::string
             }
             int connect_error = wait_for_connect(candidate, deadline, label);
             if (connect_error == 0) {
-                restore_socket_flags(candidate, flags, label);
                 fd = candidate;
                 break;
             }
@@ -352,25 +356,37 @@ void validate_owner_buffer_handle(const RemoteBufferHandle &handle, size_t reque
 }  // namespace
 
 RemoteL3SocketTransport::RemoteL3SocketTransport(
-    std::string host, uint16_t port, std::string health_host, uint16_t health_port, double timeout_s
+    std::string host, uint16_t port, std::string health_host, uint16_t health_port, double attach_timeout_s,
+    double runtime_timeout_s
 ) :
     host_(std::move(host)),
     port_(port),
     health_host_(std::move(health_host)),
     health_port_(health_port),
-    timeout_s_(timeout_s) {
+    attach_timeout_s_(attach_timeout_s),
+    runtime_timeout_s_(runtime_timeout_s) {
     if (host_.empty()) throw std::invalid_argument("RemoteL3SocketTransport: host must be non-empty");
     if (port_ == 0) throw std::invalid_argument("RemoteL3SocketTransport: port must be non-zero");
     if (health_host_.empty()) throw std::invalid_argument("RemoteL3SocketTransport: health host must be non-empty");
     if (health_port_ == 0) throw std::invalid_argument("RemoteL3SocketTransport: health port must be non-zero");
-    if (timeout_s_ <= 0.0) throw std::invalid_argument("RemoteL3SocketTransport: timeout must be positive");
+    if (attach_timeout_s_ <= 0.0)
+        throw std::invalid_argument("RemoteL3SocketTransport: attach timeout must be positive");
+    if (runtime_timeout_s_ <= 0.0)
+        throw std::invalid_argument("RemoteL3SocketTransport: runtime timeout must be positive");
+    // One absolute deadline for the whole attach phase; command-connect, the
+    // HELLO read, and health-connect all derive their remaining from it so the
+    // attach cannot exceed the caller's startup-budget slice.
+    attach_deadline_ =
+        std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(
+                                               std::chrono::duration<double>(attach_timeout_s_)
+                                           );
     connect_socket();
 }
 
 RemoteL3SocketTransport::~RemoteL3SocketTransport() { close_socket(); }
 
 void RemoteL3SocketTransport::connect_socket() {
-    fd_ = connect_tcp_socket(host_, port_, "RemoteL3SocketTransport(command)", timeout_s_);
+    fd_ = connect_tcp_socket(host_, port_, "RemoteL3SocketTransport(command)", attach_deadline_);
 }
 
 void RemoteL3SocketTransport::close_socket() {
@@ -401,7 +417,8 @@ void RemoteL3SocketTransport::check_health() {
 
 void RemoteL3SocketTransport::start_health_monitor(uint64_t session_id, int32_t worker_id) {
     if (health_thread_.joinable()) return;
-    health_fd_ = connect_tcp_socket(health_host_, health_port_, "RemoteL3SocketTransport(health)", timeout_s_);
+    // Health-connect is the last attach-phase op, so it shares attach_deadline_.
+    health_fd_ = connect_tcp_socket(health_host_, health_port_, "RemoteL3SocketTransport(health)", attach_deadline_);
     health_stop_.store(false, std::memory_order_release);
     health_failed_.store(false, std::memory_order_release);
     {
@@ -409,7 +426,9 @@ void RemoteL3SocketTransport::start_health_monitor(uint64_t session_id, int32_t
         health_error_.clear();
     }
     int fd = health_fd_;
-    double timeout_s = timeout_s_;
+    // The health-monitor loop is a runtime lane; its per-frame read uses the
+    // runtime timeout, not the (spent) attach budget.
+    double timeout_s = runtime_timeout_s_;
     health_thread_ = std::thread([this, fd, session_id, worker_id, timeout_s]() {
         auto read_exact = [&](uint8_t *data, size_t size) -> bool {
             size_t off = 0;
@@ -424,7 +443,7 @@ void RemoteL3SocketTransport::start_health_monitor(uint64_t session_id, int32_t
                 (void)poll_socket(fd, POLLIN, deadline, "timed out waiting for HEALTH frame", "poll failed");
                 ssize_t n = ::recv(fd, data + off, size - off, 0);
                 if (n < 0) {
-                    if (errno == EINTR) continue;
+                    if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue;
                     throw std::runtime_error(std::string("recv failed: ") + std::strerror(errno));
                 }
                 if (n == 0) throw std::runtime_error("health socket closed");
@@ -472,10 +491,7 @@ void RemoteL3SocketTransport::stop_health_monitor() {
     }
 }
 
-void RemoteL3SocketTransport::wait_readable() {
-    auto deadline =
-        std::chrono::steady_clock::now() +
-        std::chrono::duration_cast<std::chrono::steady_clock::duration>(std::chrono::duration<double>(timeout_s_));
+void RemoteL3SocketTransport::wait_readable(std::chrono::steady_clock::time_point deadline) {
     while (true) {
         check_health();
         auto now = std::chrono::steady_clock::now();
@@ -488,10 +504,7 @@ void RemoteL3SocketTransport::wait_readable() {
     }
 }
 
-void RemoteL3SocketTransport::wait_writable() {
-    auto deadline =
-        std::chrono::steady_clock::now() +
-        std::chrono::duration_cast<std::chrono::steady_clock::duration>(std::chrono::duration<double>(timeout_s_));
+void RemoteL3SocketTransport::wait_writable(std::chrono::steady_clock::time_point deadline) {
     while (true) {
         check_health();
         auto now = std::chrono::steady_clock::now();
@@ -504,13 +517,17 @@ void RemoteL3SocketTransport::wait_writable() {
     }
 }
 
-void RemoteL3SocketTransport::write_all(const uint8_t *data, size_t size) {
+void RemoteL3SocketTransport::write_all(
+    const uint8_t *data, size_t size, std::chrono::steady_clock::time_point deadline
+) {
     size_t off = 0;
     while (off < size) {
-        wait_writable();
+        wait_writable(deadline);
         ssize_t n = send_no_sigpipe(fd_, data + off, size - off);
         if (n < 0) {
-            if (errno == EINTR) continue;
+            // EAGAIN/EWOULDBLOCK: the buffer filled (peer not draining) — re-poll
+            // under the deadline, which throws if the write outlasts it.
+            if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue;
             throw std::runtime_error(std::string("RemoteL3SocketTransport: send failed: ") + std::strerror(errno));
         }
         if (n == 0) throw std::runtime_error("RemoteL3SocketTransport: socket closed while writing");
@@ -518,15 +535,15 @@ void RemoteL3SocketTransport::write_all(const uint8_t *data, size_t size) {
     }
 }
 
-std::vector<uint8_t> RemoteL3SocketTransport::read_frame() {
+std::vector<uint8_t> RemoteL3SocketTransport::read_frame(std::chrono::steady_clock::time_point deadline) {
     static constexpr size_t HEADER_BYTES = 40;
     std::vector<uint8_t> frame(HEADER_BYTES);
     size_t off = 0;
     while (off < HEADER_BYTES) {
-        wait_readable();
+        wait_readable(deadline);
         ssize_t n = ::recv(fd_, frame.data() + off, HEADER_BYTES - off, 0);
         if (n < 0) {
-            if (errno == EINTR) continue;
+            if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue;
             throw std::runtime_error(
                 std::string("RemoteL3SocketTransport: recv header failed: ") + std::strerror(errno)
             );
@@ -541,10 +558,10 @@ std::vector<uint8_t> RemoteL3SocketTransport::read_frame() {
     frame.resize(HEADER_BYTES + payload_bytes);
     off = HEADER_BYTES;
     while (off < frame.size()) {
-        wait_readable();
+        wait_readable(deadline);
         ssize_t n = ::recv(fd_, frame.data() + off, frame.size() - off, 0);
         if (n < 0) {
-            if (errno == EINTR) continue;
+            if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) continue;
             throw std::runtime_error(
                 std::string("RemoteL3SocketTransport: recv payload failed: ") + std::strerror(errno)
             );
@@ -558,7 +575,9 @@ std::vector<uint8_t> RemoteL3SocketTransport::read_frame() {
 void RemoteL3SocketTransport::expect_hello_ready(
     uint64_t session_id, int32_t worker_id, const std::string &comm_profile
 ) {
-    auto frame = remote_l3::decode_frame(read_frame());
+    // The HELLO read is an attach-phase op: it shares attach_deadline_ so the
+    // whole (possibly multi-recv) read cannot outlast the startup-budget slice.
+    auto frame = remote_l3::decode_frame(read_frame(attach_deadline_));
     if (frame.header.frame_type != remote_l3::FrameType::HELLO) {
         throw std::runtime_error("RemoteL3SocketTransport: expected HELLO frame");
     }
@@ -577,11 +596,13 @@ void RemoteL3SocketTransport::expect_hello_ready(
 
 void RemoteL3SocketTransport::submit_frame(const std::vector<uint8_t> &frame) {
     if (fd_ < 0) throw std::runtime_error("RemoteL3SocketTransport: socket is closed");
-    write_all(frame.data(), frame.size());
+    // Each runtime command gets a fresh runtime-timeout budget, independent of
+    // the (already-spent) attach deadline.
+    write_all(frame.data(), frame.size(), deadline_from_now(runtime_timeout_s_));
 }
 
 std::vector<uint8_t> RemoteL3SocketTransport::wait_for_reply(remote_l3::FrameType frame_type, uint64_t sequence) {
-    auto frame_bytes = read_frame();
+    auto frame_bytes = read_frame(deadline_from_now(runtime_timeout_s_));
     auto frame = remote_l3::decode_frame(frame_bytes);
     if (frame.header.frame_type != frame_type || frame.header.sequence != sequence) {
         throw std::runtime_error("RemoteL3SocketTransport: reply frame type or sequence mismatch");
diff --git a/src/common/hierarchical/remote_endpoint.h b/src/common/hierarchical/remote_endpoint.h
index 28ec44cb11..39fc86c2c8 100644
--- a/src/common/hierarchical/remote_endpoint.h
+++ b/src/common/hierarchical/remote_endpoint.h
@@ -14,6 +14,7 @@
 #include <memory>
 #include <mutex>
 #include <atomic>
+#include <chrono>
 #include <string>
 #include <thread>
 #include <vector>
@@ -32,7 +33,8 @@ class RemoteL3Transport {
 class RemoteL3SocketTransport : public RemoteL3Transport {
 public:
     RemoteL3SocketTransport(
-        std::string host, uint16_t port, std::string health_host, uint16_t health_port, double timeout_s
+        std::string host, uint16_t port, std::string health_host, uint16_t health_port, double attach_timeout_s,
+        double runtime_timeout_s
     );
     ~RemoteL3SocketTransport() override;
 
@@ -46,7 +48,12 @@ class RemoteL3SocketTransport : public RemoteL3Transport {
     uint16_t port_{0};
     std::string health_host_;
     uint16_t health_port_{0};
-    double timeout_s_{30.0};
+    // Startup-budget clock for the attach phase (command-connect, HELLO read,
+    // health-connect all share attach_deadline_) vs the per-command runtime
+    // timeout for post-attach frame I/O and the health-monitor loop.
+    double attach_timeout_s_{30.0};
+    double runtime_timeout_s_{30.0};
+    std::chrono::steady_clock::time_point attach_deadline_{};
     int fd_{-1};
     int health_fd_{-1};
     std::thread health_thread_;
@@ -61,10 +68,10 @@ class RemoteL3SocketTransport : public RemoteL3Transport {
     void stop_health_monitor();
     void mark_health_failed(const std::string &message);
     void check_health();
-    void wait_readable();
-    void wait_writable();
-    void write_all(const uint8_t *data, size_t size);
-    std::vector<uint8_t> read_frame();
+    void wait_readable(std::chrono::steady_clock::time_point deadline);
+    void wait_writable(std::chrono::steady_clock::time_point deadline);
+    void write_all(const uint8_t *data, size_t size, std::chrono::steady_clock::time_point deadline);
+    std::vector<uint8_t> read_frame(std::chrono::steady_clock::time_point deadline);
 };
 
 class RemoteL3Endpoint : public WorkerEndpoint {
diff --git a/src/common/hierarchical/worker.cpp b/src/common/hierarchical/worker.cpp
index aeb10414f4..23629a2862 100644
--- a/src/common/hierarchical/worker.cpp
+++ b/src/common/hierarchical/worker.cpp
@@ -81,10 +81,12 @@ void Worker::add_next_level_worker(int32_t worker_id, void *mailbox) {
 
 void Worker::add_remote_l3_socket(
     int32_t worker_id, uint64_t session_id, const std::string &transport_name, const std::string &host, uint16_t port,
-    const std::string &health_host, uint16_t health_port, double timeout_s
+    const std::string &health_host, uint16_t health_port, double attach_timeout_s, double runtime_timeout_s
 ) {
     if (initialized_) throw std::runtime_error("Worker: add_remote_l3_socket after init");
-    auto transport = std::make_unique<RemoteL3SocketTransport>(host, port, health_host, health_port, timeout_s);
+    auto transport = std::make_unique<RemoteL3SocketTransport>(
+        host, port, health_host, health_port, attach_timeout_s, runtime_timeout_s
+    );
     transport->expect_hello_ready(session_id, worker_id, transport_name);
     manager_.add_next_level_endpoint(
         std::make_unique<RemoteL3Endpoint>(worker_id, session_id, transport_name, std::move(transport))
diff --git a/src/common/hierarchical/worker.h b/src/common/hierarchical/worker.h
index bf7764286c..fd65eaa108 100644
--- a/src/common/hierarchical/worker.h
+++ b/src/common/hierarchical/worker.h
@@ -81,7 +81,8 @@ class Worker {
     // prestart and reported HELLO READY on the command lane.
     void add_remote_l3_socket(
         int32_t worker_id, uint64_t session_id, const std::string &transport_name, const std::string &host,
-        uint16_t port, const std::string &health_host, uint16_t health_port, double timeout_s
+        uint16_t port, const std::string &health_host, uint16_t health_port, double attach_timeout_s,
+        double runtime_timeout_s
     );
 
     // Start the scheduler thread. Must be called AFTER the parent has forked
diff --git a/tests/ut/cpp/hierarchical/test_remote_endpoint.cpp b/tests/ut/cpp/hierarchical/test_remote_endpoint.cpp
index ad4e3a826d..e5eddeb752 100644
--- a/tests/ut/cpp/hierarchical/test_remote_endpoint.cpp
+++ b/tests/ut/cpp/hierarchical/test_remote_endpoint.cpp
@@ -17,6 +17,7 @@
 #include <unistd.h>
 
 #include <array>
+#include <atomic>
 #include <cerrno>
 #include <csignal>
 #include <chrono>
@@ -123,6 +124,83 @@ uint16_t start_closing_server(std::thread &server_thread) {
     return ntohs(addr.sin_port);
 }
 
+int make_loopback_listener(uint16_t &port_out) {
+    int listener = ::socket(AF_INET, SOCK_STREAM, 0);
+    if (listener < 0) throw std::runtime_error(std::string("socket failed: ") + std::strerror(errno));
+    int one = 1;
+    (void)::setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = 0;
+    if (::bind(listener, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) != 0) {
+        int err = errno;
+        ::close(listener);
+        throw std::runtime_error(std::string("bind failed: ") + std::strerror(err));
+    }
+    if (::listen(listener, 1) != 0) {
+        int err = errno;
+        ::close(listener);
+        throw std::runtime_error(std::string("listen failed: ") + std::strerror(err));
+    }
+    socklen_t len = sizeof(addr);
+    if (::getsockname(listener, reinterpret_cast<sockaddr *>(&addr), &len) != 0) {
+        int err = errno;
+        ::close(listener);
+        throw std::runtime_error(std::string("getsockname failed: ") + std::strerror(err));
+    }
+    port_out = ntohs(addr.sin_port);
+    return listener;
+}
+
+// Accept one connection and hold it open (sending nothing) until `stop` is set,
+// so a client blocked in read_frame can only end by timing out on its own
+// deadline — never on EOF. Holding well past any client timeout is what lets a
+// test tell an attach-bounded read from a runtime-bounded one; a hard safety cap
+// keeps a failing test from hanging the suite.
+uint16_t start_stalling_server(std::thread &server_thread, std::atomic<bool> &stop) {
+    uint16_t port = 0;
+    int listener = make_loopback_listener(port);
+    server_thread = std::thread([listener, &stop]() {
+        int fd = ::accept(listener, nullptr, nullptr);
+        for (int i = 0; i < 500 && !stop.load(std::memory_order_acquire); ++i) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        }
+        if (fd >= 0) ::close(fd);
+        ::close(listener);
+    });
+    return port;
+}
+
+// Accept one connection, wait delay_ms, then send a single COMPLETION frame with
+// the given sequence so a client's wait_for_reply(COMPLETION, seq) succeeds.
+uint16_t start_delayed_reply_server(std::thread &server_thread, int delay_ms, uint64_t sequence) {
+    uint16_t port = 0;
+    int listener = make_loopback_listener(port);
+    server_thread = std::thread([listener, delay_ms, sequence]() {
+        int fd = ::accept(listener, nullptr, nullptr);
+        if (fd >= 0) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
+            remote_l3::FrameHeader header;
+            header.frame_type = remote_l3::FrameType::COMPLETION;
+            header.session_id = 1;
+            header.worker_id = 0;
+            header.sequence = sequence;
+            std::vector<uint8_t> frame = remote_l3::encode_frame(header, {});
+            size_t off = 0;
+            while (off < frame.size()) {
+                ssize_t n = ::send(fd, frame.data() + off, frame.size() - off, MSG_NOSIGNAL);
+                if (n <= 0) break;
+                off += static_cast<size_t>(n);
+            }
+            std::this_thread::sleep_for(std::chrono::milliseconds(50));
+            ::close(fd);
+        }
+        ::close(listener);
+    });
+    return port;
+}
+
 class FakeRemoteTransport : public RemoteL3Transport {
 public:
     int32_t next_error_code{0};
@@ -356,7 +434,7 @@ TEST(RemoteEndpoint, RemoteBufferControlsRejectOutOfRangeSlices) {
 TEST(RemoteSocketTransport, ClosedPeerWriteDoesNotRaiseSigpipe) {
     std::thread server_thread;
     uint16_t port = start_closing_server(server_thread);
-    RemoteL3SocketTransport transport("127.0.0.1", port, "127.0.0.1", 1, 1.0);
+    RemoteL3SocketTransport transport("127.0.0.1", port, "127.0.0.1", 1, 1.0, 1.0);
     server_thread.join();
     std::this_thread::sleep_for(std::chrono::milliseconds(50));
 
@@ -377,6 +455,74 @@ TEST(RemoteSocketTransport, ClosedPeerWriteDoesNotRaiseSigpipe) {
     transport.shutdown();
 }
 
+TEST(RemoteSocketTransport, CtorRejectsNonPositiveTimeouts) {
+    // Validation runs before connect_socket(), so no server is needed.
+    EXPECT_THROW(RemoteL3SocketTransport("127.0.0.1", 1, "127.0.0.1", 1, 0.0, 5.0), std::invalid_argument);
+    EXPECT_THROW(RemoteL3SocketTransport("127.0.0.1", 1, "127.0.0.1", 1, -1.0, 5.0), std::invalid_argument);
+    EXPECT_THROW(RemoteL3SocketTransport("127.0.0.1", 1, "127.0.0.1", 1, 5.0, 0.0), std::invalid_argument);
+    EXPECT_THROW(RemoteL3SocketTransport("127.0.0.1", 1, "127.0.0.1", 1, 5.0, -1.0), std::invalid_argument);
+}
+
+TEST(RemoteSocketTransport, HelloReadBoundedByAttachTimeout) {
+    // Server accepts the command connection but never sends HELLO, so the read
+    // can only end by timing out. A small attach budget (0.2s) and a large
+    // runtime budget (5.0s) tell the two apart: bounding the HELLO read by the
+    // runtime timeout would take ~5s.
+    std::atomic<bool> stop{false};
+    std::thread server_thread;
+    uint16_t port = start_stalling_server(server_thread, stop);
+    RemoteL3SocketTransport transport("127.0.0.1", port, "127.0.0.1", 1, /*attach*/ 0.2, /*runtime*/ 5.0);
+
+    auto t0 = std::chrono::steady_clock::now();
+    EXPECT_THROW(transport.expect_hello_ready(1, 0, "sim"), std::runtime_error);
+    double elapsed = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
+    EXPECT_LT(elapsed, 1.0);
+
+    stop.store(true, std::memory_order_release);
+    transport.shutdown();
+    server_thread.join();
+}
+
+TEST(RemoteSocketTransport, RuntimeReadSurvivesElapsedAttachDeadline) {
+    // The reply arrives after the attach deadline (0.3s) has elapsed. A runtime
+    // read that (wrongly) reused the attach deadline would throw immediately; a
+    // fresh runtime budget (2.0s) receives it. This proves the value split, not
+    // just the path.
+    std::thread server_thread;
+    uint16_t port = start_delayed_reply_server(server_thread, /*delay_ms=*/500, /*sequence=*/1);
+    RemoteL3SocketTransport transport("127.0.0.1", port, "127.0.0.1", 1, /*attach*/ 0.3, /*runtime*/ 2.0);
+
+    std::vector<uint8_t> probe(16, 0x11);
+    transport.submit_frame(probe);
+    EXPECT_NO_THROW({
+        auto reply = transport.wait_for_reply(remote_l3::FrameType::COMPLETION, 1);
+        EXPECT_FALSE(reply.empty());
+    });
+
+    transport.shutdown();
+    server_thread.join();
+}
+
+TEST(RemoteSocketTransport, RuntimeWriteToStalledReaderTimesOut) {
+    // The peer accepts but never reads; a large frame overruns the socket buffers
+    // so a single blocking send() would hang past the runtime deadline. The fd is
+    // persistently O_NONBLOCK, so the write re-polls under the deadline and throws.
+    std::atomic<bool> stop{false};
+    std::thread server_thread;
+    uint16_t port = start_stalling_server(server_thread, stop);
+    RemoteL3SocketTransport transport("127.0.0.1", port, "127.0.0.1", 1, /*attach*/ 1.0, /*runtime*/ 0.5);
+
+    std::vector<uint8_t> big(16 * 1024 * 1024, 0x7E);
+    auto t0 = std::chrono::steady_clock::now();
+    EXPECT_THROW(transport.submit_frame(big), std::runtime_error);
+    double elapsed = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
+    EXPECT_LT(elapsed, 3.0);
+
+    stop.store(true, std::memory_order_release);
+    transport.shutdown();
+    server_thread.join();
+}
+
 TEST(RemoteEndpoint, BareHostPointerWithoutSidecarIsEndpointFailure) {
     Ring ring;
     ring.init(1ULL << 20);
diff --git a/tests/ut/py/test_callable_identity.py b/tests/ut/py/test_callable_identity.py
index bd4c9ebecc..61e2a5cd9b 100644
--- a/tests/ut/py/test_callable_identity.py
+++ b/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, startup_remaining_s):
+    def fake_open_remote_session(self, *, spec, worker_id, session_id, deadline):
         opened_worker_ids.append(worker_id)
         return worker_mod._RemoteSession(  # noqa: SLF001
             worker_id=worker_id,
@@ -1616,7 +1616,7 @@ def fake_worker_ctor(*args):
 
     calls = 0
 
-    def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
+    def fake_open_remote_session(self, *, spec, worker_id, session_id, deadline):
         nonlocal calls
         calls += 1
         if calls == 2:
@@ -1673,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, startup_remaining_s):
+    def fake_open_remote_session(self, *, spec, worker_id, session_id, deadline):
         return opened_session
 
     def fake_close_remote_session(self, session):
diff --git a/tests/ut/py/test_remote_l3_lifecycle.py b/tests/ut/py/test_remote_l3_lifecycle.py
index b7d694fd1b..acd644cb3d 100644
--- a/tests/ut/py/test_remote_l3_lifecycle.py
+++ b/tests/ut/py/test_remote_l3_lifecycle.py
@@ -40,12 +40,91 @@ def test_read_runner_ready_times_out_without_payload():
     ready_r, ready_w = os.pipe()
     try:
         with pytest.raises(TimeoutError):
-            remote_l3_worker._read_runner_ready(ready_r, 0.01)
+            remote_l3_worker._read_runner_ready(ready_r, time.monotonic() + 0.01)
     finally:
         os.close(ready_r)
         os.close(ready_w)
 
 
+class _Clock:
+    """Deterministic monotonic clock advanced by explicit ticks."""
+
+    def __init__(self, t0: float = 1000.0):
+        self.t = t0
+
+    def monotonic(self) -> float:
+        return self.t
+
+    def advance(self, dt: float) -> None:
+        self.t += dt
+
+
+def test_start_session_hands_runner_absolute_deadline_and_single_ready_wait(monkeypatch):
+    # The daemon builds ONE absolute deadline up front; the runner-ready wait
+    # derives from it (so the daemon's own Popen/setup time is charged, not a
+    # fresh full slice), and the runner is handed that same ABSOLUTE deadline —
+    # so the runner's deadline cannot be re-amplified past the daemon's.
+    clock = _Clock(1000.0)
+    monkeypatch.setattr(remote_l3_worker.time, "monotonic", clock.monotonic)
+    captured: dict = {}
+
+    def fake_pipe():
+        clock.advance(1.0)  # daemon setup before the runner manifest is written
+        return (11, 12)
+
+    class _FakeTmp:
+        name = "/tmp/fake-remote-l3.json"
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_a):
+            return False
+
+        def write(self, *_a):
+            pass
+
+    class _FakePopen:
+        pid = 999
+
+        def __init__(self, *_a, **_k):
+            clock.advance(2.0)  # Popen + interpreter spawn charged to runner-ready
+
+        def wait(self, timeout=None):
+            return 0
+
+    class _SyncThread:
+        def __init__(self, *, target, args, daemon):
+            self._target, self._args = target, args
+
+        def start(self):
+            self._target(*self._args)
+
+    def fake_read_ready(fd, deadline):
+        captured["deadline"] = deadline
+        return {"ok": True}
+
+    monkeypatch.setattr(remote_l3_worker.os, "pipe", fake_pipe)
+    monkeypatch.setattr(remote_l3_worker.os, "close", lambda fd: None)
+    monkeypatch.setattr(remote_l3_worker.os, "unlink", lambda p: None)
+    monkeypatch.setattr(remote_l3_worker.tempfile, "NamedTemporaryFile", lambda *a, **k: _FakeTmp())
+    monkeypatch.setattr(remote_l3_worker.json, "dump", lambda obj, f, **k: captured.__setitem__("manifest", obj))
+    monkeypatch.setattr(remote_l3_worker.subprocess, "Popen", lambda *a, **k: _FakePopen())
+    monkeypatch.setattr(remote_l3_worker.threading, "Thread", _SyncThread)
+    monkeypatch.setattr(remote_l3_worker, "_read_runner_ready", fake_read_ready)
+
+    reply, _proc = remote_l3_worker._start_session(_manifest(startup_remaining_s=50.0, session_timeout_s=30.0))
+
+    # One deadline = 1000 + 50; the runner-ready wait uses it (the 2s Popen is
+    # charged), NOT a fresh now()+50 rebuilt after Popen.
+    assert captured["deadline"] == 1050.0
+    # The runner is handed that same ABSOLUTE monotonic deadline, so its own
+    # deadline equals the daemon's (spawn/import time charged) rather than a
+    # rebuilt now()+remaining that would run past it.
+    assert captured["manifest"]["startup_deadline_monotonic"] == 1050.0
+    assert reply["ok"] is True
+
+
 def test_start_session_kills_runner_on_ready_timeout(monkeypatch):
     class FakePopen:
         pid = 12345
@@ -77,14 +156,14 @@ def wait(self, timeout=None):
     def fake_popen(*args, **kwargs):
         return fake_proc
 
-    def fake_read_ready(fd, timeout_s):
+    def fake_read_ready(fd, deadline):
         raise TimeoutError("ready timeout")
 
     monkeypatch.setattr(remote_l3_worker.subprocess, "Popen", fake_popen)
     monkeypatch.setattr(remote_l3_worker, "_read_runner_ready", fake_read_ready)
 
     with pytest.raises(TimeoutError):
-        remote_l3_worker._start_session(_manifest())
+        remote_l3_worker._start_session(_manifest(startup_remaining_s=30.0))
 
     # Cooperative SIGTERM first, then the hard SIGKILL backstop.
     assert fake_proc.terminated
@@ -108,7 +187,7 @@ def wait(self, timeout=None):
     monkeypatch.setattr(remote_l3_worker.subprocess, "Popen", lambda *args, **kwargs: fake_proc)
     monkeypatch.setattr(remote_l3_worker, "_read_runner_ready", lambda fd, timeout_s: {"ok": True})
 
-    reply, proc = remote_l3_worker._start_session(_manifest())
+    reply, proc = remote_l3_worker._start_session(_manifest(startup_remaining_s=30.0))
 
     # A ready runner is handed back live for the caller to hand off or reclaim;
     # _start_session itself neither reaps nor kills it.
@@ -132,7 +211,7 @@ def wait(self, timeout=None):
     monkeypatch.setattr(remote_l3_worker, "_read_runner_ready", lambda fd, timeout_s: {"ok": False})
     monkeypatch.setattr(remote_l3_worker, "_wait_or_kill_runner", lambda p, **kw: reclaimed.append(p))
 
-    reply, proc = remote_l3_worker._start_session(_manifest())
+    reply, proc = remote_l3_worker._start_session(_manifest(startup_remaining_s=30.0))
 
     # A failed handshake is killed+reaped exactly once inside _start_session, so
     # the caller gets no runner to reclaim.
@@ -141,21 +220,25 @@ def wait(self, timeout=None):
     assert reclaimed == [fake_proc]
 
 
-def test_run_session_bounds_post_ready_command_accept(monkeypatch):
+def test_run_session_bounds_command_accept_by_startup_deadline_not_session_timeout(monkeypatch):
+    # Waiting for the parent to attach is still the attach phase: command accept
+    # must be bounded by the remaining startup budget, not session_timeout_s.
+    clock = _Clock(1000.0)
+    monkeypatch.setattr(remote_l3_session.time, "monotonic", clock.monotonic)
+
     class FakeWorker:
         def __init__(self, *args, **kwargs):
-            self.closed = False
+            pass
 
         def init(self, *args, **kwargs):
             pass
 
         def close(self):
-            self.closed = True
+            pass
 
     class FakeCommandSock:
         def __init__(self):
             self.timeout = None
-            self.closed = False
 
         def getsockname(self):
             return ("127.0.0.1", 12345)
@@ -169,7 +252,7 @@ def accept(self):
             raise socket.timeout("command attach timed out")
 
         def close(self):
-            self.closed = True
+            pass
 
     class FakeHealthSock:
         def getsockname(self):
@@ -189,11 +272,92 @@ def close(self):
     monkeypatch.setattr(remote_l3_session, "_health_loop", lambda *args: None)
 
     try:
-        assert remote_l3_session.run_session(_manifest(), ready_w) == 1
-        assert command_sock.timeout == 0.01
+        # 5s startup budget (deadline 1005), 30s runtime timeout — deliberately different.
+        rc = remote_l3_session.run_session(
+            _manifest(startup_deadline_monotonic=1005.0, session_timeout_s=30.0), ready_w
+        )
+        assert rc == 1
+    finally:
+        os.close(ready_r)
+
+    # accept timeout = startup_deadline - now() = 5.0, NOT the 30s session timeout.
+    assert command_sock.timeout == 5.0
+
+
+def test_run_session_forces_command_conn_blocking_for_idle(monkeypatch):
+    # A finite timeout on the command connection would self-destruct a healthy but
+    # idle session (read_frame idle-waiting for the next command would raise). The
+    # accept()'d socket inherits socket.getdefaulttimeout() (a user module could
+    # have set it), so the runner must explicitly force it blocking.
+    clock = _Clock(1000.0)
+    monkeypatch.setattr(remote_l3_session.time, "monotonic", clock.monotonic)
+
+    class FakeConn:
+        def __init__(self):
+            self.timeout = 0.05  # hostile: a non-None default timeout was inherited
+
+        def settimeout(self, t):
+            self.timeout = t
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_a):
+            return False
+
+    fake_conn = FakeConn()
+
+    class FakeWorker:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def init(self, *args, **kwargs):
+            pass
+
+        def close(self):
+            pass
+
+    class FakeCommandSock:
+        def getsockname(self):
+            return ("127.0.0.1", 12345)
+
+        def settimeout(self, timeout):
+            pass
+
+        def accept(self):
+            return fake_conn, ("127.0.0.1", 1)
+
+        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)
+    monkeypatch.setattr(remote_l3_session, "_run_command_loop", lambda *args, **kwargs: None)
+
+    try:
+        rc = remote_l3_session.run_session(
+            _manifest(startup_deadline_monotonic=1005.0, session_timeout_s=0.05), ready_w
+        )
+        assert rc == 0
     finally:
         os.close(ready_r)
 
+    # Forced blocking before the command loop, regardless of the inherited default.
+    assert fake_conn.timeout is None
+
 
 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
@@ -252,6 +416,126 @@ def close(self):
     assert 40.0 < budget <= 50.0
 
 
+def test_run_session_builds_deadline_before_registry_install(monkeypatch):
+    # The runner establishes its single deadline before registry install, so that
+    # time is charged against the budget rather than restarting a fresh slice at
+    # inner init.
+    clock = _Clock(1000.0)
+    monkeypatch.setattr(remote_l3_session.time, "monotonic", clock.monotonic)
+    captured: dict = {}
+
+    class FakeWorker:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def init(self, *args, _startup_deadline=None, **kwargs):
+            captured["deadline"] = _startup_deadline
+
+        def close(self):
+            pass
+
+    def slow_dispatch_registry(manifest):
+        clock.advance(5.0)  # registry install consumes the budget
+        return {}
+
+    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", slow_dispatch_registry)
+    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)
+
+    # deadline built at t=1000 (before the 5s registry install) => 1050, NOT the
+    # 1005 + 50 = 1055 a post-registry rebuild would give.
+    assert captured["deadline"] == 1050.0
+
+
+def test_run_session_uses_daemon_absolute_deadline_verbatim(monkeypatch):
+    # Given the daemon's shared-clock absolute deadline, the runner uses it as-is
+    # — its deadline equals the daemon's regardless of spawn/import/registry time.
+    clock = _Clock(1000.0)
+    monkeypatch.setattr(remote_l3_session.time, "monotonic", clock.monotonic)
+    captured: dict = {}
+
+    class FakeWorker:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def init(self, *args, _startup_deadline=None, **kwargs):
+            captured["deadline"] = _startup_deadline
+
+        def close(self):
+            pass
+
+    def slow_dispatch_registry(manifest):
+        clock.advance(7.0)  # spawn/import/registry cost — must NOT extend the deadline
+        return {}
+
+    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", slow_dispatch_registry)
+    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(startup_deadline_monotonic=1042.0, session_timeout_s=0.01), ready_w)
+    finally:
+        os.close(ready_r)
+
+    # Uses the daemon's absolute deadline verbatim (1042), not now()+something.
+    assert captured["deadline"] == 1042.0
+
+
 def test_health_loop_closes_active_connection_on_stop():
     stop = threading.Event()
 
diff --git a/tests/ut/py/test_worker/test_remote_startup_budget.py b/tests/ut/py/test_worker/test_remote_startup_budget.py
index a92a83cc13..69535e668c 100644
--- a/tests/ut/py/test_worker/test_remote_startup_budget.py
+++ b/tests/ut/py/test_worker/test_remote_startup_budget.py
@@ -22,6 +22,7 @@
 
 from __future__ import annotations
 
+import socket
 import time
 from unittest.mock import MagicMock
 
@@ -43,6 +44,20 @@ def _l4_with_remotes(n: int, **config) -> Worker:
     return w
 
 
+class _FakeClock:
+    """Deterministic monotonic clock; advance it by explicit ticks so every
+    derived ``deadline - now()`` value is exact (no real-sleep magnitude races)."""
+
+    def __init__(self, t0: float = 1000.0):
+        self.t = t0
+
+    def monotonic(self) -> float:
+        return self.t
+
+    def advance(self, dt: float) -> None:
+        self.t += dt
+
+
 class TestManifestBudgetField:
     def test_manifest_carries_startup_remaining_distinct_from_session_timeout(self):
         w = Worker(level=4, num_sub_workers=0, remote_session_timeout_s=30.0)
@@ -52,6 +67,11 @@ def test_manifest_carries_startup_remaining_distinct_from_session_timeout(self):
             assert manifest["session_timeout_s"] == 30.0
             # The startup budget must not be fixed to the runtime command timeout.
             assert manifest["startup_remaining_s"] != manifest["session_timeout_s"]
+            # The cross-host wire carries a duration, never an absolute deadline
+            # (monotonic clocks are not comparable across machines).
+            assert isinstance(manifest["startup_remaining_s"], float)
+            assert "startup_deadline" not in manifest
+            assert "deadline" not in manifest
         finally:
             w.close()
 
@@ -74,6 +94,29 @@ def test_positive_finite_remote_session_timeout_accepted(self):
             w.close()
 
 
+class TestNumericEndpointContract:
+    """Remote L3 endpoints are numeric-only (or localhost); a hostname is rejected
+    at add_remote_worker time, before any startup resource exists."""
+
+    def test_add_remote_worker_rejects_hostname_at_registration(self):
+        w = Worker(level=4, num_sub_workers=0)
+        try:
+            with pytest.raises(ValueError, match="numeric IP"):
+                w.add_remote_worker(RemoteWorkerSpec(endpoint="node17:19073", platform="a2a3sim"))
+            assert w._remote_worker_specs == []  # rejected before it is registered
+        finally:
+            w.close()
+
+    def test_add_remote_worker_accepts_numeric_and_localhost(self):
+        w = Worker(level=4, num_sub_workers=0)
+        try:
+            w.add_remote_worker(RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"))
+            w.add_remote_worker(RemoteWorkerSpec(endpoint="localhost:19074", platform="a2a3sim"))
+            assert len(w._remote_worker_specs) == 2
+        finally:
+            w.close()
+
+
 class TestRemoteSideValidation:
     """Both remote entry points validate the wire budget before creating resources."""
 
@@ -235,46 +278,89 @@ def _must_not_open(self, **_kwargs):
 
 
 class TestSingleRootBudget:
-    def _drive_activation(self, monkeypatch, *, n_remotes, root_budget_s, per_open_delay_s=0.02):
-        granted: list[float] = []
-        socket_timeouts: list[float] = []
-
-        def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
-            granted.append(startup_remaining_s)
-            socket_timeouts.append(timeout_s)
-            time.sleep(per_open_delay_s)
+    def _drive_activation(self, monkeypatch, *, n_remotes, root_budget_s, per_open_tick_s=0.5):
+        clock = _FakeClock()
+        monkeypatch.setattr(worker_mod.time, "monotonic", clock.monotonic)
+
+        deadlines: list[float] = []
+        granted: list[float] = []  # the startup_remaining_s the real open would derive
+
+        def fake_open(self, *, spec, worker_id, session_id, deadline):
+            deadlines.append(deadline)
+            granted.append(deadline - worker_mod.time.monotonic())
+            clock.advance(per_open_tick_s)  # models per-remote open cost, deterministically
             return _FakeSession(worker_id, session_id)
 
         monkeypatch.setattr(Worker, "_open_remote_session", fake_open)
 
         w = _l4_with_remotes(n_remotes)
-        w._worker = MagicMock()
-        deadline = time.monotonic() + root_budget_s
+        mock_worker = MagicMock()
+        w._worker = mock_worker
+        deadline = clock.monotonic() + root_budget_s
         try:
             w._activate_remote_sessions(deadline)
         finally:
             w._worker = None
             w.close()
-        return granted, socket_timeouts, w
+        return deadlines, granted, mock_worker
 
     def test_budget_not_multiplied_by_remote_count(self, monkeypatch):
-        granted, socket_timeouts, _ = self._drive_activation(monkeypatch, n_remotes=3, root_budget_s=5.0)
+        deadlines, granted, _ = self._drive_activation(monkeypatch, n_remotes=3, root_budget_s=5.0)
         assert len(granted) == 3
-        # Every remote's granted startup budget fits inside the single root budget.
+        # THE anti-multiplication invariant: one shared absolute root deadline is
+        # threaded to every remote — the bug would surface as distinct, freshly
+        # computed deadlines per remote.
+        assert len(set(deadlines)) == 1
+        # Each remote's derived startup slice fits inside the single root budget…
         for g in granted:
             assert 0 < g <= 5.0
-        # And the budget shrinks per remote (shared deadline), never a fresh full
-        # timeout each — the multiplication bug.
+        # …and shrinks per remote (shared deadline + advancing clock).
         assert granted[0] > granted[1] > granted[2]
-        # Socket handshake timeout tracks the same remaining budget.
-        for s, g in zip(socket_timeouts, granted, strict=True):
-            assert s == g
 
-    def test_attach_called_once_per_remote(self, monkeypatch):
-        granted: list[float] = []
+    def test_attach_and_runtime_timeouts_are_split(self, monkeypatch):
+        _deadlines, _granted, mock_worker = self._drive_activation(monkeypatch, n_remotes=3, root_budget_s=5.0)
+        calls = mock_worker.add_remote_l3_socket.call_args_list
+        assert len(calls) == 3
+        # attach_timeout / runtime_timeout are the last two positional args.
+        attach = [c.args[-2] for c in calls]
+        runtime = [c.args[-1] for c in calls]
+        # runtime_timeout is the full runtime command budget: constant, ==
+        # session_timeout, NOT min(session_timeout, remaining), never shrinking.
+        assert runtime == [30.0, 30.0, 30.0]
+        # attach_timeout tracks the remaining root budget: shrinks, bounded, and
+        # is a distinct quantity from runtime_timeout (the split the old min() lost).
+        assert attach[0] > attach[1] > attach[2]
+        for a in attach:
+            assert 0 < a <= 5.0
+            assert a != 30.0
+
+    def test_final_recheck_fires_when_last_attach_overruns_deadline(self, monkeypatch):
+        clock = _FakeClock()
+        monkeypatch.setattr(worker_mod.time, "monotonic", clock.monotonic)
+
+        def fake_open(self, *, spec, worker_id, session_id, deadline):
+            clock.advance(0.4)  # open leaves budget > 0, so both per-iteration rechecks pass
+            return _FakeSession(worker_id, session_id)
 
-        def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
-            granted.append(startup_remaining_s)
+        monkeypatch.setattr(Worker, "_open_remote_session", fake_open)
+        w = _l4_with_remotes(1)
+        mock_worker = MagicMock()
+        # The last attach itself consumes the remaining slice, so now >= deadline
+        # only AFTER add_remote_l3_socket returns — only a post-loop recheck catches it.
+        mock_worker.add_remote_l3_socket.side_effect = lambda *a, **k: clock.advance(1.0)
+        w._worker = mock_worker
+        deadline = clock.monotonic() + 1.0  # 0.4 (open) + 1.0 (attach) > 1.0
+        try:
+            with pytest.raises(RuntimeError, match="startup deadline exceeded after attach"):
+                w._activate_remote_sessions(deadline)
+            assert mock_worker.add_remote_l3_socket.call_count == 1  # attach did run
+            assert len(w._remote_sessions) == 1  # left recorded so init()'s rollback closes it
+        finally:
+            w._worker = None
+            w.close()
+
+    def test_attach_called_once_per_remote(self, monkeypatch):
+        def fake_open(self, *, spec, worker_id, session_id, deadline):
             return _FakeSession(worker_id, session_id)
 
         monkeypatch.setattr(Worker, "_open_remote_session", fake_open)
@@ -292,7 +378,7 @@ def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining
     def test_expired_deadline_fails_fast_without_opening(self, monkeypatch):
         opened = []
 
-        def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s):
+        def fake_open(self, *, spec, worker_id, session_id, deadline):
             opened.append(worker_id)
             return _FakeSession(worker_id, session_id)
 
@@ -309,6 +395,167 @@ def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining
             w.close()
 
 
+class TestReadyCommitGate:
+    """The final root-deadline gate lives in init()'s READY-commit critical
+    section, so a thread descheduled after attach cannot publish READY late."""
+
+    def test_gate_fires_inside_commit_before_publishing_ready(self, monkeypatch):
+        clock = _FakeClock()
+        monkeypatch.setattr(worker_mod.time, "monotonic", clock.monotonic)
+
+        def fake_start(self):
+            # Time crosses the deadline during post-startup work. No remote
+            # session on THIS worker — the gate must still fire (a local child
+            # could have remote descendants this deadline also bounds).
+            clock.advance(self._startup_timeout_s + 1.0)
+
+        monkeypatch.setattr(Worker, "_init_hierarchical", lambda self: None)
+        monkeypatch.setattr(Worker, "_start_hierarchical", fake_start)
+        monkeypatch.setattr(Worker, "_cleanup_partial_init", lambda self: None)
+
+        w = Worker(level=4, num_sub_workers=0)
+        try:
+            with pytest.raises(RuntimeError, match="startup deadline exceeded before READY"):
+                w.init()
+            assert w._lifecycle is not worker_mod._Lifecycle.READY
+        finally:
+            w._worker = None
+            w.close()
+
+
+class TestBoundedConnect:
+    """Resolution and every per-address connect are bounded by the single root
+    deadline (unlike socket.create_connection, which restarts the clock per
+    address and never bounds getaddrinfo)."""
+
+    def test_resolve_fails_fast_past_deadline_without_calling_getaddrinfo(self, monkeypatch):
+        gai = MagicMock(side_effect=AssertionError("getaddrinfo on an already-past deadline"))
+        monkeypatch.setattr(worker_mod.socket, "getaddrinfo", gai)
+        with pytest.raises(TimeoutError, match="startup deadline exceeded"):
+            Worker._resolve_within_deadline("localhost", 9, time.monotonic() - 1.0)
+        assert gai.call_count == 0
+
+    def test_resolve_rejects_hostname_instead_of_risking_unbounded_dns(self):
+        # A hostname would need an unbounded, uncancellable getaddrinfo; reject it
+        # outright rather than let a hung resolver pin init() in INITIALIZING.
+        with pytest.raises(ValueError, match="must be a numeric IP"):
+            Worker._resolve_within_deadline("example.invalid", 9, time.monotonic() + 5.0)
+
+    def test_resolve_accepts_numeric_and_localhost(self):
+        for host in ("127.0.0.1", "localhost"):
+            infos = Worker._resolve_within_deadline(host, 9, time.monotonic() + 5.0)
+            assert infos
+            assert infos[0][4][0] == "127.0.0.1"
+
+    def test_resolve_never_issues_a_dns_query(self, monkeypatch):
+        # AI_NUMERICHOST must be set so getaddrinfo cannot block on NSS/DNS.
+        calls = []
+        real = worker_mod.socket.getaddrinfo
+
+        def spy(host, port, **kwargs):
+            calls.append(kwargs.get("flags", 0))
+            return real(host, port, **kwargs)
+
+        monkeypatch.setattr(worker_mod.socket, "getaddrinfo", spy)
+        Worker._resolve_within_deadline("127.0.0.1", 9, time.monotonic() + 5.0)
+        assert calls and (calls[0] & socket.AI_NUMERICHOST)
+
+    def test_connect_bounds_each_address_by_shrinking_remaining(self, monkeypatch):
+        clock = _FakeClock()
+        monkeypatch.setattr(worker_mod.time, "monotonic", clock.monotonic)
+        addr_a = (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.1", 9))
+        addr_b = (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.2", 9))
+        monkeypatch.setattr(Worker, "_resolve_within_deadline", lambda h, p, d: [addr_a, addr_b])
+
+        timeouts: list[float] = []
+        made: list = []
+
+        class _FakeSock:
+            def settimeout(self, t):
+                timeouts.append(t)
+
+            def connect(self, sockaddr):
+                clock.advance(3.0)  # each connect attempt burns budget
+                if sockaddr == addr_a[4]:
+                    raise ConnectionRefusedError("first address black-holed")
+
+            def close(self):
+                pass
+
+        def fake_socket(*_a):
+            s = _FakeSock()
+            made.append(s)
+            return s
+
+        monkeypatch.setattr(worker_mod.socket, "socket", fake_socket)
+        sock = Worker._connect_within_deadline("host", 9, clock.monotonic() + 10.0)
+
+        # The second address gets the SHRUNKEN remaining (10 - 3), not a fresh full
+        # 10s — the per-address multiplication create_connection would allow.
+        assert timeouts == [10.0, 7.0]
+        assert sock is made[1]
+
+    def test_connect_fails_fast_when_deadline_passes_before_next_address(self, monkeypatch):
+        addr = (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.1", 9))
+        monkeypatch.setattr(Worker, "_resolve_within_deadline", lambda h, p, d: [addr])
+        made = MagicMock(side_effect=AssertionError("socket created past deadline"))
+        monkeypatch.setattr(worker_mod.socket, "socket", made)
+        with pytest.raises(TimeoutError, match="startup deadline exceeded"):
+            Worker._connect_within_deadline("h", 9, time.monotonic() - 1.0)
+        assert made.call_count == 0
+
+
+class TestOpenRemoteSession:
+    """The single-deadline contract inside _open_remote_session itself."""
+
+    def test_fails_fast_on_past_deadline_without_connecting(self, monkeypatch):
+        # Build the worker first: add_remote_worker validates the (numeric) host
+        # via getaddrinfo, which we then poison to assert the connect path is skipped.
+        w = _l4_with_remotes(1)
+        gai = MagicMock(side_effect=AssertionError("getaddrinfo on an already-past deadline"))
+        mksock = MagicMock(side_effect=AssertionError("socket() on an already-past deadline"))
+        monkeypatch.setattr(worker_mod.socket, "getaddrinfo", gai)
+        monkeypatch.setattr(worker_mod.socket, "socket", mksock)
+        try:
+            with pytest.raises(TimeoutError, match="startup deadline exceeded"):
+                w._open_remote_session(spec=_spec(), worker_id=0, session_id=1, deadline=time.monotonic() - 1.0)
+            assert gai.call_count == 0
+            assert mksock.call_count == 0
+        finally:
+            w.close()
+
+    def test_recomputes_budget_before_send_and_refuses_nonpositive_slice(self, monkeypatch):
+        clock = _FakeClock()
+        monkeypatch.setattr(worker_mod.time, "monotonic", clock.monotonic)
+        sent: list = []
+
+        class _FakeSock:
+            def settimeout(self, _t):
+                pass
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_a):
+                return False
+
+        def fake_connect(self, host, port, deadline):
+            clock.advance(10.0)  # resolve + connect consume the entire budget
+            return _FakeSock()
+
+        monkeypatch.setattr(Worker, "_connect_within_deadline", fake_connect)
+        monkeypatch.setattr(Worker, "_send_remote_daemon_json", lambda self, sock, payload: sent.append(payload))
+        w = _l4_with_remotes(1)
+        try:
+            # connect succeeds within budget, but by send time the budget is gone,
+            # so it raises before sending a <= 0 startup_remaining_s the remote bounces.
+            with pytest.raises(TimeoutError, match="startup deadline exceeded"):
+                w._open_remote_session(spec=_spec(), worker_id=0, session_id=1, deadline=clock.monotonic() + 1.0)
+            assert sent == []
+        finally:
+            w.close()
+
+
 class _FakeSession:
     """Minimal stand-in for _RemoteSession that add_remote_l3_socket can read."""