fix: PTY I/O never blocks the event loop — terminal_run fails cleanly on a stalled tab (v0.9.4)

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
     (`<AGENT>_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.

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.
 

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

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

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"
 

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

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"

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)