From 177f275b77c889a3a4635331d60cbe8836e5e4a2 Mon Sep 17 00:00:00 2001 From: brooklyn! Date: Thu, 24 Sep 2026 08:27:32 -0500 Subject: [PATCH] fix(plugins): route inject_message to the TUI session_key queue Ink TUI and desktop never registered an inject host, and sharing set_gateway_message_injector with a live messaging gateway would let the last writer win. A separate host queues the reported session_key onto that session's prompt queue and leaves other keys for the gateway. Fixes #87412 --- hermes_cli/plugins.py | 113 ++++++++++-- hermes_cli/web_server.py | 9 + tests/tui_gateway/test_plugin_inject_host.py | 178 +++++++++++++++++++ tui_gateway/entry.py | 4 + tui_gateway/plugin_inject.py | 101 +++++++++++ tui_gateway/server.py | 3 +- website/docs/user-guide/features/plugins.md | 4 +- 7 files changed, 398 insertions(+), 14 deletions(-) create mode 100644 tests/tui_gateway/test_plugin_inject_host.py create mode 100644 tui_gateway/plugin_inject.py diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 274deb117f..3a4cbfb28a 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -603,10 +603,14 @@ class PluginContext: def inject_message( self, content: str, role: str = "user", *, session_key: str | None = None, ) -> bool: - """Inject a message into a CLI or gateway conversation (new turn if idle, interrupt if running). - Gateway injection needs an existing ``session_key`` plus - ``plugins.entries..allow_gateway_injection``; ``True`` means the gateway accepted the - request for async dispatch, not that delivery completed.""" + """Inject a message into a CLI, Ink TUI/desktop, or messaging-gateway conversation. + + CLI uses the attached REPL queues. Ink TUI and desktop use a separate injector + from the messaging gateway and queue onto the live session named by ``session_key`` + (the durable key, not the ephemeral UI session id). Non-CLI injection needs that + ``session_key`` plus ``plugins.entries..allow_gateway_injection``. + ``True`` means a host accepted the request, not that the turn completed. + """ cli = self._manager._cli_ref msg = content if role == "user" else f"[{role}] {content}" if cli is not None: @@ -621,6 +625,20 @@ class PluginContext: "plugins.entries.%s.allow_gateway_injection: true to allow it", self.plugin_id, self.plugin_id) return False + # TUI/desktop host is a different slot. It accepts only when it owns this + # session_key; a miss falls through so a co-resident messaging gateway + # still receives its own keys. An exception fails closed — do not also + # hand the same text to the gateway. + if self._manager.has_tui_message_injector: + try: + if self._manager.inject_tui_message( + session_key=session_key, content=msg, plugin_id=self.plugin_id, + ): + return True + except Exception: + logger.warning("inject_message: TUI scheduling failed for plugin %s", self.plugin_id, + exc_info=True) + return False if not self._manager.has_gateway_message_injector: logger.warning("inject_message: no live gateway is available") return False @@ -1172,6 +1190,9 @@ class PluginManager(PluginLoaderMixin, PluginDispatchMixin, PluginLedgerMixin): self._discovered: bool = False self._cli_ref = None # Set by CLI after plugin discovery self._gateway_message_injector: tuple[object, Callable] | None = None + # Ink TUI / desktop. Must not alias ``_gateway_message_injector``: a live + # messaging gateway and a TUI in one process would otherwise clobber each other. + self._tui_message_injector: tuple[object, Callable] | None = None self._context_engine = None # Set by a plugin via register_context_engine() # Manager-local registries keyed by name (see the matching ``PluginContext.register_*``): # plugins, hooks, middleware, CLI + slash commands, prompt sections, skills (qualified name -> @@ -1257,6 +1278,25 @@ class PluginManager(PluginLoaderMixin, PluginDispatchMixin, PluginLedgerMixin): registered = self._gateway_message_injector return registered is not None and bool(registered[1](**kwargs)) + @property + def has_tui_message_injector(self) -> bool: + """Return whether a live Ink TUI / desktop host can accept plugin-triggered turns.""" + return self._tui_message_injector is not None + + def set_tui_message_injector(self, owner: object, injector: Callable[..., bool]) -> None: + """Publish a live TUI/desktop injector. Does not touch the messaging-gateway slot.""" + self._tui_message_injector = (owner, injector) + + def clear_tui_message_injector(self, owner: object) -> None: + """Clear the TUI injector only when it still belongs to ``owner``.""" + if self._tui_message_injector is not None and self._tui_message_injector[0] is owner: + self._tui_message_injector = None + + def inject_tui_message(self, **kwargs: Any) -> bool: + """Submit a plugin-triggered turn to the live TUI/desktop host.""" + registered = self._tui_message_injector + return registered is not None and bool(registered[1](**kwargs)) + def discover_and_load(self, force: bool = False) -> None: """Scan all plugin sources and load each plugin found; ``force`` unloads first so config changes / new bundled backends become visible in long-lived sessions.""" @@ -1562,6 +1602,12 @@ _plugin_manager: Optional[PluginManager] = None _plugin_managers_by_home: Dict[Path, PluginManager] = {} _plugin_managers_lock = threading.RLock() +# Process-wide Ink TUI / desktop host. Not the messaging-gateway slot. Stamped onto +# each profile's manager so a multiplexed desktop process does not drop injects +# aimed at a non-launch profile. ``None`` until the TUI/desktop process installs it. +_published_tui_message_injector: tuple[object, Callable] | None = None +_published_tui_host_lock = threading.Lock() + def _plugin_home_key() -> Path: """Resolved active Hermes home — the key for per-profile plugin managers (plugins capture the @@ -1588,6 +1634,45 @@ def _clear_plugin_submodules(manager: Optional[PluginManager]) -> None: _BARE_MODULE_SCOPE.pop(module_name, None) +def _known_plugin_managers() -> list[PluginManager]: + with _plugin_managers_lock: + managers = list(dict.fromkeys(_plugin_managers_by_home.values())) + if _plugin_manager is not None and _plugin_manager not in managers: + managers.append(_plugin_manager) + return managers + + +def publish_tui_message_host(owner: object, injector: Callable[..., bool]) -> None: + """Remember the process TUI/desktop host and stamp managers that already exist. + + Does not call ``set_gateway_message_injector``. + """ + global _published_tui_message_injector + with _published_tui_host_lock: + _published_tui_message_injector = (owner, injector) + for manager in _known_plugin_managers(): + manager.set_tui_message_injector(owner, injector) + + +def clear_published_tui_message_host(owner: object) -> None: + """Forget the process TUI host and clear it where this owner still holds the slot.""" + global _published_tui_message_injector + with _published_tui_host_lock: + if (_published_tui_message_injector is not None + and _published_tui_message_injector[0] is owner): + _published_tui_message_injector = None + for manager in _known_plugin_managers(): + manager.clear_tui_message_injector(owner) + + +def _attach_published_tui_host(manager: PluginManager) -> None: + """Give a newly resolved manager the process TUI host, if one is installed and the slot is empty.""" + with _published_tui_host_lock: + host = _published_tui_message_injector + if host is not None and manager._tui_message_injector is None: + manager._tui_message_injector = host + + def get_plugin_manager() -> PluginManager: """Return the plugin manager for the active Hermes profile/home (cached per resolved home; a profile switch gets its own manager and plugin submodules).""" @@ -1598,18 +1683,20 @@ def get_plugin_manager() -> PluginManager: # keyed cache doesn't know about at all. if _plugin_manager is not None and _plugin_manager not in _plugin_managers_by_home.values(): _plugin_managers_by_home[current_home] = _plugin_manager - return _plugin_manager - manager = _plugin_managers_by_home.get(current_home) - if manager is None: - manager = PluginManager(scope_key=hermes_home_key(current_home)) - _plugin_managers_by_home[current_home] = manager - _plugin_manager = manager - return manager + manager = _plugin_manager + else: + manager = _plugin_managers_by_home.get(current_home) + if manager is None: + manager = PluginManager(scope_key=hermes_home_key(current_home)) + _plugin_managers_by_home[current_home] = manager + _plugin_manager = manager + _attach_published_tui_host(manager) + return manager def _reset_plugin_managers_for_tests() -> None: """Test-only: drop every cached manager and its submodules for a fully clean slate.""" - global _plugin_manager + global _plugin_manager, _published_tui_message_injector with _plugin_managers_lock: managers = list(dict.fromkeys(_plugin_managers_by_home.values())) if _plugin_manager is not None and _plugin_manager not in managers: @@ -1622,6 +1709,8 @@ def _reset_plugin_managers_for_tests() -> None: logger.debug("test plugin-manager unload failed", exc_info=True) _plugin_managers_by_home.clear() _plugin_manager = None + with _published_tui_host_lock: + _published_tui_message_injector = None # Dashboard-auth providers are persistent and survive a routine unload, so the clean-slate # reset must clear that process-global registry explicitly or a test's provider leaks. try: diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 88d92bc234..df4e227935 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -184,6 +184,11 @@ async def _lifespan(app: "FastAPI"): from tui_gateway import methods_groups as _hosted_groups import tui_gateway.server # noqa: F401 + try: + tui_gateway.server.install_tui_message_injector() + except Exception: + _log.warning("TUI message injector did not install", exc_info=True) + hosted_room_start_cancel = threading.Event() def _start_hosted_rooms() -> None: @@ -263,6 +268,10 @@ async def _lifespan(app: "FastAPI"): try: yield finally: + try: + tui_gateway.server.clear_tui_message_injector() + except Exception: + _log.debug("TUI message injector clear skipped", exc_info=True) hosted_room_start_cancel.set() _hosted_groups.stop_hosted_room_service(timeout=5.0) hosted_room_start_thread.join(timeout=1.0) diff --git a/tests/tui_gateway/test_plugin_inject_host.py b/tests/tui_gateway/test_plugin_inject_host.py new file mode 100644 index 0000000000..ecfc1a59fe --- /dev/null +++ b/tests/tui_gateway/test_plugin_inject_host.py @@ -0,0 +1,178 @@ +"""Plugin inject_message reaches the Ink TUI / desktop session it names. + +The messaging gateway and the TUI must not share ``set_gateway_message_injector``. +A reported ``session_key`` is the routing key — not the ephemeral UI session id — +and the text lands on that session's prompt queue. +""" + +import threading +import time +from types import SimpleNamespace +from unittest.mock import MagicMock + +import yaml + +from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest +from tui_gateway import server + + +def _write_plugin_config(tmp_path, monkeypatch, entry: dict) -> None: + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + yaml.safe_dump({"plugins": {"entries": {"notify-plugin": entry}}}) + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + +def _context() -> tuple[PluginContext, PluginManager]: + manager = PluginManager() + manifest = PluginManifest(name="notify-plugin", key="notify-plugin", source="user") + return PluginContext(manifest, manager), manager + + +def _session(session_key: str, **extra) -> dict: + return { + "agent": SimpleNamespace(), + "session_key": session_key, + "history": [], + "history_lock": threading.Lock(), + "history_version": 0, + "running": False, + "transport": None, + "attached_images": [], + "last_active": 1.0, + **extra, + } + + +def _wait_until(predicate, timeout=1.0) -> bool: + deadline = time.monotonic() + timeout + while not predicate() and time.monotonic() < deadline: + time.sleep(0.01) + return predicate() + + +def test_tui_injector_slot_does_not_replace_the_gateway_slot(): + manager = PluginManager() + gateway = MagicMock(return_value=True) + tui = MagicMock(return_value=False) + gateway_owner, tui_owner = object(), object() + + manager.set_gateway_message_injector(gateway_owner, gateway) + manager.set_tui_message_injector(tui_owner, tui) + + assert manager.has_gateway_message_injector is True + assert manager.has_tui_message_injector is True + assert manager.inject_gateway_message(session_key="agent:main:telegram:dm:1") is True + assert manager.inject_tui_message(session_key="ses_tui") is False + gateway.assert_called_once_with(session_key="agent:main:telegram:dm:1") + tui.assert_called_once_with(session_key="ses_tui") + + manager.clear_tui_message_injector(tui_owner) + assert manager.has_tui_message_injector is False + assert manager.has_gateway_message_injector is True + assert manager.inject_gateway_message(session_key="kept") is True + + +def test_reported_session_key_is_queued_not_the_ui_session_id(tmp_path, monkeypatch): + """The durable session_key, not the ephemeral UI sid, selects the prompt queue.""" + _write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True}) + context, manager = _context() + gateway = MagicMock(return_value=True) + manager.set_gateway_message_injector(object(), gateway) + + origin = _session("ses_origin", running=True, last_active=10.0) + chatty = _session("ses_chatty", running=True, last_active=200.0) + # UI sids deliberately differ from the durable keys a plugin reports. + monkeypatch.setattr(server, "_sessions", {"ui-origin": origin, "ui-chatty": chatty}) + server.install_tui_message_injector(manager) + try: + assert context.inject_message("turn complete", session_key="ses_origin") is True + finally: + server.clear_tui_message_injector(manager) + + assert origin["queued_prompt"]["text"] == "turn complete" + assert chatty.get("queued_prompt") is None + gateway.assert_not_called() + + +def test_unknown_session_key_falls_through_to_the_gateway_slot(tmp_path, monkeypatch): + _write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True}) + context, manager = _context() + gateway = MagicMock(return_value=True) + manager.set_gateway_message_injector(object(), gateway) + live = _session("ses_live", running=True) + monkeypatch.setattr(server, "_sessions", {"ui-live": live}) + server.install_tui_message_injector(manager) + try: + assert context.inject_message( + "wake telegram", session_key="agent:main:telegram:dm:42", + ) is True + finally: + server.clear_tui_message_injector(manager) + + gateway.assert_called_once_with( + session_key="agent:main:telegram:dm:42", + content="wake telegram", + plugin_id="notify-plugin", + ) + assert live.get("queued_prompt") is None + + +def test_missing_session_is_not_rerouted_to_another_tui_session(tmp_path, monkeypatch): + _write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True}) + context, manager = _context() + other = _session("ses_other", running=True, last_active=200.0) + monkeypatch.setattr(server, "_sessions", {"ui-other": other}) + server.install_tui_message_injector(manager) + try: + assert context.inject_message("report", session_key="ses_gone") is False + finally: + server.clear_tui_message_injector(manager) + + assert other.get("queued_prompt") is None + + +def test_idle_session_key_drains_onto_that_prompt_queue(tmp_path, monkeypatch): + _write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True}) + context, manager = _context() + fired = {} + + def fake_run_prompt_submit(rid, sid, session, text, **kwargs): + fired["sid"] = sid + fired["text"] = text + + monkeypatch.setattr(server, "_run_prompt_submit", fake_run_prompt_submit) + monkeypatch.setattr(server, "_session_uses_compute_host", lambda session: False) + origin = _session("ses_idle", running=False) + monkeypatch.setattr(server, "_sessions", {"ui-idle": origin}) + server.install_tui_message_injector(manager) + try: + assert context.inject_message("run finished", session_key="ses_idle") is True + assert _wait_until(lambda: fired.get("text") == "run finished") + finally: + server.clear_tui_message_injector(manager) + + assert fired["sid"] == "ui-idle" + + +def test_cli_injection_still_bypasses_the_tui_host(tmp_path, monkeypatch): + _write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True}) + context, manager = _context() + pending = [] + context._manager._cli_ref = SimpleNamespace( + _agent_running=False, + _pending_input=SimpleNamespace(put=pending.append), + _interrupt_queue=SimpleNamespace(put=lambda item: None), + ) + origin = _session("ses_origin", running=True) + monkeypatch.setattr(server, "_sessions", {"ui-origin": origin}) + server.install_tui_message_injector(manager) + try: + assert context.inject_message("typed", session_key="ses_origin") is True + finally: + server.clear_tui_message_injector(manager) + + assert pending == ["typed"] + assert origin.get("queued_prompt") is None diff --git a/tui_gateway/entry.py b/tui_gateway/entry.py index aebfbeb802..ae251e8f81 100644 --- a/tui_gateway/entry.py +++ b/tui_gateway/entry.py @@ -270,6 +270,10 @@ def _write_or_exit(payload: dict, reason: str) -> None: def main(): # stdout is this process's JSON-RPC client channel: peer-less global broadcasts belong on it. server._stdio_is_rpc_channel = True + try: + server.install_tui_message_injector() + except Exception: + logger.warning("TUI message injector did not install", exc_info=True) _close_rpc_stdin_on_exec() _install_sidecar_publisher() diff --git a/tui_gateway/plugin_inject.py b/tui_gateway/plugin_inject.py new file mode 100644 index 0000000000..66dae32a52 --- /dev/null +++ b/tui_gateway/plugin_inject.py @@ -0,0 +1,101 @@ +"""Ink TUI / desktop host for ``PluginContext.inject_message``. + +Separate from ``PluginManager.set_gateway_message_injector``. A messaging gateway +and this process can both be live; they must not share one slot (last writer +would win). Routing uses the reported ``session_key``, not the ephemeral UI +session id, and lands on that session's prompt queue. + +Bodies are rebound onto ``server.py`` globals at install time +(``method_ctx.bind_module``), so ``_sessions`` and the queue helpers are bare names. +""" + +from __future__ import annotations + +import atexit + +from .method_ctx import bind_module + +# Identity for the process-owned host. A later owner can replace it; clear is +# identity-conditional so this process cannot drop that replacement. +_TUI_INJECT_OWNER = object() +_atexit_registered = False + + +def inject_tui_session_message(*, session_key: str, content: str, plugin_id: str = "") -> bool: + """Queue *content* on the live session whose ``session_key`` matches. + + Returns False when this process has no such session, so the caller can fall + through to the messaging-gateway slot. Never reroutes to a different session. + A busy session only queues (a notice must not cancel in-flight work). An idle + session drains so the queued prompt starts a turn. + """ + del plugin_id # accepted so the host matches the gateway injector kwargs + if not isinstance(session_key, str) or not session_key: + return False + if not isinstance(content, str) or not content.strip(): + return False + with _sessions_lock: + match = next( + ( + (sid, session) + for sid, session in list(_sessions.items()) + if isinstance(session, dict) and session.get("session_key") == session_key + ), + None, + ) + if match is None: + return False + sid, session = match + if session.get("lazy") or session.get("_closing") or session.get("_finalized"): + return False + if session.get("history_lock") is None: + return False + with session["history_lock"]: + queued = session.get("queued_prompt") or {} + keep_transport = queued.get("transport") if isinstance(queued, dict) else None + running = bool(session.get("running")) + _enqueue_prompt(session, content, keep_transport) + session["last_active"] = time.time() + if running: + return True + rid = f"inject-{uuid.uuid4().hex[:8]}" + threading.Thread( + target=_drain_queued_prompt, args=(rid, sid, session), + daemon=True, name=f"tui-inject-{sid[:8]}", + ).start() + return True + + +def install_tui_message_injector(manager=None) -> None: + """Publish this process's TUI host on *manager*, or the active profile's manager. + + Passing a manager (tests) does not publish process-wide. The no-arg form is + the Ink TUI / desktop startup path: every profile manager in this process + gets the host, and managers created later pick it up, without touching the + messaging-gateway slot. + """ + from hermes_cli.plugins import get_plugin_manager, publish_tui_message_host + + global _atexit_registered + if manager is None: + publish_tui_message_host(_TUI_INJECT_OWNER, inject_tui_session_message) + manager = get_plugin_manager() + if not _atexit_registered: + atexit.register(clear_tui_message_injector) + _atexit_registered = True + manager.set_tui_message_injector(_TUI_INJECT_OWNER, inject_tui_session_message) + + +def clear_tui_message_injector(manager=None) -> None: + """Drop this process's host. A different owner is left in place.""" + from hermes_cli.plugins import clear_published_tui_message_host, get_plugin_manager + + if manager is None: + clear_published_tui_message_host(_TUI_INJECT_OWNER) + manager = get_plugin_manager() + manager.clear_tui_message_injector(_TUI_INJECT_OWNER) + + +def register(server) -> None: + """Publish this module's helpers onto ``server``, rebound to its globals.""" + bind_module(globals(), server, skip=("_",)) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 475d106c4a..8f563380c7 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -3361,6 +3361,7 @@ from .mcp_rpc_helpers import summarize_server as _mcp_summarize_server # noqa: from . import ( # noqa: E402 methods_voice as _methods_voice, methods_browser as _methods_browser, methods_slash as _methods_slash, methods_complete_helpers as _methods_complete_helpers, session_auto_continue as _session_auto_continue, + plugin_inject as _plugin_inject, rpc_dispatch as _rpc_dispatch, agent_callbacks as _agent_callbacks, session_history as _session_history, prompt_attachments as _prompt_attachments, session_notifications as _session_notifications, @@ -3384,7 +3385,7 @@ from . import ( # noqa: E402 for _m in ( _session_transports, _session_reaper, _session_lifecycle, _session_workdir, _compute_host_bridge, _model_switch, _session_compression, _change_watcher, _tool_progress, _session_notifications, - _prompt_attachments, _session_history, _agent_callbacks, _session_auto_continue, _rpc_dispatch, + _prompt_attachments, _session_history, _agent_callbacks, _session_auto_continue, _plugin_inject, _rpc_dispatch, _methods_complete_helpers, _methods_slash, _methods_voice, _methods_browser, _methods_browser_control, _methods_session, _methods_prompt, _methods_config, _methods_config_set, _methods_complete, _methods_tools, _methods_profiles, _methods_images, diff --git a/website/docs/user-guide/features/plugins.md b/website/docs/user-guide/features/plugins.md index 1fa3b52fd7..668251aebe 100644 --- a/website/docs/user-guide/features/plugins.md +++ b/website/docs/user-guide/features/plugins.md @@ -816,7 +816,9 @@ In gateway mode: - The route and conversation are pinned while dispatch is pending. Hermes drops the request if topic recovery changes the route or the session rotates before handling starts. - The request enters the platform adapter's normal message path. Active sessions use the existing busy-session queue rather than starting a competing turn. - Returns `True` when the live gateway accepts the request for asynchronous dispatch. This does not confirm that the agent turn or platform delivery has completed. -- Returns `False` when `session_key` is omitted, the permission is not granted, or no live gateway can accept the request. Unknown or unroutable session keys discovered after asynchronous acceptance are written to the gateway log. +- Returns `False` when `session_key` is omitted, the permission is not granted, or no live host can accept the request. Unknown or unroutable session keys discovered after asynchronous acceptance are written to the gateway log. + +Ink TUI (`hermes --tui`) and the desktop / dashboard chat are a third host. They do not set the classic CLI reference and they do not register on the messaging-gateway injector — those two hosts stay separate so a live gateway cannot clobber the TUI (or the reverse). Pass the session's durable `session_key` (the `ses_…` id), not the ephemeral UI session id. Hermes queues the text on that session's prompt queue: a busy session keeps the message for the next turn, an idle session starts one. A key that is not a live TUI session is left for the messaging gateway when one is running, and is never rerouted to a different chat. This enables plugins like remote control viewers, messaging bridges, or webhook receivers to feed messages into the conversation from external sources.