diff --git a/src/conductor/interrupt/listener.py b/src/conductor/interrupt/listener.py index c0787998..22c69679 100644 --- a/src/conductor/interrupt/listener.py +++ b/src/conductor/interrupt/listener.py @@ -7,6 +7,11 @@ Uses a dedicated daemon thread for blocking stdin reads, delivering bytes into an ``asyncio.Queue`` via ``loop.call_soon_threadsafe``. This avoids thread leaks from abandoned ``run_in_executor`` futures. + +Terminal safety (issue #290): the original tty settings are captured once per +process into the module-level ``_CAPTURED_BASELINE_SETTINGS`` cache so a second +listener never re-captures cbreak state as "original", and the SIGTERM cleanup +handler restores the terminal before delegating to the previous disposition. """ from __future__ import annotations @@ -15,6 +20,7 @@ import atexit import contextlib import logging +import os import select import signal import sys @@ -31,6 +37,15 @@ # Timeout for disambiguating bare Esc from escape sequences (seconds) _ESC_DISAMBIGUATE_TIMEOUT = 0.05 +_CAPTURED_BASELINE_SETTINGS: Any = None +"""Process-wide tty baseline captured on the FIRST successful start(). + +Subsequent KeyboardListener instances reuse this baseline so a second +listener never re-captures cbreak state as "original". Process-lifetime +and intentionally never reset by production code: after stop() the tty +has been restored to exactly this baseline, so reuse is always +legitimate. Tests reset it via a fixture. See issue #290.""" + @dataclass class KeyboardListener: @@ -77,6 +92,9 @@ class KeyboardListener: _previous_sigterm: Any = field(default=None, repr=False) """Previous SIGTERM handler for restoration.""" + _sigterm_handler: Any = field(default=None, repr=False) + """This instance's own installed SIGTERM handler closure (issue #290).""" + _byte_queue: asyncio.Queue[int | None] = field(default_factory=asyncio.Queue, repr=False) """Async queue for delivering bytes from the reader thread.""" @@ -103,9 +121,27 @@ async def start(self) -> None: self._loop = asyncio.get_running_loop() self._stop_flag = False - # Save original terminal settings + # Save original terminal settings. A second listener must never + # re-snapshot an already-cbreak terminal as its "original" state, so + # the baseline is captured once per process into a module-level cache + # (issue #290). + global _CAPTURED_BASELINE_SETTINGS + + # Idempotent guard: a start() on a truly-active listener (baseline + # held AND reader thread running) is a no-op so it cannot overwrite + # the baseline or spawn duplicate threads. After suspend() the thread + # is None, so start-after-suspend correctly falls through below. + if self._original_settings is not None and self._reader_thread is not None: + logger.debug("Keyboard listener already active, start() is a no-op") + return + try: - self._original_settings = termios.tcgetattr(sys.stdin.fileno()) + if _CAPTURED_BASELINE_SETTINGS is not None: + # Reuse the process-wide pre-listener baseline. + self._original_settings = _CAPTURED_BASELINE_SETTINGS + else: + self._original_settings = termios.tcgetattr(sys.stdin.fileno()) + _CAPTURED_BASELINE_SETTINGS = self._original_settings # pyright: ignore[reportConstantRedefinition] except termios.error: logger.debug("Failed to get terminal settings, listener not started") return @@ -178,7 +214,7 @@ async def suspend(self) -> None: try: import termios - termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self._original_settings) + termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, self._original_settings) except (ImportError, termios.error, ValueError, OSError): pass @@ -226,10 +262,12 @@ def _restore_terminal(self) -> None: try: import termios - termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self._original_settings) + termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, self._original_settings) + # Clear the baseline only after a successful restore so a + # transient failure keeps it for a later retry (issue #290). + self._original_settings = None except (ImportError, termios.error, ValueError, OSError): pass - self._original_settings = None def _register_cleanup_handlers(self) -> None: """Register atexit and SIGTERM handlers for crash safety.""" @@ -237,17 +275,52 @@ def _register_cleanup_handlers(self) -> None: atexit.register(self._restore_terminal) self._atexit_registered = True - # Install SIGTERM handler that restores terminal then re-raises + # Install a SIGTERM handler that restores the terminal, then delegates + # to the previously-installed disposition so the process terminates + # with its expected action (issue #290). try: - self._previous_sigterm = signal.getsignal(signal.SIGTERM) + # Guard: if THIS instance's own handler is still installed, skip + # re-registration. Re-registering would capture our own closure as + # the "previous" disposition, recursing forever on invocation. The + # identity check targets this instance's stored handler only — a + # stale closure from a stopped listener must not block a new one. + if ( + self._sigterm_handler is not None + and signal.getsignal(signal.SIGTERM) is self._sigterm_handler + ): + return + + # Capture the previous disposition into a LOCAL variable so the + # closure is immutable: reading the mutable field at invocation + # time would let a later re-registration rewrite what an + # already-installed closure delegates to. + previous = signal.getsignal(signal.SIGTERM) + self._previous_sigterm = previous # backward-compat introspection def _sigterm_handler(signum: int, frame: Any) -> None: - self._restore_terminal() - # Call previous handler if it was callable - if callable(self._previous_sigterm): - self._previous_sigterm(signum, frame) + # A failed restore must never swallow the signal: swallow any + # error here so the handler always reaches the delegation path. + # The baseline is dropped unconditionally afterward — the + # process is terminating, so a stale value must not be reused. + try: + self._restore_terminal() + except Exception: + pass + finally: + self._original_settings = None + if previous is signal.SIG_DFL: + # Reset to default and re-raise so the default action + # (terminate) actually runs. + signal.signal(signal.SIGTERM, signal.SIG_DFL) + os.kill(os.getpid(), signum) + elif previous is signal.SIG_IGN: + # Caller asked to ignore SIGTERM: restore and stay alive. + return + elif callable(previous): + previous(signum, frame) signal.signal(signal.SIGTERM, _sigterm_handler) + self._sigterm_handler = _sigterm_handler except (OSError, ValueError): # Can't set signal handler (not main thread, etc.) pass diff --git a/tests/test_interrupt/conftest.py b/tests/test_interrupt/conftest.py new file mode 100644 index 00000000..673eb8ef --- /dev/null +++ b/tests/test_interrupt/conftest.py @@ -0,0 +1,13 @@ +import pytest + + +@pytest.fixture(autouse=True) +def _reset_baseline_cache(monkeypatch: pytest.MonkeyPatch) -> None: + # Requirement: the module-level baseline cache must never leak between + # tests running in the same pytest process (issue #290). raising=False + # keeps this a no-op until the implementation adds the attribute. + monkeypatch.setattr( + "conductor.interrupt.listener._CAPTURED_BASELINE_SETTINGS", + None, + raising=False, + ) diff --git a/tests/test_interrupt/test_listener.py b/tests/test_interrupt/test_listener.py index 9c41cd93..6c77f567 100644 --- a/tests/test_interrupt/test_listener.py +++ b/tests/test_interrupt/test_listener.py @@ -4,6 +4,8 @@ import asyncio import sys +from collections.abc import Callable +from typing import Any, cast from unittest.mock import MagicMock, patch import pytest @@ -121,7 +123,7 @@ async def test_stop_restores_terminal(self, listener: KeyboardListener) -> None: listener._restore_terminal() mock_termios.tcsetattr.assert_called_once_with( - 0, mock_termios.TCSADRAIN, original_settings + 0, mock_termios.TCSANOW, original_settings ) assert listener._original_settings is None @@ -328,7 +330,7 @@ async def test_call_soon_threadsafe_used_for_ctrl_g( listener._byte_queue.put_nowait(_CTRL_G_BYTE) listener._byte_queue.put_nowait(None) - threadsafe_args: list[tuple] = [] + threadsafe_args: list[tuple[Any, ...]] = [] original_call = loop.call_soon_threadsafe def tracking_call(*args, **kwargs): # type: ignore[no-untyped-def] @@ -424,8 +426,15 @@ def test_restore_clears_original_settings(self, listener: KeyboardListener) -> N assert listener._original_settings is None - def test_restore_handles_termios_error_gracefully(self, listener: KeyboardListener) -> None: - """Verify restore handles termios errors without raising.""" + def test_restore_keeps_settings_on_termios_error_for_retry( + self, listener: KeyboardListener + ) -> None: + """Verify a failed restore keeps the baseline so it can be retried. + + Requirement: a transient restore failure must not lose the only + correct baseline — the saved settings stay in place so a later + atexit/SIGTERM/stop() attempt can retry the restore (issue #290). + """ mock_termios = MagicMock() mock_termios.error = OSError mock_termios.tcsetattr.side_effect = OSError("terminal gone") @@ -438,7 +447,9 @@ def test_restore_handles_termios_error_gracefully(self, listener: KeyboardListen mock_stdin.fileno.return_value = 0 listener._restore_terminal() # Should not raise - assert listener._original_settings is None + assert listener._original_settings is not None + # Drop the fake baseline so the registered atexit handler is a no-op + listener._original_settings = None class TestConstants: @@ -455,3 +466,455 @@ def test_ctrl_g_byte_value(self) -> None: def test_disambiguate_timeout_value(self) -> None: """Verify Esc disambiguation timeout is 50ms.""" assert _ESC_DISAMBIGUATE_TIMEOUT == 0.05 + + +class TestBaselineCacheAndIdempotentStart: + """Tests for the module-level baseline cache and idempotent start (issue #290). + + Red phase: the terminal baseline must be captured once per process and + cached at module level; repeated ``start()`` calls must not re-read it. + """ + + @pytest.mark.asyncio + async def test_module_baseline_captured_once_per_process( + self, interrupt_event: asyncio.Event + ) -> None: + # Requirement: termios.tcgetattr is called exactly once per process, + # no matter how many KeyboardListener instances start (issue #290). + mock_termios = MagicMock() + mock_tty = MagicMock() + mock_termios.tcgetattr.return_value = [1, 2, 3] + mock_termios.error = OSError + + with ( + patch("sys.stdin") as mock_stdin, + patch.dict("sys.modules", {"termios": mock_termios, "tty": mock_tty}), + ): + mock_stdin.isatty.return_value = True + mock_stdin.fileno.return_value = 0 + + listener_a = KeyboardListener(interrupt_event=interrupt_event) + await listener_a.start() + mock_termios.tcgetattr.assert_called_once() + + mock_termios.reset_mock() + + listener_b = KeyboardListener(interrupt_event=interrupt_event) + await listener_b.start() + mock_termios.tcgetattr.assert_not_called() + + await listener_a.stop() + await listener_b.stop() + + @pytest.mark.asyncio + async def test_idempotent_start_is_noop_when_active( + self, interrupt_event: asyncio.Event + ) -> None: + # Requirement: calling start() on an already-active listener is a no-op + # — no tcgetattr/setcbreak, no new reader thread (issue #290). + mock_termios = MagicMock() + mock_tty = MagicMock() + mock_termios.tcgetattr.return_value = [1, 2, 3] + mock_termios.error = OSError + + with ( + patch("sys.stdin") as mock_stdin, + patch.dict("sys.modules", {"termios": mock_termios, "tty": mock_tty}), + ): + mock_stdin.isatty.return_value = True + mock_stdin.fileno.return_value = 0 + + listener = KeyboardListener(interrupt_event=interrupt_event) + await listener.start() + first_thread = listener._reader_thread + + mock_termios.reset_mock() + mock_tty.reset_mock() + + await listener.start() + + mock_termios.tcgetattr.assert_not_called() + mock_tty.setcbreak.assert_not_called() + assert listener._reader_thread is first_thread + + await listener.stop() + + @pytest.mark.asyncio + async def test_start_after_suspend_restarts_listener( + self, interrupt_event: asyncio.Event + ) -> None: + # Requirement: start() after suspend() re-enters cbreak and spawns a + # new reader thread, but reuses the cached baseline — tcgetattr must + # NOT be called again (issue #290). + mock_termios = MagicMock() + mock_tty = MagicMock() + mock_termios.tcgetattr.return_value = [1, 2, 3] + mock_termios.error = OSError + + with ( + patch("sys.stdin") as mock_stdin, + patch.dict("sys.modules", {"termios": mock_termios, "tty": mock_tty}), + ): + mock_stdin.isatty.return_value = True + mock_stdin.fileno.return_value = 0 + + listener = KeyboardListener(interrupt_event=interrupt_event) + await listener.start() + await listener.suspend() + + mock_termios.reset_mock() + mock_tty.reset_mock() + + await listener.start() + + mock_tty.setcbreak.assert_called_once_with(0) + assert listener._reader_thread is not None + mock_termios.tcgetattr.assert_not_called() + + await listener.stop() + + @pytest.mark.asyncio + async def test_suspend_keeps_baseline_for_resume(self, interrupt_event: asyncio.Event) -> None: + # Requirement: suspend() keeps _original_settings for resume(), and + # resume() re-enters cbreak without re-capturing the baseline + # (issue #290; regression guard — passes on current code). + mock_termios = MagicMock() + mock_tty = MagicMock() + mock_termios.tcgetattr.return_value = [1, 2, 3] + mock_termios.error = OSError + + with ( + patch("sys.stdin") as mock_stdin, + patch.dict("sys.modules", {"termios": mock_termios, "tty": mock_tty}), + ): + mock_stdin.isatty.return_value = True + mock_stdin.fileno.return_value = 0 + + listener = KeyboardListener(interrupt_event=interrupt_event) + await listener.start() + captured = listener._original_settings + assert captured is not None + + await listener.suspend() + assert listener._original_settings is captured + + mock_termios.reset_mock() + + await listener.resume() + mock_termios.tcgetattr.assert_not_called() + + await listener.stop() + # Drop the fake baseline so the registered atexit handler is a no-op + listener._original_settings = None + + @pytest.mark.asyncio + async def test_suspend_restores_with_tcsanow(self, interrupt_event: asyncio.Event) -> None: + # Requirement: suspend() restores the terminal with TCSANOW so the + # human gate gets a sane terminal immediately — TCSADRAIN would wait + # for pending output and can race with the gate's first read + # (issue #290). + mock_termios = MagicMock() + mock_tty = MagicMock() + mock_termios.tcgetattr.return_value = [1, 2, 3] + mock_termios.error = OSError + + with ( + patch("sys.stdin") as mock_stdin, + patch.dict("sys.modules", {"termios": mock_termios, "tty": mock_tty}), + ): + mock_stdin.isatty.return_value = True + mock_stdin.fileno.return_value = 0 + + listener = KeyboardListener(interrupt_event=interrupt_event) + await listener.start() + mock_termios.reset_mock() + + await listener.suspend() + + mock_termios.tcsetattr.assert_called_once_with( + 0, mock_termios.TCSANOW, listener._original_settings + ) + + await listener.stop() + + def test_restore_clears_baseline_only_on_success(self, interrupt_event: asyncio.Event) -> None: + # Requirement: _restore_terminal() clears _original_settings only + # after a successful tcsetattr; on termios.error the baseline must be + # kept so a later restore can retry (issue #290). + mock_termios = MagicMock() + mock_termios.error = OSError + mock_termios.tcsetattr.side_effect = [OSError("terminal gone"), None] + saved_settings = [1, 2, 3] + + listener = KeyboardListener(interrupt_event=interrupt_event) + listener._original_settings = saved_settings + + with ( + patch("sys.stdin") as mock_stdin, + patch.dict("sys.modules", {"termios": mock_termios}), + ): + mock_stdin.fileno.return_value = 0 + + listener._restore_terminal() + assert listener._original_settings is saved_settings + + listener._restore_terminal() + assert listener._original_settings is None + + +class TestSigtermHandlerDelegation: + """Tests for SIGTERM handler delegation semantics (issue #290). + + Red phase: the current ``_sigterm_handler`` closure restores the terminal + and optionally calls a callable previous handler, but it does NOT re-raise + when the previous disposition was ``SIG_DFL`` — the process silently + survives SIGTERM, leaving the terminal in cbreak. The fixed implementation + must restore, reset to SIG_DFL, and re-raise via ``os.kill``. + """ + + def test_sigterm_handler_restores_then_re_raises_when_previous_was_dfl( + self, listener: KeyboardListener + ) -> None: + # Requirement: when the previous SIGTERM disposition is SIG_DFL, the + # handler must (1) restore the terminal, (2) reset the disposition + # back to SIG_DFL, and (3) re-raise SIGTERM to self via os.kill so + # the process actually terminates with the default action (issue #290). + import os + import signal as signal_module + + captured_handler: list[tuple[int, object]] = [] + + def fake_signal(signum: int, handler): # type: ignore[no-untyped-def] + captured_handler.append((signum, handler)) + + with ( + patch("signal.getsignal", return_value=signal_module.SIG_DFL), + patch("signal.signal", side_effect=fake_signal), + patch("os.kill") as mock_kill, + ): + listener._original_settings = [1, 2, 3] + listener._register_cleanup_handlers() + + assert len(captured_handler) == 1 + registered_signum, registered_handler = captured_handler[0] + registered_handler = cast(Callable[[int, object], None], registered_handler) + assert registered_signum == signal_module.SIGTERM + + registered_handler(signal_module.SIGTERM, None) + + assert listener._original_settings is None + + reset_calls = [ + (s, h) + for (s, h) in captured_handler[1:] + if s == signal_module.SIGTERM and h == signal_module.SIG_DFL + ] + assert len(reset_calls) == 1, ( + "handler did not reset SIGTERM disposition to SIG_DFL before re-raising" + ) + + mock_kill.assert_called_once_with(os.getpid(), signal_module.SIGTERM) + + listener._original_settings = None + listener._atexit_registered = False + + def test_sigterm_handler_noop_when_previous_was_ign(self, listener: KeyboardListener) -> None: + # Requirement: when the previous SIGTERM disposition is SIG_IGN, the + # handler must restore the terminal but must NOT re-raise (the caller + # explicitly asked to ignore SIGTERM) and must NOT touch the + # disposition again (issue #290). + import signal as signal_module + + captured_handler: list[tuple[int, object]] = [] + + def fake_signal(signum: int, handler): # type: ignore[no-untyped-def] + captured_handler.append((signum, handler)) + + with ( + patch("signal.getsignal", return_value=signal_module.SIG_IGN), + patch("signal.signal", side_effect=fake_signal), + patch("os.kill") as mock_kill, + ): + listener._original_settings = [1, 2, 3] + listener._register_cleanup_handlers() + + assert len(captured_handler) == 1 + registered_signum, registered_handler = captured_handler[0] + registered_handler = cast(Callable[[int, object], None], registered_handler) + assert registered_signum == signal_module.SIGTERM + + registered_handler(signal_module.SIGTERM, None) + + assert listener._original_settings is None + + mock_kill.assert_not_called() + assert len(captured_handler) == 1, ( + "handler must not modify SIGTERM disposition when previous was SIG_IGN" + ) + + listener._original_settings = None + listener._atexit_registered = False + + def test_sigterm_handler_calls_callable_previous(self, listener: KeyboardListener) -> None: + # Requirement: when the previous SIGTERM disposition is a callable, + # the handler must restore the terminal first and then delegate to + # the callable with the same (signum, frame) arguments (issue #290). + import signal as signal_module + + previous = MagicMock() + captured_handler: list[tuple[int, object]] = [] + + def fake_signal(signum: int, handler): # type: ignore[no-untyped-def] + captured_handler.append((signum, handler)) + + with ( + patch("signal.getsignal", return_value=previous), + patch("signal.signal", side_effect=fake_signal), + patch("os.kill"), + ): + listener._original_settings = [1, 2, 3] + listener._register_cleanup_handlers() + + assert len(captured_handler) == 1 + _, registered_handler = captured_handler[0] + registered_handler = cast(Callable[[int, object], None], registered_handler) + + registered_handler(signal_module.SIGTERM, None) + + assert listener._original_settings is None + + previous.assert_called_once_with(signal_module.SIGTERM, None) + + listener._original_settings = None + listener._atexit_registered = False + + def test_register_cleanup_handlers_does_not_self_recurse( + self, listener: KeyboardListener + ) -> None: + # Requirement: calling ``_register_cleanup_handlers`` twice on the same + # instance must NOT re-capture the previously-installed own handler as + # ``_previous_sigterm`` (that would produce unbounded self-recursion + # when the handler delegates). The second call must detect that our + # own handler is still installed and skip re-registration (issue #290). + import signal as signal_module + + captured_handler: list[tuple[int, object]] = [] + + def fake_signal(signum: int, handler): # type: ignore[no-untyped-def] + captured_handler.append((signum, handler)) + + with ( + patch("signal.getsignal", return_value=signal_module.SIG_DFL), + patch("signal.signal", side_effect=fake_signal), + patch("os.kill"), + ): + listener._register_cleanup_handlers() + + assert len(captured_handler) == 1 + _, h1 = captured_handler[0] + h1 = cast(Callable[[int, object], None], h1) + + listener._sigterm_handler = h1 + + with ( + patch("signal.getsignal", return_value=h1), + patch("signal.signal", side_effect=fake_signal) as mock_signal, + patch("os.kill"), + ): + listener._register_cleanup_handlers() + + mock_signal.assert_not_called() + assert len(captured_handler) == 1, ( + "second _register_cleanup_handlers call must not re-register " + "SIGTERM handler (would self-capture and recurse)" + ) + + listener._original_settings = None + listener._atexit_registered = False + + def test_new_listener_registers_own_handler_after_stopped_listener( + self, interrupt_event: asyncio.Event + ) -> None: + # Requirement: when a second KeyboardListener starts after a first one + # was stopped (``stop()`` does not remove signal handlers), the new + # listener must install its OWN handler — not skip registration just + # because a conductor SIGTERM closure is currently installed. The new + # handler must also restore its own terminal before delegating + # (issue #290). + import signal as signal_module + + listener_a = KeyboardListener(interrupt_event=interrupt_event) + listener_b = KeyboardListener(interrupt_event=interrupt_event) + + captured_a: list[tuple[int, object]] = [] + + def fake_signal_a(signum: int, handler): # type: ignore[no-untyped-def] + captured_a.append((signum, handler)) + + with ( + patch("signal.getsignal", return_value=signal_module.SIG_DFL), + patch("signal.signal", side_effect=fake_signal_a), + patch("os.kill"), + ): + listener_a._register_cleanup_handlers() + + assert len(captured_a) == 1 + _, h_a = captured_a[0] + h_a = cast(Callable[[int, object], None], h_a) + + listener_a._sigterm_handler = h_a + + captured_b: list[tuple[int, object]] = [] + + def fake_signal_b(signum: int, handler): # type: ignore[no-untyped-def] + captured_b.append((signum, handler)) + + with ( + patch("signal.getsignal", return_value=h_a), + patch("signal.signal", side_effect=fake_signal_b), + patch("os.kill"), + ): + listener_b._register_cleanup_handlers() + + assert len(captured_b) == 1, ( + "new listener must install its own SIGTERM handler even when " + "a stale conductor handler from another instance is still installed" + ) + _, h_b = captured_b[0] + h_b = cast(Callable[[int, object], None], h_b) + assert h_b is not h_a, "new listener must not reuse the other instance's handler" + + assert listener_b._sigterm_handler is h_b + + call_order: list[str] = [] + + original_a_restore = listener_a._restore_terminal + original_b_restore = listener_b._restore_terminal + + def spy_a() -> None: + call_order.append("A") + original_a_restore() + + def spy_b() -> None: + call_order.append("B") + original_b_restore() + + listener_a._original_settings = [1, 1, 1] + listener_b._original_settings = [2, 2, 2] + + with ( + patch.object(listener_a, "_restore_terminal", side_effect=spy_a), + patch.object(listener_b, "_restore_terminal", side_effect=spy_b), + patch("os.kill"), + ): + h_b(signal_module.SIGTERM, None) + + assert call_order, "H_B must at minimum restore B's terminal" + assert call_order[0] == "B", ( + f"B's own terminal must be restored before any delegation; got call order {call_order}" + ) + + listener_a._original_settings = None + listener_a._atexit_registered = False + listener_b._original_settings = None + listener_b._atexit_registered = False diff --git a/tests/test_interrupt/test_listener_pty.py b/tests/test_interrupt/test_listener_pty.py new file mode 100644 index 00000000..23c37894 --- /dev/null +++ b/tests/test_interrupt/test_listener_pty.py @@ -0,0 +1,205 @@ +"""Real-pty integration tests for KeyboardListener terminal restore (issue #290). + +These tests allocate a genuine pseudo-terminal via ``pty.openpty()`` and point +fd 0 (and ``sys.stdin``) at its slave so the production ``KeyboardListener`` +lifecycle (``termios.tcgetattr`` / ``tty.setcbreak`` / ``termios.tcsetattr``) +runs against a real tty instead of mocks. This is the only reliable way to +reproduce the bug where a listener started on an already-cbreak terminal +snapshots cbreak as its "original" settings and later restores them. + +Expected state on unfixed code (TDD red): + +- ``test_full_lifecycle_restores_terminal`` — PASSES (regression guard) +- ``test_second_listener_does_not_capture_cbreak`` — FAILS +- ``test_double_start_same_instance_does_not_corrupt_baseline`` — FAILS +""" + +from __future__ import annotations + +import asyncio +import contextlib +import io +import os +import pty +import sys +import termios +from collections.abc import Iterator + +import pytest + +from conductor.interrupt.listener import KeyboardListener + +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="termios/pty is Unix-only") + + +class _PtyStdinShim: + """Minimal stdin replacement that satisfies the KeyboardListener contract. + + pytest's capture replaces ``sys.stdin`` with a non-tty object, so the + listener's reader thread (which calls ``sys.stdin.buffer.read(1)``) and + ``select.select([sys.stdin], ...)`` both need a shim that: + + - reports ``isatty()`` as True, + - returns fd 0 from ``fileno()`` (fd 0 is dup2'd at the pty slave), + - exposes ``.buffer`` as a real buffered binary reader on fd 0. + """ + + def __init__(self) -> None: + # Duplicate fd 0 so the shim owns its own handle to the pty slave and + # closing the shim cannot disturb fd 0 itself. + self._owned_fd = os.dup(0) + self.buffer = io.BufferedReader(io.FileIO(self._owned_fd, "rb")) + + def isatty(self) -> bool: + return True + + def fileno(self) -> int: + return 0 + + def close(self) -> None: + self.buffer.close() + + +@contextlib.contextmanager +def _replace_stdin_with_pty() -> Iterator[int]: + """Point fd 0 and ``sys.stdin`` at a fresh pty slave; yield the master fd. + + Restores fd 0 and ``sys.stdin`` in ``finally`` even on test failure, and + closes the shim's buffer BEFORE closing the pty fds so no reader thread + can block on a closed fd during teardown. + """ + try: + master_fd, slave_fd = pty.openpty() + except OSError as exc: + pytest.skip(f"pty.openpty() unavailable in this environment: {exc}") + + saved_stdin = sys.stdin + saved_fd = os.dup(0) + shim: _PtyStdinShim | None = None + try: + os.dup2(slave_fd, 0) + shim = _PtyStdinShim() + sys.stdin = shim + yield master_fd + finally: + # Restore sys.stdin first so nothing references the shim after close. + sys.stdin = saved_stdin + if shim is not None: + shim.close() + # Restore fd 0 to its pre-test target before closing anything else. + os.dup2(saved_fd, 0) + os.close(saved_fd) + os.close(master_fd) + os.close(slave_fd) + + +def _tty_flags() -> int: + """Return the ICANON|ECHO subset of the current fd-0 local-mode flags.""" + return termios.tcgetattr(0)[3] & (termios.ICANON | termios.ECHO) + + +async def test_full_lifecycle_restores_terminal() -> None: + """Requirement: start -> suspend -> resume -> stop restores the terminal. + + Regression guard: a single listener driven through its full lifecycle on a + real pty must leave ICANON|ECHO exactly as found. PASSES on unfixed code. + """ + with _replace_stdin_with_pty(): + baseline = _tty_flags() + listener = KeyboardListener(interrupt_event=asyncio.Event()) + try: + await listener.start() + # Prove the listener actually started against the pty. + assert listener._task is not None + assert listener._reader_thread is not None + + await listener.suspend() + await listener.resume() + assert listener._task is not None + assert listener._reader_thread is not None + + await listener.stop() + finally: + await listener.stop() + + assert _tty_flags() == baseline, ( + f"tty flags after full lifecycle differ from baseline: " + f"{_tty_flags():#x} != {baseline:#x}" + ) + + +async def test_second_listener_does_not_capture_cbreak() -> None: + """Requirement: a second listener must not snapshot cbreak as baseline. + + Repro for issue #290: with listener A active (terminal in cbreak), starting + listener B currently snapshots the cbreak settings as B's "original". If A + is then stopped first, B's later stop() restores its cbreak snapshot, + leaving the terminal in cbreak after both listeners have stopped. + + Stop order matters: A BEFORE B (reverse order masks the bug because B's + stop would run while A's correct settings were the last ones applied). + + FAILS on unfixed code; PASSES once start() reuses the process-wide + pre-listener baseline instead of re-snapshotting the live tty. + """ + with _replace_stdin_with_pty(): + baseline = _tty_flags() + listener_a = KeyboardListener(interrupt_event=asyncio.Event()) + listener_b = KeyboardListener(interrupt_event=asyncio.Event()) + try: + await listener_a.start() + assert listener_a._task is not None + assert listener_a._reader_thread is not None + + # Start B WITHOUT stopping A — B sees an already-cbreak terminal. + await listener_b.start() + assert listener_b._task is not None + assert listener_b._reader_thread is not None + + # Stop A first, then B (documented order above). + await listener_a.stop() + await listener_b.stop() + finally: + await listener_a.stop() + await listener_b.stop() + + assert _tty_flags() == baseline, ( + f"tty flags after stopping both listeners differ from baseline: " + f"{_tty_flags():#x} != {baseline:#x} " + f"(listener B captured cbreak as its original settings)" + ) + + +async def test_double_start_same_instance_does_not_corrupt_baseline() -> None: + """Requirement: calling start() twice on one listener keeps the baseline. + + Repro for issue #290: a second start() on the SAME instance currently + overwrites ``_original_settings`` with a snapshot of the cbreak terminal + the first start() installed. The later stop() then restores cbreak, + leaving the terminal broken. + + FAILS on unfixed code; PASSES once start() is idempotent w.r.t. baseline + capture (reuse the process-wide pre-listener baseline). + """ + with _replace_stdin_with_pty(): + baseline = _tty_flags() + listener = KeyboardListener(interrupt_event=asyncio.Event()) + try: + await listener.start() + assert listener._task is not None + assert listener._reader_thread is not None + + # Second start on the SAME instance must not clobber the baseline. + await listener.start() + assert listener._task is not None + assert listener._reader_thread is not None + + await listener.stop() + finally: + await listener.stop() + + assert _tty_flags() == baseline, ( + f"tty flags after double-start+stop differ from baseline: " + f"{_tty_flags():#x} != {baseline:#x} " + f"(second start() captured cbreak as the original settings)" + )