diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 3b76ba4a8a282..672065f017f59 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1336,6 +1336,13 @@ display: # Toggle at runtime with /verbose in the CLI tool_progress: all + # How gateway progress is grouped on platforms that support message editing. + # accumulate: Edit one bubble, starting continuations at the message limit (default) + # rolling: Keep one bounded turn-activity bubble, then send the final separately + # separate: Send one message per tool (pre-v0.9 style; noisier) + # Per-platform override: display.platforms..tool_progress_grouping + tool_progress_grouping: accumulate + # Per-platform defaults can be quieter than the global setting. Telegram # tunes for mobile: tool_progress and busy_ack_detail default off (no # per-tool breadcrumb stream, no "iteration 21/60" debug detail in busy diff --git a/gateway/display_config.py b/gateway/display_config.py index e58e6e82b22d4..b2d0672d3bc07 100644 --- a/gateway/display_config.py +++ b/gateway/display_config.py @@ -32,7 +32,7 @@ from typing import Any _GLOBAL_DEFAULTS: dict[str, Any] = { "tool_progress": "all", - "tool_progress_grouping": "accumulate", # "accumulate" = edit one bubble; "separate" = one msg per tool + "tool_progress_grouping": "accumulate", # accumulate | rolling | separate "show_reasoning": False, # How a reasoning/thinking summary is rendered when show_reasoning is on. # "code" -> 💭 **Reasoning:** + fenced code block (legacy default) @@ -298,8 +298,8 @@ def _normalise(setting: str, value: Any) -> Any: return "off" return val if val in {"full", "verb", "off"} else "full" if setting == "tool_progress_grouping": - val = str(value).lower() - return val if val in ("accumulate", "separate") else "accumulate" + val = str(value).strip().lower() + return val if val in ("accumulate", "rolling", "separate") else "accumulate" if setting == "reasoning_style": val = str(value).lower() return val if val in ("code", "blockquote", "subtext") else "code" diff --git a/gateway/run.py b/gateway/run.py index 40be1f3af9658..ffdd9a1628afd 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -752,6 +752,29 @@ def _prepare_gateway_status_message(platform: Any, event_type: str, message: str return text +def _rolling_activity_terminal_header( + result: Optional[Dict[str, Any]], + *, + timed_out: bool = False, + cancelled: bool = False, +) -> str: + """Choose the final state shown by a rolling activity message.""" + result = result or {} + if timed_out: + return "⚠️ Needs attention" + if cancelled or result.get("interrupted"): + return "⏹️ Stopped" + if ( + not result + or result.get("failed") + or result.get("partial") + or result.get("completed") is False + or bool(result.get("error")) + ): + return "⚠️ Needs attention" + return "✅ Completed" + + def render_notice_line(notice) -> str: """Render an AgentNotice to a single plaintext line for messaging platforms. @@ -3968,9 +3991,12 @@ class TurnRunner: break return - progress_lines = [] # Accumulated tool lines for the CURRENT editable bubble + progress_lines = [] # Complete tool entries in the CURRENT editable bubble progress_msg_id = None # ID of the current progress message to edit can_edit = ctx.progress_grouping != "separate" # "separate" = one message per tool (pre-v0.9 behavior) + rolling_omitted_count = 0 + rolling_header = "⏳ Working…" if ctx.progress_grouping == "rolling" else None + rolling_state_changed = False _last_edit_ts = 0.0 # Throttle edits to avoid Telegram flood control _PROGRESS_EDIT_INTERVAL = 1.5 # Minimum seconds between edits @@ -4030,7 +4056,28 @@ class TurnRunner: return await adapter.edit_message(**kwargs) def _progress_text(lines: list) -> str: - return "\n".join(str(line) for line in lines) + rendered = [str(line) for line in lines] + if rolling_header: + rendered.insert(0, rolling_header) + if rolling_header and rolling_omitted_count: + rendered.insert( + 1, + f"… {rolling_omitted_count} earlier activities omitted", + ) + return "\n".join(rendered) + + def _fit_rolling_progress_tail() -> None: + """Drop whole oldest entries until the rolling bubble fits.""" + nonlocal rolling_omitted_count + if ctx.progress_grouping != "rolling": + return + while ( + progress_lines + and _progress_len_fn(_progress_text(progress_lines)) + > _PROGRESS_TEXT_LIMIT + ): + progress_lines.pop(0) + rolling_omitted_count += 1 def _split_progress_groups(lines: list) -> list[list]: """Partition progress lines into platform-sized editable bubbles.""" @@ -4076,6 +4123,9 @@ class TurnRunner: nonlocal progress_msg_id, progress_lines, can_edit if not progress_lines or not can_edit: return False + if ctx.progress_grouping == "rolling": + _fit_rolling_progress_tail() + return False groups = _split_progress_groups(progress_lines) if len(groups) <= 1: return False @@ -4140,10 +4190,15 @@ class TurnRunner: # Handle dedup messages: update last line with repeat counter if isinstance(raw, tuple) and len(raw) == 3 and raw[0] == "__dedup__": _, base_msg, count = raw - if progress_lines: + if progress_lines and ( + progress_lines[-1] == base_msg + or str(progress_lines[-1]).startswith(f"{base_msg} (×") + ): progress_lines[-1] = f"{base_msg} (×{count + 1})" msg = progress_lines[-1] if progress_lines else base_msg elif isinstance(raw, tuple) and len(raw) >= 1 and raw[0] == "__reset__": + if ctx.progress_grouping == "rolling": + continue # Content bubble just landed on the platform — close off # the current tool-progress bubble so the next tool # starts a fresh bubble below the content. Without this, @@ -4154,9 +4209,16 @@ class TurnRunner: # linearization regression after PR #7885.) progress_msg_id = None progress_lines = [] + rolling_omitted_count = 0 ctx.last_progress_msg[0] = None ctx.repeat_count[0] = 0 continue + elif isinstance(raw, tuple) and len(raw) >= 1 and raw[0] == "__activity_start__": + msg = rolling_header or "⏳ Working…" + elif isinstance(raw, tuple) and len(raw) >= 2 and raw[0] == "__activity_state__": + rolling_header = str(raw[1]) + rolling_state_changed = True + msg = rolling_header else: msg = raw progress_lines.append(msg) @@ -4186,7 +4248,7 @@ class TurnRunner: if can_edit and progress_msg_id is not None: # Try to edit the existing progress message - full_text = "\n".join(progress_lines) + full_text = _progress_text(progress_lines) result = await _edit_progress_message(progress_msg_id, full_text) if not result.success: _err = (getattr(result, "error", "") or "").lower() @@ -4226,7 +4288,7 @@ class TurnRunner: else: if can_edit: # First tool: send all accumulated text as new message - full_text = "\n".join(progress_lines) + full_text = _progress_text(progress_lines) result = await adapter.send( chat_id=ctx.source.chat_id, content=full_text, @@ -4262,10 +4324,15 @@ class TurnRunner: raw = ctx.progress_queue.get_nowait() if isinstance(raw, tuple) and len(raw) == 3 and raw[0] == "__dedup__": _, base_msg, count = raw - if progress_lines: + if progress_lines and ( + progress_lines[-1] == base_msg + or str(progress_lines[-1]).startswith(f"{base_msg} (×") + ): progress_lines[-1] = f"{base_msg} (×{count + 1})" await _roll_progress_overflow_if_needed() elif isinstance(raw, tuple) and len(raw) >= 1 and raw[0] == "__reset__": + if ctx.progress_grouping == "rolling": + continue # Content-bubble marker during drain: close off # the current progress bubble and start a fresh # one for any tool lines that arrived after. @@ -4278,17 +4345,30 @@ class TurnRunner: pass progress_msg_id = None progress_lines = [] + rolling_omitted_count = 0 ctx.last_progress_msg[0] = None ctx.repeat_count[0] = 0 + elif isinstance(raw, tuple) and len(raw) >= 1 and raw[0] == "__activity_start__": + continue + elif isinstance(raw, tuple) and len(raw) >= 2 and raw[0] == "__activity_state__": + rolling_header = str(raw[1]) + rolling_state_changed = True else: progress_lines.append(raw) await _roll_progress_overflow_if_needed() except Exception: break - # Final edit with all remaining tools (only if editing works) - if can_edit and progress_lines and progress_msg_id: + # Final edit with all remaining tools (only if editing works). + # A rolling bubble can contain only its omission marker when a + # single oversized activity was dropped as one complete entry. + _has_final_progress = bool( + progress_lines + or rolling_state_changed + or (rolling_header and rolling_omitted_count) + ) + if can_edit and _has_final_progress and progress_msg_id: await _roll_progress_overflow_if_needed() - if can_edit and progress_lines and progress_msg_id: + if can_edit and _has_final_progress and progress_msg_id: full_text = _progress_text(progress_lines) try: await _edit_progress_message(progress_msg_id, full_text) @@ -4545,6 +4625,15 @@ class TurnRunner: if not ctx._run_still_current(): return display_text = text + if ( + ctx.progress_grouping == "rolling" + and ctx.tool_progress_enabled + and ctx.progress_queue is not None + and not already_streamed + and str(display_text or "").strip() + ): + ctx.progress_queue.put(f"💬 {display_text}") + return if _stream_consumer is not None: if already_streamed: _stream_consumer.on_segment_break() @@ -24368,7 +24457,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _env_tp and not _tool_progress_configured else (_resolved_tp or _env_tp or "all") ) - # Tool progress grouping: "accumulate" (edit one bubble) or "separate" (one msg per tool) + # Tool progress grouping: accumulated history, bounded rolling tail, or + # one separate message per tool. progress_grouping = resolve_display_setting(user_config, platform_key, "tool_progress_grouping") or "accumulate" from gateway.status_phrases import choose_status_phrase, resolve_status_phrase_catalog _generic_status_recent: List[str] = [] @@ -24841,6 +24931,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew progress_task = None if needs_progress_queue: progress_task = asyncio.create_task(send_progress_messages()) + if progress_grouping == "rolling" and tool_progress_enabled: + progress_queue.put(("__activity_start__",)) # Start the tool-call log writer when tool_progress == "log". log_task = None @@ -25040,6 +25132,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew else f"⏳ Working — {_elapsed_mins} min{_status_detail}" ) try: + if ( + progress_grouping == "rolling" + and tool_progress_enabled + and progress_queue is not None + ): + progress_queue.put(_heartbeat_text) + continue _notify_res = None if _heartbeat_msg_id: try: @@ -25105,6 +25204,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return False return False + _inactivity_timeout = False try: # Run in thread pool to not block. Use an *inactivity*-based # timeout instead of a wall-clock limit: the agent can run for @@ -25190,7 +25290,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._run_in_executor_with_context(_run_sync_with_timeout_lifecycle) ) - _inactivity_timeout = False _POLL_INTERVAL = 5.0 if _agent_timeout is None: @@ -25779,6 +25878,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew finally: # Stop progress sender, interrupt monitor, and notification task if progress_task: + if progress_grouping == "rolling" and tool_progress_enabled: + _activity_result = result_holder[0] or {} + _current_task = asyncio.current_task() + _activity_header = _rolling_activity_terminal_header( + _activity_result, + timed_out=_inactivity_timeout, + cancelled=bool( + _current_task and _current_task.cancelling() + ), + ) + progress_queue.put(("__activity_state__", _activity_header)) progress_task.cancel() if log_task: log_task.cancel() diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 41c493383ba5f..b01e97d961431 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1219,10 +1219,13 @@ DEFAULT_CONFIG = { # call in the turn reports usage. "spinner_token_flow": True, # How gateway tool-progress is grouped on platforms that support message - # editing: "accumulate" (default) edits one bubble in place; "separate" - # sends one message per tool (the pre-v0.9 behavior, noisier). Only - # applies where tool_progress is already enabled. Per-platform override - # via display.platforms..tool_progress_grouping. + # editing: "accumulate" (default) edits one bubble in place and starts + # continuation bubbles at the platform limit; "rolling" keeps one + # bounded turn-activity bubble (tools, interim commentary, heartbeats) + # by omitting its oldest complete entries; + # "separate" sends one message per tool (the pre-v0.9 behavior, noisier). + # Only applies where tool_progress is already enabled. Per-platform + # override via display.platforms..tool_progress_grouping. "tool_progress_grouping": "accumulate", # Optional custom phrases for generic long-running status messages. # Built-in defaults live in gateway/assets/status_phrases.yaml. Users diff --git a/tests/gateway/test_display_config.py b/tests/gateway/test_display_config.py index ccbfaade4d77e..76e6ba0643dcc 100644 --- a/tests/gateway/test_display_config.py +++ b/tests/gateway/test_display_config.py @@ -275,6 +275,22 @@ class TestToolProgressGrouping: == "separate" ) + def test_platform_rolling(self): + from gateway.display_config import resolve_display_setting + + config = { + "display": { + "tool_progress_grouping": "separate", + "platforms": { + "discord": {"tool_progress_grouping": " RoLlInG "}, + }, + } + } + assert ( + resolve_display_setting(config, "discord", "tool_progress_grouping") + == "rolling" + ) + class TestReasoningStyle: """Per-platform reasoning render style (code | blockquote | subtext).""" diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py index b3d646efcdffd..cffaa6b2a11e1 100644 --- a/tests/gateway/test_run_progress_topics.py +++ b/tests/gateway/test_run_progress_topics.py @@ -112,6 +112,18 @@ class SmallLimitProgressAdapter(ProgressCaptureAdapter): return SendResult(success=True, message_id=message_id) +class Utf16SmallLimitProgressAdapter(SmallLimitProgressAdapter): + """Small adapter whose platform limit is measured in UTF-16 units.""" + + @property + def message_len_fn(self): + return lambda text: len(text.encode("utf-16-le")) // 2 + + +class TinyLimitProgressAdapter(SmallLimitProgressAdapter): + MAX_MESSAGE_LENGTH = 48 + + class MetadataEditProgressCaptureAdapter(ProgressCaptureAdapter): async def edit_message( self, chat_id, message_id, content, *, finalize: bool = False, metadata=None @@ -340,6 +352,30 @@ class ManyProgressLinesAgent: } +class RollingDedupAgent: + """Overflow the rolling tail, then repeat its newest complete entry.""" + + def __init__(self, **kwargs): + self.tool_progress_callback = kwargs.get("tool_progress_callback") + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + cb = self.tool_progress_callback + assert cb is not None + cb("tool.started", "terminal", "first-short", {}) + time.sleep(0.35) + for idx in range(6): + cb("tool.started", "terminal", f"old-line-{idx}-" + "x" * 45, {}) + for _ in range(3): + cb("tool.started", "web_search", "latest repeated query", {}) + time.sleep(0.1) + return { + "final_response": "done", + "messages": [], + "api_calls": 1, + } + + class DelayedInterimAgent: def __init__(self, **kwargs): self.interim_assistant_callback = kwargs.get("interim_assistant_callback") @@ -717,6 +753,23 @@ class CommentaryAgent: } +class SlowToolAgent: + """Emit progress, then stay active long enough for a configured heartbeat.""" + + def __init__(self, **kwargs): + self.tool_progress_callback = kwargs.get("tool_progress_callback") + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + self.tool_progress_callback("tool.started", "terminal", "long command", {}) + time.sleep(0.25) + return { + "final_response": "done", + "messages": [], + "api_calls": 1, + } + + class PreviewedResponseAgent: def __init__(self, **kwargs): self.interim_assistant_callback = kwargs.get("interim_assistant_callback") @@ -959,6 +1012,186 @@ async def test_retryable_overflow_edit_keeps_editable_bubble_identity(monkeypatc assert adapter.oversized_edits == [] +@pytest.mark.asyncio +async def test_rolling_progress_keeps_one_bounded_editable_tail(monkeypatch, tmp_path): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + ManyProgressLinesAgent, + session_id="sess-progress-rolling-tail", + config_data={ + "display": { + "tool_progress": "all", + "tool_progress_grouping": "rolling", + "interim_assistant_messages": False, + } + }, + platform=Platform.DISCORD, + chat_id="C123", + chat_type="group", + thread_id="thread-1", + adapter_cls=Utf16SmallLimitProgressAdapter, + ) + + assert result["final_response"] == "done" + assert len(adapter.sent) == 1 + assert adapter.sent[0]["metadata"] is not None + assert adapter.edits + assert {call["message_id"] for call in adapter.edits} == {"progress-1"} + rendered = [adapter.sent[0]["content"], *(call["content"] for call in adapter.edits)] + effective_limit = adapter.MAX_MESSAGE_LENGTH - 64 + assert all(adapter.message_len_fn(text) <= effective_limit for text in rendered) + final_progress = rendered[-1] + assert "earlier activities omitted" in final_progress + assert "overflow-line-7" in final_progress + assert "first-short" not in final_progress + assert adapter.oversized_sends == [] + assert adapter.oversized_edits == [] + + +@pytest.mark.asyncio +async def test_rolling_progress_omits_single_oversized_entry_without_splitting( + monkeypatch, tmp_path +): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + VerboseAgent, + session_id="sess-progress-rolling-oversized-entry", + config_data={ + "display": { + "tool_progress": "verbose", + "tool_progress_grouping": "rolling", + "interim_assistant_messages": False, + } + }, + platform=Platform.DISCORD, + adapter_cls=TinyLimitProgressAdapter, + ) + + assert result["final_response"] == "done" + assert len(adapter.sent) == 1 + assert adapter.sent[0]["content"] == "⏳ Working…" + assert adapter.edits + progress_text = adapter.edits[-1]["content"] + assert progress_text == "✅ Completed\n… 1 earlier activities omitted" + assert VerboseAgent.LONG_CODE not in progress_text + assert adapter.oversized_sends == [] + assert adapter.oversized_edits == [] + + +@pytest.mark.parametrize( + ("result", "timed_out", "cancelled", "expected"), + [ + ({}, True, False, "⚠️ Needs attention"), + ({"failed": True}, False, False, "⚠️ Needs attention"), + ({"completed": False}, False, False, "⚠️ Needs attention"), + ({"partial": True}, False, False, "⚠️ Needs attention"), + ({"error": "compression deferred"}, False, False, "⚠️ Needs attention"), + ({"interrupted": True}, False, False, "⏹️ Stopped"), + (None, False, True, "⏹️ Stopped"), + ({"failed": False}, False, False, "✅ Completed"), + ], +) +def test_rolling_activity_terminal_header_states( + result, timed_out, cancelled, expected +): + gateway_run = importlib.import_module("gateway.run") + + assert gateway_run._rolling_activity_terminal_header( + result, + timed_out=timed_out, + cancelled=cancelled, + ) == expected + + +@pytest.mark.asyncio +async def test_rolling_progress_preserves_dedup_after_omitting_old_entries( + monkeypatch, tmp_path +): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + RollingDedupAgent, + session_id="sess-progress-rolling-dedup", + config_data={ + "display": { + "tool_progress": "all", + "tool_progress_grouping": "rolling", + "interim_assistant_messages": False, + } + }, + platform=Platform.DISCORD, + adapter_cls=SmallLimitProgressAdapter, + ) + + assert result["final_response"] == "done" + assert len(adapter.sent) == 1 + final_progress = adapter.edits[-1]["content"] + assert "earlier activities omitted" in final_progress + assert "latest repeated query" in final_progress + assert "(×3)" in final_progress + + +@pytest.mark.asyncio +async def test_rolling_progress_routes_interim_commentary_into_activity_bubble( + monkeypatch, tmp_path +): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + CommentaryAgent, + session_id="sess-progress-rolling-commentary", + config_data={ + "display": { + "tool_progress": "all", + "tool_progress_grouping": "rolling", + "interim_assistant_messages": True, + }, + "streaming": {"enabled": False}, + }, + platform=Platform.DISCORD, + adapter_cls=SmallLimitProgressAdapter, + ) + + assert result["final_response"] == "done" + assert len(adapter.sent) == 1 + assert adapter.edits + rendered = adapter.edits[-1]["content"] + assert "✅ Completed" in rendered + assert "I'll inspect the repo first." in rendered + assert {call["message_id"] for call in adapter.edits} == {"progress-1"} + + +@pytest.mark.asyncio +async def test_rolling_progress_routes_heartbeat_into_same_activity_bubble( + monkeypatch, tmp_path +): + monkeypatch.setenv("HERMES_AGENT_NOTIFY_INTERVAL", "0.05") + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + SlowToolAgent, + session_id="sess-progress-rolling-heartbeat", + config_data={ + "display": { + "tool_progress": "all", + "tool_progress_grouping": "rolling", + "interim_assistant_messages": False, + "long_running_notifications": True, + } + }, + platform=Platform.DISCORD, + adapter_cls=SmallLimitProgressAdapter, + ) + + assert result["final_response"] == "done" + assert len(adapter.sent) == 1 + assert adapter.edits + assert "Working — 0 min" in adapter.edits[-1]["content"] + assert {call["message_id"] for call in adapter.edits} == {"progress-1"} + + @pytest.mark.asyncio async def test_display_streaming_does_not_enable_gateway_streaming(monkeypatch, tmp_path): adapter, result = await _run_with_agent( diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index 61c47d5a421ca..31771fe3207cf 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -438,11 +438,21 @@ display: tool_progress_command: false # set to true to enable /verbose in messaging # How progress is grouped on platforms that support message editing: # accumulate (default) — edit one bubble in place as tools run + # rolling — keep one bounded activity bubble for the turn # separate — send one message per tool (pre-v0.9 style; noisier) # Only applies where tool_progress is already enabled. - tool_progress_grouping: accumulate # accumulate | separate + tool_progress_grouping: accumulate # accumulate | rolling | separate ``` +`rolling` is useful for long-running turns in notification-heavy chats. Hermes +sends one `⏳ Working…` activity bubble, edits it with tool progress, concise +interim commentary, and long-running heartbeats, then marks it complete while +the final answer is delivered separately. When the platform's message limit is +reached, Hermes drops the oldest complete activity entries, adds an omission +count, and keeps editing the same bubble. It never splits an individual entry. +The default `accumulate` mode is unchanged and starts a continuation bubble +when the current one fills. + ### `log` mode — audit file instead of chat messages Setting `display.tool_progress: log` sends **no** progress bubbles to chat. Instead, each tool call is appended as a line to `~/.hermes/logs/tool_calls.log` — a rotating audit file (5 MB × 3 backups) run through the same secret-redacting formatter as regular logs, so credentials never land on disk. Use it when you want a full tool-call trail without any chat noise.