feat(slack): native live cards — chat.startStream reply streaming + opt-in plan/task-card tool progress

cli-config.yaml.example

@@ -1060,6 +1060,11 @@ platform_toolsets:
 #         priority_mode: prepend
 #         priority:
 #           - my_plugin_command
+#   slack:
+#     extra:
+#       # Render live tool calls as Slack-native plan/task cards. This explicit
+#       # opt-in works even though Slack text tool_progress defaults to off.
+#       native_task_cards: false
 #   webhook:
 #     extra:
 #       # Route scripts default to a 30 second timeout. Scripts must live under

contributors/emails/simonvanlaak@users.noreply.github.com

@@ -0,0 +1 @@
+simonvanlaak

gateway/turn_context.py

@@ -129,3 +129,12 @@ class TurnContext:
     _step_callback_sync: Optional[Callable] = None
     _event_callback_sync: Optional[Callable] = None
     _status_callback_sync: Optional[Callable] = None
+
+    # --- Slack-native task-card progress (opt-in; #29483) ------------------
+    # True when the Slack adapter's ``native_task_cards_enabled()`` opt-in is
+    # set for this turn's platform. The ID-bearing lifecycle callbacks are
+    # published by TurnRunner (like voice_ack_callback above) so tool starts
+    # and completions correlate by real tool-call ID instead of tool name.
+    _native_slack_task_cards: bool = False
+    native_tool_start_callback: Optional[Callable] = None
+    native_tool_complete_callback: Optional[Callable] = None

tests/gateway/test_run_progress_topics.py

@@ -229,6 +229,85 @@ def run_conversation(self, message, conversation_history=None, task_id=None):
         }
 
 
+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."""

tests/gateway/test_slack.py

@@ -4559,3 +4559,125 @@ def test_hermes_slack_user_agent_prefix_format(self):
         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 == {}