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

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

Author: patrick-chinchill

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 |

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

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)
 

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/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:

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