test(cli): trim #110737 re-queue coverage to two invariants on the mixin

Drive CLIChatTurnMixin._chat_render_turn through a minimal stub instead of a full
HermesCLI (prompt_toolkit stubs, cli reload): the invariants are that a (text, images)
payload reaches _pending_input intact and that several queued parts join their text
and merge their images. Both are red on origin/main.
This commit is contained in:
teknium1
2026-09-14 17:48:40 -07:00
committed by Teknium
parent 40b22806f8
commit 24fd22b94d

View File

@@ -1,121 +1,50 @@
"""Regression for #110737: CLI silently drops interrupt messages carrying an image attachment.
"""#110737: an interrupting message that carries image attachments is a ``(text, images)``
tuple on ``_interrupt_queue``; the re-queue block in ``_chat_render_turn`` must hand it to
``_pending_input`` intact instead of string-joining it (TypeError, swallowed by ``chat()``)."""
Background
----------
``cli_tui_mixin._tui_on_enter`` bundles an Enter submission as a
``(text, images)`` tuple when images are attached, and ``_tui_enter_while_busy``
puts that tuple on ``_interrupt_queue`` in interrupt mode. ``_chat_render_turn``
then did ``"\\n".join(all_parts)`` over the raw payloads, raising ``TypeError``
on the tuple — swallowed by the outer ``chat()`` handler, so the message was
lost and no next turn ever started.
The re-queue block must unpack tuple payloads and re-attach their images to the
combined turn instead of joining them as strings.
"""
from __future__ import annotations
import importlib
import queue
import sys
from unittest.mock import MagicMock, patch
from pathlib import Path
from unittest.mock import MagicMock
from hermes_cli.cli_chat_turn_mixin import CLIChatTurnMixin
def _make_cli():
"""Build a HermesCLI instance with prompt_toolkit stubbed out.
Mirrors the helper in ``test_cli_interrupt_drain_regression.py``.
"""
_clean_config = {
"model": {
"default": "anthropic/claude-opus-4.6",
"base_url": "https://openrouter.ai/api/v1",
"provider": "auto",
},
"display": {"compact": False, "tool_progress": "all"},
"agent": {},
"terminal": {"env_type": "local"},
}
clean_env = {"LLM_MODEL": "", "HERMES_MAX_ITERATIONS": ""}
prompt_toolkit_stubs = {
"prompt_toolkit": MagicMock(),
"prompt_toolkit.history": MagicMock(),
"prompt_toolkit.styles": MagicMock(),
"prompt_toolkit.patch_stdout": MagicMock(),
"prompt_toolkit.application": MagicMock(),
"prompt_toolkit.layout": MagicMock(),
"prompt_toolkit.layout.processors": MagicMock(),
"prompt_toolkit.filters": MagicMock(),
"prompt_toolkit.layout.dimension": MagicMock(),
"prompt_toolkit.layout.menus": MagicMock(),
"prompt_toolkit.widgets": MagicMock(),
"prompt_toolkit.key_binding": MagicMock(),
"prompt_toolkit.completion": MagicMock(),
"prompt_toolkit.formatted_text": MagicMock(),
"prompt_toolkit.auto_suggest": MagicMock(),
}
with patch.dict(sys.modules, prompt_toolkit_stubs), patch.dict(
"os.environ", clean_env, clear=False
):
import cli as _cli_mod
_cli_mod = importlib.reload(_cli_mod)
with patch.object(_cli_mod, "get_tool_definitions", return_value=[]), patch.dict(
_cli_mod.__dict__, {"CLI_CONFIG": _clean_config}
):
return _cli_mod.HermesCLI()
class _Stub(CLIChatTurnMixin):
def __init__(self):
self._interrupt_queue = queue.Queue()
self._pending_input = queue.Queue()
self._voice_tts = None
self._voice_continuous = False
self.agent = MagicMock(max_iterations=500)
self._chat_print_reasoning_box = lambda turn: None
self._chat_print_response_panel = lambda turn, response: None
self._emit_focus_recovery_line = lambda: None
self._ring_bell = lambda **kwargs: None
def _stub_render_deps(cli):
"""Neutralize the display/audio side effects of ``_chat_render_turn``."""
cli._chat_print_reasoning_box = lambda turn: None
cli._chat_print_response_panel = lambda turn, response: None
cli._emit_focus_recovery_line = lambda: None
cli._ring_bell = lambda **kwargs: None
cli._voice_tts = None
def _interrupted_turn(interrupt_payload):
def _interrupted_turn(payload):
turn = MagicMock()
turn.result = {"interrupted": True, "interrupt_message": interrupt_payload}
turn.result = {"interrupted": True, "interrupt_message": payload, "final_response": ""}
turn.use_streaming_tts = False
return turn
class TestInterruptRequeueImagePayload:
"""The interrupt re-queue block preserves (text, images) tuple payloads."""
def test_tuple_payload_is_requeued_with_its_images():
cli = _Stub()
images = [Path("/tmp/clip.png")]
def test_tuple_payload_requeued_with_images_attached(self):
cli = _make_cli()
_stub_render_deps(cli)
images = ["shot1.png", "shot2.png"]
cli._chat_render_turn(_interrupted_turn(("look at this", images)), MagicMock(), None)
cli._chat_render_turn(
_interrupted_turn(("look at this", images)), MagicMock(), None
)
assert cli._pending_input.get_nowait() == ("look at this", images)
payload = cli._pending_input.get_nowait()
assert isinstance(payload, tuple)
assert payload == ("look at this", images)
def test_multiple_tuple_payloads_join_text_and_merge_images(self):
cli = _make_cli()
_stub_render_deps(cli)
cli._interrupt_queue.put(("second", ["b.png"]))
def test_queued_parts_join_text_and_merge_images():
cli = _Stub()
cli._interrupt_queue.put("second")
cli._interrupt_queue.put(("third", [Path("/tmp/b.png")]))
cli._chat_render_turn(
_interrupted_turn(("first", ["a.png"])), MagicMock(), None
)
cli._chat_render_turn(_interrupted_turn(("first", [Path("/tmp/a.png")])), MagicMock(), None)
payload = cli._pending_input.get_nowait()
assert payload == ("first\nsecond", ["a.png", "b.png"])
def test_plain_text_payload_stays_a_plain_string(self):
cli = _make_cli()
_stub_render_deps(cli)
cli._chat_render_turn(_interrupted_turn("plain text"), MagicMock(), None)
payload = cli._pending_input.get_nowait()
assert payload == "plain text"
assert isinstance(payload, str)
assert cli._pending_input.get_nowait() == (
"first\nsecond\nthird", [Path("/tmp/a.png"), Path("/tmp/b.png")],
)