fix(cli): keep monitor repaints safe during prompt handoff
This commit is contained in:
@@ -1,9 +1,13 @@
|
||||
"""Blocking prompts reclaim input from the nested monitor without a keypress."""
|
||||
import asyncio
|
||||
import threading
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
def test_prompt_paint_yields_monitor_but_ordinary_paint_does_not():
|
||||
|
||||
@pytest.mark.parametrize('paint_source', ['modal', 'tick'])
|
||||
def test_prompt_paint_yields_monitor_but_ordinary_paint_does_not(monkeypatch, paint_source):
|
||||
from hermes_cli.cli_subagent_monitor import SubagentMonitor, build_monitor_application, install_dock
|
||||
from hermes_cli.cli_terminal_mixin import CLITerminalMixin
|
||||
from prompt_toolkit.input import create_pipe_input
|
||||
@@ -29,10 +33,34 @@ def test_prompt_paint_yields_monitor_but_ordinary_paint_does_not():
|
||||
task = asyncio.create_task(app.run_async())
|
||||
await asyncio.wait_for(rendered.wait(), 3)
|
||||
try:
|
||||
rendered.clear()
|
||||
await asyncio.to_thread(CLITerminalMixin._paint_now, cli)
|
||||
await asyncio.wait_for(rendered.wait(), 3)
|
||||
assert not task.done(), 'ordinary paints must not dismiss the monitor'
|
||||
loop = asyncio.get_running_loop()
|
||||
ui_thread = threading.get_ident()
|
||||
stopped = threading.Event()
|
||||
task.add_done_callback(lambda _: stopped.set())
|
||||
schedule = loop.call_soon_threadsafe
|
||||
|
||||
def schedule_with_teardown(callback, *args, **kwargs):
|
||||
handle = schedule(callback, *args, **kwargs)
|
||||
if (callback == app.on_invalidate.fire
|
||||
and threading.get_ident() != ui_thread):
|
||||
# Let an already-queued frame see the modal and finish
|
||||
# teardown while the worker is still in invalidate().
|
||||
schedule(app._redraw)
|
||||
assert stopped.wait(3), 'monitor did not finish teardown'
|
||||
return handle
|
||||
|
||||
setattr(cli, name, {'pending': True})
|
||||
await asyncio.to_thread(CLITerminalMixin._paint_now, cli)
|
||||
with monkeypatch.context() as patch:
|
||||
patch.setattr(loop, 'call_soon_threadsafe', schedule_with_teardown)
|
||||
if paint_source == 'tick':
|
||||
patch.setattr(monitor, 'refresh', lambda: True)
|
||||
await asyncio.to_thread(monitor.tick)
|
||||
else:
|
||||
await asyncio.to_thread(CLITerminalMixin._paint_now, cli)
|
||||
done, _ = await asyncio.wait({task}, timeout=1)
|
||||
assert task in done, f'{name} remained hidden behind the monitor'
|
||||
await task
|
||||
|
||||
Reference in New Issue
Block a user