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).
89 lines
4.8 KiB
Python
89 lines
4.8 KiB
Python
"""CLI notification ownership, structured queueing and last-moment consumption."""
|
|
|
|
|
|
class CLIProcessNotificationsMixin:
|
|
def _owns_process_notification(self, event: dict) -> bool:
|
|
"""Whether this session owns a delegation event (pre-compression keys resolve to their continuation; fail closed)."""
|
|
event_key = str(event.get("session_key") or "")
|
|
current_key = str(getattr(self, "session_id", "") or "")
|
|
if not event_key or not current_key:
|
|
return False
|
|
if event_key == current_key:
|
|
return True
|
|
try:
|
|
session_db = getattr(self, "_session_db", None)
|
|
resolved_key = (
|
|
session_db.resolve_resume_session_id(event_key) if session_db is not None else event_key
|
|
) or event_key
|
|
except Exception:
|
|
resolved_key = event_key
|
|
return str(resolved_key) == current_key
|
|
|
|
def _background_notifications_suppressed(self) -> bool:
|
|
"""Whether ``display.background_process_notifications`` is ``off`` for this CLI session.
|
|
|
|
The key gates the gateway's completion injection (#9290) but the CLI drain never consulted
|
|
it, so the documented ``off`` escape hatch silently did nothing here (#123114). Mirrors the
|
|
gateway semantics: events are still drained, claimed and acknowledged — only the
|
|
turn-starting injection is suppressed."""
|
|
try:
|
|
from cli import CLI_CONFIG
|
|
mode = str((CLI_CONFIG.get("display") or {}).get("background_process_notifications") or "").strip().lower()
|
|
except Exception:
|
|
return False
|
|
return mode == "off"
|
|
|
|
def _drain_process_notifications(self, consumer: str) -> None:
|
|
from tools.process_registry import process_registry
|
|
from tools.async_delegation import claim_event_delivery, complete_event_delivery
|
|
from tools.process_registry_notifications import (
|
|
HEARTBEAT_DISPLAY_KIND, ProcessNotificationBatch, TimelineNotification, group_process_notifications,
|
|
heartbeat_display_text)
|
|
|
|
claimed = []
|
|
for event, text in process_registry.drain_notifications(
|
|
session_key=getattr(self, "session_id", "") or "", owns_event=self._owns_process_notification,
|
|
):
|
|
claim = claim_event_delivery(event, consumer)
|
|
if claim is None:
|
|
continue
|
|
claimed.append((event, text))
|
|
complete_event_delivery(event, claim)
|
|
if self._background_notifications_suppressed():
|
|
# Subagent results are not process notifications: they still land.
|
|
claimed = [(event, text) for event, text in claimed if event.get("type") == "async_delegation"]
|
|
for notifications in group_process_notifications(claimed):
|
|
event, text = notifications[0]
|
|
evt_type = event.get("type", "completion")
|
|
if evt_type == "completion":
|
|
pending = ProcessNotificationBatch(notifications)
|
|
elif evt_type == "heartbeat":
|
|
pending = TimelineNotification(text, heartbeat_display_text(event), HEARTBEAT_DISPLAY_KIND)
|
|
else:
|
|
pending = TimelineNotification.for_delegation(text, event) if evt_type == "async_delegation" else text
|
|
from agent.notification_presentation import diagnostic_process_event
|
|
if diagnostic_process_event(event) and not isinstance(pending, TimelineNotification):
|
|
pending = TimelineNotification(text, text, "internal_notification", "diagnostic")
|
|
self._pending_input.put(pending)
|
|
|
|
def _tui_unwrap_input(self, user_input):
|
|
"""Unwrap ``_VoiceInputMessage`` / ``_SeededQueryMessage`` -> ``(text_or_tuple, is_voice_input, is_seeded_query)``."""
|
|
from cli import _VoiceInputMessage, _SeededQueryMessage
|
|
from tools.process_registry import process_registry
|
|
from tools.process_registry_notifications import (
|
|
PROCESS_COMPLETE_DISPLAY_KIND, ProcessNotificationBatch, TimelineNotification)
|
|
if isinstance(user_input, ProcessNotificationBatch):
|
|
rendered = user_input.render(process_registry)
|
|
user_input = rendered and TimelineNotification(
|
|
rendered, user_input.display_text(process_registry), PROCESS_COMPLETE_DISPLAY_KIND)
|
|
# Voice-transcribed messages arrive wrapped in a sentinel so only genuine STT output gets the voice
|
|
# prefix (#65827).
|
|
is_voice_input = isinstance(user_input, _VoiceInputMessage)
|
|
if is_voice_input:
|
|
user_input = user_input.text
|
|
is_seeded_query = isinstance(user_input, _SeededQueryMessage)
|
|
if is_seeded_query:
|
|
user_input = (user_input.text, user_input.images) if user_input.images else user_input.text
|
|
return user_input, is_voice_input, is_seeded_query
|
|
|