fix: merge web terminal resilience

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: