test(slack): cover native task-card progress — ID correlation, workspace scoping, fallback
Ports the test coverage from PR #29496 onto the salvaged implementation: adapter-level serialization/workspace isolation/disconnect sealing, and gateway-level ID-correlated concurrent duplicate tools plus the editable text fallback when the native stream fails.
This commit is contained in:
Simon van Laak
@@ -229,6 +229,85 @@ class FakeAgent:
|
||||
}
|
||||
|
||||
|
||||
class NativeTaskCardAdapter(ProgressCaptureAdapter):
|
||||
def __init__(self, platform=Platform.SLACK):
|
||||
super().__init__(platform=platform)
|
||||
self.native_updates = []
|
||||
self.native_stops = 0
|
||||
|
||||
def native_task_cards_enabled(self):
|
||||
return True
|
||||
|
||||
async def send_native_task_card_progress(
|
||||
self,
|
||||
chat_id,
|
||||
tasks,
|
||||
*,
|
||||
title,
|
||||
reply_to=None,
|
||||
metadata=None,
|
||||
fallback_text=None,
|
||||
) -> SendResult:
|
||||
self.native_updates.append(
|
||||
{
|
||||
"chat_id": chat_id,
|
||||
"tasks": [dict(task) for task in tasks],
|
||||
"metadata": dict(metadata or {}),
|
||||
"fallback_text": fallback_text,
|
||||
}
|
||||
)
|
||||
return SendResult(success=True, message_id="native-stream-1")
|
||||
|
||||
async def stop_native_task_card_progress(
|
||||
self, chat_id, *, reply_to=None, metadata=None
|
||||
):
|
||||
self.native_stops += 1
|
||||
|
||||
async def edit_message(
|
||||
self, chat_id, message_id, content, *, finalize=False, metadata=None
|
||||
) -> SendResult:
|
||||
self.edits.append(
|
||||
{
|
||||
"chat_id": chat_id,
|
||||
"message_id": message_id,
|
||||
"content": content,
|
||||
"metadata": metadata,
|
||||
}
|
||||
)
|
||||
return SendResult(success=True, message_id=message_id)
|
||||
|
||||
|
||||
class FailingNativeTaskCardAdapter(NativeTaskCardAdapter):
|
||||
async def send_native_task_card_progress(self, *args, **kwargs) -> SendResult:
|
||||
await super().send_native_task_card_progress(*args, **kwargs)
|
||||
return SendResult(success=False, error="native stream unavailable", retryable=True)
|
||||
|
||||
|
||||
class DuplicateNativeToolsAgent:
|
||||
def __init__(self, **kwargs):
|
||||
self.tool_progress_callback = kwargs.get("tool_progress_callback")
|
||||
self.tool_start_callback = kwargs.get("tool_start_callback")
|
||||
self.tool_complete_callback = kwargs.get("tool_complete_callback")
|
||||
self.tools = []
|
||||
|
||||
def run_conversation(self, message, conversation_history=None, task_id=None):
|
||||
self.tool_start_callback("call-a", "web_search", {"query": "alpha"})
|
||||
time.sleep(0.15)
|
||||
self.tool_start_callback("call-b", "web_search", {"query": "beta"})
|
||||
time.sleep(0.15)
|
||||
# Complete the second same-name call first. Correlation by tool name
|
||||
# would incorrectly mark call-a as failed here.
|
||||
self.tool_complete_callback(
|
||||
"call-b", "web_search", {"query": "beta"}, '{"error": "boom"}'
|
||||
)
|
||||
time.sleep(0.15)
|
||||
self.tool_complete_callback(
|
||||
"call-a", "web_search", {"query": "alpha"}, '{"success": true}'
|
||||
)
|
||||
time.sleep(0.15)
|
||||
return {"final_response": "done", "messages": [], "api_calls": 1}
|
||||
|
||||
|
||||
class ThinkingAgent:
|
||||
"""Agent that emits _thinking scratch text (no tool calls).
|
||||
|
||||
@@ -925,6 +1004,8 @@ async def _run_with_agent(
|
||||
chat_type="group",
|
||||
thread_id="17585",
|
||||
adapter_cls=ProgressCaptureAdapter,
|
||||
user_id=None,
|
||||
scope_id=None,
|
||||
):
|
||||
if config_data:
|
||||
import yaml
|
||||
@@ -951,6 +1032,8 @@ async def _run_with_agent(
|
||||
chat_id=chat_id,
|
||||
chat_type=chat_type,
|
||||
thread_id=thread_id,
|
||||
user_id=user_id,
|
||||
scope_id=scope_id,
|
||||
)
|
||||
session_key = f"agent:main:{platform.value}:{chat_type}:{chat_id}"
|
||||
if thread_id:
|
||||
@@ -974,6 +1057,80 @@ async def _run_with_agent(
|
||||
return adapter, result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_slack_native_progress_correlates_concurrent_duplicate_tools_by_id(
|
||||
monkeypatch, tmp_path
|
||||
):
|
||||
adapter, result = await _run_with_agent(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
DuplicateNativeToolsAgent,
|
||||
session_id="sess-native-ids",
|
||||
config_data={
|
||||
"display": {"platforms": {"slack": {"tool_progress": "off"}}}
|
||||
},
|
||||
platform=Platform.SLACK,
|
||||
chat_id="C1",
|
||||
thread_id="thread-1",
|
||||
adapter_cls=NativeTaskCardAdapter,
|
||||
user_id="U1",
|
||||
scope_id="T1",
|
||||
)
|
||||
|
||||
assert result["final_response"] == "done"
|
||||
assert adapter.native_updates
|
||||
second_completed = next(
|
||||
update
|
||||
for update in adapter.native_updates
|
||||
if {task["id"]: task["status"] for task in update["tasks"]}
|
||||
== {"call-a": "in_progress", "call-b": "error"}
|
||||
)
|
||||
assert second_completed["metadata"]["recipient_team_id"] == "T1"
|
||||
assert second_completed["metadata"]["recipient_user_id"] == "U1"
|
||||
assert adapter.native_updates[-1]["tasks"] == [
|
||||
{
|
||||
"id": "call-a",
|
||||
"title": "web_search - alpha",
|
||||
"status": "complete",
|
||||
},
|
||||
{
|
||||
"id": "call-b",
|
||||
"title": "web_search - beta",
|
||||
"status": "error",
|
||||
},
|
||||
]
|
||||
assert adapter.sent == []
|
||||
assert adapter.native_stops == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_slack_native_failure_keeps_editing_one_live_text_fallback(
|
||||
monkeypatch, tmp_path
|
||||
):
|
||||
adapter, result = await _run_with_agent(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
DuplicateNativeToolsAgent,
|
||||
session_id="sess-native-fallback",
|
||||
platform=Platform.SLACK,
|
||||
chat_id="C1",
|
||||
thread_id="thread-1",
|
||||
adapter_cls=FailingNativeTaskCardAdapter,
|
||||
user_id="U1",
|
||||
scope_id="T1",
|
||||
)
|
||||
|
||||
assert result["final_response"] == "done"
|
||||
assert len(adapter.native_updates) == 1
|
||||
assert len(adapter.sent) == 1
|
||||
assert adapter.sent[0]["content"].endswith("web_search - alpha - running")
|
||||
assert len(adapter.edits) >= 2
|
||||
assert {edit["message_id"] for edit in adapter.edits} == {"progress-1"}
|
||||
assert adapter.edits[-1]["content"].endswith("web_search - beta - error")
|
||||
assert "web_search - alpha - complete" in adapter.edits[-1]["content"]
|
||||
assert adapter.native_stops == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_retryable_overflow_edit_keeps_editable_bubble_identity(monkeypatch, tmp_path):
|
||||
"""A transient split edit must retain can_edit and the current message ID."""
|
||||
|
||||
@@ -4559,3 +4559,125 @@ class TestSlackUserAgent:
|
||||
elsewhere in the codebase for platform-partner attribution."""
|
||||
assert _slack_mod._HERMES_SLACK_USER_AGENT_PREFIX.startswith("HermesAgent/")
|
||||
|
||||
|
||||
class TestNativeTaskCardProgress:
|
||||
def test_native_flag_is_an_explicit_opt_in(self):
|
||||
config = PlatformConfig(
|
||||
enabled=True,
|
||||
token="xoxb-fake-token",
|
||||
extra={"native_task_cards": "true"},
|
||||
)
|
||||
|
||||
assert SlackAdapter(config).native_task_cards_enabled() is True
|
||||
assert SlackAdapter(
|
||||
PlatformConfig(enabled=True, token="xoxb-fake-token")
|
||||
).native_task_cards_enabled() is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_native_updates_are_serialized_and_workspace_scoped(self, adapter):
|
||||
team_client = AsyncMock()
|
||||
start_count = 0
|
||||
|
||||
async def api_call(method, *, json):
|
||||
nonlocal start_count
|
||||
if method == "chat.startStream":
|
||||
start_count += 1
|
||||
await asyncio.sleep(0)
|
||||
return {"ts": "stream-1"}
|
||||
return {"ok": True}
|
||||
|
||||
team_client.api_call.side_effect = api_call
|
||||
adapter._team_clients["T1"] = team_client
|
||||
metadata = {
|
||||
"thread_id": "thread-1",
|
||||
"slack_team_id": "T1",
|
||||
"recipient_team_id": "T1",
|
||||
"recipient_user_id": "U1",
|
||||
}
|
||||
first = [{"id": "call-1", "title": "terminal", "status": "in_progress"}]
|
||||
second = [{"id": "call-1", "title": "terminal", "status": "complete"}]
|
||||
|
||||
results = await asyncio.gather(
|
||||
adapter.send_native_task_card_progress("C1", first, metadata=metadata),
|
||||
adapter.send_native_task_card_progress("C1", second, metadata=metadata),
|
||||
)
|
||||
|
||||
assert all(result.success for result in results)
|
||||
assert start_count == 1
|
||||
calls = team_client.api_call.await_args_list
|
||||
assert [call.args[0] for call in calls] == [
|
||||
"chat.startStream",
|
||||
"chat.appendStream",
|
||||
"chat.appendStream",
|
||||
]
|
||||
assert calls[0].kwargs["json"] == {
|
||||
"channel": "C1",
|
||||
"thread_ts": "thread-1",
|
||||
"task_display_mode": "plan",
|
||||
"recipient_team_id": "T1",
|
||||
"recipient_user_id": "U1",
|
||||
}
|
||||
adapter._app.client.api_call.assert_not_awaited()
|
||||
|
||||
await adapter.stop_native_task_card_progress("C1", metadata=metadata)
|
||||
|
||||
assert team_client.api_call.await_args.args[0] == "chat.stopStream"
|
||||
assert adapter._native_task_card_streams == {}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_same_channel_thread_isolated_between_workspaces(self, adapter):
|
||||
clients = {"T1": AsyncMock(), "T2": AsyncMock()}
|
||||
|
||||
def api_call_for(team_id):
|
||||
async def api_call(method, *, json):
|
||||
if method == "chat.startStream":
|
||||
return {"ts": f"stream-{team_id}"}
|
||||
return {"ok": True}
|
||||
|
||||
return api_call
|
||||
|
||||
for team_id, client in clients.items():
|
||||
client.api_call.side_effect = api_call_for(team_id)
|
||||
adapter._team_clients.update(clients)
|
||||
tasks = [{"id": "call-1", "title": "search", "status": "in_progress"}]
|
||||
|
||||
await asyncio.gather(
|
||||
*(
|
||||
adapter.send_native_task_card_progress(
|
||||
"C-shared",
|
||||
tasks,
|
||||
metadata={"thread_id": "thread-shared", "slack_team_id": team_id},
|
||||
)
|
||||
for team_id in clients
|
||||
)
|
||||
)
|
||||
|
||||
assert set(adapter._native_task_card_streams) == {
|
||||
("T1", "C-shared", "thread-shared"),
|
||||
("T2", "C-shared", "thread-shared"),
|
||||
}
|
||||
for client in clients.values():
|
||||
assert client.api_call.await_args_list[0].args[0] == "chat.startStream"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_disconnect_stops_active_native_streams(self, adapter):
|
||||
client = adapter._app.client
|
||||
client.api_call.side_effect = [
|
||||
{"ts": "stream-1"},
|
||||
{"ok": True},
|
||||
{"ok": True},
|
||||
]
|
||||
await adapter.send_native_task_card_progress(
|
||||
"C1",
|
||||
[{"id": "call-1", "title": "terminal", "status": "in_progress"}],
|
||||
metadata={"thread_id": "thread-1"},
|
||||
)
|
||||
|
||||
await adapter.disconnect()
|
||||
|
||||
assert [call.args[0] for call in client.api_call.await_args_list] == [
|
||||
"chat.startStream",
|
||||
"chat.appendStream",
|
||||
"chat.stopStream",
|
||||
]
|
||||
assert adapter._native_task_card_streams == {}
|
||||
|
||||
Reference in New Issue
Block a user