From 4e7d0f8117be40035aaca5660ef43a6bf79fefe5 Mon Sep 17 00:00:00 2001 From: patrick-chinchill Date: Fri, 19 Jun 2026 23:00:58 -0700 Subject: [PATCH 1/2] feat(core): optional ThinkingChunk StreamChunk variant (default-off, upstream-compatible; supersedes #39) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a fourth, opt-in StreamChunk variant for streaming agent reasoning/"thinking" to chat platforms. The whole design is default-off: with no opt-in, the normalized stream and the posted message are byte-for-byte identical to upstream chat@4.31. - Additive type: ThinkingChunk(type="thinking", content=str) added to the StreamChunk union. The existing three variants are unaffected. - Opt-in emit: from_full_stream(stream, emit_thinking=False) and a thread-level emit_thinking config flag (threaded into the internal _from_full_stream) surface AI-SDK reasoning/reasoning-delta (and pydantic-ai part_kind=="thinking") parts as ThinkingChunk only when enabled. Default False drops reasoning exactly as upstream does. - Graceful consume: Thread._handle_stream never accumulates a ThinkingChunk into the posted-message text, and every adapter's stream handler skips it. Slack/Teams expose an optional render_thinking hook (via shared.adapter_utils.maybe_render_thinking); the text-accumulate adapters ignore it structurally — no crash, posted message unchanged. - No state pollution: ThinkingChunk is streaming-only. Message has no thinking field, to_json() is unchanged, and a round-tripped Message is byte-identical, so cross-SDK Redis/Postgres state stays compatible. - Docs: UPSTREAM_SYNC.md Known Non-Parity row added. Rationale: upstream's chat-platform SDK drops reasoning (leaves it to the AI-SDK web UI); chinchill streams thinking to Slack/Teams out-of-band today because there is no path. This gives the SDK a first-class, opt-in one without changing any default behavior. Gauntlet: ruff check + format, audit_test_quality (0 hard failures), verify_test_fidelity --strict (732/732, 0 missing), pyrefly src (0 errors), pytest (5121 passed, 4 pre-existing skips). --- docs/UPSTREAM_SYNC.md | 1 + src/chat_sdk/__init__.py | 2 + src/chat_sdk/adapters/slack/adapter.py | 16 +- src/chat_sdk/adapters/teams/adapter.py | 20 +- src/chat_sdk/from_full_stream.py | 64 ++++- src/chat_sdk/shared/adapter_utils.py | 45 ++++ src/chat_sdk/thread.py | 58 ++++- src/chat_sdk/types.py | 34 ++- tests/test_from_full_stream.py | 117 ++++++++- tests/test_messenger_api.py | 20 ++ tests/test_slack_api.py | 78 ++++++ tests/test_teams_native_streaming.py | 52 ++++ tests/test_thinking_chunk.py | 340 +++++++++++++++++++++++++ tests/test_twilio_adapter.py | 31 +++ tests/test_types.py | 41 +++ 15 files changed, 900 insertions(+), 19 deletions(-) create mode 100644 tests/test_thinking_chunk.py diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index d89bc489..c058f081 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -684,6 +684,7 @@ stay explicit instead of being rediscovered in code review. | Teams `cards_input` empty-options default (vercel/chat#8c71411, chat@4.31) | `input_request_to_teams_adaptive_card` reads `options = request.get("options") or []` | Upstream `cards-primitives/input.ts` reads `const options = request.options ?? []` | Benign truthiness divergence. The only values that differ between `or []` and `?? []` are the falsy-but-present ones (`[]`, `None`); for an options list both produce the same empty list, so the rendered card is byte-identical. Documented (not "fixed" to `is not None`) because there is no observable behavior difference — a present empty `options` and an absent `options` both yield "no choices". | | Teams `graph` path-segment encoding (vercel/chat#8c71411, chat@4.31) | Path segments (`team_id` / `channel_id` / `chat_id` / `message_id`) are interpolated through `quote(segment, safe='')` | Upstream `graph/{channels,messages}.ts` use `encodeURIComponent(...)` | Benign over-encoding divergence. `quote(safe='')` percent-encodes a strictly larger set than `encodeURIComponent` (notably `!`, `'`, `(`, `)`, `*` — which `encodeURIComponent` leaves literal). Graph IDs (team/channel/chat/message GUIDs and thread tokens) never contain those characters, so the encoded path is identical in practice; where they did differ, the stricter `quote` is the safer choice (no URL injection via an unescaped sub-delimiter). Documented rather than narrowed to match `encodeURIComponent` exactly. | | Teams `graph` defensive shape coercion (vercel/chat#8c71411, chat@4.31) | `to_graph_message` / channel + message readers coerce unexpected Graph payload shapes with `isinstance` guards (`x if isinstance(x, Mapping) else {}`, `value if isinstance(value, list) else []`) before reading fields | Upstream `graph/messages.ts` reads `message.from?.user` / `message.body?.content ?? ""` etc. with optional chaining — a non-object where an object is expected throws at the property access | Benign defensive divergence. Upstream's optional chaining tolerates `null`/`undefined` but throws on a wrong-typed non-null (e.g. a string where an object is expected); our `isinstance` coercion fails closed to an empty mapping/list instead of raising. For well-formed Graph responses the behavior is identical; the divergence only manifests on malformed payloads, where returning an empty-shape result is more resilient than throwing. Mirrors the repo's general "more resilient than throw" stance (cf. the `renderPostable on unknown input` row). | +| `ThinkingChunk` StreamChunk variant (Python-only, default-off; supersedes PR #39) | A fourth `StreamChunk` variant — `ThinkingChunk(type="thinking", content=str)` — surfaces AI-SDK `reasoning`/`reasoning-delta` (and pydantic-ai `part_kind == "thinking"`) parts. **OPT-IN, default-off**: emitted only when a caller passes `from_full_stream(stream, emit_thinking=True)` or sets the thread-level `emit_thinking=True` config; the internal `_from_full_stream` threads the same flag. With the default (`emit_thinking=False`) the normalized stream is **byte-for-byte identical** to upstream — reasoning parts are dropped and **no** `ThinkingChunk` is produced. Consumption is graceful: `Thread._handle_stream` never accumulates a `ThinkingChunk` into the posted-message text, and every adapter's stream handler skips it (Slack/Teams expose an optional `render_thinking` hook via `chat_sdk.shared.adapter_utils.maybe_render_thinking`; the text-accumulate adapters ignore it structurally). **Streaming-only — never persisted**: `Message` has no `thinking` field, `to_json()` is unchanged, and a round-tripped `Message` is byte-identical, so cross-SDK state (Redis/Postgres shared with the TS SDK) stays compatible. | Upstream `from-full-stream.ts` forwards only `text-delta` + `finish-step`; AI-SDK `reasoning`/`reasoning-delta` parts fall through and are discarded. `StreamChunk = MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk` — no reasoning variant. Upstream leaves reasoning display to the AI-SDK web UI. | chinchill actively streams agent thinking to Slack/Teams but has to intercept the model stream out-of-band today because the chat-platform SDK has no path for it. This gives the SDK a first-class, opt-in one without changing any default behavior. The whole design constraint is that default-off == upstream: additive type, opt-in emit, graceful/skip consume, zero state pollution. Regression coverage: `tests/test_thinking_chunk.py`, `tests/test_from_full_stream.py::TestThinkingOptIn`, `tests/test_types.py::TestThinkingChunk`, plus per-adapter no-crash tests in `tests/test_slack_api.py`, `tests/test_teams_native_streaming.py`, `tests/test_twilio_adapter.py`, `tests/test_messenger_api.py`. | ### Platform-specific gaps | Area | Python | TS | Rationale | diff --git a/src/chat_sdk/__init__.py b/src/chat_sdk/__init__.py index d630a4fe..f8c2a26b 100644 --- a/src/chat_sdk/__init__.py +++ b/src/chat_sdk/__init__.py @@ -212,6 +212,7 @@ StreamChunk, StreamOptions, TaskUpdateChunk, + ThinkingChunk, Thread, ThreadInfo, ThreadSummary, @@ -450,6 +451,7 @@ "StreamChunk", "StreamOptions", "TaskUpdateChunk", + "ThinkingChunk", "Thread", "ThreadInfo", "ThreadSummary", diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index 681f03a6..a479a1b6 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -64,7 +64,12 @@ from chat_sdk.emoji import emoji_to_slack, resolve_emoji_from_slack from chat_sdk.logger import ConsoleLogger, Logger from chat_sdk.modals import ModalElement, OptionsLoadGroup, SelectOptionElement -from chat_sdk.shared.adapter_utils import extract_card, extract_files +from chat_sdk.shared.adapter_utils import ( + extract_card, + extract_files, + is_thinking_chunk, + maybe_render_thinking, +) from chat_sdk.shared.errors import AdapterRateLimitError, AuthenticationError, ValidationError from chat_sdk.types import ( ActionEvent, @@ -4006,6 +4011,15 @@ async def push_text_and_flush(text: str) -> None: await push_text_and_flush(text_value) elif hasattr(chunk, "type") and chunk.type == "markdown_text": # type: ignore[union-attr] await push_text_and_flush(chunk.text) # type: ignore[union-attr] + elif is_thinking_chunk(chunk): + # Python-only divergence: ``ThinkingChunk`` is streaming-only + # agent reasoning, NOT message content. By default it is + # skipped (no effect on the posted message). An adapter/consumer + # that wants to display thinking sets ``self.render_thinking``; + # only then is it invoked. Skipping (not routing to + # ``send_structured_chunk``) keeps the default posted message + # byte-identical and avoids disabling structured-chunk support. + await maybe_render_thinking(getattr(self, "render_thinking", None), chunk) else: await send_structured_chunk(chunk) diff --git a/src/chat_sdk/adapters/teams/adapter.py b/src/chat_sdk/adapters/teams/adapter.py index f37f515f..2e588290 100644 --- a/src/chat_sdk/adapters/teams/adapter.py +++ b/src/chat_sdk/adapters/teams/adapter.py @@ -34,7 +34,12 @@ from chat_sdk.emoji import convert_emoji_placeholders from chat_sdk.errors import ChatNotImplementedError from chat_sdk.logger import ConsoleLogger, Logger -from chat_sdk.shared.adapter_utils import extract_card, extract_files +from chat_sdk.shared.adapter_utils import ( + extract_card, + extract_files, + is_thinking_chunk, + maybe_render_thinking, +) from chat_sdk.shared.buffer_utils import buffer_to_data_uri, to_buffer from chat_sdk.shared.errors import ( AdapterPermissionError, @@ -1724,6 +1729,12 @@ async def stream( text = chunk elif isinstance(chunk, dict) and chunk.get("type") == "markdown_text": text = chunk.get("text", "") + elif is_thinking_chunk(chunk): + # Python-only divergence: streaming-only reasoning, not message + # content. Default-skip keeps the buffered post byte-identical; + # an opt-in ``render_thinking`` hook may display it. + await maybe_render_thinking(getattr(self, "render_thinking", None), chunk) + continue if not text: continue accumulated += text @@ -1796,6 +1807,13 @@ async def _on_chunk(activity: Any) -> None: text = chunk elif isinstance(chunk, dict) and chunk.get("type") == "markdown_text": text = chunk.get("text", "") + elif is_thinking_chunk(chunk): + # Python-only divergence: streaming-only reasoning, not + # message content. Default-skip keeps emitted text + # byte-identical; an opt-in ``render_thinking`` hook may + # display it. + await maybe_render_thinking(getattr(self, "render_thinking", None), chunk) + continue elif getattr(chunk, "type", None) == "markdown_text": # Dataclass ``MarkdownTextChunk`` form — mirror # ``Thread.stream``'s ``_wrapped_stream`` extraction so the diff --git a/src/chat_sdk/from_full_stream.py b/src/chat_sdk/from_full_stream.py index 0f1269d2..795979a9 100644 --- a/src/chat_sdk/from_full_stream.py +++ b/src/chat_sdk/from_full_stream.py @@ -11,23 +11,62 @@ - **StreamChunk objects** (``task_update``, ``plan_update``, ``markdown_text``) -- passed through as-is for adapters with native structured chunk support. + +Python-only divergence (default-off): when ``emit_thinking=True``, AI-SDK +``reasoning`` / ``reasoning-delta`` parts (and pydantic-ai +``part_kind == "thinking"`` parts) are surfaced as +:class:`~chat_sdk.types.ThinkingChunk` objects. With the default +``emit_thinking=False`` the output is byte-for-byte identical to upstream +chat@4.31 (reasoning is dropped, no ``ThinkingChunk`` is emitted). See +``docs/UPSTREAM_SYNC.md`` (Known Non-Parity). """ from __future__ import annotations from collections.abc import AsyncIterable, AsyncIterator -from chat_sdk.types import StreamChunk +from chat_sdk.types import StreamChunk, ThinkingChunk _STREAM_CHUNK_TYPES = frozenset({"markdown_text", "task_update", "plan_update"}) +# AI-SDK v5/v6 reasoning part types, plus pydantic-ai's ``thinking`` part kind. +# These are only consulted when ``emit_thinking=True``; otherwise they fall +# through and are dropped exactly as upstream does. +_REASONING_TYPES = frozenset({"reasoning", "reasoning-delta", "thinking"}) + +_TEXT_KEYS = ("text", "delta", "textDelta", "text_delta") +# Reasoning payloads carry the text under the same keys, with ``content`` added +# for pydantic-ai's ``ThinkingPart`` shape (``part_kind == "thinking"``). +_REASONING_KEYS = ("content", "text", "delta", "textDelta", "text_delta") + + +def _pick(event: object, keys: tuple[str, ...]) -> object | None: + """Return the first non-``None`` value among ``keys`` on a dict/object.""" + if isinstance(event, dict): + return next((v for k in keys if (v := event.get(k)) is not None), None) + return next((v for k in keys if (v := getattr(event, k, None)) is not None), None) + async def from_full_stream( stream: AsyncIterable[object], + *, + emit_thinking: bool = False, ) -> AsyncIterator[str | StreamChunk]: """Normalize an async iterable stream for use with ``thread.post()``. Yields either plain ``str`` chunks or ``StreamChunk`` objects. + + Args: + stream: The source async iterable (text stream, full stream, or + pre-built ``StreamChunk`` objects). + emit_thinking: **Opt-in, default off.** When ``False`` (the default), + behavior is byte-for-byte upstream: AI-SDK ``reasoning`` / + ``reasoning-delta`` parts are dropped and **no** + :class:`~chat_sdk.types.ThinkingChunk` is emitted. When ``True``, + such parts (and pydantic-ai ``thinking`` parts) are surfaced as + ``ThinkingChunk`` objects so a consumer/adapter can render agent + reasoning. Thinking is never accumulated into the posted message + text, so the posted message is unchanged either way. """ needs_separator = False has_emitted_text = False @@ -52,23 +91,24 @@ async def from_full_stream( if not event_type: continue - # Pass through StreamChunk objects + # Pass through StreamChunk objects (includes pre-built ThinkingChunk). if event_type in _STREAM_CHUNK_TYPES: yield event # type: ignore[misc] continue + # Opt-in reasoning surfacing. Default-off => this whole branch is + # skipped and reasoning parts fall through (dropped) exactly as + # upstream chat@4.31 does. + if event_type in _REASONING_TYPES: + if emit_thinking: + content = _pick(event, _REASONING_KEYS) + if isinstance(content, str) and content: + yield ThinkingChunk(content=content) + continue + # AI SDK v6 uses "text", v5 uses "textDelta"; also accept "delta" # Priority: text > delta > textDelta > text_delta (matches TS) - if isinstance(event, dict): - text_content = next( - (v for k in ("text", "delta", "textDelta", "text_delta") if (v := event.get(k)) is not None), - None, - ) - else: - text_content = next( - (v for k in ("text", "delta", "textDelta", "text_delta") if (v := getattr(event, k, None)) is not None), - None, - ) + text_content = _pick(event, _TEXT_KEYS) if event_type == "text-delta" and isinstance(text_content, str): if needs_separator and has_emitted_text: diff --git a/src/chat_sdk/shared/adapter_utils.py b/src/chat_sdk/shared/adapter_utils.py index 0250344e..5246005b 100644 --- a/src/chat_sdk/shared/adapter_utils.py +++ b/src/chat_sdk/shared/adapter_utils.py @@ -2,9 +2,54 @@ from __future__ import annotations +import inspect +from collections.abc import Awaitable, Callable +from typing import Any + from chat_sdk.cards import CardElement, is_card_element from chat_sdk.types import AdapterPostableMessage, Attachment, FileUpload +# Optional opt-in hook signature an adapter/consumer may set on the adapter +# instance as ``render_thinking`` to receive streaming ``ThinkingChunk`` +# content. May be sync or async; both are supported by +# :func:`maybe_render_thinking`. +RenderThinking = Callable[[str], Awaitable[None] | None] + + +def is_thinking_chunk(chunk: Any) -> bool: + """Return True if a stream chunk is a Python-only ``ThinkingChunk``. + + Matches both the dataclass form (``chunk.type == "thinking"``) and the + dict-normalized form (``chunk["type"] == "thinking"``). Used by adapter + stream loops to *skip* thinking when accumulating the posted message — + thinking is streaming-only reasoning, not message content. + """ + if isinstance(chunk, dict): + return chunk.get("type") == "thinking" + return getattr(chunk, "type", None) == "thinking" + + +def thinking_content(chunk: Any) -> str: + """Extract the reasoning text from a ``ThinkingChunk`` (dict or dataclass).""" + value = chunk.get("content") if isinstance(chunk, dict) else getattr(chunk, "content", None) + return value if isinstance(value, str) else "" + + +async def maybe_render_thinking(hook: RenderThinking | None, chunk: Any) -> None: + """Invoke an opt-in ``render_thinking`` hook for a ``ThinkingChunk``, else skip. + + Default behavior (``hook is None``) is a no-op: the thinking chunk is + silently skipped so the posted message stays byte-identical to upstream and + adapters that don't render thinking never crash. When a hook is provided it + is called with the reasoning text; both sync and async hooks are supported. + """ + if hook is None: + return + content = thinking_content(chunk) + result = hook(content) + if inspect.isawaitable(result): + await result + def extract_card(message: AdapterPostableMessage) -> CardElement | None: """Extract CardElement from an AdapterPostableMessage if present.""" diff --git a/src/chat_sdk/thread.py b/src/chat_sdk/thread.py index 8dba3c4b..53ed808b 100644 --- a/src/chat_sdk/thread.py +++ b/src/chat_sdk/thread.py @@ -47,6 +47,7 @@ StreamChunk, StreamOptions, TaskUpdateChunk, + ThinkingChunk, ) if TYPE_CHECKING: @@ -266,6 +267,11 @@ class _ThreadImplConfig: is_subscribed_context: bool = False logger: Logger | None = None streaming_update_interval_ms: int = 500 + # Python-only divergence (default-off). When True, raw AI-SDK + # ``reasoning`` / ``reasoning-delta`` parts passed to ``post()`` are + # surfaced as ``ThinkingChunk`` objects for adapters that render thinking. + # Default False keeps the stream byte-for-byte upstream (reasoning dropped). + emit_thinking: bool = False # Direct adapter mode adapter: Adapter | None = None @@ -294,6 +300,7 @@ def __init__(self, config: _ThreadImplConfig) -> None: self._logger = config.logger self._streaming_update_interval_ms = config.streaming_update_interval_ms self._fallback_streaming_placeholder_text = config.fallback_streaming_placeholder_text + self._emit_thinking = config.emit_thinking # Recent messages cache self._recent_messages: list[Message] = [] @@ -687,8 +694,11 @@ async def _handle_stream( are built into ``StreamOptions`` before both ``adapter.stream`` and ``fallbackStream`` are invoked. """ - # Build text-only stream from raw_stream - text_stream = _from_full_stream(raw_stream) + # Build text-only stream from raw_stream. ``emit_thinking`` is a + # Python-only, default-off divergence: when enabled, raw reasoning + # parts become ``ThinkingChunk`` objects; when off (the default) the + # stream is byte-for-byte upstream. + text_stream = _from_full_stream(raw_stream, emit_thinking=self._emit_thinking) # Build streaming options from current message context + caller # options. Upstream parity (vercel/chat#340): seed @@ -1283,12 +1293,24 @@ def _to_message(sent: SentMessage) -> Message: # --------------------------------------------------------------------------- -async def _from_full_stream(raw_stream: Any) -> AsyncIterator[str | StreamChunk]: +async def _from_full_stream( + raw_stream: Any, + *, + emit_thinking: bool = False, +) -> AsyncIterator[str | StreamChunk]: """Normalise a raw async iterable into str or StreamChunk items. Handles plain strings, AI SDK fullStream events, and StreamChunk objects. Mirrors from-full-stream.ts: tracks ``finish-step`` events so that a ``"\n\n"`` separator is emitted between consecutive steps. + + ``emit_thinking`` is a Python-only, default-off divergence. When ``False`` + (the default) the output is byte-for-byte upstream: AI-SDK ``reasoning`` / + ``reasoning-delta`` parts are dropped and pre-built ``ThinkingChunk`` + objects are *not* forwarded. When ``True``, reasoning parts are surfaced as + ``ThinkingChunk`` objects and pre-built ``ThinkingChunk`` objects pass + through (for adapters that render thinking). Either way thinking is never + accumulated into the posted message text. """ needs_separator = False has_emitted_text = False @@ -1307,6 +1329,21 @@ async def _from_full_stream(raw_stream: Any) -> AsyncIterator[str | StreamChunk] yield item continue + # Opt-in reasoning surfacing (default-off => dropped as upstream). + if item_type in ("reasoning", "reasoning-delta", "thinking"): + if emit_thinking: + content = next( + ( + v + for k in ("content", "text", "delta", "textDelta", "text_delta") + if (v := getattr(item, k, None)) is not None + ), + "", + ) + if isinstance(content, str) and content: + yield ThinkingChunk(content=content) + continue + # AI SDK v6 uses "text", v5 uses "textDelta"; also accept "delta" if item_type == "text-delta": text_content = next( @@ -1337,6 +1374,21 @@ async def _from_full_stream(raw_stream: Any) -> AsyncIterator[str | StreamChunk] yield cast("MarkdownTextChunk | PlanUpdateChunk | TaskUpdateChunk", item) continue + # Opt-in reasoning surfacing (default-off => dropped as upstream). + if t in ("reasoning", "reasoning-delta", "thinking"): + if emit_thinking: + content = next( + ( + v + for k in ("content", "text", "delta", "textDelta", "text_delta") + if (v := item.get(k)) is not None + ), + "", + ) + if isinstance(content, str) and content: + yield ThinkingChunk(content=content) + continue + if t == "text-delta": text_content = next( (v for k in ("text", "delta", "textDelta", "text_delta") if (v := item.get(k)) is not None), diff --git a/src/chat_sdk/types.py b/src/chat_sdk/types.py index db503889..060e5047 100644 --- a/src/chat_sdk/types.py +++ b/src/chat_sdk/types.py @@ -1002,7 +1002,39 @@ class PlanUpdateChunk: title: str = "" -StreamChunk = MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk +@dataclass +class ThinkingChunk: + """Streamed agent reasoning / "thinking" content. + + **Python-only divergence — not present in upstream chat@4.31.** Upstream's + chat-platform SDK drops AI-SDK ``reasoning`` / ``reasoning-delta`` parts in + ``from-full-stream.ts`` (only ``text-delta`` and ``finish-step`` are + forwarded); reasoning is left to the AI-SDK web UI. This variant gives the + chat-platform SDK a first-class, *opt-in* path so consumers can stream + agent thinking to chat platforms (e.g. Slack context blocks, Teams styled + text). + + Design guarantees (see ``docs/UPSTREAM_SYNC.md`` Known Non-Parity): + + - **Default-off.** No ``ThinkingChunk`` is ever produced unless a caller + explicitly opts in (``from_full_stream(..., emit_thinking=True)``), so the + default stream is byte-for-byte upstream. + - **Streaming-only.** It is never persisted into :class:`Message` / history + / state, so cross-SDK state (Redis/Postgres shared with the TS SDK) stays + identical. + - **Gracefully skipped.** It is not message content: the default + message-rendering path never accumulates it into the posted text, and + adapters that do not render thinking simply ignore it. + + The shape mirrors chinchill's thinking-delta model (``part_kind == "thinking"`` + with a ``content`` string) so it can be consumed directly. + """ + + type: Literal["thinking"] = "thinking" + content: str = "" + + +StreamChunk = MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk | ThinkingChunk @dataclass diff --git a/tests/test_from_full_stream.py b/tests/test_from_full_stream.py index 7c814146..6a119ffd 100644 --- a/tests/test_from_full_stream.py +++ b/tests/test_from_full_stream.py @@ -11,7 +11,7 @@ import pytest from chat_sdk.from_full_stream import from_full_stream -from chat_sdk.types import StreamChunk +from chat_sdk.types import StreamChunk, ThinkingChunk # --------------------------------------------------------------------------- # Helpers @@ -358,3 +358,118 @@ async def test_prefers_text_over_textdelta_when_both_present(self): async def test_ignores_invalid_events_null_primitives_missing_type(self): events: list = [None, 42, True, {"no_type": 1}, {"type": "text-delta", "textDelta": "ok"}] assert await _collect(from_full_stream(_async_iter(events))) == ["ok"] + + +# --------------------------------------------------------------------------- +# ThinkingChunk opt-in (Python-only divergence, default-off == upstream) +# --------------------------------------------------------------------------- + + +class TestThinkingOptIn: + """``emit_thinking`` surfaces AI-SDK reasoning as ``ThinkingChunk``. + + Default-off must be byte-for-byte upstream: reasoning parts are dropped + and no ``ThinkingChunk`` is ever produced. Opt-in turns them into chunks. + """ + + async def test_default_off_drops_reasoning_delta(self): + # Upstream chat@4.31 drops `reasoning-delta`; default-off must match. + events: list[Any] = [ + {"type": "reasoning-delta", "text": "thinking hard"}, + {"type": "text-delta", "text": "answer"}, + ] + result = await _collect(from_full_stream(_async_iter(events))) + assert result == ["answer"] + assert all(not isinstance(c, ThinkingChunk) for c in result) + + async def test_default_off_drops_reasoning_part(self): + events: list[Any] = [ + {"type": "reasoning", "text": "step 1"}, + {"type": "text-delta", "text": "out"}, + ] + result = await _collect(from_full_stream(_async_iter(events))) + assert result == ["out"] + + async def test_default_off_drops_pydantic_thinking_part(self): + # pydantic-ai shape: type == "thinking", content carries the text. + events: list[Any] = [ + {"type": "thinking", "content": "reasoning"}, + {"type": "text-delta", "text": "final"}, + ] + result = await _collect(from_full_stream(_async_iter(events))) + assert result == ["final"] + + async def test_default_off_byte_identical_to_no_reasoning(self): + # Output with reasoning parts (default-off) must equal output of the + # same stream with the reasoning parts removed entirely. + with_reasoning: list[Any] = [ + {"type": "text-delta", "text": "A"}, + {"type": "reasoning-delta", "text": "secret thought"}, + {"type": "finish-step"}, + {"type": "reasoning", "text": "more thought"}, + {"type": "text-delta", "text": "B"}, + ] + without_reasoning: list[Any] = [ + {"type": "text-delta", "text": "A"}, + {"type": "finish-step"}, + {"type": "text-delta", "text": "B"}, + ] + assert await _collect(from_full_stream(_async_iter(with_reasoning))) == await _collect( + from_full_stream(_async_iter(without_reasoning)) + ) + + async def test_opt_in_yields_thinking_chunk_from_reasoning_delta(self): + events: list[Any] = [{"type": "reasoning-delta", "text": "analyzing"}] + result = await _collect(from_full_stream(_async_iter(events), emit_thinking=True)) + assert len(result) == 1 + chunk = result[0] + assert isinstance(chunk, ThinkingChunk) + assert chunk.type == "thinking" + assert chunk.content == "analyzing" + + async def test_opt_in_yields_thinking_chunk_from_reasoning_part(self): + events: list[Any] = [{"type": "reasoning", "text": "deliberating"}] + result = await _collect(from_full_stream(_async_iter(events), emit_thinking=True)) + assert result == [ThinkingChunk(content="deliberating")] + + async def test_opt_in_yields_thinking_chunk_from_pydantic_content(self): + # pydantic-ai ThinkingPart carries text in `content`, not `text`. + events: list[Any] = [{"type": "thinking", "content": "pondering"}] + result = await _collect(from_full_stream(_async_iter(events), emit_thinking=True)) + assert result == [ThinkingChunk(content="pondering")] + + async def test_opt_in_interleaves_thinking_and_text(self): + events: list[Any] = [ + {"type": "reasoning-delta", "text": "let me think"}, + {"type": "text-delta", "text": "the answer is"}, + {"type": "reasoning", "text": "double-checking"}, + {"type": "text-delta", "text": " 42"}, + ] + result = await _collect(from_full_stream(_async_iter(events), emit_thinking=True)) + assert result == [ + ThinkingChunk(content="let me think"), + "the answer is", + ThinkingChunk(content="double-checking"), + " 42", + ] + + async def test_opt_in_skips_empty_reasoning(self): + # No content => nothing emitted even with opt-in on. + events: list[Any] = [ + {"type": "reasoning-delta", "text": ""}, + {"type": "reasoning"}, + {"type": "text-delta", "text": "x"}, + ] + result = await _collect(from_full_stream(_async_iter(events), emit_thinking=True)) + assert result == ["x"] + + async def test_opt_in_does_not_change_text_output(self): + # The text path is identical with the flag on; only thinking is added. + events: list[Any] = [ + {"type": "text-delta", "text": "A"}, + {"type": "finish-step"}, + {"type": "text-delta", "text": "B"}, + ] + off = await _collect(from_full_stream(_async_iter(events))) + on = await _collect(from_full_stream(_async_iter(events), emit_thinking=True)) + assert off == on == ["A", "\n\n", "B"] diff --git a/tests/test_messenger_api.py b/tests/test_messenger_api.py index 97e7c31c..78dc4bc4 100644 --- a/tests/test_messenger_api.py +++ b/tests/test_messenger_api.py @@ -345,6 +345,26 @@ async def _chunks() -> AsyncIterator[str | StreamChunk]: body = adapter._graph_api_fetch.call_args.kwargs["body"] assert body["message"]["text"] == "Structured plain content" + @pytest.mark.asyncio + async def test_ignores_thinking_chunk(self) -> None: + """A ``ThinkingChunk`` is skipped (streaming-only reasoning, not message + content); the buffered post excludes it and does not crash.""" + from chat_sdk.types import ThinkingChunk + + adapter = _make_adapter() + adapter._graph_api_fetch = AsyncMock(return_value=_send_api_response("mid.t")) + + async def _chunks() -> AsyncIterator[str | StreamChunk]: + yield ThinkingChunk(content="reasoning") + yield "Hello " + yield ThinkingChunk(content="more") + yield "world" + + result = await adapter.stream(THREAD_ID, _chunks()) + assert result.id == "mid.t" + body = adapter._graph_api_fetch.call_args.kwargs["body"] + assert body["message"]["text"] == "Hello world" + # --------------------------------------------------------------------------- # Typing indicator diff --git a/tests/test_slack_api.py b/tests/test_slack_api.py index 8f7b9ec7..0d5c3f65 100644 --- a/tests/test_slack_api.py +++ b/tests/test_slack_api.py @@ -1213,6 +1213,84 @@ async def chunk_gen() -> AsyncIterator[StreamChunk | str]: assert result.id == "777.777" + @pytest.mark.asyncio + async def test_stream_skips_thinking_chunk_by_default(self): + """A ``ThinkingChunk`` must not change the posted message and must not + be sent as a structured chunk (which would disable structured-chunk + support for the rest of the stream). + + Python-only divergence: thinking is streaming-only reasoning, not + message content. By default the adapter skips it — the appended text + equals exactly the text chunks, and no ``{"type": "thinking"}`` + structured chunk is ever appended. + """ + adapter, client, _ = await _init_adapter() + + mock_streamer = MagicMock() + mock_streamer.append = AsyncMock() + mock_streamer.stop = AsyncMock(return_value={"message": {"ts": "555.555"}}) + client.chat_stream = AsyncMock(return_value=mock_streamer) + + from chat_sdk.types import ThinkingChunk + + async def chunk_gen() -> AsyncIterator[Any]: + yield ThinkingChunk(content="let me reason") + yield "Hello " + yield ThinkingChunk(content="still reasoning") + yield "world" + + result = await adapter.stream( + "slack:C123:1234567890.000000", + chunk_gen(), + StreamOptions(recipient_user_id="U1", recipient_team_id="T1"), + ) + + assert result.id == "555.555" + # The accumulated markdown is exactly the text chunks (no thinking). + appended_markdown = "".join( + call.kwargs.get("markdown_text", "") + for call in mock_streamer.append.call_args_list + if "markdown_text" in call.kwargs + ) + assert appended_markdown == "Hello world" + # No structured "thinking" chunk was ever appended. + for call in mock_streamer.append.call_args_list: + for chunk in call.kwargs.get("chunks", []) or []: + assert chunk.get("type") != "thinking" + + @pytest.mark.asyncio + async def test_stream_thinking_chunk_invokes_render_hook(self): + """An opt-in ``render_thinking`` hook receives the reasoning text while + the posted message stays text-only.""" + adapter, client, _ = await _init_adapter() + + mock_streamer = MagicMock() + mock_streamer.append = AsyncMock() + mock_streamer.stop = AsyncMock(return_value={"message": {"ts": "444.444"}}) + client.chat_stream = AsyncMock(return_value=mock_streamer) + + seen: list[str] = [] + + async def render_thinking(content: str) -> None: + seen.append(content) + + adapter.render_thinking = render_thinking # type: ignore[attr-defined] + + from chat_sdk.types import ThinkingChunk + + async def chunk_gen() -> AsyncIterator[Any]: + yield ThinkingChunk(content="thought A") + yield "answer" + yield ThinkingChunk(content="thought B") + + await adapter.stream( + "slack:C123:1234567890.000000", + chunk_gen(), + StreamOptions(recipient_user_id="U1", recipient_team_id="T1"), + ) + + assert seen == ["thought A", "thought B"] + @pytest.mark.asyncio async def test_stream_accepts_dict_markdown_text_chunks(self): """Dict-shaped markdown_text chunks must flow into the renderer the diff --git a/tests/test_teams_native_streaming.py b/tests/test_teams_native_streaming.py index a5790baf..0742345a 100644 --- a/tests/test_teams_native_streaming.py +++ b/tests/test_teams_native_streaming.py @@ -251,6 +251,58 @@ async def gen(): assert streamer.emitted == ["kept"] assert result.raw["text"] == "kept" + @pytest.mark.asyncio + async def test_skips_thinking_chunk_by_default(self): + """``ThinkingChunk`` is streaming-only reasoning, not message content. + + The default emit path skips it: it is never emitted to the SDK + ``IStreamer`` and the recorded text equals the text chunks only. + """ + from chat_sdk.types import ThinkingChunk + + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = FakeStreamer() + await _register_streamer(adapter, tid, streamer) + + async def gen(): + yield ThinkingChunk(content="reasoning A") + yield "Hello " + yield ThinkingChunk(content="reasoning B") + yield "World" + + result = await adapter.stream(tid, gen()) + assert streamer.emitted == ["Hello ", "World"] + assert result.raw["text"] == "Hello World" + + @pytest.mark.asyncio + async def test_thinking_chunk_invokes_render_hook(self): + """An opt-in ``render_thinking`` hook receives the reasoning while the + emitted text stays thinking-free.""" + from chat_sdk.types import ThinkingChunk + + adapter = _make_adapter() + tid = _dm_thread_id(adapter) + streamer = FakeStreamer() + await _register_streamer(adapter, tid, streamer) + + seen: list[str] = [] + + async def render_thinking(content: str) -> None: + seen.append(content) + + adapter.render_thinking = render_thinking # type: ignore[attr-defined] + + async def gen(): + yield ThinkingChunk(content="thought A") + yield "text" + yield ThinkingChunk(content="thought B") + + result = await adapter.stream(tid, gen()) + assert seen == ["thought A", "thought B"] + assert streamer.emitted == ["text"] + assert result.raw["text"] == "text" + @pytest.mark.asyncio async def test_never_calls_close(self): """``stream()`` / ``_stream_via_emit`` must NEVER call ``stream.close()``. diff --git a/tests/test_thinking_chunk.py b/tests/test_thinking_chunk.py new file mode 100644 index 00000000..941f7370 --- /dev/null +++ b/tests/test_thinking_chunk.py @@ -0,0 +1,340 @@ +"""Integration tests for the Python-only ``ThinkingChunk`` divergence. + +Covers the cross-cutting guarantees of the opt-in thinking-stream feature: + +- ``Thread._handle_stream`` gracefully *skips* a ``ThinkingChunk`` so the + posted message is byte-identical with or without thinking in the stream. +- The thread-level ``emit_thinking`` flag (default-off) threads through to + ``_from_full_stream`` so a raw AI-SDK ``reasoning`` part becomes a + ``ThinkingChunk`` only when enabled. +- The persisted ``Message`` round-trips byte-identically — no thinking ever + reaches history/state. +- Every adapter's stream handler ignores a ``ThinkingChunk`` without crashing. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from datetime import datetime, timezone +from typing import Any + +import pytest + +from chat_sdk.testing import create_mock_adapter, create_mock_state +from chat_sdk.thread import ThreadImpl, _ThreadImplConfig, _to_message +from chat_sdk.types import ( + Author, + MarkdownTextChunk, + Message, + MessageMetadata, + PostableMarkdown, + RawMessage, + ThinkingChunk, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_thread(*, emit_thinking: bool = False, adapter: Any = None) -> ThreadImpl: + return ThreadImpl( + _ThreadImplConfig( + id="slack:C123:1234.5678", + adapter=adapter or create_mock_adapter(), + state_adapter=create_mock_state(), + channel_id="C123", + emit_thinking=emit_thinking, + # Edit on every chunk so the fallback path's accumulated text is + # fully exercised. + streaming_update_interval_ms=0, + ) + ) + + +async def _stream(items: list[Any]) -> AsyncIterator[Any]: + for item in items: + yield item + + +# --------------------------------------------------------------------------- +# _handle_stream graceful skip (fallback post+edit path) +# --------------------------------------------------------------------------- + + +class TestHandleStreamGracefulSkip: + @pytest.mark.asyncio + async def test_posted_message_identical_with_and_without_thinking(self): + # Same text chunks, one stream interleaved with ThinkingChunks. + thread_plain = _make_thread() + plain = await thread_plain.post(_stream(["Hello ", "world"])) + + thread_thinking = _make_thread() + with_thinking = await thread_thinking.post( + _stream( + [ + ThinkingChunk(content="reasoning A"), + "Hello ", + ThinkingChunk(content="reasoning B"), + "world", + ] + ) + ) + + assert plain.text == with_thinking.text == "Hello world" + + @pytest.mark.asyncio + async def test_thinking_chunk_does_not_appear_in_posted_text(self): + thread = _make_thread() + sent = await thread.post(_stream([ThinkingChunk(content="SECRET-THOUGHT"), "visible"])) + assert "SECRET-THOUGHT" not in sent.text + assert sent.text == "visible" + + @pytest.mark.asyncio + async def test_thinking_only_stream_posts_no_thinking_text(self): + thread = _make_thread() + sent = await thread.post(_stream([ThinkingChunk(content="just thinking")])) + assert "just thinking" not in sent.text + + +# --------------------------------------------------------------------------- +# emit_thinking flag threading through _handle_stream -> adapter.stream +# --------------------------------------------------------------------------- + + +class TestEmitThinkingThreading: + @pytest.mark.asyncio + async def test_default_off_drops_raw_reasoning_before_adapter(self): + # A native-stream adapter records every chunk it receives. With the + # default-off flag, raw reasoning parts must NOT reach it. + received: list[Any] = [] + + async def mock_stream(thread_id: str, text_stream: Any, options: Any = None) -> RawMessage: + async for chunk in text_stream: + received.append(chunk) + return RawMessage(id="m1", thread_id=thread_id, raw={}) + + adapter = create_mock_adapter() + adapter.stream = mock_stream # type: ignore[attr-defined] + + thread = _make_thread(adapter=adapter) # emit_thinking defaults to False + await thread.post( + _stream( + [ + {"type": "reasoning", "text": "hidden"}, + {"type": "text-delta", "text": "answer"}, + ] + ) + ) + + assert all(not isinstance(c, ThinkingChunk) for c in received) + assert "answer" in received + + @pytest.mark.asyncio + async def test_opt_in_surfaces_thinking_chunk_to_adapter(self): + received: list[Any] = [] + + async def mock_stream(thread_id: str, text_stream: Any, options: Any = None) -> RawMessage: + async for chunk in text_stream: + received.append(chunk) + return RawMessage(id="m1", thread_id=thread_id, raw={}) + + adapter = create_mock_adapter() + adapter.stream = mock_stream # type: ignore[attr-defined] + + thread = _make_thread(adapter=adapter, emit_thinking=True) + sent = await thread.post( + _stream( + [ + {"type": "reasoning", "text": "thinking"}, + {"type": "text-delta", "text": "answer"}, + ] + ) + ) + + thinking = [c for c in received if isinstance(c, ThinkingChunk)] + assert len(thinking) == 1 + assert thinking[0].content == "thinking" + # Even with thinking surfaced, the posted message text excludes it. + assert sent.text == "answer" + + @pytest.mark.asyncio + async def test_opt_in_passes_prebuilt_thinking_chunk_through(self): + received: list[Any] = [] + + async def mock_stream(thread_id: str, text_stream: Any, options: Any = None) -> RawMessage: + async for chunk in text_stream: + received.append(chunk) + return RawMessage(id="m1", thread_id=thread_id, raw={}) + + adapter = create_mock_adapter() + adapter.stream = mock_stream # type: ignore[attr-defined] + + thread = _make_thread(adapter=adapter, emit_thinking=True) + await thread.post(_stream([ThinkingChunk(content="pre"), "text"])) + + assert any(isinstance(c, ThinkingChunk) and c.content == "pre" for c in received) + + +# --------------------------------------------------------------------------- +# No state pollution: persisted Message round-trips byte-identically +# --------------------------------------------------------------------------- + + +def _make_message(text: str) -> Message: + return Message( + id="m1", + thread_id="slack:C123:1234.5678", + text=text, + formatted={"type": "root", "children": []}, + author=Author(user_id="U1", user_name="a", full_name="A", is_bot=True, is_me=True), + metadata=MessageMetadata(date_sent=datetime(2024, 1, 1, tzinfo=timezone.utc), edited=False), + ) + + +class TestNoStatePollution: + def test_message_has_no_thinking_field(self): + msg = _make_message("hi") + assert not hasattr(msg, "thinking") + assert "thinking" not in msg.to_json() + + def test_message_roundtrip_byte_identical(self): + msg = _make_message("response body") + serialized = msg.to_json() + restored = Message.from_json(serialized) + # Re-serialization must be byte-identical: thinking never enters state. + assert restored.to_json() == serialized + assert "thinking" not in restored.to_json() + + @pytest.mark.asyncio + async def test_streamed_message_persisted_without_thinking(self): + state = create_mock_state() + thread = ThreadImpl( + _ThreadImplConfig( + id="slack:C123:1234.5678", + adapter=create_mock_adapter(), + state_adapter=state, + channel_id="C123", + streaming_update_interval_ms=0, + ) + ) + sent = await thread.post(_stream([ThinkingChunk(content="SECRET"), "posted text"])) + persisted = _to_message(sent) + assert "SECRET" not in persisted.to_json() + assert persisted.text == "posted text" + + +# --------------------------------------------------------------------------- +# from_full_stream pre-built ThinkingChunk passthrough symmetry +# --------------------------------------------------------------------------- + + +class TestStreamChunkUnion: + def test_markdown_and_thinking_share_no_discriminant(self): + assert MarkdownTextChunk().type == "markdown_text" + assert ThinkingChunk().type == "thinking" + + @pytest.mark.asyncio + async def test_postable_markdown_unaffected_by_thinking(self): + # Sanity: posting a normal PostableMarkdown is unchanged. + thread = _make_thread() + sent = await thread.post(PostableMarkdown(markdown="plain message")) + assert sent.text == "plain message" + + +# --------------------------------------------------------------------------- +# Every adapter's text-accumulate stream handler ignores a ThinkingChunk +# without crashing. (Slack/Teams native-stream paths + Twilio/Messenger are +# covered in their own adapter test modules.) +# --------------------------------------------------------------------------- + + +def _build_text_accumulate_adapters() -> list[tuple[str, Any, str]]: + """Construct the text-accumulate adapters with their encoded thread IDs. + + Returns ``(name, adapter, thread_id)`` triples. Each adapter's ``stream`` + accumulates only ``str`` / ``markdown_text`` text and posts via + ``post_message`` / ``edit_message``, which we stub per-adapter so no + network call is made. + """ + from chat_sdk.logger import ConsoleLogger + + triples: list[tuple[str, Any, str]] = [] + + # Discord + from chat_sdk.adapters.discord.adapter import DiscordAdapter + from chat_sdk.adapters.discord.types import DiscordAdapterConfig, DiscordThreadId + + discord = DiscordAdapter(DiscordAdapterConfig(bot_token="t", public_key="a" * 64, application_id="app")) + triples.append(("discord", discord, discord.encode_thread_id(DiscordThreadId(guild_id="g", channel_id="c")))) + + # WhatsApp + from chat_sdk.adapters.whatsapp.adapter import WhatsAppAdapter + from chat_sdk.adapters.whatsapp.types import WhatsAppAdapterConfig, WhatsAppThreadId + + whatsapp = WhatsAppAdapter( + WhatsAppAdapterConfig( + access_token="t", + app_secret="s", + phone_number_id="111", + verify_token="v", + user_name="bot", + logger=ConsoleLogger("error"), + ) + ) + triples.append( + ("whatsapp", whatsapp, whatsapp.encode_thread_id(WhatsAppThreadId(phone_number_id="111", user_wa_id="222"))) + ) + + # GitHub + from chat_sdk.adapters.github.adapter import GitHubAdapter + from chat_sdk.adapters.github.types import GitHubThreadId + + github = GitHubAdapter({"webhook_secret": "s", "token": "ghp_t", "logger": ConsoleLogger("error")}) + triples.append(("github", github, github.encode_thread_id(GitHubThreadId(owner="o", repo="r", pr_number=1)))) + + # Linear + from chat_sdk.adapters.linear.adapter import LinearAdapter + from chat_sdk.adapters.linear.types import LinearAdapterAPIKeyConfig, LinearThreadId + + linear = LinearAdapter( + LinearAdapterAPIKeyConfig(api_key="k", webhook_secret="s", user_name="bot", logger=ConsoleLogger("error")) + ) + triples.append(("linear", linear, linear.encode_thread_id(LinearThreadId(issue_id="abc123-def456-789")))) + + # Stub the network-touching post/edit on each adapter. + for _name, adapter, _tid in triples: + captured: dict[str, str] = {"text": ""} + + async def _post(thread_id: str, message: Any, _cap: dict[str, str] = captured) -> RawMessage: + _cap["text"] = getattr(message, "markdown", None) or getattr(message, "raw", "") or "" + return RawMessage(id="posted", thread_id=thread_id, raw={}) + + async def _edit(thread_id: str, message_id: str, message: Any, _cap: dict[str, str] = captured) -> RawMessage: + _cap["text"] = getattr(message, "markdown", None) or getattr(message, "raw", "") or "" + return RawMessage(id=message_id, thread_id=thread_id, raw={}) + + adapter.post_message = _post # type: ignore[attr-defined] + adapter.edit_message = _edit # type: ignore[attr-defined] + adapter._captured = captured # type: ignore[attr-defined] + + return triples + + +@pytest.mark.parametrize("name,adapter,thread_id", _build_text_accumulate_adapters()) +@pytest.mark.asyncio +async def test_adapter_stream_ignores_thinking_chunk(name: str, adapter: Any, thread_id: str) -> None: + """Each text-accumulate adapter's ``stream`` skips a ``ThinkingChunk`` + without crashing, and the posted text contains only the text chunks.""" + + async def gen() -> AsyncIterator[Any]: + yield ThinkingChunk(content="reasoning") + yield "Hello " + yield ThinkingChunk(content="more reasoning") + yield "world" + + # Must not raise. + await adapter.stream(thread_id, gen()) + posted = adapter._captured["text"] + assert "reasoning" not in posted, f"{name}: thinking leaked into posted text" + assert posted == "Hello world", f"{name}: unexpected posted text {posted!r}" diff --git a/tests/test_twilio_adapter.py b/tests/test_twilio_adapter.py index e2014a89..7eea8471 100644 --- a/tests/test_twilio_adapter.py +++ b/tests/test_twilio_adapter.py @@ -468,6 +468,37 @@ async def chunks() -> Any: http.assert_awaited_once() assert _sent_form(http)["Body"] == "hello world" + @pytest.mark.asyncio + async def test_stream_ignores_thinking_chunk(self): + """A ``ThinkingChunk`` is streaming-only reasoning, not message content. + + The text-accumulating adapter must skip it without crashing and the + posted body must equal the text chunks only. + """ + from chat_sdk.types import ThinkingChunk + + http = _mock_http( + { + "body": "hello world", + "direction": "outbound-api", + "from": "+15550000001", + "sid": "SM124", + "to": "+15550000002", + } + ) + adapter = create_twilio_adapter(account_sid="AC123", auth_token="token", http_request=http) + + async def chunks() -> Any: + yield ThinkingChunk(content="reasoning") + yield "hello " + yield ThinkingChunk(content="more reasoning") + yield "world" + + result = await adapter.stream("twilio:%2B15550000001:%2B15550000002", chunks()) + + assert result.id == "SM124" + assert _sent_form(http)["Body"] == "hello world" + class TestFetchMessages: """fetch_message / fetch_messages over the Messages API.""" diff --git a/tests/test_types.py b/tests/test_types.py index 9f5f605d..74c6c22e 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -14,11 +14,15 @@ Author, EmojiFormats, EmojiValue, + MarkdownTextChunk, Message, MessageMetadata, MessageSubject, MessageSubjectParty, + PlanUpdateChunk, RawMessage, + TaskUpdateChunk, + ThinkingChunk, _get_message_adapter, _message_adapter_map, set_message_adapter, @@ -632,3 +636,40 @@ def test_re_registered_message_cleans_up_exactly_once(self): del msg gc.collect() assert key not in _message_adapter_map + + +class TestThinkingChunk: + """Tests for the Python-only ``ThinkingChunk`` StreamChunk variant.""" + + def test_defaults(self): + chunk = ThinkingChunk() + assert chunk.type == "thinking" + assert chunk.content == "" + + def test_with_content(self): + chunk = ThinkingChunk(content="Analyzing the user's request") + assert chunk.type == "thinking" + assert chunk.content == "Analyzing the user's request" + + def test_type_discriminant_is_fixed(self): + # The discriminant is what adapters/normalizers switch on; it must be + # exactly "thinking" so it is never confused with the other variants. + assert ThinkingChunk(content="x").type == "thinking" + assert ThinkingChunk().type != MarkdownTextChunk().type + assert ThinkingChunk().type != TaskUpdateChunk().type + assert ThinkingChunk().type != PlanUpdateChunk().type + + def test_is_member_of_streamchunk_union(self): + # Structural runtime check that ThinkingChunk is one of the StreamChunk + # variants alongside the upstream three. + from typing import get_args + + from chat_sdk.types import StreamChunk + + variants = (MarkdownTextChunk, TaskUpdateChunk, PlanUpdateChunk, ThinkingChunk) + assert set(get_args(StreamChunk)) == set(variants) + assert isinstance(ThinkingChunk(content="reasoning"), variants) + + def test_equality_by_value(self): + assert ThinkingChunk(content="a") == ThinkingChunk(content="a") + assert ThinkingChunk(content="a") != ThinkingChunk(content="b") From 0c52cbb3f7d710950460bc6081c0fee88c7be250 Mon Sep 17 00:00:00 2001 From: patrick-chinchill Date: Fri, 19 Jun 2026 23:26:47 -0700 Subject: [PATCH 2/2] refactor(core): keep StreamChunk upstream-exact, ThinkingChunk as opt-in StreamInput Revert the public ``StreamChunk`` union to upstream's three variants (``MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk``) so a consumer doing an exhaustive ``match`` over it sees zero change on upgrade. The Python-only ``ThinkingChunk`` is no longer a member of that union. Introduce a public ``StreamInput = str | StreamChunk | ThinkingChunk`` alias for what a stream may yield, and widen only the stream-input/output boundaries to accept it: the ``Adapter.stream()`` protocol signature, ``from_full_stream``/``_from_full_stream`` returns, ``Thread._wrapped_stream``, and each receiving adapter's ``stream()`` signature. A producer can yield ``ThinkingChunk`` (opt-in) and the adapters type-check; code referencing ``StreamChunk`` itself is unaffected. All other thinking behavior is preserved byte-for-byte: ``emit_thinking`` defaults to False (default stream identical to upstream), the ``_handle_stream`` graceful skip, the adapter skip / ``render_thinking`` branches, ``maybe_render_thinking``, and the no-``Message.thinking``-field / round-trip-identical guarantee. The UPSTREAM_SYNC divergence row is updated to state that ``StreamChunk`` is NOT widened and ``ThinkingChunk`` is a separate opt-in input type. Tests now assert ``get_args(StreamChunk)`` has exactly the three upstream variants and that ``ThinkingChunk`` is excluded from ``StreamChunk`` but accepted by ``StreamInput``. --- docs/UPSTREAM_SYNC.md | 2 +- src/chat_sdk/__init__.py | 2 ++ src/chat_sdk/adapters/github/adapter.py | 4 +-- src/chat_sdk/adapters/google_chat/adapter.py | 4 +-- src/chat_sdk/adapters/messenger/adapter.py | 4 +-- src/chat_sdk/adapters/slack/adapter.py | 10 ++++-- src/chat_sdk/adapters/telegram/adapter.py | 4 +-- src/chat_sdk/adapters/twilio/adapter.py | 4 +-- src/chat_sdk/adapters/whatsapp/adapter.py | 4 +-- src/chat_sdk/from_full_stream.py | 11 +++--- src/chat_sdk/thread.py | 6 ++-- src/chat_sdk/types.py | 19 ++++++++-- tests/test_types.py | 38 ++++++++++++++++---- 13 files changed, 81 insertions(+), 31 deletions(-) diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index c058f081..403bc9e1 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -684,7 +684,7 @@ stay explicit instead of being rediscovered in code review. | Teams `cards_input` empty-options default (vercel/chat#8c71411, chat@4.31) | `input_request_to_teams_adaptive_card` reads `options = request.get("options") or []` | Upstream `cards-primitives/input.ts` reads `const options = request.options ?? []` | Benign truthiness divergence. The only values that differ between `or []` and `?? []` are the falsy-but-present ones (`[]`, `None`); for an options list both produce the same empty list, so the rendered card is byte-identical. Documented (not "fixed" to `is not None`) because there is no observable behavior difference — a present empty `options` and an absent `options` both yield "no choices". | | Teams `graph` path-segment encoding (vercel/chat#8c71411, chat@4.31) | Path segments (`team_id` / `channel_id` / `chat_id` / `message_id`) are interpolated through `quote(segment, safe='')` | Upstream `graph/{channels,messages}.ts` use `encodeURIComponent(...)` | Benign over-encoding divergence. `quote(safe='')` percent-encodes a strictly larger set than `encodeURIComponent` (notably `!`, `'`, `(`, `)`, `*` — which `encodeURIComponent` leaves literal). Graph IDs (team/channel/chat/message GUIDs and thread tokens) never contain those characters, so the encoded path is identical in practice; where they did differ, the stricter `quote` is the safer choice (no URL injection via an unescaped sub-delimiter). Documented rather than narrowed to match `encodeURIComponent` exactly. | | Teams `graph` defensive shape coercion (vercel/chat#8c71411, chat@4.31) | `to_graph_message` / channel + message readers coerce unexpected Graph payload shapes with `isinstance` guards (`x if isinstance(x, Mapping) else {}`, `value if isinstance(value, list) else []`) before reading fields | Upstream `graph/messages.ts` reads `message.from?.user` / `message.body?.content ?? ""` etc. with optional chaining — a non-object where an object is expected throws at the property access | Benign defensive divergence. Upstream's optional chaining tolerates `null`/`undefined` but throws on a wrong-typed non-null (e.g. a string where an object is expected); our `isinstance` coercion fails closed to an empty mapping/list instead of raising. For well-formed Graph responses the behavior is identical; the divergence only manifests on malformed payloads, where returning an empty-shape result is more resilient than throwing. Mirrors the repo's general "more resilient than throw" stance (cf. the `renderPostable on unknown input` row). | -| `ThinkingChunk` StreamChunk variant (Python-only, default-off; supersedes PR #39) | A fourth `StreamChunk` variant — `ThinkingChunk(type="thinking", content=str)` — surfaces AI-SDK `reasoning`/`reasoning-delta` (and pydantic-ai `part_kind == "thinking"`) parts. **OPT-IN, default-off**: emitted only when a caller passes `from_full_stream(stream, emit_thinking=True)` or sets the thread-level `emit_thinking=True` config; the internal `_from_full_stream` threads the same flag. With the default (`emit_thinking=False`) the normalized stream is **byte-for-byte identical** to upstream — reasoning parts are dropped and **no** `ThinkingChunk` is produced. Consumption is graceful: `Thread._handle_stream` never accumulates a `ThinkingChunk` into the posted-message text, and every adapter's stream handler skips it (Slack/Teams expose an optional `render_thinking` hook via `chat_sdk.shared.adapter_utils.maybe_render_thinking`; the text-accumulate adapters ignore it structurally). **Streaming-only — never persisted**: `Message` has no `thinking` field, `to_json()` is unchanged, and a round-tripped `Message` is byte-identical, so cross-SDK state (Redis/Postgres shared with the TS SDK) stays compatible. | Upstream `from-full-stream.ts` forwards only `text-delta` + `finish-step`; AI-SDK `reasoning`/`reasoning-delta` parts fall through and are discarded. `StreamChunk = MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk` — no reasoning variant. Upstream leaves reasoning display to the AI-SDK web UI. | chinchill actively streams agent thinking to Slack/Teams but has to intercept the model stream out-of-band today because the chat-platform SDK has no path for it. This gives the SDK a first-class, opt-in one without changing any default behavior. The whole design constraint is that default-off == upstream: additive type, opt-in emit, graceful/skip consume, zero state pollution. Regression coverage: `tests/test_thinking_chunk.py`, `tests/test_from_full_stream.py::TestThinkingOptIn`, `tests/test_types.py::TestThinkingChunk`, plus per-adapter no-crash tests in `tests/test_slack_api.py`, `tests/test_teams_native_streaming.py`, `tests/test_twilio_adapter.py`, `tests/test_messenger_api.py`. | +| `ThinkingChunk` opt-in stream-input type (Python-only, default-off; supersedes PR #39) | A **separate, opt-in** dataclass — `ThinkingChunk(type="thinking", content=str)` — surfaces AI-SDK `reasoning`/`reasoning-delta` (and pydantic-ai `part_kind == "thinking"`) parts. **`StreamChunk` is NOT widened**: the canonical union stays `StreamChunk = MarkdownTextChunk \| TaskUpdateChunk \| PlanUpdateChunk` — byte-identical to upstream's three variants, so a consumer doing an exhaustive `match` over `StreamChunk` sees zero change on upgrade. `ThinkingChunk` is accepted only at the **stream-input/output boundaries** via the public alias `StreamInput = str \| StreamChunk \| ThinkingChunk` (the `Adapter.stream()` protocol signature, `from_full_stream`/`_from_full_stream` returns, `Thread._wrapped_stream`, and each receiving adapter's `stream()` signature). A producer can yield `ThinkingChunk` (opt-in) and the adapters that receive the stream type-check; code that only references `StreamChunk` never touches it. **OPT-IN, default-off**: emitted only when a caller passes `from_full_stream(stream, emit_thinking=True)` or sets the thread-level `emit_thinking=True` config; the internal `_from_full_stream` threads the same flag. With the default (`emit_thinking=False`) the normalized stream is **byte-for-byte identical** to upstream — reasoning parts are dropped and **no** `ThinkingChunk` is produced. Consumption is graceful: `Thread._handle_stream` never accumulates a `ThinkingChunk` into the posted-message text, and every adapter's stream handler skips it (Slack/Teams expose an optional `render_thinking` hook via `chat_sdk.shared.adapter_utils.maybe_render_thinking`; the text-accumulate adapters ignore it structurally). **Streaming-only — never persisted**: `Message` has no `thinking` field, `to_json()` is unchanged, and a round-tripped `Message` is byte-identical, so cross-SDK state (Redis/Postgres shared with the TS SDK) stays compatible. | Upstream `from-full-stream.ts` forwards only `text-delta` + `finish-step`; AI-SDK `reasoning`/`reasoning-delta` parts fall through and are discarded. `StreamChunk = MarkdownTextChunk \| TaskUpdateChunk \| PlanUpdateChunk` — no reasoning variant and no stream-input alias. Upstream leaves reasoning display to the AI-SDK web UI. | chinchill actively streams agent thinking to Slack/Teams but has to intercept the model stream out-of-band today because the chat-platform SDK has no path for it. This gives the SDK a first-class, opt-in one without changing any default behavior — and crucially **without widening the public `StreamChunk` union**, so consumers referencing it are unaffected. The whole design constraint is that default-off == upstream and `StreamChunk` == upstream: separate opt-in input type, opt-in emit, graceful/skip consume, zero state pollution. Regression coverage: `tests/test_thinking_chunk.py`, `tests/test_from_full_stream.py::TestThinkingOptIn`, `tests/test_types.py::TestThinkingChunk`, plus per-adapter no-crash tests in `tests/test_slack_api.py`, `tests/test_teams_native_streaming.py`, `tests/test_twilio_adapter.py`, `tests/test_messenger_api.py`. | ### Platform-specific gaps | Area | Python | TS | Rationale | diff --git a/src/chat_sdk/__init__.py b/src/chat_sdk/__init__.py index f8c2a26b..f2742571 100644 --- a/src/chat_sdk/__init__.py +++ b/src/chat_sdk/__init__.py @@ -210,6 +210,7 @@ SlashCommandEvent, StateAdapter, StreamChunk, + StreamInput, StreamOptions, TaskUpdateChunk, ThinkingChunk, @@ -449,6 +450,7 @@ "SlashCommandEvent", "StateAdapter", "StreamChunk", + "StreamInput", "StreamOptions", "TaskUpdateChunk", "ThinkingChunk", diff --git a/src/chat_sdk/adapters/github/adapter.py b/src/chat_sdk/adapters/github/adapter.py index bf7630b6..351fad8a 100644 --- a/src/chat_sdk/adapters/github/adapter.py +++ b/src/chat_sdk/adapters/github/adapter.py @@ -55,7 +55,7 @@ MessageMetadata, PostableMarkdown, RawMessage, - StreamChunk, + StreamInput, StreamOptions, Thread, ThreadInfo, @@ -645,7 +645,7 @@ async def edit_message(self, thread_id: str, message_id: str, message: AdapterPo async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage: """Stream by accumulating text and posting once.""" diff --git a/src/chat_sdk/adapters/google_chat/adapter.py b/src/chat_sdk/adapters/google_chat/adapter.py index aec6ad16..f8aa1730 100644 --- a/src/chat_sdk/adapters/google_chat/adapter.py +++ b/src/chat_sdk/adapters/google_chat/adapter.py @@ -77,7 +77,7 @@ RawMessage, ReactionEvent, StateAdapter, - StreamChunk, + StreamInput, StreamOptions, ThreadInfo, ThreadSummary, @@ -1729,7 +1729,7 @@ async def delete_message(self, thread_id: str, message_id: str) -> None: async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage: """Stream by accumulating all chunks and posting as a single message.""" diff --git a/src/chat_sdk/adapters/messenger/adapter.py b/src/chat_sdk/adapters/messenger/adapter.py index 2d795a60..613476f1 100644 --- a/src/chat_sdk/adapters/messenger/adapter.py +++ b/src/chat_sdk/adapters/messenger/adapter.py @@ -70,7 +70,7 @@ PostableMarkdown, RawMessage, ReactionEvent, - StreamChunk, + StreamInput, StreamOptions, ThreadInfo, UserInfo, @@ -585,7 +585,7 @@ async def edit_message( async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage: """Buffer all stream chunks and send as a single message. diff --git a/src/chat_sdk/adapters/slack/adapter.py b/src/chat_sdk/adapters/slack/adapter.py index a479a1b6..b9b6574f 100644 --- a/src/chat_sdk/adapters/slack/adapter.py +++ b/src/chat_sdk/adapters/slack/adapter.py @@ -105,7 +105,9 @@ ScheduledMessage, SlashCommandEvent, StreamChunk, + StreamInput, StreamOptions, + ThinkingChunk, ThreadInfo, ThreadSummary, UserInfo, @@ -3878,7 +3880,7 @@ async def start_typing(self, thread_id: str, status: str | None = None) -> None: async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage: """Stream a message using Slack's native streaming API. @@ -3950,7 +3952,11 @@ async def flush_markdown_delta(delta: str) -> None: return await streamer.append(markdown_text=delta, token=token) - async def send_structured_chunk(chunk: StreamChunk | dict[str, Any]) -> None: + # Accepts the residual stream-input union: ``is_thinking_chunk`` filters + # ``ThinkingChunk`` out before this runs (it is never reached with one), + # but it stays in the static type since that runtime guard does not + # narrow it. The generic ``_read``-based body handles any chunk shape. + async def send_structured_chunk(chunk: StreamChunk | ThinkingChunk | dict[str, Any]) -> None: nonlocal last_appended, structured_chunks_supported if not structured_chunks_supported: return diff --git a/src/chat_sdk/adapters/telegram/adapter.py b/src/chat_sdk/adapters/telegram/adapter.py index 3d57edb0..b0af9b6b 100644 --- a/src/chat_sdk/adapters/telegram/adapter.py +++ b/src/chat_sdk/adapters/telegram/adapter.py @@ -89,7 +89,7 @@ RawMessage, ReactionEvent, SlashCommandEvent, - StreamChunk, + StreamInput, StreamOptions, ThreadInfo, UserInfo, @@ -1729,7 +1729,7 @@ async def start_typing(self, thread_id: str, status: str | None = None) -> None: async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage | None: """Stream a message to a Telegram private chat via draft updates. diff --git a/src/chat_sdk/adapters/twilio/adapter.py b/src/chat_sdk/adapters/twilio/adapter.py index f49354bf..60faf05d 100644 --- a/src/chat_sdk/adapters/twilio/adapter.py +++ b/src/chat_sdk/adapters/twilio/adapter.py @@ -85,7 +85,7 @@ MessageMetadata, PostableMarkdown, RawMessage, - StreamChunk, + StreamInput, StreamOptions, ThreadInfo, UserInfo, @@ -327,7 +327,7 @@ async def start_typing(self, thread_id: str, status: str | None = None) -> None: async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage: """Buffer all stream chunks and send as a single message. diff --git a/src/chat_sdk/adapters/whatsapp/adapter.py b/src/chat_sdk/adapters/whatsapp/adapter.py index c50a39b6..fe4c3a28 100644 --- a/src/chat_sdk/adapters/whatsapp/adapter.py +++ b/src/chat_sdk/adapters/whatsapp/adapter.py @@ -57,7 +57,7 @@ PostableMarkdown, RawMessage, ReactionEvent, - StreamChunk, + StreamInput, StreamOptions, ThreadInfo, UserInfo, @@ -887,7 +887,7 @@ async def edit_message( async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage: """Stream a message by buffering all chunks and sending as a single message.""" diff --git a/src/chat_sdk/from_full_stream.py b/src/chat_sdk/from_full_stream.py index 795979a9..2ecd1c20 100644 --- a/src/chat_sdk/from_full_stream.py +++ b/src/chat_sdk/from_full_stream.py @@ -25,7 +25,7 @@ from collections.abc import AsyncIterable, AsyncIterator -from chat_sdk.types import StreamChunk, ThinkingChunk +from chat_sdk.types import StreamInput, ThinkingChunk _STREAM_CHUNK_TYPES = frozenset({"markdown_text", "task_update", "plan_update"}) @@ -51,10 +51,11 @@ async def from_full_stream( stream: AsyncIterable[object], *, emit_thinking: bool = False, -) -> AsyncIterator[str | StreamChunk]: +) -> AsyncIterator[StreamInput]: """Normalize an async iterable stream for use with ``thread.post()``. - Yields either plain ``str`` chunks or ``StreamChunk`` objects. + Yields plain ``str`` chunks, canonical ``StreamChunk`` objects, or — only + when ``emit_thinking=True`` — the opt-in, Python-only ``ThinkingChunk``. Args: stream: The source async iterable (text stream, full stream, or @@ -91,7 +92,9 @@ async def from_full_stream( if not event_type: continue - # Pass through StreamChunk objects (includes pre-built ThinkingChunk). + # Pass through canonical StreamChunk objects. (Pre-built ThinkingChunk + # has ``type == "thinking"`` and is handled by the reasoning branch + # below, gated on ``emit_thinking``.) if event_type in _STREAM_CHUNK_TYPES: yield event # type: ignore[misc] continue diff --git a/src/chat_sdk/thread.py b/src/chat_sdk/thread.py index 53ed808b..182e7829 100644 --- a/src/chat_sdk/thread.py +++ b/src/chat_sdk/thread.py @@ -44,7 +44,7 @@ ScheduledMessage, SentMessage, StateAdapter, - StreamChunk, + StreamInput, StreamOptions, TaskUpdateChunk, ThinkingChunk, @@ -732,7 +732,7 @@ async def _handle_stream( if hasattr(self.adapter, "stream") and self.adapter.stream: # type: ignore[union-attr] accumulated = "" - async def _wrapped_stream() -> AsyncGenerator[str | StreamChunk, None]: + async def _wrapped_stream() -> AsyncGenerator[StreamInput, None]: nonlocal accumulated async for chunk in text_stream: if isinstance(chunk, str): @@ -1297,7 +1297,7 @@ async def _from_full_stream( raw_stream: Any, *, emit_thinking: bool = False, -) -> AsyncIterator[str | StreamChunk]: +) -> AsyncIterator[StreamInput]: """Normalise a raw async iterable into str or StreamChunk items. Handles plain strings, AI SDK fullStream events, and StreamChunk objects. diff --git a/src/chat_sdk/types.py b/src/chat_sdk/types.py index 060e5047..8482b826 100644 --- a/src/chat_sdk/types.py +++ b/src/chat_sdk/types.py @@ -1034,7 +1034,22 @@ class ThinkingChunk: content: str = "" -StreamChunk = MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk | ThinkingChunk +StreamChunk = MarkdownTextChunk | TaskUpdateChunk | PlanUpdateChunk +"""Canonical stream-chunk union — **byte-identical to upstream chat@4.31's three +variants**. A consumer doing an exhaustive ``match`` over ``StreamChunk`` sees +zero change on upgrade; the Python-only :class:`ThinkingChunk` is *not* a member +(see :data:`StreamInput`).""" + + +StreamInput = str | StreamChunk | ThinkingChunk +"""What a stream may *yield* at the stream-input/output boundaries. + +This is the canonical :data:`StreamChunk` union plus plain ``str`` text and the +Python-only, **opt-in** :class:`ThinkingChunk`. ``ThinkingChunk`` is accepted +only here — a producer may yield it when thinking is enabled, and adapters that +receive the stream type-check against it — but it is deliberately kept out of +:data:`StreamChunk` itself so consumers referencing that union are unaffected. +""" @dataclass @@ -1511,7 +1526,7 @@ def persist_thread_history(self) -> bool | None: async def stream( self, thread_id: str, - text_stream: AsyncIterable[str | StreamChunk], + text_stream: AsyncIterable[StreamInput], options: StreamOptions | None = None, ) -> RawMessage | None: """Stream a message using platform-native streaming APIs. diff --git a/tests/test_types.py b/tests/test_types.py index 74c6c22e..515d5ec3 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -639,7 +639,7 @@ def test_re_registered_message_cleans_up_exactly_once(self): class TestThinkingChunk: - """Tests for the Python-only ``ThinkingChunk`` StreamChunk variant.""" + """Tests for the Python-only, opt-in ``ThinkingChunk`` stream-input type.""" def test_defaults(self): chunk = ThinkingChunk() @@ -659,16 +659,40 @@ def test_type_discriminant_is_fixed(self): assert ThinkingChunk().type != TaskUpdateChunk().type assert ThinkingChunk().type != PlanUpdateChunk().type - def test_is_member_of_streamchunk_union(self): - # Structural runtime check that ThinkingChunk is one of the StreamChunk - # variants alongside the upstream three. + def test_streamchunk_union_is_upstream_exact(self): + # ``StreamChunk`` must stay byte-identical to upstream's THREE variants. + # ``ThinkingChunk`` is deliberately NOT a member: a consumer doing an + # exhaustive ``match`` over ``StreamChunk`` must see zero change on + # upgrade. from typing import get_args from chat_sdk.types import StreamChunk - variants = (MarkdownTextChunk, TaskUpdateChunk, PlanUpdateChunk, ThinkingChunk) - assert set(get_args(StreamChunk)) == set(variants) - assert isinstance(ThinkingChunk(content="reasoning"), variants) + upstream_variants = (MarkdownTextChunk, TaskUpdateChunk, PlanUpdateChunk) + assert set(get_args(StreamChunk)) == set(upstream_variants) + # Exactly three — no fourth (thinking) variant leaked into the union. + assert len(get_args(StreamChunk)) == 3 + assert ThinkingChunk not in get_args(StreamChunk) + assert not isinstance(ThinkingChunk(content="reasoning"), upstream_variants) + + def test_thinking_chunk_is_accepted_by_stream_input(self): + # ``ThinkingChunk`` is opt-in input only: it lives in ``StreamInput`` + # (what a stream may yield), NOT in the canonical ``StreamChunk`` union. + from typing import get_args + + from chat_sdk.types import StreamChunk, StreamInput + + # StreamInput = str | StreamChunk | ThinkingChunk. ``get_args`` flattens + # the nested ``StreamChunk`` alias into its three members, so the input + # alias resolves to str + the three canonical variants + ThinkingChunk. + input_args = set(get_args(StreamInput)) + assert input_args == {str, *get_args(StreamChunk), ThinkingChunk} + assert str in input_args + assert ThinkingChunk in input_args + # Every canonical ``StreamChunk`` variant is reachable via the input. + assert set(get_args(StreamChunk)) <= input_args + # ...but the canonical union itself never gains the thinking variant. + assert ThinkingChunk not in get_args(StreamChunk) def test_equality_by_value(self): assert ThinkingChunk(content="a") == ThinkingChunk(content="a")