diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index 1d9778ac0f8e..a9ea279fce8c 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -2376,3 +2376,135 @@ class TestHandleProcessRedaction: monkeypatch.setattr(pr, "process_registry", reg) out = json.loads(pr._handle_process({"action": "log", "session_id": sess.id})) assert "zzzopaque1234567890abcdef" in out["output"] + + +# ========================================================================= +# Reader loop: orphaned grandchild holding the stdout pipe (issue #68915) +# ========================================================================= + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only: select() on pipes") +class TestReaderLoopOrphanedPipe: + """Regression tests for issue #68915. + + When an agent command backgrounds a long-lived process (``node server.js + &``), the grandchild inherits the write end of the reader's stdout pipe. + The direct bash child exits, but the pipe never EOFs — the old blocking + ``read1()`` parked the reader thread forever, ``session.exited`` never + flipped on its own, and ``notify_on_complete`` never fired. The reader + must instead terminate shortly after the direct child exits, even while + a descendant still holds the pipe open. + """ + + def test_reader_exits_when_orphan_holds_pipe(self, registry): + """Reader loop must return promptly after the direct child exits, + even though a backgrounded descendant keeps the pipe open.""" + proc = subprocess.Popen( + ["sh", "-c", "echo started; sleep 30 & exit 0"], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + preexec_fn=os.setsid, + ) + s = _make_session(sid="proc_orphan_reader") + s.process = proc + s.pid = proc.pid + registry._running[s.id] = s + + done = threading.Event() + + def _run(): + registry._reader_loop(s) + done.set() + + t = threading.Thread(target=_run, daemon=True) + t.start() + try: + # The direct child exits immediately; the reader must notice and + # return well before the 30s descendant releases the pipe. + assert done.wait(timeout=10.0), ( + "_reader_loop is still blocked on the orphan-held pipe " + "(issue #68915) — session.exited would never flip and " + "notify_on_complete would never fire" + ) + assert s.exited is True + assert s.exit_code == 0 + assert s.completion_reason == "exited" + assert "started" in s.output_buffer + assert s.id in registry._finished + finally: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + except (ProcessLookupError, PermissionError): + pass + + def test_reader_exit_fires_notify_on_complete(self, registry): + """The autonomous completion notification must not depend on a + poll()/wait() call when an orphan holds the pipe.""" + proc = subprocess.Popen( + ["sh", "-c", "sleep 30 & echo bg-started"], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + preexec_fn=os.setsid, + ) + s = _make_session(sid="proc_orphan_notify") + s.process = proc + s.pid = proc.pid + s.notify_on_complete = True + registry._running[s.id] = s + + done = threading.Event() + + def _run(): + registry._reader_loop(s) + done.set() + + t = threading.Thread(target=_run, daemon=True) + t.start() + try: + assert done.wait(timeout=10.0), ( + "_reader_loop blocked — completion notification lost (#68915)" + ) + # Exactly one completion event must have been queued. + item = registry.completion_queue.get_nowait() + assert item["type"] == "completion" + assert item["session_id"] == s.id + assert item["exit_code"] == 0 + finally: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + except (ProcessLookupError, PermissionError): + pass + + def test_reader_still_streams_full_output_to_eof(self, registry): + """No-orphan case: the reader must still capture ALL output through + true EOF (the early-exit path must not race away buffered tail).""" + script = ( + "for i in 1 2 3 4 5; do echo line-$i; done; " + "sleep 0.3; echo tail-after-sleep" + ) + proc = subprocess.Popen( + ["sh", "-c", script], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + preexec_fn=os.setsid, + ) + s = _make_session(sid="proc_orphan_fulldrain") + s.process = proc + s.pid = proc.pid + registry._running[s.id] = s + + registry._reader_loop(s) + + assert s.exited is True + assert s.exit_code == 0 + for i in range(1, 6): + assert f"line-{i}" in s.output_buffer + assert "tail-after-sleep" in s.output_buffer diff --git a/tools/process_registry.py b/tools/process_registry.py index aed6be8933f5..7daf74b2e336 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -943,36 +943,96 @@ class ProcessRegistry: block until EOF (or a large buffer fills), which makes "live" output land in one burst at process exit. ``buffer.read1(4096)`` yields incremental chunks as bytes become available, then we decode to text. + + Orphaned-pipe guard (issue #68915): when the user's command backgrounds + a long-lived process (``node server.js &``, ``sleep 300 &``), that + grandchild inherits the write end of our stdout pipe via ``fork()``. + The direct ``bash`` child exits promptly, but the pipe never reaches + EOF while the grandchild lives — so a blocking read would park this + thread forever, ``session.exited`` would never flip, and + ``notify_on_complete`` would never fire (``_reconcile_local_exit`` + only runs lazily from poll()/wait(), which an autonomous notification + can't rely on). On POSIX we therefore ``select()`` with a short poll + interval and stop draining shortly after the direct child exits, even + if the pipe hasn't EOF'd — mirroring the foreground fix in + ``tools/environments/base.py::_wait_for_process`` (#8340). Windows + pipes don't support select(); the blocking path is kept there and the + lazy reconcile in poll()/wait() remains the safety net. """ first_chunk = True + + def _append_chunk(chunk: str): + nonlocal first_chunk + if first_chunk: + chunk = self._clean_shell_noise(chunk) + first_chunk = False + with session._lock: + session.output_buffer += chunk + if len(session.output_buffer) > session.max_output_chars: + session.output_buffer = session.output_buffer[-session.max_output_chars:] + self._check_watch_patterns(session, chunk) + self._emit_output(session, chunk) + try: - stdout = session.process.stdout - if stdout is None: + proc = session.process + if proc is None or proc.stdout is None: return + stdout = proc.stdout raw_read = getattr(getattr(stdout, "buffer", None), "read1", None) - while True: - if raw_read is not None: - raw = raw_read(4096) - if not raw: - break - chunk = raw.decode("utf-8", errors="replace") - else: - # Fallback for mocked/alternate streams without a buffered raw - # interface. This may be less "live", but keeps compatibility. - chunk = stdout.read(4096) - if not chunk: - break - if first_chunk: - chunk = self._clean_shell_noise(chunk) - first_chunk = False - with session._lock: - session.output_buffer += chunk - if len(session.output_buffer) > session.max_output_chars: - session.output_buffer = session.output_buffer[-session.max_output_chars:] - self._check_watch_patterns(session, chunk) - self._emit_output(session, chunk) + # Resolve a real OS fd for the select() path. Mocked streams + # (unit tests, adapters) may lack fileno() — fall back to the + # historical blocking loop for those. + fd = None + if raw_read is not None and not _IS_WINDOWS: + fileno = getattr(stdout, "fileno", None) + try: + candidate = fileno() if callable(fileno) else None + except Exception: + candidate = None + if isinstance(candidate, int) and candidate >= 0: + fd = candidate + + if fd is not None: + import select as _select + + idle_after_exit = 0 + while True: + try: + ready, _, _ = _select.select([fd], [], [], 0.2) + except (ValueError, OSError): + break # fd already closed + if ready: + raw = raw_read(4096) + if not raw: + break # true EOF — all writers closed + _append_chunk(raw.decode("utf-8", errors="replace")) + idle_after_exit = 0 + elif proc.poll() is not None: + # Direct child is gone and the pipe was idle for + # ~200ms. Give it a few more cycles to catch any + # buffered tail, then stop — otherwise we would wait + # forever on a pipe held open by an orphaned + # grandchild (issue #68915). + idle_after_exit += 1 + if idle_after_exit >= 3: + break + else: + while True: + if raw_read is not None: + raw = raw_read(4096) + if not raw: + break + chunk = raw.decode("utf-8", errors="replace") + else: + # Fallback for mocked/alternate streams without a buffered raw + # interface. This may be less "live", but keeps compatibility. + chunk = stdout.read(4096) + if not chunk: + break + + _append_chunk(chunk) except Exception as e: logger.debug("Process stdout reader ended: %s", e) finally: