From 8dc0fe2429973515f38b1f748986caf69194416e Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Sun, 2 Aug 2026 00:42:13 -0700 Subject: [PATCH 1/2] fix: dedicate a non-blocking fd for terminal input to stop TTY-reader freezes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: prompt_toolkit's PosixStdinReader.read() does a non-atomic select-then-read on fd 0 (blocking, shared with the parent shell and any inherited children). If a competing reader (ssh without -n, a digital-twin exec, a nested amplifier invocation) drains the pending byte between the readiness check and the os.read() call, the read blocks forever -- ON THE EVENT LOOP THREAD, since it's registered via loop.add_reader(). This wedged real sessions for hours (21 confirmed freeze events across 19,028 session logs; one bash tool call with a 35s timeout returned 4h04m later), always resolving the instant Enter was pressed. Fix: amplifier_app_cli/dedicated_tty_input.py opens /dev/tty with a fresh O_RDONLY | O_NONBLOCK file description (never touching fd 0's own OFD, which would leak O_NONBLOCK to the parent shell) and builds a prompt_toolkit Vt100Input on it. PosixStdinReader.read() already swallows BlockingIOError (a subclass of OSError) from a non-blocking read, so the exact race degrades to an empty read instead of a hang -- verified directly, including a deterministic (non-probabilistic) reproduction of the select-says-ready-but-nothing-to-read race via a monkeypatched select. A fail-loud guard (_assert_posix_stdin_reader_degrades_nonblocking_reads) verifies that core prompt_toolkit assumption at construction time and raises, naming the installed version, if a future release changes it. Falls back to None (prompt_toolkit's own default, zero UX change) whenever the dedicated fd isn't available: Windows, stdin not a tty, or no controlling terminal / /dev/tty unopenable. Wired into both PromptSession construction sites that read fd 0 by default: the main REPL prompt (_create_prompt_session in main.py) and the steering prompt active during agent turns (SteeringInputManager.run in steering_input.py) -- the latter is the actual reader active when a bash tool spawns a competing TTY reader, matching the confirmed freeze scenario. Both share one process-wide dedicated fd (get_dedicated_tty_input), closed once during session teardown (close_dedicated_tty_input) so nothing leaks across sessions or spawned sub-sessions. Tests: tests/test_dedicated_tty_input.py -- attaches a real pty to fd 0 and proves (1) the dedicated fd is distinct and non-blocking, (2) fd 0's own OFD is left untouched (the leak-guard the fix specifically avoids), (3) a real os.read() on the dedicated fd with nothing pending raises BlockingIOError instead of blocking, (4) PosixStdinReader.read() degrades that race to '' deterministically, (5-7) graceful fallback for non-tty stdin, unopenable /dev/tty, and Windows, and (8) the fail-loud version guard. Full existing suite: 1248 passed (14 pre-existing failures, confirmed identical with and without this change via git stash comparison), 0 regressions. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- amplifier_app_cli/dedicated_tty_input.py | 310 +++++++++++++++++++++++ amplifier_app_cli/main.py | 11 + amplifier_app_cli/steering_input.py | 15 +- tests/test_dedicated_tty_input.py | 240 ++++++++++++++++++ 4 files changed, 575 insertions(+), 1 deletion(-) create mode 100644 amplifier_app_cli/dedicated_tty_input.py create mode 100644 tests/test_dedicated_tty_input.py diff --git a/amplifier_app_cli/dedicated_tty_input.py b/amplifier_app_cli/dedicated_tty_input.py new file mode 100644 index 00000000..03a2e815 --- /dev/null +++ b/amplifier_app_cli/dedicated_tty_input.py @@ -0,0 +1,310 @@ +"""Dedicated, non-blocking terminal input for prompt_toolkit -- stops a +competing TTY reader from freezing the CLI's asyncio event loop. + +CONFIRMED BUG: real sessions froze completely for hours (one bash tool call +configured with a 35s timeout returned "Command timed out after 35 seconds" +4h04m later). 21 distinct freeze events across 19,028 real session logs, each +showing total process silence -- zero events anywhere in the project -- for +the entire gap. The freeze always ends the instant the user presses Enter. + +Root cause, proven by a loop-thread stack captured mid-stall:: + + prompt_toolkit/input/posix_utils.py:87 in read <-- os.read(fd0, 1024) BLOCKED + prompt_toolkit/input/vt100.py:98 in read_keys + prompt_toolkit/application/application.py:686 in read_from_input + asyncio/events.py:89 in _run + asyncio/base_events.py:2050 in _run_once + asyncio/base_events.py:683 in run_forever + +``PosixStdinReader.read()`` (``prompt_toolkit/input/posix_utils.py``) does a +NON-ATOMIC select-then-read on a blocking fd:: + + if not select.select([self.stdin_fd], [], [], 0)[0]: # readiness check + return "" + ... + data = os.read(self.stdin_fd, count) # BLOCKS + +fd 0 has no ``O_NONBLOCK``; the tty is raw with ``VMIN=1``. If any other +process (``ssh`` without ``-n``, a digital-twin ``exec``, a nested +``amplifier`` invocation -- all observed in real sessions) consumes the +pending byte in the gap between the ``select()`` readiness check and the +``os.read()`` call, ``os.read`` parks forever. This is registered via +``loop.add_reader`` (``prompt_toolkit/input/vt100.py``), so it runs ON THE +EVENT LOOP THREAD -- ``run_forever`` cannot advance, and no asyncio timer, +task, or callback (including an unrelated bash tool's own +``asyncio.wait_for(...)`` timeout) can fire until the blocked read returns. + +THE FIX: give prompt_toolkit its own dedicated file description for +terminal input, opened with ``O_NONBLOCK``, instead of sharing fd 0. +``PosixStdinReader.read()`` already wraps its ``os.read()`` call in +``except OSError: data = b""`` (the "In case of SIGWINCH" branch). +``BlockingIOError`` is a subclass of ``OSError``. So on a non-blocking fd, +the exact race this module exists to close degrades to an empty read +instead of a hang -- verified directly in +``tests/test_dedicated_tty_input.py`` (including a deterministic +reproduction of the select-says-ready-but-read-finds-nothing race, with no +timing dependence). + +CRITICAL DETAIL -- why this does NOT just call +``os.set_blocking(0, False)``: ``O_NONBLOCK`` lives on the *open file +description* (OFD), which fd 0 shares with the parent shell and every +inherited child process. Setting it there would leak out of this process +and could break the user's shell (or any inherited child) after exit. +Instead, ``os.open("/dev/tty", os.O_RDONLY | os.O_NONBLOCK)`` creates a +FRESH OFD -- the flag is private to this new fd and cannot leak onto fd 0. +(As a bonus, Python's ``os.open()`` also makes the new fd non-inheritable +by default since PEP 446, so it isn't handed to bash-tool children either.) + +WHY CONSTRUCTION OVER MONKEYPATCHING (contrast with +``stdout_offload.py``'s scoped monkeypatch of ``run_in_terminal``): here, +prompt_toolkit already exposes a fully PUBLIC, documented seam for this -- +``Input`` objects are constructed from a plain ``TextIO``-like object and +handed to ``PromptSession(input=...)`` / ``Application(input=...)``. No +private name needs to be patched at all. Building a ``Vt100Input`` on our +own dedicated stream is simpler and more robust than intercepting an +internal call chain, so that is what this module does. The one true +internal dependency -- that ``PosixStdinReader.read()`` swallows +``BlockingIOError`` -- is guarded by +``_assert_posix_stdin_reader_degrades_nonblocking_reads()`` below, which +fails loud (naming the installed prompt_toolkit version) if that +assumption ever stops holding, rather than silently reintroducing the +freeze. + +GRACEFUL FALLBACK: ``open_dedicated_tty_input()`` returns ``None`` -- +never raises -- whenever the dedicated fd isn't available or applicable: +no controlling terminal, ``/dev/tty`` cannot be opened (CI, containers), +stdin is not a tty (piped input, non-interactive use), or Windows (no +``/dev/tty`` / POSIX ``Vt100Input`` concept there). Callers pass the +result straight through as ``PromptSession(input=...)``; ``None`` is +exactly prompt_toolkit's own default, so this is a transparent, safe +drop-in with zero UX change when the dedicated fd isn't available. +""" + +from __future__ import annotations + +import os +import sys +import threading +from collections.abc import Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from prompt_toolkit.input.base import Input + +# Seam for tests: production code always opens the process's controlling +# terminal. Tests point this at a real pty slave's own path instead, since +# constructing a genuine controlling-terminal setup (setsid + TIOCSCTTY) +# isn't available inside a pytest worker. +_TTY_DEVICE_PATH = "/dev/tty" + +__all__ = [ + "DedicatedTtyInput", + "close_dedicated_tty_input", + "get_dedicated_tty_input", + "open_dedicated_tty_input", +] + + +@dataclass +class DedicatedTtyInput: + """A prompt_toolkit ``Input`` built on its own non-blocking fd, plus a + ``close()`` to release that fd deterministically (no leak across + sessions or spawned sub-sessions). + """ + + input: Input + close: Callable[[], None] + + +def _assert_posix_stdin_reader_degrades_nonblocking_reads() -> None: + """Fail loud if ``PosixStdinReader.read()`` no longer swallows a + non-blocking read's ``BlockingIOError`` and instead lets it propagate. + + This is the one internal prompt_toolkit behavior this whole fix + depends on (see module docstring). Verified here with a real + ``os.pipe()`` and a monkeypatched ``select`` that unconditionally + reports the fd ready -- deterministically reproducing the exact + select-says-ready-but-nothing-to-read race a competing reader can + cause, with no timing dependence and no real tty required. + + Raises ``RuntimeError`` naming the installed prompt_toolkit version if + the assumption no longer holds, rather than silently constructing a + dedicated input that can hang exactly like the bug this module fixes. + """ + import prompt_toolkit + from prompt_toolkit.input import posix_utils as pt_posix_utils + + installed_version = prompt_toolkit.__version__ + + read_fd, write_fd = os.pipe() + try: + os.set_blocking(read_fd, False) + reader = pt_posix_utils.PosixStdinReader(read_fd, encoding="utf-8") + + class _AlwaysReadyFakeSelect: + """Stand-in for the ``select`` module: reports the fd ready + unconditionally, forcing the race deterministically instead + of relying on real (probabilistic) scheduling timing. + """ + + @staticmethod + def select(rlist, wlist, xlist, timeout): + return (rlist, [], []) + + # `select` is a plain module-level name inside prompt_toolkit's + # posix_utils module (imported there via `import select`), not part + # of its public `__all__` -- hence getattr/setattr (pyright + # suppressed) instead of attribute access, matching + # stdout_offload.py's own convention for touching internal names. + real_select = getattr( # noqa: B009 - pyright: ignore[reportPrivateImportUsage] + pt_posix_utils, "select" + ) + setattr( # noqa: B010 - intentional scoped patch, restored in finally + pt_posix_utils, "select", _AlwaysReadyFakeSelect + ) + try: + result = reader.read() + except OSError as exc: + raise RuntimeError( + "amplifier_app_cli.dedicated_tty_input: prompt_toolkit " + f"(=={installed_version})'s PosixStdinReader.read() no longer " + f"swallows a non-blocking read's BlockingIOError ({exc!r}). " + "This fix depends on that behavior to make a dedicated " + "non-blocking input fd degrade a raced read to empty instead " + "of propagating into the event loop. Update " + "amplifier_app_cli/dedicated_tty_input.py or pin " + "prompt_toolkit to a known-compatible version." + ) from exc + finally: + setattr(pt_posix_utils, "select", real_select) # noqa: B010 + + if result != "": + raise RuntimeError( + "amplifier_app_cli.dedicated_tty_input: prompt_toolkit " + f"(=={installed_version})'s PosixStdinReader.read() on an " + f"empty non-blocking pipe returned {result!r} instead of " + "'' -- the assumption this fix depends on no longer holds. " + "Update amplifier_app_cli/dedicated_tty_input.py or pin " + "prompt_toolkit to a known-compatible version." + ) + finally: + os.close(write_fd) + os.close(read_fd) + + +def open_dedicated_tty_input() -> DedicatedTtyInput | None: + """Build a prompt_toolkit ``Input`` on a fresh, non-blocking fd opened + against the controlling terminal, instead of sharing fd 0. + + Returns ``None`` (never raises) whenever the dedicated fd isn't + available or applicable -- Windows, stdin not a tty, no controlling + terminal, or ``/dev/tty`` otherwise unopenable. Callers pass the + result straight through as ``PromptSession(input=...)`` / + ``Application(input=...)``: ``None`` is prompt_toolkit's own default, + so falling back is a transparent, zero-UX-change no-op. + """ + if sys.platform == "win32": + # No /dev/tty, no POSIX Vt100Input on Windows -- fall back to + # prompt_toolkit's own default (Win32Input on sys.stdin). + return None + + try: + # Checked against fd 0 directly (``os.isatty(0)``) rather than + # ``sys.stdin.isatty()`` -- the fd is the thing that actually + # matters here (it's what a competing reader races against), and + # ``sys.stdin`` can be swapped for an unrelated object by test + # runners, logging wrappers, etc. without changing fd 0 at all. + if not os.isatty(0): + # Piped/redirected stdin (CI, non-interactive use, tests): + # nothing to dedicate a fd for -- fall back to the default. + return None + except OSError: + # Defensive: fd 0 could be closed entirely in some embeddings. + # Fall back rather than risk constructing something broken. + return None + + try: + fd = os.open(_TTY_DEVICE_PATH, os.O_RDONLY | os.O_NONBLOCK) + except OSError: + # No controlling terminal (containers, detached processes) or + # /dev/tty otherwise unopenable -- fall back to the default. + return None + + try: + _assert_posix_stdin_reader_degrades_nonblocking_reads() + + # Wrap the fd in a text file object -- Vt100Input needs + # .fileno(), .isatty(), and .encoding. closefd=True (the default) + # means closing this file object closes the underlying fd. + stream = os.fdopen(fd, "r", encoding="utf-8", closefd=True) + except BaseException: + # Anything going wrong past this point (including the fail-loud + # guard above) must not leak the fd we just opened. + os.close(fd) + raise + + try: + from prompt_toolkit.input.vt100 import Vt100Input + + pt_input = Vt100Input(stream) + except Exception: # noqa: BLE001 - intentional: any construction failure must fall back, not crash the CLI + # Construction failed for some reason not anticipated above + # (e.g. a future prompt_toolkit release changing Vt100Input's + # constructor contract) -- fall back rather than crash the CLI. + stream.close() + return None + + return DedicatedTtyInput(input=pt_input, close=stream.close) + + +# --------------------------------------------------------------------------- +# Process-wide singleton +# +# Every PromptSession the CLI creates (the main REPL prompt, and a fresh +# SteeringInputManager prompt each turn) should read the terminal through +# the SAME dedicated fd rather than each opening -- and leaking -- its own. +# Lazily created on first use, explicitly torn down via +# ``close_dedicated_tty_input()`` in the CLI's session-teardown path (no fd +# leak across sessions or spawned sub-sessions). +# --------------------------------------------------------------------------- + +_singleton_lock = threading.Lock() +_singleton: DedicatedTtyInput | None = None +_singleton_attempted = False + + +def get_dedicated_tty_input() -> Input | None: + """Return the process-wide dedicated terminal ``Input``, creating it on + first call. Returns ``None`` (matching prompt_toolkit's own default) + whenever a dedicated fd isn't available -- see + ``open_dedicated_tty_input()`` for the fallback conditions. + + Safe to call repeatedly and from multiple call sites (the main REPL + prompt, a fresh steering prompt each turn): construction is attempted + at most once per process; every caller shares the same fd. + """ + global _singleton, _singleton_attempted + + with _singleton_lock: + if not _singleton_attempted: + _singleton_attempted = True + _singleton = open_dedicated_tty_input() + return _singleton.input if _singleton is not None else None + + +def close_dedicated_tty_input() -> None: + """Close the process-wide dedicated terminal fd, if one was created. + + Idempotent: safe to call even if ``get_dedicated_tty_input()`` was + never called, or already closed. Call this once during session + teardown -- never leave the dedicated fd open past the session that + opened it. + """ + global _singleton, _singleton_attempted + + with _singleton_lock: + if _singleton is not None: + _singleton.close() + _singleton = None + _singleton_attempted = False diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index 613729ef..520c094e 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -46,6 +46,7 @@ from .commands.update import update as update_cmd from .commands.version import version as version_cmd from .console import Markdown, console +from .dedicated_tty_input import close_dedicated_tty_input, get_dedicated_tty_input from .effective_config import get_effective_config_summary from .key_manager import KeyManager from .session_runner import SessionConfig, create_initialized_session @@ -2775,6 +2776,12 @@ def get_prompt(): # including mid-word wraps -- requiring manual cleanup after paste. prompt_continuation="", enable_history_search=True, # Enables Ctrl-R + # Dedicated, non-blocking fd instead of sharing fd 0 -- stops a + # competing TTY reader (ssh, digital-twin exec, a nested amplifier + # invocation) from freezing this event loop mid-read. Returns None + # (prompt_toolkit's own default) whenever a dedicated fd isn't + # available; see dedicated_tty_input.py for the full mechanism. + input=get_dedicated_tty_input(), ) @@ -3696,6 +3703,10 @@ def _goal_sigint_handler(signum, frame): # session:end is emitted by session.cleanup() (the canonical kernel path). # Do NOT emit it explicitly here — that would duplicate the event. await initialized.cleanup() + # Close the dedicated terminal-input fd (see dedicated_tty_input.py) + # opened for this session's PromptSessions -- no fd leak past this + # session's teardown. + close_dedicated_tty_input() # --- cleanup:finally_end (after cleanup so its duration is visible) --- if hooks: await hooks.emit(CLEANUP_FINALLY_END, {"session_id": actual_session_id}) diff --git a/amplifier_app_cli/steering_input.py b/amplifier_app_cli/steering_input.py index 03394383..9e7be34d 100644 --- a/amplifier_app_cli/steering_input.py +++ b/amplifier_app_cli/steering_input.py @@ -399,13 +399,26 @@ async def run(self) -> None: if self._input_provider is None: from prompt_toolkit import PromptSession + from .dedicated_tty_input import get_dedicated_tty_input + # interrupt_exception=_CtrlCInterrupt: replace the default # KeyboardInterrupt with a plain Exception subclass so that asyncio's # Task.__step_run_and_handle_result() does NOT re-raise after storing # the exception. The re-raise only applies to KeyboardInterrupt / # SystemExit (CPython 3.11+) and would otherwise propagate through # Handle._run() → _run_once() → asyncio.run(), crashing the process. - self._pt_session = PromptSession(interrupt_exception=_CtrlCInterrupt) + # + # input=get_dedicated_tty_input(): this steering prompt is the + # ACTIVE reader during agent turns -- exactly when a bash tool + # can spawn a competing TTY reader (ssh, digital-twin exec, a + # nested amplifier invocation). A dedicated, non-blocking fd + # stops that competing read from freezing this event loop + # (see dedicated_tty_input.py). Returns None (prompt_toolkit's + # own default) whenever a dedicated fd isn't available. + self._pt_session = PromptSession( + interrupt_exception=_CtrlCInterrupt, + input=get_dedicated_tty_input(), + ) # Pass a callable so prompt_toolkit re-evaluates the message on each # app.invalidate() call, keeping the queued count live in the prompt. _message: Any = self._prompt_message diff --git a/tests/test_dedicated_tty_input.py b/tests/test_dedicated_tty_input.py new file mode 100644 index 00000000..aa451815 --- /dev/null +++ b/tests/test_dedicated_tty_input.py @@ -0,0 +1,240 @@ +"""Regression tests for ``amplifier_app_cli.dedicated_tty_input``. + +Proves the mechanism that stops a competing TTY reader from freezing the +event loop: prompt_toolkit's default input path shares fd 0 with the +parent shell and any inherited children. If another process (ssh, a +digital-twin `exec`, a nested `amplifier` invocation) drains a byte off +that shared fd between prompt_toolkit's readiness check (``select.select``) +and its actual read (``os.read``), the read blocks forever on the +event-loop thread -- see the module docstring in +``amplifier_app_cli/dedicated_tty_input.py`` for the full mechanism. + +These tests attach a REAL pty to fd 0 (mirroring an actual interactive +session) and verify: + +1. The dedicated input's fd is distinct from fd 0 (its own open file + description, not a share of fd 0's). +2. The dedicated fd privately carries ``O_NONBLOCK``. +3. fd 0's own open file description does NOT carry ``O_NONBLOCK`` -- + the regression guard against the "leaked O_NONBLOCK onto the shared + fd" trap the fix must avoid (see module docstring). +4. A real, non-blocking ``os.read()`` on the dedicated fd with nothing + pending does not block -- it raises ``BlockingIOError`` immediately + instead of hanging (the OS-level guarantee the whole fix rests on). +5. prompt_toolkit's own ``PosixStdinReader.read()`` -- the exact call + path in the confirmed freeze stack trace -- degrades that same + ``BlockingIOError`` to an empty string instead of propagating it, + reproducing the select-says-ready-but-read-finds-nothing race + deterministically (no timing dependence, no forking, no real second + reader process). + +Fallback behavior (no controlling terminal, stdin not a tty, Windows, +``/dev/tty`` unavailable) is covered separately and does not require a +real pty. +""" + +from __future__ import annotations + +import fcntl +import os +import pty +import sys + +import pytest +from amplifier_app_cli import dedicated_tty_input as dti + + +def _is_nonblocking(fd: int) -> bool: + flags = fcntl.fcntl(fd, fcntl.F_GETFL) + return bool(flags & os.O_NONBLOCK) + + +@pytest.fixture +def pty_on_stdin(monkeypatch): + """Attach a real pty's slave side to fd 0, saving/restoring the original. + + Also points ``dedicated_tty_input``'s tty-device seam at the pty + slave's own path so production code (which normally opens + ``/dev/tty``) opens a REAL, controllable device in the test instead + of requiring a genuine controlling-terminal setup (``setsid`` + + ``TIOCSCTTY``), which pytest's own process doesn't have. + """ + master_fd, slave_fd = pty.openpty() + slave_path = os.ttyname(slave_fd) + + saved_stdin_fd = os.dup(0) + os.dup2(slave_fd, 0) + monkeypatch.setattr(dti, "_TTY_DEVICE_PATH", slave_path) + + try: + yield master_fd, slave_fd, slave_path + finally: + os.dup2(saved_stdin_fd, 0) + os.close(saved_stdin_fd) + os.close(slave_fd) + os.close(master_fd) + + +def test_dedicated_fd_is_distinct_and_nonblocking_without_leaking_to_fd0( + pty_on_stdin, +): + """Core regression guard: dedicated fd is separate + non-blocking; + fd 0 itself is left completely untouched. + """ + handle = dti.open_dedicated_tty_input() + assert handle is not None, ( + "Expected a dedicated input to be created against the test pty" + ) + + try: + dedicated_fd = handle.input.fileno() + + assert dedicated_fd != 0, "Dedicated input must use its own fd, not share fd 0" + + assert _is_nonblocking(dedicated_fd) is True, ( + "Dedicated fd must carry O_NONBLOCK on its own open file description" + ) + + assert _is_nonblocking(0) is False, ( + "fd 0's open file description must be untouched -- setting " + "O_NONBLOCK there would leak out to the parent shell and any " + "inherited children (the OFD-leak trap)" + ) + finally: + handle.close() + + +def test_dedicated_fd_read_does_not_block_with_nothing_pending(pty_on_stdin): + """Direct OS-level guarantee the whole fix rests on: a real ``os.read()`` + on the dedicated fd, with nothing written to the pty, must not block -- + it must raise ``BlockingIOError`` immediately. + """ + handle = dti.open_dedicated_tty_input() + assert handle is not None + + try: + dedicated_fd = handle.input.fileno() + with pytest.raises(BlockingIOError): + os.read(dedicated_fd, 1024) + finally: + handle.close() + + +def test_posix_stdin_reader_degrades_race_to_empty_read(pty_on_stdin): + """Reproduces the CONFIRMED freeze mechanism's exact call path + deterministically: ``select.select`` reports the fd ready, but by the + time ``os.read`` executes there is nothing left (a competing reader + drained it in between). On the dedicated non-blocking fd, + prompt_toolkit's own ``PosixStdinReader.read()`` -- the function at + the top of the confirmed freeze stack trace -- must degrade this to + an empty string rather than raising or blocking. + """ + from prompt_toolkit.input import posix_utils as pt_posix_utils + + handle = dti.open_dedicated_tty_input() + assert handle is not None + + try: + dedicated_fd = handle.input.fileno() + reader = pt_posix_utils.PosixStdinReader(dedicated_fd, encoding="utf-8") + + class _AlwaysReadyFakeSelect: + """Stand-in for the ``select`` module: reports the fd ready + unconditionally, forcing the exact race prompt_toolkit's real + ``select.select(..., timeout=0)`` guard is meant to prevent + but occasionally loses to under real scheduling timing. + """ + + @staticmethod + def select(rlist, wlist, xlist, timeout): + return (rlist, [], []) + + real_select = getattr( # noqa: B009 - pyright: ignore[reportPrivateImportUsage] + pt_posix_utils, "select" + ) + setattr(pt_posix_utils, "select", _AlwaysReadyFakeSelect) # noqa: B010 + try: + result = reader.read() + finally: + setattr(pt_posix_utils, "select", real_select) # noqa: B010 + + assert result == "", ( + "PosixStdinReader.read() must degrade a would-block race to an " + f"empty string, not raise or hang. Got: {result!r}" + ) + finally: + handle.close() + + +# --------------------------------------------------------------------------- +# Fallback behavior -- no real pty required. +# --------------------------------------------------------------------------- + + +def test_open_dedicated_tty_input_returns_none_when_stdin_not_a_tty(monkeypatch): + """Piped/redirected stdin (CI, non-interactive use): must gracefully + fall back to ``None`` rather than raising. + + Checked against fd 0 directly (``os.isatty(0)``, what the production + code actually calls) rather than ``sys.stdin.isatty()`` -- the latter + is unreliable under pytest, which replaces ``sys.stdin`` with a + capture stand-in whose ``isatty()`` doesn't reflect fd 0's real state. + """ + monkeypatch.setattr(dti.os, "isatty", lambda fd: False) + assert dti.open_dedicated_tty_input() is None + + +def test_open_dedicated_tty_input_returns_none_when_tty_device_unopenable( + monkeypatch, +): + """No controlling terminal / ``/dev/tty`` unavailable (containers): + must gracefully fall back to ``None`` rather than raising. + """ + monkeypatch.setattr(sys.stdin, "isatty", lambda: True, raising=False) + + def _raise_open(*args, **kwargs): + raise OSError("simulated: no controlling terminal") + + monkeypatch.setattr(dti.os, "open", _raise_open) + assert dti.open_dedicated_tty_input() is None + + +def test_open_dedicated_tty_input_returns_none_on_windows(monkeypatch): + """Windows has no ``/dev/tty`` / POSIX ``Vt100Input`` concept: must + gracefully fall back to ``None``. + """ + monkeypatch.setattr(dti.sys, "platform", "win32") + assert dti.open_dedicated_tty_input() is None + + +# --------------------------------------------------------------------------- +# Fail-loud guard -- prompt_toolkit internals this fix depends on. +# --------------------------------------------------------------------------- + + +def test_assert_guard_raises_if_posix_stdin_reader_stops_degrading_would_block(): + """If a future prompt_toolkit release stops swallowing non-blocking + read errors in ``PosixStdinReader.read()``, the fix's core assumption + is broken. The guard must fail loud (naming the installed version) + instead of silently reintroducing the freeze. + """ + import prompt_toolkit.input.posix_utils as pt_posix_utils + + class _BrokenPosixStdinReader: + def __init__(self, fd, encoding): + pass + + def read(self, count=1024): + raise BlockingIOError("simulated: no longer swallowed") + + original = pt_posix_utils.PosixStdinReader + pt_posix_utils.PosixStdinReader = _BrokenPosixStdinReader + try: + with pytest.raises(RuntimeError) as exc_info: + dti._assert_posix_stdin_reader_degrades_nonblocking_reads() + finally: + pt_posix_utils.PosixStdinReader = original + + import prompt_toolkit + + assert prompt_toolkit.__version__ in str(exc_info.value) From 531973cc8c125849dfccae50a914fb2c37f210db Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Sun, 2 Aug 2026 00:53:10 -0700 Subject: [PATCH 2/2] fix: close dedicated TTY input fd on interactive_chat teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit interactive_chat()'s finally block called initialized.cleanup() but never close_dedicated_tty_input() -- even though interactive_chat is the REPL path that actually opens the dedicated fd (via _create_prompt_session() and each turn's SteeringInputManager prompt). execute_single() had the close call wired but never opens the fd in the first place, so the close was backwards relative to the freeze scenario this fix targets. Add close_dedicated_tty_input() to interactive_chat()'s finally block, mirroring the existing call in execute_single(). The call is idempotent and safe even when the fd was never opened (e.g. non-tty stdin, no controlling terminal). Failing-test-first evidence: tests/test_interactive_chat_tty_teardown.py fails against pre-fix code (close_dedicated_tty_input never called from interactive_chat's finally block) and passes once the call is added. Full suite: 1252 passed (1248 baseline + 4 new), same 14 pre-existing failures, no new regressions. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- amplifier_app_cli/main.py | 6 + tests/test_interactive_chat_tty_teardown.py | 294 ++++++++++++++++++++ 2 files changed, 300 insertions(+) create mode 100644 tests/test_interactive_chat_tty_teardown.py diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index 520c094e..067681c8 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -3327,6 +3327,12 @@ def sigint_handler(signum, frame): # session:end is emitted by session.cleanup() (the canonical kernel path). # Do NOT emit it here — that would duplicate the event. await initialized.cleanup() + # Close the dedicated terminal-input fd (see dedicated_tty_input.py) -- + # this is the REPL path that actually opens it (via + # _create_prompt_session() and each turn's SteeringInputManager + # prompt), so it must be the path that closes it too. Idempotent and + # safe even if the fd was never opened. + close_dedicated_tty_input() # --- cleanup:finally_end (after cleanup so its duration is visible) --- if hooks: await hooks.emit(CLEANUP_FINALLY_END, {"session_id": actual_session_id}) diff --git a/tests/test_interactive_chat_tty_teardown.py b/tests/test_interactive_chat_tty_teardown.py new file mode 100644 index 00000000..b133ca52 --- /dev/null +++ b/tests/test_interactive_chat_tty_teardown.py @@ -0,0 +1,294 @@ +"""Regression test: interactive_chat() must close the dedicated TTY input fd. + +Background +~~~~~~~~~~ +``dedicated_tty_input.py`` opens a process-wide, dedicated non-blocking fd +against ``/dev/tty`` so prompt_toolkit never races a competing reader on fd 0 +(see that module's docstring for the full freeze mechanism). The contract is: +whoever opens the fd (via ``get_dedicated_tty_input()``) must close it again +via ``close_dedicated_tty_input()`` once the session that opened it tears +down -- otherwise the fd leaks across sessions and spawned sub-sessions. + +``execute_single()``'s ``finally`` block already calls +``close_dedicated_tty_input()`` (main.py:3706-3709) -- but that code path +never *opens* the dedicated fd in the first place (it doesn't build a +``PromptSession``). ``interactive_chat()`` is the REPL that DOES build +``PromptSession``s (both the main prompt via ``_create_prompt_session()`` +and a fresh steering prompt each turn via ``SteeringInputManager``) -- i.e. +it is the path that actually opens the dedicated fd, and the exact scenario +where the freeze this fix targets was observed. Its ``finally`` block was +missing the matching close call. + +RED phase: these tests FAIL on the pre-fix code because +``interactive_chat()``'s ``finally`` block never calls +``close_dedicated_tty_input()``. + +GREEN phase: once the call is added (mirroring ``execute_single()``), both +tests pass. +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +_MODULE = "amplifier_app_cli.main" + + +# --------------------------------------------------------------------------- +# Helpers (mirrors test_always_render_final_response.py / test_session_lifecycle_events.py) +# --------------------------------------------------------------------------- + + +def _make_mock_session() -> MagicMock: + mock_ctx = MagicMock() + mock_ctx.get_messages = AsyncMock(return_value=[]) + + def _coordinator_get(key: str): + if key == "context": + return mock_ctx + if key == "providers": + return {} + return None # hooks=None -> no hook emits + + session = MagicMock() + session.session_id = "test-session-id" + session.execute = AsyncMock(return_value="Hello!") + session.coordinator = MagicMock() + session.coordinator.get = _coordinator_get + session.coordinator.cancellation = MagicMock() + session.coordinator.cancellation.is_cancelled = False + session.coordinator.cancellation.is_immediate = False + session.coordinator.session_state = {} + session.config = {} + return session + + +def _make_initialized(session: MagicMock) -> MagicMock: + mock = MagicMock() + mock.session = session + mock.session_id = "test-session-id" + mock.configurator = None + mock.cleanup = AsyncMock() + return mock + + +def _run_interactive_chat(tmp_path: Path, mock_close: MagicMock) -> None: + """Run interactive_chat() through a single EOFError-terminated turn, + with dedicated-tty-input's close patched so we can observe teardown.""" + from amplifier_app_cli.main import interactive_chat + + session = _make_mock_session() + initialized = _make_initialized(session) + + mock_ps = MagicMock() + mock_ps.prompt_async = AsyncMock(side_effect=EOFError) + + with ( + patch( + f"{_MODULE}.create_initialized_session", + new=AsyncMock(return_value=initialized), + ), + patch(f"{_MODULE}._create_prompt_session", return_value=mock_ps), + patch("amplifier_app_cli.incremental_save.register_incremental_save"), + patch(f"{_MODULE}.SessionStore") as MockStore, + patch(f"{_MODULE}.console"), + patch( + f"{_MODULE}.process_runtime_mentions", + new=AsyncMock(side_effect=lambda s, t: t), + ), + patch(f"{_MODULE}.get_effective_config_summary"), + patch(f"{_MODULE}.close_dedicated_tty_input", new=mock_close), + ): + store_instance = MockStore.return_value + store_instance.get_metadata.return_value = {} + store_instance.save.return_value = None + + import asyncio + + asyncio.get_event_loop().run_until_complete( + interactive_chat( + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + ) + ) + + +class TestInteractiveChatClosesDedicatedTtyOnTeardown: + """interactive_chat()'s finally block must close the dedicated TTY fd. + + Regression guard for the gap left by PR #14 / the initial + dedicated-tty-input fix: execute_single() closed it, but + interactive_chat() -- the path that actually opens it -- did not. + """ + + @pytest.mark.asyncio + async def test_close_dedicated_tty_input_called_on_normal_exit( + self, tmp_path: Path + ): + """close_dedicated_tty_input() must be called exactly once when the + REPL loop exits normally (EOFError, e.g. Ctrl+D).""" + from amplifier_app_cli.main import interactive_chat + + session = _make_mock_session() + initialized = _make_initialized(session) + + mock_ps = MagicMock() + mock_ps.prompt_async = AsyncMock(side_effect=EOFError) + mock_close = MagicMock() + + with ( + patch( + f"{_MODULE}.create_initialized_session", + new=AsyncMock(return_value=initialized), + ), + patch(f"{_MODULE}._create_prompt_session", return_value=mock_ps), + patch("amplifier_app_cli.incremental_save.register_incremental_save"), + patch(f"{_MODULE}.SessionStore") as MockStore, + patch(f"{_MODULE}.console"), + patch( + f"{_MODULE}.process_runtime_mentions", + new=AsyncMock(side_effect=lambda s, t: t), + ), + patch(f"{_MODULE}.get_effective_config_summary"), + patch(f"{_MODULE}.close_dedicated_tty_input", new=mock_close), + ): + store_instance = MockStore.return_value + store_instance.get_metadata.return_value = {} + store_instance.save.return_value = None + + await interactive_chat( + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + ) + + mock_close.assert_called_once() + + @pytest.mark.asyncio + async def test_close_dedicated_tty_input_called_even_after_initial_prompt( + self, tmp_path: Path + ): + """Same guard, but through the initial_prompt auto-execute path (the + shape used by the other interactive_chat regression tests in this + suite) -- teardown must still close the fd.""" + from amplifier_app_cli.main import interactive_chat + + session = _make_mock_session() + initialized = _make_initialized(session) + + mock_ps = MagicMock() + mock_ps.prompt_async = AsyncMock(side_effect=EOFError) + mock_close = MagicMock() + + with ( + patch( + f"{_MODULE}.create_initialized_session", + new=AsyncMock(return_value=initialized), + ), + patch(f"{_MODULE}._create_prompt_session", return_value=mock_ps), + patch("amplifier_app_cli.incremental_save.register_incremental_save"), + patch(f"{_MODULE}.SessionStore") as MockStore, + patch(f"{_MODULE}.console"), + patch( + f"{_MODULE}.process_runtime_mentions", + new=AsyncMock(side_effect=lambda s, t: t), + ), + patch(f"{_MODULE}.get_effective_config_summary"), + patch("amplifier_app_cli.ui.render_message"), + patch(f"{_MODULE}.close_dedicated_tty_input", new=mock_close), + ): + store_instance = MockStore.return_value + store_instance.get_metadata.return_value = {} + store_instance.save.return_value = None + + await interactive_chat( + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + initial_prompt="Hi", + ) + + mock_close.assert_called_once() + + +class TestCloseDedicatedTtyTeardownIsRobust: + """The finally-block call site must not be able to mask the real error. + + ``close_dedicated_tty_input()`` itself is already idempotent by + contract (safe even if never opened, or already closed) -- these tests + guard that contract at the real call site: a session that never opened + the dedicated fd (e.g. non-tty stdin, no controlling terminal) must + still tear down cleanly, and calling close twice must not raise. + """ + + def test_close_is_idempotent_and_safe_when_never_opened(self): + """Calling close_dedicated_tty_input() when get_dedicated_tty_input() + was never called (e.g. stdin isn't a tty) must not raise.""" + from amplifier_app_cli import dedicated_tty_input as dti + + # Ensure clean slate regardless of prior test/process state. + dti.close_dedicated_tty_input() + # Calling again with nothing open must be a no-op, not an error. + dti.close_dedicated_tty_input() + dti.close_dedicated_tty_input() + + @pytest.mark.asyncio + async def test_interactive_chat_teardown_does_not_raise_when_fd_never_opened( + self, tmp_path: Path + ): + """End-to-end: even when the dedicated fd was never opened (this + test's own environment has no attached tty), interactive_chat()'s + finally block calling close_dedicated_tty_input() must not raise + and must not mask the underlying session cleanup.""" + from amplifier_app_cli import dedicated_tty_input as dti + from amplifier_app_cli.main import interactive_chat + + # Real, un-mocked close_dedicated_tty_input -- proves the actual + # idempotent implementation is safe at this call site too. + dti.close_dedicated_tty_input() # clean slate + + session = _make_mock_session() + initialized = _make_initialized(session) + + mock_ps = MagicMock() + mock_ps.prompt_async = AsyncMock(side_effect=EOFError) + + with ( + patch( + f"{_MODULE}.create_initialized_session", + new=AsyncMock(return_value=initialized), + ), + patch(f"{_MODULE}._create_prompt_session", return_value=mock_ps), + patch("amplifier_app_cli.incremental_save.register_incremental_save"), + patch(f"{_MODULE}.SessionStore") as MockStore, + patch(f"{_MODULE}.console"), + patch( + f"{_MODULE}.process_runtime_mentions", + new=AsyncMock(side_effect=lambda s, t: t), + ), + patch(f"{_MODULE}.get_effective_config_summary"), + ): + store_instance = MockStore.return_value + store_instance.get_metadata.return_value = {} + store_instance.save.return_value = None + + # Must not raise -- proves the real close_dedicated_tty_input() + # is safe to call from interactive_chat()'s finally block even + # when nothing was ever opened. + await interactive_chat( + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + ) + + # cleanup() (session teardown) still ran despite no dedicated fd + # having been opened -- the close call didn't short-circuit anything. + initialized.cleanup.assert_awaited_once()