fix(slack): never re-post a successfully streamed answer
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)
This commit is contained in:
ms-elbdev
kshitij
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user