diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index e4369f0fda..9ead6f0445 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -703,6 +703,99 @@ class TestPruning: assert total <= MAX_PROCESSES +class TestFinishedHandleRelease: + """Finished sessions must release their Popen/PTY OS handles immediately. + + Regression for the "file descriptor limit" symptom: a finished-but- + unpruned session previously kept its Popen stdout pipe (or PTY master) + FD open until the finished-process TTL (FINISHED_TTL_SECONDS) elapsed. + Under heavy background churn the gateway could exhaust its FD limit even + though the registry never rejects spawns (it prunes oldest-finished at + MAX_PROCESSES instead) — the symptom was a retained-handle leak, not a + registry-cap rejection. poll()/wait()/read_log() serve from the buffered + output_buffer, never from the pipe, so closing the handles at finish is + lossless. + """ + + def test_move_to_finished_closes_popen_pipes(self, registry): + proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(0.2)"], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, + ) + session = _make_session(sid="proc_handle_close", exited=False) + session.process = proc + registry._running[session.id] = session + + assert proc.stdout is not None + assert not proc.stdout.closed + + # Simulate the reader loop finishing (process exits, EOF drained). + proc.wait(timeout=5) + session.exited = True + session.exit_code = proc.returncode + session.completion_reason = "exited" + registry._move_to_finished(session) + + assert session.id in registry._finished + assert proc.stdout.closed, "finished session must release its stdout pipe FD" # type: ignore[union-attr] + + def test_move_to_finished_closes_pty(self, registry): + """PTY-backed sessions release the PTY master on finish too.""" + pty_closed = {"closed": False} + + class _FakePty: + def close(self): + pty_closed["closed"] = True + + session = _make_session(sid="proc_pty_close", exited=True) + session._pty = _FakePty() + registry._finished[session.id] = session + + registry._move_to_finished(session) + assert pty_closed["closed"] + + def test_move_to_finished_safe_without_handles(self, registry): + """Env-backed / detached sessions have no local Popen or PTY; the + release must be a no-op, not a crash.""" + session = _make_session(sid="proc_no_handles", exited=True) + registry._finished[session.id] = session + registry._move_to_finished(session) # must not raise + assert session.id in registry._finished + + def test_poll_still_serves_output_after_handle_release(self, registry): + """Output remains queryable after the pipes close — poll() reads the + buffered output, never the (now-closed) pipe.""" + proc = subprocess.Popen( + [sys.executable, "-c", "print('hello-finish'); import time; time.sleep(0.2)"], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, + text=True, + ) + session = _make_session(sid="proc_poll_after_close", exited=False) + session.process = proc + registry._running[session.id] = session + + # Drain output like the reader loop would. + proc.wait(timeout=5) + try: + tail = proc.stdout.read() if proc.stdout else "" + except ValueError: + tail = "" + session.output_buffer = tail or "" + session.exited = True + session.exit_code = proc.returncode + session.completion_reason = "exited" + registry._move_to_finished(session) + + assert proc.stdout.closed + result = registry.poll("proc_poll_after_close") + assert result["status"] == "exited" + assert "hello-finish" in result["output_preview"] + + # ========================================================================= # Spawn env sanitization # ========================================================================= diff --git a/tools/process_registry.py b/tools/process_registry.py index 224e2df371..8a86f0a01e 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1335,6 +1335,16 @@ class ProcessRegistry(ProcessCheckpointMixin): save_completed_result(session) self._running.pop(session.id) self._finished[session.id] = session + # The reader thread has drained the pipe at this point (EOF reached + # before _move_to_finished runs). Release the retained Popen/PTY + # handles now so finished sessions stop holding OS file descriptors — + # otherwise every finished-but-unpruned session keeps its stdout pipe + # (or PTY master) FD open until the finished-process TTL elapses, and + # heavy background churn can exhaust the gateway's FD limit. + # poll()/wait()/read_log() serve output from the buffered + # ``output_buffer``, never from the pipe, so closing the handles here + # is lossless. + self._release_finished_handles(session) self._write_checkpoint() if was_running and session.notify_on_complete: notification = { @@ -1363,6 +1373,31 @@ class ProcessRegistry(ProcessCheckpointMixin): "termination_source": session.termination_source, } + def _release_finished_handles(self, session: ProcessSession): + """Close a finished session's OS handles (Popen pipes / PTY master). + + Best-effort and idempotent: the session may have no local Popen (env + backends, detached recovery), or the handles may already be closed by + the reader loop / kill path. Closing a Popen's stream objects does not + kill anything — the child has already exited — it only releases the + parent's pipe FDs, which is exactly the retained-resource leak. + """ + proc = getattr(session, "process", None) + if proc is not None: + for stream_name in ("stdout", "stderr", "stdin"): + stream = getattr(proc, stream_name, None) + if stream is not None: + try: + stream.close() + except Exception: + pass + pty = getattr(session, "_pty", None) + if pty is not None: + try: + pty.close() + except Exception: + pass + # ----- Query Methods ----- def is_completion_consumed(self, session_id: str) -> bool: