From edd54a427c61f6768e9229a4f923477b4eaa2c41 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Sat, 18 Jul 2026 16:26:23 +0800 Subject: [PATCH] core/tools/process: drop host TTY restore; rely on CSI sanitize only --- CLAUDE.md | 2 +- README.md | 4 +- src/plyngent/tools/process/__init__.py | 1 - src/plyngent/tools/process/pty_session.py | 14 ++--- src/plyngent/tools/process/pty_terminal.py | 68 ++++++-------------- tests/test_tools/test_process.py | 73 +++------------------- 6 files changed, 35 insertions(+), 127 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index d60e14e..3129936 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -81,7 +81,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use. - **`workspace`**: path resolve under root; path substring denylist; command basename denylist; clearer deny messages. - **`file`**: `read_file`, `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace`, `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`. - **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep). -- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (CSI sanitized for host safety); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); session limit/idle TTL/output budget; close terminate→kill + **host TTY restore**. +- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); session limit/idle TTL/output budget; close terminate→kill. - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget. - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn). - **`vcs`**: read-only VCS tools (`vcs_kind` / `vcs_status` / `vcs_diff` / `vcs_log` / `vcs_branch`) via `VcsBackend` protocol; **git** implemented; detectors are pluggable for other systems. diff --git a/README.md b/README.md index a468412..da6a46f 100644 --- a/README.md +++ b/README.md @@ -232,8 +232,8 @@ Safety defaults: - Command basename denylist (e.g. dangerous shells/utilities). - Destructive tools (delete/move/overwrite) can require confirm (`confirm_destructive`; default deny in non-TTY). Override for the session with `/yolo on|off|once` or startup `--yes` (path/command denylists still apply). - PTY sessions: caps, idle TTL, output budget; master FD is non-inheritable; sessions closed on chat exit. - Prefer file tools over full-screen editors (`vim`/`nano`) for edits. `read_pty` escapes CSI so tool - results cannot reprogram the host TTY; `close_pty` / chat exit also restore the host terminal. + Prefer file tools over full-screen editors (`vim`/`nano`) for edits. `read_pty` sanitizes CSI/controls + so tool results cannot reprogram the host TTY (no host terminal reset on exit). `write_pty` is literal text only; use `write_pty_keys` for `\xHH`, `ctrl+x`, `key=esc|enter|…`. ## Usage / context (CLI) diff --git a/src/plyngent/tools/process/__init__.py b/src/plyngent/tools/process/__init__.py index d4adb25..be0e8f8 100644 --- a/src/plyngent/tools/process/__init__.py +++ b/src/plyngent/tools/process/__init__.py @@ -1,7 +1,6 @@ from .close_pty import close_pty as close_pty from .open_pty import open_pty as open_pty from .pty_terminal import decode_write_data as decode_write_data -from .pty_terminal import restore_host_terminal as restore_host_terminal from .pty_terminal import sanitize_pty_output_for_tool as sanitize_pty_output_for_tool from .read_pty import read_pty as read_pty from .run_command import run_command as run_command diff --git a/src/plyngent/tools/process/pty_session.py b/src/plyngent/tools/process/pty_session.py index 62897ab..a25ad02 100644 --- a/src/plyngent/tools/process/pty_session.py +++ b/src/plyngent/tools/process/pty_session.py @@ -8,7 +8,7 @@ from threading import Lock from typing import TYPE_CHECKING, ClassVar from plyngent.tools.process.pty_backend import PtyHandle, pty_available, spawn_pty -from plyngent.tools.process.pty_terminal import restore_host_terminal, sanitize_pty_output_for_tool +from plyngent.tools.process.pty_terminal import sanitize_pty_output_for_tool from plyngent.tools.workspace import WorkspaceError, check_command_allowed, resolve_path if TYPE_CHECKING: @@ -330,7 +330,6 @@ class PtyManager: message="unknown session", ) if session.closed: - # No host restore: no child was torn down this call. return PtyCloseResult( session_id=session_id, closed=True, @@ -358,8 +357,8 @@ class PtyManager: with cls._lock: _ = cls._sessions.pop(session_id, None) - # Host TTY may have been reprogrammed if tool results echoed CSI. - restore_host_terminal() + # No host TTY reset: tool-facing read_pty sanitizes CSI so chat echo + # cannot reprogram the user terminal. return PtyCloseResult( session_id=session_id, closed=True, @@ -387,12 +386,7 @@ class PtyManager: @classmethod def close_all(cls) -> None: - """Close every open session. - - Host TTY restore runs only when at least one session was actually closed - (via :meth:`close`). Empty registry is a no-op — important so Ctrl-D / - chat exit does not flash-reset the terminal when no PTY was used. - """ + """Close every open session (no host TTY reset).""" with cls._lock: ids = list(cls._sessions.keys()) for session_id in ids: diff --git a/src/plyngent/tools/process/pty_terminal.py b/src/plyngent/tools/process/pty_terminal.py index 1ad6576..3e0a6e6 100644 --- a/src/plyngent/tools/process/pty_terminal.py +++ b/src/plyngent/tools/process/pty_terminal.py @@ -1,34 +1,16 @@ -"""Host-terminal safety helpers for agent PTY sessions. +"""PTY payload helpers for agent tools. -Child PTY I/O is isolated on the master FD. The practical leak is that -``read_pty`` returns raw CSI/alt-screen sequences which the CLI then echoes -to the *user's* stdout as tool results — those bytes reprogram the host TTY. +Child PTY I/O stays on the master FD. Tool results are printed on the *user* +TTY by the CLI, so any CSI/control bytes in ``read_pty`` data would reprogram +the host terminal. We sanitize on the tool boundary instead of resetting the +host after close (which flashes the screen on every chat exit). """ from __future__ import annotations -import contextlib import re -import sys -from typing import Final - -# Leave alt-screen, restore cursor visibility, reset SGR, keypad, mouse, etc. -# Intentional over-reset: cheap and safe after TUI children. -_HOST_RESTORE: Final[str] = ( - "\x1b[?1049l" # leave alternate screen buffer - "\x1b[?47l" # leave alt screen (legacy) - "\x1b[?25h" # show cursor - "\x1b[0m" # reset SGR - "\x1b[?1l" # normal cursor keys - "\x1b[?1000l" # mouse off - "\x1b[?1002l" - "\x1b[?1003l" - "\x1b[?1006l" - "\x1b[?2004l" # bracketed paste off - "\x1b[?7h" # wraparound on - "\r\n" -) +# Match common \xNN / \uNNNN / named ctrl+ / esc sequences in write_pty_keys data. _HEX_BYTE = re.compile(r"\\x([0-9A-Fa-f]{2})") _HEX_U = re.compile(r"\\u([0-9A-Fa-f]{4})") _CTRL = re.compile(r"ctrl\+([a-z@\[\\\]\^_])", re.IGNORECASE) @@ -54,39 +36,30 @@ _KEY_MAP = { } _SIMPLE_ESCAPE_LEN = 2 - -def restore_host_terminal() -> None: - """Best-effort reset of the *user* terminal (stdout), if it is a TTY.""" - try: - out = sys.stdout - if not out.isatty(): - return - buffer = getattr(out, "buffer", None) - if buffer is not None: - with contextlib.suppress(OSError, ValueError): - _ = buffer.write(_HOST_RESTORE.encode("ascii", errors="replace")) - _ = buffer.flush() - return - with contextlib.suppress(OSError, ValueError): - _ = out.write(_HOST_RESTORE) - _ = out.flush() - except AttributeError, OSError, ValueError: - return +# C0 controls except TAB/LF/CR (those stay for readable logs). +_UNSAFE_CTRL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") def sanitize_pty_output_for_tool(data: str) -> str: """Make PTY bytes safe to print on the host as tool/chat text. - Escapes ESC (``\\\\x1b``) so CSI sequences are visible text instead of - control codes when the CLI echoes tool results. + - ESC becomes the two-character sequence ``\\\\x1b`` so CSI is not executed + when the CLI echoes tool results. + - Other C0 controls (except tab/LF/CR) become ``\\\\xHH``. """ - if not data or "\x1b" not in data: + if not data: return data - return data.replace("\x1b", "\\x1b") + + def _esc_ctrl(match: re.Match[str]) -> str: + return f"\\x{ord(match.group(0)):02x}" + + # ESC first as a stable, readable form used in docs/tests. + out = data.replace("\x1b", "\\x1b") + return _UNSAFE_CTRL.sub(_esc_ctrl, out) def decode_write_data(data: str) -> str: - """Expand agent-friendly escapes into raw PTY input. + """Expand agent-friendly escapes into raw PTY input (for ``write_pty_keys`` only). Supported (as literal characters in ``data``): @@ -98,7 +71,6 @@ def decode_write_data(data: str) -> str: if not data: return data - # Order: multi-char named tokens first, then hex, then simple backslash pairs. out: list[str] = [] i = 0 n = len(data) diff --git a/tests/test_tools/test_process.py b/tests/test_tools/test_process.py index c663fe9..167e9ec 100644 --- a/tests/test_tools/test_process.py +++ b/tests/test_tools/test_process.py @@ -4,8 +4,6 @@ import asyncio import sys from pathlib import Path -import pytest - from plyngent.tools.process import ( close_pty, open_pty, @@ -314,82 +312,28 @@ def test_decode_write_data_escapes() -> None: def test_sanitize_pty_output_escapes_csi() -> None: from plyngent.tools.process.pty_terminal import sanitize_pty_output_for_tool - raw = "\x1b[?1049hhello\x1b[0m" + raw = chr(0x1B) + "[?1049hhello" + chr(0x1B) + "[0m" + chr(7) safe = sanitize_pty_output_for_tool(raw) - assert "\x1b" not in safe + assert chr(0x1B) not in safe assert "\\x1b" in safe assert "hello" in safe + assert "\\x07" in safe -def test_close_all_empty_does_not_restore(monkeypatch: pytest.MonkeyPatch) -> None: - """Ctrl-D / chat exit calls close_all with no sessions — must not flash TTY.""" - import plyngent.tools.process.pty_session as ps - - calls: list[int] = [] - - def fake_restore() -> None: - calls.append(1) - - monkeypatch.setattr(ps, "restore_host_terminal", fake_restore) +def test_close_all_empty_is_noop() -> None: + """Ctrl-D / chat exit with no sessions must not touch the host TTY.""" PtyManager.close_all() - assert calls == [] - - -def test_restore_host_terminal_noop_when_not_tty(monkeypatch: pytest.MonkeyPatch) -> None: - import sys - - from plyngent.tools.process.pty_terminal import restore_host_terminal - - class FakeOut: - def isatty(self) -> bool: - return False - - def write(self, _s: str) -> int: - raise AssertionError("should not write") - - monkeypatch.setattr(sys, "stdout", FakeOut()) - restore_host_terminal() # no-op - - -def test_restore_host_terminal_writes_when_tty(monkeypatch: pytest.MonkeyPatch) -> None: - import sys - - from plyngent.tools.process.pty_terminal import restore_host_terminal - - written: list[bytes] = [] - - class Buf: - def write(self, data: bytes) -> int: - written.append(data) - return len(data) - - def flush(self) -> None: - return None - - class FakeOut: - buffer = Buf() - - def isatty(self) -> bool: - return True - - monkeypatch.setattr(sys, "stdout", FakeOut()) - restore_host_terminal() - assert written - blob = b"".join(written) - assert b"\x1b[?1049l" in blob async def test_read_pty_sanitizes_esc(workspace: object) -> None: del workspace try: - # Print ESC so tool-facing read must escape it. - opened = call_sync(open_pty, _py("print('\\x1b[31mred\\x1b[0m')")) + opened = call_sync(open_pty, _py("print(chr(0x1B)+'[31mred'+chr(0x1B)+'[0m')")) session_id = _session_id(opened) text = await call_async(read_pty, session_id, timeout=2.0, until="red") assert "red" in text - # Raw ESC must not appear in tool payload. payload = text.split("--- data ---", 1)[-1] - assert "\x1b" not in payload + assert chr(0x1B) not in payload assert "\\x1b" in payload or "red" in payload _ = await call_async(close_pty, session_id) finally: @@ -424,8 +368,7 @@ def test_write_pty_is_literal() -> None: from plyngent.tools.process.pty_terminal import decode_write_data - # Decoder is intentionally aggressive; that is why write_pty stays literal. - assert decode_write_data("press ctrl+c to cancel") == "press \x03 to cancel" + assert decode_write_data("press ctrl+c to cancel") == "press " + chr(3) + " to cancel" src = inspect.getsource(write_pty.handler) assert "decode_write_data" not in src doc = write_pty.description or write_pty.handler.__doc__ or ""