mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
47895f8fe0
The CLI answering a plugin command's clarify request between turns now tracks the open request: the gateway's request.cancel (another viewer picked, or its wait ended) closes the panel, and so does the end of the command's wait. A choice with a description still accepts its plain label typed back. Greptile P1/P2 on #142563.
516 lines
27 KiB
Python
516 lines
27 KiB
Python
"""A classic CLI plugin command runs in the gateway and its ``ctx.call_mcp`` reaches the gateway's MCP
|
||
registry (#134889 CLI-13, PLG-7).
|
||
|
||
A real daemon (``gateway.run``) loads a user plugin whose ``/probe-mcp-status`` calls ``probe-mcp``, a
|
||
fake server in the daemon's own MCP registry (tests/gateway/fixtures/plugin_mcp_daemon.py). The CLI's
|
||
dispatch reads the daemon's ``commands.catalog`` and sends ``slash.exec`` over its websocket, so the
|
||
command goes through ``gateway/session_commands.py`` -> ``_hm_run_plugin_handler``.
|
||
"""
|
||
import asyncio
|
||
from contextlib import asynccontextmanager, contextmanager
|
||
import json
|
||
import os
|
||
import queue
|
||
import types
|
||
from http.server import ThreadingHTTPServer
|
||
from pathlib import Path
|
||
import signal
|
||
import threading
|
||
|
||
import pytest
|
||
|
||
from tests.gateway.fixtures.local_recovery_probe import Model, child_env, daemon, rpc, websocket
|
||
from tests.waiting import wait_until
|
||
|
||
_PLUGIN = '''import json
|
||
import os
|
||
|
||
|
||
def register(ctx):
|
||
def status(arg):
|
||
reply = ctx.call_mcp("probe-mcp", "echo", {"text": arg})
|
||
return "PLUGIN_PID=%d MCP=%s" % (os.getpid(), json.dumps(reply, default=str))
|
||
ctx.register_command("probe-mcp-status", status, description="Ask the fake MCP server")
|
||
ctx.register_command("probe_under", lambda arg: "EXACT_UNDERSCORE", description="Exact name")
|
||
ctx.register_command("probe-under", lambda arg: "HYPHEN_ALIAS", description="Telegram alias form")
|
||
ctx.register_command("probe_only", lambda arg: "UNDERSCORE_ONLY", description="No hyphen twin")
|
||
|
||
def boom(arg):
|
||
raise RuntimeError("probe handler exploded")
|
||
ctx.register_command("probe-boom", boom, description="Always fails")
|
||
|
||
def where(arg): # what the plugin docs tell a handler to read instead of os.getcwd()/os.environ
|
||
from agent.runtime_cwd import resolve_agent_cwd
|
||
from gateway.session_context import get_session_env
|
||
return "WHERE=" + json.dumps({"cwd": str(resolve_agent_cwd()), "getcwd": os.getcwd(),
|
||
"source": get_session_env("HERMES_SESSION_SOURCE"),
|
||
"platform": get_session_env("HERMES_SESSION_PLATFORM"),
|
||
"key": get_session_env("HERMES_SESSION_KEY")})
|
||
ctx.register_command("probe-cwd", where, description="Where the handler runs")
|
||
|
||
def sid(arg): # the session a background job started from the handler records as its parent
|
||
from gateway.session_context import get_session_env
|
||
from tools.process_registry import process_registry
|
||
started = json.loads(ctx.dispatch_tool("terminal", {"command": "true", "background": True}))
|
||
job = process_registry.get(started.get("session_id") or "")
|
||
return "SID=" + json.dumps({"id": get_session_env("HERMES_SESSION_ID"),
|
||
"key": get_session_env("HERMES_SESSION_KEY"),
|
||
"parent": getattr(job, "parent_session_id", None), "started": started})
|
||
ctx.register_command("probe-sid", sid, description="The calling session")
|
||
ctx.register_command("probe-args", lambda arg: "ARGS=" + json.dumps(arg), description="Echo the argument")
|
||
'''
|
||
# One command per gateway life: A serves /alpha, the restarted B only /beta.
|
||
_LIFE = '''import os
|
||
|
||
|
||
def register(ctx):
|
||
ctx.register_command("NAME", lambda arg: "RAN_NAME PID=%d" % os.getpid(), description="NAME")
|
||
'''
|
||
|
||
|
||
@contextmanager
|
||
def _plugin_home(tmp_path, source):
|
||
"""``(root, home, env)``: a profile home with plugin ``probe-cmd`` and a loopback model."""
|
||
root = Path(__file__).resolve().parents[2]
|
||
home, user = tmp_path / 'state', tmp_path / 'user'
|
||
home.mkdir(mode=0o700)
|
||
user.mkdir()
|
||
plugin = home / 'plugins' / 'probe-cmd'
|
||
plugin.mkdir(parents=True)
|
||
(plugin / 'plugin.yaml').write_text("name: probe-cmd\nversion: '1.0'\n")
|
||
(plugin / '__init__.py').write_text(source)
|
||
peer = ThreadingHTTPServer(('127.0.0.1', 0), Model)
|
||
peer.requests = []
|
||
peer.blocked, peer.release = threading.Event(), threading.Event()
|
||
thread = threading.Thread(target=peer.serve_forever, daemon=True)
|
||
thread.start()
|
||
url = f'http://127.0.0.1:{peer.server_port}/v1'
|
||
(home / 'config.yaml').write_text(json.dumps({
|
||
'gateway': {'multiplex_profiles': False},
|
||
'model': {'provider': 'custom', 'default': 'plugin-model', 'base_url': url},
|
||
'auxiliary': {'title_generation': {'enabled': False}}, 'platform_toolsets': {'cli': []},
|
||
'plugins': {'enabled': ['probe-cmd'], 'entries': {'probe-cmd': {'mcp_allowlist': ['probe-mcp']}}},
|
||
}))
|
||
env = child_env()
|
||
env.update(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(home), PYTHONPATH=str(root),
|
||
OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=url, PYTHONUNBUFFERED='1',
|
||
PYTHONDONTWRITEBYTECODE='1')
|
||
try:
|
||
yield root, home, env
|
||
finally:
|
||
peer.shutdown()
|
||
peer.server_close()
|
||
thread.join(timeout=5)
|
||
|
||
|
||
def _no_local_plugins(monkeypatch):
|
||
def no_local_plugin(*_a, **_k):
|
||
pytest.fail('the CLI process loaded or ran plugin code')
|
||
for name in ('get_plugin_command_handler', 'discover_plugins', 'get_plugin_commands'):
|
||
monkeypatch.setattr(f'hermes_cli.plugins.{name}', no_local_plugin)
|
||
|
||
|
||
class _Wire:
|
||
"""The CLI connection's ``call`` over the daemon's real websocket, on one loop thread."""
|
||
closed = final = False
|
||
epoch = 1
|
||
|
||
def __init__(self, home, desc):
|
||
self.loop = asyncio.new_event_loop()
|
||
self.thread = threading.Thread(target=self.loop.run_forever, daemon=True)
|
||
self.thread.start()
|
||
self.calls = []
|
||
self.events = queue.Queue() # the session's frames (none reach this wire's ``call``)
|
||
self.listeners = types.SimpleNamespace(request_listeners=[])
|
||
|
||
async def _open():
|
||
return await websocket(home, desc)
|
||
self.ws = self._run(_open())
|
||
|
||
def _run(self, coro):
|
||
return asyncio.run_coroutine_threadsafe(coro, self.loop).result(60)
|
||
|
||
def state(self, session_id):
|
||
return self.listeners
|
||
|
||
def call(self, method, **params):
|
||
from hermes_cli.gateway_client import GatewayRPCError
|
||
self.calls.append(method)
|
||
reply = self._run(rpc(self.ws, method, **params))
|
||
if 'error' in reply: # an owner refusal, as gateway_client raises it
|
||
raise GatewayRPCError(str(reply['error'].get('message')), reply['error'].get('code'))
|
||
return reply['result']
|
||
|
||
def call_tagged(self, method, **params):
|
||
return self.call(method, **params), self.epoch
|
||
|
||
def call_until(self, method, _wake, **params):
|
||
return self.call(method, **params)
|
||
|
||
def close(self):
|
||
self._run(self.ws.close())
|
||
self.loop.call_soon_threadsafe(self.loop.stop)
|
||
self.thread.join(timeout=5)
|
||
|
||
|
||
@pytest.mark.platforms("linux")
|
||
def test_cli_plugin_command_runs_in_the_gateway_and_reaches_its_mcp_registry(tmp_path, monkeypatch, capsys):
|
||
import cli as cli_mod
|
||
from cli import HermesCLI
|
||
_no_local_plugins(monkeypatch)
|
||
monkeypatch.setattr(cli_mod, '_ensure_skill_commands', dict)
|
||
monkeypatch.setattr(cli_mod, 'get_skill_bundles', dict)
|
||
work = tmp_path / 'work'
|
||
work.mkdir()
|
||
with _plugin_home(tmp_path, _PLUGIN) as (root, home, env):
|
||
with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (proc, desc):
|
||
wire = _Wire(home, desc)
|
||
try:
|
||
sid = wire.call('session.create', request_id='plugin-cli', source='cli', toolsets=[],
|
||
cwd=str(work))['session_id']
|
||
monkeypatch.setattr('hermes_cli.gateway_side.connection_for', lambda _cli: wire)
|
||
c = HermesCLI.__new__(HermesCLI)
|
||
c.config, c.session_id = {}, sid
|
||
capsys.readouterr()
|
||
assert c._process_unregistered_slash('/probe-mcp-status Deep Check',
|
||
'/probe-mcp-status deep check') is True
|
||
out = capsys.readouterr().out
|
||
calls = list(wire.calls)
|
||
typed_outs = {}
|
||
for typed in ('/probe_under', '/probe_only', '/probe-boom', '/probe-cwd', '/probe-sid',
|
||
'/probe-args 2026–2027 — x'):
|
||
assert c._process_unregistered_slash(typed, typed) is True
|
||
typed_outs[typed] = capsys.readouterr().out
|
||
finally:
|
||
wire.close()
|
||
proc.send_signal(signal.SIGINT)
|
||
proc.wait(timeout=20)
|
||
assert calls[-2:] == ['commands.catalog', 'slash.exec'], calls
|
||
assert 'EXACT_UNDERSCORE' in typed_outs['/probe_under'], typed_outs
|
||
assert 'UNDERSCORE_ONLY' in typed_outs['/probe_only'], typed_outs
|
||
# main's local path printed a failing handler's error; the gateway path must not swallow it
|
||
assert 'Plugin command error: probe handler exploded' in typed_outs['/probe-boom'], typed_outs
|
||
where = json.loads(typed_outs['/probe-cwd'].split('WHERE=', 1)[1].strip())
|
||
# the session's folder and session vars are bound around the handler; the process cwd is the gateway's
|
||
assert where['cwd'] == str(work) and where['getcwd'] != str(work), where
|
||
assert where['platform'] == 'local' and where['key'], where
|
||
# the calling session's identity, as a turn binds it: a background job records it as its parent
|
||
ident = json.loads(typed_outs['/probe-sid'].split('SID=', 1)[1].strip())
|
||
assert ident['id'] == sid and ident['key'] == where['key'], ident
|
||
assert ident['parent'] == sid, ident
|
||
# as typed: the messaging path's dash repair (MessageEvent.get_command_args) is not applied
|
||
args = json.loads(typed_outs['/probe-args 2026–2027 — x'].split('ARGS=', 1)[1].strip())
|
||
assert args == '2026–2027 — x', args
|
||
assert proc.pid != os.getpid()
|
||
assert f'PLUGIN_PID={proc.pid} ' in out, out # the handler ran in the gateway process
|
||
mcp = json.loads(json.loads(out.split('MCP=', 1)[1].strip())['result'])
|
||
assert mcp == {'tool': 'echo', 'arguments': {'text': 'Deep Check'}, 'registry_pid': proc.pid}, out
|
||
|
||
|
||
def _dial_running(home):
|
||
"""The production dial pinned to an owner already running (``connect_gateway(endpoint)``), so a
|
||
redial while none is up fails instead of starting one the test does not own."""
|
||
from hermes_cli.gateway_client import GatewayClientError, connect_gateway
|
||
from hermes_cli.gateway_runtime import discover_gateway_endpoint
|
||
|
||
@asynccontextmanager
|
||
async def dial():
|
||
found = await asyncio.to_thread(discover_gateway_endpoint, home)
|
||
if found.state != 'ready' or found.endpoint is None:
|
||
raise GatewayClientError(f'no gateway is serving this home ({found.state})')
|
||
async with connect_gateway(found.endpoint) as client:
|
||
yield client
|
||
return dial
|
||
|
||
|
||
@pytest.mark.platforms("linux")
|
||
def test_a_redial_to_a_restarted_gateway_and_slash_new_reread_the_plugin_commands(tmp_path, monkeypatch, capsys):
|
||
"""The classic CLI's own GatewayConnection and GatewaySessionStore across a real owner restart:
|
||
gateway A serves /alpha, B (same home and socket) only /beta. A cached /alpha never runs: while no
|
||
gateway answers it is a named refusal (no same-named skill stands in), on B it is no plugin
|
||
command. /new through the gateway session store re-reads B's catalog for the new session."""
|
||
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)
|
||
skills, ran_skill = {'/alpha': {'name': 'alpha'}}, []
|
||
monkeypatch.setattr(cli_mod, '_ensure_skill_commands', lambda: dict(skills))
|
||
monkeypatch.setattr(cli_mod, 'get_skill_bundles', dict)
|
||
monkeypatch.setattr(HermesCLI, '_run_skill_slash_command', lambda self, *a: ran_skill.append(a[0]))
|
||
with _plugin_home(tmp_path, _LIFE.replace('NAME', 'alpha')) as (root, home, env):
|
||
monkeypatch.setenv('HERMES_HOME', str(home))
|
||
conn = GatewayConnection(connect=_dial_running(home))
|
||
sent, real, real_until = [], conn.call_tagged, conn.call_until
|
||
|
||
def spy(method, **params):
|
||
sent.append((method, params.get('session_id')))
|
||
return real(method, **params)
|
||
|
||
def spy_until(method, wake, **params): # slash.exec: no answer deadline
|
||
sent.append((method, params.get('session_id')))
|
||
return real_until(method, wake, **params)
|
||
monkeypatch.setattr(conn, 'call_tagged', spy)
|
||
monkeypatch.setattr(conn, 'call_until', spy_until)
|
||
c = HermesCLI()
|
||
c.agent, c._session_db, c._gateway_session_open = None, GatewaySessionStore(conn), False
|
||
|
||
def typed(text):
|
||
del sent[:]
|
||
assert c._process_unregistered_slash(text, text) is True
|
||
return capsys.readouterr().out, [m for m, _ in sent]
|
||
try:
|
||
with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (a, _desc):
|
||
c._ensure_gateway_session() # the CLI's startup session.create on A
|
||
first = c.session_id
|
||
capsys.readouterr()
|
||
on_a = typed('/alpha')
|
||
a.send_signal(signal.SIGINT)
|
||
a.wait(timeout=20)
|
||
wait_until(lambda: conn.closed, timeout=20, what='the CLI connection to see gateway A hang up')
|
||
gone = typed('/alpha')
|
||
(home / 'plugins' / 'probe-cmd' / '__init__.py').write_text(_LIFE.replace('NAME', 'beta'))
|
||
skills.clear()
|
||
with daemon(root, home, env, barrier=False) as (b, _desc):
|
||
on_b = typed('/beta')
|
||
epoch_b = conn.epoch
|
||
stale = typed('/alpha')
|
||
c.new_session(silent=True)
|
||
capsys.readouterr()
|
||
del sent[:]
|
||
assert c._process_unregistered_slash('/beta', '/beta') is True
|
||
after_new, new_calls = capsys.readouterr().out, list(sent)
|
||
b.send_signal(signal.SIGINT)
|
||
b.wait(timeout=20)
|
||
finally:
|
||
conn.close()
|
||
assert on_a[1] == ['commands.catalog', 'slash.exec'] and f'RAN_alpha PID={a.pid}' in on_a[0], on_a
|
||
# A hung up and nothing serves the home: the cached /alpha is not sent and the skill does not run
|
||
assert 'slash.exec' not in gone[1] and "command list is unavailable" in gone[0], gone
|
||
assert 'nothing was run' in gone[0] and ran_skill == [], (gone, ran_skill)
|
||
assert epoch_b == 2 and on_b[1] == ['commands.catalog', 'slash.exec'], (epoch_b, on_b)
|
||
assert f'RAN_beta PID={b.pid}' in on_b[0] and b.pid != a.pid, on_b
|
||
# B's catalog has no /alpha: one refresh on the miss, never a slash.exec, no plugin output
|
||
assert stale[1] == ['commands.catalog'] and 'RAN_alpha' not in stale[0], stale
|
||
assert c.session_id != first and conn.epoch == epoch_b
|
||
# /new reads the new session's own catalog (its skills come from that session's folder)
|
||
assert new_calls == [('commands.catalog', c.session_id), ('slash.exec', c.session_id)], new_calls
|
||
assert f'RAN_beta PID={b.pid}' in after_new, after_new
|
||
|
||
|
||
# /pidprobe stays; /pptwin shares its name with a skill. ``TWIN`` is dropped to stop registering it.
|
||
_RELOADED = '''import os
|
||
|
||
|
||
def register(ctx):
|
||
ctx.register_command("pidprobe", lambda arg: "PIDPROBE PID=%d" % os.getpid(), description="Probe")
|
||
TWIN
|
||
'''
|
||
_TWIN = 'ctx.register_command("pptwin", lambda arg: "PLUGIN_TWIN", description="Shadows a skill")'
|
||
|
||
|
||
@pytest.mark.platforms("linux")
|
||
@pytest.mark.parametrize("unload", ["disable", "stop_registering"])
|
||
def test_a_command_a_plugin_reload_removed_is_dispatched_again(tmp_path, monkeypatch, capsys, unload):
|
||
"""The open CLI cached the session's catalog; then the running gateway reloads its plugins (what
|
||
``hermes plugins disable`` asks for) without the command, in the same session and gateway life.
|
||
The cached name gets ``unsupported_command`` from ``slash.exec`` once: the CLI re-reads the catalog
|
||
and dispatches the name again, so a same-named skill runs, and a name nothing else matches gets
|
||
the "no longer has this plugin command" line, never the raw code, on that try or a later one."""
|
||
import cli as cli_mod
|
||
from cli import HermesCLI
|
||
from gateway.control_socket import reload_gateway_plugins
|
||
_no_local_plugins(monkeypatch)
|
||
ran_skill = []
|
||
monkeypatch.setattr(cli_mod, '_ensure_skill_commands', dict) # the skill is the gateway's, below
|
||
monkeypatch.setattr(cli_mod, 'get_skill_bundles', dict)
|
||
monkeypatch.setattr(HermesCLI, '_run_skill_slash_command', lambda self, *a: ran_skill.append(a[0]))
|
||
work = tmp_path / 'work'
|
||
work.mkdir()
|
||
with _plugin_home(tmp_path, _RELOADED.replace('TWIN', _TWIN)) as (root, home, env):
|
||
(home / 'skills' / 'pptwin').mkdir(parents=True)
|
||
(home / 'skills' / 'pptwin' / 'SKILL.md').write_text('---\nname: pptwin\ndescription: Twin skill\n---\nSay SKILL_TWIN.\n')
|
||
with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (proc, desc):
|
||
wire = _Wire(home, desc)
|
||
try:
|
||
sid = wire.call('session.create', request_id='plugin-reload', source='cli', toolsets=[],
|
||
cwd=str(work))['session_id']
|
||
monkeypatch.setattr('hermes_cli.gateway_side.connection_for', lambda _cli: wire)
|
||
c = HermesCLI.__new__(HermesCLI)
|
||
c.config, c.session_id = {}, sid
|
||
|
||
def typed(text):
|
||
del wire.calls[:]
|
||
assert c._process_unregistered_slash(text, text) is True
|
||
return capsys.readouterr().out, list(wire.calls)
|
||
capsys.readouterr()
|
||
before = typed('/pidprobe'), typed('/pptwin')
|
||
if unload == 'disable':
|
||
config = json.loads((home / 'config.yaml').read_text())
|
||
config['plugins'].update(enabled=[], disabled=['probe-cmd'])
|
||
(home / 'config.yaml').write_text(json.dumps(config))
|
||
else:
|
||
(home / 'plugins' / 'probe-cmd' / '__init__.py').write_text(_RELOADED.replace('TWIN', 'pass'))
|
||
answer = reload_gateway_plugins(home)
|
||
order = ('/pidprobe', '/pptwin', '/pidprobe') if unload == 'disable' else ('/pptwin', '/pidprobe', '/pptwin')
|
||
after = [typed(text) for text in order]
|
||
finally:
|
||
wire.close()
|
||
proc.send_signal(signal.SIGINT)
|
||
proc.wait(timeout=20)
|
||
assert f'PIDPROBE PID={proc.pid}' in before[0][0] and 'PLUGIN_TWIN' in before[1][0], before
|
||
assert (answer or {}).get('reloaded') is True, answer
|
||
assert not any('unsupported_command' in out for out, _ in after), after
|
||
if unload == 'disable':
|
||
# the cached /pidprobe: refused once, one re-read, the existing line; nothing else ran
|
||
assert after[0][1] == ['slash.exec', 'commands.catalog'], after[0]
|
||
assert 'no longer has this plugin command' in after[0][0] and 'PIDPROBE' not in after[0][0], after[0]
|
||
# the re-read is the session's catalog now: the skill runs without asking the gateway to exec
|
||
assert 'slash.exec' not in after[1][1] and 'slash.exec' not in after[2][1], after
|
||
assert ran_skill == ['/pptwin'] and 'PIDPROBE' not in after[2][0], (ran_skill, after)
|
||
else:
|
||
# the cached /pptwin: refused once, one re-read, then the same-named skill runs
|
||
assert after[0][1] == ['slash.exec', 'commands.catalog'] and ran_skill == ['/pptwin', '/pptwin'], after
|
||
assert f'PIDPROBE PID={proc.pid}' in after[1][0], after # still the plugin's, in the gateway
|
||
assert 'slash.exec' not in after[2][1] and 'PLUGIN_TWIN' not in after[2][0], after
|
||
|
||
|
||
_BROKEN = '''def register(ctx):
|
||
ctx.register_command("pptwin", lambda arg: "PLUGIN_TWIN", description="Shadows a skill")
|
||
raise RuntimeError("register exploded")
|
||
'''
|
||
|
||
|
||
@pytest.mark.platforms("linux")
|
||
def test_a_plugin_whose_register_fails_leaves_its_name_to_a_same_named_skill(tmp_path, monkeypatch, capsys):
|
||
"""Pins the review-bot question "failed plugins select other commands". The loader catches a failed
|
||
``register()``, drops what it registered and records ``loaded.error`` (plugins_loader); the
|
||
plugin has no command, so the catalog lists none and reports no discovery error, and the typed
|
||
name goes on to the same-named skill. Main's CLI did the same: its own load dropped the
|
||
command, ``_get_plugin_cmd_handler_names()`` missed it, and the skill branch ran."""
|
||
import cli as cli_mod
|
||
from cli import HermesCLI
|
||
_no_local_plugins(monkeypatch)
|
||
ran_skill = []
|
||
monkeypatch.setattr(cli_mod, '_ensure_skill_commands', dict) # the skill is the gateway's, below
|
||
monkeypatch.setattr(cli_mod, 'get_skill_bundles', dict)
|
||
monkeypatch.setattr(HermesCLI, '_run_skill_slash_command', lambda self, *a: ran_skill.append(a[0]))
|
||
work = tmp_path / 'work'
|
||
work.mkdir()
|
||
with _plugin_home(tmp_path, _BROKEN) as (root, home, env):
|
||
(home / 'skills' / 'pptwin').mkdir(parents=True)
|
||
(home / 'skills' / 'pptwin' / 'SKILL.md').write_text('---\nname: pptwin\ndescription: Twin skill\n---\nSay SKILL_TWIN.\n')
|
||
with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (proc, desc):
|
||
wire = _Wire(home, desc)
|
||
try:
|
||
sid = wire.call('session.create', request_id='plugin-broken', source='cli', toolsets=[],
|
||
cwd=str(work))['session_id']
|
||
catalog = wire.call('commands.catalog', session_id=sid)
|
||
monkeypatch.setattr('hermes_cli.gateway_side.connection_for', lambda _cli: wire)
|
||
c = HermesCLI.__new__(HermesCLI)
|
||
c.config, c.session_id = {}, sid
|
||
capsys.readouterr()
|
||
del wire.calls[:]
|
||
assert c._process_unregistered_slash('/pptwin', '/pptwin') is True
|
||
out, calls = capsys.readouterr().out, list(wire.calls)
|
||
finally:
|
||
wire.close()
|
||
proc.send_signal(signal.SIGINT)
|
||
proc.wait(timeout=20)
|
||
plugin_rows = [row for cat in catalog['categories'] if cat['name'] == 'Plugin commands' for row in cat['pairs']]
|
||
assert plugin_rows == [] and not catalog.get('plugin_discovery_error'), catalog
|
||
assert calls == ['commands.catalog'] and ran_skill == ['/pptwin'], (calls, ran_skill)
|
||
assert 'PLUGIN_TWIN' not in out, out
|
||
|
||
|
||
# /ppslow blocks on the FIFO GATE until the test writes a note; MARKS/started-<arg> and
|
||
# MARKS/done-<arg> show when the handler started and finished in the gateway.
|
||
_SLOW = '''import os
|
||
|
||
|
||
def register(ctx):
|
||
def slow(arg):
|
||
open(os.path.join(MARKS, "started-" + arg), "w").close()
|
||
with open(GATE) as gate:
|
||
note = gate.read()
|
||
open(os.path.join(MARKS, "done-" + arg), "w").close()
|
||
return "SLOW_DONE %s %s" % (arg, note)
|
||
ctx.register_command("ppslow", slow, description="Waits for the test")
|
||
ctx.register_command("ppquick", lambda arg: "QUICK_DONE", description="Answers at once")
|
||
'''
|
||
|
||
|
||
def _typed_in_thread(c, text):
|
||
thread = threading.Thread(target=c._process_unregistered_slash, args=(text, text.lower()), daemon=True)
|
||
thread.start()
|
||
return thread
|
||
|
||
|
||
def _release(gate, note):
|
||
with open(gate, 'w') as pipe: # blocks until the handler opened its end
|
||
pipe.write(note)
|
||
|
||
|
||
@pytest.mark.platforms("linux")
|
||
def test_a_plugin_command_outliving_the_rpc_limit_prints_its_reply_and_ctrl_c_stops_the_wait(
|
||
tmp_path, monkeypatch, capsys):
|
||
"""Main's CLI ran the handler itself and waited for it. Through the gateway the CLI waits for the
|
||
reply as long as the handler runs, not the 30 s RPC budget (injected here as 0, so a request on
|
||
that budget gives up at once). Ctrl+C stops the wait with a line saying what happens next: the
|
||
handler finishes in the gateway and its reply is dropped, while the next command already runs."""
|
||
import cli as cli_mod
|
||
from types import SimpleNamespace
|
||
from prompt_toolkit.buffer import Buffer
|
||
from cli import HermesCLI
|
||
from hermes_cli import gateway_plugin_commands
|
||
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)
|
||
gate, marks = tmp_path / 'gate', tmp_path / 'marks'
|
||
os.mkfifo(gate)
|
||
marks.mkdir()
|
||
source = _SLOW.replace('GATE', repr(str(gate))).replace('MARKS', repr(str(marks)))
|
||
with _plugin_home(tmp_path, source) as (root, home, env):
|
||
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
|
||
try:
|
||
with daemon(root, home, env, barrier=True, fixture='plugin_mcp_daemon.py') as (proc, _desc):
|
||
c._ensure_gateway_session()
|
||
assert 'ppslow' in (gateway_plugin_commands.names(c) or {}) # cached: no catalog read below
|
||
monkeypatch.setattr('hermes_cli.gateway_connection.DEFAULT_RPC_TIMEOUT', 0)
|
||
capsys.readouterr()
|
||
waiting = _typed_in_thread(c, '/ppslow long')
|
||
wait_until(lambda: (marks / 'started-long').exists(), timeout=30, what='the handler to start')
|
||
_release(gate, 'released')
|
||
waiting.join(60)
|
||
long_out = capsys.readouterr().out
|
||
assert not waiting.is_alive()
|
||
assert 'SLOW_DONE long released' in long_out, long_out
|
||
assert 'did not answer' not in long_out, long_out
|
||
|
||
stopped = _typed_in_thread(c, '/ppslow stop')
|
||
wait_until(lambda: (marks / 'started-stop').exists(), timeout=30, what='the handler to start')
|
||
app = SimpleNamespace(current_buffer=Buffer(), invalidate=lambda: None,
|
||
exit=lambda: setattr(c, '_should_exit', True))
|
||
c._tui_handle_ctrl_c(SimpleNamespace(app=app)) # the classic TUI's Ctrl+C binding
|
||
stopped.join(60)
|
||
stop_out = capsys.readouterr().out
|
||
stop_alive, should_exit = stopped.is_alive(), c._should_exit
|
||
# The stopped handler ran on a socket of its own: the next command does not wait for it.
|
||
assert c._process_unregistered_slash('/ppquick', '/ppquick') is True
|
||
after_out = capsys.readouterr().out
|
||
_release(gate, 'late')
|
||
wait_until(lambda: (marks / 'done-stop').exists(), timeout=30, what='the handler to finish')
|
||
proc.send_signal(signal.SIGINT)
|
||
proc.wait(timeout=20)
|
||
finally:
|
||
conn.close()
|
||
assert not stop_alive and not should_exit, (stop_alive, should_exit)
|
||
assert 'stopped waiting' in stop_out and 'keeps running in the gateway' in stop_out, stop_out
|
||
assert 'SLOW_DONE' not in stop_out, stop_out
|
||
assert 'QUICK_DONE' in after_out and 'SLOW_DONE' not in after_out, after_out # the late reply is dropped
|