From 72893ca6a1e64f3703901476fe479f0a0418ab8e Mon Sep 17 00:00:00 2001 From: ms-elbdev <97030771+ms-elbdev@users.noreply.github.com> Date: Mon, 21 Sep 2026 16:30:04 +0200 Subject: [PATCH] fix(slack): never re-post a successfully streamed answer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem: with native streaming (chat.startStream/appendStream/stopStream) the same answer could land twice in a thread — once as the streamed message, once as a fresh chat.postMessage — while the streamed message kept its live-typing indicator. Mechanism: `_try_finalize_stream` matched the turn-final against the streamed text with a raw `startswith`. The agent strips `final_response` and joins footers with `rstrip()`, so any surrounding-whitespace difference made the finalize fall through to a plain post although the open stream already showed the whole answer (and was never sealed). A `chat.stopStream` failure took the same fresh-post path even when the streamed text equalled the final. Streams were also keyed per `chat_id` only, so two concurrent turns in two threads of one channel sealed or overwrote each other's stream. Fix: - Key native streams per `(team_id, chat_id, thread_ts)`; the stream consumer stamps the same `thread_id` on every draft frame and on the turn-final `send()`, so both resolve to the same key. - Honor the streaming contract (gateway/AGENTS.md): sends carrying `_interim_send` or `expect_edits` never seal a stream. - Classify the final against the streamed text as equal / extends / unrelated with edge-whitespace tolerance (`_stream_relation`). The stopStream delta is sliced from the RAW final, so nothing inside the answer (blank lines, fences, tables) is dropped or repeated. - Commit rule: one `chat.stopStream`, one retry only when no tail is appended (`markdown_text` APPENDS, so an ambiguous failure must not repeat it), then `edit_message(finalize=True)` on the stream ts as the idempotent in-place commit — it already owns format/truncate/Block Kit and the block-rejection retry. Only when both fail does `send()` post a fresh message (a duplicate beats a lost answer). - Oversized tails and rewritten finals (`notify=True`) seal the stale stream on what is visible before falling back, so no stream is left with a live-typing indicator. - `_seal_stream` takes the exact unsent delta instead of recomputing it from `final_text`; `disconnect()` and the stream API calls route through the stream's own team client. Tests: tests/gateway/test_slack_native_streaming.py covers the whitespace-only difference, the stopStream-failure commit path, the bounded retry, the uncommittable fallback, interim/preview sends, per-thread keying, oversized tails, rewritten finals and the GatewayStreamConsumer end-to-end path. (cherry picked from commit a64d10071ce7816b124467e29407d7d47bfdde8d) --- plugins/platforms/slack/adapter.py | 158 +++++--- .../test_native_stream_warning_composition.py | 6 +- tests/gateway/test_slack_native_streaming.py | 383 +++++++++++++++++- 3 files changed, 482 insertions(+), 65 deletions(-) diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index b1382acd48..3ade2bad86 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -1116,7 +1116,7 @@ class SlackAdapter(BasePlatformAdapter): self._slash_command_contexts: Dict[Tuple[str, ...], Dict[str, Any]] = {} # Native streaming state per chat_id: {"ts", "draft_id", "sent", "started"}. # ``sent`` is raw pre-mrkdwn text; the API is append-only so deltas diff against it. - self._active_streams: Dict[str, Dict[str, Any]] = {} + self._active_streams: Dict[Tuple[str, str, str], Dict[str, Any]] = {} # see _stream_key # Set once startStream reports the app lacks streaming (Agents & AI Apps # off / missing scope); later responses skip straight to edit-based streaming. self._native_stream_unsupported = False @@ -1899,8 +1899,8 @@ class SlackAdapter(BasePlatformAdapter): """Disconnect from Slack.""" self._running = False # Seal dangling native streams so no live-typing indicator survives a restart. - for chat_id, stream in list(self._active_streams.items()): - await self._seal_stream(chat_id, stream) + for stream in list(self._active_streams.values()): + await self._seal_stream(str(stream.get("chat_id", "")), stream) self._active_streams.clear() # A watchdog that lost the cancel race must not block cleanup/lock release. await self._cancel_socket_watchdog("[Slack] Watchdog task raised during disconnect") @@ -2230,7 +2230,7 @@ class SlackAdapter(BasePlatformAdapter): return await self._send_slash_reply(chat_id, slash_ctx, content, metadata) # An active native stream that this content finalizes IS the final # message: seal it instead of posting a duplicate. - stream_result = await self._try_finalize_stream(chat_id, content) + stream_result = await self._try_finalize_stream(chat_id, content, metadata) if stream_result is not None: return stream_result formatted = self.format_message(content) @@ -2480,13 +2480,15 @@ class SlackAdapter(BasePlatformAdapter): 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) + stream_key = self._stream_key(chat_id, metadata) + client = self._get_client(chat_id, team_id=stream_key[0] or None) + stream = self._active_streams.get(stream_key) try: if stream is not None and stream.get("draft_id") != draft_id: # New segment 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) + self._active_streams.pop(stream_key, None) stream = None if stream is None: return await self._start_stream(client, chat_id, draft_id, text, metadata) @@ -2497,14 +2499,14 @@ class SlackAdapter(BasePlatformAdapter): # Text was rewritten mid-segment: seal the stream, then fail # the frame so the consumer falls back to the edit path. await self._seal_stream(chat_id, stream) - self._active_streams.pop(chat_id, None) + self._active_streams.pop(stream_key, 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) + self._active_streams.pop(stream_key, None) err = str(e) # Feature-gate errors: remember unsupported so later responses # skip the native attempt instead of erroring each time. @@ -2530,7 +2532,8 @@ class SlackAdapter(BasePlatformAdapter): start_kwargs: Dict[str, Any] = {"channel": chat_id, "thread_ts": thread_ts} md = metadata or {} user_id = md.get("user_id") or md.get("sender_id") - team_id = self._channel_team.get(chat_id) + stream_key = self._stream_key(chat_id, metadata) + team_id = stream_key[0] or self._channel_team.get(chat_id) if user_id: start_kwargs["recipient_user_id"] = str(user_id) if team_id: @@ -2541,61 +2544,124 @@ class SlackAdapter(BasePlatformAdapter): 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._active_streams[stream_key] = { + "ts": str(ts), "draft_id": draft_id, "sent": text, "started": time.time(), + "chat_id": str(chat_id), "team_id": stream_key[0]} self._bot_message_ts.add(str(ts)) return SendResult(success=True, message_id=str(ts)) + def _stream_key( + self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> Tuple[str, str, str]: + """Identity of a native stream: ``(team_id, chat_id, thread_ts)``. The stream consumer + stamps the same ``thread_id`` metadata on every draft frame AND on the turn-final + ``send()``, so both resolve to the same key; keying per thread keeps concurrent turns in + two threads of one channel from sealing or overwriting each other's stream.""" + md = metadata or {} + team_id = self._metadata_team_id(md) or self._channel_team.get(chat_id, "") or "" + return (str(team_id), str(chat_id), str(self._resolve_thread_ts(None, md) or "")) + + @staticmethod + def _stream_relation(sent: str, text: str) -> Tuple[str, str]: + """Classify turn-final ``text`` against the streamed ``sent``: ``("equal", "")`` — the + streamed message already shows the whole final; ``("extends", delta)`` — the final + continues it, ``delta`` being the exact raw tail still to append; ``("unrelated", "")`` — + not this stream's final (interim commentary, a rewritten answer, an empty stream). + + The agent strips its final response and appends footers with ``rstrip()``, so the final + may differ from the streamed frames by surrounding whitespace only. The delta is sliced + from the RAW ``text`` where the streamed core ends — never from a normalized copy — so + nothing inside the answer (blank lines, fences, tables) is dropped or repeated.""" + core = sent.strip() + if not core: + return "unrelated", "" + if text.startswith(sent): + delta = text[len(sent):] + return ("extends" if delta else "equal"), delta + lead = len(text) - len(text.lstrip()) + if text[lead:].startswith(core): + delta = text[lead + len(core):] + return ("extends" if delta.strip() else "equal"), delta + return "unrelated", "" + 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.""" + self, chat_id: str, stream: Dict[str, Any], delta: Optional[str] = None) -> bool: + """Best-effort chat.stopStream for an open stream. ``delta`` is the exact unsent tail of + the final content (the API is append-only: ``stopStream.markdown_text`` APPENDS). 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) + if delta and delta.strip(): + kwargs["markdown_text"] = delta + await self._get_client(chat_id, team_id=stream.get("team_id") or None).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]: + async def _try_finalize_stream( + self, chat_id: str, content: str, metadata: Optional[Dict[str, Any]] = None + ) -> Optional[SendResult]: """Seal the active native stream if ``content`` is its final text: SendResult when the - stream IS the final message; None when unrelated (interim commentary), leaving it open.""" - stream = self._active_streams.get(chat_id) + stream IS the final message; None when unrelated (interim commentary), leaving it open. + + Invariant (duplicate-final class): a successfully streamed answer is NEVER posted a + second time. A fresh post happens only when the streamed message is demonstrably + uncommittable — the seal failed AND an in-place ``chat.update`` of it failed — or when + the content is unrelated to the stream.""" + md = metadata or {} + # Streaming contract: interim sends (commentary, segment-tail flushes) and editable + # previews are never the turn final, whatever their text looks like. + if md.get("_interim_send") or md.get("expect_edits"): + return None + key = self._stream_key(chat_id, md) + stream = self._active_streams.get(key) if stream is None: return None - sent = stream.get("sent", "") text = self._strip_stream_cursor(content) - # Only claim sends that extend what was streamed; an empty ``sent`` - # prefix would match everything. - if not sent or not text.startswith(sent): + kind, delta = self._stream_relation(stream.get("sent", ""), text) + if kind == "unrelated": + if md.get("notify"): + # A turn-final that does not continue the stream (the model rewrote its answer): + # close the stale stream on what is visible so no live-typing indicator hangs, + # then let send() post the real final. + self._active_streams.pop(key, None) + await self._seal_stream(chat_id, stream) return None - self._active_streams.pop(chat_id, None) + if kind == "extends" and len(delta) > self.MAX_MESSAGE_LENGTH: + # Tail too large for one append: close the stream on what is visible and let the + # normal split path deliver the full final. + self._active_streams.pop(key, None) + await self._seal_stream(chat_id, stream) + return None + self._active_streams.pop(key, None) ts = stream["ts"] - ok = await self._seal_stream(chat_id, stream, final_text=text) - if not ok: - # Stop failed — post normally; the dangling stream times out on Slack's side. - return None - # Streams render markdown natively; rich blocks are applied via - # chat_update on the sealed message (mirrors edit_message finalize). - 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) + # Bounded: stopStream, one retry ONLY when nothing is appended (``markdown_text`` APPENDS: + # a first attempt that landed server-side but raised here would repeat the tail), then one + # in-place update as the commit signal — idempotent, it carries the full text — then the + # caller's plain post. + sealed = await self._seal_stream(chat_id, stream, delta=delta) + if not sealed and not (delta and delta.strip()): + sealed = await self._seal_stream(chat_id, stream) + if sealed: + # A sealed native stream is a regular message: the rich layout (if any) is applied by + # edit_message, which owns format/truncate/blocks/block-rejection retry. + # Failure is non-fatal — the streamed markdown stands. + if self._maybe_blocks(text): + await self.edit_message(chat_id, ts, text, finalize=True, metadata=metadata) + else: + # The answer is visible but not provably durable: commit it in place. Only when THAT + # fails too does send() post a fresh message (a duplicate beats a lost answer). + committed = await self.edit_message(chat_id, ts, text, finalize=True, metadata=metadata) + if not committed.success: + logger.warning( + "[Slack] Native stream %s/%s could not be sealed or updated in place; " + "delivering the final as a fresh post", chat_id, ts) + return None + logger.warning( + "[Slack] chat.stopStream failed for %s/%s; final committed via chat.update " + "instead (no duplicate post)", chat_id, ts) + await self.stop_typing(chat_id, metadata) return SendResult(success=True, message_id=ts) async def send_typing(self, chat_id: str, metadata=None) -> None: diff --git a/tests/gateway/test_native_stream_warning_composition.py b/tests/gateway/test_native_stream_warning_composition.py index 8215296b9b..fcf658c9ac 100644 --- a/tests/gateway/test_native_stream_warning_composition.py +++ b/tests/gateway/test_native_stream_warning_composition.py @@ -10,7 +10,7 @@ from gateway.run_turn_runner import TurnRunner from gateway.turn_context import TurnContext from gateway.session import SessionSource from gateway.config import Platform -from tests.gateway.test_slack_native_streaming import _make_adapter, META +from tests.gateway.test_slack_native_streaming import _make_adapter, _open_streams, META @pytest.mark.asyncio @@ -52,11 +52,11 @@ async def test_warning_and_media_failure_do_not_seal_requested_final(tmp_path, m assert media.success is not (setting is True) # legacy text fallback receipt assert client.chat_postMessage.await_count == (0 if setting is True else 2) assert client.chat_stopStream.await_count == 0 - assert "D1" in adapter._active_streams + assert _open_streams(adapter, "D1") # keyed per (team, chat, thread) await adapter.send_draft("D1", 7, "Requested final", metadata=META) result = await adapter.send("D1", "Requested final", metadata=META) assert result.success assert client.chat_startStream.await_count == 1 assert client.chat_stopStream.await_count == 1 - assert "D1" not in adapter._active_streams + assert not _open_streams(adapter, "D1") assert client.chat_postMessage.await_count == (0 if setting is True else 2) diff --git a/tests/gateway/test_slack_native_streaming.py b/tests/gateway/test_slack_native_streaming.py index 56f6be7fdf..eed333fac4 100644 --- a/tests/gateway/test_slack_native_streaming.py +++ b/tests/gateway/test_slack_native_streaming.py @@ -16,13 +16,28 @@ Behaviour contract: * startStream feature-gate error: caches _native_stream_unsupported so future supports_draft_streaming() returns False. * disconnect(): dangling streams sealed. + +Duplicate-reply invariant: + * A successfully streamed answer is NEVER posted a second time as a fresh + message — not when the agent's final differs from the streamed frames + only by surrounding whitespace (``final_response.strip()`` / + ``rstrip() + footer``), and not when chat.stopStream fails after the + whole answer is already visible (the final is then committed in place + via chat.update). + * A genuinely uncommittable stream (stopStream AND chat.update fail) still + falls back to a fresh post so the answer is not lost. + * Interim sends (``_interim_send`` / ``expect_edits``) never seal a stream. + * Streams are keyed per (team, channel, thread): two threads in one channel + never seal each other's stream. """ +import asyncio from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import PlatformConfig +from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig from plugins.platforms.slack.adapter import SlackAdapter @@ -42,7 +57,13 @@ def _make_adapter(extra=None): return a, client +def _open_streams(adapter, chat_id="D1"): + """Stream entries currently open for ``chat_id`` (any thread/team).""" + return [s for k, s in adapter._active_streams.items() if k[1] == chat_id] + + META = {"thread_id": "111.000", "user_id": "U123"} +META_B = {"thread_id": "222.000", "user_id": "U123"} class TestSupportsDraftStreaming: @@ -100,16 +121,16 @@ class TestSendDraft: @pytest.mark.asyncio async def test_cursor_glyph_stripped(self): adapter, client = _make_adapter() - await adapter.send_draft("D1", 7, "Hello \u2589", metadata=META) + await adapter.send_draft("D1", 7, "Hello ▉", metadata=META) assert client.chat_startStream.await_args.kwargs["markdown_text"] == "Hello" - await adapter.send_draft("D1", 7, "Hello world \u2589", metadata=META) + await adapter.send_draft("D1", 7, "Hello world ▉", 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) + result = await adapter.send_draft("D1", 7, "Hello ▉", metadata=META) assert result.success client.chat_appendStream.assert_not_awaited() @@ -120,7 +141,7 @@ class TestSendDraft: 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 + assert not _open_streams(adapter) @pytest.mark.asyncio async def test_no_thread_ts_fails_cleanly(self): @@ -137,7 +158,18 @@ class TestSendDraft: 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" + (stream,) = _open_streams(adapter) + assert stream["ts"] == "124.000" + + @pytest.mark.asyncio + async def test_streams_are_keyed_per_thread(self): + """Two threads in one channel: frames for B must not seal A.""" + adapter, client = _make_adapter() + await adapter.send_draft("C1", 7, "Thread A answer", metadata=META) + client.chat_startStream.return_value = {"ok": True, "ts": "456.000"} + await adapter.send_draft("C1", 8, "Thread B answer", metadata=META_B) + client.chat_stopStream.assert_not_awaited() + assert {s["ts"] for s in _open_streams(adapter, "C1")} == {"123.456", "456.000"} class TestFeatureGateFallback: @@ -152,13 +184,36 @@ class TestFeatureGateFallback: 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 TestStreamRelation: + """Pure classification of the turn-final against the streamed text.""" + + def test_exact_prefix_is_raw_delta(self): + assert SlackAdapter._stream_relation("Hello", "Hello world") == ("extends", " world") + assert SlackAdapter._stream_relation("Hello", "Hello") == ("equal", "") + + def test_surrounding_whitespace_is_tolerated(self): + assert SlackAdapter._stream_relation("\n\nHello world\n", "Hello world") == ("equal", "") + assert SlackAdapter._stream_relation("Hello world", "\n\nHello world\n")[0] == "equal" + + def test_delta_is_sliced_from_raw_final(self): + # Streamed frame ended with a trailing space+newline; the agent + # rstrip()s before appending the footer. The delta must be the exact + # raw tail of the final, whitespace preserved. + kind, delta = SlackAdapter._stream_relation("Answer \n", "Answer\n\n-# footer") + assert kind == "extends" + assert delta == "\n\n-# footer" + + def test_crlf_and_fence_closure_deltas_are_exact(self): + kind, delta = SlackAdapter._stream_relation("line1\r\n", "line1\r\nline2") + assert (kind, delta) == ("extends", "line2") + kind, delta = SlackAdapter._stream_relation("```py\nx = 1", "```py\nx = 1\n```") + assert (kind, delta) == ("extends", "\n```") + + def test_unrelated_and_empty_stream(self): + assert SlackAdapter._stream_relation("Streaming text", "Unrelated notice") == ("unrelated", "") + assert SlackAdapter._stream_relation(" ", "Hello")[0] == "unrelated" + assert SlackAdapter._stream_relation("", "Hello")[0] == "unrelated" class TestSendFinalization: @@ -172,10 +227,11 @@ class TestSendFinalization: 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 + assert not _open_streams(adapter) @pytest.mark.asyncio async def test_final_send_equal_content_seals_without_delta(self): + """A: streamed == final → one Slack message only.""" adapter, client = _make_adapter() await adapter.send_draft("D1", 7, "Hello world", metadata=META) result = await adapter.send("D1", "Hello world", metadata=META) @@ -183,6 +239,30 @@ class TestSendFinalization: kwargs = client.chat_stopStream.await_args.kwargs assert "markdown_text" not in kwargs client.chat_postMessage.assert_not_awaited() + client.chat_update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_whitespace_only_difference_does_not_duplicate(self): + """B: the agent strips final_response; the streamed frames were not.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "\n\nHello world\n", metadata=META) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + assert result.message_id == "123.456" + assert client.chat_stopStream.await_count == 1 + assert "markdown_text" not in client.chat_stopStream.await_args.kwargs + client.chat_postMessage.assert_not_awaited() + assert not _open_streams(adapter) + + @pytest.mark.asyncio + async def test_footer_after_rstrip_appends_exact_tail(self): + """B: ``final.rstrip() + "\\n\\n" + footer`` vs a streamed trailing space.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Answer \n", metadata=META) + result = await adapter.send("D1", "Answer\n\n-# footer", metadata=META) + assert result.success + assert client.chat_stopStream.await_args.kwargs["markdown_text"] == "\n\n-# footer" + client.chat_postMessage.assert_not_awaited() @pytest.mark.asyncio async def test_unrelated_send_passes_through(self): @@ -192,16 +272,231 @@ class TestSendFinalization: assert result.success client.chat_postMessage.assert_awaited() # Stream stays open for its own finalization. - assert "D1" in adapter._active_streams + assert _open_streams(adapter) @pytest.mark.asyncio - async def test_stop_stream_failure_falls_back_to_post(self): + async def test_stop_stream_failure_with_full_answer_visible_commits_in_place(self): + """C: stopStream fails after the whole answer streamed → no fresh post.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello world", metadata=META) + client.chat_stopStream = AsyncMock(side_effect=Exception("boom")) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + assert result.message_id == "123.456" + assert client.chat_stopStream.await_count == 2 # one bounded retry + client.chat_update.assert_awaited_once() + assert client.chat_update.await_args.kwargs["ts"] == "123.456" + client.chat_postMessage.assert_not_awaited() + assert not _open_streams(adapter) + + @pytest.mark.asyncio + async def test_stop_and_update_both_fail_falls_back_to_fresh_post(self): + """C2/D: an uncommittable stream still delivers the answer (loss-safe).""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello", metadata=META) + client.chat_stopStream = AsyncMock(side_effect=Exception("boom")) + client.chat_update = AsyncMock(side_effect=Exception("update boom")) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + client.chat_postMessage.assert_awaited_once() + assert client.chat_postMessage.await_args.kwargs["text"] == "Hello world" + # A stop that carries a tail is never retried: ``markdown_text`` APPENDS. + assert client.chat_stopStream.await_count == 1 + assert client.chat_update.await_count == 1 + assert not _open_streams(adapter) + + @pytest.mark.asyncio + async def test_an_ambiguous_stop_failure_never_appends_the_tail_twice(self): + """stopStream landed server-side but raised here (timeout on the response): a retry with + the same ``markdown_text`` would show "Hello world world". The final is committed by the + idempotent in-place update instead.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello", metadata=META) + client.chat_stopStream = AsyncMock(side_effect=Exception("read timeout")) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + tails = [c.kwargs.get("markdown_text") for c in client.chat_stopStream.await_args_list] + assert tails == [" world"] + assert client.chat_update.await_args.kwargs["text"] == "Hello world" + client.chat_postMessage.assert_not_awaited() + + @pytest.mark.asyncio + async def test_partial_stream_stop_failure_commits_full_final_in_place(self): + """D: partial stream, stopStream fails → final committed via update.""" 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() + assert client.chat_update.await_args.kwargs["text"] == "Hello world" + client.chat_postMessage.assert_not_awaited() + + @pytest.mark.asyncio + async def test_persistent_stop_failure_is_bounded(self): + """H: stopStream raising on every call → ≤2 stop calls, one output.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello world", metadata=META) + client.chat_stopStream = AsyncMock(side_effect=Exception("always")) + client.chat_update = AsyncMock(side_effect=Exception("always")) + await adapter.send("D1", "Hello world", metadata=META) + assert client.chat_stopStream.await_count == 2 + assert client.chat_update.await_count == 1 + assert client.chat_postMessage.await_count == 1 + + @pytest.mark.asyncio + async def test_whitespace_only_stream_is_unusable(self): + """E: a stream with no substance never claims the final.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, " ", metadata=META) + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + client.chat_postMessage.assert_awaited_once() + client.chat_stopStream.assert_not_awaited() + + @pytest.mark.asyncio + async def test_non_streaming_send_unchanged(self): + """F: no active stream → plain post, no stream API calls.""" + adapter, client = _make_adapter() + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.success + client.chat_postMessage.assert_awaited_once() + client.chat_stopStream.assert_not_awaited() + client.chat_update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_interim_send_never_seals_even_when_equal(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello world", metadata=META) + interim = dict(META, _interim_send=True) + result = await adapter.send("D1", "Hello world", metadata=interim) + assert result.success + client.chat_stopStream.assert_not_awaited() + client.chat_postMessage.assert_awaited_once() + assert _open_streams(adapter) + # The real final still seals exactly once, with no extra post. + result = await adapter.send("D1", "Hello world", metadata=META) + assert result.message_id == "123.456" + assert client.chat_stopStream.await_count == 1 + assert client.chat_postMessage.await_count == 1 + + @pytest.mark.asyncio + async def test_expect_edits_preview_never_seals(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello world", metadata=META) + preview = dict(META, expect_edits=True) + await adapter.send("D1", "Hello world", metadata=preview) + client.chat_stopStream.assert_not_awaited() + assert _open_streams(adapter) + + @pytest.mark.asyncio + async def test_two_threads_finalize_their_own_streams(self): + adapter, client = _make_adapter() + await adapter.send_draft("C1", 7, "Thread A answer", metadata=META) + client.chat_startStream.return_value = {"ok": True, "ts": "456.000"} + await adapter.send_draft("C1", 8, "Thread B answer", metadata=META_B) + rb = await adapter.send("C1", "Thread B answer", metadata=META_B) + assert rb.message_id == "456.000" + assert client.chat_stopStream.await_args.kwargs["ts"] == "456.000" + assert [s["ts"] for s in _open_streams(adapter, "C1")] == ["123.456"] + ra = await adapter.send("C1", "Thread A answer", metadata=META) + assert ra.message_id == "123.456" + assert client.chat_stopStream.await_count == 2 + client.chat_postMessage.assert_not_awaited() + assert not _open_streams(adapter, "C1") + + @pytest.mark.asyncio + async def test_team_id_routes_stream_calls(self): + adapter, client = _make_adapter() + meta = dict(META, slack_team_id="T999") + await adapter.send_draft("C1", 7, "Hello", metadata=meta) + client.chat_stopStream = AsyncMock(side_effect=Exception("boom")) + await adapter.send("C1", "Hello world", metadata=meta) + teams = {c.kwargs.get("team_id") for c in adapter._get_client.call_args_list} + assert teams == {"T999"} + client.chat_postMessage.assert_not_awaited() + + @pytest.mark.asyncio + async def test_oversized_tail_uses_normal_split_path(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Intro", metadata=META) + tail = "x" * (adapter.MAX_MESSAGE_LENGTH + 10) + result = await adapter.send("D1", "Intro" + tail, metadata=META) + assert result.success + # Stream closed on what was visible (no oversized append), then the + # normal split path delivers the full final; nothing dangles. + client.chat_stopStream.assert_awaited_once() + assert "markdown_text" not in client.chat_stopStream.await_args.kwargs + assert client.chat_postMessage.await_count >= 1 + assert not _open_streams(adapter) + + @pytest.mark.asyncio + async def test_rewritten_turn_final_seals_stale_stream_then_posts(self): + """A turn-final (notify=True) that no longer continues the stream.""" + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Draft answer that got rewritten", metadata=META) + result = await adapter.send("D1", "Completely new answer", metadata=dict(META, notify=True)) + assert result.success + client.chat_stopStream.assert_awaited_once() + assert "markdown_text" not in client.chat_stopStream.await_args.kwargs + client.chat_postMessage.assert_awaited_once() + assert not _open_streams(adapter) + + @pytest.mark.asyncio + async def test_explicit_team_metadata_fills_recipient_team(self): + adapter, client = _make_adapter() + await adapter.send_draft("C1", 7, "Hello", metadata=dict(META, slack_team_id="T999")) + assert client.chat_startStream.await_args.kwargs["recipient_team_id"] == "T999" + + @pytest.mark.asyncio + async def test_thread_status_cleared_with_metadata(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, "Hello", metadata=META) + await adapter.send("D1", "Hello world", metadata=META) + adapter.stop_typing.assert_awaited_once_with("D1", META) + + +RICH_MD = "# Title\n\nbody text with **bold**\n\n| a | b |\n|---|---|\n| 1 | 2 |" + + +class TestRichBlocksAfterSeal: + """G: with ``rich_blocks`` the sealed stream gets its layout via chat.update, once.""" + + @pytest.mark.asyncio + async def test_rich_final_seals_then_renders_blocks_once(self): + adapter, client = _make_adapter({"rich_blocks": True}) + body = f"Intro.\n\n{RICH_MD}\n" + await adapter.send_draft("D1", 7, body, metadata=META) + result = await adapter.send("D1", body.strip(), metadata=META) + assert result.success and result.message_id == "123.456" + assert client.chat_stopStream.await_count == 1 + client.chat_postMessage.assert_not_awaited() + client.chat_update.assert_awaited_once() + kwargs = client.chat_update.await_args.kwargs + assert kwargs["ts"] == "123.456" + assert kwargs["blocks"] + assert "header" in [b["type"] for b in kwargs["blocks"]] + + @pytest.mark.asyncio + async def test_block_rejection_retries_without_blocks_no_post(self): + adapter, client = _make_adapter({"rich_blocks": True}) + await adapter.send_draft("D1", 7, RICH_MD, metadata=META) + client.chat_update = AsyncMock( + side_effect=[Exception("invalid_blocks"), {"ok": True}] + ) + result = await adapter.send("D1", RICH_MD, metadata=META) + assert result.success + assert client.chat_update.await_count == 2 + assert client.chat_update.await_args.kwargs["blocks"] == [] + client.chat_postMessage.assert_not_awaited() + + @pytest.mark.asyncio + async def test_block_update_failure_after_seal_keeps_markdown(self): + adapter, client = _make_adapter({"rich_blocks": True}) + await adapter.send_draft("D1", 7, RICH_MD, metadata=META) + client.chat_update = AsyncMock(side_effect=Exception("ratelimited")) + result = await adapter.send("D1", RICH_MD, metadata=META) + assert result.success and result.message_id == "123.456" + client.chat_postMessage.assert_not_awaited() @pytest.mark.asyncio async def test_rich_blocks_applied_after_seal(self): @@ -213,6 +508,61 @@ class TestSendFinalization: client.chat_update.assert_awaited() assert client.chat_update.await_args.kwargs["blocks"] + @pytest.mark.asyncio + async def test_plain_final_without_rich_blocks_never_updates(self): + adapter, client = _make_adapter() + await adapter.send_draft("D1", 7, RICH_MD[:8], metadata=META) + result = await adapter.send("D1", RICH_MD, metadata=META) + assert result.success and result.message_id == "123.456" + client.chat_update.assert_not_awaited() + client.chat_postMessage.assert_not_awaited() + + +class TestEndToEndConsumer: + """GatewayStreamConsumer → real SlackAdapter (mocked Slack client).""" + + async def _run(self, adapter, deltas, final, *, segment_break_after=None): + cfg = StreamConsumerConfig( + transport="auto", chat_type="dm", edit_interval=0.01, buffer_threshold=1, + ) + consumer = GatewayStreamConsumer(adapter, "D1", cfg, metadata=dict(META)) + task = asyncio.create_task(consumer.run()) + for i, d in enumerate(deltas): + consumer.on_delta(d) + await asyncio.sleep(0.05) + if segment_break_after is not None and i == segment_break_after: + consumer.on_segment_break() + await asyncio.sleep(0.05) + consumer.finish(final) + await task + return consumer + + @pytest.mark.asyncio + async def test_stripped_final_is_one_message(self): + adapter, client = _make_adapter() + consumer = await self._run(adapter, ["\n\nHello ", "world\n"], "Hello world") + assert client.chat_startStream.await_count == 1 + assert client.chat_stopStream.await_count == 1 + client.chat_postMessage.assert_not_awaited() + assert not _open_streams(adapter) + assert consumer.final_response_sent + assert consumer.delivered_final_matches("Hello world") is True + + @pytest.mark.asyncio + async def test_tool_boundary_yields_two_sealed_streams_no_post(self): + adapter, client = _make_adapter() + client.chat_startStream = AsyncMock( + side_effect=[{"ok": True, "ts": "1.0"}, {"ok": True, "ts": "2.0"}] + ) + await self._run( + adapter, ["First segment ", "here.", "Second segment."], + "Second segment.", segment_break_after=1, + ) + assert client.chat_startStream.await_count == 2 + assert client.chat_stopStream.await_count == 2 + client.chat_postMessage.assert_not_awaited() + assert not _open_streams(adapter) + class TestDisconnectCleanup: @pytest.mark.asyncio @@ -223,4 +573,5 @@ class TestDisconnectCleanup: adapter._release_platform_lock = MagicMock() await adapter.disconnect() client.chat_stopStream.assert_awaited() + assert client.chat_stopStream.await_args.kwargs["channel"] == "D1" assert not adapter._active_streams