test(streaming): cover interrupt join, orphan drain, and EIO paint freeze

tests/hermes_cli/test_relay_shared_metrics_runtime.py

@@ -1303,6 +1303,49 @@ def test_direct_runtime_fake_enforces_lifo_scope_contract(direct_runtime):
     runtime.run_in_session(session, direct_runtime.scope.pop, first)
 
 
+def test_close_session_drains_orphaned_scopes_before_session_pop(direct_runtime):
+    """Orphaned physical scopes must not permanently wedge session close (#81521)."""
+    runtime = relay_runtime.get_runtime()
+    assert runtime is not None
+    session = runtime.ensure_session({"session_id": "orphan-drain"})
+    assert session is not None
+    session_handle = session.handle
+
+    orphan = runtime.run_in_session(
+        session,
+        direct_runtime.scope.push,
+        "orphaned-physical-llm",
+        direct_runtime.ScopeType.Function,
+        handle=session_handle,
+    )
+    assert orphan is not None
+
+    # Without drain, popping the session while the orphan is on top fails
+    # with "scope handle is not at the top of the stack".
+    runtime.close_session({"session_id": "orphan-drain"})
+
+    assert runtime.get_session("orphan-drain") is None
+    rejected = [
+        event
+        for event in direct_runtime.events
+        if event[0] == "scope.pop.rejected" and event[1] == session_handle
+    ]
+    # First attempt may reject; drain + retry must succeed so the session
+    # handle is eventually popped (not left rejected-only).
+    session_pops = [
+        event
+        for event in direct_runtime.events
+        if event[0] == "scope.pop" and event[1] == session_handle
+    ]
+    orphan_pops = [
+        event
+        for event in direct_runtime.events
+        if event[0] == "scope.pop" and event[1] == orphan
+    ]
+    assert orphan_pops, "orphaned physical scope was not drained"
+    assert session_pops, f"session scope never closed (rejected={rejected!r})"
+
+
 def test_concurrent_turn_skips_relay_before_scope_stack_can_interleave(
     direct_runtime,
 ):

tests/hermes_cli/test_terminal_io_broken_81521.py

@@ -0,0 +1,59 @@
+"""CLI freezes UI paints after stdout/PTY EIO (#81521).
+
+A stream-stall interrupt that corrupts the PTY used to leave the classic
+CLI invalidating hundreds of times per second (escape-sequence flood).
+Once EIO is observed, paints must stop.
+"""
+
+from __future__ import annotations
+
+import errno
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def _make_cli_stub():
+    from cli import HermesCLI
+
+    cli = object.__new__(HermesCLI)
+    cli._terminal_io_broken = False
+    cli._resize_recovery_pending = False
+    cli._last_invalidate = 0.0
+    cli._pet_anim_running = False
+    cli._app = MagicMock()
+    return cli
+
+
+class TestTerminalIoBrokenFreeze:
+    def test_mark_terminal_io_broken_is_idempotent(self):
+        cli = _make_cli_stub()
+        cli._mark_terminal_io_broken("first")
+        cli._mark_terminal_io_broken("second")
+        assert cli._terminal_io_broken is True
+
+    def test_invalidate_stops_after_eio(self):
+        cli = _make_cli_stub()
+        cli._app.invalidate.side_effect = OSError(errno.EIO, "Input/output error")
+
+        cli._invalidate(min_interval=0.0)
+
+        assert cli._terminal_io_broken is True
+        assert cli._app.invalidate.call_count == 1
+
+        cli._invalidate(min_interval=0.0)
+        # Frozen — no further paints.
+        assert cli._app.invalidate.call_count == 1
+
+    def test_force_full_redraw_skipped_when_broken(self):
+        cli = _make_cli_stub()
+        cli._terminal_io_broken = True
+        cli._force_full_redraw()
+        cli._app.invalidate.assert_not_called()
+
+    def test_recover_terminal_after_interrupt_skips_when_broken(self):
+        cli = _make_cli_stub()
+        cli._terminal_io_broken = True
+        cli._force_full_redraw = MagicMock()
+        cli._recover_terminal_after_interrupt()
+        cli._force_full_redraw.assert_not_called()

tests/run_agent/test_stream_interrupt_retry.py

@@ -240,3 +240,71 @@ class TestStreamInterruptBeforeRetry:
         assert "new final" in delivered
         assert response.choices[0].message.content == "new final"
         assert mock_abort.called
+
+
+class TestStreamInterruptJoinsWorkerBeforeRaise:
+    """#81521: interrupt must join the stream worker before raising.
+
+    Raising InterruptedError immediately lets Relay turn teardown race a
+    still-open physical LLM scope ("scope handle is not at the top of the
+    stack") and cascade into the CLI EIO / redraw storm.
+    """
+
+    @pytest.mark.filterwarnings(
+        "ignore::pytest.PytestUnhandledThreadExceptionWarning"
+    )
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_interrupt_joins_worker_before_raising(
+        self, mock_close, mock_create, monkeypatch
+    ):
+        import threading
+
+        import httpx
+
+        join_timeouts: list[float | None] = []
+        original_join = threading.Thread.join
+
+        def spy_join(self, *args, **kwargs):
+            timeout = kwargs.get("timeout", args[0] if args else None)
+            join_timeouts.append(timeout)
+            return original_join(self, *args, **kwargs)
+
+        monkeypatch.setattr(threading.Thread, "join", spy_join)
+
+        class HangUntilClosedStream:
+            response = SimpleNamespace(headers={})
+
+            def __iter__(self):
+                # Block until the poll loop force-closes / cancels; then
+                # surface a transport error like a real aborted SSE body.
+                import time
+
+                deadline = time.time() + 5.0
+                while time.time() < deadline:
+                    time.sleep(0.05)
+                raise httpx.RemoteProtocolError("connection closed by interrupt")
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.return_value = HangUntilClosedStream()
+        mock_create.return_value = mock_client
+
+        agent = _make_agent()
+        agent._interrupt_requested = False
+
+        def fire_interrupt():
+            import time
+
+            time.sleep(0.2)
+            agent._interrupt_requested = True
+
+        threading.Thread(target=fire_interrupt, daemon=True).start()
+
+        with pytest.raises(InterruptedError, match="interrupted"):
+            agent._interruptible_streaming_api_call({})
+
+        assert 2.0 in join_timeouts, (
+            f"Expected a 2.0s worker join before InterruptedError; "
+            f"saw join timeouts {join_timeouts!r}. Without the join, Relay "
+            f"scope teardown races the stream worker (#81521)."
+        )