diff --git a/hermes_cli/pty_bridge.py b/hermes_cli/pty_bridge.py
index faaed7344b..90d6a61a6c 100644
--- a/hermes_cli/pty_bridge.py
+++ b/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``.
diff --git a/hermes_cli/pty_session.py b/hermes_cli/pty_session.py
index d5cb58a517..c53e50bfa1 100644
--- a/hermes_cli/pty_session.py
+++ b/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:
diff --git a/hermes_cli/web_routers/chat_ws.py b/hermes_cli/web_routers/chat_ws.py
index 64f792d3f3..6d4dc9f84e 100644
--- a/hermes_cli/web_routers/chat_ws.py
+++ b/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:
diff --git a/hermes_cli/web_server_chat.py b/hermes_cli/web_server_chat.py
index d92d6e7b79..047bc867af 100644
--- a/hermes_cli/web_server_chat.py
+++ b/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:
diff --git a/hermes_cli/win_pty_bridge.py b/hermes_cli/win_pty_bridge.py
index 05deb36a5c..5a11238347 100644
--- a/hermes_cli/win_pty_bridge.py
+++ b/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:
diff --git a/tests/hermes_cli/test_pty_bridge.py b/tests/hermes_cli/test_pty_bridge.py
index c260b9b609..8b7c88f53d 100644
--- a/tests/hermes_cli/test_pty_bridge.py
+++ b/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:
diff --git a/tests/hermes_cli/test_web_server_pty_idle_backoff.py b/tests/hermes_cli/test_web_server_pty_idle_backoff.py
index 49712fcf61..7bb8bdc591 100644
--- a/tests/hermes_cli/test_web_server_pty_idle_backoff.py
+++ b/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
diff --git a/tests/hermes_cli/test_win_pty_bridge.py b/tests/hermes_cli/test_win_pty_bridge.py
index fff0da000d..465880055c 100644
--- a/tests/hermes_cli/test_win_pty_bridge.py
+++ b/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()
-
diff --git a/tests/test_pty_keepalive_ws.py b/tests/test_pty_keepalive_ws.py
index 86c85f60f0..47ca9c284c 100644
--- a/tests/test_pty_keepalive_ws.py
+++ b/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
diff --git a/tests/test_pty_session.py b/tests/test_pty_session.py
index 85a91c77ca..64d0173f9c 100644
--- a/tests/test_pty_session.py
+++ b/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