mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
ba18ab4d61
The poller claimed parked events before the guarded try, so an exception from take_ready between the claim and the try leaked the claimed copies' held marks, and the ledger sweep then treated those delegations as live forever. Taking the dequeue inside the try puts the claim under the same finally that releases it.
914 lines
49 KiB
Python
914 lines
49 KiB
Python
"""Per-session notification poller: kanban/loop/delegation events routed to the owning session,
|
|
desktop UI wiring, HUD surface note. Bodies are rebound onto server.py's globals at install time
|
|
(method_ctx.bind_module), so they reference server.py globals bare."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
|
|
from .method_ctx import bind_module
|
|
|
|
|
|
def _notif_locked_sessions(fn, default):
|
|
"""Run ``fn(_sessions)`` under ``_sessions_lock``; ``default`` on failure (poller must never crash)."""
|
|
try:
|
|
with _sessions_lock:
|
|
return fn(_sessions)
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def _notif_current_keys(sid: str, session: dict) -> set:
|
|
return {str(session.get("session_key") or ""), _session_lookup_key(session, fallback=sid)}
|
|
|
|
|
|
def _notif_session_matches(s: dict, keys) -> bool:
|
|
return str(s.get("session_key") or "") in keys or _session_lookup_key(s, fallback="") in keys
|
|
|
|
|
|
def _notif_live_session_matches(keys, exclude: dict | None = None) -> bool:
|
|
"""Any non-finalized live session (other than ``exclude``) matches ``keys``; False if the registry can't be read
|
|
(fail open rather than drop the event)."""
|
|
return _notif_locked_sessions(
|
|
lambda ss: any(s is not exclude and not s.get("_finalized") and _notif_session_matches(s, keys)
|
|
for s in ss.values()),
|
|
False)
|
|
|
|
|
|
def _notif_resolve_event_key(evt_key: str, session: dict | None = None) -> str:
|
|
"""Resolve a compression-rotated session key to its continuation tip (or itself). Looked up in
|
|
``session``'s own store: a named-profile session's lineage lives in ``profiles/<x>/state.db``,
|
|
where the launch handle cannot see it."""
|
|
try:
|
|
with _session_db(session or {}) as db:
|
|
return (db.resolve_resume_session_id(evt_key) if db is not None else evt_key) or evt_key
|
|
except Exception:
|
|
return evt_key
|
|
|
|
|
|
def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> bool:
|
|
"""True if ``evt`` is owned by a *different* live session. Background completions carry the ``session_key`` of the
|
|
session that started the work; async delegation completions also carry ``origin_ui_session_id`` (the live TUI tab)."""
|
|
evt_ui_sid = str(evt.get("origin_ui_session_id") or "")
|
|
if evt_ui_sid:
|
|
if evt_ui_sid == str(sid or "") and not session.get("_finalized"):
|
|
return False
|
|
if _notif_locked_sessions(lambda ss: evt_ui_sid in ss and not ss[evt_ui_sid].get("_finalized"), False):
|
|
return True
|
|
# Exact UI tab gone: fall through to durable session_key routing so a resumed continuation with the same
|
|
# key/lineage can still claim it.
|
|
evt_key = str(evt.get("session_key") or "")
|
|
if not evt_key:
|
|
return False
|
|
current_keys = _notif_current_keys(sid, session)
|
|
# Compression can rotate AIAgent.session_id while the detached child is still running: map the event's original
|
|
# key to its continuation tip so it reaches the live session instead of becoming an orphan any poller may consume.
|
|
# A live continuation wins over the compressed parent, else a stale parent tab could consume the event first.
|
|
resolved_key = _notif_resolve_event_key(evt_key, session)
|
|
if resolved_key != evt_key:
|
|
if resolved_key in current_keys:
|
|
return False
|
|
if _notif_live_session_matches({resolved_key}):
|
|
return True
|
|
if evt_key in current_keys:
|
|
return False
|
|
if resolved_key == evt_key and _notif_other_profile_session_owns(sid, session, evt):
|
|
return True
|
|
return _notif_live_session_matches({evt_key, resolved_key}, exclude=session)
|
|
|
|
|
|
def _notif_other_profile_session_owns(sid: str, session: dict, evt: dict) -> bool:
|
|
"""True when a live session on ANOTHER profile store provably owns ``evt`` (its compression lineage
|
|
resolves there). Every poller drains one process-wide queue, but lineage is looked up in the
|
|
dequeuer's own store; without this, profile B dequeuing an event keyed on profile A's compressed
|
|
parent found no owner anywhere and dropped it for good. Snapshot under the lock, resolve outside it."""
|
|
own_home = str(session.get("profile_home") or "")
|
|
candidates = _notif_locked_sessions(
|
|
lambda ss: [(other_sid, other) for other_sid, other in ss.items()
|
|
if other is not session and not other.get("_finalized")
|
|
and str(other.get("profile_home") or "") != own_home],
|
|
[])
|
|
return any(_session_owns_notification_event(other_sid, other, evt) for other_sid, other in candidates)
|
|
|
|
|
|
def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool:
|
|
"""True iff *this* session PROVABLY owns ``evt`` (UI origin is this live session, or ``session_key`` raw/compression-
|
|
resolved matches) — the fail-closed gate for addressed notifications, without the orphan-adoption fallback."""
|
|
if session.get("_finalized"):
|
|
return False
|
|
# A completion replayed at process start has no in-process owner. A matching
|
|
# persisted key alone can belong to a detached runtime whose old lease survived;
|
|
# requiring a current client prevents it from spending a full model turn unseen.
|
|
# Shared peer membership excludes stdio, even when it is the real TUI RPC channel
|
|
# (possibly wrapped for a dashboard sidecar), rather than a headless fallback sink.
|
|
if (evt.get("type") == "async_delegation" and evt.get("startup_replay")
|
|
and not (_session_has_live_transport(session)
|
|
or (_stdio_is_rpc_channel and session.get("transport") is _stdio_transport))):
|
|
return False
|
|
if str(evt.get("origin_ui_session_id") or "") == str(sid or ""):
|
|
return True
|
|
evt_key = str(evt.get("session_key") or "")
|
|
current_keys = _notif_current_keys(sid, session)
|
|
return bool(evt_key) and (evt_key in current_keys or _notif_resolve_event_key(evt_key, session) in current_keys)
|
|
|
|
|
|
def _notification_event_requires_owner(evt: dict) -> bool:
|
|
"""Whether ``evt`` must be positively claimed before TUI delivery."""
|
|
return evt.get("type") == "async_delegation" or bool(evt.get("origin_ui_session_id") or evt.get("session_key"))
|
|
|
|
|
|
# Extra dedup fields per event type. Completions are terminal (one-shot per process session); watch events are not —
|
|
# one process can match patterns many times, so their content is part of the key.
|
|
_DEDUP_EXTRA_FIELDS = {
|
|
"watch_match": ("command", "pattern", "output", "suppressed", "message_id"),
|
|
"heartbeat": ("seq",),
|
|
"watch_disabled": ("command", "message", "suppressed"),
|
|
"watch_overflow_": ("command", "message", "suppressed"), # prefix match
|
|
}
|
|
|
|
|
|
def _notification_event_dedup_key(evt: dict) -> tuple:
|
|
"""UI-emission identity for a process notification event."""
|
|
evt_type = evt.get("type", "completion")
|
|
if evt_type in {"watch_match", "watch_disabled"} and evt.get("event_id"):
|
|
return (evt.get("session_id", ""), evt_type, evt["event_id"])
|
|
if evt_type == "async_delegation":
|
|
# No process session_id: else every completion keys as ("", "async_delegation") and the second is suppressed forever.
|
|
# An early per-task failure notice must not collapse with the batch's final result (nor with a sibling's notice).
|
|
if evt.get("task_failure_notice"):
|
|
task_idx = ((evt.get("results") or [{}])[0] or {}).get("task_index", "")
|
|
return (evt.get("delegation_id", ""), evt_type, "task_failure", task_idx)
|
|
return (evt.get("delegation_id", ""), evt_type)
|
|
extra = _DEDUP_EXTRA_FIELDS.get("watch_overflow_" if evt_type.startswith("watch_overflow_") else evt_type, ())
|
|
return (evt.get("session_id", ""), evt_type, *(evt.get(f, 0 if f == "suppressed" else "") for f in extra))
|
|
|
|
|
|
# Mirror gateway/kanban_watchers.py TERMINAL_KINDS: claim silent kinds (archived/unblocked) too so the cursor advances
|
|
# past them and they can't wedge a later completed/blocked event behind an unclaimed row.
|
|
_KANBAN_NOTIFY_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked")
|
|
# kanban, /loop + /heartbeat and the bot mailbox share one idle-poll cadence; probing the lease registry on
|
|
# every 0.5s queue timeout cost ~a core at 11 sessions (#108005).
|
|
_KANBAN_POLL_SECONDS = _LOOP_POLL_SECONDS = 5.0
|
|
|
|
|
|
def _notif_release_turn(session: dict) -> None:
|
|
with session["history_lock"]:
|
|
session["running"] = False
|
|
|
|
|
|
def _notif_claim_turn(session: dict) -> bool:
|
|
"""Claim the idle session (running=True) under history_lock; False if a turn is live.
|
|
After the user's Stop no automatic turn starts: the cancel latch holds notifications
|
|
(requeued by the callers) until the next user prompt clears it."""
|
|
with _session_turn_admission(session) as admitted:
|
|
if not admitted or session.get("running") or session.get("_turn_cancel_requested"):
|
|
return False
|
|
session["running"] = True
|
|
return True
|
|
|
|
|
|
def _notif_log_failure(what: str, exc: BaseException) -> None:
|
|
print(f"[tui_gateway] {what}: {type(exc).__name__}: {exc}", file=sys.stderr)
|
|
|
|
|
|
def _notif_unowned_event(sid: str, evt: dict, deferred) -> None:
|
|
"""Dispose of an addressed event this session cannot prove it owns (never adopt an orphan): park it for its
|
|
owner's return (``notification_park``). One the park refuses: a delegation is dropped at WARNING with its offer
|
|
cleared (``return_completion_offer``; the row stays pending) or, during a shutdown drain, goes onto ``deferred``
|
|
to be requeued; anything else drops."""
|
|
evt_type, is_delegation = evt.get("type", "completion"), evt.get("type") == "async_delegation"
|
|
origin, key = str(evt.get("origin_ui_session_id") or ""), str(evt.get("session_key") or "")
|
|
from tui_gateway import notification_park
|
|
if notification_park.park(evt):
|
|
logger.debug("Parked unowned %s notification (origin=%r key=%r) until its owner returns",
|
|
evt_type, origin, key)
|
|
elif deferred is not None:
|
|
if is_delegation:
|
|
deferred.append(evt)
|
|
else:
|
|
logger.debug("Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
|
|
evt_type, origin, key)
|
|
else:
|
|
# A durable replay stays pending: hand it back so the orphan sweep re-offers it once its owner
|
|
# is live (#97202), and keep that retry out of WARNING.
|
|
restored = is_delegation and bool(evt.get("restored"))
|
|
(logger.warning if is_delegation and not restored else logger.debug)(
|
|
"Dropping unowned %s notification (origin=%r key=%r) instead of delivering to session %s",
|
|
evt_type, origin, key, sid)
|
|
if is_delegation:
|
|
from tools.async_delegation import return_completion_offer
|
|
return_completion_offer(evt)
|
|
|
|
|
|
# A retried notice the canonical ledger refuses (SESSION_NOT_OWNED persists while a gateway owns the
|
|
# chat) waits before its requeue, doubling per session up to the kanban/loop poll period so the
|
|
# poller thread never delays those polls by more than one period.
|
|
_NOTIF_REFUSAL_BACKOFF_S = 0.25
|
|
_NOTIF_REFUSAL_BACKOFF_MAX_S = _KANBAN_POLL_SECONDS
|
|
|
|
|
|
def _notif_refusal_backoff(session: dict) -> None:
|
|
"""Wait before requeueing a refused notice: this session's next delay doubles, to a cap."""
|
|
delay = session.get("_notif_refusal_delay") or _NOTIF_REFUSAL_BACKOFF_S
|
|
session["_notif_refusal_delay"] = min(delay * 2, _NOTIF_REFUSAL_BACKOFF_MAX_S)
|
|
time.sleep(delay)
|
|
|
|
|
|
def _notif_submit(rid: str, sid: str, session: dict, text: str, what: str, **kwargs) -> bool:
|
|
"""Only the execution gate can acknowledge a claimed notification turn."""
|
|
try:
|
|
accepted = _run_prompt_submit(rid, sid, session, text, **kwargs)
|
|
if accepted:
|
|
session.pop("_notif_refusal_delay", None)
|
|
else:
|
|
# Admission can refuse even after the local running flag was claimed. A caller that
|
|
# requeues the notice backs off first (_notif_refusal_backoff); the rest return at once.
|
|
_notif_release_turn(session)
|
|
return accepted
|
|
except Exception as exc:
|
|
_notif_log_failure(what, exc)
|
|
_notif_release_turn(session)
|
|
raise
|
|
|
|
|
|
def _notif_loop_status(sid: str, text: str) -> None:
|
|
_emit("status.update", sid, {"kind": "loop", "text": text})
|
|
|
|
|
|
def _notif_slash_loop_tick(rid: str, sid: str, session: dict, mgr, wakeup: str) -> None:
|
|
"""Slash-command /loop wakeup: route through the slash pipeline, not the model. No model reply to evaluate, so the
|
|
tick completes immediately — unless the command resolves to a prompt (skill command etc.), which runs as a normal
|
|
turn whose post-turn hook completes the tick."""
|
|
_notif_release_turn(session)
|
|
try:
|
|
parts = wakeup.lstrip()[1:].split(None, 1)
|
|
resp = _methods["command.dispatch"](
|
|
rid, {"name": parts[0] if parts else "", "arg": parts[1] if len(parts) > 1 else "", "session_id": sid})
|
|
payload = (resp or {}).get("result") or {}
|
|
if out := str(payload.get("output") or "").strip():
|
|
_notif_loop_status(sid, out)
|
|
if payload.get("type") == "send" and payload.get("message"):
|
|
if not _notif_claim_turn(session):
|
|
mgr.abandon_tick()
|
|
_publish_session_control_snapshot(sid, session) # back to the countdown, not "running now"
|
|
return
|
|
# Releases the claim on failure: the swallow below would otherwise leave the session busy for good.
|
|
_notif_submit(rid, sid, session, payload["message"], "loop wakeup send failed")
|
|
return
|
|
except Exception:
|
|
pass
|
|
decision = mgr.complete_tick("")
|
|
if decision.get("message"):
|
|
_notif_loop_status(sid, decision["message"])
|
|
_publish_session_control_snapshot(sid, session) # next_due_at moved: clients re-anchor the countdown
|
|
|
|
|
|
def _notif_gateway_owns_heartbeat(session: dict, session_key: str) -> bool:
|
|
"""Whether the gateway's heartbeat poller owns this session's due tick.
|
|
|
|
Desktop/TUI can attach to a messaging conversation, but its session-owner poller has no adapter
|
|
route for the reply. Ownership is the gateway's LIVE routing index, not the row's immutable
|
|
``source``: ``gateway/run_heartbeat_restore.py`` only registers watches for a non-suspended,
|
|
origin-bearing key whose current ``session_id`` is this one, so a row archived by /reset,
|
|
auto-reset or compression rotation belongs to nobody there and must keep firing here. The index
|
|
lives in the gateway's home store (the launch handle for a multiplexed gateway, the profile's
|
|
own store for a per-profile gateway), so both are consulted; no entry is fail-open.
|
|
"""
|
|
try:
|
|
with _session_db(session) as db:
|
|
for store in {id(d): d for d in (db, _get_db()) if d is not None}.values():
|
|
if (entry := store.gateway_routing_entry_for_session(session_key)) is not None:
|
|
return bool(entry.get("origin")) and not entry.get("suspended")
|
|
return False
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _maybe_fire_tui_heartbeat_tick(sid: str, session: dict) -> None:
|
|
"""Fire a due /heartbeat prompt for an idle TUI/Desktop/dashboard session (#102056, #103044).
|
|
|
|
``/heartbeat`` runs in the slash worker, whose CLI watchdog queues the due prompt into a
|
|
``_pending_input`` no turn loop ever drains — armed-but-dead. State is durable in SessionDB, so
|
|
the session-owner process drives firing exactly like ``_maybe_fire_tui_loop_tick``: claim the
|
|
idle session first (a racing user prompt wins), then re-enter through ``_run_prompt_submit`` as a
|
|
plain user turn. A dispatch that never starts a turn rewinds the persisted fire so the tick stays
|
|
due instead of being silently consumed.
|
|
"""
|
|
try:
|
|
from hermes_cli.heartbeat import HeartbeatManager
|
|
except Exception:
|
|
return
|
|
if not (sid_key := session.get("session_key") or ""):
|
|
return
|
|
mgr = HeartbeatManager(session_id=sid_key)
|
|
if not mgr.is_active() or not mgr.state.is_due() or _notif_gateway_owns_heartbeat(session, sid_key):
|
|
return # not due, or the gateway poller owns the routed conversation — stays due there
|
|
if not _notif_claim_turn(session):
|
|
return # busy — the tick coalesces to the next idle poll
|
|
if not (prompt := mgr.due_prompt()):
|
|
_notif_release_turn(session)
|
|
return
|
|
started = False
|
|
try:
|
|
_emit("status.update", sid, {"kind": "heartbeat", "text": f"♥ heartbeat #{mgr.state.fire_count} firing…"})
|
|
started = bool(_run_prompt_submit(f"__heartbeat__{int(time.time() * 1000)}", sid, session, prompt))
|
|
except Exception as exc:
|
|
_notif_log_failure("heartbeat dispatch failed", exc)
|
|
if not started:
|
|
# _run_prompt_submit releases ``running`` itself when it refuses the turn; make it unconditional.
|
|
_notif_release_turn(session)
|
|
with contextlib.suppress(Exception):
|
|
mgr.abandon_fire()
|
|
|
|
|
|
def _loop_route_is_gateway_chat(state) -> bool:
|
|
"""A /loop set from a messaging chat carries the gateway's ``route`` (platform + chat_id); its wakeup scanner
|
|
(``gateway/run_goals.py::_loop_wakeup_fire_one``) fires those and skips route-less CLI/TUI loops. Mirror it here
|
|
so a Desktop viewer of the same session never consumes the tick and strands the reply off the chat."""
|
|
route = getattr(state, "route", None) or {}
|
|
return bool(route.get("platform") and route.get("chat_id"))
|
|
|
|
|
|
def _session_owns_live_wakeup_schedule(session: dict) -> bool:
|
|
"""True while THIS process's notification poller is the only driver of an active /loop or /heartbeat for
|
|
``session`` — the reapers must then keep the detached session (and with it the poller) alive, or the schedule
|
|
freezes until a client reattaches. Gateway-routed schedules are fired by the gateway, so they never pin the
|
|
session here. Bounded: a loop pauses at its tick budget and a stopped/paused schedule releases the session.
|
|
Fail-open to False: an unreadable store must not make a session unreapable."""
|
|
if not (sid_key := session.get("session_key") or ""):
|
|
return False
|
|
try:
|
|
from hermes_cli.heartbeat import HeartbeatManager
|
|
from hermes_cli.loops import LoopManager
|
|
with _session_profile_runtime_scope(session, hydrate_secrets=False):
|
|
loop = LoopManager(session_id=sid_key)
|
|
if loop.is_active() and not _loop_route_is_gateway_chat(loop.state):
|
|
return True
|
|
heartbeat = HeartbeatManager(session_id=sid_key)
|
|
return heartbeat.is_active() and not _notif_gateway_owns_heartbeat(session, sid_key)
|
|
except Exception:
|
|
logger.debug("wakeup-schedule check failed for %s", sid_key, exc_info=True)
|
|
return False
|
|
|
|
|
|
def _session_owns_pending_process_notification(session: dict) -> bool:
|
|
"""True while a background process this session started will still report back into it (``notify_on_complete``
|
|
or ``watch_patterns``): running, or its notice not yet routed (see ``tools.process_registry_awaiting``). The notice
|
|
reaches the model only through THIS session's notification poller, and closing the session kills the process
|
|
(``agent_close``), so the reapers must keep it alive. Ownership is the set close() kills by — the agent's raw
|
|
spawning task ids, which keep pre-compression ids after the session key rotates. Bounded by delivery: the session
|
|
is released once routing admits the notice's turn (``running`` then protects it), requeues or drops it."""
|
|
if session.get("_turn_cancel_requested"):
|
|
# After Stop no notification turn can start until the user's next prompt (``_notif_claim_turn``), so a
|
|
# pending notice is no reason to keep the session: the user's prompt resumes it like any other.
|
|
return False
|
|
from tools.process_registry import process_registry
|
|
from tools.process_registry_awaiting import awaits_notification
|
|
return awaits_notification(process_registry, _session_process_owners(session))
|
|
|
|
|
|
def _session_process_owners(session: dict) -> set:
|
|
"""The raw spawning ids ``close()`` kills by (pre-compression ids included), plus the current key."""
|
|
agent = session.get("agent")
|
|
return {*getattr(agent, "_process_owner_task_ids", ()), str(session.get("session_key") or "")} - {""}
|
|
|
|
|
|
def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
|
|
"""Fire a due /loop wakeup for an idle TUI/Desktop/dashboard session (per-session poller, coarse cadence). Claims
|
|
the session (running=True) before dispatching so a racing user prompt wins; the post-turn hook completes the tick."""
|
|
try:
|
|
from hermes_cli.loops import LoopManager, goal_blocks_loop_tick
|
|
except Exception:
|
|
return
|
|
if not (sid_key := session.get("session_key") or ""):
|
|
return
|
|
mgr = LoopManager(session_id=sid_key)
|
|
if not mgr.is_due() or goal_blocks_loop_tick(sid_key) or _loop_route_is_gateway_chat(mgr.state):
|
|
return # not due, or the gateway's wakeup scanner owns the routed chat — stays due there
|
|
if not _notif_claim_turn(session):
|
|
return # busy — stays due, next poll retries
|
|
if not (wakeup := mgr.fire_tick()):
|
|
_notif_release_turn(session)
|
|
return
|
|
rid = f"__loop__{int(time.time() * 1000)}"
|
|
_publish_session_control_snapshot(sid, session) # awaiting_response: clients swap the countdown for "running"
|
|
try:
|
|
_notif_loop_status(sid, f"↻ /loop wakeup #{mgr.state.ticks_fired if mgr.state else '?'} firing…")
|
|
if wakeup.lstrip().startswith("/"):
|
|
_notif_slash_loop_tick(rid, sid, session, mgr, wakeup)
|
|
else:
|
|
_emit("message.start", sid)
|
|
_run_prompt_submit(rid, sid, session, wakeup)
|
|
except Exception as exc:
|
|
_notif_log_failure("loop wakeup dispatch failed", exc)
|
|
_notif_release_turn(session)
|
|
with contextlib.suppress(Exception):
|
|
mgr.abandon_tick()
|
|
_publish_session_control_snapshot(sid, session) # back to the countdown, not "running now"
|
|
|
|
|
|
def _kb_first_line(value: Any, limit: int) -> str:
|
|
lines = str(value).strip().splitlines()
|
|
return f"\n{lines[0][:limit]}" if lines else ""
|
|
|
|
|
|
def _kb_completed(task, payload: dict, title: str) -> str:
|
|
handoff = (_kb_first_line(payload["summary"], 200) if payload.get("summary")
|
|
else _kb_first_line(task.result, 160) if getattr(task, "result", None) else "")
|
|
return f" done — {title}{handoff}"
|
|
|
|
|
|
def _kb_timed_out(task, payload: dict, title: str) -> str:
|
|
with contextlib.suppress(TypeError, ValueError):
|
|
return f" timed out (max_runtime={int(payload.get('limit_seconds') or 0)}s); will retry"
|
|
return " timed out (max_runtime=0s); will retry"
|
|
|
|
|
|
# kind -> (glyph, suffix after "Kanban <id>"); silent kinds (archived/unblocked) are absent → None.
|
|
_KANBAN_EVENT_FORMATTERS = {
|
|
"completed": ("✔", _kb_completed),
|
|
"blocked": ("⏸", lambda t, p, title: " blocked" + (f": {str(p.get('reason'))[:160]}" if p.get("reason") else "")),
|
|
"gave_up": ("✖", lambda t, p, title: " gave up after repeated spawn failures"
|
|
+ (f"\n{str(p.get('error'))[:200]}" if p.get("error") else "")),
|
|
"crashed": ("✖", lambda t, p, title: " worker crashed (pid gone); dispatcher will retry"),
|
|
"timed_out": ("⏱", _kb_timed_out),
|
|
"status": ("🔄", lambda t, p, title: f" → {p.get('status') or ''}"),
|
|
}
|
|
|
|
|
|
def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[str]:
|
|
"""Single-line notification text for one kanban event; wording mirrors gateway/kanban_watchers.py (reads the same
|
|
as on Telegram). None for silent kinds."""
|
|
if (entry := _KANBAN_EVENT_FORMATTERS.get(getattr(ev, "kind", ""))) is None:
|
|
return None
|
|
glyph, fmt = entry
|
|
task_id = sub.get("task_id", "")
|
|
title = (getattr(task, "title", None) or task_id)[:120]
|
|
who = getattr(task, "assignee", None) or ""
|
|
prefix = f"{glyph} " + (f"[{board_slug}] " if board_slug else "") + (f"@{who} " if who else "")
|
|
return f"{prefix}Kanban {task_id}{fmt(task, getattr(ev, 'payload', None) or {}, title)}"
|
|
|
|
|
|
def _kb_board_key(_kb, board_meta) -> tuple[str, str]:
|
|
"""(slug, resolved DB identity) — multiple slugs can point at one DB when HERMES_KANBAN_DB pins it."""
|
|
slug = (board_meta or {}).get("slug") or _kb.DEFAULT_BOARD
|
|
db_path = (board_meta or {}).get("db_path")
|
|
try:
|
|
return slug, str(Path(db_path).expanduser().resolve() if db_path else _kb.kanban_db_path(slug).resolve())
|
|
except Exception:
|
|
return slug, f"slug:{slug}"
|
|
|
|
|
|
def _kb_poll_board(_kb, slug: str, session_key: str) -> list:
|
|
"""Claim + format this session's unseen events on one board. One poller per live session: the board is not opened
|
|
writable unless it has a subscription owned by this exact session (a failed read-only probe — locked/corrupt DB —
|
|
falls through so delivery is preserved)."""
|
|
from hermes_cli import kanban_db_connect as _kbc
|
|
from hermes_cli import kanban_db_notify as _kbn
|
|
with contextlib.suppress(Exception):
|
|
if _kbn.count_notify_subs(board=slug, platform="tui", chat_id=session_key) == 0:
|
|
return []
|
|
try:
|
|
conn = _kbc.connect(board=slug)
|
|
except Exception:
|
|
return []
|
|
texts: list = []
|
|
with contextlib.closing(conn):
|
|
try:
|
|
subs = _kbn.list_notify_subs(conn)
|
|
except Exception:
|
|
return []
|
|
for sub in subs:
|
|
if (sub.get("platform") or "").lower() != "tui" or sub.get("chat_id") != session_key:
|
|
continue
|
|
sub_ident = dict(task_id=sub["task_id"], platform=sub["platform"], chat_id=sub["chat_id"],
|
|
thread_id=sub.get("thread_id") or "")
|
|
_old, _new, events = _kbn.claim_unseen_events_for_sub(conn, kinds=_KANBAN_NOTIFY_KINDS, **sub_ident)
|
|
if not events:
|
|
continue
|
|
task = _kb.get_task(conn, sub["task_id"])
|
|
from gateway.kanban_watchers_notifier import diagnostic_event
|
|
from gateway.warning_notifications import DiagnosticText
|
|
for ev in events:
|
|
text = _format_kanban_event_text(sub, task, ev, slug)
|
|
if text:
|
|
texts.append(DiagnosticText(text) if diagnostic_event(ev) else text)
|
|
# Unsubscribe only on archive: ``done`` is reversible in review/controller flows, so keeping the sub lets a
|
|
# later reopen notify the same session. The claimed cursor prevents replay.
|
|
if task and getattr(task, "status", "") == "archived":
|
|
with contextlib.suppress(Exception):
|
|
_kbn.remove_notify_sub(conn, **sub_ident)
|
|
return texts
|
|
|
|
|
|
def _collect_kanban_notifications(session: dict) -> list:
|
|
"""Claim unseen terminal kanban events for this session's ``platform="tui"`` subscriptions (``kanban_create``
|
|
auto-subscribes with ``chat_id=HERMES_SESSION_KEY``; no "tui" messaging adapter exists, so this poller is the
|
|
delivery path). Same atomic cursor-claim as the gateway notifier: exactly-once even if a gateway polls the same DB.
|
|
|
|
See #59890.
|
|
"""
|
|
session_key = str(session.get("session_key") or "")
|
|
if not session_key or session.get("_finalized"):
|
|
return []
|
|
try:
|
|
from hermes_cli import kanban_db as _kb
|
|
except Exception:
|
|
return []
|
|
try:
|
|
boards = _kb.list_boards(include_archived=False)
|
|
except Exception:
|
|
try:
|
|
boards = [_kb.read_board_metadata(_kb.DEFAULT_BOARD)]
|
|
except Exception:
|
|
return []
|
|
# dict keyed by resolved DB identity: first slug per DB wins (a pinned HERMES_KANBAN_DB aliases slugs).
|
|
unique = {}
|
|
for slug, resolved in (_kb_board_key(_kb, board_meta) for board_meta in boards):
|
|
unique.setdefault(resolved, slug)
|
|
return [t for slug in unique.values() for t in _kb_poll_board(_kb, slug, session_key)]
|
|
|
|
|
|
def _notif_poll_kanban(sid: str, session: dict) -> None:
|
|
with _session_profile_runtime_scope(session):
|
|
_notif_poll_kanban_scoped(sid, session)
|
|
|
|
|
|
def _notif_poll_kanban_scoped(sid: str, session: dict) -> None:
|
|
"""One kanban poll: emit new texts, buffer them, and run the buffered batch as a turn if idle. Events are
|
|
cursor-claimed (never re-queued), so they wait in the buffer instead of dropping the agent turn."""
|
|
try:
|
|
texts = _collect_kanban_notifications(session)
|
|
except Exception as exc:
|
|
_notif_log_failure("kanban notification poll failed", exc)
|
|
texts = []
|
|
for text in texts:
|
|
from gateway.warning_notifications import DiagnosticText, render_notification
|
|
render_notification(lambda: _emit("status.update", sid, {"kind": "process", "text": text}),
|
|
platform="tui", diagnostic=isinstance(text, DiagnosticText))
|
|
if texts:
|
|
session.setdefault("_kanban_pending", []).extend(texts)
|
|
if not session.get("_kanban_pending") or not _notif_claim_turn(session):
|
|
return
|
|
with session["history_lock"]:
|
|
pending = session.get("_kanban_pending") or []
|
|
from gateway.warning_notifications import DiagnosticText, warning_notifications_enabled
|
|
split = not warning_notifications_enabled("tui")
|
|
diagnostic = split and isinstance(pending[0], DiagnosticText)
|
|
batch = [text for text in pending if not split or isinstance(text, DiagnosticText) == diagnostic]
|
|
session["_kanban_pending"] = [text for text in pending if split and isinstance(text, DiagnosticText) != diagnostic]
|
|
with contextlib.suppress(Exception):
|
|
_notif_submit(f"__notif__{int(time.time() * 1000)}", sid, session, "\n".join(batch),
|
|
"kanban notification dispatch failed",
|
|
**({"display_metadata": {"notification_category": "diagnostic"}} if diagnostic else {}))
|
|
|
|
|
|
def _notif_defer_event(evt, claim, put=None):
|
|
from tools.async_delegation import defer_completion_delivery
|
|
from tools.process_registry import process_registry
|
|
if claim:
|
|
defer_completion_delivery(evt['delegation_id'], claim)
|
|
(put or process_registry.completion_queue.put)(evt)
|
|
|
|
|
|
def _background_notifications_off(session: dict) -> bool:
|
|
"""Whether the owning profile set ``display.background_process_notifications: off``. Same
|
|
gate the messaging gateway applies to its process-event injection; only ``off`` matters
|
|
here (the other modes shape gateway chat receipts, not agent wakes)."""
|
|
with _session_profile_runtime_scope(session):
|
|
raw = (_load_cfg().get("display") or {}).get("background_process_notifications")
|
|
return raw is False or str(raw or "").strip().lower() == "off"
|
|
|
|
|
|
def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None:
|
|
"""Run the claimed (running=True) agent turn for one notification event."""
|
|
from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery
|
|
try:
|
|
claim = claim_event_delivery(evt, "tui-poller")
|
|
except Exception as exc: # shared ledger busy/unreadable: the durable row stays pending and replays
|
|
_notif_log_failure("notification delivery claim failed", exc)
|
|
claim = None
|
|
if claim is None:
|
|
# Another consumer holds the durable row — a gateway sharing this home claims before it verifies
|
|
# the target. No turn will run, and nothing else clears ``running``: a busy session is exempt
|
|
# from the reaper, keeps its lease, and never reaches its bot mailbox again.
|
|
_notif_release_turn(session)
|
|
return
|
|
evt_type = evt.get("type")
|
|
kwargs: dict = {}
|
|
if evt_type == "async_delegation":
|
|
kwargs = {"display_kind": "async_delegation_complete", "display_metadata": _async_delegation_display_metadata(evt)}
|
|
elif evt_type == "heartbeat":
|
|
# Model-facing scaffolding: the process row on the status stack already says it is running,
|
|
# so the wake never paints as a user bubble (Desktop, TUI and the transcript preview all
|
|
# honour ``hidden``). Only what the agent says about the new output is visible.
|
|
from tools.process_registry_notifications import HEARTBEAT_DISPLAY_KIND
|
|
kwargs = {"display_kind": HEARTBEAT_DISPLAY_KIND}
|
|
from agent.notification_presentation import diagnostic_process_event
|
|
if diagnostic_process_event(evt):
|
|
kwargs.setdefault("display_metadata", {})["notification_category"] = "diagnostic"
|
|
try:
|
|
if not _notif_submit(f"__notif__{int(time.time() * 1000)}", sid, session, text,
|
|
"notification poller dispatch failed", report_ownership_refusal=False, **kwargs):
|
|
_notif_defer_event(evt, claim)
|
|
_notif_refusal_backoff(session) # requeued: queue.get(timeout=...) would never wait
|
|
return
|
|
except Exception:
|
|
release_event_delivery(evt, claim)
|
|
return
|
|
complete_event_delivery(evt, claim)
|
|
|
|
|
|
def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred, completions=None, *, owned=False) -> bool:
|
|
"""Route one dequeued event: foreign (another live session owns it) → requeued, or onto ``deferred`` during the
|
|
shutdown drain; unowned (addressed but unprovable — never adopt an orphan) → parked for a bounded window so the
|
|
owner's resume can still claim it (``_notif_unowned_event``); ours (or ownerless legacy,
|
|
kept process-global) → status.update once, then an agent turn if idle. False = the drain must stop (session
|
|
busy). ``owned`` skips the ownership gates for events a caller already drained through
|
|
``_session_owns_notification_event`` (the post-turn safety net), so lineage resolves once."""
|
|
queue = registry.completion_queue
|
|
evt_type, is_delegation = evt.get("type", "completion"), evt.get("type") == "async_delegation"
|
|
if not owned and _notification_event_belongs_elsewhere(sid, session, evt):
|
|
if deferred is not None:
|
|
deferred.append(evt)
|
|
else: # otherwise a process started in session A surfaces in whichever poller wakes first
|
|
queue.put(evt)
|
|
time.sleep(0.1)
|
|
return True
|
|
if not owned and _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
|
|
_notif_unowned_event(sid, evt, deferred)
|
|
return True
|
|
if evt_type == "completion" and registry.is_completion_consumed(evt.get("session_id", "")):
|
|
return True
|
|
text = fmt(evt)
|
|
if not text:
|
|
return True
|
|
# Emit once per dedup key: a re-queued completion would otherwise re-emit every 0.5s while the session is busy,
|
|
# while distinct watch_match events from one process must stay visible.
|
|
dedup_key = _notification_event_dedup_key(evt)
|
|
if dedup_key not in emitted:
|
|
from tools.process_registry_notifications import async_delegation_display_text, process_completion_display_text
|
|
display_text = (async_delegation_display_text(evt) if is_delegation
|
|
else process_completion_display_text([evt]) if evt_type == "completion" else text)
|
|
from agent.notification_presentation import diagnostic_process_event
|
|
from gateway.warning_notifications import render_notification
|
|
render_notification(lambda: _emit("status.update", sid, {"kind": "process", "text": display_text}),
|
|
platform="tui", diagnostic=diagnostic_process_event(evt))
|
|
emitted.add(dedup_key)
|
|
if evt_type != "async_delegation" and _background_notifications_off(session):
|
|
# The user opted out of process-driven agent wakes: the status row above is the whole
|
|
# delivery. Subagent results are not process notifications and still land.
|
|
return True
|
|
if evt_type == "completion" and completions is not None:
|
|
completions.append((evt, text))
|
|
return True
|
|
if not _notif_claim_turn(session):
|
|
queue.put(evt)
|
|
if deferred is not None:
|
|
return False
|
|
time.sleep(0.25) # back off: the re-queued event keeps the queue non-empty, else this loop spins at 100% CPU
|
|
return True
|
|
_notif_dispatch_event(sid, session, evt, text)
|
|
return True
|
|
|
|
|
|
def _notif_dispatch_completions(sid, session, notifications, registry, deferred):
|
|
from tools.process_registry_notifications import PROCESS_COMPLETE_DISPLAY_KIND, ProcessNotificationBatch
|
|
from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery
|
|
|
|
if not notifications:
|
|
return
|
|
if not _notif_claim_turn(session):
|
|
for event, _text in notifications:
|
|
(deferred.append if deferred is not None else registry.completion_queue.put)(event)
|
|
if deferred is None:
|
|
time.sleep(0.25)
|
|
return
|
|
claimed: list = []
|
|
try:
|
|
for event, event_text in notifications:
|
|
if (claim := claim_event_delivery(event, "tui-completion-batch")) is not None:
|
|
claimed.append((event, event_text, claim))
|
|
batch = ProcessNotificationBatch(tuple((event, event_text) for event, event_text, _claim in claimed))
|
|
text = batch.render(registry)
|
|
except Exception as exc:
|
|
_notif_log_failure("completion batch preparation failed", exc)
|
|
_notif_release_turn(session)
|
|
for event, _text, claim in claimed:
|
|
release_event_delivery(event, claim)
|
|
return
|
|
if text is None:
|
|
_notif_release_turn(session)
|
|
try:
|
|
if text is not None:
|
|
if not _notif_submit(f"__notif__{int(time.time() * 1000)}", sid, session, text,
|
|
"completion batch dispatch failed", display_kind=PROCESS_COMPLETE_DISPLAY_KIND,
|
|
report_ownership_refusal=False,
|
|
display_metadata={"display_text": batch.display_text(registry)}):
|
|
for event, _text, claim in claimed:
|
|
_notif_defer_event(event, claim,
|
|
deferred.append if deferred is not None else registry.completion_queue.put)
|
|
if deferred is None: # requeued onto the live queue (a shutdown drain hands them back)
|
|
_notif_refusal_backoff(session)
|
|
return
|
|
except Exception:
|
|
for event, _text, claim in claimed:
|
|
release_event_delivery(event, claim)
|
|
return
|
|
for event, _text, claim in claimed:
|
|
complete_event_delivery(event, claim)
|
|
|
|
|
|
def _notif_handle_ready(sid, session, events, emitted, registry, fmt, deferred, *, owned=False):
|
|
"""One ready snapshot: ownership and UI emission per event, one turn per completion run. Events this pass
|
|
does not route (busy session, or an event whose routing raised) go back to ``deferred`` / the queue."""
|
|
completions = []
|
|
hand_back = deferred.append if deferred is not None else registry.completion_queue.put
|
|
for index, event in enumerate(events):
|
|
if event.get("type", "completion") != "completion":
|
|
_notif_dispatch_completions(sid, session, completions, registry, deferred)
|
|
completions = []
|
|
try:
|
|
routed = _notif_handle_event(sid, session, event, emitted, registry, fmt, deferred, completions,
|
|
owned=owned)
|
|
except Exception:
|
|
# The snapshot is the only in-memory copy of everything after this event: hand it back, and the
|
|
# failed delegation's offer to the ledger, before the caller logs the failure.
|
|
for remaining in events[index + 1:]:
|
|
hand_back(remaining)
|
|
_notif_dispatch_completions(sid, session, completions, registry, deferred)
|
|
from tools.async_delegation import return_completion_offer
|
|
return_completion_offer(event)
|
|
raise
|
|
if not routed:
|
|
for remaining in events[index + 1:]:
|
|
hand_back(remaining)
|
|
break
|
|
_notif_dispatch_completions(sid, session, completions, registry, deferred)
|
|
|
|
|
|
def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None:
|
|
with _session_profile_runtime_scope(session):
|
|
_notification_poller_scoped_loop(stop_event, sid, session)
|
|
|
|
|
|
def _notification_poller_scoped_loop(stop_event: threading.Event, sid: str, session: dict) -> None:
|
|
"""Daemon thread (started by _init_session()) that drains the process-global completion_queue for this session
|
|
(ownership routing: _notif_handle_event), claims the events parked while their owner was not live
|
|
(``notification_park``), and polls ``kanban_notify_subs`` every ``_KANBAN_POLL_SECONDS`` — the
|
|
delivery path for platform="tui" rows.
|
|
|
|
Also polls ``kanban_notify_subs`` every ``_KANBAN_POLL_SECONDS`` for this session's TUI kanban
|
|
subscriptions and delivers terminal task events the same way (status.update + agent turn) — the delivery
|
|
path tools/kanban_tools.py documents for platform="tui" rows (issue #59890).
|
|
"""
|
|
from tools import async_delegation
|
|
from tools.process_registry import process_registry
|
|
from tools.process_registry_awaiting import release as release_in_flight, take_ready
|
|
from tui_gateway import notification_park
|
|
from tools.process_registry_notifications import format_process_notification
|
|
process_registry.restore_completions() # first consumer in a TUI process (#123265)
|
|
queue = process_registry.completion_queue
|
|
emitted = session.setdefault("_notification_emitted", set())
|
|
handle = lambda events, deferred: _notif_handle_ready(
|
|
sid, session, events, emitted, process_registry, format_process_notification, deferred)
|
|
last_kanban_poll = last_loop_poll = 0.0
|
|
while not stop_event.is_set() and not session.get("_finalized"):
|
|
now = time.monotonic()
|
|
# Completions whose owner process died after this one started (#97202); throttled per profile home.
|
|
async_delegation.maybe_sweep_orphaned_completions(queue)
|
|
# /loop and /heartbeat wakeup drivers: fire a due tick for THIS session while idle (same claim-under-lock
|
|
# as kanban dispatch). An active non-parked /goal owns the idle boundary and defers the loop tick.
|
|
if now - last_loop_poll >= _LOOP_POLL_SECONDS:
|
|
last_loop_poll = now
|
|
for what, fire in (("loop wakeup", _maybe_fire_tui_loop_tick), ("heartbeat", _maybe_fire_tui_heartbeat_tick)):
|
|
try:
|
|
fire(sid, session)
|
|
except Exception as tick_exc:
|
|
_notif_log_failure(f"{what} poll failed", tick_exc)
|
|
if now - last_kanban_poll >= _KANBAN_POLL_SECONDS:
|
|
last_kanban_poll = now
|
|
_notif_poll_kanban(sid, session)
|
|
# Dequeue and in-flight hold are one step: until routing admits a turn, requeues or drops each event, its
|
|
# owner session stays protected from the reapers (it is in neither the queue nor ``running`` meanwhile).
|
|
# Addressed events parked while their owner was not live (notification_park): once this session proves it
|
|
# owns one, it is this session's to deliver, ahead of the shared queue.
|
|
parked = notification_park.claim(lambda evt: _session_owns_notification_event(sid, session, evt))
|
|
ready = []
|
|
try:
|
|
ready = take_ready(queue, 0 if parked else 0.5)
|
|
if not parked and not ready:
|
|
continue
|
|
handle(parked + ready, None)
|
|
except Exception as exc:
|
|
# This thread is the session's only path to notifications, /loop, /heartbeat and its
|
|
# bot mailbox; one bad event must not end all four.
|
|
_notif_log_failure("notification dispatch failed", exc)
|
|
finally:
|
|
release_in_flight(ready)
|
|
notification_park.unhold(parked)
|
|
# Drain remaining events after the stop signal so nothing is lost on shutdown: unowned ones park for their
|
|
# owner's return; foreign ones (and delegations the park refused) go back to the shared queue afterwards, even
|
|
# when routing one of them raised (#140718).
|
|
deferred: list = []
|
|
ready = take_ready(queue, 0)
|
|
try:
|
|
handle(ready, deferred)
|
|
finally:
|
|
for evt in deferred:
|
|
queue.put(evt)
|
|
release_in_flight(ready)
|
|
|
|
|
|
def _async_delegation_display_metadata(evt: dict) -> dict:
|
|
"""Build display-only metadata before the completion event is formatted."""
|
|
from tools.process_registry_notifications import async_delegation_display_text
|
|
raw_results = evt.get("results")
|
|
results: list[dict] = [r for r in raw_results if isinstance(r, dict)] if isinstance(raw_results, list) else []
|
|
task_count = len(results) or 1
|
|
completed_count = sum(1 for r in results if r.get("status") in {"completed", "success"})
|
|
failed_count = sum(1 for r in results if r.get("status") in {"failed", "error"})
|
|
duration = evt.get("total_duration_seconds") or evt.get("duration_seconds")
|
|
return {"display_text": async_delegation_display_text(evt),
|
|
"delegation_id": str(evt.get("delegation_id") or ""), "task_count": task_count,
|
|
"completed_count": completed_count or task_count - failed_count, "failed_count": failed_count,
|
|
**({"duration_seconds": duration} if isinstance(duration, (int, float)) else {})}
|
|
|
|
|
|
_desktop_ui_wired = False
|
|
|
|
|
|
def _wire_desktop_sinks() -> None:
|
|
"""Idempotently wire process-registry and desktop-tool sinks to renderer events: `agent.terminal.output` and
|
|
`terminal.close` (drops a tab without killing the process) route to the window owning the process; desktop-only
|
|
tools pass the turn's ``HERMES_UI_SESSION_ID`` as ``sid``. `_emit` is thread-safe."""
|
|
global _desktop_ui_wired
|
|
from tools.process_registry import process_registry
|
|
|
|
def _owner_sid(session) -> str:
|
|
# session may be None (process already finished/pruned) — the tab can still linger and be closed.
|
|
session_key = str(getattr(session, "session_key", "") or "") if session is not None else ""
|
|
if not session_key:
|
|
return ""
|
|
with _sessions_lock:
|
|
return next((sid for sid, s in _sessions.items() if str(s.get("session_key") or "") == session_key), "")
|
|
if getattr(process_registry, "on_output", None) is None:
|
|
process_registry.on_output = lambda session, chunk: _emit(
|
|
"agent.terminal.output", _owner_sid(session), {"process_id": session.id, "chunk": chunk})
|
|
if getattr(process_registry, "on_close", None) is None:
|
|
process_registry.on_close = lambda session, pid: _emit("terminal.close", _owner_sid(session), {"process_id": pid})
|
|
if not _desktop_ui_wired:
|
|
with contextlib.suppress(Exception):
|
|
from tools import desktop_ui
|
|
desktop_ui.set_emitter(lambda sid, event, payload: _emit(event, sid, payload))
|
|
_desktop_ui_wired = True
|
|
|
|
|
|
# (stop_event, thread) for every poller started in this process, pruned of dead threads on each spawn. Test teardowns
|
|
# reap leaked pollers through it: an unjoined poller steals events off the process-global queue in a LATER test.
|
|
_notification_pollers: list = []
|
|
|
|
|
|
def _start_notification_poller(sid: str, session: dict) -> threading.Event:
|
|
"""Start the background notification poller for a TUI session (thread name is greppable)."""
|
|
_wire_desktop_sinks()
|
|
stop = threading.Event()
|
|
t = threading.Thread(target=_notification_poller_loop, args=(stop, sid, session), daemon=True, name=f"tui-notif-poller-{sid}")
|
|
_notification_pollers[:] = [(s, th) for (s, th) in _notification_pollers if th.is_alive()] + [(stop, t)]
|
|
t.start()
|
|
return stop
|
|
|
|
|
|
def _hud_surface_note(session: dict) -> str:
|
|
"""The per-surface note for this turn ("" for the plain app window): HUD -> the read-the-window-below
|
|
prior; voice-live -> the spoken-delegation contract with the recent transcript."""
|
|
from gateway.session_surface import surface_note
|
|
surface, agent = session.get("client_surface"), session.get("agent")
|
|
direct = getattr(agent, "valid_tool_names", None) or set()
|
|
if surface == "hud":
|
|
from tools.tool_search_catalog import TOOL_CALL_NAME
|
|
if TOOL_CALL_NAME in direct:
|
|
# Tool search defers the desktop tools by default: they stay callable through the
|
|
# tool_call bridge, so the HUD note must count them and name the bridge (main 312f6bc2fd6).
|
|
from agent.prompt_builder import hud_surface_note
|
|
from agent.tool_executor import _tool_search_scoped_names
|
|
return hud_surface_note(direct, _tool_search_scoped_names(agent))
|
|
committed = {"surface": surface, "voice_context": session.get("voice_live_context")}
|
|
return surface_note(committed, direct)
|
|
|
|
|
|
def _prepend_note(run_message: Any, note: str) -> Any:
|
|
"""Prefix a per-turn note onto the MODEL INPUT, leaving the prompt alone: everything the model must know that the
|
|
user did not type (interrupted reply, reactions, surface) arrives this way, so no scaffolding reaches the
|
|
transcript and no sent message is rewritten — the cached prefix survives."""
|
|
if note and isinstance(run_message, str):
|
|
return f"{note}\n\n{run_message}"
|
|
if note and isinstance(run_message, list):
|
|
return [{"type": "text", "text": note}, *run_message]
|
|
return run_message
|
|
|
|
|
|
def register(server) -> None:
|
|
"""Publish this module's helpers + handlers onto ``server``, rebound to its globals."""
|
|
bind_module(globals(), server, skip=("_",))
|