mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
0eb4f62a8b
The plugin-command dispatch binds the live SessionSource (receiving-bot provenance included) task-locally, so a post from a command that is the chat first message goes through the bot that received it instead of a source rebuilt from session env vars.
196 lines
8.4 KiB
Python
196 lines
8.4 KiB
Python
"""``PluginContext.post_message``: one message in the current conversation, editable in place.
|
|
|
|
The conversation is the session bound in the calling context (a tool or hook running in its turn,
|
|
its slash command, a hook fired for it): the plugin never names a chat, so it cannot reach another
|
|
one. Outside any session (a timer, a plugin thread, cron, a process with no live gateway) the call
|
|
is refused with a result, never an exception. Routing and the edit fallback live in
|
|
``gateway/plugin_messages.py``; this side only hands work to the gateway loop and back.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import concurrent.futures
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
logger = logging.getLogger("hermes_cli.plugins")
|
|
|
|
_TIMEOUT_S = 30.0
|
|
|
|
|
|
def _err(code: str, detail: str = "") -> dict[str, Any]:
|
|
return {"ok": False, "error": code, **({"detail": detail} if detail else {})}
|
|
|
|
|
|
class PluginMessage:
|
|
"""Handle for one posted message. ``ok`` / ``message_id`` / ``error`` describe the last
|
|
operation; ``edit(text)`` replaces the text in place (or posts anew where the surface cannot
|
|
edit, and the handle then follows the new message). A refused post returns a handle whose
|
|
edits are refused the same way. Operations stay ordered: one the gateway has not finished
|
|
(``error == "timeout"``, ``pending`` set) is waited for before the next runs, and its outcome
|
|
still lands on the handle, so a slow post never leaves an orphan message behind a dead handle."""
|
|
|
|
def __init__(self, plugin_id: str, target: Any, result: dict[str, Any]):
|
|
self._plugin_id, self._target = plugin_id, target
|
|
self.result = result
|
|
self.message_id: Optional[str] = result.get("message_id")
|
|
self._pending: Optional[concurrent.futures.Future] = None
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return bool(self.result.get("ok"))
|
|
|
|
@property
|
|
def error(self) -> Optional[str]:
|
|
return self.result.get("error")
|
|
|
|
@property
|
|
def pending(self) -> bool:
|
|
return self._pending is not None and not self._pending.done()
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<PluginMessage ok={self.ok} message_id={self.message_id!r} error={self.error!r}>"
|
|
|
|
def _adopt(self, other: PluginMessage) -> None:
|
|
self._target, self.result, self.message_id = other._target, other.result, other.message_id
|
|
|
|
def _refused(self) -> Optional[dict[str, Any]]:
|
|
if self._target is None:
|
|
return _err(self.error or "no_conversation", self.result.get("detail", ""))
|
|
return None
|
|
|
|
def _settle(self, result: dict[str, Any]) -> dict[str, Any]:
|
|
self.result = result
|
|
if result.get("ok") and result.get("message_id"):
|
|
self.message_id = result["message_id"]
|
|
return result
|
|
|
|
async def _edit(self, text: str) -> dict[str, Any]:
|
|
refused = self._refused()
|
|
if refused is not None:
|
|
return refused
|
|
from gateway import plugin_messages
|
|
return self._settle(await plugin_messages.edit(self._target, self.message_id, str(text),
|
|
plugin_id=self._plugin_id))
|
|
|
|
async def aedit(self, text: str) -> dict[str, Any]:
|
|
"""``edit`` for code already running on the gateway's event loop (async hooks/commands)."""
|
|
if self.pending:
|
|
await asyncio.wrap_future(self._pending)
|
|
return await self._edit(text)
|
|
|
|
def edit(self, text: str) -> dict[str, Any]:
|
|
"""Replace the message text: ``{"ok", "message_id", "edited", ["fallback"]}``. Thread-safe."""
|
|
if _on_gateway_loop(): # never block the loop the pending operation needs
|
|
return _err("on_event_loop", "called from the gateway event loop: use await handle.aedit(...)")
|
|
if self.pending:
|
|
try:
|
|
self._pending.result(_TIMEOUT_S)
|
|
except concurrent.futures.TimeoutError:
|
|
return _err("timeout", "the previous operation on this message has not finished")
|
|
except Exception: # its own outcome is already on the handle
|
|
logger.debug("plugin_message: previous operation failed", exc_info=True)
|
|
refused = self._refused()
|
|
if refused is not None:
|
|
return refused
|
|
return _run_on_gateway_loop(self._edit(text), self)
|
|
|
|
|
|
def _gateway_loop():
|
|
try:
|
|
from gateway.run import _gateway_runner_ref
|
|
runner = _gateway_runner_ref()
|
|
except ImportError:
|
|
runner = None
|
|
loop = getattr(runner, "_gateway_loop", None)
|
|
if runner is None or not getattr(runner, "_running", False) or loop is None or loop.is_closed():
|
|
return None, None
|
|
return runner, loop
|
|
|
|
|
|
def _on_gateway_loop() -> bool:
|
|
_runner, loop = _gateway_loop()
|
|
try:
|
|
return loop is not None and asyncio.get_running_loop() is loop
|
|
except RuntimeError:
|
|
return False
|
|
|
|
|
|
def _run_on_gateway_loop(coro, handle: PluginMessage) -> dict[str, Any]:
|
|
"""Run *coro* on the gateway loop and wait up to ``_TIMEOUT_S``; past that it keeps running as
|
|
the handle's pending operation (its result still lands on the handle)."""
|
|
_runner, loop = _gateway_loop()
|
|
if loop is None:
|
|
coro.close()
|
|
return _err("gateway_unavailable", "no live gateway is running in this process")
|
|
if _on_gateway_loop():
|
|
coro.close()
|
|
return _err("on_event_loop", "called from the gateway event loop: use the async form (await ...)")
|
|
future = asyncio.run_coroutine_threadsafe(coro, loop)
|
|
handle._pending = future
|
|
try:
|
|
return future.result(_TIMEOUT_S)
|
|
except concurrent.futures.TimeoutError:
|
|
result = _err("timeout", f"the gateway did not answer within {_TIMEOUT_S:.0f}s; the operation "
|
|
"is still running and its result will land on this handle")
|
|
handle.result = result
|
|
return result
|
|
except Exception as exc: # the plugin gets a result, never the raise
|
|
logger.warning("plugin_message: gateway hand-off failed", exc_info=True)
|
|
return _err("send_failed", str(exc)[:512])
|
|
|
|
|
|
_NO_SESSION = ("no session is bound in the calling context (a tool call, slash command or hook "
|
|
"fired for a session)")
|
|
|
|
|
|
async def _post_for(ctx: Any, text: str, session_key: str, route: Any) -> PluginMessage:
|
|
plugin_id = ctx.plugin_id
|
|
runner, _loop = _gateway_loop()
|
|
result: dict[str, Any]
|
|
if not session_key:
|
|
result = _err("no_conversation", _NO_SESSION)
|
|
elif runner is None:
|
|
result = _err("gateway_unavailable", "no live gateway is running in this process")
|
|
else:
|
|
from gateway import plugin_messages
|
|
# The session store read is disk-bound: keep it off the gateway loop.
|
|
target, refusal = await asyncio.to_thread(plugin_messages.resolve, runner, session_key=session_key,
|
|
route=route, plugin_home=ctx._manager.home_path, plugin_id=plugin_id)
|
|
if target is None:
|
|
result = refusal or _err("no_conversation")
|
|
else:
|
|
result = await plugin_messages.post(target, str(text), plugin_id=plugin_id)
|
|
if result.get("ok"):
|
|
return PluginMessage(plugin_id, target, result)
|
|
logger.info("plugin_message plugin=%s op=post refused: %s", plugin_id, result.get("error"))
|
|
return PluginMessage(plugin_id, None, result)
|
|
|
|
|
|
async def apost(ctx: Any, text: str) -> PluginMessage:
|
|
"""``post`` for code already running on the gateway's event loop (async hooks/commands)."""
|
|
from gateway.plugin_messages import bound_route
|
|
from gateway.session_context import bound_session_key
|
|
return await _post_for(ctx, text, bound_session_key(), bound_route())
|
|
|
|
|
|
def post(ctx: Any, text: str) -> PluginMessage:
|
|
"""Post *text* to the calling session from a tool handler or hook on a worker thread. The session
|
|
is read here, on the caller's own context, then the send runs on the gateway loop."""
|
|
from gateway.session_context import bound_session_key
|
|
session_key = bound_session_key()
|
|
if not session_key:
|
|
return PluginMessage(ctx.plugin_id, None, _err("no_conversation", _NO_SESSION))
|
|
from gateway.plugin_messages import bound_route
|
|
route = bound_route()
|
|
handle = PluginMessage(ctx.plugin_id, None, _err("timeout"))
|
|
|
|
async def _post() -> dict[str, Any]:
|
|
handle._adopt(await _post_for(ctx, text, session_key, route))
|
|
return handle.result
|
|
|
|
result = _run_on_gateway_loop(_post(), handle)
|
|
if handle._target is None and not handle.pending:
|
|
handle.result = result
|
|
return handle
|