Skip to content
Merged
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
15 changes: 11 additions & 4 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -2394,7 +2394,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
history=history,
source=source,
session_id=session_entry.session_id,
session_key=session_key
session_key=session_key,
event_message_id=event.message_id,
)

# Stop persistent typing indicator now that the agent is done
Expand Down Expand Up @@ -4842,6 +4843,7 @@ async def _run_agent(
session_id: str,
session_key: str = None,
_interrupt_depth: int = 0,
event_message_id: Optional[str] = None,
) -> Dict[str, Any]:
"""
Run the agent with the given message and context.
Expand Down Expand Up @@ -4978,7 +4980,12 @@ def progress_callback(tool_name: str, preview: str = None, args: dict = None):

# Background task to send progress messages
# Accumulates tool lines into a single message that gets edited
_progress_metadata = {"thread_id": source.thread_id} if source.thread_id else None
# 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
_progress_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None

async def send_progress_messages():
if not progress_queue:
Expand Down Expand Up @@ -5093,7 +5100,7 @@ def _step_callback_sync(iteration: int, tool_names: list) -> None:
# Bridge sync status_callback → async adapter.send for context pressure
_status_adapter = self.adapters.get(source.platform)
_status_chat_id = source.chat_id
_status_thread_metadata = {"thread_id": source.thread_id} if source.thread_id else None
_status_thread_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None

def _status_callback_sync(event_type: str, message: str) -> None:
if not _status_adapter:
Expand Down Expand Up @@ -5174,7 +5181,7 @@ def run_sync():
adapter=_adapter,
chat_id=source.chat_id,
config=_consumer_cfg,
metadata={"thread_id": source.thread_id} if source.thread_id else None,
metadata={"thread_id": _progress_thread_id} if _progress_thread_id else None,
)
_stream_delta_cb = _stream_consumer.on_delta
stream_consumer_holder[0] = _stream_consumer
Expand Down
97 changes: 97 additions & 0 deletions tests/gateway/test_slack.py
Original file line number Diff line number Diff line change
Expand Up @@ -946,3 +946,100 @@ async def test_send_image_file_fallback_includes_caption(self, adapter, tmp_path

call_kwargs = adapter._app.client.chat_postMessage.call_args.kwargs
assert "important screenshot" in call_kwargs["text"]


# ---------------------------------------------------------------------------
# TestProgressMessageThread
# ---------------------------------------------------------------------------

class TestProgressMessageThread:
"""Verify that progress messages go to the correct thread.

Issue #2954: For Slack DM top-level messages, source.thread_id is None
but the final reply is threaded under the user's message via reply_to.
Progress messages must use the same thread anchor (the original message's
ts) so they appear in the thread instead of the DM root.
"""

@pytest.mark.asyncio
async def test_dm_toplevel_progress_uses_message_ts_as_thread(self, adapter):
"""Progress messages for a top-level DM should go into the reply thread."""
# Simulate a top-level DM: no thread_ts in the event
event = {
"channel": "D_DM",
"channel_type": "im",
"user": "U_USER",
"text": "Hello bot",
"ts": "1234567890.000001",
# No thread_ts — this is a top-level DM
}

captured_events = []
adapter.handle_message = AsyncMock(side_effect=lambda e: captured_events.append(e))

# Patch _resolve_user_name to avoid async Slack API call
with patch.object(adapter, "_resolve_user_name", new=AsyncMock(return_value="testuser")):
await adapter._handle_slack_message(event)

assert len(captured_events) == 1
msg_event = captured_events[0]
source = msg_event.source

# For a top-level DM: source.thread_id should remain None
# (session keying must not be affected)
assert source.thread_id is None, (
"source.thread_id must stay None for top-level DMs "
"so they share one continuous session"
)

# The message_id should be the event's ts — this is what the gateway
# passes as event_message_id so progress messages can thread correctly
assert msg_event.message_id == "1234567890.000001", (
"message_id must equal the event ts so _run_agent can use it as "
"the fallback thread anchor for progress messages"
)

# Verify that the Slack send() method correctly threads a message
# when metadata contains thread_id equal to the original ts
adapter._app.client.chat_postMessage = AsyncMock(return_value={"ts": "reply_ts"})
result = await adapter.send(
chat_id="D_DM",
content="⚙️ working...",
metadata={"thread_id": msg_event.message_id},
)
assert result.success
call_kwargs = adapter._app.client.chat_postMessage.call_args[1]
assert call_kwargs.get("thread_ts") == "1234567890.000001", (
"send() must pass thread_ts when metadata has thread_id, "
"ensuring progress messages land in the thread"
)

@pytest.mark.asyncio
async def test_channel_mention_progress_uses_thread_ts(self, adapter):
"""Progress messages for a channel @mention should go into the reply thread."""
# Simulate an @mention in a channel: the event ts becomes the thread anchor
event = {
"channel": "C_CHAN",
"channel_type": "channel",
"user": "U_USER",
"text": f"<@U_BOT> help me",
"ts": "2000000000.000001",
# No thread_ts — top-level channel message
}

captured_events = []
adapter.handle_message = AsyncMock(side_effect=lambda e: captured_events.append(e))

with patch.object(adapter, "_resolve_user_name", new=AsyncMock(return_value="testuser")):
await adapter._handle_slack_message(event)

assert len(captured_events) == 1
msg_event = captured_events[0]
source = msg_event.source

# For channel @mention: thread_id should equal the event ts (fallback)
assert source.thread_id == "2000000000.000001", (
"source.thread_id must equal the event ts for channel messages "
"so each @mention starts its own thread"
)
assert msg_event.message_id == "2000000000.000001"
Loading