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

feat(plugins): inject_message(origin=...) starts a gateway session in the plugin's own profile

Browse Source 
`ctx.inject_message` could only reach an existing gateway session (`session_key`), so an
orchestrator plugin that opens a chat/thread itself (a bot-authored Discord forum post never
comes back as an inbound message) had no way to get an agent turn there.

`origin` takes the SessionSource.to_dict() shape. The gateway creates (or continues) that chat's
session through the adapter's normal message path, in the profile whose plugin manager loaded the
plugin, and the reply goes back to the origin chat:

- Profile: resolved from the manager's immutable home, never an ambient one. The origin may not
  name another profile, the gateway's own routing for that chat through that profile's adapter
  must land in the plugin's profile, and the profile needs its own connected adapter for the
  platform (another profile's bot is never borrowed). Refusals are synchronous False + one log line.
- Authorization: same current-allowlist check as the session_key path (no bypass for host code).
- Multiplex: handle_message runs inside the profile's runtime scope so the spawned turn AND the
  reply send after it inherit that profile's home and secrets (applies to the session_key path too,
  which used to send from the default loop context).
- allow_gateway_injection is read from the plugin's own profile config.
- Plugin/system turns (event.internal) skip the human first-contact notes (onboarding intro and
  the "no home channel" nudge), which were posted into the plugin's fresh thread.

Requested by canonicle-desk in #132609.

Closes #132609
This commit is contained in:
Teknium
2026-10-05 16:05:39 -07:00
parent 3a2588f014
commit 7b2ff7a7d4
7 changed files with 528 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
gateway/run_plugin_injection.py
View File

@@ -1,12 +1,23 @@
"""Plugin-triggered gateway turns (``PluginContext.inject_message``): the process-wide injector this
runner publishes, the thread-safe scheduler, and the dispatch through the adapter message path."""
runner publishes, the thread-safe scheduler, and the dispatch through the adapter message path.
Two targets: an existing session (``session_key``; route and session id pinned from the store) or a
messaging chat (``origin``) whose session is created — or continued — through the normal session path
in the calling plugin's OWN profile. The origin path never chooses a profile: the gateway's routing
for that chat through that profile's adapter must already land in the plugin's profile, so a human
follow-up in the same chat keys into the same session."""
from __future__ import annotations
import asyncio
import concurrent.futures
import contextlib
import logging
import weakref
from pathlib import Path
from typing import Any, Mapping, Optional
from gateway.platforms.event import MessageEvent, MessageType
from gateway.session import SessionSource
logger = logging.getLogger("gateway.run")
@@ -25,17 +36,29 @@ class GatewayPluginInjectionMixin:
clear_published_gateway_message_host(self)
def _schedule_plugin_message_injection(
self, *, session_key: str, content: str, plugin_id: str
self, *, content: str, plugin_id: str, session_key: Optional[str] = None,
origin: Optional[Mapping[str, Any]] = None, plugin_home: Optional[Path] = None,
) -> bool:
"""Schedule a plugin-triggered turn on the live gateway loop (thread-safe)."""
"""Schedule a plugin-triggered turn on the live gateway loop (thread-safe).
With ``origin`` the profile, origin and adapter checks run here, synchronously, so the plugin
gets ``False`` for a chat it may not start; authorization is rechecked at dispatch."""
from gateway.run import safe_schedule_threadsafe
loop = getattr(self, "_gateway_loop", None)
if not getattr(self, "_running", False) or loop is None or loop.is_closed():
return False
coro = self._dispatch_plugin_message_injection(
session_key=session_key, content=content, plugin_id=plugin_id,
)
if origin is not None:
source = self._plugin_origin_source(origin, plugin_home=plugin_home, plugin_id=plugin_id)
if source is None:
return False
target = f"origin={source.platform.value}/{source.chat_id}"
coro = self._dispatch_plugin_origin_injection(source=source, content=content, plugin_id=plugin_id)
else:
target = f"session={session_key}"
coro = self._dispatch_plugin_message_injection(
session_key=session_key, content=content, plugin_id=plugin_id,
)
try:
current_loop = asyncio.get_running_loop()
except RuntimeError:
@@ -62,63 +85,160 @@ class GatewayPluginInjectionMixin:
try:
if completed.result():
return
what, exc = "was not routed", None
except (asyncio.CancelledError, concurrent.futures.CancelledError):
return
except Exception as err:
what, exc = "failed", err
logger.warning(
"Plugin message injection %s: plugin=%s session=%s", what, plugin_id, session_key, exc_info=exc,
)
except Exception: # dispatch boundary: the plugin already got True; log, never raise
logger.warning("Plugin message injection failed: plugin=%s %s", plugin_id, target, exc_info=True)
return
logger.warning("Plugin message injection was not routed: plugin=%s %s", plugin_id, target)
future.add_done_callback(_log_result)
return True
def _plugin_injection_accepting(self) -> bool:
return getattr(self, "_running", False) and not getattr(self, "_draining", False)
def _plugin_injection_scope(self, source: SessionSource):
"""Multiplex: enter the source's profile runtime scope around ``handle_message`` so the
background task it spawns — the turn AND the reply send after the handler returns — inherits
that profile's home and secrets (this coroutine runs from the default loop context, not from
the profile adapter's own intake task)."""
if getattr(getattr(self, "config", None), "multiplex_profiles", False) is not True:
return contextlib.nullcontext()
from gateway.run import _async_profile_runtime_scope
return _async_profile_runtime_scope(self._resolve_profile_home_for_source(source))
def _plugin_injection_authorized(self, source: SessionSource, *, plugin_id: str, target: str) -> bool:
"""Current gateway authorization for the route the injected turn will run as. Host-side
plugin code is trusted to ask, never to widen access: a stored or supplied route must pass
today's allowlists / pairing / allow-all, and adapter-time grants (Discord roles) do not count."""
try:
authorized = self._is_user_authorized_for_source(source, allow_adapter_delegation=False)
except Exception:
logger.warning("Plugin message injection authorization check failed: plugin=%s %s",
plugin_id, target, exc_info=True)
return False
if not authorized:
logger.warning("Plugin message injection denied by current gateway authorization: plugin=%s %s",
plugin_id, target)
return authorized
async def _dispatch_plugin_message_injection(
self, *, session_key: str, content: str, plugin_id: str
) -> bool:
"""Route a plugin-triggered turn through the session's live adapter."""
def _accepting() -> bool:
return getattr(self, "_running", False) and not getattr(self, "_draining", False)
if not _accepting():
if not self._plugin_injection_accepting():
return False
entry = await self.async_session_store.lookup_by_session_key(session_key)
if entry is None or entry.origin is None or not _accepting():
if entry is None or entry.origin is None or not self._plugin_injection_accepting():
return False
from gateway.session_identity import replace_source
source = replace_source(self._restored_source(entry))
try:
authorized = self._is_user_authorized_for_source(source, allow_adapter_delegation=False)
except Exception:
logger.warning(
"Plugin message injection authorization check failed: plugin=%s session=%s",
plugin_id, session_key, exc_info=True,
)
return False
if not authorized:
logger.warning(
"Plugin message injection denied by current gateway authorization: "
"plugin=%s session=%s", plugin_id, session_key,
)
if not self._plugin_injection_authorized(source, plugin_id=plugin_id, target=f"session={session_key}"):
return False
adapter = self._delivery_adapter_for(source)
if adapter is None:
return False
await adapter.handle_message(MessageEvent(
text=content, message_type=MessageType.TEXT, source=source, internal=True,
allow_gateway_control=False,
metadata={
"hermes_plugin_id": plugin_id, "hermes_plugin_injection": True,
"gateway_session_key": session_key, "gateway_session_id": entry.session_id,
"gateway_session_strict": True,
},
))
async with self._plugin_injection_scope(source):
await adapter.handle_message(MessageEvent(
text=content, message_type=MessageType.TEXT, source=source, internal=True,
allow_gateway_control=False,
metadata={
"hermes_plugin_id": plugin_id, "hermes_plugin_injection": True,
"gateway_session_key": session_key, "gateway_session_id": entry.session_id,
"gateway_session_strict": True,
},
))
logger.info(
"Plugin message injection dispatched: plugin=%s session=%s session_id=%s",
plugin_id, session_key, entry.session_id,
)
return True
def _served_profile_for_home(self, home: Optional[Path]) -> Optional[str]:
"""Name of the profile this gateway serves at *home* (a plugin manager's immutable home), or
``None`` when this gateway does not serve it."""
if home is None:
return None
from hermes_constants import get_routing_process_hermes_home, hermes_home_key
key = hermes_home_key(home)
if getattr(getattr(self, "config", None), "multiplex_profiles", False):
from gateway.run import _multiplex_profile_homes
try:
served = _multiplex_profile_homes(self.config)
except Exception:
logger.warning("Plugin message injection: served-profile set unavailable", exc_info=True)
return None
return next((name for name, served_home in served if hermes_home_key(served_home) == key), None)
if hermes_home_key(get_routing_process_hermes_home()) != key:
return None
return getattr(self, "_primary_profile_name", None) or self._active_profile_name()
def _plugin_origin_source(
self, origin: Mapping[str, Any], *, plugin_home: Optional[Path], plugin_id: str,
) -> Optional[SessionSource]:
"""The canonicalized source for a plugin-supplied *origin*, or ``None`` (logged) when the
plugin may not start a turn there: its profile is not served, the origin is malformed or
names another profile, that profile has no live adapter for the platform, or the gateway's
own routing for the chat lands in a different profile."""
def _refuse(reason: str, *args: Any) -> None:
logger.warning("Plugin message injection refused: plugin=%s " + reason, plugin_id, *args)
profile = self._served_profile_for_home(plugin_home)
if profile is None:
return _refuse("its profile home %s is not served by this gateway", plugin_home)
fields = dict(origin)
requested = fields.pop("profile", None)
if requested and requested != profile:
return _refuse("origin names profile %r; plugins may only start sessions in their own profile (%s)",
requested, profile)
try:
source = SessionSource.from_dict(fields)
except (KeyError, TypeError, ValueError) as exc:
return _refuse("invalid origin (%s)", exc)
where = f"{source.platform.value}/{source.chat_id}"
adapter = self._adapters_for_profile(profile).get(source.platform)
if adapter is None:
return _refuse("no connected %s adapter for profile %s", source.platform.value, profile)
# Same provenance build_source() stamps on a live inbound event from this adapter, so the
# identity (receiving bot, runtime profile, authorization home) resolves exactly as it would
# for a human message in this chat.
source._transport_adapter_ref = weakref.ref(adapter)
identity = self._canonicalize(source)
if identity is None:
return _refuse("origin %s does not resolve to a served profile", where)
if identity.runtime_profile != profile:
return _refuse("origin %s routes to profile %s, not the plugin's profile %s",
where, identity.runtime_profile, profile)
return source
async def _dispatch_plugin_origin_injection(
self, *, source: SessionSource, content: str, plugin_id: str
) -> bool:
"""Start (or continue) the session at a plugin-supplied origin through the adapter's normal
message path: ``get_or_create_session`` creates it, the reply goes back to that chat."""
target = f"origin={source.platform.value}/{source.chat_id}"
if not self._plugin_injection_accepting():
return False
adapter = self._delivery_adapter_for(source)
if adapter is None:
logger.warning("Plugin message injection: adapter for %s went away before dispatch", target)
return False
if not self._plugin_injection_authorized(source, plugin_id=plugin_id, target=target):
return False
session_key = self._session_key_for_source(source)
async with self._plugin_injection_scope(source):
await adapter.handle_message(MessageEvent(
text=content, message_type=MessageType.TEXT, source=source, internal=True,
allow_gateway_control=False,
# Non-strict: the key is pinned against topic recovery, the session may not exist yet.
metadata={
"hermes_plugin_id": plugin_id, "hermes_plugin_injection": True,
"gateway_session_key": session_key,
},
))
logger.info("Plugin message injection dispatched: plugin=%s %s session=%s", plugin_id, target, session_key)
return True

6
gateway/run_turn.py
View File

@@ -1409,12 +1409,12 @@ class GatewayTurnMixin:
)
return bounded
async def _hmwa_first_contact_notes(self, source, history, turn_sidecar_notes):
async def _hmwa_first_contact_notes(self, source, history, turn_sidecar_notes, internal=False):
"""First-ever-message onboarding note + one-time 'no home channel' prompt (both only when
the session has no history). Delivered on the user message (sidecar), NOT the ephemeral
system prompt: present-on-turn-1/absent-on-turn-2 was a guaranteed prompt diff + rebuild."""
from gateway.run import _gateway_config_home, _home_target_env_var, _load_gateway_config
if history:
if history or internal: # internal = plugin/system turn: no human made first contact
return
human_platform = bool(source.platform) and source.platform not in (Platform.LOCAL, Platform.WEBHOOK)
if human_platform and source.chat_type == "dm" and not await self.async_session_store.has_any_sessions():
@@ -2101,7 +2101,7 @@ class GatewayTurnMixin:
self._clear_session_env(_session_env_tokens)
return t("gateway.errors.history_unavailable"), _session_env_tokens
await self._hmwa_first_contact_notes(source, history, turn_sidecar_notes)
await self._hmwa_first_contact_notes(source, history, turn_sidecar_notes, internal=event.internal)
# Voice channel state rides the user message ONLY when changed (in the system prompt it
# forced a rebuild + prompt-cache re-key per message).

56
hermes_cli/plugins.py
View File

@@ -602,59 +602,29 @@ class PluginContext:
# manager's home, never the active profile's (#65593 constraint).
def inject_message(
self, content: str, role: str = "user", *, session_key: str | None = None,
origin: Mapping[str, Any] | None = None,
) -> bool:
"""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``.
(the durable key, not the ephemeral UI session id). ``origin`` (a
``SessionSource.to_dict()``-shaped mapping: platform, chat_id, chat_type, thread_id,
user_id, ...) instead names a messaging chat: the gateway starts (or continues) that
chat's session in THIS plugin's own profile and runs a turn there; the plugin cannot
target another profile. Non-CLI injection needs ``session_key`` or ``origin`` plus
``plugins.entries.<plugin_id>.allow_gateway_injection`` in the plugin's profile config.
``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:
queue_ = cli._interrupt_queue if getattr(cli, "_agent_running", False) else cli._pending_input
queue_.put(msg)
return True
if not session_key:
logger.warning("inject_message: gateway mode requires an existing session_key")
return False
if not self._gateway_injection_allowed():
logger.warning("inject_message: gateway injection denied for plugin %s; set "
"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
try:
return bool(self._manager.inject_gateway_message(
session_key=session_key, content=msg, plugin_id=self.plugin_id,
))
except Exception:
logger.warning("inject_message: gateway scheduling failed for plugin %s", self.plugin_id,
exc_info=True)
return False
from hermes_cli.plugins_injection import inject_plugin_message
return inject_plugin_message(self, content, role, session_key=session_key, origin=origin)
def _gateway_injection_allowed(self) -> bool:
"""Return whether this plugin may trigger gateway session turns."""
"""Return whether this plugin may trigger gateway session turns (read from the plugin's
own profile config, never the calling thread's ambient home)."""
try:
cfg = load_config_readonly() or {}
with _plugin_home_scope(self._manager.home_path):
cfg = load_config_readonly() or {}
except Exception:
return False
return (_plugin_settings_entry(cfg, self.plugin_id) or {}).get("allow_gateway_injection") is True

62
hermes_cli/plugins_injection.py Normal file
View File

@@ -0,0 +1,62 @@
"""Host routing for ``PluginContext.inject_message``: classic CLI REPL, Ink TUI / desktop, messaging
gateway (existing ``session_key`` or a new session at ``origin``)."""
from __future__ import annotations
import logging
from collections.abc import Mapping
from typing import Any, Optional
logger = logging.getLogger("hermes_cli.plugins")
def inject_plugin_message(
ctx: Any, content: str, role: str, *, session_key: Optional[str],
origin: Optional[Mapping[str, Any]],
) -> bool:
manager = ctx._manager
msg = content if role == "user" else f"[{role}] {content}"
cli = manager._cli_ref
# An origin names a messaging chat; the local REPL is never that chat.
if cli is not None and origin is None:
queue_ = cli._interrupt_queue if getattr(cli, "_agent_running", False) else cli._pending_input
queue_.put(msg)
return True
if session_key and origin is not None:
logger.warning("inject_message: pass session_key or origin, not both")
return False
if not session_key and origin is None:
logger.warning("inject_message: gateway mode requires an existing session_key or an origin")
return False
if origin is not None and not isinstance(origin, Mapping):
logger.warning("inject_message: origin must be a mapping in the SessionSource.to_dict() shape")
return False
if not ctx._gateway_injection_allowed():
logger.warning("inject_message: gateway injection denied for plugin %s; set "
"plugins.entries.%s.allow_gateway_injection: true to allow it",
ctx.plugin_id, ctx.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 session_key and manager.has_tui_message_injector:
try:
if manager.inject_tui_message(session_key=session_key, content=msg, plugin_id=ctx.plugin_id):
return True
except Exception:
logger.warning("inject_message: TUI scheduling failed for plugin %s", ctx.plugin_id, exc_info=True)
return False
if not manager.has_gateway_message_injector:
logger.warning("inject_message: no live gateway is available")
return False
if origin is None:
target: dict[str, Any] = {"session_key": session_key}
else:
# The manager's immutable home is the plugin's profile: the gateway creates the session
# there or nowhere (never the ambient HERMES_HOME of whatever thread called us).
target = {"origin": dict(origin), "plugin_home": manager.home_path}
try:
return bool(manager.inject_gateway_message(**target, content=msg, plugin_id=ctx.plugin_id))
except Exception:
logger.warning("inject_message: gateway scheduling failed for plugin %s", ctx.plugin_id, exc_info=True)
return False

200
tests/gateway/test_plugin_origin_injection.py Normal file
View File

@@ -0,0 +1,200 @@
"""``ctx.inject_message(origin=...)`` starts a gateway session in the plugin's OWN profile (#132609).
Real ``GatewayRunner`` + real ``BasePlatformAdapter`` ingress with a recording transport; only the
model call (``_run_agent_inner``) is replaced, and it records the home and secret scope the turn
actually ran under.
"""
import asyncio
from types import SimpleNamespace
from unittest.mock import patch
import pytest
import hermes_yaml as yaml
import gateway.run as gateway_run
from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult
from gateway.profile_routing import parse_profile_routes
from gateway.run import GatewayRunner
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
PLUGIN = "desk"
class _RecordingAdapter(BasePlatformAdapter):
def __init__(self, label: str):
super().__init__(PlatformConfig(enabled=True, token=f"tok-{label}"), Platform.TELEGRAM)
self.label = label
self.sent: list[tuple[str, str]] = []
self.send_scopes: list[tuple] = [] # (home, PROBE_MARKER) each send ran under
self._running = True
async def connect(self, *, is_reconnect: bool = False) -> bool:
return True
async def disconnect(self) -> None:
self._mark_disconnected()
async def send(self, chat_id, content, reply_to=None, metadata=None):
from agent.secret_scope import get_secret
from hermes_constants import get_hermes_home
self.sent.append((str(chat_id), content))
self.send_scopes.append((get_hermes_home(), get_secret("PROBE_MARKER")))
return SendResult(success=True, message_id=str(len(self.sent)))
async def send_typing(self, chat_id, metadata=None):
return None
async def get_chat_info(self, chat_id):
return {"id": chat_id, "type": "group"}
def _write_home(home, marker: str) -> None:
home.mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text(yaml.safe_dump({
"plugins": {"entries": {PLUGIN: {"allow_gateway_injection": True}}},
}), encoding="utf-8")
(home / ".env").write_text(f"TELEGRAM_ALLOWED_USERS=u1\nPROBE_MARKER={marker}\n", encoding="utf-8")
def _context(home) -> tuple[PluginContext, PluginManager]:
manager = PluginManager(scope_key=str(home))
return PluginContext(PluginManifest(name=PLUGIN, key=PLUGIN, source="user"), manager), manager
def _origin(chat_id: str, **extra) -> dict:
return {"platform": "telegram", "chat_id": chat_id, "chat_type": "group", "thread_id": "7",
"user_id": "u1", "user_name": "Requester", **extra}
@pytest.fixture
def gateway(tmp_path, monkeypatch):
"""Multiplexed gateway serving ``default`` (bot P), ``alpha`` (its own bot A) and ``beta`` (its
own bot B); a route sends chat 999 on bot A to ``beta``."""
home = tmp_path / "hh"
homes = {"default": home, "alpha": home / "profiles" / "alpha", "beta": home / "profiles" / "beta"}
for name, profile_home in homes.items():
_write_home(profile_home, marker=name)
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(gateway_run, "_hermes_home", home)
config = GatewayConfig(multiplex_profiles=True, sessions_dir=home / "sessions")
config.profile_routes = parse_profile_routes([
{"name": "beta-on-a", "platform": "telegram", "profile": "beta", "chat_id": "999", "bot_profile": "alpha"},
])
runner = GatewayRunner(config)
adapters = {name: _RecordingAdapter(name) for name in homes}
runner.adapters = {Platform.TELEGRAM: adapters["default"]}
runner._profile_adapters = {name: {Platform.TELEGRAM: adapters[name]} for name in ("alpha", "beta")}
for name, adapter in adapters.items():
adapter.gateway_runner = runner
if name != "default":
adapter.set_owner_profile(name)
adapter.set_message_handler(runner._make_profile_message_handler(name))
else:
adapter.set_message_handler(runner._handle_message)
turns = []
async def _fake_turn(message, context_prompt, history, source, session_id, **_kwargs):
from agent.secret_scope import get_secret
from hermes_constants import get_hermes_home
turns.append(SimpleNamespace(message=message, history=list(history), session_id=session_id,
home=get_hermes_home(), marker=get_secret("PROBE_MARKER")))
return {"final_response": f"reply-{len(turns)}", "messages": [], "tools": [],
"history_offset": len(history), "last_prompt_tokens": 0}
runner._run_agent_inner = _fake_turn
served = list(homes.items())
with patch("hermes_cli.profiles.profiles_to_serve", return_value=served), \
patch("hermes_cli.profiles.get_profile_dir", side_effect=lambda n: homes[n]), \
patch("hermes_cli.profiles.profile_exists", side_effect=lambda n: n in homes):
yield SimpleNamespace(runner=runner, homes=homes, adapters=adapters, turns=turns)
async def _publish(runner):
runner._gateway_loop = asyncio.get_running_loop()
runner._running = True
runner._install_plugin_message_injector()
async def _settle(runner, adapter, rounds: int = 200):
for _ in range(rounds):
await asyncio.sleep(0.01)
if not runner._background_tasks and not adapter._session_tasks and not adapter._active_sessions:
return
@pytest.mark.asyncio
async def test_origin_creates_session_in_own_profile_and_replies_to_origin(gateway):
"""No session at the origin: the plugin's profile gets a fresh session, the turn runs under that
profile's home and secrets, the reply goes back to the origin chat, and a human follow-up in the
same chat continues the SAME session (A -> B -> A: beta in between, alpha's scope comes back)."""
runner, alpha, beta = gateway.runner, gateway.adapters["alpha"], gateway.adapters["beta"]
ctx_alpha, mgr_alpha = _context(gateway.homes["alpha"])
ctx_beta, mgr_beta = _context(gateway.homes["beta"])
with patch("hermes_cli.plugins._known_plugin_managers", return_value=[mgr_alpha, mgr_beta]):
await _publish(runner)
assert ctx_alpha.inject_message("Kick off the review", origin=_origin("100")) is True
await _settle(runner, alpha)
assert ctx_beta.inject_message("Beta's own post", origin=_origin("200")) is True
await _settle(runner, beta)
assert ctx_alpha.inject_message("Second post", origin=_origin("101")) is True
await _settle(runner, alpha)
first, middle, last = gateway.turns
assert (first.home, first.marker) == (gateway.homes["alpha"], "alpha")
assert (middle.home, middle.marker) == (gateway.homes["beta"], "beta")
assert (last.home, last.marker) == (gateway.homes["alpha"], "alpha")
assert first.message.endswith("Kick off the review") and first.history == []
assert alpha.sent == [("100", "reply-1"), ("101", "reply-3")]
assert beta.sent == [("200", "reply-2")]
# The reply is sent after the handler returns: it must still run in the plugin's profile scope.
assert alpha.send_scopes == [(gateway.homes["alpha"], "alpha")] * 2
assert beta.send_scopes == [(gateway.homes["beta"], "beta")]
assert gateway.adapters["default"].sent == []
entry = runner.session_store._entries["agent:alpha:telegram:group:100:7"]
assert entry.session_id == first.session_id
assert not any(key.startswith(("agent:main:", "agent:beta:")) and ":100:" in key
for key in runner.session_store._entries)
# A human reply in the same chat keys into the plugin-created session. A secondary adapter's
# intake task runs inside its profile scope (it is spawned from connect() under that scope).
reply = alpha.build_source(chat_id="100", chat_type="group", thread_id="7", user_id="u1")
async with gateway_run._async_profile_runtime_scope(gateway.homes["alpha"]):
await alpha.handle_message(MessageEvent(text="thanks", message_type=MessageType.TEXT, source=reply))
await _settle(runner, alpha)
assert gateway.turns[-1].session_id == first.session_id
assert alpha.sent[-1] == ("100", "reply-4")
@pytest.mark.asyncio
async def test_origin_cannot_reach_another_profile(gateway):
"""A plugin in ``alpha`` cannot start a session in ``beta``: not by naming the profile, not
through a chat the gateway routes to ``beta``. Refusal is synchronous and nothing is created."""
runner, alpha = gateway.runner, gateway.adapters["alpha"]
ctx_alpha, mgr_alpha = _context(gateway.homes["alpha"])
with patch("hermes_cli.plugins._known_plugin_managers", return_value=[mgr_alpha]):
await _publish(runner)
assert ctx_alpha.inject_message("x", origin=_origin("100", profile="beta")) is False
assert ctx_alpha.inject_message("x", origin=_origin("999")) is False # routed to beta
assert ctx_alpha.inject_message("x", origin={"platform": "telegram"}) is False # malformed
await _settle(runner, alpha)
assert gateway.turns == [] and alpha.sent == [] and gateway.adapters["beta"].sent == []
assert runner.session_store._entries == {}
@pytest.mark.asyncio
async def test_origin_without_connected_adapter_returns_false(gateway):
runner = gateway.runner
ctx_alpha, mgr_alpha = _context(gateway.homes["alpha"])
runner._profile_adapters["alpha"] = {} # alpha's bot is down; never borrow default's
with patch("hermes_cli.plugins._known_plugin_managers", return_value=[mgr_alpha]):
await _publish(runner)
assert ctx_alpha.inject_message("x", origin=_origin("100")) is False
assert ctx_alpha.inject_message("x", origin={**_origin("100"), "platform": "discord"}) is False
assert runner._background_tasks == set()
assert gateway.adapters["default"].sent == []

64
tests/hermes_cli/test_plugin_message_injection.py
View File

@@ -252,3 +252,67 @@ def test_published_gateway_host_reaches_existing_and_future_profile_managers(tmp
plugins_mod.clear_published_gateway_message_host(owner)
plugins_mod._reset_plugin_managers_for_tests()
secret_scope.set_multiplex_active(was_active)
ORIGIN = {"platform": "discord", "chat_id": "42", "chat_type": "thread", "thread_id": "42", "user_id": "u1"}
def test_origin_injection_targets_the_plugins_own_home(tmp_path, monkeypatch):
"""``origin`` is forwarded with the manager's immutable home — never an ambient one — so the
gateway can only start the session in the profile that loaded the plugin."""
_write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True})
context, manager = _context()
injector = MagicMock(return_value=True)
manager.set_gateway_message_injector(object(), injector)
other = tmp_path / "other"
other.mkdir()
token = set_hermes_home_override(other)
try:
assert context.inject_message("Kick off", origin=ORIGIN) is True
finally:
reset_hermes_home_override(token)
injector.assert_called_once_with(
origin=ORIGIN, plugin_home=manager.home_path, content="Kick off", plugin_id="notify-plugin",
)
def test_origin_injection_permission_comes_from_the_plugins_own_profile(tmp_path, monkeypatch):
"""The allow_gateway_injection gate reads the plugin's profile config even when the calling
thread's ambient home is another profile that grants it."""
_write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": False})
context, manager = _context()
injector = MagicMock(return_value=True)
manager.set_gateway_message_injector(object(), injector)
granting = tmp_path / "granting"
granting.mkdir()
(granting / "config.yaml").write_text(
yaml.safe_dump({"plugins": {"entries": {"notify-plugin": {"allow_gateway_injection": True}}}}),
encoding="utf-8",
)
token = set_hermes_home_override(granting)
try:
assert context.inject_message("x", origin=ORIGIN) is False
finally:
reset_hermes_home_override(token)
injector.assert_not_called()
def test_origin_injection_rejects_ambiguous_or_malformed_targets(tmp_path, monkeypatch):
_write_plugin_config(tmp_path, monkeypatch, {"allow_gateway_injection": True})
context, manager = _context()
injector = MagicMock(return_value=True)
manager.set_gateway_message_injector(object(), injector)
assert context.inject_message("x", session_key="agent:main:discord:thread:42", origin=ORIGIN) is False
assert context.inject_message("x", origin="discord:42") is False
injector.assert_not_called()
def test_origin_injection_never_lands_in_the_attached_cli():
"""An origin names a messaging chat; a CLI-hosted plugin gets False, not a REPL queue entry."""
context, manager = _context()
cli = SimpleNamespace(_agent_running=False, _pending_input=SimpleQueue(), _interrupt_queue=SimpleQueue())
manager._cli_ref = cli
assert context.inject_message("x", origin=ORIGIN) is False
assert cli._pending_input.empty() and cli._interrupt_queue.empty()

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

@@ -103,7 +103,7 @@ Every `ctx.*` API below is available inside a plugin's `register(ctx)` function.
| Add slash commands | `ctx.register_command(name, handler, description)` — adds `/name` in CLI and gateway sessions |
| Dispatch tools from commands | `ctx.dispatch_tool(name, args)` — invokes a registered tool with parent-agent context auto-wired |
| Add CLI commands | `ctx.register_cli_command(name, help, setup_fn, handler_fn)` — adds `hermes <plugin> <subcommand>` |
| Inject messages | `ctx.inject_message(content, role="user", session_key=...)` - see [Injecting Messages](#injecting-messages) |
| Inject messages | `ctx.inject_message(content, role="user", session_key=... \| origin=...)` - see [Injecting Messages](#injecting-messages) |
| Ship data files | `Path(__file__).parent / "data" / "file.yaml"` |
| Bundle skills | `ctx.register_skill(name, path)` — namespaced as `plugin:skill`, loaded via `skill_view("plugin:skill")` |
| Gate on env vars | `requires_env: [API_KEY]` in plugin.yaml — prompted during `hermes plugins install` |
@@ -892,7 +892,7 @@ In a running session, `/plugins` shows which plugins are currently loaded.
## Injecting Messages
Plugins can inject messages into a CLI conversation or a known gateway session using `ctx.inject_message()`:
Plugins can inject messages into a CLI conversation, a known gateway session, or a new gateway session in a given chat using `ctx.inject_message()`:
```python
# Active CLI conversation
@@ -904,9 +904,22 @@ ctx.inject_message(
role="user",
session_key="agent:main:telegram:dm:123456789",
)
# Start (or continue) the session for a chat/thread, e.g. a Discord forum post the plugin just opened
ctx.inject_message(
"Review the attached spec and post a plan here.",
origin={
"platform": "discord",
"chat_id": "1290000000000000000", # the forum post (thread) id
"chat_type": "thread",
"thread_id": "1290000000000000000",
"user_id": "123456789012345678", # who the turn runs as; must pass the gateway's allowlist
"user_name": "Desk Orchestrator",
},
)
```
**Signature:** `ctx.inject_message(content: str, role: str = "user", *, session_key: str | None = None) -> bool`
**Signature:** `ctx.inject_message(content: str, role: str = "user", *, session_key: str | None = None, origin: Mapping[str, Any] | None = None) -> bool`
In CLI mode:
@@ -918,14 +931,22 @@ In CLI mode:
In gateway mode:
- `session_key` is required and must identify an existing gateway session. It is the stable routing key, not the CLI session ID.
- Hermes reuses that session's stored platform, chat, thread, profile, and conversation history. Plugins cannot supply a new chat route through this API.
- Hermes reuses that session's stored platform, chat, thread, profile, and conversation history. Plugins cannot change that route through `session_key`.
- Hermes rechecks the stored route against the gateway's current authorisation rules before dispatch.
- Routes that relied only on an adapter-time or upstream authorisation decision are rejected unless Hermes can revalidate them from current core allowlists, pairing, or explicit allow-all configuration.
- Injected text is always conversational input. It cannot invoke slash commands, approve tools, or resolve pending confirmation and clarification prompts.
- 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 host can accept the request. Unknown or unroutable session keys discovered after asynchronous acceptance are written to the gateway log.
- Returns `False` when neither `session_key` nor `origin` is given (or both are), 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.
Starting a session with `origin` (gateway mode only):
- `origin` uses the same shape as a stored session origin (`SessionSource.to_dict()`): `platform` and `chat_id` are required; `chat_type`, `thread_id`, `user_id`, `user_name`, `chat_name` and the other origin fields are optional. Use it when no session exists yet — for example, a bot-authored forum post, which platforms like Discord never deliver back to the bot as an inbound message.
- The session is created in **the plugin's own profile** (the profile whose plugin manager loaded it) through the normal session path, with a fresh prompt and an empty history, and the reply is delivered to that chat/thread. If the chat already has a session in that profile, the text joins it exactly like a human message would (the busy-session queue applies), so a human follow-up in the same thread continues the same session.
- A plugin can never target another profile. `False` is returned (with the reason in the gateway log) when the origin names a different `profile`, when the gateway's own profile routing sends that chat to a different profile, when the plugin's profile is not served by this gateway, or when that profile has no connected adapter for `platform` — another profile's bot is never borrowed. On a multiplexed gateway the turn runs under the plugin's profile home and secrets.
- Authorisation is not bypassed: the origin's user must pass the gateway's current allowlists, pairing, or allow-all configuration for that profile, the same check as the `session_key` path. Choose a `user_id` that is authorised, or the request is dropped at dispatch and logged.
- The same text rules apply: no slash commands, approvals, or prompt resolution. The permission below is read from the plugin's own profile `config.yaml`.
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.
@@ -941,7 +962,7 @@ plugins:
```
:::warning
Only grant gateway injection to plugins you trust. Hermes checks this host API permission and restricts it to existing session routes, but Python plugins run in-process and this setting is not a sandbox.
Only grant gateway injection to plugins you trust. Hermes checks this host API permission, restricts it to existing session routes or new sessions in the plugin's own profile, and applies current authorisation, but Python plugins run in-process and this setting is not a sandbox.
:::
:::note