diff --git a/PROTO.md b/PROTO.md index 3ebd351..6a02504 100644 --- a/PROTO.md +++ b/PROTO.md @@ -99,6 +99,14 @@ Dev dependencies are in `requirements-dev.txt` plus `ruff` - Operator/agent **secrets are scrubbed** from the child shell's environment (`_API_KEY`, `A2A_AUTH_TOKEN`, …). +- **No PTY I/O on the event loop** (#20, v0.9.4). Reads and writes each run on the + session's own thread (`InputWriter` in `pty_session.py`); a write to a PTY blocks + whenever the child stops reading, and since the child's echo is drained by the loop's + own pump, a write on the loop deadlocks the whole server. Viewer keystrokes are + queued (`write`, never blocks); `terminal_run` types with the bounded `send()` and + fails the call on `PtyWriteTimeout`. Tests for this use real processes + (`tests/test_input_backpressure.py`), never a mocked write. + - **Agent tools are gated by `agent_access`** (`off | read | run`), and **`terminal_run` must never type into a tab where a program holds the foreground** (vim, a build, a REPL) — it would feed keystrokes into that program. diff --git a/README.md b/README.md index 5132bb1..449a198 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,9 @@ The agent can work **with** your terminal, in plain sight — four tools, gated Guardrails: `terminal_run` never types into a tab where a program holds the foreground (vim, a build, a REPL) — it would feed keystrokes into that program. It only uses one of *your* tabs when you ask it to (`session="active"` or an id), and it never switches your -console view. The agent already has a hidden shell; this is the visible one — what it +console view. If a tab stops accepting input (a stuck program that isn't reading its +keyboard), `terminal_run` gives up typing after 10 seconds and says so — terminal I/O +never runs on the server's event loop, so a wedged tab can't freeze the agent. The agent already has a hidden shell; this is the visible one — what it runs, you can watch, scroll back through, and take over. Whatever it reads goes to your model gateway, like any other tool result. diff --git a/api.py b/api.py index 81250a6..f91929b 100644 --- a/api.py +++ b/api.py @@ -367,8 +367,12 @@ async def _bridge(ws, hello: dict, conf) -> None: return queue: asyncio.Queue = asyncio.Queue() - # Label the tab by the shell's LIVE cwd (it may have cd'd since it started). - here = await asyncio.to_thread(cwd_of, sess.pty.pid) if getattr(sess.pty, "pid", None) else "" + # Label a RESUMED tab by the shell's LIVE cwd (it may have cd'd since it started). A + # shell forked a moment ago may not have reached its chdir yet — the OS would still + # report the server's cwd — so a new tab is labelled by where it is starting. + here = "" + if resumed and getattr(sess.pty, "pid", None): + here = await asyncio.to_thread(cwd_of, sess.pty.pid) sess.attach(queue, resumed=resumed, label=display_path(here) if here else "") writer = asyncio.create_task(_write_out(ws, queue)) killed = False diff --git a/protoagent.plugin.yaml b/protoagent.plugin.yaml index 24563b9..b255d8c 100644 --- a/protoagent.plugin.yaml +++ b/protoagent.plugin.yaml @@ -1,6 +1,6 @@ id: terminal name: Terminal -version: 0.9.3 +version: 0.9.4 description: >- A full terminal in the protoAgent console — an xterm.js view wired to a real PTY shell over a WebSocket. The terminal is themed from the protoAgent design system diff --git a/pty_session.py b/pty_session.py index 5e41495..5dfbc54 100644 --- a/pty_session.py +++ b/pty_session.py @@ -7,18 +7,31 @@ the manifest); validated by the CI windows job (tests/test_winpty.py, a real cmd.exe). ``open_session(...)`` picks the right backend for the platform. The POSIX session owns -a child shell behind a pseudo-terminal: read its output off the master fd (in a thread, -so the loop never blocks), write keystrokes, resize (``TIOCSWINSZ``), reap the group. +a child shell behind a pseudo-terminal: read its output off the master fd, write +keystrokes, resize (``TIOCSWINSZ``), reap the group. + +**No PTY I/O ever runs on the event loop.** Reads AND writes each have their own +per-session thread. A write to a PTY blocks whenever the child stops reading its input +(a program in raw mode that never reads, a full input queue) — and, worse, the child can +only consume input as fast as its echo/redraw OUTPUT is drained, which is the loop's own +pump. A write on the loop therefore deadlocks the whole server: the loop waits on the +child, the child waits on the loop (terminal-plugin#20 — a member frozen until restart). """ from __future__ import annotations import asyncio +import concurrent.futures import errno +import logging import os +import select import signal import struct import sys +import threading +from collections import deque +from collections.abc import Callable # Unix PTY primitives — guarded so the module still imports on Windows (which uses the # pywinpty backend). The POSIX PtySession references these only at runtime, on POSIX. @@ -124,8 +137,174 @@ def default_shell() -> str: return os.environ.get("SHELL") or "/bin/bash" +log = logging.getLogger("protoagent.plugins.terminal") + +WRITE_CHUNK = 1024 # bytes per write syscall — small, so a stall is noticed between chunks +MAX_PENDING_INPUT = 4 << 20 # queued-but-unwritten input per session before keystrokes drop +_POLL = 0.2 # seconds the writer thread waits for the PTY to accept input before rechecking + + class PtyError(Exception): - """A PTY lifecycle failure (start / resize).""" + """A PTY lifecycle failure (start / resize / a closed session).""" + + +class PtyWriteTimeout(PtyError): + """The PTY stopped accepting input: ``written`` of ``total`` bytes went in before the + deadline. The rest is abandoned — never typed later, except at most the one chunk + (``WRITE_CHUNK``) already handed to the kernel, which lands if the PTY drains.""" + + def __init__(self, written: int, total: int, timeout: float): + self.written, self.total, self.timeout = written, total, timeout + super().__init__(f"the terminal stopped accepting input — {written} of {total} bytes written in {timeout:g}s") + + +class _Input: + __slots__ = ("data", "done", "written", "cancelled") + + def __init__(self, data: bytes, done: concurrent.futures.Future | None): + self.data, self.done, self.written, self.cancelled = data, done, 0, False + + +class InputWriter: + """Feeds input to a PTY from ONE dedicated thread, strictly in order, so the event + loop never makes a write syscall. ``write_chunk(bytes) -> int`` does the (possibly + blocking) write; ``ready(timeout) -> bool`` says whether the PTY can take input now + (POSIX: ``select``), so the thread notices a close/cancel between waits instead of + sitting in a blocked write. + + ``submit`` never blocks. A submit carrying a Future resolves it when every byte is + in (or fails it); ``cancel`` abandons an item that has not finished.""" + + def __init__(self, write_chunk: Callable[[bytes], int], ready: Callable[[float], bool] | None = None, *, name=""): + self._write_chunk = write_chunk + self._ready = ready or (lambda _t: True) + self._name = name or "terminal-input" + self._cond = threading.Condition() + self._queue: deque[_Input] = deque() + self._pending = 0 # bytes queued but not yet written + self._closed = False + self._warned = False # one warning per backlog-full episode, not one per keystroke + self._thread: threading.Thread | None = None + + @property + def closed(self) -> bool: + return self._closed + + def submit(self, data: bytes, done: concurrent.futures.Future | None = None) -> _Input | None: + """Queue ``data``. None (and ``done`` failed) when the writer is closed or the + backlog is full — keystrokes into a PTY that has stopped reading are dropped + rather than queued without bound.""" + with self._cond: + if self._closed or self._pending + len(data) > MAX_PENDING_INPUT: + reason = "the terminal is closed" if self._closed else "the terminal's input backlog is full" + if done is not None and not done.done(): + done.set_exception(PtyError(reason)) + elif not self._closed and not self._warned: + self._warned = True + log.warning("[terminal] input dropped: %s (the shell isn't reading)", reason) + return None + self._warned = False + item = _Input(data, done) + self._queue.append(item) + self._pending += len(data) + if self._thread is None: + self._thread = threading.Thread(target=self._run, name=self._name, daemon=True) + self._thread.start() + self._cond.notify() + return item + + def cancel(self, item: _Input) -> None: + with self._cond: + item.cancelled = True + + def close(self, join: float = 0.0) -> None: + """Stop: fail whatever is queued and let the thread exit. ``join`` waits that + long for it (so the caller can close the fd only once nothing can write to it).""" + with self._cond: + self._closed = True + pending = list(self._queue) + self._queue.clear() + self._pending = 0 + self._cond.notify_all() + for item in pending: + _fail(item, PtyError("the terminal is closed")) + t = self._thread + if join and t is not None and t is not threading.current_thread(): + t.join(join) + + def _run(self) -> None: + while True: + with self._cond: + while not self._queue and not self._closed: + self._cond.wait() + if self._closed: + return + item = self._queue[0] + err = self._feed(item) + with self._cond: + if self._queue and self._queue[0] is item: + self._queue.popleft() + self._pending -= len(item.data) + if err is not None: # the PTY is gone — nothing queued can be written + self._closed = True + rest = list(self._queue) + self._queue.clear() + self._pending = 0 + if err is not None: + _fail(item, err) + for other in rest: + _fail(other, err) + return + if item.cancelled: + _fail(item, PtyError("cancelled")) + elif item.done is not None and not item.done.done(): + item.done.set_result(item.written) + + def _feed(self, item: _Input) -> Exception | None: + """Write one item, chunk by chunk. None on success or cancel; the error when the + PTY is gone (EIO / EBADF / closed).""" + view = memoryview(item.data) + while item.written < len(view): + if self._closed: + return PtyError("the terminal is closed") + if item.cancelled: + return None + try: + if not self._ready(_POLL): + continue # still full — recheck closed/cancelled, then wait again + n = self._write_chunk(bytes(view[item.written : item.written + WRITE_CHUNK])) + except InterruptedError: + continue + except BlockingIOError: # aclose() switched the fd to non-blocking to drain it + return PtyError("the terminal is closing") + except Exception as exc: # noqa: BLE001 — EIO (child gone), EBADF, a backend error + return PtyError(f"the terminal stopped accepting input: {exc}") + item.written += max(0, int(n or 0)) + return None + + +def _fail(item: _Input, exc: Exception) -> None: + if item.done is not None and not item.done.done(): + item.done.set_exception(exc) + + +async def send_input(writer: InputWriter, data: bytes, timeout: float) -> int: + """Queue ``data`` on ``writer`` and wait (without blocking the loop) until it is all + written. Raises ``PtyWriteTimeout`` after ``timeout`` seconds — the unwritten rest is + abandoned so a half-typed command can't finish typing itself later — and + ``PtyError`` when the terminal is closed or gone.""" + done: concurrent.futures.Future = concurrent.futures.Future() + item = writer.submit(data, done) + fut = asyncio.wrap_future(done) + # After a timeout nobody awaits it any more; consume its late outcome quietly. + fut.add_done_callback(lambda f: f.cancelled() or f.exception()) + if item is None: + return await fut # raises the submit failure + try: + return await asyncio.wait_for(asyncio.shield(fut), timeout) + except asyncio.TimeoutError: + writer.cancel(item) + raise PtyWriteTimeout(item.written, len(data), timeout) from None class PtySession: @@ -153,6 +332,11 @@ def __init__( self.pid: int | None = None self._fd: int | None = None self._exit_code: int | None = None + self._writer: InputWriter | None = None + # Reads get their OWN thread: a pump parks in os.read for the session's whole life, + # and on the loop's shared default executor every open tab would hold one of its + # few workers — enough tabs and the host's to_thread / DNS calls starve. + self._reads: concurrent.futures.ThreadPoolExecutor | None = None # ── env ───────────────────────────────────────────────────────────────────── def _build_env(self) -> dict[str, str]: @@ -187,15 +371,20 @@ def start(self) -> None: os._exit(127) self.pid = pid self._fd = fd + self._writer = InputWriter(self._write_chunk, self._writable, name=f"terminal-input-{pid}") + self._reads = concurrent.futures.ThreadPoolExecutor(1, thread_name_prefix=f"terminal-read-{pid}") self.resize(self.cols, self.rows) async def read(self, n: int = 65536) -> bytes: """A chunk of shell output; ``b""`` on EOF (the child exited). The blocking - read runs in a thread so the event loop keeps serving other sessions.""" - if self._fd is None: + read runs on the session's own thread so the event loop keeps serving.""" + if self._fd is None or self._reads is None: return b"" loop = asyncio.get_running_loop() - return await loop.run_in_executor(None, self._blocking_read, n) + try: + return await loop.run_in_executor(self._reads, self._blocking_read, n) + except RuntimeError: # the read executor was shut down by aclose() + return b"" def _blocking_read(self, n: int) -> bytes: try: @@ -208,15 +397,31 @@ def _blocking_read(self, n: int) -> bytes: raise def write(self, data: str | bytes) -> None: - """Send keystrokes/paste to the shell.""" - if self._fd is None: + """Send keystrokes/paste to the shell. Never blocks: the bytes are queued for the + session's input thread (a shell that has stopped reading can't stall the loop).""" + if self._writer is None or self._fd is None: return - if isinstance(data, str): - data = data.encode("utf-8", "replace") - try: - os.write(self._fd, data) - except OSError: - pass # the shell may have just exited; the read loop will see EOF + self._writer.submit(_as_bytes(data)) + + async def send(self, data: str | bytes, timeout: float = 10.0) -> int: + """Type ``data`` and wait until the PTY has taken all of it. Raises + ``PtyWriteTimeout`` when it stops accepting input for ``timeout`` seconds, and + ``PtyError`` when the session is closed.""" + if self._writer is None or self._fd is None: + raise PtyError("the terminal is closed") + return await send_input(self._writer, _as_bytes(data), timeout) + + def _writable(self, timeout: float) -> bool: + fd = self._fd + if fd is None: + raise PtyError("the terminal is closed") + return bool(select.select([], [fd], [], timeout)[1]) + + def _write_chunk(self, chunk: bytes) -> int: + fd = self._fd + if fd is None: + raise PtyError("the terminal is closed") + return os.write(fd, chunk) def resize(self, cols: int, rows: int) -> None: """Set the PTY window size (``TIOCSWINSZ``) so the shell + TUIs reflow.""" @@ -271,6 +476,12 @@ async def aclose(self) -> int | None: if not await self._wait_exit(1.0): self._signal_group(signal.SIGKILL) await self._wait_exit(2.0) + if self._writer is not None: + # The child is gone, so a write in flight fails (EIO) at once; wait briefly for + # the input thread to let go BEFORE the fd number can be closed and reused. + await asyncio.to_thread(self._writer.close, 1.0) + if self._reads is not None: + self._reads.shutdown(wait=False, cancel_futures=True) if self._fd is not None: try: os.close(self._fd) @@ -351,6 +562,8 @@ def __init__( self.pid: int | None = None self._proc = None self._exit_code: int | None = None + self._writer: InputWriter | None = None + self._reads: concurrent.futures.ThreadPoolExecutor | None = None def _build_env(self) -> dict[str, str]: env = {k: v for k, v in os.environ.items() if k not in self._scrub_env} @@ -358,6 +571,15 @@ def _build_env(self) -> dict[str, str]: env.update(self._env_overrides) return env + def _write_chunk(self, chunk: bytes) -> int: + proc = self._proc + if proc is None: + raise PtyError("the terminal is closed") + # Decode leniently per chunk: a UTF-8 sequence split at a chunk edge degrades to + # U+FFFD rather than failing; WRITE_CHUNK is far above a typed line. + proc.write(chunk.decode("utf-8", "replace")) + return len(chunk) + def start(self) -> None: try: from winpty import PtyProcess # optional dep (requires_pip on Windows) @@ -367,13 +589,15 @@ def start(self) -> None: self.shell, cwd=self.cwd, env=self._build_env(), dimensions=(self.rows, self.cols) ) self.pid = getattr(self._proc, "pid", None) + self._writer = InputWriter(self._write_chunk, name=f"terminal-input-{self.pid}") + self._reads = concurrent.futures.ThreadPoolExecutor(1, thread_name_prefix=f"terminal-read-{self.pid}") async def read(self, n: int = 65536) -> bytes: - if self._proc is None: + if self._proc is None or self._reads is None: return b"" loop = asyncio.get_running_loop() try: - data = await loop.run_in_executor(None, self._proc.read, n) + data = await loop.run_in_executor(self._reads, self._proc.read, n) except EOFError: return b"" except Exception: # noqa: BLE001 — treat any read failure as EOF @@ -383,14 +607,14 @@ async def read(self, n: int = 65536) -> bytes: return data.encode("utf-8", "replace") if isinstance(data, str) else data def write(self, data: str | bytes) -> None: - if self._proc is None: + if self._proc is None or self._writer is None: return - if isinstance(data, bytes): - data = data.decode("utf-8", "replace") - try: - self._proc.write(data) - except Exception: # noqa: BLE001 - pass + self._writer.submit(_as_bytes(data)) + + async def send(self, data: str | bytes, timeout: float = 10.0) -> int: + if self._proc is None or self._writer is None: + raise PtyError("the terminal is closed") + return await send_input(self._writer, _as_bytes(data), timeout) def resize(self, cols: int, rows: int) -> None: self.cols, self.rows = max(1, int(cols)), max(1, int(rows)) @@ -418,10 +642,18 @@ async def aclose(self) -> int | None: self._proc.terminate(force=True) except Exception: # noqa: BLE001 pass + if self._writer is not None: + await asyncio.to_thread(self._writer.close, 1.0) + if self._reads is not None: + self._reads.shutdown(wait=False, cancel_futures=True) self._proc = None return self._exit_code +def _as_bytes(data: str | bytes) -> bytes: + return data.encode("utf-8", "replace") if isinstance(data, str) else bytes(data) + + def open_session(**kw): """Construct the right PTY session for the platform: ``WinPtySession`` on Windows (pywinpty), else the POSIX ``PtySession``.""" diff --git a/pyproject.toml b/pyproject.toml index 4de6c49..7b21aa4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "terminal-plugin" -version = "0.9.3" # keep in lockstep with protoagent.plugin.yaml (a test enforces it) +version = "0.9.4" # keep in lockstep with protoagent.plugin.yaml (a test enforces it) description = "A full terminal (xterm.js + a real PTY over WebSocket) as a protoAgent console plugin." requires-python = ">=3.11" diff --git a/sessions.py b/sessions.py index 33c9e9d..77c5a3b 100644 --- a/sessions.py +++ b/sessions.py @@ -208,8 +208,14 @@ def detach(self, queue: asyncio.Queue) -> bool: # ── input ──────────────────────────────────────────────────────────────────── def write(self, data) -> None: + """Keystrokes from a viewer — queued, never blocks the loop.""" self.pty.write(data) + async def send(self, data, timeout: float = 10.0) -> int: + """Type ``data`` and wait until the shell has taken all of it; raises + ``PtyWriteTimeout`` when it stops accepting input (see pty_session).""" + return await self.pty.send(data, timeout) + def resize(self, cols: int, rows: int) -> None: if self._redraw_pending and rows > 1: # Two size changes → SIGWINCH even when the final size equals the old one, so diff --git a/tests/test_input_backpressure.py b/tests/test_input_backpressure.py new file mode 100644 index 0000000..2156937 --- /dev/null +++ b/tests/test_input_backpressure.py @@ -0,0 +1,178 @@ +"""terminal-plugin#20 — a PTY that stops accepting input must never block the event loop. + +Real PTYs and real processes only (a mocked write would pass whether or not the syscall +can block). Two shapes, both reproduced against v0.9.3, where each froze the loop until +the child was killed: + +- a program in raw mode that never reads stdin — the PTY input queue fills and the + master ``write`` blocks for as long as the program lives; +- a program that DOES read and echoes — it can only take input as fast as its output is + drained, and the drain is the loop's own pump; a blocking write on the loop deadlocks. + +Each scenario runs its own event loop in a worker thread, watched from the test thread: +a stall is reported (and the child killed) instead of hanging the suite. +""" + +from __future__ import annotations + +import asyncio +import os +import signal +import sys +import threading +import time + +import pytest + +from terminal import api, sessions, tools +from terminal.pty_session import PtySession, PtyWriteTimeout +from terminal.sessions import SessionManager + +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX PTY backend") + +# Far beyond any PTY input buffer (macOS ~1 KiB raw queue; Linux up to 640 KiB of +# tty buffers), so a non-reading child is guaranteed to stop accepting it. +BIG = 2 * 1024 * 1024 +MAX_LOOP_GAP = 0.75 # seconds — a blocked loop shows up as a heartbeat gap + + +def _script(tmp_path, name: str, body: str) -> str: + path = tmp_path / name + path.write_text("#!/bin/sh\n" + body + "\n") + path.chmod(0o755) + return str(path) + + +@pytest.fixture +def stops_reading(tmp_path): + """A 'shell' that puts its tty in raw mode, then never reads its input.""" + return _script(tmp_path, "stops_reading.sh", "stty raw -echo\nexec sleep 600") + + +@pytest.fixture +def raw_echo(tmp_path): + """A 'shell' that reads every byte and echoes it (raw mode + echo, then cat).""" + return _script(tmp_path, "raw_echo.sh", "stty raw\nexec cat") + + +def _run_watched(scenario, *, limit: float = 30.0): + """Run ``scenario(state)`` (an async fn) on a fresh loop in a worker thread, with a + heartbeat task beside it. Returns ``(result, max_heartbeat_gap)``. If the loop wedges, + kill the child whose pid the scenario stored in ``state['pids']`` and fail.""" + state: dict = {"pids": [], "beat": time.monotonic(), "gap": 0.0} + out: dict = {} + + async def main(): + async def heartbeat(): + while True: + now = time.monotonic() + state["gap"] = max(state["gap"], now - state["beat"]) + state["beat"] = now + await asyncio.sleep(0.02) + + hb = asyncio.create_task(heartbeat()) + try: + out["result"] = await scenario(state) + except BaseException as exc: # noqa: BLE001 — surfaced in the test thread + out["error"] = exc + finally: + hb.cancel() + + th = threading.Thread(target=lambda: asyncio.run(main()), daemon=True) + th.start() + th.join(limit) + if th.is_alive(): + silent = time.monotonic() - state["beat"] + for pid in state["pids"]: + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + th.join(5) + pytest.fail(f"the event loop wedged — no heartbeat for {silent:.1f}s (terminal-plugin#20)") + if "error" in out: + raise out["error"] + return out.get("result"), state["gap"] + + +def test_a_pty_that_stops_reading_fails_the_write_without_blocking_the_loop(stops_reading): + async def scenario(state): + sess = PtySession(shell=stops_reading) + sess.start() + state["pids"].append(sess.pid) + try: + await asyncio.sleep(0.3) # let it reach `stty raw` + sleep + t0 = time.monotonic() + sess.write(b"k" * 4096) # fire-and-forget keystrokes: returns at once + assert time.monotonic() - t0 < 0.1 + t0 = time.monotonic() + with pytest.raises(PtyWriteTimeout) as exc: + await sess.send(b"x" * BIG, timeout=1.0) + took = time.monotonic() - t0 + assert 0.9 <= took < 3.0, took + assert exc.value.total == BIG and exc.value.written < BIG + assert "stopped accepting input" in str(exc.value) + finally: + t0 = time.monotonic() + await asyncio.wait_for(sess.aclose(), 8) + closed_in = time.monotonic() - t0 + return closed_in + + closed_in, gap = _run_watched(scenario) + assert gap < MAX_LOOP_GAP, f"the loop stalled {gap:.2f}s" + assert closed_in < 5.0 # the input thread let go once the child died + + +def test_a_big_write_to_an_echoing_pty_cannot_deadlock_the_pump(raw_echo): + """The deadlock shape: the child echoes, so it stops reading input until its output is + drained — and the session's pump (on the loop) is what drains it.""" + + async def scenario(state): + mgr = SessionManager() + sess = mgr.create(shell=raw_echo) + state["pids"].append(sess.pty.pid) + try: + await asyncio.sleep(0.3) + payload = b"y" * (256 * 1024) + written = await sess.send(payload, timeout=20) + assert written == len(payload) + # cat writes everything back (plus the tty echo): wait for the pump to see it + deadline = time.monotonic() + 10 + while sess.emitted < len(payload) and time.monotonic() < deadline: + await asyncio.sleep(0.05) + return sess.emitted + finally: + await asyncio.wait_for(mgr.close_all(), 8) + + emitted, gap = _run_watched(scenario) + assert emitted >= 256 * 1024 + assert gap < MAX_LOOP_GAP, f"the loop stalled {gap:.2f}s" + + +def test_terminal_run_fails_cleanly_when_the_tab_stops_accepting_input(stops_reading, registry, monkeypatch): + """The tool from the issue: it returns a clear error well inside its own budget and + the loop keeps serving meanwhile.""" + monkeypatch.setattr(tools, "TYPE_TIMEOUT", 1.0) + + async def scenario(state): + mgr = SessionManager() + for mod in (sessions, tools, api): + monkeypatch.setattr(mod, "MANAGER", mgr) + cfg = {"shell": stops_reading, "login_shell": False} + t = {x.name: x for x in tools.build_tools(registry, lambda: api.resolve(cfg))} + try: + run = asyncio.create_task(t["terminal_run"].ainvoke({"command": "echo " + "z" * BIG, "wait_seconds": 30})) + while mgr.agent_session() is None and not run.done(): + await asyncio.sleep(0.01) + if mgr.agent_session() is not None: + state["pids"].append(mgr.agent_session().pty.pid) + t0 = time.monotonic() + out = await asyncio.wait_for(run, 15) + return out, time.monotonic() - t0 + finally: + await asyncio.wait_for(mgr.close_all(), 8) + + (out, took), gap = _run_watched(scenario) + assert "stopped accepting input" in out and "session='new'" in out + assert took < 8.0 # ≈ settle (≤3s) + the 1s typing budget, never the 30s wait + assert gap < MAX_LOOP_GAP, f"the loop stalled {gap:.2f}s" diff --git a/tools.py b/tools.py index 818f118..2e94010 100644 --- a/tools.py +++ b/tools.py @@ -26,11 +26,13 @@ import time from .procinfo import command_name, cwd_of +from .pty_session import PtyError, PtyWriteTimeout from .sessions import MANAGER, SessionLimitError from .textutil import clean, tail_lines MAX_OUTPUT_CHARS = 12_000 # what one tool result may carry back into the context AGENT_TAB = "Agent" +TYPE_TIMEOUT = 10.0 # seconds the tab may take to accept the typed command line ACCESS_LEVELS = ("off", "read", "run") @@ -202,7 +204,20 @@ async def terminal_run(command: str, session: str = "agent", cwd: str = "", wait ) line = f"cd {shlex.quote(os.path.expanduser(cwd))} && {command}" if cwd else command start = sess.emitted - sess.write(line + "\r") + try: + # Off-loop and bounded: a tab that has stopped reading its input must fail the + # call, never stall it (or, before #20, the whole server's event loop). + await sess.send(line + "\r", TYPE_TIMEOUT) + except PtyWriteTimeout as exc: + return ( + f"Terminal {sess.id} stopped accepting input — only {exc.written} of {exc.total} bytes of " + f"the command went in within {exc.timeout:g}s, so the rest was dropped and nothing ran " + "cleanly. Something in that tab isn't reading its keyboard (a stuck program or a full input " + "buffer). Check it with terminal_read, or use session='new'. Keep very long scripts in a file " + "and run the file instead of typing them." + ) + except PtyError as exc: + return f"Couldn't type into terminal {sess.id}: {exc}. terminal_list shows what's open." finished = await _wait_for_prompt(sess, max(0.0, min(float(wait_seconds), 300.0))) raw, complete = sess.text_since(start) out = clean(raw)