diff --git a/gateway/run.py b/gateway/run.py index 43853f9b66b80..3446b3c5e3143 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -31,6 +31,7 @@ import inspect import json import logging +import math import os import queue import re @@ -40,7 +41,7 @@ import signal import threading import time -from collections import OrderedDict +from collections import OrderedDict, deque from contextvars import copy_context from pathlib import Path from datetime import datetime @@ -75,6 +76,13 @@ # wall deadlines plus readiness; other platforms retain the 30s isolation bound. _TELEGRAM_CONNECT_TIMEOUT_SECS_DEFAULT = 180.0 _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0 +_ASYNC_DELEGATION_PANEL_CYCLE_BUDGET_SECS = 0.2 +_ASYNC_DELEGATION_PANEL_REQUEST_TIMEOUT_SECS = 0.1 +_ASYNC_DELEGATION_PANEL_INITIAL_SEND_SPACING_SECS = 1.1 +_ASYNC_DELEGATION_PANEL_EDIT_SPACING_SECS = 5.0 +_ASYNC_DELEGATION_PANEL_WRITE_WINDOW_SECS = 60.0 +_ASYNC_DELEGATION_PANEL_MAX_WRITES_PER_WINDOW = 6 +_ASYNC_DELEGATION_PANEL_MAX_CHARS = 500 _GATEWAY_PROXY_SSE_BUFFER_MAX_CHARS = 16 * 1024 * 1024 _TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(? float: self._running = False self._draining = True + panel_snapshot_task = getattr( + self, "_async_delegation_panel_snapshot_task", None, + ) + if panel_snapshot_task is not None and not panel_snapshot_task.done(): + panel_snapshot_task.cancel() + self._async_delegation_panel_snapshot_task = None stop_watchdog = getattr(self, "_stop_systemd_watchdog", None) if callable(stop_watchdog): @@ -15492,7 +15509,11 @@ def _cache_session_source(self, session_key: str, source) -> None: cached_sources = OrderedDict() self._session_sources = cached_sources try: - cached_sources[session_key] = dataclasses.replace(source) + cached_source = dataclasses.replace(source) + transport_ref = getattr(source, "_transport_adapter_ref", None) + if transport_ref is not None: + cached_source._transport_adapter_ref = transport_ref + cached_sources[session_key] = cached_source except Exception: logger.debug("Failed to cache live session source for %s", session_key, exc_info=True) return @@ -21248,6 +21269,290 @@ def _enrich_async_delegation_routing(self, evt: dict) -> None: if parsed.get("thread_id"): evt["thread_id"] = parsed["thread_id"] + @staticmethod + def _async_delegation_panel_status(raw_status: object) -> tuple[str, bool]: + """Normalize internal lifecycle state and identify terminal records.""" + key = str(raw_status or "").strip().lower().replace("-", "_").replace(" ", "_") + states = { + "queued": ("pending", False), + "pending": ("pending", False), + "running": ("running", False), + "active": ("running", False), + "stalling": ("stalling", False), + "finalizing": ("finalizing", False), + "completing": ("finalizing", False), + "stalled": ("stalled", True), + "completed": ("completed", True), + "success": ("completed", True), + "error": ("error", True), + "failed": ("error", True), + "interrupted": ("interrupted", True), + "cancelled": ("interrupted", True), + "canceled": ("interrupted", True), + "timed_out": ("timed out", True), + "timeout": ("timed out", True), + } + # Unknown internal states are not safe to treat as indefinitely active. + return states.get(key, ("unknown", True)) + + @staticmethod + def _async_delegation_panel_clock() -> float: + return time.monotonic() + + @staticmethod + def _async_delegation_panel_retry_after_seconds(value: Any) -> float | None: + if value is None: + return None + try: + seconds = value.total_seconds() if hasattr(value, "total_seconds") else float(value) + except (TypeError, ValueError, OverflowError): + return None + return seconds if math.isfinite(seconds) and seconds > 0 else None + + @staticmethod + def _async_delegation_panel_text(record: dict, now: float) -> tuple[str, tuple]: + """Render a bounded allowlist; temporal ages never enter the signature.""" + def clean(value: object, limit: int) -> str: + raw = str(value or "")[: max(256, limit * 4)] + redacted = _redact_gateway_user_facing_secrets(raw) + printable = "".join(char if char.isprintable() else " " for char in redacted) + text = " ".join(printable.split()) + if len(text) <= limit: + return text.rstrip() + return text[: max(0, limit - 3)].rstrip() + "..." + + def age_seconds(value: object) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, int): + return min(max(0, value), 999_999_999) + if not isinstance(value, float) or not math.isfinite(value): + return None + return min(max(0, int(value)), 999_999_999) + + delegation_id = str(record.get("delegation_id") or "") + status, _terminal = GatewayRunner._async_delegation_panel_status(record.get("status")) + goal = clean(record.get("goal"), 100) or "(no goal)" + role = clean(record.get("role"), 40) or "leaf" + in_tool = record.get("in_tool") is True + lines = [ + f"Delegation {delegation_id}", + f"status: {status} · {'in tool' if in_tool else 'waiting'}", + f"goal: {goal}", + f"role: {role}", + ] + visible_signature = list(lines) + + children = record.get("children_activity") + if not isinstance(children, (list, tuple)): + children = [] + for index, child in enumerate(children, start=1): + if not isinstance(child, dict): + continue + api_calls = child.get("api_calls") + if isinstance(api_calls, int) and not isinstance(api_calls, bool): + bounded_calls = min(max(0, api_calls), 999_999_999) + call_text = f"{bounded_calls}{'+' if api_calls > bounded_calls else ''}" + else: + call_text = "?" + tool_match = re.match(r"[A-Za-z0-9_.-]+", str(child.get("current_tool") or "")) + tool = clean(tool_match.group(0) if tool_match else "waiting", 40) or "waiting" + base = f"child {index}: {call_text} calls · {tool}" + if len("\n".join([*lines, base])) > _ASYNC_DELEGATION_PANEL_MAX_CHARS: + break + activity_age = age_seconds(child.get("seconds_since_activity")) + with_age = f"{base} · active {activity_age}s ago" if activity_age is not None else base + lines.append( + with_age + if len("\n".join([*lines, with_age])) <= _ASYNC_DELEGATION_PANEL_MAX_CHARS + else base + ) + visible_signature.append(base) + + ages = [] + dispatched_at = record.get("dispatched_at") + if isinstance(dispatched_at, (int, float)) and not isinstance(dispatched_at, bool): + try: + dispatched = float(dispatched_at) + except OverflowError: + dispatched = math.inf + if math.isfinite(dispatched) and math.isfinite(now): + ages.append(f"elapsed: {max(0, int(now - dispatched))}s") + progress_age = age_seconds(record.get("seconds_since_progress")) + if progress_age is not None: + ages.append(f"progress: {progress_age}s ago") + if ages: + age_line = " · ".join(ages) + if len("\n".join([*lines, age_line])) <= _ASYNC_DELEGATION_PANEL_MAX_CHARS: + lines.append(age_line) + + content = "\n".join(lines) + return content, (delegation_id, tuple(visible_signature)) + + def _canonical_async_delegation_panel_source(self, session_key: str): + """Return a retained origin only; a parsed key is not canonical routing.""" + if not session_key: + return None + cached_source = self._get_cached_session_source(session_key) + if ( + getattr(getattr(self, "config", None), "multiplex_profiles", False) + and cached_source is not None + and self._registered_transport_adapter(cached_source) is not None + ): + return cached_source + try: + self.session_store._ensure_loaded() + entry = self.session_store._entries.get(session_key) + source = getattr(entry, "origin", None) if entry else None + if source is not None: + return source + except Exception: + logger.debug("Delegation panel origin lookup failed", exc_info=True) + return cached_source + + async def _update_async_delegation_telegram_panels(self) -> None: + """Best-effort Telegram status projection; never affects completion delivery.""" + from tools.async_delegation import list_async_delegations + + clock = self._async_delegation_panel_clock + deadline = clock() + _ASYNC_DELEGATION_PANEL_CYCLE_BUDGET_SECS + panels = getattr(self, "_async_delegation_panels", None) + if panels is None: + panels = self._async_delegation_panels = {} + chats = getattr(self, "_async_delegation_panel_chats", None) + if chats is None: + chats = self._async_delegation_panel_chats = {} + snapshot_task = getattr(self, "_async_delegation_panel_snapshot_task", None) + if snapshot_task is None: + snapshot_task = asyncio.create_task(asyncio.to_thread(list_async_delegations)) + self._async_delegation_panel_snapshot_task = snapshot_task + try: + remaining = deadline - clock() + if remaining <= 0: + return + records = await asyncio.wait_for(asyncio.shield(snapshot_task), timeout=remaining) + except Exception: + logger.debug("Delegation panel snapshot failed", exc_info=True) + return + finally: + if snapshot_task.done(): + self._async_delegation_panel_snapshot_task = None + live_ids = {str(record.get("delegation_id") or "") for record in records} + for delegation_id, panel in list(panels.items()): + if delegation_id in live_ids: + continue + status_ids = getattr(panel.get("adapter"), "_status_message_ids", None) + if isinstance(status_ids, dict): + status_ids.pop((panel.get("chat_id"), panel.get("status_key")), None) + panels.pop(delegation_id, None) + + for record in records: + if clock() >= deadline: + break + delegation_id = str(record.get("delegation_id") or "") + source = self._canonical_async_delegation_panel_source(str(record.get("session_key") or "")) + if source is None or getattr(source, "platform", None) != Platform.TELEGRAM: + continue + transport_adapter = None + if getattr(getattr(self, "config", None), "multiplex_profiles", False): + transport_adapter = self._registered_transport_adapter(source) + if transport_adapter is None: + continue + adapter = self._adapter_for_source(source) + if transport_adapter is not None and adapter is not transport_adapter: + continue + if getattr(adapter, "platform", None) != Platform.TELEGRAM: + continue + sender = getattr(adapter, "send_or_update_status", None) + if not delegation_id or not callable(sender): + continue + content, signature = self._async_delegation_panel_text(record, time.time()) + _status, terminal = self._async_delegation_panel_status(record.get("status")) + panel = panels.get(delegation_id) + panel = panels.setdefault(delegation_id, { + "adapter": adapter, + "chat_id": str(source.chat_id), + "status_key": f"delegation:{delegation_id}", + "signature": None, + "sent": False, + "terminal": terminal, + }) + if terminal and not panel["sent"]: + panel["terminal"] = True + continue + if panel["terminal"] or panel["signature"] == signature: + continue + chat_key = (id(adapter), str(source.chat_id)) + chat = chats.setdefault(chat_key, { + "last_send": -math.inf, + "last_edit": -math.inf, + "blocked_until": 0.0, + "writes": deque(), + }) + mono_now = clock() + writes = chat["writes"] + while writes and writes[0] <= mono_now - _ASYNC_DELEGATION_PANEL_WRITE_WINDOW_SECS: + writes.popleft() + if ( + mono_now < chat["blocked_until"] + or len(writes) >= _ASYNC_DELEGATION_PANEL_MAX_WRITES_PER_WINDOW + or ( + panel["sent"] + and mono_now - chat["last_edit"] < _ASYNC_DELEGATION_PANEL_EDIT_SPACING_SECS + ) + or ( + not panel["sent"] + and mono_now - chat["last_send"] + < _ASYNC_DELEGATION_PANEL_INITIAL_SEND_SPACING_SECS + ) + ): + continue + # Consume an admitted update before I/O: failures must not replay it. + panel["signature"] = signature + writes.append(mono_now) + if panel["sent"]: + chat["last_edit"] = mono_now + else: + chat["last_send"] = mono_now + try: + timeout = min( + _ASYNC_DELEGATION_PANEL_REQUEST_TIMEOUT_SECS, + max(0.0, deadline - clock()), + ) + if timeout <= 0: + break + result = await asyncio.wait_for( + sender( + str(source.chat_id), panel["status_key"], content, + metadata=self._thread_metadata_for_source(source), + best_effort=True, terminal=terminal, + ), + timeout=timeout, + ) + if getattr(result, "success", False) and getattr(result, "message_id", None): + panel["sent"] = True + seconds = self._async_delegation_panel_retry_after_seconds( + getattr(result, "retry_after", None), + ) + if seconds is not None: + chat["blocked_until"] = max( + chat["blocked_until"], mono_now + seconds, + ) + except Exception as exc: + seconds = self._async_delegation_panel_retry_after_seconds( + getattr(exc, "retry_after", None), + ) + if seconds is not None: + chat["blocked_until"] = max(chat["blocked_until"], mono_now + seconds) + logger.debug("Delegation panel delivery failed", exc_info=True) + finally: + if terminal: + panel["terminal"] = True + active_chat_keys = {(id(panel["adapter"]), panel["chat_id"]) for panel in panels.values()} + for chat_key in list(chats): + if chat_key not in active_chat_keys: + chats.pop(chat_key, None) + async def _async_delegation_watcher(self, interval: float = 2.0) -> None: """Drain async-delegation completions and inject them as new turns. @@ -21296,6 +21601,10 @@ async def _async_delegation_watcher(self, interval: float = 2.0) -> None: logger.error("Async delegation injection error: %s", e) except Exception as e: logger.debug("Async delegation watcher error: %s", e) + try: + await self._update_async_delegation_telegram_panels() + except Exception: + logger.debug("Async delegation panel update failed", exc_info=True) await asyncio.sleep(interval) async def _run_process_watcher(self, watcher: dict) -> None: diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 4a430702e85e3..ff5a5d7abc5d2 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -4716,6 +4716,8 @@ async def send_or_update_status( content: str, *, metadata: Optional[Dict[str, Any]] = None, + best_effort: bool = False, + terminal: bool = False, ) -> SendResult: """Send a status message, or edit the previous one with the same key. @@ -4725,7 +4727,16 @@ async def send_or_update_status( subsequent calls with the same (chat_id, status_key) edit that same message in place. If the edit fails (message deleted, too old, etc.) we drop the cached id and send fresh. + + ``best_effort`` is deliberately opt-in for bounded informational panels. + It makes exactly one plain Bot API request (send *or* edit), with no + formatting, retry, fallback send, overflow continuations, or typing. + The normal status path remains unchanged. """ + if best_effort: + return await self._send_or_update_status_best_effort( + chat_id, status_key, content, metadata=metadata, terminal=terminal, + ) key = (str(chat_id), str(status_key)) cached_id = self._status_message_ids.get(key) if cached_id is not None: @@ -4743,6 +4754,87 @@ async def send_or_update_status( self._status_message_ids[key] = str(result.message_id) return result + async def _send_or_update_status_best_effort( + self, + chat_id: str, + status_key: str, + content: str, + *, + metadata: Optional[Dict[str, Any]] = None, + terminal: bool = False, + ) -> SendResult: + """Deliver one plain-text status request without normal recovery paths.""" + if not self._bot: + return SendResult(success=False, error="Not connected") + key = (str(chat_id), str(status_key)) + cached_id = self._status_message_ids.get(key) + thread_id = self._metadata_thread_id(metadata) + private_dm_topic_send = self._is_private_dm_topic_send( + chat_id, thread_id, metadata, + ) + reply_to_id = self._reply_to_message_id_for_send( + None, metadata, self._reply_to_mode, + ) + if cached_id is None and private_dm_topic_send and reply_to_id is None: + return SendResult( + success=False, + error=self._dm_topic_missing_anchor_error(), + retryable=False, + ) + try: + if cached_id is not None: + await self._bot.edit_message_text( + chat_id=normalize_telegram_chat_id(chat_id), + message_id=int(cached_id), + text=content, + ) + result = SendResult(success=True, message_id=str(cached_id)) + else: + thread_kwargs = self._thread_kwargs_for_send( + chat_id, + thread_id, + metadata, + reply_to_message_id=reply_to_id, + reply_to_mode=self._reply_to_mode, + ) + if reply_to_id is not None: + thread_kwargs["reply_to_message_id"] = reply_to_id + message = await self._bot.send_message( + chat_id=normalize_telegram_chat_id(chat_id), + text=content, + **thread_kwargs, + **self._notification_kwargs(metadata), + ) + result = SendResult( + success=True, + message_id=str(getattr(message, "message_id", "")) or None, + ) + if result.message_id and not terminal: + self._status_message_ids[key] = str(result.message_id) + return result + except Exception as exc: + raw_retry_after: Any = getattr(exc, "retry_after", None) + retry_after: Optional[float] + if raw_retry_after is None: + retry_after = None + else: + try: + retry_after = float( + raw_retry_after.total_seconds() + if hasattr(raw_retry_after, "total_seconds") + else raw_retry_after + ) + except (TypeError, ValueError, OverflowError): + retry_after = None + return SendResult( + success=False, + error=_redact_telegram_error_text(exc), + retry_after=retry_after, + ) + finally: + if terminal: + self._status_message_ids.pop(key, None) + async def edit_message( self, chat_id: str, diff --git a/tests/gateway/test_async_delegation_telegram_panel.py b/tests/gateway/test_async_delegation_telegram_panel.py new file mode 100644 index 0000000000000..0084e5933665e --- /dev/null +++ b/tests/gateway/test_async_delegation_telegram_panel.py @@ -0,0 +1,596 @@ +"""Telegram-only async-delegation status-panel behavior.""" + +from __future__ import annotations + +import asyncio +import queue +import threading +import time +from collections import deque +from datetime import timedelta +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import AsyncMock + +import pytest + +from gateway.config import Platform +from gateway.platforms.base import BasePlatformAdapter, SendResult +from gateway.run import GatewayRunner +from gateway.session import SessionSource + + +def _runner(adapter): + if not hasattr(adapter, "platform"): + adapter.platform = Platform.TELEGRAM + runner = object.__new__(GatewayRunner) + runner.config = SimpleNamespace(multiplex_profiles=False) + runner._async_delegation_panels = {} + runner._async_delegation_panel_chats = {} + runner._canonical_async_delegation_panel_source = lambda session_key: SessionSource( + platform=Platform.TELEGRAM, + chat_id="-100123", + chat_type="group", + thread_id="42", + profile="default", + ) + runner._adapter_for_source = lambda source: adapter + runner._thread_metadata_for_source = lambda source: {"thread_id": source.thread_id} + return runner + + +def test_running_panel_uses_canonical_telegram_source_and_allowlisted_plain_text(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock( + return_value=SendResult(success=True, message_id="7"), + )) + runner = _runner(adapter) + records = [{ + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "RUNNING", + "goal": "Investigate\ncredentials=do-not-show " + "x" * 110, + "role": "leaf " + "y" * 50, + "dispatched_at": 10.0, + "seconds_since_progress": 4.0, + "in_tool": True, + "children_activity": [{ + "api_calls": 3, + "current_tool": "terminal --token private " + "z" * 60, + "seconds_since_activity": 2.0, + }], + "context": "must never be displayed", + "model": "must never be displayed", + "result": "must never be displayed", + "error": "must never be displayed", + }] + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: records) + runner._async_delegation_panel_clock = lambda: 20.0 + + asyncio.run(runner._update_async_delegation_telegram_panels()) + + adapter.send_or_update_status.assert_awaited_once() + args, kwargs = adapter.send_or_update_status.await_args + assert args[:2] == ("-100123", "delegation:deleg_abcdef123456") + content = args[2] + assert kwargs["metadata"] == {"thread_id": "42"} + assert kwargs["best_effort"] is True + assert content.splitlines()[0] == "Delegation deleg_abcdef123456" + assert "running" in content + assert "in tool" in content + assert "credentials" in content + assert "do-not-show" not in content + assert "credentials=***" in content + assert "must never" not in content + assert "private" not in content + + +def test_panel_rejects_non_telegram_delivery_adapters(monkeypatch): + relay_adapter = SimpleNamespace( + platform=Platform.RELAY, + send_or_update_status=AsyncMock( + return_value=SendResult(success=True, message_id="7"), + ), + ) + runner = _runner(relay_adapter) + monkeypatch.setattr( + "tools.async_delegation.list_async_delegations", + lambda: [{ + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "safe goal", + "role": "leaf", + }], + ) + + asyncio.run(runner._update_async_delegation_telegram_panels()) + + relay_adapter.send_or_update_status.assert_not_awaited() + + +def test_multiplex_panel_requires_registered_transport_even_for_profile_stamped_source(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock( + return_value=SendResult(success=True, message_id="7"), + )) + runner = _runner(adapter) + runner.config = SimpleNamespace(multiplex_profiles=True) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="99", + chat_type="dm", + profile="secondary", + ) + runner._canonical_async_delegation_panel_source = lambda _key: source + transport_known = False + runner._registered_transport_adapter = lambda _source: adapter if transport_known else None + records = [{ + "delegation_id": "deleg_abcdef123456", + "session_key": "derived-only-key", + "status": "running", + "goal": "safe goal", + "role": "leaf", + "dispatched_at": time.time(), + }] + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: records) + + asyncio.run(runner._update_async_delegation_telegram_panels()) + adapter.send_or_update_status.assert_not_awaited() + + transport_known = True + asyncio.run(runner._update_async_delegation_telegram_panels()) + adapter.send_or_update_status.assert_awaited_once() + + +def test_multiplex_live_source_provenance_survives_cache_and_beats_persisted_origin(monkeypatch): + class LiveAdapter: + platform = Platform.TELEGRAM + + def __init__(self): + self.send_or_update_status = AsyncMock( + return_value=SendResult(success=True, message_id="7"), + ) + self.gateway_runner = SimpleNamespace( + _profile_name_for_source=lambda _source: "secondary", + ) + + adapter = LiveAdapter() + source = BasePlatformAdapter.build_source( + adapter, + chat_id="99", + chat_type="dm", + ) + session_key = "agent:main:telegram:dm:99" + persisted_source = SessionSource.from_dict(source.to_dict()) + + runner = object.__new__(GatewayRunner) + runner.config = SimpleNamespace(multiplex_profiles=True) + runner.adapters = {Platform.TELEGRAM: adapter} + runner._profile_adapters = {} + runner._async_delegation_panels = {} + runner._async_delegation_panel_chats = {} + runner._thread_metadata_for_source = lambda _source: {} + runner.session_store = SimpleNamespace( + _ensure_loaded=lambda: None, + _entries={session_key: SimpleNamespace(origin=persisted_source)}, + ) + runner._cache_session_source(session_key, source) + + cached_source = runner._get_cached_session_source(session_key) + assert runner._registered_transport_adapter(cached_source) is adapter + assert runner._registered_transport_adapter(persisted_source) is None + + monkeypatch.setattr( + "tools.async_delegation.list_async_delegations", + lambda: [{ + "delegation_id": "deleg_abcdef123456", + "session_key": session_key, + "status": "running", + "goal": "safe goal", + "role": "leaf", + }], + ) + + asyncio.run(runner._update_async_delegation_telegram_panels()) + + adapter.send_or_update_status.assert_awaited_once() + + +def test_terminal_first_observation_stays_a_complete_silent_tombstone(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock()) + runner = _runner(adapter) + record = { + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "completed", + "goal": "already done", + "role": "leaf", + } + monkeypatch.setattr( + "tools.async_delegation.list_async_delegations", + lambda: [record], + ) + + asyncio.run(runner._update_async_delegation_telegram_panels()) + asyncio.run(runner._update_async_delegation_telegram_panels()) + + adapter.send_or_update_status.assert_not_awaited() + assert runner._async_delegation_panels[record["delegation_id"]] == { + "adapter": adapter, + "chat_id": "-100123", + "status_key": "delegation:deleg_abcdef123456", + "signature": None, + "sent": False, + "terminal": True, + } + + +def test_panel_normalizes_lifecycle_statuses_through_an_explicit_allowlist(): + expected = { + "queued": "pending", "ACTIVE": "running", "completing": "finalizing", + "success": "completed", "failed": "error", "cancelled": "interrupted", + "timed_out": "timed out", "arbitrary internal detail": "unknown", + } + for raw, normalized in expected.items(): + content, _signature = GatewayRunner._async_delegation_panel_text( + {"delegation_id": "deleg_abcdef123456", "status": raw, "goal": "g", "role": "leaf"}, + now=0.0, + ) + assert f"status: {normalized}" in content + + +@pytest.mark.parametrize("delegation_id", ["deleg_abcdef123456", "abc"]) +def test_panel_displays_the_full_actionable_delegation_id(delegation_id): + content, signature = GatewayRunner._async_delegation_panel_text( + {"delegation_id": delegation_id, "status": "running", "goal": "g", "role": "leaf"}, + now=0.0, + ) + + assert content.splitlines()[0] == f"Delegation {delegation_id}" + assert signature[0] == delegation_id + + +def test_panel_renders_every_visible_child_and_temporal_ages_do_not_change_signature(): + record = { + "delegation_id": "deleg_abcdef123456", + "status": "running", + "goal": "fan out safely", + "role": "orchestrator", + "dispatched_at": 10.0, + "seconds_since_progress": 4.0, + "children_activity": [ + {"api_calls": 3, "current_tool": "terminal", "seconds_since_activity": 2.0}, + {"api_calls": 7, "current_tool": "web_search", "seconds_since_activity": 1.0}, + {"api_calls": 1, "current_tool": "read_file", "seconds_since_activity": 0.5}, + ], + } + + first, first_signature = GatewayRunner._async_delegation_panel_text(record, now=20.0) + record["seconds_since_progress"] = 40.0 + for child in record["children_activity"]: + child["seconds_since_activity"] += 30.0 + second, second_signature = GatewayRunner._async_delegation_panel_text(record, now=80.0) + + assert "child 1: 3 calls · terminal" in first + assert "child 2: 7 calls · web_search" in first + assert "child 3: 1 calls · read_file" in first + assert len(first) <= 500 + assert first != second + assert first_signature == second_signature + + +def test_panel_bounds_pathological_child_api_call_counts(): + content, _signature = GatewayRunner._async_delegation_panel_text( + { + "delegation_id": "deleg_abcdef123456", + "status": "running", + "goal": "bounded", + "role": "leaf", + "children_activity": [{ + "api_calls": 10 ** 10_000, + "current_tool": "terminal", + "seconds_since_activity": 10 ** 10_000, + }], + "seconds_since_progress": 10 ** 10_000, + "dispatched_at": 10 ** 10_000, + }, + now=0.0, + ) + + assert "child 1: 999999999+ calls · terminal" in content + assert len(content) <= 500 + + +@pytest.mark.asyncio +async def test_snapshot_acquisition_is_inside_the_cycle_budget(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock()) + runner = _runner(adapter) + release = threading.Event() + + def slow_snapshot(): + release.wait(timeout=1.0) + return [] + + monkeypatch.setattr("tools.async_delegation.list_async_delegations", slow_snapshot) + started = time.perf_counter() + await runner._update_async_delegation_telegram_panels() + + assert time.perf_counter() - started < 0.35 + adapter.send_or_update_status.assert_not_awaited() + release.set() + snapshot_task = runner._async_delegation_panel_snapshot_task + assert snapshot_task is not None + await snapshot_task + + +def test_rate_rejection_preserves_signature_eligibility_and_terminal_without_a_sent_panel_is_silent(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock( + return_value=SendResult(success=True, message_id="7"), + )) + runner = _runner(adapter) + record = { + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "safe goal", + "role": "leaf", + "dispatched_at": 10.0, + } + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: [record]) + now = 20.0 + runner._async_delegation_panel_clock = lambda: now + chat_key = (id(adapter), "-100123") + runner._async_delegation_panel_chats[chat_key] = { + "last_send": 19.5, + "last_edit": 0.0, + "blocked_until": 0.0, + "writes": deque(), + } + + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 0 + assert runner._async_delegation_panels[record["delegation_id"]]["signature"] is None + + now = 21.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 1 + + # If the admitted send had failed to create a message, terminal state must + # tombstone instead of creating a new cosmetic message after completion. + panel = runner._async_delegation_panels[record["delegation_id"]] + panel["sent"] = False + record["status"] = "completed" + now = 30.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 1 + assert panel["terminal"] is True + + +def test_retry_after_timedelta_opens_adapter_chat_circuit_and_rolling_cap_is_finite(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock( + return_value=SendResult( + success=False, + error="flood controlled", + retry_after=cast(Any, timedelta(seconds=30)), + ), + )) + runner = _runner(adapter) + record = { + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "first state", + "role": "leaf", + "dispatched_at": 10.0, + } + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: [record]) + now = 20.0 + runner._async_delegation_panel_clock = lambda: now + + asyncio.run(runner._update_async_delegation_telegram_panels()) + chat = runner._async_delegation_panel_chats[(id(adapter), "-100123")] + assert chat["blocked_until"] == pytest.approx(50.0) + assert adapter.send_or_update_status.await_count == 1 + + record["goal"] = "changed while blocked" + now = 25.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 1 + + # Once the circuit expires, six prior admitted writes in the rolling + # window still suppress the update; expiring those writes admits it. + now = 51.0 + chat["writes"] = deque([45.0, 46.0, 47.0, 48.0, 49.0, 50.0]) + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 1 + + now = 111.0 + adapter.send_or_update_status.return_value = SendResult(success=True, message_id="7") + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 2 + + +def test_first_send_is_immediately_eligible_and_chat_send_spacing_is_1_1_seconds(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock( + return_value=SendResult(success=True, message_id="7"), + )) + runner = _runner(adapter) + records = [ + { + "delegation_id": "deleg_abcdef111111", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "first", + "role": "leaf", + "dispatched_at": 0.0, + }, + { + "delegation_id": "deleg_abcdef222222", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "second", + "role": "leaf", + "dispatched_at": 0.0, + }, + ] + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: records) + now = 0.1 + runner._async_delegation_panel_clock = lambda: now + + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 1 + + now = 1.19 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 1 + + now = 1.21 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 2 + + +def test_consecutive_edits_share_a_five_second_chat_spacing(monkeypatch): + adapter = SimpleNamespace(send_or_update_status=AsyncMock( + return_value=SendResult(success=True, message_id="7"), + )) + runner = _runner(adapter) + record = { + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "initial", + "role": "leaf", + "dispatched_at": 0.0, + } + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: [record]) + now = 10.0 + runner._async_delegation_panel_clock = lambda: now + + asyncio.run(runner._update_async_delegation_telegram_panels()) + record["goal"] = "first edit" + now = 11.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 2 + + record["goal"] = "second edit" + now = 15.9 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 2 + + now = 16.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 3 + + +def test_terminal_edit_is_attempted_once_then_absent_state_prunes_panel_and_adapter_cache(monkeypatch): + adapter = SimpleNamespace( + platform=Platform.TELEGRAM, + _status_message_ids={}, + send_or_update_status=AsyncMock(side_effect=[ + SendResult(success=True, message_id="7"), + SendResult(success=False, error="cosmetic edit failed"), + ]), + ) + runner = _runner(adapter) + record = { + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "initial", + "role": "leaf", + "dispatched_at": 0.0, + } + records = [record] + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: records) + now = 10.0 + runner._async_delegation_panel_clock = lambda: now + + asyncio.run(runner._update_async_delegation_telegram_panels()) + cache_key = ("-100123", "delegation:deleg_abcdef123456") + adapter._status_message_ids[cache_key] = "7" + + record["status"] = "completed" + now = 20.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 2 + assert runner._async_delegation_panels[record["delegation_id"]]["terminal"] is True + + record["goal"] = "must not trigger another terminal edit" + now = 30.0 + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert adapter.send_or_update_status.await_count == 2 + + records.clear() + asyncio.run(runner._update_async_delegation_telegram_panels()) + assert record["delegation_id"] not in runner._async_delegation_panels + assert cache_key not in adapter._status_message_ids + + +@pytest.mark.asyncio +async def test_slow_panel_request_is_cancelled_inside_the_cycle_budget(monkeypatch): + never = asyncio.Event() + + async def slow_status(*_args, **_kwargs): + await never.wait() + + adapter = SimpleNamespace(send_or_update_status=AsyncMock(side_effect=slow_status)) + runner = _runner(adapter) + records = [{ + "delegation_id": "deleg_abcdef123456", + "session_key": "agent:main:telegram:group:-100123:42", + "status": "running", + "goal": "bounded request", + "role": "leaf", + "dispatched_at": time.time(), + }] + monkeypatch.setattr("tools.async_delegation.list_async_delegations", lambda: records) + + started = time.perf_counter() + await runner._update_async_delegation_telegram_panels() + + assert time.perf_counter() - started < 0.35 + assert adapter.send_or_update_status.await_count == 1 + + +def test_completion_delivery_precedes_nonfatal_panel_failure(monkeypatch): + from tools import process_registry as process_registry_module + + events = [] + completion_queue = queue.Queue() + completion_queue.put({ + "type": "async_delegation", + "delegation_id": "deleg_abcdef123456", + "status": "completed", + }) + monkeypatch.setattr( + process_registry_module, + "process_registry", + SimpleNamespace(completion_queue=completion_queue), + ) + monkeypatch.setattr("gateway.run._format_gateway_process_notification", lambda _evt: "done") + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._enrich_async_delegation_routing = lambda _evt: None + + async def deliver(_text, _evt): + events.append("completion") + return True + + async def panel(): + events.append("panel") + raise RuntimeError("cosmetic failure") + + runner._deliver_completion_notification = deliver + runner._update_async_delegation_telegram_panels = panel + sleep_calls = 0 + + async def stop_after_one_cycle(_delay): + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls == 2: + runner._running = False + + monkeypatch.setattr("gateway.run.asyncio.sleep", stop_after_one_cycle) + asyncio.run(runner._async_delegation_watcher(interval=0)) + + assert events == ["completion", "panel"] diff --git a/tests/gateway/test_telegram_status_delivery.py b/tests/gateway/test_telegram_status_delivery.py new file mode 100644 index 0000000000000..87fcb77fad79c --- /dev/null +++ b/tests/gateway/test_telegram_status_delivery.py @@ -0,0 +1,171 @@ +"""Best-effort Telegram status delivery stays a one-request plain-text path.""" + +from __future__ import annotations + +import sys +import types +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig + + +def _install_fake_telegram(monkeypatch): + fake_telegram = types.ModuleType("telegram") + fake_telegram.Update = SimpleNamespace(ALL_TYPES=()) + fake_telegram.Bot = object + fake_telegram.Message = object + fake_telegram.InlineKeyboardButton = object + fake_telegram.InlineKeyboardMarkup = object + + fake_error = types.ModuleType("telegram.error") + fake_error.NetworkError = type("NetworkError", (Exception,), {}) + fake_error.BadRequest = type("BadRequest", (Exception,), {}) + fake_error.TimedOut = type("TimedOut", (Exception,), {}) + fake_telegram.error = fake_error + + fake_constants = types.ModuleType("telegram.constants") + fake_constants.ParseMode = SimpleNamespace(MARKDOWN_V2="MarkdownV2") + fake_constants.ChatType = SimpleNamespace( + GROUP="group", SUPERGROUP="supergroup", CHANNEL="channel", PRIVATE="private", + ) + fake_telegram.constants = fake_constants + + fake_ext = types.ModuleType("telegram.ext") + fake_ext.Application = object + fake_ext.CommandHandler = object + fake_ext.CallbackQueryHandler = object + fake_ext.MessageHandler = object + fake_ext.ContextTypes = SimpleNamespace(DEFAULT_TYPE=object) + fake_ext.filters = object + + fake_request = types.ModuleType("telegram.request") + fake_request.HTTPXRequest = object + + monkeypatch.setitem(sys.modules, "telegram", fake_telegram) + monkeypatch.setitem(sys.modules, "telegram.error", fake_error) + monkeypatch.setitem(sys.modules, "telegram.constants", fake_constants) + monkeypatch.setitem(sys.modules, "telegram.ext", fake_ext) + monkeypatch.setitem(sys.modules, "telegram.request", fake_request) + + +@pytest.fixture +def adapter(monkeypatch): + _install_fake_telegram(monkeypatch) + from plugins.platforms.telegram.adapter import TelegramAdapter + + result = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + result._bot = MagicMock() + result._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=77)) + result._bot.edit_message_text = AsyncMock() + return result + + +@pytest.mark.asyncio +async def test_best_effort_status_sends_plain_text_once_without_formatting_or_fallback(adapter): + result = await adapter.send_or_update_status( + "123", + "delegation:abc123", + "Delegation abc123\nstatus: running", + metadata={"thread_id": "42"}, + best_effort=True, + ) + + assert result.success is True + assert result.message_id == "77" + adapter._bot.send_message.assert_awaited_once_with( + chat_id=123, + text="Delegation abc123\nstatus: running", + message_thread_id=42, + disable_notification=True, + ) + + +@pytest.mark.asyncio +async def test_best_effort_status_preserves_private_dm_topic_reply_anchor(adapter): + result = await adapter.send_or_update_status( + "123", + "delegation:abc123", + "Delegation abc123\nstatus: running", + metadata={ + "thread_id": "42", + "telegram_dm_topic_reply_fallback": True, + "telegram_reply_to_message_id": 462, + }, + best_effort=True, + ) + + assert result.success is True + adapter._bot.send_message.assert_awaited_once_with( + chat_id=123, + text="Delegation abc123\nstatus: running", + reply_to_message_id=462, + message_thread_id=42, + disable_notification=True, + ) + + +@pytest.mark.asyncio +async def test_best_effort_status_fails_closed_when_private_dm_topic_anchor_is_missing(adapter): + result = await adapter.send_or_update_status( + "123", + "delegation:abc123", + "Delegation abc123\nstatus: running", + metadata={ + "thread_id": "42", + "telegram_dm_topic_reply_fallback": True, + }, + best_effort=True, + ) + + assert result.success is False + assert result.retryable is False + assert "requires a reply anchor" in (result.error or "") + adapter._bot.send_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_best_effort_status_edits_once_without_topic_kwargs_or_fallback_and_terminal_cleans_cache(adapter): + adapter._status_message_ids[("123", "delegation:abc123")] = "77" + + result = await adapter.send_or_update_status( + "123", + "delegation:abc123", + "Delegation abc123\nstatus: completed", + metadata={"thread_id": "42"}, + best_effort=True, + terminal=True, + ) + + assert result.success is True + adapter._bot.edit_message_text.assert_awaited_once_with( + chat_id=123, + message_id=77, + text="Delegation abc123\nstatus: completed", + ) + adapter._bot.send_message.assert_not_awaited() + assert ("123", "delegation:abc123") not in adapter._status_message_ids + + +@pytest.mark.asyncio +async def test_best_effort_retry_after_returns_once_without_sleep_or_fallback(adapter): + class RetryAfterError(Exception): + retry_after = timedelta(seconds=45) + + adapter._bot.send_message.side_effect = RetryAfterError("retry after 45") + + result = await adapter.send_or_update_status( + "123", + "delegation:abc123", + "Delegation abc123\nstatus: running", + best_effort=True, + ) + + assert result.success is False + assert result.retry_after == 45.0 + adapter._bot.send_message.assert_awaited_once() + adapter._bot.edit_message_text.assert_not_awaited() + assert ("123", "delegation:abc123") not in adapter._status_message_ids diff --git a/website/docs/user-guide/features/delegation.md b/website/docs/user-guide/features/delegation.md index 14fbad48058e3..cfe3c451f0d57 100644 --- a/website/docs/user-guide/features/delegation.md +++ b/website/docs/user-guide/features/delegation.md @@ -269,6 +269,16 @@ A delegation the stall monitor has flagged shows as children show their quiet time so you can tell "slow" from "stuck" at a glance. +### Telegram status panel + +For background delegations started from Telegram, Hermes maintains one small +plain-text status message in the originating chat or topic. It is updated in +place only when visible task state changes and shows the full actionable delegation ID, +normalized lifecycle state, goal/role, and bounded child activity. It never +includes the delegation context, model, result/error payload, tool arguments, +or session key. The panel is best-effort and locally rate-limited; it does not +change normal Telegram delivery or global flood-control behavior. + ## Live Transcripts Every `delegate_task` dispatch also creates one **append-only, human-readable log per task** so you (or the parent agent) can watch a subagent work in real time instead of waiting for the consolidated summary: