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

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

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

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