diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 5e121904ca4a..a14a7cdea1d5 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -877,6 +877,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 diff --git a/gateway/run.py b/gateway/run.py index 01fddf2dfaaa..3f74c141ed66 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -19112,7 +19112,27 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview: require_platform_override_for={Platform.MATTERMOST}, ) _thinking_enabled = _thinking_mode != "off" - needs_progress_queue = tool_progress_enabled or _thinking_enabled + _progress_adapter = self._adapter_for_source(source) + _native_slack_task_cards = False + if ( + source.platform == Platform.SLACK + and _progress_adapter is not None + and hasattr(_progress_adapter, "native_task_cards_enabled") + ): + try: + _native_slack_task_cards = bool( + _progress_adapter.native_task_cards_enabled() + ) + except Exception: + logger.debug("Slack native task-card config check failed", exc_info=True) + # The native flag is itself an explicit progress opt-in. Slack keeps + # ordinary text tool_progress off by default, so requiring both flags + # would silently leave the native feature inactive. + needs_progress_queue = ( + tool_progress_enabled + or _thinking_enabled + or _native_slack_task_cards + ) # Queue for progress messages (thread-safe) @@ -19272,6 +19292,16 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non progress_queue.put(msg) return + # Native progress consumes the authoritative ID-bearing + # tool_start/tool_complete callbacks below. Do not also enqueue + # name-correlated text events, which would duplicate cards and + # mispair concurrent calls to the same tool. + if _native_slack_task_cards and event_type in { + "tool.started", + "tool.completed", + }: + return + # If tool_progress is off, only _thinking passes through (above). # Regular tool calls are suppressed. if not tool_progress_enabled: @@ -19447,12 +19477,75 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non else {"thread_id": _progress_thread_id} ) if _progress_thread_id else None _progress_metadata = _non_conversational_metadata(_progress_metadata, platform=source.platform) + if _native_slack_task_cards: + _progress_metadata = dict(_progress_metadata or {}) + if source.scope_id: + _progress_metadata.setdefault("recipient_team_id", source.scope_id) + _progress_metadata.setdefault("slack_team_id", source.scope_id) + if source.user_id: + _progress_metadata.setdefault("recipient_user_id", source.user_id) _progress_reply_to = ( event_message_id if source.platform in (Platform.FEISHU, Platform.MATTERMOST) and source.thread_id and event_message_id else None ) + def _native_tool_start_callback(call_id, tool_name, args): + """Queue an ID-correlated native progress start from the agent thread.""" + if not progress_queue or not _run_still_current(): + return + try: + _agent_for_interrupt = agent_holder[0] if agent_holder else None + if _agent_for_interrupt is not None and getattr( + _agent_for_interrupt, "is_interrupted", False + ): + return + except Exception: + pass + from agent.display import build_tool_preview + + progress_queue.put( + { + "type": "tool.started", + "tool_call_id": str(call_id or ""), + "tool_name": str(tool_name or "tool"), + "preview": build_tool_preview( + str(tool_name or "tool"), args or {}, max_len=64 + ) + or "", + } + ) + + def _native_tool_complete_callback(call_id, tool_name, args, result): + """Queue the matching native completion using the real tool-call ID.""" + if not progress_queue or not _run_still_current(): + return + try: + _agent_for_interrupt = agent_holder[0] if agent_holder else None + if _agent_for_interrupt is not None and getattr( + _agent_for_interrupt, "is_interrupted", False + ): + return + except Exception: + pass + from agent.display import _detect_tool_failure + + is_error, _ = _detect_tool_failure(str(tool_name or "tool"), result) + progress_queue.put( + { + "type": "tool.completed", + "tool_call_id": str(call_id or ""), + "tool_name": str(tool_name or "tool"), + "is_error": bool(is_error), + } + ) + + def _combined_tool_start_callback(call_id, tool_name, args): + if _voice_ack_guild[0] is not None: + voice_ack_callback(call_id, tool_name, args) + if _native_slack_task_cards: + _native_tool_start_callback(call_id, tool_name, args) + async def write_tool_log(): """Drain log_queue and append tool-call lines to tool_calls.log. @@ -19516,6 +19609,187 @@ async def send_progress_messages(): if not adapter: return + if ( + _native_slack_task_cards + and hasattr(adapter, "send_native_task_card_progress") + ): + tasks: Dict[str, Dict[str, str]] = {} + task_order: List[str] = [] + fallback_msg_id: Optional[str] = None + native_failed = False + anonymous_seq = 0 + + def _compact(value: Any, limit: int = 120) -> str: + text = re.sub(r"\s+", " ", str(value or "")).strip() + if len(text) <= limit: + return text + return text[: limit - 3].rstrip() + "..." + + def _visible_tasks() -> List[Dict[str, str]]: + return [tasks[task_id] for task_id in task_order[-8:]] + + def _fallback_text() -> str: + labels = { + "in_progress": "running", + "complete": "complete", + "error": "error", + } + lines = [ + f"- {task['title']} - {labels.get(task['status'], task['status'])}" + for task in _visible_tasks() + ] + return "Hermes is working\n" + "\n".join(lines) + + def _apply_native_event(raw: Any) -> bool: + nonlocal anonymous_seq + if not isinstance(raw, dict): + return False + event_type = raw.get("type") + if event_type not in {"tool.started", "tool.completed"}: + return False + call_id = str(raw.get("tool_call_id") or "") + if not call_id: + anonymous_seq += 1 + call_id = f"anonymous_{anonymous_seq}" + tool_name = str(raw.get("tool_name") or "tool") + + if event_type == "tool.started": + title = tool_name + preview = _compact(raw.get("preview"), 64) + if preview: + title = f"{tool_name} - {preview}" + if call_id not in tasks: + task_order.append(call_id) + tasks[call_id] = { + "id": call_id, + "title": _compact(title), + "status": "in_progress", + } + return True + + task = tasks.get(call_id) + if task is None: + # Completion-only events are rare but valid on some + # runtimes. Keep their real ID instead of guessing a + # same-name pending call. + task = { + "id": call_id, + "title": _compact(tool_name), + "status": "in_progress", + } + tasks[call_id] = task + task_order.append(call_id) + task["status"] = "error" if raw.get("is_error") else "complete" + return True + + async def _send_or_edit_fallback() -> None: + nonlocal fallback_msg_id + text = _fallback_text() + if fallback_msg_id: + result = await adapter.edit_message( + chat_id=source.chat_id, + message_id=fallback_msg_id, + content=text, + metadata=_progress_metadata, + ) + if getattr(result, "success", False): + return + result = await adapter.send( + chat_id=source.chat_id, + content=text, + reply_to=_progress_reply_to, + metadata=_progress_metadata, + ) + if getattr(result, "success", False) and getattr( + result, "message_id", None + ): + fallback_msg_id = str(result.message_id) + if _cleanup_progress: + _cleanup_msg_ids.append(fallback_msg_id) + + async def _publish_native_progress() -> None: + nonlocal native_failed + if not tasks: + return + if not native_failed: + result = await adapter.send_native_task_card_progress( + chat_id=source.chat_id, + tasks=_visible_tasks(), + title="Hermes is working", + reply_to=_progress_reply_to, + metadata=_progress_metadata, + fallback_text=_fallback_text(), + ) + if getattr(result, "success", False): + return + native_failed = True + logger.warning( + "Slack native task-card progress failed; falling back " + "to an editable text update: %s", + getattr(result, "error", "unknown error"), + ) + # Once the native rail fails, every later lifecycle event + # edits the same fallback message so progress remains live. + await _send_or_edit_fallback() + + def _drain_native_queue() -> bool: + changed = False + while True: + try: + changed = _apply_native_event( + progress_queue.get_nowait() + ) or changed + except queue.Empty: + return changed + except Exception: + logger.debug( + "Slack native progress queue drain failed", + exc_info=True, + ) + return changed + + try: + while True: + if not _run_still_current(): + return + try: + raw = progress_queue.get_nowait() + except queue.Empty: + await asyncio.sleep(0.1) + continue + + try: + _agent_for_interrupt = agent_holder[0] if agent_holder else None + if _agent_for_interrupt is not None and getattr( + _agent_for_interrupt, "is_interrupted", False + ): + continue + except Exception: + pass + + if _apply_native_event(raw): + await _publish_native_progress() + except asyncio.CancelledError: + if _drain_native_queue() and _run_still_current(): + try: + _agent_for_interrupt = agent_holder[0] if agent_holder else None + _interrupted = bool( + _agent_for_interrupt is not None + and getattr(_agent_for_interrupt, "is_interrupted", False) + ) + except Exception: + _interrupted = False + if not _interrupted: + await _publish_native_progress() + return + finally: + if hasattr(adapter, "stop_native_task_card_progress"): + await adapter.stop_native_task_card_progress( + source.chat_id, + reply_to=_progress_reply_to, + metadata=_progress_metadata, + ) + # Skip tool progress for platforms that don't support message # editing (e.g. iMessage/BlueBubbles) — each progress update # would become a separate message bubble, which is noisy. @@ -20405,10 +20679,16 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: ) else None ) - # Discord voice verbal-ack hook (fires once per turn on first tool - # call; armed only when in a voice channel with the mixer running). + # Compose ID-bearing lifecycle consumers: Discord's one-time voice + # ack and Slack's native task cards both use the authoritative + # start callback, so neither has to infer identity from tool names. agent.tool_start_callback = ( - voice_ack_callback if _voice_ack_guild[0] is not None else None + _combined_tool_start_callback + if _voice_ack_guild[0] is not None or _native_slack_task_cards + else None + ) + agent.tool_complete_callback = ( + _native_tool_complete_callback if _native_slack_task_cards else None ) agent.step_callback = _step_callback_sync if _hooks_ref.loaded_hooks else None agent.stream_delta_callback = _stream_delta_cb diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index bde89f4ad1af..46bd730bc57d 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -85,6 +85,18 @@ class _ThreadContextCache: parent_text: str = "" # Raw text of the thread parent (for reply_to_text injection) +@dataclass +class _NativeTaskCardStream: + """Serialized state for one workspace-scoped Slack progress stream.""" + + team_id: str + channel: str + thread_ts: str + stream_ts: str = "" + stopped: bool = False + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + def check_slack_requirements() -> bool: """Check if Slack dependencies are available. @@ -486,6 +498,12 @@ def __init__(self, config: PlatformConfig): # Track active Assistant statuses by (team_id, channel_id, thread_ts) # so cleanup cannot clear an overlapping Slack Connect workspace. self._active_status_threads: Dict[Tuple[str, str, str], Dict[str, str]] = {} + # Native progress streams share Slack's workspace/thread isolation. + # Each stream owns a lock so concurrent start/append/stop calls cannot + # race into duplicate streams or append after finalization. + self._native_task_card_streams: Dict[ + Tuple[str, str, str], _NativeTaskCardStream + ] = {} # Best-effort guard so automatic Slack AI thread titles are set once # per visible DM thread instead of on every reply. self._titled_assistant_threads: set = set() @@ -1369,6 +1387,12 @@ async def disconnect(self) -> None: "[Slack] Watchdog task raised during disconnect", exc_info=True ) + # Finalize native streams while workspace clients are still live. The + # gateway normally stops each turn's stream first; this is the shutdown + # safety net for cancellation/reconnect races. + for key, stream in list(self._native_task_card_streams.items()): + await self._stop_native_task_card_stream(key, stream) + await self._stop_socket_mode_handler() self._app = None self._app_token = None @@ -1399,6 +1423,183 @@ def _get_client(self, chat_id: str, team_id: Optional[str] = None) -> Any: return self._team_clients[team_id] return self._app.client # fallback to primary + @staticmethod + def _truthy_config(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes", "on"} + return bool(value) + + def native_task_cards_enabled(self) -> bool: + """Return whether Slack-native tool progress is explicitly enabled.""" + extra = self.config.extra if isinstance(self.config.extra, dict) else {} + direct = extra.get("native_task_cards", extra.get("nativeTaskCards")) + if direct is not None: + return self._truthy_config(direct) + + streaming = extra.get("streaming") + if isinstance(streaming, dict): + progress = streaming.get("progress") + if isinstance(progress, dict): + nested = progress.get( + "native_task_cards", progress.get("nativeTaskCards") + ) + if nested is not None: + return self._truthy_config(nested) + return False + + def _native_task_card_key( + self, + chat_id: str, + reply_to: Optional[str], + metadata: Optional[Dict[str, Any]], + ) -> Optional[Tuple[str, str, str]]: + thread_ts = self._resolve_thread_ts(reply_to, metadata) + if not thread_ts: + return None + return self._workspace_thread_key( + self._metadata_team_id(metadata), chat_id, str(thread_ts) + ) + + async def send_native_task_card_progress( + self, + chat_id: str, + tasks: List[Dict[str, str]], + *, + title: str = "Hermes is working", + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + fallback_text: Optional[str] = None, + ) -> SendResult: + """Start or update a Slack-native plan/task progress stream.""" + if not self._app: + return SendResult(success=False, error="Not connected") + if not tasks: + return SendResult(success=False, error="No tasks") + + key = self._native_task_card_key(chat_id, reply_to, metadata) + if key is None: + return SendResult(success=False, error="No Slack thread target") + + stream = self._native_task_card_streams.get(key) + if stream is None or stream.stopped: + stream = _NativeTaskCardStream( + team_id=key[0], + channel=chat_id, + thread_ts=key[2], + ) + # There is no await between lookup and assignment, so competing + # coroutines on this event loop will observe this same lock. + self._native_task_card_streams[key] = stream + + async with stream.lock: + if stream.stopped: + return SendResult(success=False, error="Progress stream already stopped") + try: + client = self._get_client(chat_id, team_id=stream.team_id) + if not stream.stream_ts: + start_payload: Dict[str, Any] = { + "channel": chat_id, + "thread_ts": stream.thread_ts, + "task_display_mode": "plan", + } + md = metadata or {} + recipient_team_id = ( + md.get("recipient_team_id") + or md.get("team_id") + or md.get("slack_team_id") + ) + recipient_user_id = md.get("recipient_user_id") or md.get("user_id") + if recipient_team_id: + start_payload["recipient_team_id"] = recipient_team_id + if recipient_user_id: + start_payload["recipient_user_id"] = recipient_user_id + + result = await client.api_call( + "chat.startStream", json=start_payload + ) + if hasattr(result, "get"): + stream.stream_ts = str( + result.get("ts") or result.get("message_ts") or "" + ) + if not stream.stream_ts: + raise RuntimeError("Slack startStream returned no stream timestamp") + + chunks: List[Dict[str, Any]] = [ + {"type": "plan_update", "title": str(title)[:256]} + ] + for task in tasks: + status = str(task.get("status") or "in_progress") + if status not in {"in_progress", "complete", "error"}: + status = "in_progress" + task_id = str(task.get("id") or task.get("task_id") or "task") + chunks.append( + { + "type": "task_update", + "id": task_id, + "title": str(task.get("title") or task_id)[:256], + "status": status, + } + ) + + append_payload: Dict[str, Any] = { + "channel": chat_id, + "ts": stream.stream_ts, + "chunks": chunks, + } + if fallback_text: + append_payload["markdown_text"] = fallback_text + await client.api_call("chat.appendStream", json=append_payload) + return SendResult(success=True, message_id=stream.stream_ts) + except Exception as exc: # pragma: no cover - defensive logging + logger.error( + "[Slack] Native task-card progress error: %s", + exc, + exc_info=True, + ) + return SendResult(success=False, error=str(exc), retryable=True) + + async def _stop_native_task_card_stream( + self, + key: Tuple[str, str, str], + stream: _NativeTaskCardStream, + ) -> None: + async with stream.lock: + if stream.stopped: + return + stream.stopped = True + try: + if self._app and stream.stream_ts: + await self._get_client( + stream.channel, team_id=stream.team_id + ).api_call( + "chat.stopStream", + json={"channel": stream.channel, "ts": stream.stream_ts}, + ) + except Exception as exc: # pragma: no cover - defensive logging + logger.debug( + "[Slack] Native task-card stopStream failed: %s", exc + ) + finally: + if self._native_task_card_streams.get(key) is stream: + self._native_task_card_streams.pop(key, None) + + async def stop_native_task_card_progress( + self, + chat_id: str, + *, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> None: + """Finalize an active Slack-native progress stream exactly once.""" + key = self._native_task_card_key(chat_id, reply_to, metadata) + if key is None: + return + stream = self._native_task_card_streams.get(key) + if stream is not None: + await self._stop_native_task_card_stream(key, stream) + async def send( self, chat_id: str, diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py index 9892b38b90d3..d478915bf4c1 100644 --- a/tests/gateway/test_run_progress_topics.py +++ b/tests/gateway/test_run_progress_topics.py @@ -145,6 +145,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). @@ -788,6 +867,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 @@ -814,6 +895,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: @@ -837,6 +920,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_run_agent_rolls_progress_bubble_before_platform_limit(monkeypatch, tmp_path): """Tool progress should start a second editable bubble before Telegram's limit. diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py index dd61bd95f5bb..041346ee8df7 100644 --- a/tests/gateway/test_slack.py +++ b/tests/gateway/test_slack.py @@ -4995,3 +4995,126 @@ async def test_auth_check_exception_does_not_crash_fetch(self, adapter): # Renders successfully without trust tag (exception → unknown trust). assert "U_X: hello" in content assert "[unverified]" not in content + + +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 == {} diff --git a/website/docs/user-guide/messaging/slack.md b/website/docs/user-guide/messaging/slack.md index f6b7d585a51d..ab66faea6de9 100644 --- a/website/docs/user-guide/messaging/slack.md +++ b/website/docs/user-guide/messaging/slack.md @@ -382,6 +382,12 @@ platforms: # Requires rich_blocks: true. Default: false. feedback_buttons: false + # Render live tool calls as Slack-native plan/task cards. This explicit + # opt-in activates native progress even when text tool_progress is off. + # If Slack rejects the native stream, Hermes keeps one editable text + # fallback current for the rest of the turn. + native_task_cards: false + # Suggested prompts pinned at the top of Agent view's Messages tab. # Either a list of {title, message} rows, or a titled object: # {title: "Start here", prompts: [{title: "Plan", message: "..."}]} @@ -406,6 +412,7 @@ platforms: | `platforms.slack.extra.reply_broadcast` | `false` | When `true`, thread replies are also posted to the main channel. Only the first chunk is broadcast. | | `platforms.slack.extra.rich_blocks` | `false` | When `true`, agent messages are rendered as [Block Kit](https://docs.slack.dev/block-kit/) blocks (headers, dividers, true nested lists, and native tables). A plain-text fallback is always sent. Tables over Slack's limits fall back to aligned monospace. No app reinstall required — it's a send-side change only. | | `platforms.slack.extra.feedback_buttons` | `false` | When `true` with `rich_blocks`, appends Slack-native feedback controls to final replies. | +| `platforms.slack.extra.native_task_cards` | `false` | When `true`, renders live tool calls as Slack-native plan/task cards. This is an explicit progress opt-in independent of Slack's default `tool_progress: off`; native API failures fall back to one continuously edited text update. | | `platforms.slack.extra.suggested_prompts` | `[]` | Up to four `{title, message}` prompts for Agent/Assistant DM entry points; accepts either a list or `{title, prompts}`. | | `platforms.slack.extra.assistant_thread_titles` | `true` | When `true`, names Agent/Assistant DM threads from the first user message. | | `platforms.slack.extra.cron_continuable_surface` | `"thread"` | Delivery surface for [continuable cron jobs](../features/cron.md#flat-in-channel-continuation-slack). `"thread"` opens a dedicated thread per delivery (default); `"in_channel"` delivers flat into the channel timeline. Pair `in_channel` with `reply_in_thread: false` (and `require_mention: false`) so a plain channel reply continues the job. |