feat(core): optional ThinkingChunk StreamChunk variant (default-off, upstream-compatible; supersedes #39)

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` 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 |

src/chat_sdk/__init__.py

@@ -210,8 +210,10 @@
     SlashCommandEvent,
     StateAdapter,
     StreamChunk,
+    StreamInput,
     StreamOptions,
     TaskUpdateChunk,
+    ThinkingChunk,
     Thread,
     ThreadInfo,
     ThreadSummary,
@@ -448,8 +450,10 @@
     "SlashCommandEvent",
     "StateAdapter",
     "StreamChunk",
+    "StreamInput",
     "StreamOptions",
     "TaskUpdateChunk",
+    "ThinkingChunk",
     "Thread",
     "ThreadInfo",
     "ThreadSummary",

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."""

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."""

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.

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,
@@ -100,7 +105,9 @@
     ScheduledMessage,
     SlashCommandEvent,
     StreamChunk,
+    StreamInput,
     StreamOptions,
+    ThinkingChunk,
     ThreadInfo,
     ThreadSummary,
     UserInfo,
@@ -3873,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.
@@ -3945,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
@@ -4006,6 +4017,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)
 

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

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.

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.

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."""

src/chat_sdk/from_full_stream.py

@@ -11,23 +11,63 @@
 - **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 StreamInput, 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],
-) -> AsyncIterator[str | StreamChunk]:
+    *,
+    emit_thinking: bool = False,
+) -> 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
+            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 +92,26 @@ async def from_full_stream(
         if not event_type:
             continue
 
-        # Pass through StreamChunk objects
+        # 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
 
+        # 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: