fix(dashboard): prevent PTY input from blocking event loop (#93565)

* fix(dashboard): prevent PTY input from blocking event loop

* fix(win-pty): don't terminate a healthy ConPTY on write cancellation; log leaked write workers

Review follow-up to the backpressure fix.

CancelledError on WinPtyBridge.write() ran the same path as a timeout and
force-terminated the ConPTY. Cancellation means the owning socket went away
mid-write, which is the keep-alive session's normal reattach case, not a
wedged child; killing the process there defeats the PTY-outlives-socket
design. Give the in-flight write the shutdown grace window and only
terminate if it never lands.

When terminate() fails to unblock pywinpty, the worker stays parked in the
default executor. That was swallowed by a bare except; log it so a slow
thread-pool starvation is diagnosable.

---------

Co-authored-by: Austin Pickett <pickett.austin@gmail.com>

hermes_cli/pty_bridge.py

@@ -8,6 +8,7 @@ Windows would need a separate ConPTY/``pywinpty`` implementation).
 
 from __future__ import annotations
 
+import asyncio
 import errno
 import fcntl
 import os
@@ -62,13 +63,16 @@ class PtyUnavailableError(RuntimeError):
 
 class PtyBridge:
     """Thin wrapper around ``ptyprocess.PtyProcess`` for byte streaming. Not thread-safe: owned by
-    the WebSocket handler that spawned it; reads run in an executor thread, writes on the loop.
+    the WebSocket handler that spawned it; reads run in an executor thread, writes are awaited on
+    the loop. The master fd is non-blocking so input backpressure suspends only the owning
+    WebSocket task, never the dashboard event loop.
     """
 
     def __init__(self, proc: "ptyprocess.PtyProcess"):  # type: ignore[name-defined]
         self._proc = proc
         self._fd: int = proc.fd
         self._closed = False
+        os.set_blocking(self._fd, False)
 
     @classmethod
     def is_available(cls) -> bool:
@@ -130,25 +134,75 @@ class PtyBridge:
             # EIO on Linux = slave side closed.  EBADF = already closed.
             if exc.errno in {errno.EIO, errno.EBADF}:
                 return None
+            # The fd is deliberately non-blocking. Readiness can disappear
+            # between select() and os.read() when close/output races occur.
+            if exc.errno in {errno.EAGAIN, errno.EWOULDBLOCK}:
+                return b""
             raise
         return data or None
 
-    def write(self, data: bytes) -> None:
-        """Write raw bytes to the PTY master (i.e. the child's stdin)."""
-        if self._closed or not data:
-            return
-        # os.write can return a short write under load; loop until drained.
+    async def _wait_writable(self, timeout: float) -> bool:
+        """Wait without blocking the event loop until the master accepts input."""
+        if self._closed or timeout <= 0:
+            return False
+        loop = asyncio.get_running_loop()
+        ready = loop.create_future()
+
+        def _mark_ready() -> None:
+            if not ready.done():
+                ready.set_result(None)
+
+        try:
+            loop.add_writer(self._fd, _mark_ready)
+            await asyncio.wait_for(ready, timeout=timeout)
+            return not self._closed
+        except (asyncio.TimeoutError, OSError, ValueError):
+            return False
+        finally:
+            try:
+                loop.remove_writer(self._fd)
+            except (OSError, ValueError):
+                pass
+
+    async def write(self, data: bytes, *, timeout: float = 10.0) -> bool:
+        """Write all raw bytes without ever blocking the dashboard event loop.
+
+        Returns ``False`` when the bridge closes or the child leaves its input
+        buffer full for ``timeout`` seconds. Callers can then recycle only the
+        affected terminal session while the rest of the dashboard stays live.
+        """
+        if self._closed:
+            return False
+        if not data:
+            return True
+
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + max(0.0, timeout)
         view = memoryview(data)
         while view:
+            if self._closed:
+                return False
             try:
                 n = os.write(self._fd, view)
             except OSError as exc:
                 if exc.errno in {errno.EIO, errno.EBADF, errno.EPIPE}:
-                    return
-                raise
-            if n <= 0:
-                return
-            view = view[n:]
+                    return False
+                if exc.errno in {errno.EAGAIN, errno.EWOULDBLOCK}:
+                    n = 0
+                else:
+                    raise
+            if n > 0:
+                view = view[n:]
+                # A very large paste can otherwise monopolize the loop while
+                # the child drains quickly enough to keep the fd writable.
+                if view:
+                    await asyncio.sleep(0)
+                continue
+
+            remaining = deadline - loop.time()
+            if not await self._wait_writable(remaining):
+                return False
+        return True
 
     def resize(self, cols: int, rows: int) -> None:
         """Forward a terminal resize to the child via ``TIOCSWINSZ``.

hermes_cli/pty_session.py

@@ -52,7 +52,9 @@ class PtySession:
         self.last_detached_at: Optional[float] = None
         self._read_timeout = read_timeout
         self._ws = None
+        self._attach_generation = 0
         self._drain_task: Optional[asyncio.Task] = None
+        self._write_lock = asyncio.Lock()
 
     async def start(self) -> None:
         self._drain_task = asyncio.create_task(self._drain())
@@ -75,7 +77,25 @@ class PtySession:
             except Exception:
                 pass                                 # detached mid-send; keep buffering
 
-    async def attach(self, ws, *, force_redraw: bool = False) -> None:
+    async def write(self, ws, data: bytes) -> bool:
+        """Serialize input and discard bytes from a superseded socket."""
+        async with self._write_lock:
+            if self._ws is not ws:
+                return True
+            generation = self._attach_generation
+            delivered = await self.bridge.write(data)
+            # A replacement socket can attach while the bridge write is
+            # suspended on backpressure. A late failure from the superseded
+            # socket must not poison the replacement's shared PTY session.
+            if (
+                not delivered
+                and self._ws is ws
+                and self._attach_generation == generation
+            ):
+                self.alive = False
+            return delivered
+
+    async def attach(self, ws, *, force_redraw: bool = False) -> bool:
         """Attach a browser terminal and replay buffered PTY output.
 
         The TUI renders differentially on an alternate screen, so a bounded ANSI tail is not a
@@ -84,12 +104,14 @@ class PtySession:
         if self._ws is not ws:
             await _close_ws(self._ws, WS_CLOSE_SUPERSEDED)
         self._ws = ws
+        self._attach_generation += 1
         self.attached = True
         self.last_detached_at = None
         if snap := self.buffer.snapshot():
             await ws.send_bytes(snap)
         if force_redraw:
-            self.bridge.write(TUI_FORCE_REDRAW)
+            return await self.write(ws, TUI_FORCE_REDRAW)
+        return True
 
     def detach(self, ws) -> None:
         # Only the currently-attached socket may mark the session detached: a superseded socket's
@@ -102,6 +124,7 @@ class PtySession:
         self.last_detached_at = time.monotonic()
 
     async def close(self) -> None:
+        self.alive = False
         if self._drain_task is not None:
             self._drain_task.cancel()
             try:

hermes_cli/web_routers/chat_ws.py

@@ -18,7 +18,8 @@ from fastapi import APIRouter, FastAPI, HTTPException, WebSocket, WebSocketDisco
 from hermes_cli.pty_session import RegistryFull
 from hermes_cli.web_deps import LateState, late
 from hermes_cli.web_server_chat import (
-    _build_sidecar_url, _get_console_executor, _legacy_pump, _ws_auth_ok, _ws_request_is_allowed,
+    _build_sidecar_url, _close_stalled_pty_input, _get_console_executor, _legacy_pump, _ws_auth_ok,
+    _ws_request_is_allowed,
 )
 
 _log = logging.getLogger("hermes_cli.web_server")
@@ -492,7 +493,10 @@ async def pty_ws(ws: WebSocket) -> None:
 
     # A fresh xterm can't rebuild the TUI from an arbitrary tail of alternate-
     # screen differential output; reused PTYs emit a full frame after replay.
-    await session.attach(ws, force_redraw=not _created)
+    if not await session.attach(ws, force_redraw=not _created):
+        await _close_stalled_pty_input(ws, path="keepalive-redraw")
+        PTY_REGISTRY.detach(attach_token, ws)
+        return
 
     # Writer loop only: the session's drain task (one per PTY, inside the
     # registry) forwards output to whichever socket is attached and ring-buffers
@@ -517,7 +521,9 @@ async def pty_ws(ws: WebSocket) -> None:
             if match and match.end() == len(raw):
                 session.bridge.resize(cols=int(match.group(1)), rows=int(match.group(2)))
                 continue
-            session.bridge.write(raw)
+            if not await session.write(ws, raw):
+                await _close_stalled_pty_input(ws, path="keepalive")
+                break
     except WebSocketDisconnect:
         pass
     finally:

hermes_cli/web_server_chat.py

@@ -49,6 +49,15 @@ PTY_REGISTRY = PtySessionRegistry(
     ttl=30 * 60, max_sessions=16, buffer_cap=1 * 1024 * 1024, read_timeout=_PTY_READ_CHUNK_TIMEOUT)
 
 
+async def _close_stalled_pty_input(ws: "WebSocket", *, path: str) -> None:
+    """Close only the terminal socket when its child stops accepting input."""
+    _log.warning("pty input stalled path=%s; recycling terminal session", path)
+    try:
+        await ws.close(code=1013, reason="PTY input stalled")
+    except Exception:
+        pass
+
+
 async def _legacy_pump(ws: "WebSocket", bridge) -> None:
     """Original 1:1 socket<->PTY pump: stream until disconnect, then close the
     bridge. Used when no ``?attach=`` token is supplied (keep-alive opt-in).
@@ -108,7 +117,9 @@ async def _legacy_pump(ws: "WebSocket", bridge) -> None:
             if match and match.end() == len(raw):
                 bridge.resize(cols=int(match.group(1)), rows=int(match.group(2)))
                 continue
-            bridge.write(raw)
+            if not await bridge.write(raw):
+                await _close_stalled_pty_input(ws, path="legacy")
+                break
     except WebSocketDisconnect:
         pass
     finally:

hermes_cli/win_pty_bridge.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
 import sys
 import time
 from typing import Optional, Sequence
@@ -13,6 +15,8 @@ except ImportError:  # pragma: no cover - non-Windows or pywinpty missing
     PtyProcess = None  # type: ignore
     _PTY_AVAILABLE = False
 
+_log = logging.getLogger(__name__)
+
 
 __all__ = ["WinPtyBridge", "PtyUnavailableError"]
 
@@ -21,6 +25,7 @@ __all__ = ["WinPtyBridge", "PtyUnavailableError"]
 _MIN_DIMENSION = 1
 _MAX_COLS = 2000
 _MAX_ROWS = 1000
+_WRITE_SHUTDOWN_GRACE = 1.0
 
 
 def _clamp(value: int, maximum: int) -> int:
@@ -37,7 +42,8 @@ class PtyUnavailableError(RuntimeError):
 
 class WinPtyBridge:
     """pywinpty-backed bridge with the same interface as ``PtyBridge``. ``read`` runs inside
-    ``run_in_executor``; ConPTY has no selectable fd, so it polls with a short sleep."""
+    ``run_in_executor``; ConPTY has no selectable fd, so reads poll and writes run in a worker
+    thread to keep the same non-blocking event-loop contract as the POSIX bridge."""
 
     def __init__(self, proc: "PtyProcess") -> None:  # type: ignore[name-defined]
         self._proc = proc
@@ -92,13 +98,75 @@ class WinPtyBridge:
         # xterm.js tolerates the rare replacement char (the one fidelity tradeoff vs POSIX).
         return data.encode("utf-8", errors="replace")
 
-    def write(self, data: bytes) -> None:
-        if self._closed or not data:
-            return
+    def _write_blocking(self, data: bytes) -> bool:
+        if self._closed:
+            return False
+        if not data:
+            return True
         try:
             self._proc.write(data.decode("utf-8", errors="replace"))  # pywinpty wants text
         except Exception:
+            return False
+        return True
+
+    async def write(self, data: bytes, *, timeout: float = 10.0) -> bool:
+        """Write off-loop and tear down ConPTY when its input pipe wedges.
+
+        ``wait_for(to_thread(...))`` alone only cancels the asyncio wrapper;
+        the worker remains blocked inside pywinpty. Keep the worker future,
+        force-close the ConPTY on timeout, and wait briefly for that close to
+        release the blocked write before returning.
+
+        Cancellation (the owning socket went away mid-write) is not evidence
+        of a wedged child: the PTY outlives its socket by design, so give the
+        in-flight write the grace window and only terminate if it never lands.
+        """
+        if self._closed:
+            return False
+        if not data:
+            return True
+        loop = asyncio.get_running_loop()
+        write_future = loop.run_in_executor(None, self._write_blocking, data)
+        try:
+            return await asyncio.wait_for(
+                asyncio.shield(write_future),
+                timeout=max(0.0, timeout),
+            )
+        except asyncio.TimeoutError:
+            await self._stop_stalled_write(write_future)
+            return False
+        except asyncio.CancelledError:
+            await asyncio.shield(self._settle_or_stop_write(write_future))
+            raise
+
+    async def _settle_or_stop_write(self, write_future: asyncio.Future) -> None:
+        """Let a cancelled write finish within the grace window; terminate only if it stalls."""
+        try:
+            await asyncio.wait_for(asyncio.shield(write_future), timeout=_WRITE_SHUTDOWN_GRACE)
             return
+        except asyncio.TimeoutError:
+            pass
+        except Exception:
+            return
+        await self._stop_stalled_write(write_future)
+
+    async def _stop_stalled_write(self, write_future: asyncio.Future) -> None:
+        """Close ConPTY and reap the worker that was blocked in ``write``."""
+        await asyncio.to_thread(self.close)
+        try:
+            await asyncio.wait_for(
+                asyncio.shield(write_future),
+                timeout=_WRITE_SHUTDOWN_GRACE,
+            )
+        except asyncio.TimeoutError:
+            # The worker is still parked inside pywinpty after terminate(); it
+            # now occupies a default-executor thread until the process exits.
+            _log.warning(
+                "ConPTY write worker did not exit within %.1fs of terminate(); thread leaked",
+                _WRITE_SHUTDOWN_GRACE,
+            )
+        except Exception:
+            pass
 
     def resize(self, cols: int, rows: int) -> None:
         if self._closed:

tests/hermes_cli/test_pty_bridge.py

@@ -6,7 +6,10 @@ printf) to verify it behaves like a PTY you can read/write/resize/close.
 
 from __future__ import annotations
 
+import asyncio
+import errno
 import os
+import select
 import shutil
 import signal
 import sys
@@ -70,17 +73,91 @@ class TestPtyBridgeSpawn:
 @skip_on_windows
 class TestPtyBridgeIO:
 
-    def test_write_sends_to_child_stdin(self):
+    @pytest.mark.asyncio
+    async def test_write_sends_to_child_stdin(self):
         # `cat` with no args echoes stdin back to stdout.  We write a line,
         # read it back, then signal EOF to let cat exit cleanly.
         bridge = PtyBridge.spawn([shutil.which("cat") or "cat"])
         try:
-            bridge.write(b"hello-pty\n")
+            assert await bridge.write(b"hello-pty\n") is True
             output = _read_until(bridge, b"hello-pty")
             assert b"hello-pty" in output
         finally:
             bridge.close()
 
+    @pytest.mark.asyncio
+    async def test_write_yields_while_input_is_backpressured(self, monkeypatch):
+        bridge = PtyBridge.__new__(PtyBridge)
+        bridge._fd = 123
+        bridge._closed = False
+        wait_started = asyncio.Event()
+        release_write = asyncio.Event()
+        write_calls = 0
+
+        def fake_write(fd, data):
+            nonlocal write_calls
+            assert fd == 123
+            write_calls += 1
+            if write_calls == 1:
+                raise BlockingIOError(errno.EAGAIN, "buffer full")
+            return len(data)
+
+        async def fake_wait_writable(timeout):
+            assert timeout > 0
+            wait_started.set()
+            await release_write.wait()
+            return True
+
+        monkeypatch.setattr(os, "write", fake_write)
+        monkeypatch.setattr(bridge, "_wait_writable", fake_wait_writable)
+
+        write_task = asyncio.create_task(bridge.write(b"queued input"))
+        await wait_started.wait()
+
+        # The stalled PTY write yielded control instead of pinning asyncio.
+        heartbeat_ran = False
+
+        async def heartbeat():
+            nonlocal heartbeat_ran
+            heartbeat_ran = True
+
+        await heartbeat()
+        assert heartbeat_ran is True
+
+        release_write.set()
+        assert await write_task is True
+        assert write_calls == 2
+
+    @pytest.mark.asyncio
+    async def test_write_reports_sustained_backpressure(self, monkeypatch):
+        bridge = PtyBridge.__new__(PtyBridge)
+        bridge._fd = 123
+        bridge._closed = False
+
+        def fake_write(_fd, _data):
+            raise BlockingIOError(errno.EAGAIN, "buffer full")
+
+        async def never_writable(_timeout):
+            return False
+
+        monkeypatch.setattr(os, "write", fake_write)
+        monkeypatch.setattr(bridge, "_wait_writable", never_writable)
+
+        assert await bridge.write(b"queued input") is False
+
+    def test_read_treats_nonblocking_read_race_as_idle(self, monkeypatch):
+        bridge = PtyBridge.__new__(PtyBridge)
+        bridge._fd = 123
+        bridge._closed = False
+
+        monkeypatch.setattr(select, "select", lambda *_args: ([123], [], []))
+
+        def would_block(_fd, _size):
+            raise BlockingIOError(errno.EAGAIN, "try again")
+
+        monkeypatch.setattr(os, "read", would_block)
+        assert bridge.read(timeout=0.01) == b""
+
     def test_read_returns_none_after_child_exits(self):
         bridge = PtyBridge.spawn(["/bin/sh", "-c", "printf done"])
         try:

tests/hermes_cli/test_web_server_pty_idle_backoff.py

@@ -20,8 +20,8 @@ class _FakeBridge:
             return self._reads.pop(0)
         return None
 
-    def write(self, data):
-        pass
+    async def write(self, data):
+        return True
 
     def resize(self, cols, rows):
         pass

tests/hermes_cli/test_win_pty_bridge.py

@@ -14,15 +14,20 @@ actually live on native Windows.
 
 from __future__ import annotations
 
+import asyncio
+import logging
 import os
 import sys
+import threading
 import time
+from unittest.mock import patch
 
 import pytest
 
 # WinPtyBridge can be imported on every platform — ``is_available`` just
 # returns False when pywinpty isn't usable.  Importing the module itself
 # must never raise, otherwise the web_server import branch becomes a trap.
+from hermes_cli import win_pty_bridge
 from hermes_cli.win_pty_bridge import PtyUnavailableError, WinPtyBridge
 
 # ``pytest.mark.windows_only`` rather than a local ``skipif`` alias: the
@@ -70,6 +75,133 @@ class TestWinPtyBridgeUnavailable:
         assert WinPtyBridge is not None
         assert callable(WinPtyBridge.is_available)
 
+    @pytest.mark.asyncio
+    async def test_write_has_nonblocking_async_contract(self):
+        class _FakeProc:
+            pid = 1
+
+            def __init__(self):
+                self.written = []
+
+            def write(self, text):
+                self.written.append(text)
+
+        proc = _FakeProc()
+        bridge = WinPtyBridge(proc)
+
+        assert await bridge.write(b"hello") is True
+        assert proc.written == ["hello"]
+
+    @pytest.mark.asyncio
+    async def test_write_timeout_terminates_conpty_and_reaps_worker(self):
+        class _BlockingProc:
+            pid = 1
+
+            def __init__(self):
+                self.write_started = threading.Event()
+                self.release_write = threading.Event()
+                self.write_finished = threading.Event()
+                self.terminated = threading.Event()
+
+            def write(self, _text):
+                self.write_started.set()
+                self.release_write.wait(timeout=2.0)
+                self.write_finished.set()
+
+            def terminate(self, force=False):
+                assert force is True
+                self.terminated.set()
+                self.release_write.set()
+
+        proc = _BlockingProc()
+        bridge = WinPtyBridge(proc)
+
+        assert await bridge.write(b"blocked", timeout=0.01) is False
+        assert proc.write_started.is_set()
+        assert proc.terminated.is_set()
+        assert proc.write_finished.is_set()
+        assert bridge._closed is True
+
+    @pytest.mark.asyncio
+    async def test_cancelled_write_keeps_healthy_conpty_alive(self):
+        """A socket dropping mid-write is not a wedged child: the PTY outlives
+        its socket by design, so a write that lands within the grace window
+        must not terminate the process."""
+        class _SlowProc:
+            pid = 1
+
+            def __init__(self):
+                self.release_write = threading.Event()
+                self.terminated = threading.Event()
+
+            def write(self, _text):
+                self.release_write.wait(timeout=2.0)
+
+            def terminate(self, force=False):
+                self.terminated.set()
+
+        proc = _SlowProc()
+        bridge = WinPtyBridge(proc)
+        task = asyncio.create_task(bridge.write(b"x"))
+        await asyncio.sleep(0.05)
+        task.cancel()
+        proc.release_write.set()  # the child drains right after the socket left
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert not proc.terminated.is_set()
+        assert bridge._closed is False
+
+    @pytest.mark.asyncio
+    async def test_cancelled_write_that_never_lands_terminates_conpty(self):
+        class _WedgedProc:
+            pid = 1
+
+            def __init__(self):
+                self.release_write = threading.Event()
+                self.terminated = threading.Event()
+
+            def write(self, _text):
+                self.release_write.wait(timeout=5.0)
+
+            def terminate(self, force=False):
+                self.terminated.set()
+                self.release_write.set()
+
+        proc = _WedgedProc()
+        bridge = WinPtyBridge(proc)
+        with patch.object(win_pty_bridge, "_WRITE_SHUTDOWN_GRACE", 0.05):
+            task = asyncio.create_task(bridge.write(b"x"))
+            await asyncio.sleep(0.02)
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+        assert proc.terminated.is_set()
+        assert bridge._closed is True
+
+    @pytest.mark.asyncio
+    async def test_leaked_write_worker_is_logged(self, caplog):
+        """terminate() that fails to unblock pywinpty leaves a thread parked in
+        the default executor; that must be observable, not swallowed."""
+        class _StuckProc:
+            pid = 1
+
+            def __init__(self):
+                self.release_write = threading.Event()
+
+            def write(self, _text):
+                self.release_write.wait(timeout=5.0)
+
+            def terminate(self, force=False):
+                pass  # does NOT release the write
+
+        proc = _StuckProc()
+        bridge = WinPtyBridge(proc)
+        with patch.object(win_pty_bridge, "_WRITE_SHUTDOWN_GRACE", 0.05), \
+                caplog.at_level(logging.WARNING, logger="hermes_cli.win_pty_bridge"):
+            assert await bridge.write(b"x", timeout=0.01) is False
+        proc.release_write.set()
+        assert any("thread leaked" in r.getMessage() for r in caplog.records)
+
     @pytest.mark.skipif(sys.platform.startswith("win"), reason="non-Windows only")
     def test_spawn_raises_unavailable_off_windows(self):
         with pytest.raises(PtyUnavailableError):
@@ -101,7 +233,8 @@ class TestWinPtyBridgeSpawn:
 @pytest.mark.windows_only
 class TestWinPtyBridgeIO:
 
-    def test_write_sends_to_child_stdin(self):
+    @pytest.mark.asyncio
+    async def test_write_sends_to_child_stdin(self):
         # python -c reads stdin, echoes a marker, exits.  More reliable than
         # ``cat`` (not on Windows) and doesn't depend on a particular shell.
         script = (
@@ -112,7 +245,7 @@ class TestWinPtyBridgeIO:
         )
         bridge = WinPtyBridge.spawn([sys.executable, "-c", script])
         try:
-            bridge.write(b"hello-pty\r\n")
+            assert await bridge.write(b"hello-pty\r\n") is True
             output = _read_until(bridge, b"GOT:hello-pty")
             assert b"GOT:hello-pty" in output
         finally:
@@ -248,4 +381,3 @@ class TestWinPtyBridgeEnv:
             assert b"pty-env-works" in output
         finally:
             bridge.close()
-

tests/test_pty_keepalive_ws.py

@@ -9,13 +9,17 @@ import hermes_cli.web_server_chat as _web_server_chat
 class FakeBridge:
     def __init__(self):
         self.alive = True
+        self.accept_input = True
         self.written = bytearray()
 
     def read(self, timeout):
         return b""        # idle forever
 
-    def write(self, data):
+    async def write(self, data):
+        if not self.accept_input:
+            return False
         self.written.extend(data)
+        return True
 
     def resize(self, cols, rows):
         pass
@@ -70,6 +74,25 @@ async def test_attach_token_reuses_same_session(pty_keepalive_harness):
     assert bytes(pty_keepalive_harness.bridges[0].written) == b"hi\x0cagain"
 
 
+@pytest.mark.asyncio
+async def test_stalled_input_closes_only_the_keepalive_socket(
+    pty_keepalive_harness,
+):
+    from starlette.testclient import TestClient
+    from starlette.websockets import WebSocketDisconnect
+
+    client = TestClient(web_server.app)
+    with client.websocket_connect("/api/pty?attach=TOK1") as ws:
+        bridge = pty_keepalive_harness.bridges[0]
+        bridge.accept_input = False
+        ws.send_bytes(b"input")
+        with pytest.raises(WebSocketDisconnect) as exc_info:
+            ws.receive_bytes()
+
+    assert exc_info.value.code == 1013
+    assert web_server.PTY_REGISTRY._sessions["TOK1"].alive is False
+
+
 @pytest.mark.asyncio
 async def test_attach_token_reuses_same_resume(pty_keepalive_harness):
     from starlette.testclient import TestClient

tests/test_pty_session.py

@@ -26,9 +26,10 @@ def test_ringbuffer_drops_oldest_over_capacity():
 class FakeBridge:
     """Implements the bridge contract PtySession depends on."""
 
-    def __init__(self, chunks):
+    def __init__(self, chunks, *, write_result=True):
         self._chunks = list(chunks)   # bytes; b"" = idle tick; None = EOF
         self.written = bytearray()
+        self.write_result = write_result
         self.closed = False
         self.resized = None
 
@@ -37,8 +38,10 @@ class FakeBridge:
             return b""                # idle
         return self._chunks.pop(0)
 
-    def write(self, data):
-        self.written.extend(data)
+    async def write(self, data):
+        if self.write_result:
+            self.written.extend(data)
+        return self.write_result
 
     def resize(self, cols, rows):
         self.resized = (cols, rows)
@@ -87,7 +90,7 @@ async def test_reattach_can_force_complete_tui_redraw_after_replay():
     await asyncio.sleep(0.05)
 
     ws = FakeWS()
-    await s.attach(ws, force_redraw=True)
+    assert await s.attach(ws, force_redraw=True) is True
 
     replay = b"".join(p for kind, p in ws.sent if kind == "bytes")
     assert replay == b"partial differential frame"
@@ -95,6 +98,101 @@ async def test_reattach_can_force_complete_tui_redraw_after_replay():
     await s.close()
 
 
+@pytest.mark.asyncio
+async def test_failed_redraw_marks_session_dead_for_replacement():
+    from hermes_cli.pty_session import PtySession
+
+    bridge = FakeBridge([b""], write_result=False)
+    s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01)
+    await s.start()
+    ws = FakeWS()
+
+    assert await s.attach(ws, force_redraw=True) is False
+    assert s.alive is False
+    await s.close()
+
+
+@pytest.mark.asyncio
+async def test_session_serializes_input_across_socket_tasks():
+    from hermes_cli.pty_session import PtySession
+
+    class OrderedBridge(FakeBridge):
+        def __init__(self):
+            super().__init__([b""])
+            self.first_started = asyncio.Event()
+            self.release_first = asyncio.Event()
+
+        async def write(self, data):
+            if not self.written:
+                self.first_started.set()
+                await self.release_first.wait()
+            self.written.extend(data)
+            return True
+
+    bridge = OrderedBridge()
+    s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01)
+    await s.start()
+    ws = FakeWS()
+    await s.attach(ws)
+
+    first = asyncio.create_task(s.write(ws, b"first"))
+    await bridge.first_started.wait()
+    second = asyncio.create_task(s.write(ws, b"second"))
+    await asyncio.sleep(0)
+    assert bytes(bridge.written) == b""
+
+    bridge.release_first.set()
+    assert await first is True
+    assert await second is True
+    assert bytes(bridge.written) == b"firstsecond"
+    await s.close()
+
+
+@pytest.mark.asyncio
+async def test_superseded_failed_write_does_not_kill_replacement_session():
+    from hermes_cli.pty_session import PtySession
+
+    class SupersededBridge(FakeBridge):
+        def __init__(self):
+            super().__init__([b""])
+            self.old_write_started = asyncio.Event()
+            self.release_old_write = asyncio.Event()
+            self.calls = 0
+
+        async def write(self, data):
+            self.calls += 1
+            if self.calls == 1:
+                self.old_write_started.set()
+                await self.release_old_write.wait()
+                return False
+            self.written.extend(data)
+            return True
+
+    bridge = SupersededBridge()
+    s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01)
+    await s.start()
+    old_ws = FakeWS()
+    new_ws = FakeWS()
+    await s.attach(old_ws)
+
+    old_write = asyncio.create_task(s.write(old_ws, b"old input"))
+    await bridge.old_write_started.wait()
+    new_attach = asyncio.create_task(s.attach(new_ws, force_redraw=True))
+    for _ in range(10):
+        if s._ws is new_ws:
+            break
+        await asyncio.sleep(0)
+    assert s._ws is new_ws
+
+    bridge.release_old_write.set()
+    assert await old_write is False
+    assert await new_attach is True
+    assert s.alive is True
+    assert await s.write(new_ws, b"new input") is True
+    assert bytes(bridge.written) == b"\x0cnew input"
+    await s.close()
+
+
 @pytest.mark.asyncio
 async def test_detach_keeps_draining_into_buffer():
     from hermes_cli.pty_session import PtySession