fix(cli): monitor steer/stop run off the UI loop; one subagent backend for the dock

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.
This commit is contained in:
Teknium committed 2026-10-19 07:12:58 -07:00
1 parent 2aef21193c
commit 08a0b58f93
21 files changed
+132 -42

No files matched your search

+41 -41
View File
@@ -66,28 +66,23 @@ class SubagentMonitor:
row = self.selected
return row if row and row.get('kind') == 'process' else None
def _subagent_rows(self):
"""``(rows, activity)``: the session's live children and their lifecycle records. Rewired, the
children run in the gateway, so both come from its ``subagent.list`` (CLI-12)."""
from hermes_cli.gateway_tools import process_view
def subagents(self):
"""The subagent backend: the gateway session's view when its children run there (CLI-12),
else this process's own registry; both answer the same ``subagent_*`` calls."""
from hermes_cli.gateway_tools import LocalSubagents, process_view
view = process_view()
if view is not None:
rows = view.subagent_rows(busy=bool(getattr(self.cli, '_agent_running', False)))
return rows, {r['subagent_id']: r for r in rows}
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;
# its matching public lifecycle record supplies the latest observed tool.
return entries, ({r['subagent_id']: r for r in list_active_subagents()} if entries else {})
return view
if getattr(self, '_local', None) is None:
self._local = LocalSubagents(self.cli)
return self._local
def refresh(self, now=None):
now = time.time() if now is None else now
entries, activity = self._subagent_rows()
entries = self.subagents().subagent_rows(busy=bool(getattr(self.cli, '_agent_running', False)))
for row in entries:
live = activity.get(row['subagent_id'], {})
row['elapsed'] = max(0, int(now - (live.get('started_at') or now)))
row['last_tool'] = live.get('last_tool') or ''
row['elapsed'] = max(0, int(now - (row.get('started_at') or now)))
row['last_tool'] = row.get('last_tool') or ''
row['key'] = row['subagent_id']
row.pop('running_seconds', None)
processes = procs.process_rows(now)
@@ -138,12 +133,12 @@ class SubagentMonitor:
def _poll_tail(self):
"""Off the render path (the spinner thread's tick): refresh the gateway tail the open detail
view paints from the cache. True when the text changed."""
from hermes_cli.gateway_tools import process_view
row, view = self.selected, process_view()
if not self.detail or view is None or row is None or row.get('kind') == 'process':
row = self.selected
if not self.detail or row is None or row.get('kind') == 'process':
return False
before = view.subagent_tail(row['subagent_id'])
return view.subagent_tail(row['subagent_id'], fetch=True) != before
backend = self.subagents()
before = backend.subagent_tail(row['subagent_id'])
return backend.subagent_tail(row['subagent_id'], fetch=True) != before
def select(self, delta):
roster = self.roster
@@ -157,12 +152,7 @@ class SubagentMonitor:
if action != 'stop':
return {'error': t('cli.subagents.cannot_steer_process')}
return procs.kill(target)
from hermes_cli.gateway_tools import process_view
view = process_view()
if view is not None:
return view.subagent_control(action, target, message)
from tools.delegate_tool_registry import _handle_control_action
return json.loads(_handle_control_action(action, target, message, getattr(self.cli, 'agent', None)))
return self.subagents().subagent_control(action, target, message)
def _counts(self, *, session=True):
"""Count fragment: ``2 live``, ``1 proc``, ``2 live · 3 procs``; the collapsed heading
@@ -290,15 +280,10 @@ def read_tail(path):
return t('cli.subagents.transcript_unavailable')
def subagent_tail(row):
"""The child's transcript tail: the gateway's ``subagent.tail`` when it runs the child (fetched and
filtered by the monitor's tick, never here: the detail view calls this every frame), else the
transcript file this process wrote."""
from hermes_cli.gateway_tools import process_view
view = process_view()
if view is None:
return read_tail(row.get('live_transcript'))
text = view.subagent_tail(row['subagent_id'])
def subagent_tail(monitor, row):
"""The child's transcript tail from the monitor's backend (the gateway's is fetched and filtered by
the monitor's tick, never here: the detail view calls this every frame)."""
text = monitor.subagents().subagent_tail(row['subagent_id'])
return t('cli.subagents.transcript_unavailable') if text is None else text
@@ -354,7 +339,7 @@ def build_monitor_application(monitor, **kwargs):
elif row.get('kind') == 'process':
text = procs.process_tail(row['id'])
else:
text = subagent_tail(row)
text = subagent_tail(monitor, row)
if text != tail.text:
following = tail.buffer.cursor_position == len(tail.text)
position = tail.buffer.cursor_position
@@ -384,6 +369,23 @@ def build_monitor_application(monitor, **kwargs):
steer = '' if process else t('cli.subagents.footer_steer_key_long')
return t('cli.subagents.footer_detail' if monitor.detail else 'cli.subagents.footer_roster', steer=steer)
def run_control(action, message, target):
"""Steer/stop off the UI loop: a stalled gateway call (up to its timeout) must not freeze the
monitor's repaint, Esc or Ctrl+C. The answer lands in the notice line when it arrives."""
import asyncio
state['notice'] = t('cli.subagents.control_pending')
def done(future):
try:
result = future.result()
except Exception as exc: # health: allow BLE001 -- a control failure is shown, never raised into the UI loop
result = {'error': str(exc)}
state['notice'] = result.get('error') or result.get('note') or str(result)
app.invalidate()
asyncio.get_running_loop().run_in_executor(
None, lambda: monitor.control(action, message, target=target)).add_done_callback(done)
kb = KeyBindings()
normal = Condition(lambda: not state['steering'] and not state['confirm'])
listing = normal & Condition(lambda: not monitor.detail)
@@ -415,8 +417,7 @@ def build_monitor_application(monitor, **kwargs):
def send_steer(event):
if not steer.text.strip():
return
result = monitor.control('steer', steer.text, target=state['target'])
state['notice'] = result.get('error') or result.get('note') or str(result)
run_control('steer', steer.text, state['target'])
steer.text = ''
state['steering'] = False
app.layout.focus(tail if monitor.detail else roster)
@@ -429,8 +430,7 @@ def build_monitor_application(monitor, **kwargs):
@kb.add('y', filter=Condition(lambda: state['confirm']))
def confirm(event):
result = monitor.control('stop', target=state['target'])
state['notice'] = result.get('error') or result.get('note') or str(result)
run_control('stop', None, state['target'])
state['confirm'] = False
@kb.add('escape', eager=True)
+31
View File
@@ -205,6 +205,37 @@ class GatewayProcessView:
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."""
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Tou · {count} in tou · /queue list{controls}"
more_queued: "+{count} meer"
transcript_unavailable: "Lewendige transkripsie is nog nie beskikbaar nie."
control_pending: "Stuur tans…"
steer_prompt: "Stuur: "
roster_empty: "Geen lewendige subagente of agtergrondprosesse nie. Resultate kom in die gesprek aan."
entry_gone: "Hierdie inskrywing is nie meer lewendig nie."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "قائمة الانتظار · {count} في الانتظار · /queue list{controls}"
more_queued: "+{count} أخرى"
transcript_unavailable: "النص المباشر غير متاح بعد."
control_pending: "جارٍ الإرسال…"
steer_prompt: "توجيه: "
roster_empty: "لا وكلاء فرعيين نشطين ولا عمليات في الخلفية. تصل النتائج في المحادثة."
entry_gone: "هذا العنصر لم يعد نشطًا."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Warteschlange · {count} wartend · /queue list{controls}"
more_queued: "+{count} weitere"
transcript_unavailable: "Live-Transkript noch nicht verfügbar."
control_pending: "Wird gesendet…"
steer_prompt: "Steuern: "
roster_empty: "Keine aktiven Subagenten oder Hintergrundprozesse. Ergebnisse erscheinen in der Konversation."
entry_gone: "Dieser Eintrag ist nicht mehr aktiv."
+1
View File
@@ -2346,6 +2346,7 @@ cli:
queue_heading: "Queue · {count} queued · /queue list{controls}"
more_queued: "+{count} more"
transcript_unavailable: "Live transcript not available yet."
control_pending: "Sending…"
steer_prompt: "Steer: "
roster_empty: "No live subagents or background processes. Results arrive in the conversation."
entry_gone: "This entry is no longer live."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Cola · {count} en cola · /queue list{controls}"
more_queued: "+{count} más"
transcript_unavailable: "La transcripción en vivo aún no está disponible."
control_pending: "Enviando…"
steer_prompt: "Redirigir: "
roster_empty: "No hay subagentes activos ni procesos en segundo plano. Los resultados llegan a la conversación."
entry_gone: "Esta entrada ya no está activa."
+1
View File
@@ -2520,6 +2520,7 @@ cli:
queue_heading: "File d'attente · {count} en file · /queue list{controls}"
more_queued: "+{count} de plus"
transcript_unavailable: "Transcription en direct pas encore disponible."
control_pending: "Envoi…"
steer_prompt: "Orienter : "
roster_empty: "Aucun sous-agent ni processus en arrière-plan en cours. Les résultats arrivent dans la conversation."
entry_gone: "Cette entrée n'est plus active."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Scuaine · {count} i scuaine · /queue list{controls}"
more_queued: "+{count} eile"
transcript_unavailable: "Níl an traschríbhinn bheo ar fáil fós."
control_pending: "Á sheoladh…"
steer_prompt: "Stiúir: "
roster_empty: "Níl aon fho-ghníomhairí beo ná próisis chúlra ann. Tagann na torthaí isteach sa chomhrá."
entry_gone: "Níl an iontráil seo beo a thuilleadh."
+1
View File
@@ -2264,6 +2264,7 @@ cli:
queue_heading: "Sor · {count} sorban · /queue list{controls}"
more_queued: "+{count} további"
transcript_unavailable: "Élő átirat még nem érhető el."
control_pending: "Küldés…"
steer_prompt: "Irányítás: "
roster_empty: "Nincs élő alügynök vagy háttérfolyamat. Az eredmények a beszélgetésben érkeznek."
entry_gone: "Ez a bejegyzés már nem élő."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Coda · {count} in coda · /queue list{controls}"
more_queued: "+{count} altri"
transcript_unavailable: "Trascrizione in tempo reale non ancora disponibile."
control_pending: "Invio in corso…"
steer_prompt: "Indirizza: "
roster_empty: "Nessun subagente attivo né processo in background. I risultati arrivano nella conversazione."
entry_gone: "Questa voce non è più attiva."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "キュー · {count} 件キュー待ち · /queue list{controls}"
more_queued: "+{count} 件"
transcript_unavailable: "ライブトランスクリプトはまだ利用できません。"
control_pending: "送信中…"
steer_prompt: "ステア: "
roster_empty: "稼働中のサブエージェントやバックグラウンドプロセスはありません。結果は会話内に届きます。"
entry_gone: "この項目はすでに終了しています。"
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "대기열 · {count}개 대기 중 · /queue list{controls}"
more_queued: "+{count}개 더"
transcript_unavailable: "실시간 기록을 아직 볼 수 없어요."
control_pending: "보내는 중…"
steer_prompt: "조정: "
roster_empty: "진행 중인 서브에이전트나 백그라운드 프로세스가 없어요. 결과는 대화에 도착해요."
entry_gone: "이 항목은 더 이상 진행 중이 아니에요."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Fila · {count} na fila · /queue list{controls}"
more_queued: "+{count} mais"
transcript_unavailable: "Transcrição ao vivo ainda não disponível."
control_pending: "Enviando…"
steer_prompt: "Redirecionar: "
roster_empty: "Nenhum subagente ativo nem processo em segundo plano. Os resultados chegam na conversa."
entry_gone: "Esta entrada não está mais ativa."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Очередь · в очереди: {count} · /queue list{controls}"
more_queued: "+{count} ещё"
transcript_unavailable: "Живой транскрипт пока недоступен."
control_pending: "Отправка…"
steer_prompt: "Корректировка: "
roster_empty: "Нет активных подагентов или фоновых процессов. Результаты появятся в разговоре."
entry_gone: "Эта запись больше не активна."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Kuyruk · {count} sırada · /queue list{controls}"
more_queued: "+{count} tane daha"
transcript_unavailable: "Canlı transkript henüz kullanılamıyor."
control_pending: "Gönderiliyor…"
steer_prompt: "Yönlendir: "
roster_empty: "Canlı alt ajan veya arka plan süreci yok. Sonuçlar konuşmaya gelir."
entry_gone: "Bu kayıt artık canlı değil."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "Черга · {count} у черзі · /queue list{controls}"
more_queued: "+{count} ще"
transcript_unavailable: "Транскрипт у реальному часі ще недоступний."
control_pending: "Надсилання…"
steer_prompt: "Скерувати: "
roster_empty: "Немає активних субагентів чи фонових процесів. Результати з'являться в розмові."
entry_gone: "Цей запис більше не активний."
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "佇列 · {count} 個排隊中 · /queue list{controls}"
more_queued: "還有 {count} 個"
transcript_unavailable: "即時逐字稿尚未可用。"
control_pending: "傳送中…"
steer_prompt: "引導:"
roster_empty: "沒有進行中的子代理或背景程序。結果會出現在對話中。"
entry_gone: "此項目已不再進行中。"
+1
View File
@@ -2247,6 +2247,7 @@ cli:
queue_heading: "队列 · {count} 个排队中 · /queue list{controls}"
more_queued: "+{count} 个"
transcript_unavailable: "实时记录尚不可用。"
control_pending: "发送中…"
steer_prompt: "引导:"
roster_empty: "没有运行中的子代理或后台进程。结果会出现在对话中。"
entry_gone: "此条目已不再运行。"
@@ -84,7 +84,7 @@ def test_monitor_lists_tails_and_controls_the_gateway_sessions_children(gateway)
monitor.detail = True # the open detail view: the spinner thread's tick fetches its tail
client.tail_text = "step 1\r\n\x1b[2Jstep 2\n" # a child's terminal control codes
monitor.tick()
assert cli_subagent_monitor.subagent_tail(monitor.selected) == "step 1\n[2Jstep 2\n" # stripped
assert cli_subagent_monitor.subagent_tail(monitor, monitor.selected) == "step 1\n[2Jstep 2\n" # stripped
assert "note" in monitor.control("steer", "Focus on the tokenizer")
assert client.steered == ["Focus on the tokenizer"]
@@ -60,3 +60,45 @@ def test_extended_tail_is_bounded_and_literal(tmp_path):
assert 'last activity' in tail and '\x1b' not in tail
assert len(tail) <= 32768
assert 'not available' in read_tail(str(tmp_path / 'missing.log'))
def test_a_stalled_steer_never_freezes_the_monitor(monkeypatch):
"""Steer runs off the UI loop: while the backend call hangs, the monitor still repaints and Esc
still closes it; the answer lands in the notice line when it arrives."""
import threading
from hermes_cli.cli_subagent_monitor import SubagentMonitor, build_monitor_application
from prompt_toolkit.input import create_pipe_input
from prompt_toolkit.output import DummyOutput
from tools import delegate_tool_registry as registry
monkeypatch.setattr(registry, '_active_subagents', {})
registry._register_subagent(dict(subagent_id='only', goal='only', owner_agent_session_id='owner',
started_at=1, status='running', agent=SimpleNamespace()))
dock = SubagentMonitor(SimpleNamespace(agent=SimpleNamespace(session_id='owner')))
dock.refresh()
release, entered = threading.Event(), threading.Event()
def stalled(action, message=None, *, target=None):
entered.set()
release.wait(5)
return {'note': 'steered late'}
monkeypatch.setattr(dock, 'control', stalled)
async def run():
with create_pipe_input() as pipe:
app = build_monitor_application(dock, input=pipe, output=DummyOutput())
rendered = asyncio.Event()
app.after_render += lambda app: rendered.set()
task = asyncio.create_task(app.run_async())
await asyncio.wait_for(rendered.wait(), 3)
rendered.clear()
pipe.send_text('sFocus\r')
await asyncio.wait_for(rendered.wait(), 3) # repainted while the call is still hanging
await asyncio.get_running_loop().run_in_executor(None, entered.wait, 3)
assert entered.is_set() and not release.is_set()
rendered.clear()
pipe.send_text('\x1b') # Esc still closes it mid-call
await asyncio.wait_for(rendered.wait(), 3)
pipe.send_text('\x1b')
await asyncio.wait_for(task, 3)
release.set()
asyncio.run(run())