8cb4fdc925
A `terminal(background=true, heartbeat=N)` tick queued a notification every N seconds
whether or not the process had printed anything, and every queued event costs the owning
session a full model turn. On Desktop and the TUI that turn painted the wake as a user
bubble ("[Background process ... heartbeat #9 ... (no new output since the last
heartbeat)]") followed by the model's "Still running normally." — over and over, for a
process whose row on the status stack already said it was running — and while the wake
held the session's turn, the user's own prompt sat queued behind it.
- `ProcessRegistry._emit_heartbeat` skips a tick with no new output. The sequence counts
delivered beats only; the "(no new output)" placeholder in the formatter is gone.
- TUI/Desktop type heartbeat rows `display_kind: hidden` (the kind both clients and the
transcript preview already honour); the CLI paints a one-line receipt and persists the
row hidden, so reopening the session in Desktop shows only the agent's reply.
- Desktop hydration drops heartbeat rows persisted by older backends the same way.
- `display.background_process_notifications: off` is honored by the TUI/Desktop poller and
the CLI drain, not just the messaging gateway. `off` mutes process-driven wakes only:
a finished `delegate_task(background=true)` still lands.
Supersedes #123123 (cherry-picked; scoped so `off` keeps subagent results) and #119202
(cherry-picked; `heartbeat: 0` is schema-valid so models that materialize every field
stop tripping the foreground guard).
863 lines
44 KiB
Python
863 lines
44 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
|
||
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 == "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 = _BOT_DELIVERY_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."""
|
||
with _session_turn_admission(session) as admitted:
|
||
if not admitted or session.get("running"):
|
||
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_submit(rid: str, sid: str, session: dict, text: str, what: str, **kwargs) -> None:
|
||
"""message.start + _run_prompt_submit for a claimed (running=True) turn; releases on failure."""
|
||
try:
|
||
from gateway.warning_notifications import render_notification
|
||
with _session_profile_runtime_scope(session):
|
||
render_notification(lambda: _emit("message.start", sid), platform="tui",
|
||
diagnostic=(kwargs.get("display_metadata") or {}).get("notification_category") == "diagnostic")
|
||
_run_prompt_submit(rid, sid, session, text, **kwargs)
|
||
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()
|
||
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"])
|
||
|
||
|
||
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 _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)}"
|
||
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()
|
||
|
||
|
||
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 _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:
|
||
_notif_submit(f"__notif__{int(time.time() * 1000)}", sid, session, text, "notification poller dispatch failed", **kwargs)
|
||
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) → dropped, except delegation payloads
|
||
deferred for a resume; 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):
|
||
origin, key = str(evt.get("origin_ui_session_id") or ""), str(evt.get("session_key") or "")
|
||
if deferred is None:
|
||
# 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)
|
||
elif is_delegation:
|
||
deferred.append(evt)
|
||
else:
|
||
logger.debug("Dropping unowned %s notification during shutdown drain (origin=%r key=%r)", evt_type, origin, key)
|
||
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:
|
||
_notif_submit(f"__notif__{int(time.time() * 1000)}", sid, session, text,
|
||
"completion batch dispatch failed", display_kind=PROCESS_COMPLETE_DISPLAY_KIND,
|
||
display_metadata={"display_text": batch.display_text(registry)})
|
||
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."""
|
||
completions = []
|
||
for index, event in enumerate(events):
|
||
if event.get("type", "completion") != "completion":
|
||
_notif_dispatch_completions(sid, session, completions, registry, deferred)
|
||
completions = []
|
||
if not _notif_handle_event(sid, session, event, emitted, registry, fmt, deferred, completions, owned=owned):
|
||
for remaining in events[index + 1:]:
|
||
(deferred.append if deferred is not None else registry.completion_queue.put)(remaining)
|
||
break
|
||
_notif_dispatch_completions(sid, session, completions, registry, deferred)
|
||
|
||
|
||
def _poll_bot_live_delivery_once(sid: str, session: dict) -> bool:
|
||
"""Run one durable envelope only after local FIFO/continuations yield the idle boundary."""
|
||
from tools.bot_live_delivery import claim_pending_delivery, complete_delivery, find_canonical_live_owner, has_mailbox
|
||
|
||
home = _session_home(session)
|
||
# Most profiles never receive a delivery: without a mailbox there is nothing to claim, and the owner
|
||
# lookup below costs a state.db open plus the exclusive active-session registry lock every pass (#111719).
|
||
if not has_mailbox(home):
|
||
return False
|
||
with _session_turn_admission(session) as admitted:
|
||
if not admitted or any(session.get(key) for key in (
|
||
"running", "_closing", "_finalized", "queued_prompt", "queued_prompts",
|
||
"_auto_continue_scheduled")) or session.get("agent") is None:
|
||
return False
|
||
lease = session.get("active_session_lease")
|
||
if lease is None or getattr(lease, "released", False):
|
||
return False
|
||
owner = find_canonical_live_owner(home)
|
||
if (not owner or owner.get("lease_id") != lease.lease_id
|
||
or owner.get("live_session_id") != sid
|
||
or owner.get("session_id") != session.get("session_key")):
|
||
return False
|
||
# The mailbox matches each envelope to this pinned lease/live id and compression lineage.
|
||
claimed = claim_pending_delivery(home, owner)
|
||
if claimed is None:
|
||
return False
|
||
session["running"] = True
|
||
|
||
delivery_id = str(claimed["id"])
|
||
|
||
def terminal_receipt(terminal: dict) -> None:
|
||
status = str(terminal.get("status") or "failed")
|
||
error = str(terminal.get("error") or "")
|
||
reason = "cancelled" if status == "cancelled" else ""
|
||
if status not in {"settled", "cancelled"}:
|
||
from tools.bot_failure_reasons import classify_agent_error
|
||
reason = classify_agent_error(error)
|
||
# Let a failed write propagate: the turn must not retire its crash marker without its receipt.
|
||
complete_delivery(home, delivery_id, status=status,
|
||
reply=str(terminal.get("text") or "") if status == "settled" else "",
|
||
error=error, reason=reason)
|
||
|
||
try:
|
||
started = _run_prompt_submit(f"__bot_dm__{delivery_id}", sid, session, claimed["message"],
|
||
image_paths=[], terminal_callback=terminal_receipt,
|
||
turn_author=claimed.get("author") or None,
|
||
**({"display_metadata": {"notification_category": "diagnostic"}}
|
||
if claimed.get("notification_category") == "diagnostic" else {}))
|
||
except Exception as exc:
|
||
_notif_release_turn(session)
|
||
terminal_receipt({"status": "failed", "error": str(exc)})
|
||
raise
|
||
if not started:
|
||
_notif_release_turn(session)
|
||
terminal_receipt({"status": "failed", "error": "live session owner could not start the delivery turn"})
|
||
return started
|
||
|
||
|
||
# A failing mailbox poll (typically the active-session registry lock unavailable under contention) is
|
||
# retried on the next ``_BOT_DELIVERY_POLL_SECONDS`` pass; log the failure once per window, not per attempt.
|
||
_BOT_POLL_WARN_INTERVAL_S = 60.0
|
||
|
||
|
||
def _poll_bot_live_delivery_guarded(sid: str, session: dict, now: float) -> None:
|
||
"""One poller-loop pass of the mailbox poll. A failure is logged at WARNING once per
|
||
``_BOT_POLL_WARN_INTERVAL_S`` (with the count of suppressed repeats) and at DEBUG otherwise. An
|
||
unthrottled poll logged ``Bot live-owner delivery poll failed`` ~2×/minute per session for days,
|
||
91% of an install's WARNING output (#111719)."""
|
||
try:
|
||
_poll_bot_live_delivery_once(sid, session)
|
||
except Exception:
|
||
suppressed = int(session.get("_bot_poll_warn_suppressed", 0))
|
||
if now - session.get("_bot_poll_warned_at", -_BOT_POLL_WARN_INTERVAL_S) < _BOT_POLL_WARN_INTERVAL_S:
|
||
session["_bot_poll_warn_suppressed"] = suppressed + 1
|
||
logger.debug("Bot live-owner delivery poll failed (repeat)", exc_info=True)
|
||
return
|
||
session["_bot_poll_warned_at"], session["_bot_poll_warn_suppressed"] = now, 0
|
||
logger.warning("Bot live-owner delivery poll failed (%d repeat(s) suppressed since the last report)",
|
||
suppressed, exc_info=True)
|
||
return
|
||
session["_bot_poll_warn_suppressed"] = 0
|
||
|
||
|
||
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) 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_notifications import format_process_notification
|
||
queue = process_registry.completion_queue
|
||
emitted = session.setdefault("_notification_emitted", set())
|
||
handle = lambda events, deferred: _notif_handle_ready( # noqa: E731
|
||
sid, session, events, emitted, process_registry, format_process_notification, deferred)
|
||
last_kanban_poll = last_loop_poll = last_bot_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)
|
||
if now - last_bot_poll >= _BOT_DELIVERY_POLL_SECONDS: # bot DM → live-owner delivery latency ≤ 5 s
|
||
last_bot_poll = now
|
||
_poll_bot_live_delivery_guarded(sid, session, now)
|
||
# /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)
|
||
try:
|
||
evt = queue.get(timeout=0.5)
|
||
except Exception:
|
||
continue
|
||
ready = [evt]
|
||
for _ in range(queue.qsize()):
|
||
try:
|
||
ready.append(queue.get_nowait())
|
||
except Exception:
|
||
break
|
||
try:
|
||
handle(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)
|
||
# Drain remaining events after the stop signal so nothing is lost on shutdown; foreign and orphaned-delegation
|
||
# events are handed back to the shared queue afterwards.
|
||
deferred: list = []
|
||
ready = []
|
||
for _ in range(queue.qsize()):
|
||
try:
|
||
ready.append(queue.get_nowait())
|
||
except Exception:
|
||
break
|
||
handle(ready, deferred)
|
||
for evt in deferred:
|
||
queue.put(evt)
|
||
|
||
|
||
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 (transcript in, speakable prose out)."""
|
||
surface = session.get("client_surface")
|
||
if surface == "hud":
|
||
from agent.prompt_builder import hud_surface_note
|
||
return hud_surface_note(getattr(session.get("agent"), "valid_tool_names", None))
|
||
if surface == "voice-live":
|
||
from tools.voice_live import voice_live_turn_note
|
||
return voice_live_turn_note(session.get("voice_live_context") or "")
|
||
return ""
|
||
|
||
|
||
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=("_",))
|