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
17 changes: 11 additions & 6 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -4974,12 +4974,17 @@ def progress_callback(tool_name: str, preview: str = None, args: dict = None):
progress_queue.put(msg)

# Background task to send progress messages
# Accumulates tool lines into a single message that gets edited
# For DM top-level Slack messages, source.thread_id is None but the
# final reply will be threaded under the original message via reply_to.
# Use event_message_id as fallback so progress messages land in the
# same thread as the final response instead of going to the DM root.
_progress_thread_id = source.thread_id or event_message_id
# Accumulates tool lines into a single message that gets edited.
#
# Threading metadata is platform-specific:
# - Slack DM threading needs event_message_id fallback (reply thread)
# - Telegram uses message_thread_id only for forum topics; passing a
# normal DM/group message id as thread_id causes send failures
# - Other platforms should use explicit source.thread_id only
if source.platform == Platform.SLACK:
_progress_thread_id = source.thread_id or event_message_id
else:
_progress_thread_id = source.thread_id
_progress_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None

async def send_progress_messages():
Expand Down
90 changes: 87 additions & 3 deletions tests/gateway/test_run_progress_topics.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@


class ProgressCaptureAdapter(BasePlatformAdapter):
def __init__(self):
super().__init__(PlatformConfig(enabled=True, token="fake-token"), Platform.TELEGRAM)
def __init__(self, platform=Platform.TELEGRAM):
super().__init__(PlatformConfig(enabled=True, token="***"), platform)
self.sent = []
self.edits = []
self.typing = []
Expand Down Expand Up @@ -76,7 +76,7 @@ def _make_runner(adapter):
GatewayRunner = gateway_run.GatewayRunner

runner = object.__new__(GatewayRunner)
runner.adapters = {Platform.TELEGRAM: adapter}
runner.adapters = {adapter.platform: adapter}
runner._voice_mode = {}
runner._prefill_messages = []
runner._ephemeral_system_prompt = ""
Expand Down Expand Up @@ -133,3 +133,87 @@ async def test_run_agent_progress_stays_in_originating_topic(monkeypatch, tmp_pa
]
assert adapter.edits
assert all(call["metadata"] == {"thread_id": "17585"} for call in adapter.typing)


@pytest.mark.asyncio
async def test_run_agent_progress_does_not_use_event_message_id_for_telegram_dm(monkeypatch, tmp_path):
"""Telegram DM progress must not reuse event message id as thread metadata."""
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 = FakeAgent
monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)

adapter = ProgressCaptureAdapter(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="12345",
chat_type="dm",
thread_id=None,
)

result = await runner._run_agent(
message="hello",
context_prompt="",
history=[],
source=source,
session_id="sess-2",
session_key="agent:main:telegram:dm:12345",
event_message_id="777",
)

assert result["final_response"] == "done"
assert adapter.sent
assert adapter.sent[0]["metadata"] is None
assert all(call["metadata"] is None for call in adapter.typing)


@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."""
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 = FakeAgent
monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)

adapter = ProgressCaptureAdapter(platform=Platform.SLACK)
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.SLACK,
chat_id="D123",
chat_type="dm",
thread_id=None,
)

result = await runner._run_agent(
message="hello",
context_prompt="",
history=[],
source=source,
session_id="sess-3",
session_key="agent:main:slack:dm:D123",
event_message_id="1234567890.000001",
)

assert result["final_response"] == "done"
assert adapter.sent
assert adapter.sent[0]["metadata"] == {"thread_id": "1234567890.000001"}
assert all(call["metadata"] == {"thread_id": "1234567890.000001"} for call in adapter.typing)
Loading