From 81c4f3a143fe3feb80fa43cbdb8a3ea767d3baf6 Mon Sep 17 00:00:00 2001 From: HexLab98 Date: Sat, 8 Aug 2026 15:26:46 +0700 Subject: [PATCH] test(streaming): cover interrupt join, orphan drain, and EIO paint freeze --- .../test_relay_shared_metrics_runtime.py | 43 ++++++++++++ .../test_terminal_io_broken_81521.py | 59 ++++++++++++++++ .../run_agent/test_stream_interrupt_retry.py | 68 +++++++++++++++++++ 3 files changed, 170 insertions(+) create mode 100644 tests/hermes_cli/test_terminal_io_broken_81521.py diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index bbda0b8c4f..629afe79c7 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/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, ): diff --git a/tests/hermes_cli/test_terminal_io_broken_81521.py b/tests/hermes_cli/test_terminal_io_broken_81521.py new file mode 100644 index 0000000000..05d8ec1ffc --- /dev/null +++ b/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() diff --git a/tests/run_agent/test_stream_interrupt_retry.py b/tests/run_agent/test_stream_interrupt_retry.py index 09ceeef86d..acb618175b 100644 --- a/tests/run_agent/test_stream_interrupt_retry.py +++ b/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)." + )