diff --git a/cli-config.yaml.example b/cli-config.yaml.example index fb36868ce837..4ec0a24a4045 100644 --- a/cli-config.yaml.example +++ b/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 diff --git a/contributors/emails/simonvanlaak@users.noreply.github.com b/contributors/emails/simonvanlaak@users.noreply.github.com new file mode 100644 index 000000000000..832bfb1d642d --- /dev/null +++ b/contributors/emails/simonvanlaak@users.noreply.github.com @@ -0,0 +1 @@ +simonvanlaak diff --git a/gateway/run.py b/gateway/run.py index 7ff4055ca035..697c3be268dc 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3932,6 +3932,16 @@ def progress_callback(self, event_type: str, tool_name: str = None, preview: str ctx.progress_queue.put(msg) return + # Native task cards consume the authoritative ID-bearing + # tool_start/tool_complete callbacks instead. Do not also enqueue + # name-correlated text events, which would duplicate cards and + # mispair concurrent calls to the same tool. + if ctx._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 ctx.tool_progress_enabled: @@ -4108,6 +4118,188 @@ def progress_callback(self, event_type: str, tool_name: str = None, preview: str ctx.progress_queue.put(msg) + async def _send_native_task_card_progress(self, adapter) -> None: + """Drain the progress queue into Slack-native plan/task cards (#29483). + + Consumes the ID-bearing lifecycle dicts queued by + native_tool_start_callback / native_tool_complete_callback and renders + them through the adapter's chat.startStream plan/task-card stream. + On any native failure, falls back to an editable in-thread text + message so progress stays live for the rest of the turn. + """ + ctx = self._ctx + 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=ctx.source.chat_id, + message_id=fallback_msg_id, + content=text, + metadata=ctx._progress_metadata, + ) + if getattr(result, "success", False): + return + result = await adapter.send( + chat_id=ctx.source.chat_id, + content=text, + reply_to=ctx._progress_reply_to, + metadata=ctx._progress_metadata, + ) + if getattr(result, "success", False) and getattr( + result, "message_id", None + ): + fallback_msg_id = str(result.message_id) + if ctx._cleanup_progress: + ctx._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=ctx.source.chat_id, + tasks=_visible_tasks(), + title="Hermes is working", + reply_to=ctx._progress_reply_to, + metadata=ctx._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( + ctx.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 + + def _agent_interrupted() -> bool: + try: + _agent = ctx.agent_holder[0] if ctx.agent_holder else None + return bool( + _agent is not None and getattr(_agent, "is_interrupted", False) + ) + except Exception: + return False + + try: + while True: + if not ctx._run_still_current(): + return + try: + raw = ctx.progress_queue.get_nowait() + except queue.Empty: + await asyncio.sleep(0.1) + continue + + if _agent_interrupted(): + continue + + if _apply_native_event(raw): + await _publish_native_progress() + except asyncio.CancelledError: + if _drain_native_queue() and ctx._run_still_current(): + if not _agent_interrupted(): + await _publish_native_progress() + return + finally: + if hasattr(adapter, "stop_native_task_card_progress"): + await adapter.stop_native_task_card_progress( + ctx.source.chat_id, + reply_to=ctx._progress_reply_to, + metadata=ctx._progress_metadata, + ) + async def send_progress_messages(self): ctx = self._ctx if not ctx.progress_queue: @@ -4117,6 +4309,12 @@ async def send_progress_messages(self): if not adapter: return + if ctx._native_slack_task_cards and hasattr( + adapter, "send_native_task_card_progress" + ): + await self._send_native_task_card_progress(adapter) + return + # 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. @@ -4484,6 +4682,68 @@ def voice_ack_callback(self, call_id, tool_name, args): except Exception as _ack_err: logger.debug("voice ack schedule failed: %s", _ack_err) + # ── Slack-native task cards: ID-bearing lifecycle callbacks (#29483) ── + # These ride agent.tool_start_callback / agent.tool_complete_callback so + # start/completion events correlate by the REAL tool-call id — the + # name-correlated text events in progress_callback would duplicate cards + # and mispair concurrent calls to the same tool. + + def native_tool_start_callback(self, call_id, tool_name, args): + """Queue an ID-correlated native progress start from the agent thread.""" + ctx = self._ctx + if not ctx.progress_queue or not ctx._run_still_current(): + return + try: + _agent = ctx.agent_holder[0] if ctx.agent_holder else None + if _agent is not None and getattr(_agent, "is_interrupted", False): + return + except Exception: + pass + from agent.display import build_tool_preview + + ctx.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(self, call_id, tool_name, args, result): + """Queue the matching native completion using the real tool-call ID.""" + ctx = self._ctx + if not ctx.progress_queue or not ctx._run_still_current(): + return + try: + _agent = ctx.agent_holder[0] if ctx.agent_holder else None + if _agent is not None and getattr(_agent, "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) + ctx.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(self, call_id, tool_name, args): + """Compose the voice ack + native task-card start consumers.""" + ctx = self._ctx + if ctx._voice_ack_guild[0] is not None: + self.voice_ack_callback(call_id, tool_name, args) + if ctx._native_slack_task_cards: + self.native_tool_start_callback(call_id, tool_name, args) + def _step_callback_sync(self, iteration: int, prev_tools: list) -> None: ctx = self._ctx if not ctx._run_still_current(): @@ -5082,10 +5342,23 @@ 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 ride the authoritative + # start callback, so neither has to infer identity from tool names. + _combined_start_cb = ctx.native_tool_start_callback or ctx.voice_ack_callback agent.tool_start_callback = ( - ctx.voice_ack_callback if ctx._voice_ack_guild[0] is not None else None + _combined_start_cb + if ( + ctx._voice_ack_guild[0] is not None + or ctx._native_slack_task_cards + ) + else None + ) + agent.tool_complete_callback = ( + ctx.native_tool_complete_callback + if ctx._native_slack_task_cards + and ctx.native_tool_complete_callback is not None + else None ) agent.step_callback = ctx._step_callback_sync if ctx._hooks_ref.loaded_hooks else None agent.stream_delta_callback = _stream_delta_cb @@ -26019,7 +26292,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 + # Slack-native task cards (#29483): when the Slack adapter's opt-in + # is set, tool progress renders as native plan/task cards via + # chat.startStream — the progress queue is needed even though Slack + # keeps ordinary text tool_progress off by default (requiring both + # flags would silently leave the native feature inactive). + _progress_adapter_for_native = self._adapter_for_source(source) + _native_slack_task_cards = False + if ( + source.platform == Platform.SLACK + and _progress_adapter_for_native is not None + and hasattr(_progress_adapter_for_native, "native_task_cards_enabled") + ): + try: + _native_slack_task_cards = bool( + _progress_adapter_for_native.native_task_cards_enabled() + ) + except Exception: + logger.debug("Slack native task-card config check failed", exc_info=True) + needs_progress_queue = ( + tool_progress_enabled or _thinking_enabled or _native_slack_task_cards + ) # Queue for progress messages (thread-safe) @@ -26111,6 +26404,7 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview: log_mode_enabled=log_mode_enabled, interim_assistant_messages_enabled=interim_assistant_messages_enabled, needs_progress_queue=needs_progress_queue, + _native_slack_task_cards=_native_slack_task_cards, _voice_ack_fired=_voice_ack_fired, _voice_ack_guild=_voice_ack_guild, _voice_ack_loop=_voice_ack_loop, @@ -26131,6 +26425,10 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview: # TurnRunner.progress_callback (bound method, same signature). turn_ctx.progress_callback = turn_runner.progress_callback turn_ctx.voice_ack_callback = turn_runner.voice_ack_callback + turn_ctx.native_tool_start_callback = turn_runner.combined_tool_start_callback + turn_ctx.native_tool_complete_callback = ( + turn_runner.native_tool_complete_callback + ) # Background task to send progress messages # Accumulates tool lines into a single message that gets edited. @@ -26208,6 +26506,15 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview: # reply anchor; carry it so progress joins that thread. _progress_metadata = {"reply_to_message_id": event_message_id} _progress_metadata = _non_conversational_metadata(_progress_metadata, platform=source.platform) + if _native_slack_task_cards: + # chat.startStream in channels requires the recipient team/user + # pair; harmless extras elsewhere, so stamp them whenever known. + _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 ( diff --git a/gateway/turn_context.py b/gateway/turn_context.py index ac727e2b4f3a..302272909c50 100644 --- a/gateway/turn_context.py +++ b/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 diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index ecc90bb96925..33d047bee9bf 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -294,6 +294,18 @@ def slack_deps_present() -> bool: return SLACK_AVAILABLE +@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. @@ -1017,6 +1029,12 @@ def __init__(self, config: PlatformConfig): # eviction (key[2] is the thread ts). self._active_status_threads: Dict[Tuple[str, str, str], Dict[str, Any]] = {} self._ACTIVE_STATUS_THREADS_MAX = 1000 + # 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() @@ -1028,6 +1046,17 @@ def __init__(self, config: PlatformConfig): # commands that arrived without a workspace id. # Each value: {"response_url": str, "ts": float} self._slash_command_contexts: Dict[Tuple[str, ...], Dict[str, Any]] = {} + # Native streaming (chat.startStream/appendStream/stopStream) state. + # One active stream per chat, keyed by chat_id. Each value: + # {"ts": str, "draft_id": int, "sent": str, "started": float} + # ``sent`` is the raw (pre-mrkdwn) text streamed so far — deltas are + # computed against it because the streaming API is append-only. + self._active_streams: Dict[str, Dict[str, Any]] = {} + # Set after the first startStream failure that indicates the Slack + # app lacks the streaming feature (Agents & AI Apps not enabled / + # missing scope). Future runs then skip straight to edit-based + # streaming without an error round-trip per response. + self._native_stream_unsupported = False # Socket Mode resilience: track runtime connection state so we can # self-heal when Slack silently drops the websocket. self._app_token: Optional[str] = None @@ -2276,6 +2305,12 @@ async def disconnect(self) -> None: """Disconnect from Slack.""" self._running = False + # Seal any dangling native streams so chats aren't left with a + # live-typing indicator across a restart. + for chat_id, stream in list(self._active_streams.items()): + await self._seal_stream(chat_id, stream) + self._active_streams.clear() + watchdog_task = self._socket_watchdog_task self._socket_watchdog_task = None if watchdog_task is not None and not watchdog_task.done(): @@ -2292,6 +2327,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() await self._close_workspace_clients() self._app = None @@ -2447,6 +2488,183 @@ def _is_ignored_channel(self, channel_id: str) -> bool: ignored = self._slack_ignored_channels() return "*" in ignored or parent_channel_id in ignored + @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, @@ -2519,6 +2737,14 @@ async def send( ) return fallback_result + # Native streaming finalization: when this chat has an active + # chat.startStream stream and this send carries its final + # content, seal the stream instead of posting a duplicate + # message (the streamed message IS the final message). + stream_result = await self._try_finalize_stream(chat_id, content) + if stream_result is not None: + return stream_result + # Convert standard markdown → Slack mrkdwn formatted = self.format_message(content) @@ -2857,6 +3083,248 @@ async def delete_message(self, chat_id: str, message_id: str) -> bool: ) return False + # ── Native streaming (chat.startStream / appendStream / stopStream) ── + # + # Slack's Agents & AI Apps feature ships a native streaming surface: the + # bot starts a stream (which renders a live "typing into the message" + # bubble), appends markdown deltas, and stops the stream to finalize. + # Unlike Telegram drafts (ephemeral, replaced by a real sendMessage at the + # end), a Slack stream IS the final message — so ``send()`` intercepts the + # turn-final delivery for a chat with an active stream and seals it via + # chat.stopStream instead of posting a duplicate. + # + # Availability: requires the Slack app to have the AI features enabled. + # When chat.startStream fails with a permission/feature error we cache + # ``_native_stream_unsupported`` and all future runs fall back to the + # edit-based path (the stream consumer handles the per-run fallback on + # the first send_draft failure automatically). + + # Trailing cursor glyph appended by the stream consumer to in-progress + # frames (streaming.cursor, default " ▉"). Stripped before computing + # append deltas because the streaming API is append-only. + _STREAM_CURSOR_GLYPHS = ("\u2589", "▍", "▌", "…") + + def supports_draft_streaming( + self, + chat_type: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> bool: + """Slack native streaming works in DMs, threads, and channels.""" + if self._native_stream_unsupported: + return False + return self._app is not None + + def _strip_stream_cursor(self, text: str) -> str: + """Strip the consumer's trailing cursor glyph from a frame.""" + stripped = text.rstrip() + for glyph in self._STREAM_CURSOR_GLYPHS: + if stripped.endswith(glyph): + return stripped[: -len(glyph)].rstrip() + return text + + async def send_draft( + self, + chat_id: str, + draft_id: int, + content: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Stream a frame via Slack's native streaming APIs. + + First frame for a (chat, draft_id) starts the stream; subsequent + frames append the delta. ``content`` is the full accumulated text + so far (append-only invariant holds because the stream consumer + accumulates monotonically within one text segment). + """ + if not self._app: + return SendResult(success=False, error="Not connected") + if self._native_stream_unsupported: + return SendResult(success=False, error="native streaming unsupported") + + text = self._strip_stream_cursor(content) + client = self._get_client(chat_id) + stream = self._active_streams.get(chat_id) + + try: + if stream is not None and stream.get("draft_id") != draft_id: + # New segment started while a prior stream is open — seal the + # old one so it doesn't hang with a live-typing indicator. + await self._seal_stream(chat_id, stream) + stream = None + + if stream is None: + thread_ts = self._resolve_thread_ts(None, metadata) + if not thread_ts: + # Streamed messages must anchor to a thread_ts. The + # gateway sets metadata.thread_id even for top-level + # messages (the message's own ts), so this is rare. + return SendResult( + success=False, error="no thread_ts for native stream" + ) + start_kwargs: Dict[str, Any] = { + "channel": chat_id, + "thread_ts": thread_ts, + } + # Channels require the recipient team/user pair; harmless + # extras for DMs, so include them whenever known. + md = metadata or {} + user_id = md.get("user_id") or md.get("sender_id") + team_id = self._channel_team.get(chat_id) + if user_id: + start_kwargs["recipient_user_id"] = str(user_id) + if team_id: + start_kwargs["recipient_team_id"] = str(team_id) + if text: + start_kwargs["markdown_text"] = text + response = await client.chat_startStream(**start_kwargs) + ts = response.get("ts") if response else None + if not ts: + raise RuntimeError("chat.startStream returned no ts") + self._active_streams[chat_id] = { + "ts": str(ts), + "draft_id": draft_id, + "sent": text, + "started": time.time(), + } + self._bot_message_ts.add(str(ts)) + return SendResult(success=True, message_id=str(ts)) + + # Append path: compute the delta against what we already sent. + sent = stream.get("sent", "") + if text == sent: + return SendResult(success=True, message_id=stream["ts"]) + if not text.startswith(sent): + # Accumulated text was rewritten (shouldn't happen within a + # segment). Fail the frame so the consumer falls back to the + # edit path; seal the stream first so it doesn't dangle. + await self._seal_stream(chat_id, stream) + self._active_streams.pop(chat_id, None) + return SendResult( + success=False, error="stream prefix mismatch" + ) + delta = text[len(sent):] + await client.chat_appendStream( + channel=chat_id, + ts=stream["ts"], + markdown_text=delta, + ) + stream["sent"] = text + return SendResult(success=True, message_id=stream["ts"]) + + except Exception as e: # pragma: no cover - network/API errors + self._active_streams.pop(chat_id, None) + err = str(e) + # Feature-gate errors: cache unsupported so future runs skip the + # native attempt entirely instead of erroring once per response. + if any( + marker in err + for marker in ( + "not_allowed", + "missing_scope", + "feature_not_enabled", + "invalid_method", + "unknown_method", + "method_deprecated", + "not_authed", + "streaming_not_allowed", + ) + ): + self._native_stream_unsupported = True + logger.warning( + "[Slack] Native streaming unavailable (%s). Falling back " + "to edit-based streaming. To enable native streaming, " + "turn on the Agents & AI Apps feature for this Slack app " + "(and ensure the assistant:write scope).", + err, + ) + else: + logger.debug("[Slack] Native stream frame failed: %s", err) + return SendResult(success=False, error=err) + + async def _seal_stream( + self, + chat_id: str, + stream: Dict[str, Any], + final_text: Optional[str] = None, + blocks: Optional[list] = None, + ) -> bool: + """Best-effort chat.stopStream for an open stream. + + ``final_text`` is the complete final content; only the unsent delta + is passed to stopStream (append-only API). Returns True on success. + """ + try: + kwargs: Dict[str, Any] = { + "channel": chat_id, + "ts": stream["ts"], + } + if final_text is not None: + sent = stream.get("sent", "") + if final_text.startswith(sent) and len(final_text) > len(sent): + kwargs["markdown_text"] = final_text[len(sent):] + if blocks: + kwargs["blocks"] = blocks + await self._get_client(chat_id).chat_stopStream(**kwargs) + return True + except Exception as e: # pragma: no cover - defensive + logger.debug( + "[Slack] chat.stopStream failed for %s/%s: %s", + chat_id, stream.get("ts"), e, + ) + return False + + async def _try_finalize_stream( + self, + chat_id: str, + content: str, + ) -> Optional[SendResult]: + """Finalize an active native stream with the turn-final content. + + Called from ``send()``. Returns a SendResult when the final content + belongs to the active stream (the stream is sealed and the streamed + message IS the final message — no new post needed). Returns None + when the content is unrelated (e.g. interim commentary), leaving the + stream open and letting ``send()`` proceed normally. + """ + stream = self._active_streams.get(chat_id) + if stream is None: + return None + sent = stream.get("sent", "") + text = self._strip_stream_cursor(content) + # Only treat this send as the stream's finalization when it extends + # (or equals) what was streamed. Unrelated sends (e.g. interim + # commentary) pass through. An empty ``sent`` prefix would match + # everything, so require substance before claiming the send. + if not sent or not text.startswith(sent): + return None + self._active_streams.pop(chat_id, None) + ts = stream["ts"] + ok = await self._seal_stream(chat_id, stream, final_text=text) + if not ok: + # Could not stop the stream — post normally so the user still + # gets the final answer; the dangling stream times out on + # Slack's side. + return None + # Final Block Kit pass: streamed messages render markdown natively, + # but the rich block layout (if any) is applied via chat_update on + # the sealed message, mirroring the finalize path in edit_message. + blocks = self._maybe_blocks(text) + if blocks: + try: + await self._get_client(chat_id).chat_update( + channel=chat_id, + ts=ts, + text=self.format_message(text), + blocks=blocks, + ) + except Exception as e: + logger.debug( + "[Slack] Post-stream Block Kit update failed " + "(markdown fallback stands): %s", e, + ) + await self.stop_typing(chat_id) + return SendResult(success=True, message_id=ts) + async def send_typing(self, chat_id: str, metadata=None) -> None: """Show a typing/status indicator using assistant.threads.setStatus. diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py index da3856fb3587..da3dd2efc7bf 100644 --- a/tests/gateway/test_run_progress_topics.py +++ b/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.""" diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py index d9acaefaf6b0..2d770c236b0b 100644 --- a/tests/gateway/test_slack.py +++ b/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 == {} diff --git a/tests/gateway/test_slack_native_streaming.py b/tests/gateway/test_slack_native_streaming.py new file mode 100644 index 000000000000..21f66738f09e --- /dev/null +++ b/tests/gateway/test_slack_native_streaming.py @@ -0,0 +1,213 @@ +"""Tests: SlackAdapter native streaming (chat.startStream/appendStream/stopStream). + +Behaviour contract: + * supports_draft_streaming: True when connected, False after a cached + feature-gate failure or when disconnected. + * send_draft first frame: chat_startStream with thread_ts + initial text; + returns the stream ts as message_id. + * send_draft subsequent frames: chat_appendStream with only the delta; + trailing cursor glyph stripped before delta computation. + * identical frame: no API call, success. + * prefix mismatch: stream sealed, frame fails (consumer falls back to edits). + * send() finalization: active stream sealed via chat_stopStream with the + remaining delta instead of chat_postMessage (no duplicate message). + * send() with unrelated content: stream left open, normal post proceeds. + * startStream feature-gate error: caches _native_stream_unsupported so + future supports_draft_streaming() returns False. + * disconnect(): dangling streams sealed. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.slack.adapter import SlackAdapter + + +def _make_adapter(extra=None): + config = PlatformConfig(enabled=True, token="xoxb-fake", extra=extra or {}) + a = SlackAdapter(config) + a._app = MagicMock() + client = AsyncMock() + client.chat_postMessage = AsyncMock(return_value={"ts": "999.111"}) + client.chat_update = AsyncMock(return_value={"ts": "999.111"}) + client.chat_startStream = AsyncMock(return_value={"ok": True, "ts": "123.456"}) + client.chat_appendStream = AsyncMock(return_value={"ok": True}) + client.chat_stopStream = AsyncMock(return_value={"ok": True}) + a._get_client = MagicMock(return_value=client) + a.stop_typing = AsyncMock() + a._running = True + return a, client + + +META = {"thread_id": "111.000", "user_id": "U123"} + + +class TestSupportsDraftStreaming: + def test_supported_when_connected(self): + adapter, _ = _make_adapter() + assert adapter.supports_draft_streaming(chat_type="dm") is True + + def test_unsupported_when_disconnected(self): + adapter, _ = _make_adapter() + adapter._app = None + assert adapter.supports_draft_streaming() is False + + def test_unsupported_after_feature_gate_failure(self): + adapter, _ = _make_adapter() + adapter._native_stream_unsupported = True + assert adapter.supports_draft_streaming() is False + + +class TestSendDraft: + @pytest.mark.asyncio + async def test_first_frame_starts_stream(self): + adapter, client = _make_adapter() + result = await adapter.send_draft("D1", 7, "Hello wo", metadata=META) + assert result.success + assert result.message_id == "123.456" + kwargs = client.chat_startStream.await_args.kwargs + assert kwargs["channel"] == "D1" + assert kwargs["thread_ts"] == "111.000" + assert kwargs["markdown_text"] == "Hello wo" + assert kwargs["recipient_user_id"] == "U123" + client.chat_appendStream.assert_not_awaited() + + @pytest.mark.asyncio + async def test_subsequent_frame_appends_delta_only(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello wo", metadata=META) + result = await adapter.send_draft("D1", 7, "Hello world!", metadata=META) + assert result.success + kwargs = client.chat_appendStream.await_args.kwargs + assert kwargs["markdown_text"] == "rld!" + assert kwargs["ts"] == "123.456" + + @pytest.mark.asyncio + async def test_cursor_glyph_stripped(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello \u2589", metadata=META) + assert client.chat_startStream.await_args.kwargs["markdown_text"] == "Hello" + await adapter.send_draft("D1", 7, "Hello world \u2589", metadata=META) + assert client.chat_appendStream.await_args.kwargs["markdown_text"] == " world" + + @pytest.mark.asyncio + async def test_identical_frame_is_noop(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello", metadata=META) + result = await adapter.send_draft("D1", 7, "Hello \u2589", metadata=META) + assert result.success + client.chat_appendStream.assert_not_awaited() + + @pytest.mark.asyncio + async def test_prefix_mismatch_seals_and_fails(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello", metadata=META) + result = await adapter.send_draft("D1", 7, "Rewritten text", metadata=META) + assert not result.success + client.chat_stopStream.assert_awaited() + assert "D1" not in adapter._active_streams + + @pytest.mark.asyncio + async def test_no_thread_ts_fails_cleanly(self): + adapter, client = _make_adapter() + result = await adapter.send_draft("D1", 7, "Hello", metadata={}) + assert not result.success + client.chat_startStream.assert_not_awaited() + + @pytest.mark.asyncio + async def test_new_draft_id_seals_prior_stream(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Segment one", metadata=META) + client.chat_startStream.return_value = {"ok": True, "ts": "124.000"} + result = await adapter.send_draft("D1", 8, "Segment two", metadata=META) + assert result.success + client.chat_stopStream.assert_awaited() # sealed segment one + assert adapter._active_streams["D1"]["ts"] == "124.000" + + +class TestFeatureGateFallback: + @pytest.mark.asyncio + async def test_not_allowed_caches_unsupported(self): + adapter, client = _make_adapter() + client.chat_startStream = AsyncMock( + side_effect=Exception("The request to the Slack API failed. (not_allowed)") + ) + result = await adapter.send_draft("D1", 7, "Hello", metadata=META) + assert not result.success + assert adapter._native_stream_unsupported is True + assert adapter.supports_draft_streaming() is False + + @pytest.mark.asyncio + async def test_transient_error_does_not_cache(self): + adapter, client = _make_adapter() + client.chat_startStream = AsyncMock(side_effect=Exception("timeout")) + result = await adapter.send_draft("D1", 7, "Hello", metadata=META) + assert not result.success + assert adapter._native_stream_unsupported is False + + +class TestSendFinalization: + @pytest.mark.asyncio + async def test_final_send_seals_stream_no_duplicate_post(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello wo", metadata=META) + result = await adapter.send("D1", "Hello world, done.", metadata=META) + assert result.success + assert result.message_id == "123.456" + kwargs = client.chat_stopStream.await_args.kwargs + assert kwargs["markdown_text"] == "rld, done." + client.chat_postMessage.assert_not_awaited() + assert "D1" not in adapter._active_streams + + @pytest.mark.asyncio + async def test_final_send_equal_content_seals_without_delta(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello world", metadata=META) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + kwargs = client.chat_stopStream.await_args.kwargs + assert "markdown_text" not in kwargs + client.chat_postMessage.assert_not_awaited() + + @pytest.mark.asyncio + async def test_unrelated_send_passes_through(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Streaming text here", metadata=META) + result = await adapter.send("D1", "Unrelated notice", metadata=META) + assert result.success + client.chat_postMessage.assert_awaited() + # Stream stays open for its own finalization. + assert "D1" in adapter._active_streams + + @pytest.mark.asyncio + async def test_stop_stream_failure_falls_back_to_post(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello", metadata=META) + client.chat_stopStream = AsyncMock(side_effect=Exception("boom")) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + client.chat_postMessage.assert_awaited() + + @pytest.mark.asyncio + async def test_rich_blocks_applied_after_seal(self): + adapter, client = _make_adapter({"rich_blocks": True}) + rich = "# Title\n\nbody text" + await adapter.send_draft("D1", 7, rich[:5], metadata=META) + result = await adapter.send("D1", rich, metadata=META) + assert result.success + client.chat_update.assert_awaited() + assert client.chat_update.await_args.kwargs["blocks"] + + +class TestDisconnectCleanup: + @pytest.mark.asyncio + async def test_disconnect_seals_dangling_streams(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Dangling", metadata=META) + adapter._stop_socket_mode_handler = AsyncMock() + adapter._release_platform_lock = MagicMock() + await adapter.disconnect() + client.chat_stopStream.assert_awaited() + assert not adapter._active_streams diff --git a/website/docs/user-guide/messaging/slack.md b/website/docs/user-guide/messaging/slack.md index 683d1a823c7d..544ed727e914 100644 --- a/website/docs/user-guide/messaging/slack.md +++ b/website/docs/user-guide/messaging/slack.md @@ -423,6 +423,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: "..."}]} @@ -454,6 +460,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.allow_bots` | `"none"` | Controls messages from other Slack bots: `"none"` ignores them, `"mentions"` accepts a bot message only when **that message itself** @mentions Hermes, and `"all"` accepts all of them. Use `"mentions"` for the safest bot-to-bot collaboration mode. See [Accepting messages from other bots](#accepting-messages-from-other-bots-allow_bots). | @@ -518,6 +525,54 @@ display: |-----|---------|-------------| | `display.live_status` | `"full"` | Live per-tool status line. `full` shows verb + argument preview; `verb` shows the verb only (keeps file paths and commands out of shared channels); `off` restores the static text. Requires the `assistant:write` scope, same as the static status line. | +### Native Streaming (live-typing replies) + +Slack's [Agents & AI Apps](https://docs.slack.dev/ai/) feature ships a native +streaming surface (`chat.startStream` / `chat.appendStream` / +`chat.stopStream`) that renders the reply as a live-typing message — much +smoother than the edit-based progressive updates used otherwise. When +`streaming.enabled` is on (transport `auto` or `draft`), Hermes uses native +streaming automatically wherever it's available: + +- The stream starts on the first frame and appends only deltas (the API is + append-only). The streamed message **is** the final message — Hermes seals + it via `chat.stopStream` instead of posting a duplicate final reply. +- If your Slack app doesn't have the AI features enabled (or lacks the + `assistant:write` scope), the first failure is cached and Hermes falls back + to edit-based streaming with a single log warning naming the fix. +- Opt-in Block Kit (`rich_blocks: true`) is applied to the sealed message, + same as the edit-based finalize path. + +No extra configuration is needed beyond enabling streaming: + +```yaml +streaming: + enabled: true # transport auto/draft lights up Slack native streaming +``` + +### Native Task Cards (live tool progress) + +With `platforms.slack.extra.native_task_cards: true`, live tool calls render +as Slack-native **plan/task cards** (the same UI Slack's own AI features use) +instead of text progress bubbles: one card per turn, one row per tool call, +with per-task running/complete/error states updating in place. + +```yaml +platforms: + slack: + extra: + native_task_cards: true +``` + +- This is an explicit progress opt-in — it works even though Slack's default + is `tool_progress: off` (text bubbles spam channels; native cards don't). +- Concurrent calls to the same tool are correlated by real tool-call ID, so + parallel `web_search` calls each get their own row with the right status. +- If the native stream can't start or update, Hermes falls back to a single + continuously edited text message so progress stays live for the turn. +- The card stream is stopped exactly once when the turn finalizes, including + on interrupt/disconnect, so no dangling live indicator is left behind. + ### Session Isolation ```yaml