"""A classic CLI plugin command never holds the CLI's own gateway connection (#134889 PLG-7). The gateway answers one socket's requests one at a time, in arrival order (``tui_gateway/ws.py`` ``_dispatch_loop``), and a plugin handler may run for minutes. So ``slash.exec`` for a plugin command goes over a socket of its own to the same gateway, for the same session: while the handler runs, even after Ctrl+C stopped the wait, the CLI's reads, prompt admissions and interrupts are answered at once. Main's CLI ran the handler itself, without occupying any gateway connection. """ import os import signal import threading from types import SimpleNamespace import pytest from tests.gateway.fixtures.local_recovery_probe import Model, daemon from tests.gateway.test_cli_plugin_command_in_gateway import ( _SLOW, _dial_running, _no_local_plugins, _plugin_home, _release, _typed_in_thread, ) from tests.waiting import wait_until # The command socket's thread (hermes_cli/gateway_connection.py); none may outlive its command. _COMMAND_THREAD = 'hermes-gateway-command' def _command_threads(): return [thread for thread in threading.enumerate() if thread.name == _COMMAND_THREAD] def _cli(monkeypatch, home): """``(cli, connection)``: a classic CLI on its own production connection to ``home``.""" import cli as cli_mod from cli import HermesCLI from hermes_cli.gateway_connection import GatewayConnection from hermes_cli.gateway_session_store import GatewaySessionStore _no_local_plugins(monkeypatch) monkeypatch.setattr(cli_mod, '_ensure_skill_commands', dict) monkeypatch.setattr(cli_mod, 'get_skill_bundles', dict) monkeypatch.setenv('HERMES_HOME', str(home)) conn = GatewayConnection(connect=_dial_running(home)) c = HermesCLI() c.agent, c._session_db, c._gateway_session_open = None, GatewaySessionStore(conn), False return c, conn def _slow_source(tmp_path): gate, marks = tmp_path / 'gate', tmp_path / 'marks' os.mkfifo(gate) marks.mkdir() return gate, marks, _SLOW.replace('GATE', repr(str(gate))).replace('MARKS', repr(str(marks))) def _ctrl_c(c): """The classic TUI's Ctrl+C binding, with an empty draft.""" from prompt_toolkit.buffer import Buffer app = SimpleNamespace(current_buffer=Buffer(), invalidate=lambda: None, exit=lambda: setattr(c, '_should_exit', True)) c._tui_handle_ctrl_c(SimpleNamespace(app=app)) @pytest.mark.platforms("linux") def test_a_stopped_plugin_command_leaves_the_cli_connection_free(tmp_path, monkeypatch, capsys): """Ctrl+C stops the wait while the handler stays blocked in the gateway. Before it is released, the CLI's connection answers a prompt admission, a receipt read and the interrupt of that turn, in order. Released, the handler's late reply is never shown and no command socket or pending request is left.""" from hermes_cli import gateway_plugin_commands from hermes_cli.gateway_client import GatewayClientError models, model_post = [], Model.do_POST def seen_post(handler): # the loopback model server, to hold and release the admitted turn models.append(handler.server) model_post(handler) monkeypatch.setattr(Model, 'do_POST', seen_post) gate, marks, source = _slow_source(tmp_path) outcomes = {} def step(name, method, **params): # the primary connection's ordinary call, its default budget try: outcomes[name] = conn.call(method, **params) except GatewayClientError as exc: outcomes[name] = exc return outcomes[name] with _plugin_home(tmp_path, source) as (root, home, env): c, conn = _cli(monkeypatch, home) try: with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (proc, _desc): c._ensure_gateway_session() sid = c.session_id assert 'ppslow' in (gateway_plugin_commands.names(c) or {}) # cached: no catalog read below capsys.readouterr() stopped = _typed_in_thread(c, '/ppslow held') wait_until(lambda: (marks / 'started-held').exists(), timeout=30, what='the handler to start') _ctrl_c(c) stopped.join(60) stop_out, stop_alive = capsys.readouterr().out, stopped.is_alive() try: receipt = step('submit', 'prompt.submit', session_id=sid, submission_id='held-turn', text='BLOCK_STARTED') if isinstance(receipt, dict): wait_until(lambda: models and models[-1].blocked.is_set(), timeout=60, what='the admitted turn to reach the model') read = step('read', 'prompt.receipt', session_id=sid, admission_id=receipt['admission_id']) if isinstance(read, dict): step('interrupt', 'session.interrupt', session_id=sid, execution_generation=read['execution_generation']) handler_held = not (marks / 'done-held').exists() finally: for model in models: model.release.set() _release(gate, 'late') wait_until(lambda: (marks / 'done-held').exists(), timeout=30, what='the handler to finish') after = c._process_unregistered_slash('/ppquick', '/ppquick'), capsys.readouterr().out wait_until(lambda: not _command_threads(), timeout=30, what='the command sockets to close') pending = dict(conn.client.pending) if conn.client is not None else {} proc.send_signal(signal.SIGINT) proc.wait(timeout=20) finally: conn.close() assert not stop_alive and not c._should_exit, (stop_alive, c._should_exit) assert 'stopped waiting' in stop_out and 'keeps running in the gateway' in stop_out, stop_out # answered while the stopped handler was still blocked in the gateway assert handler_held, 'the handler finished before the requests were answered' assert isinstance(outcomes.get('submit'), dict), outcomes assert outcomes['submit']['status'] in {'queued', 'started'}, outcomes assert isinstance(outcomes.get('read'), dict) and outcomes['read']['status'] == 'started', outcomes assert isinstance(outcomes.get('interrupt'), dict), outcomes assert after[0] is True and 'QUICK_DONE' in after[1] and 'SLOW_DONE' not in after[1], after assert pending == {}, pending @pytest.mark.platforms("linux") def test_a_gateway_hang_up_during_a_plugin_command_ends_the_wait_and_leaves_nothing_pending( tmp_path, monkeypatch, capsys): """The gateway dies while the handler runs: the wait ends with the transport's error (the command is not sent again), and no command socket, thread or pending request is left.""" _gate, marks, source = _slow_source(tmp_path) with _plugin_home(tmp_path, source) as (root, home, env): c, conn = _cli(monkeypatch, home) try: with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (proc, _desc): c._ensure_gateway_session() from hermes_cli import gateway_plugin_commands assert 'ppslow' in (gateway_plugin_commands.names(c) or {}) capsys.readouterr() waiting = _typed_in_thread(c, '/ppslow hangup') wait_until(lambda: (marks / 'started-hangup').exists(), timeout=30, what='the handler to start') proc.kill() proc.wait(timeout=20) waiting.join(60) out, alive = capsys.readouterr().out, waiting.is_alive() wait_until(lambda: not _command_threads(), timeout=30, what='the command socket to close') pending = dict(conn.client.pending) if conn.client is not None else {} finally: conn.close() assert not alive, 'the wait outlived the gateway' assert '/ppslow: Gateway disconnected' in out and 'stopped waiting' not in out, out assert out.count('/ppslow') == 1 and 'SLOW_DONE' not in out, out wait = getattr(c, '_gateway_command_wait', None) assert wait is None and pending == {}, (wait, pending)