Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 73 additions & 6 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -670,6 +670,32 @@ def _reload_runtime_env_preserving_config_authority() -> None:
_AGENT_PENDING_SENTINEL = object()


def _platform_message_limit(adapter: Any) -> Optional[int]:
"""Return the platform message limit used for progressive edits, if known."""
limit = getattr(adapter, "MAX_MESSAGE_LENGTH", None)
if limit is None:
return None
try:
parsed = int(limit)
except (TypeError, ValueError):
return None
return parsed if parsed > 0 else None


def _platform_message_len_fn(adapter: Any):
"""Return the platform-aware length function for outbound text."""
fn = getattr(adapter, "message_len_fn", None)
return fn if callable(fn) else len


def _text_exceeds_platform_limit(adapter: Any, text: str) -> bool:
"""Whether *text* would exceed the adapter's single-message edit budget."""
limit = _platform_message_limit(adapter)
if limit is None:
return False
return _platform_message_len_fn(adapter)(text) > limit


def _resolve_runtime_agent_kwargs() -> dict:
"""Resolve provider credentials for gateway-created AIAgent instances.

Expand Down Expand Up @@ -14647,12 +14673,17 @@ async def send_progress_messages():
except Exception:
pass

rollover_line = None

# 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:
progress_lines[-1] = f"{base_msg} (Γ—{count + 1})"
msg = progress_lines[-1] if progress_lines else base_msg
msg = f"{base_msg} (Γ—{count + 1})"
candidate_lines = (
[*progress_lines[:-1], msg]
if progress_lines
else [msg]
)
elif isinstance(raw, tuple) and len(raw) >= 1 and raw[0] == "__reset__":
# Content bubble just landed on the platform β€” close off
# the current tool-progress bubble so the next tool
Expand All @@ -14669,7 +14700,26 @@ async def send_progress_messages():
continue
else:
msg = raw
progress_lines.append(msg)
candidate_lines = [*progress_lines, msg]

if (
can_edit
and progress_msg_id is not None
and progress_lines
and _text_exceeds_platform_limit(
adapter,
"\n".join(candidate_lines),
)
):
# Start a fresh progress bubble before we overflow the
# platform edit limit. Telegram can split oversized
# edits reactively, but live tool-progress updates need
# a stable "current bubble" so later edits don't keep
# replaying the full historical transcript into older
# messages.
rollover_line = msg
else:
progress_lines = candidate_lines

# Throttle edits: batch rapid tool updates into fewer
# API calls to avoid hitting Telegram flood control.
Expand All @@ -14687,15 +14737,32 @@ async def send_progress_messages():
if not _run_still_current():
return

if can_edit and progress_msg_id is not None:
if rollover_line is not None:
result = await adapter.send(
chat_id=source.chat_id,
content=rollover_line,
reply_to=_progress_reply_to,
metadata=_progress_metadata,
)
if result.success and result.message_id:
progress_msg_id = result.message_id
progress_lines = [rollover_line]
if _cleanup_progress:
_cleanup_msg_ids.append(str(result.message_id))
else:
can_edit = False
progress_lines = [rollover_line]
elif can_edit and progress_msg_id is not None:
# Try to edit the existing progress message
full_text = "\n".join(progress_lines)
result = await adapter.edit_message(
chat_id=source.chat_id,
message_id=progress_msg_id,
content=full_text,
)
if not result.success:
if result.success and result.message_id:
progress_msg_id = result.message_id
elif not result.success:
_err = (getattr(result, "error", "") or "").lower()
if "flood" in _err or "retry after" in _err:
# Flood control hit β€” disable further edits,
Expand Down
87 changes: 87 additions & 0 deletions tests/gateway/test_run_progress_topics.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,26 @@ async def get_chat_info(self, chat_id: str):
return {"id": chat_id}


class RolloverProgressCaptureAdapter(ProgressCaptureAdapter):
MAX_MESSAGE_LENGTH = 45

def __init__(self, platform=Platform.TELEGRAM):
super().__init__(platform=platform)
self._message_counter = 0

async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
self._message_counter += 1
self.sent.append(
{
"chat_id": chat_id,
"content": content,
"reply_to": reply_to,
"metadata": metadata,
}
)
return SendResult(success=True, message_id=f"progress-{self._message_counter}")


class NonEditingProgressCaptureAdapter(ProgressCaptureAdapter):
SUPPORTS_MESSAGE_EDITING = False

Expand Down Expand Up @@ -123,6 +143,25 @@ def run_conversation(self, message, conversation_history=None, task_id=None):
}


class TelegramRolloverAgent:
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", "first command", {})
time.sleep(1.65)
self.tool_progress_callback("tool.started", "terminal", "second longer", {})
time.sleep(1.65)
self.tool_progress_callback("tool.started", "terminal", "x", {})
time.sleep(0.2)
return {
"final_response": "done",
"messages": [],
"api_calls": 1,
}


class DelayedInterimAgent:
def __init__(self, **kwargs):
self.interim_assistant_callback = kwargs.get("interim_assistant_callback")
Expand Down Expand Up @@ -253,6 +292,54 @@ async def test_run_agent_progress_does_not_use_event_message_id_for_telegram_dm(
assert all(call["metadata"] is None for call in adapter.typing)


@pytest.mark.asyncio
async def test_run_agent_telegram_progress_rolls_over_before_edit_overflow(monkeypatch, tmp_path):
monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all")

fake_dotenv = types.ModuleType("dotenv")
fake_dotenv.load_dotenv = lambda *args, **kwargs: None
monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)

fake_run_agent = types.ModuleType("run_agent")
fake_run_agent.AIAgent = TelegramRolloverAgent
monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
import tools.terminal_tool # noqa: F401 - register terminal emoji for this fake-agent test

adapter = RolloverProgressCaptureAdapter(platform=Platform.TELEGRAM)
runner = _make_runner(adapter)
gateway_run = importlib.import_module("gateway.run")
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"})

source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="-1001",
chat_type="group",
thread_id="17585",
)

result = await runner._run_agent(
message="hello",
context_prompt="",
history=[],
source=source,
session_id="sess-rollover",
session_key="agent:main:telegram:group:-1001:17585",
)

progress_sends = [call["content"] for call in adapter.sent if call["content"] != "done"]
progress_edits = adapter.edits

assert result["final_response"] == "done"
assert progress_sends[:2] == [
'πŸ’» terminal: "first command"',
'πŸ’» terminal: "second longer"',
]
assert progress_edits, "expected the newest rollover bubble to remain editable"
assert progress_edits[-1]["message_id"] == "progress-2"
assert progress_edits[-1]["content"] == 'πŸ’» terminal: "second longer"\nπŸ’» terminal: "x"'


@pytest.mark.asyncio
async def test_run_agent_progress_uses_event_message_id_for_slack_dm(monkeypatch, tmp_path):
"""Slack DM progress should keep event ts fallback threading."""
Expand Down
Loading