Fix: isolate Remote L3 daemon per-connection send failures

python/simpler/remote_l3_worker.py

@@ -141,7 +141,14 @@ def _reap_session_runner(proc: subprocess.Popen[Any]) -> None:
         pass
 
 
-def _start_session(manifest: dict[str, Any]) -> dict[str, Any]:
+def _start_session(manifest: dict[str, Any]) -> tuple[dict[str, Any], subprocess.Popen[Any] | None]:
+    # Returns (reply, proc). proc is the live, ready runner handle when the reply
+    # is ok — the caller owns it: hand it to a background reaper once the reply
+    # send to the parent succeeds, or reclaim it if the send raises. proc is None
+    # when no runner survives (a failed handshake has already been killed and
+    # reaped here), so a failed send then leaves nothing to reclaim. A successful
+    # send only means the bytes were queued locally, not that the parent read
+    # them; unobserved receipt would need an ACK / lease, which sim does not have.
     _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
@@ -185,9 +192,8 @@ def _start_session(manifest: dict[str, Any]) -> dict[str, Any]:
         ready["pid"] = int(proc.pid)
         if not ready.get("ok", False):
             _wait_or_kill_runner(proc)
-        else:
-            threading.Thread(target=_reap_session_runner, args=(proc,), daemon=True).start()
-        return ready
+            return ready, None
+        return ready, proc
     finally:
         if ready_w >= 0:
             try:
@@ -205,22 +211,63 @@ def _start_session(manifest: dict[str, Any]) -> dict[str, Any]:
                 pass
 
 
+def _serve_connection(conn: socket.socket) -> None:
+    # One parent connection, fully isolated: any Exception on this socket affects
+    # only this session, never the shared accept loop. KeyboardInterrupt /
+    # SystemExit are BaseException, not Exception, so they propagate out for a
+    # clean daemon shutdown instead of turning into a spurious error reply. A live
+    # runner is disposed of exactly once on every exit path — handed to the reaper
+    # only once its reply send returns, otherwise reclaimed by the finally — so no
+    # session runner is orphaned (until its own timeout) or double-reaped, even
+    # when the send fails, the reaper cannot be launched, or a control exception
+    # unwinds mid-connection.
+    proc: subprocess.Popen[Any] | None = None
+    try:
+        try:
+            manifest = _read_json(conn)
+            reply, proc = _start_session(manifest)
+        except Exception as exc:  # noqa: BLE001
+            # Handshake failed (bad manifest, runner start error); _start_session
+            # already reaped any runner it started. Best-effort error reply,
+            # swallowing only a send that itself fails on a dead parent.
+            with contextlib.suppress(OSError):
+                _send_json(conn, {"ok": False, "error": f"{type(exc).__name__}: {exc}"})
+            return
+        try:
+            _send_json(conn, reply)
+            if proc is not None:
+                threading.Thread(target=_reap_session_runner, args=(proc,), daemon=True).start()
+                proc = None
+        except Exception:  # noqa: BLE001
+            # Local send raised (parent gone) or the reaper thread could not be
+            # started; the finally reclaims the still-owned runner.
+            return
+    finally:
+        if proc is not None:
+            _wait_or_kill_runner(proc)
+
+
+def _serve_loop(server: socket.socket) -> None:
+    while True:
+        try:
+            conn, _addr = server.accept()
+        except OSError:
+            # Listener closed (shutdown); stop accepting.
+            return
+        with conn:
+            _serve_connection(conn)
+
+
 def serve(host: str, port: int) -> int:
     server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     server.bind((host, port))
     server.listen()
     try:
-        while True:
-            conn, _addr = server.accept()
-            with conn:
-                try:
-                    manifest = _read_json(conn)
-                    _send_json(conn, _start_session(manifest))
-                except BaseException as exc:  # noqa: BLE001
-                    _send_json(conn, {"ok": False, "error": f"{type(exc).__name__}: {exc}"})
+        _serve_loop(server)
     finally:
         server.close()
+    return 0
 
 
 def main(argv: list[str] | None = None) -> int: