fix(streaming-plan): address PR #74 review feedback

- StreamingPlanOptions mapping: replace truthy checks with `is not None`
  so update_interval_ms=0 and end_with=[] propagate correctly (thread.py
  post dispatcher and _fallback_stream interval guard). Per CLAUDE.md
  Port Rule #1; diverges from upstream thread.ts, which has the same
  latent bug.
- StreamingPlan: raise RuntimeError from is_supported() and
  get_fallback_text() so misroutes through post_postable_object /
  Channel.post fail loudly instead of posting "" or attempting a
  wrong-shape adapter.post_object("stream", ...). Also guard
  post_postable_object() with an early kind=="stream" check for a
  clearer error message. Diverges from upstream postable-object.ts,
  which silently posts the empty fallback string.
- Fallback test: spy on _fallback_stream to capture the StreamOptions
  actually reaching the fallback path and assert task_display_mode,
  stop_blocks, and update_interval_ms all propagate -- previous test
  passed even when options were silently dropped.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: patrick-chinchill

src/chat_sdk/plan.py

@@ -168,6 +168,20 @@ async def post_postable_object(
         Optional logger for error reporting.
     """
 
+    # StreamingPlan (kind == "stream") is a nominal PostableObject that only
+    # Thread.post() knows how to consume (via its native-or-fallback streaming
+    # path). Reject it here so callers get a clear error instead of blank
+    # posts or a wrong-shape adapter.post_object("stream", ...) call. Diverges
+    # from upstream postable-object.ts, which posts ``getFallbackText() == ""``
+    # as an empty message.
+    if getattr(obj, "kind", None) == "stream":
+        raise RuntimeError(
+            "StreamingPlan cannot be posted via post_postable_object / "
+            "Channel.post -- its stream is consumed only by Thread.post(), "
+            "which special-cases kind=='stream' for native or fallback "
+            "streaming. Use thread.post(streaming_plan) instead."
+        )
+
     def _make_context(raw: Any) -> PostableObjectContext:
         return PostableObjectContext(
             adapter=adapter,
@@ -518,20 +532,52 @@ def options(self) -> StreamingPlanOptions:
         return self._options
 
     # -- PostableObject protocol ------------------------------------------------
+    #
+    # StreamingPlan is a "nominal" PostableObject: it satisfies the duck-typing
+    # protocol (so ``is_postable_object()`` detects it and ``Thread.post``'s
+    # ``kind == "stream"`` branch fires), but it cannot actually round-trip
+    # through the generic ``post_postable_object`` helper -- there is no static
+    # fallback text to post and no meaningful ``adapter.post_object("stream",
+    # ...)`` shape.
+    #
+    # Upstream TS has the same latent gap: ``ChannelImpl.post`` routes any
+    # PostableObject through ``postPostableObject``, which would post an empty
+    # string for StreamingPlan. We diverge by failing loudly rather than
+    # silently posting blanks, per CLAUDE.md adversarial-review discipline.
 
     def get_fallback_text(self) -> str:
-        """StreamingPlan has no static fallback text -- streaming is handled
-        by Thread.post() directly via the native-or-fallback streaming path."""
-        return ""
+        """StreamingPlan has no static fallback text.
+
+        Raises ``RuntimeError`` to fail loudly if a generic posting path
+        (e.g. ``Channel.post`` or ``post_postable_object``) tries to
+        consume a StreamingPlan as a normal PostableObject. StreamingPlan
+        must be posted via :meth:`Thread.post`, which special-cases
+        ``kind == "stream"`` and consumes the wrapped async iterable.
+        """
+        raise RuntimeError(
+            "StreamingPlan cannot be posted via the generic PostableObject "
+            "path (no static fallback text). Post it with Thread.post(), "
+            "which routes kind=='stream' to native or fallback streaming."
+        )
 
     def get_post_data(self) -> _StreamingPlanData:
         """Return the underlying stream + options for Thread.post to route."""
         return _StreamingPlanData(stream=self._stream, options=self._options)
 
     def is_supported(self, _adapter: Adapter) -> bool:
-        """StreamingPlan is always supported -- fallback path handles adapters
-        without native streaming."""
-        return True
+        """StreamingPlan is not generically postable -- see
+        :meth:`get_fallback_text`.
+
+        Raises ``RuntimeError`` so misroutes through
+        ``post_postable_object`` fail loudly rather than silently trying
+        ``adapter.post_object("stream", ...)`` on adapters that don't
+        understand the shape.
+        """
+        raise RuntimeError(
+            "StreamingPlan cannot be posted via the generic PostableObject "
+            "path. Post it with Thread.post(), which routes kind=='stream' "
+            "to native or fallback streaming."
+        )
 
     def on_posted(self, _context: PostableObjectContext) -> None:
         """Streams are one-shot, no lifecycle binding needed."""

src/chat_sdk/thread.py

@@ -472,11 +472,16 @@ async def post(
                     group_tasks = getattr(plan_options, "group_tasks", None)
                     end_with = getattr(plan_options, "end_with", None)
                     update_interval_ms = getattr(plan_options, "update_interval_ms", None)
-                    if group_tasks:
+                    # Port Rule #1: use `is not None` so explicit falsy values
+                    # (``end_with=[]``, ``update_interval_ms=0``) still
+                    # propagate to the adapter/fallback instead of being
+                    # silently dropped by a truthiness check. Diverges from
+                    # upstream thread.ts, which has the same latent bug.
+                    if group_tasks is not None:
                         extra.task_display_mode = group_tasks
-                    if end_with:
+                    if end_with is not None:
                         extra.stop_blocks = end_with
-                    if update_interval_ms:
+                    if update_interval_ms is not None:
                         extra.update_interval_ms = update_interval_ms
                 await self._handle_stream(stream_iter, extra_options=extra)
                 return message
@@ -650,8 +655,13 @@ async def _fallback_stream(
         Posts an initial placeholder, then edits the message at intervals as
         new text arrives from the stream.
         """
+        # ``is not None`` so explicit ``update_interval_ms=0`` (edit-on-every-
+        # chunk) from ``StreamingPlan`` is honored rather than silently reset
+        # to the thread default by a truthiness check.
         interval_ms = (
-            options.update_interval_ms if options and options.update_interval_ms else self._streaming_update_interval_ms
+            options.update_interval_ms
+            if options is not None and options.update_interval_ms is not None
+            else self._streaming_update_interval_ms
         )
         interval_s = interval_ms / 1000.0
         placeholder_text = self._fallback_streaming_placeholder_text

tests/test_thread_faithful.py

@@ -949,6 +949,20 @@ async def test_should_route_streamingplan_through_fallback_when_adapter_has_no_n
         assert not hasattr(adapter, "stream") or getattr(adapter, "stream", None) is None
 
         thread = _make_thread(adapter, state)
+
+        # Capture StreamOptions at the fallback entry point so we can prove
+        # that StreamingPlanOptions actually reach the fallback path -- this
+        # guards issue #56 against silent truthiness drops that would leave
+        # observable streaming behavior unchanged.
+        captured_options: list[Any] = []
+        original_fallback = thread._fallback_stream
+
+        async def _spy_fallback(text_stream: Any, options: Any = None) -> Any:
+            captured_options.append(options)
+            return await original_fallback(text_stream, options)
+
+        thread._fallback_stream = _spy_fallback  # type: ignore[method-assign]
+
         text_stream = _create_text_stream(["Hello", " ", "World"])
         await thread.post(
             StreamingPlan(
@@ -971,6 +985,18 @@ async def test_should_route_streamingplan_through_fallback_when_adapter_has_no_n
         assert isinstance(last_edit[2], PostableMarkdown)
         assert last_edit[2].markdown == "Hello World"
 
+        # All three StreamingPlanOptions fields must reach the fallback path.
+        # Before the truthiness->`is not None` fix, end_with=[] and
+        # update_interval_ms=0 would be dropped; asserting full propagation
+        # here catches both the mapping bug and any future regression that
+        # forgets to thread options through to _fallback_stream.
+        assert len(captured_options) == 1
+        options = captured_options[0]
+        assert options is not None
+        assert options.task_display_mode == "plan"
+        assert options.stop_blocks == [{"type": "actions"}]
+        assert options.update_interval_ms == 2000
+
     # it("should still work without options (backward compat)")
     @pytest.mark.asyncio
     async def test_should_still_work_without_options_backward_compat(self):