fix(process_registry): reader loop no longer hangs when an orphaned grandchild holds the stdout pipe

When a background terminal() command backgrounds its own long-lived
child (`node server.js &`, `sleep 300 &`), the grandchild inherits the
write end of the reader thread's stdout pipe. The direct bash child
exits promptly, but the pipe never reaches EOF while the grandchild
lives — so `_reader_loop`'s blocking `read1()` parked the thread
forever, `session.exited` never flipped on its own, and
`notify_on_complete` was silently lost. `_reconcile_local_exit`
(#17327) only runs lazily from poll()/wait(), so nothing autonomous
ever surfaced the exit; each occurrence also leaked a reader thread
and pipe fd for the grandchild's lifetime.

Fix: on POSIX, drain via select() with a short poll interval and stop
shortly after the direct child exits even if the pipe hasn't EOF'd —
the same pattern the foreground path uses in
tools/environments/base.py::_wait_for_process (#8340). Windows pipes
don't support select(), so the blocking path is kept there with the
existing lazy reconcile as the safety net; mocked/iterator stdout
streams (no usable fileno) also keep the historical path.

Fixes #68915

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

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: