fix: harden web terminal paste and reconnect

- Queue bounded ordered terminal input so acknowledgements remain responsive
- Prevent stale sockets and retries from replacing healthy connections
- Preserve desktop scrollback behavior and add terminal regression coverage

Quality gates: frontend tests (91 passed), typecheck, lint, build, Python compilation, LSP diagnostics. Backend pytest skipped by user request.

apps/api/src/api/system/terminal.py

@@ -3,6 +3,8 @@
 import asyncio
 import json
 import logging
+from asyncio import QueueFull
+from json import JSONDecodeError
 import uuid
 from contextlib import suppress
 
@@ -31,6 +33,9 @@ from src.services.terminal.terminal_manager import (
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
+MAX_PENDING_INPUT_MESSAGES = 64
+MAX_TERMINAL_INPUT_BYTES = 1024 * 1024
+
 
 class SessionRef:
     """Mutable reference to a terminal session, allowing updates during reset."""
@@ -320,8 +325,38 @@ async def _handle_terminal_websocket(
                 )
 
 
+async def _input_write_loop(input_queue: asyncio.Queue[tuple[object, bytes]]) -> None:
+    """Serialize PTY writes without blocking terminal control messages."""
+    while True:
+        session, data = await input_queue.get()
+        try:
+            await session.write_input(data)  # type: ignore[attr-defined]
+        except Exception:
+            logger.debug("Terminal input write failed", exc_info=True)
+        finally:
+            input_queue.task_done()
+
+
+def _queue_terminal_input(
+    input_queue: asyncio.Queue[tuple[object, bytes]], session: object, data: bytes
+) -> bool:
+    """Queue bounded terminal input without blocking control-message processing."""
+    if len(data) > MAX_TERMINAL_INPUT_BYTES:
+        return False
+    try:
+        input_queue.put_nowait((session, data))
+    except QueueFull:
+        return False
+    return True
+
+
 async def _write_loop(session_ref: SessionRef, websocket, instance_id: str) -> None:
-    """Read input from WebSocket and send to container."""
+    """Receive terminal messages while a dedicated worker serializes PTY input."""
+    input_queue: asyncio.Queue[tuple[object, bytes]] = asyncio.Queue(
+        maxsize=MAX_PENDING_INPUT_MESSAGES
+    )
+    input_writer = asyncio.create_task(_input_write_loop(input_queue))
+
     try:
         while True:
             session = session_ref.session
@@ -331,7 +366,12 @@ async def _write_loop(session_ref: SessionRef, websocket, instance_id: str) -> N
             message = await websocket.receive()
             if message["type"] == "websocket.receive":
                 if "bytes" in message:
-                    await session.write_input(message["bytes"])
+                    if not _queue_terminal_input(
+                        input_queue, session, message["bytes"]
+                    ):
+                        logger.warning("Terminal input buffer exceeded for %s", instance_id)
+                        await websocket.close(code=1009, reason="Terminal input buffer full")
+                        break
                 elif "text" in message:
                     text = message["text"]
                     # A text frame that parses to a JSON object with a
@@ -345,13 +385,22 @@ async def _write_loop(session_ref: SessionRef, websocket, instance_id: str) -> N
                     if text.startswith("{"):
                         try:
                             parsed = json.loads(text)
-                        except json.JSONDecodeError:
+                        except JSONDecodeError:
                             parsed = None
                         if isinstance(parsed, dict) and "type" in parsed:
                             ctrl = parsed
 
                     if ctrl is None:
-                        await session.write_input(text.encode("utf-8"))
+                        if not _queue_terminal_input(
+                            input_queue, session, text.encode("utf-8")
+                        ):
+                            logger.warning(
+                                "Terminal input buffer exceeded for %s", instance_id
+                            )
+                            await websocket.close(
+                                code=1009, reason="Terminal input buffer full"
+                            )
+                            break
                         continue
 
                     msg_type = ctrl["type"]
@@ -403,8 +452,14 @@ async def _write_loop(session_ref: SessionRef, websocket, instance_id: str) -> N
                         continue
             elif message["type"] == "websocket.disconnect":
                 break
-    except Exception:
+    except WebSocketDisconnect:
         pass
+    except (RuntimeError, TypeError, ValueError) as exc:
+        logger.debug("Terminal WebSocket receive loop ended: %s", exc)
+    finally:
+        input_writer.cancel()
+        with suppress(asyncio.CancelledError):
+            await input_writer
 
 
 async def _heartbeat_loop(websocket: WebSocket) -> None:

apps/api/tests/api/test_terminal_ws_multi.py

@@ -1,7 +1,15 @@
 """Integration tests for multi-session terminal WebSocket and REST API."""
 
+import asyncio
+
 import pytest
 from fastapi.testclient import TestClient
+from src.api.system.terminal import (
+    MAX_TERMINAL_INPUT_BYTES,
+    SessionRef,
+    _queue_terminal_input,
+    _write_loop,
+)
 from src.main import app
 
 
@@ -19,12 +27,79 @@ class TestTerminalWebSocketMultiSession:
         # the route exists by checking for a 403 (no auth cookie)
         response = client.get("/ws/tool-instances/test-instance/terminal/test-session")
         # WebSocket endpoint returns 403 when accessed via HTTP GET
-        assert response.status_code in (403, 404)
+        assert response.status_code == 403 or response.status_code == 404
 
     def test_default_session_alias_route_exists(self, client):
         """The default session alias route should still exist."""
         response = client.get("/ws/tool-instances/test-instance/terminal")
-        assert response.status_code in (403, 404)
+        assert response.status_code == 403 or response.status_code == 404
+
+
+def test_terminal_input_queue_rejects_excess_input_without_blocking() -> None:
+    """A stalled PTY writer cannot make the input queue grow without limit."""
+    input_queue: asyncio.Queue[tuple[object, bytes]] = asyncio.Queue(maxsize=1)
+    session = object()
+
+    assert _queue_terminal_input(input_queue, session, b"first")
+    assert not _queue_terminal_input(input_queue, session, b"second")
+    assert not _queue_terminal_input(
+        asyncio.Queue(), session, b"x" * (MAX_TERMINAL_INPUT_BYTES + 1)
+    )
+
+
+@pytest.mark.asyncio
+async def test_ack_is_processed_while_a_pty_write_is_waiting() -> None:
+    """A blocked paste writer must not block flow-control acknowledgements."""
+
+    write_started = asyncio.Event()
+
+    class Session:
+        _closed = False
+
+        def __init__(self) -> None:
+            self.acks: list[int] = []
+            self.write_finished = False
+
+        def is_alive(self) -> bool:
+            return True
+
+        async def write_input(self, _data: bytes) -> None:
+            write_started.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                self.write_finished = True
+
+        def acknowledge_data(self, char_count: int) -> None:
+            self.acks.append(char_count)
+
+    class WebSocket:
+        def __init__(self) -> None:
+            self.messages = iter(
+                [
+                    {"type": "websocket.receive", "bytes": b"large paste"},
+                    {"type": "websocket.receive", "text": '{"type":"ack","chars":4096}'},
+                    {"type": "websocket.disconnect"},
+                ]
+            )
+            self.receive_count = 0
+
+        async def receive(self):
+            self.receive_count += 1
+            if self.receive_count > 1:
+                await write_started.wait()
+            return next(self.messages)
+
+    session = Session()
+    websocket = WebSocket()
+    task = asyncio.create_task(_write_loop(SessionRef(session), websocket, "instance"))
+
+    await asyncio.wait_for(write_started.wait(), timeout=0.1)
+    await asyncio.sleep(0)
+    assert session.acks == [4096]
+
+    await task
+    assert session.write_finished
 
 
 class TestTerminalRestApi: