diff --git a/CHANGELOG.md b/CHANGELOG.md index 001881ff..624c86ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -75,6 +75,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 failure is reported as a new `agent_compaction_skipped` event (`reason: "estimate_unavailable"`) rather than vanishing into stderr. See [Workflow Syntax → Context Compaction](docs/workflow-syntax.md#context-compaction). +- **Terminal left in cbreak mode (no echo/ICANON) after a run exited** — a + `KeyboardListener` started while the terminal was already in cbreak mode + (after an Esc pause/resume cycle, or a second listener in the same process) + captured that cbreak state as its "original" settings and restored it on + `stop()`, leaving the user's interactive shell without echo or canonical + mode. The tty baseline is now captured once per process, before the first + `tty.setcbreak()`, and reused by every later listener; `start()` is + idempotent while active so a duplicate call can't overwrite the baseline; + restore uses `TCSANOW` so a blocked output drain can't delay it; and the + saved baseline is cleared only after a successful `tcsetattr` so a + transient failure can be retried by `atexit`/`SIGTERM`/`stop()`. The + SIGTERM cleanup handler also no longer swallows the signal: after + restoring the terminal it delegates to the previously-installed + disposition (reset-and-re-raise for `SIG_DFL`, ignore for `SIG_IGN`, + invoke a callable previous handler), and re-registration captures the + previous disposition in the handler closure so the listener can no + longer recurse into itself. The run and resume commands now also reapply + and retire the baseline at their outermost cleanup boundary, after provider + shutdown and every other teardown step. This closes a later race where + normal completion or Ctrl+C could restore the terminal correctly and then + a provider's cleanup could put it back into cbreak; SIGTERM during that same + late-cleanup window restores the process baseline before terminating. + ([#290](https://github.com/microsoft/conductor/issues/290)) ## [0.1.36](https://github.com/microsoft/conductor/compare/v0.1.35...v0.1.36) - 2026-09-02 diff --git a/src/conductor/cli/run.py b/src/conductor/cli/run.py index 2fcbaf18..ccd2bb13 100644 --- a/src/conductor/cli/run.py +++ b/src/conductor/cli/run.py @@ -2531,46 +2531,55 @@ async def run_workflow_async( terminal_error_message = str(exc) raise finally: - # Write the terminal run record (MCP server plan E2) before - # removing the live one below, so a completed run remains - # resolvable by run_id after this process exits. Never raises -- - # see the helper's own docstring. - _write_terminal_record_for_current_process( - event_log_subscriber=event_log_subscriber, - workflow_path=workflow_path, - started_at=started_at_iso, - status=terminal_status, - output=terminal_output, - error_type=terminal_error_type, - error_message=terminal_error_message, - engine=engine, - ) + try: + # Write the terminal run record (MCP server plan E2) before + # removing the live one below, so a completed run remains + # resolvable by run_id after this process exits. Never raises -- + # see the helper's own docstring. + _write_terminal_record_for_current_process( + event_log_subscriber=event_log_subscriber, + workflow_path=workflow_path, + started_at=started_at_iso, + status=terminal_status, + output=terminal_output, + error_type=terminal_error_type, + error_message=terminal_error_message, + engine=engine, + ) - # Clean up the Fleet Manager run record on every exit path (E2 — - # normal completion, an explicit WorkflowTerminated re-raise, or an - # unexpected exception all funnel through this finally). Unlike the - # legacy PID file (removed only by a background child), this runs - # unconditionally: foreground and foreground-with-dashboard runs now - # write a record too and must remove it on exit just the same. - # Guarded (never raises) so a failure here cannot prevent the - # dashboard/event-log/file-logging cleanup below from running. - _remove_run_record_for_current_process_safe() - - # Stop dashboard if it was started - if dashboard is not None: - await dashboard.stop() + # Clean up the Fleet Manager run record on every exit path (E2 — + # normal completion, an explicit WorkflowTerminated re-raise, or an + # unexpected exception all funnel through this finally). Unlike the + # legacy PID file (removed only by a background child), this runs + # unconditionally: foreground and foreground-with-dashboard runs now + # write a record too and must remove it on exit just the same. + # Guarded (never raises) so a failure here cannot prevent the + # dashboard/event-log/file-logging cleanup below from running. + _remove_run_record_for_current_process_safe() + + # Stop dashboard if it was started + if dashboard is not None: + await dashboard.stop() - # Close JSONL event log and report path - if event_log_subscriber is not None: - event_log_subscriber.close() - _verbose_console.print( - styled("[dim]Event log written to: {}[/dim]", event_log_subscriber.path) - ) + # Close JSONL event log and report path + if event_log_subscriber is not None: + event_log_subscriber.close() + _verbose_console.print( + styled("[dim]Event log written to: {}[/dim]", event_log_subscriber.path) + ) + + # Report log file path to stderr and close file logging + if log_file is not None and _file_console is not None: + _verbose_console.print(styled("[dim]Log written to: {}[/dim]", log_file)) + close_file_logging() + finally: + # Provider shutdown occurs after the listener's inner ``finally`` + # and may itself touch the controlling TTY. Reapply the process + # baseline at the outermost boundary so normal exit, Ctrl+C, and a + # later cleanup failure all return a sane terminal to the shell. + from conductor.interrupt.listener import restore_terminal_baseline - # Report log file path to stderr and close file logging - if log_file is not None and _file_console is not None: - _verbose_console.print(styled("[dim]Log written to: {}[/dim]", log_file)) - close_file_logging() + restore_terminal_baseline(clear=True) def format_routes(routes: list[dict[str, Any]]) -> Text: @@ -3334,44 +3343,50 @@ async def resume_workflow_async( terminal_error_message = str(exc) raise finally: - # Write the terminal run record (MCP server plan E2) before - # removing the live one below -- mirrors run_workflow_async. A - # resumed run reuses its predecessor's run_id, so this call - # replaces the earlier terminal record rather than duplicating it. - # Never raises -- see the helper's own docstring. - _write_terminal_record_for_current_process( - event_log_subscriber=event_log_subscriber, - workflow_path=resolved_workflow_path, - started_at=started_at_iso, - status=terminal_status, - output=terminal_output, - error_type=terminal_error_type, - error_message=terminal_error_message, - engine=engine, - ) + try: + # Write the terminal run record (MCP server plan E2) before + # removing the live one below -- mirrors run_workflow_async. A + # resumed run reuses its predecessor's run_id, so this call + # replaces the earlier terminal record rather than duplicating it. + # Never raises -- see the helper's own docstring. + _write_terminal_record_for_current_process( + event_log_subscriber=event_log_subscriber, + workflow_path=resolved_workflow_path, + started_at=started_at_iso, + status=terminal_status, + output=terminal_output, + error_type=terminal_error_type, + error_message=terminal_error_message, + engine=engine, + ) - # Clean up the Fleet Manager run record on every exit path (E2 — - # mirrors run_workflow_async so a resumed run's record is removed - # the same way a fresh run's is). Guarded (never raises) so a - # failure here cannot prevent the dashboard/event-log/file-logging - # cleanup below from running. - _remove_run_record_for_current_process_safe() + # Clean up the Fleet Manager run record on every exit path (E2 — + # mirrors run_workflow_async so a resumed run's record is removed + # the same way a fresh run's is). Guarded (never raises) so a + # failure here cannot prevent the dashboard/event-log/file-logging + # cleanup below from running. + _remove_run_record_for_current_process_safe() - # Stop dashboard if it was started - if dashboard is not None: - await dashboard.stop() + # Stop dashboard if it was started + if dashboard is not None: + await dashboard.stop() - # Close JSONL event log and report path - if event_log_subscriber is not None: - event_log_subscriber.close() - _verbose_console.print( - styled("[dim]Event log written to: {}[/dim]", event_log_subscriber.path) - ) + # Close JSONL event log and report path + if event_log_subscriber is not None: + event_log_subscriber.close() + _verbose_console.print( + styled("[dim]Event log written to: {}[/dim]", event_log_subscriber.path) + ) + + # Report log file path to stderr and close file logging + if log_file is not None and _file_console is not None: + _verbose_console.print(styled("[dim]Log written to: {}[/dim]", log_file)) + close_file_logging() + finally: + # Keep resume teardown parity with ``run_workflow_async``. + from conductor.interrupt.listener import restore_terminal_baseline - # Report log file path to stderr and close file logging - if log_file is not None and _file_console is not None: - _verbose_console.print(styled("[dim]Log written to: {}[/dim]", log_file)) - close_file_logging() + restore_terminal_baseline(clear=True) async def _prefetch_plugin_sources(config: Any, workflow_path: Path) -> dict[str, Any]: diff --git a/src/conductor/interrupt/listener.py b/src/conductor/interrupt/listener.py index 2a4112dd..5d07d41e 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 @@ -32,6 +37,32 @@ # 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". The outer CLI cleanup +retires it after every teardown step has finished, so a later invocation in +the same process cannot apply stale settings to a new terminal. Tests also +reset it via a fixture. See issue #290.""" + + +def restore_terminal_baseline(*, clear: bool = False) -> None: + """Restore the process-wide TTY baseline and optionally retire it.""" + global _captured_baseline_settings + + if _captured_baseline_settings is None: + return + + try: + import termios + + termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, _captured_baseline_settings) + if clear: + _captured_baseline_settings = None + except (ImportError, ValueError, OSError): + pass + @dataclass class KeyboardListener: @@ -78,6 +109,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.""" @@ -104,9 +138,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 except termios.error: logger.debug("Failed to get terminal settings, listener not started") return @@ -179,8 +231,8 @@ async def suspend(self) -> None: try: import termios - termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self._original_settings) - except (ImportError, termios.error, ValueError, OSError): + termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, self._original_settings) + except (ImportError, ValueError, OSError): pass logger.debug("Keyboard listener suspended") @@ -227,48 +279,73 @@ def _restore_terminal(self) -> None: try: import termios - termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self._original_settings) - except (ImportError, termios.error, ValueError, OSError): + 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, ValueError, OSError): pass - self._original_settings = None def _register_cleanup_handlers(self) -> None: """Register atexit and SIGTERM handlers for crash safety.""" if not self._atexit_registered: - atexit.register(self._restore_terminal) + atexit.register(restore_terminal_baseline) 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. In an unmodified - # process (the common case) `signal.getsignal(SIGTERM)` is - # `signal.Handlers.SIG_DFL` -- an IntEnum member, not - # callable -- so falling through here would silently - # swallow the SIGTERM: the process would survive and keep - # running forever (Fleet Manager E3-T9; see Open Question 1 - # in docs/projects/fleet-manager/fleet-manager.plan.md). - # Restore the default disposition and re-raise the signal - # against ourselves instead, so the process actually - # terminates the way an unhandled SIGTERM normally would. - if callable(self._previous_sigterm): - self._previous_sigterm(signum, frame) - elif self._previous_sigterm is signal.SIG_IGN: - # An inherited SIG_IGN is a deliberate "this process - # does not die on SIGTERM", set by a supervisor or - # container init shim. It is an IntEnum member like - # SIG_DFL, so without this branch it would take the - # re-raise path below and terminate a process that was - # explicitly configured not to. - return - else: + # 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() + restore_terminal_baseline() + except Exception: + pass + finally: + self._original_settings = None + if previous is signal.SIG_DFL: + # In an unmodified process (the common case) + # `signal.getsignal(SIGTERM)` is `signal.Handlers.SIG_DFL` + # — an IntEnum member, not callable — so falling through + # here would silently swallow the SIGTERM: the process + # would survive and keep running forever (Fleet Manager + # E3-T9). Reset to default and re-raise so the default + # action (terminate) actually runs. signal.signal(signal.SIGTERM, signal.SIG_DFL) - os.kill(os.getpid(), signal.SIGTERM) + os.kill(os.getpid(), signum) + elif previous is signal.SIG_IGN: + # An inherited SIG_IGN is a deliberate "this process does + # not die on SIGTERM", set by a supervisor or container + # init shim: 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..389d4aca --- /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_cli_terminal_pty.py b/tests/test_interrupt/test_cli_terminal_pty.py new file mode 100644 index 00000000..c0d42143 --- /dev/null +++ b/tests/test_interrupt/test_cli_terminal_pty.py @@ -0,0 +1,148 @@ +"""Real-PTY coverage for the outer ``conductor run`` terminal boundary.""" + +from __future__ import annotations + +import contextlib +import os +import pty +import signal +import sys +import termios +import time +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from tests.test_interrupt.test_listener_pty import _replace_stdin_with_pty + + +def _write_wait_workflow(path: Path) -> Path: + workflow = path / "wait.yaml" + workflow.write_text( + "workflow:\n" + " name: tty-cleanup\n" + " entry_point: wait\n" + "agents:\n" + " - name: wait\n" + " type: wait\n" + " duration: 1ms\n" + " routes:\n" + " - to: $end\n" + ) + return workflow + + +@pytest.mark.parametrize("cleanup_fails", [False, True], ids=["success", "failure"]) +async def test_run_restores_after_provider_teardown( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cleanup_fails: bool +) -> None: + """Requirement: provider teardown cannot bypass final TTY restoration.""" + from conductor.cli.run import run_workflow_async + + monkeypatch.setenv("CONDUCTOR_HOME", str(tmp_path / "home")) + with _replace_stdin_with_pty(): + baseline = termios.tcgetattr(0) + engine = MagicMock() + engine.run = AsyncMock(return_value={}) + engine.config.workflow.cost.show_summary = False + registry = AsyncMock() + registry.__aenter__ = AsyncMock(return_value=registry) + + async def alter_terminal(*_args: object) -> None: + import tty + + tty.setcbreak(0) + if cleanup_fails: + raise RuntimeError("provider cleanup failed") + + registry.__aexit__ = AsyncMock(side_effect=alter_terminal) + expected_error = ( + pytest.raises(RuntimeError, match="provider cleanup failed") + if cleanup_fails + else contextlib.nullcontext() + ) + with ( + patch("conductor.cli.run.ProviderRegistry", return_value=registry), + patch("conductor.cli.run.WorkflowEngine", return_value=engine), + expected_error, + ): + await run_workflow_async(_write_wait_workflow(tmp_path), {}) + + assert termios.tcgetattr(0) == baseline + + +async def test_non_interactive_run_does_not_reapply_retired_baseline( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Requirement: a later non-interactive run must not reuse an old TTY baseline.""" + from conductor.cli.run import run_workflow_async + + monkeypatch.setenv("CONDUCTOR_HOME", str(tmp_path / "home")) + workflow = _write_wait_workflow(tmp_path) + with _replace_stdin_with_pty(): + engine = MagicMock() + engine.run = AsyncMock(return_value={}) + engine.config.workflow.cost.show_summary = False + registry = AsyncMock() + registry.__aenter__ = AsyncMock(return_value=registry) + registry.__aexit__ = AsyncMock(return_value=None) + + with ( + patch("conductor.cli.run.ProviderRegistry", return_value=registry), + patch("conductor.cli.run.WorkflowEngine", return_value=engine), + ): + await run_workflow_async(workflow, {}) + + import tty + + tty.setcbreak(0) + expected = termios.tcgetattr(0) + await run_workflow_async(workflow, {}, no_interactive=True) + + assert termios.tcgetattr(0) == expected + + +@pytest.mark.parametrize("interrupt", [False, True], ids=["normal-exit", "ctrl-c"]) +def test_cli_run_restores_terminal(interrupt: bool) -> None: + """Requirement: the real CLI restores exact TTY attrs on exit and Ctrl+C.""" + command = [ + sys.executable, + "-m", + "conductor.cli.app", + "--silent", + "run", + str(Path("examples/wait-smoke.yaml").resolve()), + ] + if interrupt: + command.extend(["--input", "middle_duration_ms=30000"]) + + pid, master_fd = pty.fork() + if pid == 0: + os.execv(command[0], command) + + baseline = termios.tcgetattr(master_fd) + interrupted = False + deadline = time.monotonic() + 20 + try: + while time.monotonic() < deadline: + if interrupt and not interrupted and time.monotonic() > deadline - 19: + os.write(master_fd, b"\x03") + interrupted = True + completed_pid, status = os.waitpid(pid, os.WNOHANG) + if completed_pid: + assert os.waitstatus_to_exitcode(status) == 0 + break + time.sleep(0.02) + else: + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + pytest.fail("conductor run did not exit before the PTY test deadline") + + assert termios.tcgetattr(master_fd) == baseline + finally: + with contextlib.suppress(OSError, ChildProcessError): + os.kill(pid, signal.SIGKILL) + with contextlib.suppress(ChildProcessError): + os.waitpid(pid, 0) + os.close(master_fd) diff --git a/tests/test_interrupt/test_listener.py b/tests/test_interrupt/test_listener.py index 82708de1..a0632689 100644 --- a/tests/test_interrupt/test_listener.py +++ b/tests/test_interrupt/test_listener.py @@ -9,6 +9,8 @@ import signal import sys import time +from collections.abc import Callable +from typing import Any, cast from unittest.mock import MagicMock, patch import pytest @@ -126,7 +128,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 @@ -333,7 +335,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] @@ -429,8 +431,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") @@ -443,7 +452,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: @@ -627,3 +638,484 @@ def test_process_exits_on_sigterm(self) -> None: os.close(master_fd) assert exited, "child process ignored SIGTERM and did not exit (E3-T9 regression)" + + +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_restores_process_baseline_after_listener_stop( + self, listener: KeyboardListener + ) -> None: + # Requirement: SIGTERM during provider teardown restores the cached + # process baseline even though listener.stop() cleared instance state. + 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("conductor.interrupt.listener.restore_terminal_baseline") as restore_baseline, + ): + listener._register_cleanup_handlers() + registered_handler = cast(Callable[[int, object], None], captured_handler[0][1]) + listener._original_settings = None + + registered_handler(signal_module.SIGTERM, None) + + restore_baseline.assert_called_once_with() + previous.assert_called_once_with(signal_module.SIGTERM, 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..0a907b68 --- /dev/null +++ b/tests/test_interrupt/test_listener_pty.py @@ -0,0 +1,228 @@ +"""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, restore_terminal_baseline + +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)" + ) + + +async def test_process_baseline_restores_late_teardown_change() -> None: + """Requirement: final cleanup repairs TTY changes made after listener.stop().""" + with _replace_stdin_with_pty(): + baseline = _tty_flags() + listener = KeyboardListener(interrupt_event=asyncio.Event()) + try: + await listener.start() + await listener.stop() + + # Simulate a provider/runtime cleanup that touches the controlling + # terminal after the keyboard listener has already stopped. + import tty + + tty.setcbreak(0) + assert _tty_flags() != baseline + + restore_terminal_baseline(clear=True) + finally: + restore_terminal_baseline(clear=True) + + assert _tty_flags() == baseline