feat(llm): capture Claude/OpenAI reasoning as a real subtype:reasoning channel (ENG-1109)

anton/core/llm/anthropic.py

@@ -16,6 +16,7 @@
     ProviderConnectionInfo,
     StreamComplete,
     StreamEvent,
+    StreamReasoningDelta,
     StreamTextDelta,
     StreamToolUseDelta,
     StreamToolUseEnd,
@@ -262,6 +263,11 @@ async def stream(
                                 "json_parts": [],
                             }
                             yield StreamToolUseStart(id=block.id, name=block.name)
+                        elif block.type in ("thinking", "redacted_thinking"):
+                            # Adaptive thinking (triggered by output_config.effort,
+                            # set above when self._reasoning_effort is configured)
+                            # — the model's own reasoning, not the final answer.
+                            blocks[idx] = {"type": "thinking"}
                         else:
                             blocks[idx] = {"type": "text"}
 
@@ -278,6 +284,10 @@ async def stream(
                                 yield StreamToolUseDelta(
                                     id=info["id"], json_delta=delta.partial_json
                                 )
+                        elif delta.type == "thinking_delta":
+                            yield StreamReasoningDelta(text=delta.thinking)
+                        # signature_delta carries only a verification signature
+                        # for the thinking block, no user-facing text — ignored.
 
                     elif event.type == "content_block_stop":
                         idx = event.index

anton/core/llm/openai.py

@@ -20,6 +20,7 @@
     ProviderConnectionInfo,
     StreamComplete,
     StreamEvent,
+    StreamReasoningDelta,
     StreamTextDelta,
     StreamToolUseDelta,
     StreamToolUseEnd,
@@ -953,6 +954,17 @@ async def stream(
                     content_text += delta.content
                     yield StreamTextDelta(text=delta.content)
 
+                # Reasoning content — chat.completions has no first-party
+                # reasoning-summary field (that's Responses-API-only), but
+                # `reasoning_content` is the de facto convention several
+                # OpenAI-compatible reasoning gateways use (DeepSeek, vLLM's
+                # reasoning parser, and possibly the mdb.ai passthrough for
+                # non-Anthropic reasoning models). Read defensively via
+                # getattr since the SDK's Delta model doesn't declare it.
+                reasoning_delta = getattr(delta, "reasoning_content", None)
+                if reasoning_delta:
+                    yield StreamReasoningDelta(text=reasoning_delta)
+
                 # Tool call deltas
                 if delta.tool_calls:
                     for tc_delta in delta.tool_calls:
@@ -1112,7 +1124,14 @@ def _build_responses_kwargs(
         if system:
             kwargs["instructions"] = system
         if self._reasoning_effort:
-            kwargs["reasoning"] = {"effort": self._reasoning_effort}
+            # "summary": "auto" asks the Responses API to also stream a
+            # natural-language summary of the reasoning itself (as
+            # response.reasoning_summary_text.delta events) — without it,
+            # reasoning happens server-side but is never surfaced to us at
+            # all. Safe to always pair with effort: only sent when effort
+            # is already set, which is itself gated to reasoning-capable
+            # models by the caller (see AntonSettings.planning_reasoning_effort).
+            kwargs["reasoning"] = {"effort": self._reasoning_effort, "summary": "auto"}
 
         merged_tools: list[dict] = []
         if tools:
@@ -1214,6 +1233,14 @@ async def _stream_via_responses(
                         content_text += delta
                         yield StreamTextDelta(text=delta)
 
+                # The model's own reasoning summary — not the final answer.
+                # Only arrives when `reasoning.summary` was requested (see
+                # _build_responses_kwargs, gated on self._reasoning_effort).
+                elif etype == "response.reasoning_summary_text.delta":
+                    delta = getattr(event, "delta", "")
+                    if delta:
+                        yield StreamReasoningDelta(text=delta)
+
                 # New output item (could be a function_call, server-tool call,
                 # or message). We only need to react when a function_call
                 # appears so we can emit the StreamToolUseStart with id+name.

anton/core/llm/provider.py

@@ -105,6 +105,21 @@ class StreamContextCompacted:
     message: str
 
 
+@dataclass
+class StreamReasoningDelta:
+    """A chunk of the model's own extended-thinking/reasoning text.
+
+    NOT part of the final answer — Anthropic's `thinking_delta` content
+    blocks (surfaced via `output_config.effort`'s adaptive thinking) and
+    OpenAI's `response.reasoning_summary_text.delta` Responses-API events
+    both map to this. Kept distinct from `StreamTextDelta` so the harness
+    layer can route it to a separate "current thought" channel instead of
+    the persisted answer body.
+    """
+
+    text: str
+
+
 StreamEvent = (
     StreamTextDelta
     | StreamToolUseStart
@@ -114,6 +129,7 @@ class StreamContextCompacted:
     | StreamTaskProgress
     | StreamToolResult
     | StreamContextCompacted
+    | StreamReasoningDelta
 )
 
 

anton/core/session.py

@@ -36,6 +36,7 @@
     StreamComplete,
     StreamContextCompacted,
     StreamEvent,
+    StreamReasoningDelta,
     StreamTaskProgress,
     StreamTextDelta,
     StreamToolResult,
@@ -2605,9 +2606,12 @@ async def _stream_and_handle_tools(
                 async for event in self.plan_stream_with_recovery(
                     system=system, tools=tools
                 ):
-                    # Capture reasoning elapsed on first text or tool event
+                    # Capture reasoning elapsed on first text, reasoning, or
+                    # tool event — a StreamReasoningDelta means the model has
+                    # already started reasoning, same signal as the first
+                    # text token.
                     if _reasoning_t0 and isinstance(
-                        event, (StreamTextDelta, StreamComplete)
+                        event, (StreamTextDelta, StreamReasoningDelta, StreamComplete)
                     ):
                         _reasoning_elapsed = _time.monotonic() - _reasoning_t0
                         _reasoning_t0 = 0  # only fire once

tests/test_openai_provider.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -670,3 +671,134 @@ def test_byok_openai_uses_openai_flavor(self):
             )
             client = LLMClient.from_settings(settings)
             assert client._planning_provider._flavor == OpenAIProvider.FLAVOR_OPENAI
+
+
+async def _fake_async_iter(items):
+    for item in items:
+        yield item
+
+
+class TestResponsesAPIReasoningSummary:
+    """ENG-1109: the Responses API only streams a reasoning summary when
+    explicitly asked for it via `reasoning.summary` — check the request
+    kwargs carry it, and that the resulting delta event maps correctly."""
+
+    def test_build_responses_kwargs_requests_summary_when_effort_set(self):
+        with patch("anton.core.llm.openai.openai"):
+            provider = OpenAIProvider(
+                api_key="k", flavor=OpenAIProvider.FLAVOR_OPENAI, reasoning_effort="high",
+            )
+            kwargs = provider._build_responses_kwargs(
+                model="gpt-5", system="s", messages=[{"role": "user", "content": "hi"}],
+                tools=None, tool_choice=None, max_tokens=100, native_web_tools=None,
+            )
+            assert kwargs["reasoning"] == {"effort": "high", "summary": "auto"}
+
+    def test_build_responses_kwargs_omits_reasoning_when_effort_unset(self):
+        with patch("anton.core.llm.openai.openai"):
+            provider = OpenAIProvider(api_key="k", flavor=OpenAIProvider.FLAVOR_OPENAI)
+            kwargs = provider._build_responses_kwargs(
+                model="gpt-5", system="s", messages=[{"role": "user", "content": "hi"}],
+                tools=None, tool_choice=None, max_tokens=100, native_web_tools=None,
+            )
+            assert "reasoning" not in kwargs
+
+    async def test_stream_maps_reasoning_summary_delta_not_output_text(self):
+        from anton.core.llm.provider import StreamReasoningDelta, StreamTextDelta
+
+        events = [
+            SimpleNamespace(type="response.reasoning_summary_text.delta", delta="Checking the docs first."),
+            SimpleNamespace(type="response.output_text.delta", delta="The real answer."),
+            SimpleNamespace(type="response.completed", response=SimpleNamespace(usage=None, status="completed")),
+        ]
+        with patch("anton.core.llm.openai.openai") as mock_openai:
+            mock_client = AsyncMock()
+            mock_openai.AsyncOpenAI.return_value = mock_client
+            mock_client.responses.create = AsyncMock(return_value=_fake_async_iter(events))
+
+            provider = OpenAIProvider(
+                api_key="k", flavor=OpenAIProvider.FLAVOR_OPENAI, reasoning_effort="high",
+            )
+            yielded = [
+                e async for e in provider.stream(
+                    model="gpt-5", system="s", messages=[{"role": "user", "content": "hi"}],
+                )
+            ]
+
+        assert [e for e in yielded if isinstance(e, StreamReasoningDelta)] == [
+            StreamReasoningDelta(text="Checking the docs first.")
+        ]
+        assert [e for e in yielded if isinstance(e, StreamTextDelta)] == [
+            StreamTextDelta(text="The real answer.")
+        ]
+
+
+class TestChatCompletionsReasoningContent:
+    """ENG-1109: mdb.ai passthrough / openai-compatible reasoning gateways —
+    `delta.reasoning_content` is read best-effort, never crashes when the
+    SDK's Delta model doesn't declare the field."""
+
+    async def test_stream_reads_reasoning_content_delta(self):
+        from anton.core.llm.provider import StreamReasoningDelta, StreamTextDelta
+
+        reasoning_chunk = MagicMock()
+        reasoning_chunk.usage = None
+        reasoning_chunk.choices = [MagicMock(delta=MagicMock(content=None, tool_calls=None), finish_reason=None)]
+        reasoning_chunk.choices[0].delta.reasoning_content = "Thinking about the best approach."
+
+        text_chunk = MagicMock()
+        text_chunk.usage = None
+        text_chunk.choices = [MagicMock(delta=MagicMock(content="Final answer.", tool_calls=None), finish_reason="stop")]
+        text_chunk.choices[0].delta.reasoning_content = None
+
+        with patch("anton.core.llm.openai.openai") as mock_openai:
+            mock_client = AsyncMock()
+            mock_openai.AsyncOpenAI.return_value = mock_client
+            mock_client.chat.completions.create = AsyncMock(
+                return_value=_fake_async_iter([reasoning_chunk, text_chunk])
+            )
+
+            provider = OpenAIProvider(
+                api_key="k", flavor=OpenAIProvider.FLAVOR_MINDS_PASSTHROUGH, reasoning_effort="high",
+            )
+            yielded = [
+                e async for e in provider.stream(
+                    model="claude-sonnet-4-6", system="s", messages=[{"role": "user", "content": "hi"}],
+                )
+            ]
+
+        assert [e for e in yielded if isinstance(e, StreamReasoningDelta)] == [
+            StreamReasoningDelta(text="Thinking about the best approach.")
+        ]
+        assert [e for e in yielded if isinstance(e, StreamTextDelta)] == [
+            StreamTextDelta(text="Final answer.")
+        ]
+
+    async def test_stream_tolerates_missing_reasoning_content_field(self):
+        # A Delta object that doesn't declare `reasoning_content` at all
+        # (e.g. a real openai.types.chat.ChoiceDelta from a non-reasoning
+        # model) must not raise — getattr's default covers this, but a
+        # plain SimpleNamespace (no attr at all, unlike MagicMock which
+        # auto-vivifies) is the real test that we're not assuming the
+        # field always exists.
+        chunk = SimpleNamespace(
+            usage=None,
+            choices=[SimpleNamespace(
+                delta=SimpleNamespace(content="hi", tool_calls=None),
+                finish_reason="stop",
+            )],
+        )
+        with patch("anton.core.llm.openai.openai") as mock_openai:
+            mock_client = AsyncMock()
+            mock_openai.AsyncOpenAI.return_value = mock_client
+            mock_client.chat.completions.create = AsyncMock(return_value=_fake_async_iter([chunk]))
+
+            provider = OpenAIProvider(api_key="k", flavor=OpenAIProvider.FLAVOR_MINDS_PASSTHROUGH)
+            yielded = [
+                e async for e in provider.stream(
+                    model="claude-sonnet-4-6", system="s", messages=[{"role": "user", "content": "hi"}],
+                )
+            ]
+        from anton.core.llm.provider import StreamReasoningDelta, StreamTextDelta
+        assert [e for e in yielded if isinstance(e, StreamReasoningDelta)] == []
+        assert [e for e in yielded if isinstance(e, StreamTextDelta)] == [StreamTextDelta(text="hi")]

tests/test_provider.py

@@ -188,6 +188,114 @@ async def test_provider_without_api_key(self):
             mock_anthropic.AsyncAnthropic.assert_called_once_with()
 
 
+class _FakeAnthropicStream:
+    """Minimal async-context-manager + async-iterator stand-in for the
+    object `client.messages.stream(**kwargs)` returns (NOT awaited itself
+    — used via `async with ... as stream: async for event in stream:`)."""
+
+    def __init__(self, events):
+        self._events = events
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_exc):
+        return False
+
+    def __aiter__(self):
+        return self._iter()
+
+    async def _iter(self):
+        for event in self._events:
+            yield event
+
+
+class TestAnthropicProviderReasoningStream:
+    """ENG-1109: extended-thinking content blocks (triggered server-side by
+    `output_config.effort`, already sent whenever reasoning_effort is set)
+    must surface as StreamReasoningDelta, not get misclassified as text or
+    silently dropped."""
+
+    async def test_thinking_delta_becomes_stream_reasoning_delta(self):
+        from anton.core.llm.provider import StreamReasoningDelta, StreamTextDelta
+
+        events = [
+            SimpleNamespace(
+                type="message_start",
+                message=SimpleNamespace(usage=SimpleNamespace(input_tokens=5, output_tokens=0)),
+            ),
+            SimpleNamespace(
+                type="content_block_start", index=0,
+                content_block=SimpleNamespace(type="thinking"),
+            ),
+            SimpleNamespace(
+                type="content_block_delta", index=0,
+                delta=SimpleNamespace(type="thinking_delta", thinking="Let me check that first."),
+            ),
+            SimpleNamespace(
+                type="content_block_delta", index=0,
+                delta=SimpleNamespace(type="signature_delta", signature="sig-abc"),
+            ),
+            SimpleNamespace(type="content_block_stop", index=0),
+            SimpleNamespace(
+                type="content_block_start", index=1,
+                content_block=SimpleNamespace(type="text"),
+            ),
+            SimpleNamespace(
+                type="content_block_delta", index=1,
+                delta=SimpleNamespace(type="text_delta", text="The real answer."),
+            ),
+            SimpleNamespace(type="content_block_stop", index=1),
+            SimpleNamespace(
+                type="message_delta",
+                delta=SimpleNamespace(stop_reason="end_turn"),
+                usage=SimpleNamespace(output_tokens=12),
+            ),
+        ]
+
+        with patch("anton.core.llm.anthropic.anthropic") as mock_anthropic:
+            mock_client = AsyncMock()
+            mock_client.messages.stream = MagicMock(return_value=_FakeAnthropicStream(events))
+            mock_anthropic.AsyncAnthropic.return_value = mock_client
+
+            provider = AnthropicProvider(api_key="test-key", reasoning_effort="medium")
+            yielded = [
+                e async for e in provider.stream(
+                    model="claude-sonnet-4-6",
+                    system="be helpful",
+                    messages=[{"role": "user", "content": "hi"}],
+                )
+            ]
+
+        reasoning_events = [e for e in yielded if isinstance(e, StreamReasoningDelta)]
+        text_events = [e for e in yielded if isinstance(e, StreamTextDelta)]
+        assert reasoning_events == [StreamReasoningDelta(text="Let me check that first.")]
+        assert text_events == [StreamTextDelta(text="The real answer.")]
+
+    async def test_stream_passes_effort_via_extra_body(self):
+        with patch("anton.core.llm.anthropic.anthropic") as mock_anthropic:
+            mock_client = AsyncMock()
+            mock_client.messages.stream = MagicMock(
+                return_value=_FakeAnthropicStream([
+                    SimpleNamespace(
+                        type="message_delta",
+                        delta=SimpleNamespace(stop_reason="end_turn"),
+                        usage=SimpleNamespace(output_tokens=0),
+                    ),
+                ])
+            )
+            mock_anthropic.AsyncAnthropic.return_value = mock_client
+
+            provider = AnthropicProvider(api_key="k", reasoning_effort="high")
+            async for _ in provider.stream(
+                model="claude-sonnet-4-6", system="s", messages=[{"role": "user", "content": "hi"}],
+            ):
+                pass
+
+            call_kwargs = mock_client.messages.stream.call_args[1]
+            assert call_kwargs["extra_body"] == {"output_config": {"effort": "high"}}
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Native server-side web tools (web_search / web_fetch)
 # ─────────────────────────────────────────────────────────────────────────────