diff --git a/anton/core/llm/anthropic.py b/anton/core/llm/anthropic.py index bf817bc1..317d467e 100644 --- a/anton/core/llm/anthropic.py +++ b/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 diff --git a/anton/core/llm/openai.py b/anton/core/llm/openai.py index 19e5c539..92507707 100644 --- a/anton/core/llm/openai.py +++ b/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. diff --git a/anton/core/llm/provider.py b/anton/core/llm/provider.py index 719ba67b..214ee240 100644 --- a/anton/core/llm/provider.py +++ b/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 ) diff --git a/anton/core/session.py b/anton/core/session.py index c6920bba..e76c9c1b 100644 --- a/anton/core/session.py +++ b/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 diff --git a/tests/test_openai_provider.py b/tests/test_openai_provider.py index 7b2c8737..613d0d7b 100644 --- a/tests/test_openai_provider.py +++ b/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")] diff --git a/tests/test_provider.py b/tests/test_provider.py index 65c28baf..42ee24af 100644 --- a/tests/test_provider.py +++ b/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) # ─────────────────────────────────────────────────────────────────────────────