mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
08a0b58f93
Review findings on the gateway subagent dock (#140677): - steer and stop called the gateway synchronously on the monitor UI loop, so a stalled gateway froze repaint, Esc and Ctrl+C for up to the 5 s timeout. They now run in the executor; the notice line says "Sending..." and then the answer. - the gateway-or-local choice was repeated in four places in the monitor. It now lives in one place, SubagentMonitor.subagents(): the gateway view, or a LocalSubagents backend with the same subagent_rows / subagent_tail / subagent_control surface, so the monitor only manages selection and display.
336 lines
15 KiB
Python
336 lines
15 KiB
Python
"""The background-process registry main's classic CLI reads, served by the gateway session.
|
|
|
|
On main the CLI process ran every ``terminal(background=true)`` itself, so the live-work dock,
|
|
the status bar's ``⚙ N`` segment, ``/stop`` and the dock's tail view read
|
|
``tools.process_registry`` in-process. Rewired, the processes run in the gateway: this view
|
|
answers the same calls (``list_sessions`` / ``get`` / ``count_running`` / ``kill_process`` /
|
|
``kill_all``) from the session's ``process.list`` / ``process.kill`` / ``process.stop`` verbs, so
|
|
main's dock and command code render unchanged.
|
|
|
|
Polling: the dock ticks once a second. The view only asks the gateway while it has a reason to
|
|
(a tool finished since the last poll, or a listed process is still running or recently
|
|
exited); the status bar reads the cached count and never blocks a repaint on an RPC.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
|
|
from hermes_cli.gateway_client import GatewayClientError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_POLL_SECONDS = 1.0
|
|
# A finished process stays interesting this long (main's dock RETAIN_SECONDS).
|
|
_RETAIN_SECONDS = 60
|
|
|
|
_view = None
|
|
|
|
|
|
class _Proc:
|
|
"""The ``ProcessSession`` attributes main's dock reads."""
|
|
|
|
def __init__(self, row):
|
|
self.id = row.get("session_id") or ""
|
|
self.command = row.get("command") or ""
|
|
self.exit_code = row.get("exit_code")
|
|
self.started_at = float(row.get("started_epoch") or 0.0)
|
|
self.exited_at = float(row.get("exited_at") or 0.0)
|
|
self.output_buffer = row.get("output_tail") or ""
|
|
self.completion_reason = row.get("completion_reason") or "exited"
|
|
self.exited = row.get("status") == "exited"
|
|
|
|
|
|
class GatewayProcessView:
|
|
def __init__(self, agent):
|
|
self.agent = agent
|
|
self.connection = agent.connection
|
|
self.rows: list = []
|
|
self.dirty = True
|
|
self.delegations = 0
|
|
self._polled = 0.0
|
|
self._lock = threading.Lock()
|
|
self.subagents: list = []
|
|
self._subagents_polled = 0.0
|
|
self._subagents_seen = 0.0
|
|
self._subagents_lock = threading.Lock()
|
|
self._tail = None
|
|
self._subagents_watch_until = 0.0
|
|
|
|
@property
|
|
def session_id(self):
|
|
return self.agent.session_id # follows /new, /resume and compression rotations
|
|
|
|
# -- polling ---------------------------------------------------------------------------
|
|
def mark_dirty(self):
|
|
self.dirty = True
|
|
|
|
def _interesting(self, now):
|
|
return any(not p.exited or (p.exited_at and now - p.exited_at <= _RETAIN_SECONDS) for p in self.rows)
|
|
|
|
def refresh(self, *, force=False):
|
|
now = time.time()
|
|
if not force and (now - self._polled < _POLL_SECONDS or not (self.dirty or self._interesting(now))):
|
|
return
|
|
if not self._lock.acquire(blocking=False):
|
|
return
|
|
try:
|
|
self._polled = now
|
|
self.dirty = False
|
|
result = self.connection.call("process.list", session_id=self.session_id, timeout=5)
|
|
self.rows = [_Proc(row) for row in (result or {}).get("processes") or []]
|
|
except (GatewayClientError, RuntimeError) as exc:
|
|
logger.debug("process.list failed: %s", exc)
|
|
finally:
|
|
self._lock.release()
|
|
|
|
# -- the tools.process_registry surface main's CLI calls ---------------------------------
|
|
def list_sessions(self):
|
|
self.refresh()
|
|
now = time.time()
|
|
return [{"session_id": p.id, "command": p.command[:200],
|
|
"status": "exited" if p.exited else "running", "exit_code": p.exit_code,
|
|
"completion_reason": p.completion_reason if p.exited else None,
|
|
"uptime_seconds": int(now - p.started_at) if p.started_at else 0} for p in self.rows]
|
|
|
|
def get(self, process_id):
|
|
return next((p for p in self.rows if p.id == process_id), None)
|
|
|
|
def count_running(self):
|
|
"""Cached (the status bar repaints off the render thread)."""
|
|
return sum(1 for p in self.rows if not p.exited)
|
|
|
|
def kill_process(self, process_id, **_kw):
|
|
try:
|
|
result = self.connection.call("process.kill", session_id=self.session_id, process_id=process_id)
|
|
except GatewayClientError as exc:
|
|
return {"status": "error", "error": str(exc)}
|
|
self.refresh(force=True)
|
|
return result
|
|
|
|
def kill_all(self, **_kw):
|
|
"""``/stop``: the session's own processes plus its background delegations (``process.stop``)."""
|
|
result = self.connection.call("process.stop", session_id=self.session_id) or {}
|
|
self.delegations = int(result.get("delegations") or 0)
|
|
self.refresh(force=True)
|
|
return int(result.get("killed") or 0)
|
|
|
|
def interrupt_delegations(self):
|
|
"""The delegation half of ``/stop`` (``kill_all`` already ran it when processes were live)."""
|
|
if self.delegations:
|
|
count, self.delegations = self.delegations, 0
|
|
return count
|
|
self.kill_all()
|
|
count, self.delegations = self.delegations, 0
|
|
return count
|
|
|
|
def active_delegations(self):
|
|
"""Running background subagents (the gateway's ``session.usage.active_subagents``)."""
|
|
try:
|
|
usage = self.connection.state(self.session_id).usage or {}
|
|
except Exception: # health: allow BLE001 -- a missing usage cache reads as none running
|
|
return 0
|
|
value = usage.get("active_subagents")
|
|
return value if isinstance(value, int) else 0
|
|
|
|
# -- the subagent registry main's Ctrl+T monitor and dock read ---------------------------
|
|
def watch_subagents(self, seconds=_RETAIN_SECONDS):
|
|
"""Poll ``subagent.list`` for a while even with no turn running: an off-turn spawn
|
|
(``/review``) moves neither ``busy`` nor the usage count the gate below reads."""
|
|
self._subagents_watch_until = time.time() + seconds
|
|
self._subagents_polled = 0.0
|
|
|
|
def subagent_rows(self, *, busy):
|
|
"""The session's live children (``subagent.list``). Asks at most once a second, and only
|
|
while a turn runs (``busy``), children were listed last time, the gateway counts some, or
|
|
an off-turn spawn asked to be watched."""
|
|
now = time.time()
|
|
wanted = (busy or self.subagents or now < self._subagents_watch_until
|
|
or self.active_delegations())
|
|
if wanted and now - self._subagents_polled >= _POLL_SECONDS and self._subagents_lock.acquire(blocking=False):
|
|
try:
|
|
self._subagents_polled = now
|
|
result = self.connection.call("subagent.list", session_id=self.session_id, timeout=5)
|
|
self.subagents = list((result or {}).get("subagents") or [])
|
|
self._subagents_seen = now
|
|
except (GatewayClientError, RuntimeError) as exc:
|
|
logger.debug("subagent.list failed: %s", exc)
|
|
if now - self._subagents_seen > _RETAIN_SECONDS:
|
|
self.subagents = [] # no answer for a minute: drop the rows and stop polling
|
|
finally:
|
|
self._subagents_lock.release()
|
|
return [dict(row) for row in self.subagents]
|
|
|
|
def subagent_tail(self, subagent_id, *, fetch=False):
|
|
"""The child's live transcript tail (``subagent.tail``); None when the gateway has none.
|
|
The detail view repaints every frame and reads only the cache; ``fetch`` (the monitor's
|
|
tick, on the spinner thread) asks the gateway at most once a second, failures included."""
|
|
now = time.time()
|
|
cached = self._tail if self._tail and self._tail[0] == subagent_id else None
|
|
if not fetch or (cached and now - cached[1] < _POLL_SECONDS):
|
|
return cached[2] if cached else None
|
|
try:
|
|
result = self.connection.call("subagent.tail", session_id=self.session_id,
|
|
subagent_id=subagent_id, timeout=5) or {}
|
|
text = result.get("text") if result.get("available") else None
|
|
if text is not None: # filter once here, not on every repaint of the detail view
|
|
text = "".join(c for c in text if c.isprintable() or c in "\n\t")
|
|
except (GatewayClientError, RuntimeError) as exc:
|
|
logger.debug("subagent.tail failed: %s", exc)
|
|
text = None
|
|
self._tail = (subagent_id, now, text)
|
|
return text
|
|
|
|
def subagent_control(self, action, subagent_id, message=None):
|
|
"""Steer (``subagent.steer``) or stop (``subagent.interrupt``) one child of this session, in
|
|
the shape ``delegate_tool_registry._handle_control_action`` answers: ``note`` or ``error``."""
|
|
from tools.delegate_tool_registry import _CONTROL_OUTCOMES
|
|
|
|
status, note, failure = _CONTROL_OUTCOMES[action]
|
|
try:
|
|
if action == "steer":
|
|
result = self.connection.call("subagent.steer", session_id=self.session_id,
|
|
subagent_id=subagent_id, text=message, timeout=5) or {}
|
|
ok = result.get("status") == "queued"
|
|
else:
|
|
result = self.connection.call("subagent.interrupt", session_id=self.session_id,
|
|
subagent_id=subagent_id, timeout=5) or {}
|
|
ok = bool(result.get("found"))
|
|
except GatewayClientError as exc:
|
|
return {"error": str(exc)}
|
|
self._subagents_polled = 0.0 # the next tick re-reads the roster
|
|
if ok:
|
|
return {"action": action, "subagent_id": subagent_id, "status": status, "note": note}
|
|
return {"error": failure.format(sid=subagent_id)}
|
|
|
|
|
|
class LocalSubagents:
|
|
"""This process's own subagent registry behind the view's ``subagent_*`` surface, for a CLI that
|
|
runs its agent in-process: the dock and monitor talk to one backend shape either way."""
|
|
|
|
def __init__(self, cli):
|
|
self.cli = cli
|
|
self._paths: dict = {}
|
|
|
|
def subagent_rows(self, *, busy):
|
|
from tools.delegate_tool_registry import _list_payload, list_active_subagents
|
|
parent = getattr(self.cli, "agent", None)
|
|
entries = _list_payload(parent)["subagents"] if parent is not None else []
|
|
# The scoped control-plane snapshot supplies authority and transcript paths; the public
|
|
# lifecycle record supplies start time and the latest observed tool (as subagent.list does).
|
|
live = {r["subagent_id"]: r for r in list_active_subagents()} if entries else {}
|
|
for row in entries:
|
|
record = live.get(row["subagent_id"], {})
|
|
row["started_at"], row["last_tool"] = record.get("started_at"), record.get("last_tool") or ""
|
|
self._paths = {row["subagent_id"]: row.get("live_transcript") for row in entries}
|
|
return entries
|
|
|
|
def subagent_tail(self, subagent_id, *, fetch=False):
|
|
from hermes_cli.cli_subagent_monitor import read_tail
|
|
return read_tail(self._paths.get(subagent_id))
|
|
|
|
def subagent_control(self, action, subagent_id, message=None):
|
|
import json
|
|
from tools.delegate_tool_registry import _handle_control_action
|
|
return json.loads(_handle_control_action(action, subagent_id, message, getattr(self.cli, "agent", None)))
|
|
|
|
|
|
class _OpenedSession:
|
|
"""The CLI's open gateway session before any chat agent exists (``_init_agent`` builds that on the
|
|
first turn): the view's ``connection`` and ``session_id``, following the CLI's id."""
|
|
|
|
def __init__(self, cli, connection):
|
|
self.cli, self.connection = cli, connection
|
|
|
|
@property
|
|
def session_id(self):
|
|
return self.cli.session_id
|
|
|
|
|
|
def bind(agent):
|
|
"""Called when the CLI's gateway-backed agent is built."""
|
|
global _view
|
|
_view = GatewayProcessView(agent) if getattr(agent, "connection", None) is not None else None
|
|
return _view
|
|
|
|
|
|
def bind_session(cli, connection):
|
|
"""Called when the CLI opens its gateway session, so ``/agents``, ``/stop``, the dock and the status
|
|
bar read the session's processes before the first turn (a resumed session's work keeps running)."""
|
|
return bind(_OpenedSession(cli, connection))
|
|
|
|
|
|
def mark_dirty():
|
|
if _view is not None:
|
|
_view.mark_dirty()
|
|
|
|
|
|
def process_view():
|
|
"""The gateway-backed registry, or None when this process runs its own tools (not rewired) or
|
|
its gateway connection is final (a hang-up keeps the view; its next call redials)."""
|
|
view = _view
|
|
return view if view is not None and not view.connection.final else None
|
|
|
|
|
|
def process_registry():
|
|
"""``tools.process_registry.process_registry`` for main's dock, status bar and ``/stop``."""
|
|
view = process_view()
|
|
if view is not None:
|
|
return view
|
|
from tools.process_registry import process_registry as local
|
|
return local
|
|
|
|
|
|
def background_counts():
|
|
"""The status bar's ``⚙ N`` processes and running background subagents (cached, no RPC)."""
|
|
view = process_view()
|
|
if view is not None:
|
|
return {"active_background_processes": view.count_running(),
|
|
"active_background_subagents": view.active_delegations()}
|
|
from tools.async_delegation import active_count
|
|
from tools.process_registry import process_registry as local
|
|
return {"active_background_processes": local.count_running(), "active_background_subagents": active_count()}
|
|
|
|
|
|
def session_tools():
|
|
"""``/tools``: the gateway session's tool catalog (``tools.show``) as
|
|
``(tools, toolset_of)``: ``get_tool_definitions``-shaped rows (first description sentence) and
|
|
a name -> toolset lookup. None when the CLI is not gateway-backed."""
|
|
view = process_view()
|
|
if view is None:
|
|
return None
|
|
result = view.connection.call("tools.show", session_id=view.session_id) or {}
|
|
tools, toolset_of = [], {}
|
|
for section in result.get("sections") or []:
|
|
for row in section.get("tools") or []:
|
|
toolset_of[row["name"]] = section.get("name")
|
|
tools.append({"type": "function", "function": {"name": row["name"],
|
|
"description": row.get("description") or ""}})
|
|
return tools, toolset_of.get
|
|
|
|
|
|
def reload_gateway_skills():
|
|
"""Rescan skills in the gateway session too (its skill-command map serves this session's
|
|
``/<skill>`` and bundle invocations); the CLI's own rescan still drives main's output."""
|
|
view = process_view()
|
|
if view is None:
|
|
return None
|
|
try:
|
|
return (view.connection.call("skills.reload", session_id=view.session_id) or {}).get("result")
|
|
except GatewayClientError as exc:
|
|
logger.debug("skills.reload failed: %s", exc)
|
|
return None
|
|
|
|
|
|
def handle_skills_slash(cmd, console, cli=None):
|
|
"""Main's ``/skills`` (hub search / install / uninstall write the local skills tree), then the
|
|
gateway's rescan so an install reaches the skill map the session's turns use, and ``cli``'s copy of
|
|
the gateway's skill list is read again."""
|
|
from hermes_cli.skills_hub import handle_skills_slash as local
|
|
local(cmd, console)
|
|
reload_gateway_skills()
|
|
if cli is not None:
|
|
from hermes_cli.gateway_plugin_commands import invalidate
|
|
invalidate(cli)
|