fix: yield subagent monitor to incoming CLI prompts

hermes_cli/callbacks.py

@@ -10,7 +10,9 @@ from hermes_constants import display_hermes_home
 
 
 def _invalidate(cli) -> None:
-    if getattr(cli, "_app", None):
+    if hasattr(cli, "_paint_now"):
+        cli._paint_now()
+    elif getattr(cli, "_app", None):
         cli._app.invalidate()
 
 

hermes_cli/cli_modal_mixin.py

@@ -853,11 +853,16 @@ class CLIModalMixin:
         self._invalidate()
 
     def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict:
-        return prompt_for_secret(self, var_name, prompt, metadata)
+        self._capture_modal_input_snapshot()
+        try:
+            return prompt_for_secret(self, var_name, prompt, metadata)
+        finally:
+            self._restore_modal_input_snapshot()
+            self._paint_now()
 
     def _capture_modal_input_snapshot(self) -> None:
         """Temporarily clear the input buffer and save the user's in-progress draft."""
-        if self._modal_input_snapshot is not None or not getattr(self, "_app", None):
+        if getattr(self, "_modal_input_snapshot", None) is not None or not getattr(self, "_app", None):
             return
         try:
             buf = self._app.current_buffer
@@ -868,7 +873,7 @@ class CLIModalMixin:
 
     def _restore_modal_input_snapshot(self) -> None:
         """Restore any draft text that was present before a modal prompt opened."""
-        snapshot = self._modal_input_snapshot
+        snapshot = getattr(self, "_modal_input_snapshot", None)
         self._modal_input_snapshot = None
         if not snapshot or not getattr(self, "_app", None):
             return

hermes_cli/cli_subagent_monitor.py

@@ -105,6 +105,12 @@ def read_tail(path):
         return 'Live transcript not available yet.'
 
 
+def modal_prompt_active(cli):
+    return any(getattr(cli, name, None) for name in (
+        '_clarify_state', '_approval_state', '_slash_confirm_state', '_sudo_state',
+        '_secret_state', '_model_picker_state', '_command_palette_state'))
+
+
 def build_monitor_application(monitor, **kwargs):
     from prompt_toolkit.application import Application
     from prompt_toolkit.data_structures import Point
@@ -235,8 +241,16 @@ def build_monitor_application(monitor, **kwargs):
         Window(FormattedTextControl(lambda: _clip(state['notice'], app.output.get_size().columns)), height=1),
         Window(FormattedTextControl(footer), height=1),
     ]), focused_element=roster)
+    def before_render(app):
+        # Prompts arrive on worker threads; exit on the UI loop, including the
+        # first frame if a prompt won the race with in_terminal() acquisition.
+        if modal_prompt_active(monitor.cli) and not app.is_done:
+            app.exit()
+        elif state['detail']:
+            update_tail()
+
     app = Application(layout=layout, key_bindings=kb, full_screen=True, mouse_support=False,
-                      before_render=lambda app: update_tail() if state['detail'] else None, **kwargs)
+                      before_render=before_render, **kwargs)
     return app
 
 
@@ -277,5 +291,5 @@ def install_dock(cli):
 
     cli._subagent_dock_widget = ConditionalContainer(
         Window(FormattedTextControl(text), wrap_lines=False),
-        filter=Condition(lambda: bool(monitor.entries)),
+        filter=Condition(lambda: bool(monitor.entries) and not modal_prompt_active(cli)),
     )

hermes_cli/cli_terminal_mixin.py

@@ -103,6 +103,9 @@ class CLITerminalMixin:
         """
         if getattr(self, "_terminal_io_broken", False):
             return
+        monitor = getattr(self, "_subagent_monitor", None)
+        if monitor is not None and monitor.app is not None:
+            monitor.app.invalidate()
         app = getattr(self, "_app", None)
         if app is not None:
             self._app_invalidate(app, "paint_now", swallow=True)

hermes_cli/cli_tui_mixin.py

@@ -1456,8 +1456,9 @@ class CLITuiMixin:
             event.app.invalidate()
             return True
         if self._secret_state:
-            self._submit_secret_response(buf.text)
+            value = buf.text
             buf.reset()
+            self._submit_secret_response(value)
             event.app.invalidate()
             return True
         if self._approval_state:
@@ -1857,12 +1858,9 @@ class CLITuiMixin:
         kb.add(Keys.BracketedPaste, eager=True)(self._tui_handle_paste)
         kb.add('c-v')(self._tui_handle_ctrl_v)
         kb.add('escape', 'v')(self._tui_handle_alt_v)
-        from hermes_cli.cli_subagent_monitor import open_monitor
-        kb.add('f6', filter=Condition(lambda: not any(
-            getattr(self, name, None) for name in (
-                '_clarify_state', '_approval_state', '_slash_confirm_state', '_sudo_state',
-                '_secret_state', '_model_picker_state', '_command_palette_state'))))(
-                    lambda event: open_monitor(self))
+        from hermes_cli.cli_subagent_monitor import modal_prompt_active, open_monitor
+        kb.add('f6', filter=Condition(lambda: not modal_prompt_active(self)))(
+            lambda event: open_monitor(self))
         return kb
 
     def _tui_bind_editor_and_stash(self, kb) -> None:

tests/cli/test_subagent_monitor_prompts.py

@@ -0,0 +1,89 @@
+"""Blocking prompts reclaim input from the nested monitor without a keypress."""
+import asyncio
+from types import SimpleNamespace
+
+
+def test_prompt_paint_yields_monitor_but_ordinary_paint_does_not():
+    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
+    from prompt_toolkit.output import DummyOutput
+
+    async def run():
+        for name in ('_approval_state', '_clarify_state', '_secret_state', '_sudo_state',
+                     '_slash_confirm_state'):
+            dock_cli = SimpleNamespace(agent=None)
+            install_dock(dock_cli)
+            dock_cli._subagent_monitor.entries = [{}]
+            assert dock_cli._subagent_dock_widget.filter()
+            setattr(dock_cli, name, {'pending': True})
+            assert not dock_cli._subagent_dock_widget.filter(), 'dock crowds the blocking prompt'
+            cli = SimpleNamespace(_app=None)
+            monitor = SubagentMonitor(cli)
+            cli._subagent_monitor = monitor
+            with create_pipe_input() as pipe:
+                app = build_monitor_application(monitor, input=pipe, output=DummyOutput())
+                monitor.app = app
+                rendered = asyncio.Event()
+                app.after_render += lambda app: rendered.set()
+                task = asyncio.create_task(app.run_async())
+                await asyncio.wait_for(rendered.wait(), 3)
+                try:
+                    await asyncio.to_thread(CLITerminalMixin._paint_now, cli)
+                    assert not task.done(), 'ordinary paints must not dismiss the monitor'
+                    setattr(cli, name, {'pending': True})
+                    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
+                finally:
+                    if not task.done():
+                        app.exit()
+                        await task
+    asyncio.run(run())
+
+
+def test_secret_callback_yields_monitor_and_restores_composer():
+    from hermes_cli.cli_subagent_monitor import SubagentMonitor, build_monitor_application
+    from hermes_cli.cli_terminal_mixin import CLITerminalMixin
+    from hermes_cli.cli_modal_mixin import CLIModalMixin
+    from prompt_toolkit.buffer import Buffer
+    from prompt_toolkit.document import Document
+    from prompt_toolkit.input import create_pipe_input
+    from prompt_toolkit.output import DummyOutput
+
+    class CLI(CLIModalMixin, CLITerminalMixin):
+        def _ring_bell(self, **kwargs):
+            pass
+
+    async def run():
+        cli = CLI()
+        draft = Document('keep my draft', 4)
+        buffer = Buffer(document=draft)
+        cli._app = SimpleNamespace(current_buffer=buffer, invalidate=lambda: None)
+        cli._modal_input_snapshot = None
+        monitor = cli._subagent_monitor = SubagentMonitor(cli)
+        with create_pipe_input() as pipe:
+            monitor.app = build_monitor_application(monitor, input=pipe, output=DummyOutput())
+            rendered = asyncio.Event()
+            monitor.app.after_render += lambda app: rendered.set()
+            task = asyncio.create_task(monitor.app.run_async())
+            await asyncio.wait_for(rendered.wait(), 3)
+            callback = asyncio.create_task(asyncio.to_thread(
+                cli._secret_capture_callback, 'FIXTURE_SECRET', 'Owned fixture'))
+            try:
+                done, _ = await asyncio.wait({task}, timeout=1)
+                assert task in done, 'secret arrival did not immediately reclaim input'
+                assert buffer.text == '', 'draft must not be submitted as a secret'
+                cli._submit_secret_response('')
+                result = await asyncio.wait_for(callback, 3)
+                assert result['skipped']
+                assert buffer.document == draft
+            finally:
+                if not task.done():
+                    monitor.app.exit()
+                await task
+                if not callback.done():
+                    cli._submit_secret_response('')
+                    await callback
+    asyncio.run(run())