hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(plugins): route inject_message to the TUI session_key queue

Browse Source 
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
This commit is contained in:
brooklyn!
2026-09-24 08:27:32 -05:00
parent 13141c6dcb
commit 177f275b77
7 changed files with 398 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
hermes_cli/plugins.py
View File

@@ -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.<plugin_id>.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.<plugin_id>.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:

9
hermes_cli/web_server.py
View File

@@ -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)

178
tests/tui_gateway/test_plugin_inject_host.py Normal file
View File

@@ -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

4
tui_gateway/entry.py
View File

@@ -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()

101
tui_gateway/plugin_inject.py Normal file
View File

@@ -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=("_",))

3
tui_gateway/server.py
View File

@@ -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,

4
website/docs/user-guide/features/plugins.md
View File

@@ -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.