cd3de040ab
Squash of the 54 commits on victor-kyriazakos:feat/user-channel-warning-suppression (PR #112302, head f45c640e55) so the contributor's authorship survives a rebase-merge; the commits interleave with a cron delivery-ledger rework that the salvage removes in follow-up commits, so per-commit cherry-picks were not practical. Adds display.suppress_warning_notifications (global + per-platform, default false): one resolver (gateway/warning_notifications.py), BasePlatformAdapter.emit_warning / emit_media_warning / warning_text, a notification_category classification carried through wakes, queues and persistence, and render/present boundaries for CLI/TUI.
229 lines
12 KiB
Python
229 lines
12 KiB
Python
"""User-facing status / warning / notice plumbing for ``AIAgent``.
|
|
|
|
Safe printing, quiet-mode gating, deduped context-overflow warnings, and the buffered retry
|
|
chatter that is shown only when every retry/fallback is exhausted.
|
|
Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged.
|
|
"""
|
|
import logging
|
|
import sys
|
|
|
|
from agent.session_activity import ActivityProvenance
|
|
|
|
# Same logger name as the origin module so log records / caplog filters are unchanged.
|
|
logger = logging.getLogger("run_agent")
|
|
|
|
|
|
class StatusOutputMixin:
|
|
"""Status/warning/notice emission and retry-chatter buffering (see module docstring)."""
|
|
|
|
def _safe_print(self, *args, diagnostic: bool = False, **kwargs):
|
|
"""Print that swallows broken pipes / closed stdout (headless stdout can vanish mid-session);
|
|
routes through ``self._print_fn`` so the CLI can inject an ANSI-aware renderer."""
|
|
if getattr(self, "_mute_notification_reply", False):
|
|
return
|
|
if diagnostic and not self._warning_presentation_enabled():
|
|
return
|
|
try:
|
|
(self._print_fn or print)(*args, **kwargs)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
def _vprint(self, *args, force: bool = False, diagnostic: bool = False, **kwargs):
|
|
"""Verbose print — suppressed while tokens are streaming (allowed during tool execution) and after
|
|
the main response; ``force=True`` bypasses both. ``suppress_status_output`` (``hermes chat -q``) wins."""
|
|
if getattr(self, "suppress_status_output", False):
|
|
return
|
|
if diagnostic and not self._warning_presentation_enabled():
|
|
return
|
|
if force or not (getattr(self, "_mute_post_response", False) or (self._has_stream_consumers() and not self._executing_tools)):
|
|
self._safe_print(*args, **kwargs)
|
|
|
|
def _should_start_quiet_spinner(self) -> bool:
|
|
"""True when quiet-mode spinner output has a safe sink (``_print_fn`` or a real TTY); a raw spinner
|
|
on a non-TTY stdout can corrupt protocol streams (ACP JSON-RPC)."""
|
|
if self._print_fn is not None:
|
|
return True
|
|
try:
|
|
return bool(sys.stdout.isatty())
|
|
except (AttributeError, ValueError, OSError):
|
|
return False
|
|
|
|
def _should_emit_quiet_tool_messages(self) -> bool:
|
|
"""True when quiet-mode tool summaries should print directly (CLI, no callback owns rendering);
|
|
``suppress_status_output`` always wins so ``[tool]``/``[done]`` never land in captured stdout.
|
|
|
|
``suppress_status_output`` (the strict machine-readable mode used by ``hermes chat -Q``) always
|
|
wins: those flows neutralize the rendering callbacks, and without this gate the "no callback owns
|
|
rendering" fallback would print ``[tool]``/``[done]`` spinner lines into the captured stdout it
|
|
exists to keep clean (#93220).
|
|
"""
|
|
if getattr(self, "suppress_status_output", False):
|
|
return False
|
|
return self.quiet_mode and not self.tool_progress_callback and getattr(self, "platform", "") == "cli"
|
|
|
|
def _call_callback(self, name: str, *args, origin: str) -> None:
|
|
"""Invoke ``self.<name>(*args)`` if set, swallowing errors — a driver callback must never break the loop."""
|
|
cb = getattr(self, name, None)
|
|
if cb:
|
|
try:
|
|
cb(*args)
|
|
except Exception:
|
|
logger.debug("%s error in %s", name, origin, exc_info=True)
|
|
|
|
def _emit_status_kind(self, kind: str, message: str, *, origin: str) -> None:
|
|
"""Print to the CLI (``_vprint(force=True)``) and forward to ``status_callback(kind, message)``. Never raises."""
|
|
from gateway.warning_notifications import is_warning_status
|
|
try:
|
|
if not is_warning_status(kind, message) or self._warning_presentation_enabled():
|
|
self._vprint(f"{self.log_prefix}{message}", force=True)
|
|
except Exception:
|
|
pass
|
|
self._call_callback("status_callback", kind, message, origin=origin)
|
|
|
|
def _warning_presentation_enabled(self) -> bool:
|
|
from gateway.warning_notifications import warning_notifications_enabled
|
|
try:
|
|
return warning_notifications_enabled(
|
|
getattr(self, "_notification_platform", getattr(self, "platform", "cli")),
|
|
getattr(self, "_notification_config", None),
|
|
)
|
|
except Exception:
|
|
# A presentation preference must never turn a recoverable notice into a failed turn.
|
|
return True
|
|
|
|
def _emit_diagnostic_status(self, message: str) -> None:
|
|
"""A diagnostic on the lifecycle rail, without changing legacy formatting."""
|
|
from gateway.warning_notifications import DiagnosticText
|
|
self._emit_status(DiagnosticText(message))
|
|
|
|
def _emit_status(self, message: str) -> None:
|
|
"""Emit a lifecycle status message (CLI + gateway ``status_callback``)."""
|
|
self._emit_status_kind("lifecycle", message, origin="_emit_status")
|
|
|
|
def _emit_warning(self, message: str) -> None:
|
|
"""Emit a user-visible warning for degraded side paths where the turn continues but the user must know."""
|
|
self._emit_status_kind("warn", message, origin="_emit_warning")
|
|
|
|
def _warn_context_overflow_blocked(self, reason: str, preflight_tokens: int, threshold_tokens: int) -> None:
|
|
"""Warn (deduped on the block *kind* — ``cooldown`` / ``ineffective`` — not the countdown string;
|
|
cleared by ``_clear_context_overflow_warn``) when context is over the threshold but compression is blocked."""
|
|
_warn_kind = (reason or "unknown").split(":", 1)[0]
|
|
_warn_key = ("ctx_overflow_blocked", _warn_kind)
|
|
if getattr(self, "_last_ctx_overflow_warn", None) == _warn_key:
|
|
return
|
|
self._last_ctx_overflow_warn = _warn_key
|
|
from agent.conversation_compression import CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE
|
|
|
|
# cooldown + anti-thrash (ineffective) are both "compression blocked".
|
|
if _warn_kind in ("cooldown", "ineffective"):
|
|
self._touch_activity(f"compression blocked ({reason})", provenance=ActivityProvenance.AGENT_COMPRESSION_COOLDOWN)
|
|
self._emit_warning(CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE.format(
|
|
tokens=preflight_tokens, threshold=threshold_tokens, reason=reason,
|
|
))
|
|
|
|
def _warn_uncompressed_context_overflow(self, preflight_tokens: int, context_length: int) -> None:
|
|
"""Deduped warning when uncompressed context exceeds the model limit; points the user at /compact.
|
|
|
|
When compression is explicitly disabled (compression.enabled: false), long sessions can grow past
|
|
the model context window with no compression to shrink them (#89297). Surface an actionable warning
|
|
so the user knows to run /compact or enable compression.
|
|
"""
|
|
_warn_key = ("uncompressed_ctx_overflow", context_length)
|
|
if getattr(self, "_last_ctx_overflow_warn", None) != _warn_key:
|
|
self._last_ctx_overflow_warn = _warn_key
|
|
self._emit_warning(
|
|
f"⚠️ Session context (~{preflight_tokens:,} tokens) exceeds the model "
|
|
f"context window (~{context_length:,} tokens) with compression disabled "
|
|
f"(compression.enabled: false). Use /compact to compress history or "
|
|
f"enable compression in config.yaml."
|
|
)
|
|
|
|
def _clear_context_overflow_warn(self) -> None:
|
|
"""Reset the blocked-overflow warning dedup so it can re-fire on the next blocked turn."""
|
|
self._last_ctx_overflow_warn = None
|
|
|
|
def _emit_notice(self, notice) -> None:
|
|
"""Fire a structured ``AgentNotice`` to the active driver (TUI / CLI)."""
|
|
self._call_callback("notice_callback", notice, origin="_emit_notice")
|
|
|
|
def _emit_notice_clear(self, key: str) -> None:
|
|
"""Clear a previously-fired sticky notice by ``key`` (e.g. on recovery)."""
|
|
self._call_callback("notice_clear_callback", key, origin="_emit_notice_clear")
|
|
|
|
def _emit_wait_notice(self, text: str) -> None:
|
|
"""Rewrite the live status line (CLI spinner, TUI ``thinking.delta``, gateway activity)
|
|
so long provider waits are not an anonymous spinner."""
|
|
self._touch_activity(text)
|
|
self._call_callback("thinking_callback", text, origin="_emit_wait_notice")
|
|
|
|
def _emit_diagnostic_wait(self, text: str) -> None:
|
|
from gateway.warning_notifications import DiagnosticText
|
|
self._emit_wait_notice(DiagnosticText(text))
|
|
|
|
# ── Buffered retry/fallback status: shown only when every retry/fallback is exhausted, dropped on
|
|
# success. Backend logs are unaffected (every site still logs). ──
|
|
|
|
def _buffer_retry_message(self, kind: str, message: str) -> None:
|
|
"""Buffer a retry/fallback line as ``(kind, text)`` until we know whether the turn recovered.
|
|
|
|
``kind`` is ``"status"`` (replays via ``_emit_status``), ``"vprint"`` (``_vprint(force=True)``) or
|
|
``"warn"`` (``_emit_warning``).
|
|
"""
|
|
buf = getattr(self, "_retry_status_buffer", None)
|
|
if buf is None:
|
|
buf = self._retry_status_buffer = []
|
|
buf.append((kind, message))
|
|
|
|
def _buffer_status(self, message: str) -> None:
|
|
self._buffer_retry_message("status", message)
|
|
|
|
def _buffer_diagnostic_status(self, message: str) -> None:
|
|
from gateway.warning_notifications import DiagnosticText
|
|
self._buffer_status(DiagnosticText(message))
|
|
|
|
def _buffer_vprint(self, message: str) -> None:
|
|
self._buffer_retry_message("vprint", message)
|
|
|
|
def _clear_status_buffer(self) -> None:
|
|
"""Drop buffered retry messages — call on successful recovery."""
|
|
buf = getattr(self, "_retry_status_buffer", None)
|
|
if buf:
|
|
buf.clear()
|
|
|
|
def _emit_pending_fallback_notice(self) -> None:
|
|
"""Surface the one-shot fallback-switch notice on successful recovery: a provider switch is durable
|
|
state operators must see, unlike the retry chatter ``_clear_status_buffer`` drops. Emitted once, then
|
|
cleared; on terminal failure the buffered switch line is flushed instead (``_flush_status_buffer``)."""
|
|
notice = getattr(self, "_pending_fallback_notice", None)
|
|
if not notice:
|
|
return
|
|
# Clear before emitting so a (swallowed) callback error can't leave a stale re-emit.
|
|
self._pending_fallback_notice = None
|
|
for item in notice if isinstance(notice, list) else [notice]:
|
|
try:
|
|
self._emit_diagnostic_status(item)
|
|
except Exception:
|
|
# One surface failure must not hide later switches from the same chain.
|
|
continue
|
|
|
|
def _flush_status_buffer(self) -> None:
|
|
"""Emit buffered retry messages — call on terminal failure so the user sees what was tried."""
|
|
# The buffered trace already carries the switch line; drop the one-shot notice.
|
|
self._pending_fallback_notice = None
|
|
buf = getattr(self, "_retry_status_buffer", None)
|
|
if not buf:
|
|
return
|
|
# Drain first so a callback exception doesn't double-emit.
|
|
messages = list(buf)
|
|
buf.clear()
|
|
replay = {"status": self._emit_status, "warn": self._emit_warning}
|
|
for kind, msg in messages:
|
|
try:
|
|
if kind in replay:
|
|
replay[kind](msg)
|
|
else:
|
|
if self._warning_presentation_enabled():
|
|
self._vprint(f"{self.log_prefix}{msg}", force=True)
|
|
except Exception:
|
|
pass
|