fix(cli): /stop mid-turn interrupts at once and stops jobs off the UI thread

/stop typed while a turn runs is dispatched inline from the prompt_toolkit
Enter handler (busy_policy="interrupt_then_dispatch"), so the whole handler
ran on the UI thread. That included process.list, process.stop and
kill_all, which stops each process serially and waits out its grace
period, so the input froze until every job had exited.

The handler now keeps only the turn interrupt on the UI thread. The
background cleanup (process refresh, kill_all, delegation interrupt) moves
into _stop_background_work, which runs on a spawn_context_thread worker
when /stop arrives mid-turn and prints its result from there. Idle /stop
runs it inline, as before, on the process loop.

Refs NousResearch#134889

hermes_cli/cli_commands_session_tools.py

@@ -98,33 +98,20 @@ class CLICommandsSessionToolsMixin:
     def _handle_stop_command(self):
         """Handle /stop — kill all running background processes and background (async) delegations.
         Typed while a turn runs it stops that turn first, as Ctrl+C does (the registry's
-        ``busy_policy="interrupt_then_dispatch"``; the input loop dispatches it inline).
+        ``busy_policy="interrupt_then_dispatch"``; the input loop dispatches it inline, on the UI
+        thread). The interrupt is immediate; the background cleanup then runs on a worker thread,
+        since ``kill_all`` waits out each process's stop grace period and would freeze input.
 
         See #14602.
         """
-        from hermes_cli.gateway_tools import process_registry as _registry, process_view
-        if getattr(self, "_agent_running", False) and getattr(self, "agent", None):
+        if not getattr(self, "_agent_running", False):
+            return _stop_background_work()
+        if getattr(self, "agent", None):
             from agent.interrupt_compat import request_hard_interrupt
             print(f"  {t('cli.tui.interrupting')}")
             request_hard_interrupt(self.agent)
-        process_registry, view = _registry(), process_view()
-        if view is not None:
-            view.refresh(force=True)
-        running = [p for p in process_registry.list_sessions() if p.get("status") == "running"]
-        # Background subagents live in their own registry, not the process registry.
-        n_async = view.active_delegations() if view is not None else _probe("tools.async_delegation", "active_count", 0)
-        if not running and not n_async:
-            return print(f"  {_t('stop.none_running')}")
-        if running:
-            print(f"  {_t('stop.stopping', count=len(running))}")
-            print(f"  {_t('stop.stopped', count=process_registry.kill_all(source='cli.stop'))}")
-        if n_async:
-            if view is not None:
-                interrupted = view.interrupt_delegations()  # the gateway interrupts this session's own
-            else:
-                from tools.async_delegation import interrupt_all
-                interrupted = interrupt_all(reason='/stop')
-            print(f"  {_t('stop.interrupted_delegations', count=interrupted)}")
+        from agent.memory_provider import spawn_context_thread
+        spawn_context_thread(_stop_background_work, name="cli-stop").start()
 
     def _handle_agents_command(self):
         """Handle /agents — show background processes and agent status. On the gateway both run there:
@@ -291,6 +278,30 @@ class CLICommandsSessionToolsMixin:
                    f"  {_t('profile.home', home=reply.data['home'])}")
 
 
+def _stop_background_work() -> None:
+    """``/stop``'s cleanup: kill the background processes, then interrupt the background (async)
+    delegations. Mid-turn it runs on a worker thread and prints its result from there."""
+    from hermes_cli.gateway_tools import process_registry as _registry, process_view
+    process_registry, view = _registry(), process_view()
+    if view is not None:
+        view.refresh(force=True)
+    running = [p for p in process_registry.list_sessions() if p.get("status") == "running"]
+    # Background subagents live in their own registry, not the process registry.
+    n_async = view.active_delegations() if view is not None else _probe("tools.async_delegation", "active_count", 0)
+    if not running and not n_async:
+        return print(f"  {_t('stop.none_running')}")
+    if running:
+        print(f"  {_t('stop.stopping', count=len(running))}")
+        print(f"  {_t('stop.stopped', count=process_registry.kill_all(source='cli.stop'))}")
+    if n_async:
+        if view is not None:
+            interrupted = view.interrupt_delegations()  # the gateway interrupts this session's own
+        else:
+            from tools.async_delegation import interrupt_all
+            interrupted = interrupt_all(reason='/stop')
+        print(f"  {_t('stop.interrupted_delegations', count=interrupted)}")
+
+
 def print_session_diff(cli, result: dict, stat_only: bool) -> None:
     """Render a ``CheckpointManager.session_diff`` result (``/diff session``): this process's agent's,
     or the gateway session's ``rollback.session_diff`` (``diff_lines``: its uncut body's line count)."""

tests/hermes_cli/test_gateway_tools.py

@@ -133,12 +133,26 @@ def test_agents_before_the_first_turn_reads_the_opened_sessions_processes(gatewa
     assert ("process.list", {"session_id": SID}) in client.calls
 
 
-def test_stop_typed_mid_turn_interrupts_the_turn_then_stops_processes(gateway, capsys):
+def test_stop_typed_mid_turn_interrupts_the_turn_then_stops_processes(gateway, monkeypatch, capsys):
     """/stop typed while a turn runs is dispatched inline and interrupts it like Ctrl+C (tracker
-    #134889 N4), instead of being queued until the turn it should stop has finished."""
+    #134889 N4), instead of being queued until the turn it should stop has finished. Only the
+    interrupt runs on the UI thread: ``kill_all`` waits out each process's stop grace period, so it
+    runs on a worker and the input is back at once (Greptile 4277259986)."""
+    import threading
     from types import SimpleNamespace
     from cli import HermesCLI
     _agent, client = gateway
+    view = gateway_tools.process_view()
+    assert view is not None
+    kill_all, release, stopped = view.kill_all, threading.Event(), threading.Event()
+
+    def slow_kill_all(**kw):  # process.stop still inside its grace periods
+        release.wait(5)
+        try:
+            return kill_all(**kw)
+        finally:
+            stopped.set()
+    monkeypatch.setattr(view, "kill_all", slow_kill_all)
     stops = []
 
     class _Agent:
@@ -150,9 +164,13 @@ def test_stop_typed_mid_turn_interrupts_the_turn_then_stops_processes(gateway, c
     buffer = SimpleNamespace(reset=lambda **_k: None)
     event = SimpleNamespace(app=SimpleNamespace(current_buffer=buffer, invalidate=lambda: None))
     assert cli._tui_enter_inline_command(event, "/stop", False) is True
-    assert stops == [None]
+    assert stops == [None] and not stopped.is_set()  # interrupted; the UI thread did not wait on kill_all
+    release.set()
+    for worker in [t for t in threading.enumerate() if t.name == "cli-stop"]:
+        worker.join(5)
+    out = capsys.readouterr().out
+    assert stopped.is_set() and "Interrupting agent" in out and "Stopped 1 process(es)." in out
     assert any(method == "process.stop" for method, _ in client.calls)
-    assert "Interrupting agent" in capsys.readouterr().out
 
 
 def test_dock_stop_kills_one_process_through_the_gateway(gateway):