ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-27 06:14:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

feat(reasoning): OpenAI Responses + Chat Completions capture/replay (Phase 3+4)

Browse Source 
Wire reasoning capture and (where the API supports it) replay for the
two remaining provider paths.  Phase 3 was originally scoped as
"OpenAI Responses + Gemini" but a spike against the OpenAI SDK source
revealed that Gemini routes through the OpenAI-compatible endpoint
(``/v1beta/openai/``), which is structurally identical to vLLM /
llama.cpp / any other Chat-Completions-shaped local model.  Phase 3
and Phase 4 collapse into one feature with two distinct sub-paths:

* **Path 2 (OpenAI Responses)** — full capture+replay.  ``include=
  ["reasoning.encrypted_content"]`` on the request makes the API
  surface ``encrypted_content`` on reasoning items in
  ``provider_blocks``; ``_convert_messages`` round-trips them as
  ``ResponseReasoningItemParam`` input items on subsequent turns.
  Verified against the OpenAI Python SDK 2.33.0 source
  (``response_reasoning_item.py:31-62``,
  ``response_reasoning_item_param.py:33-37``,
  ``response_create_params.py:70-74``).  Even with ``store=False``,
  ``encrypted_content`` round-trips correctly per the SDK's own
  documentation.

* **Path 3 (Chat Completions / vLLM / llama.cpp / Gemini-compat)** —
  persist-only.  Canonical OpenAI Chat Completions has no reasoning
  field on the wire, but several local-model servers tack on
  ``delta.reasoning_content`` as Pydantic extras.  ``ChatSession.
  _maybe_synth_reasoning_block`` stamps a synthetic ``{type:
  "reasoning_text", text, source?}`` block onto ``_provider_content``
  at end-of-stream when no native ``provider_blocks`` were emitted but
  ``reasoning_parts`` accumulated text.  The ``source`` field carries
  ``server_compat.server_type`` (vllm, llama.cpp, sglang, …) for
  diagnostic value — informational only, doesn't gate behaviour.
  Reasoning text NEVER replays back to the model on this path; it
  rides ``_provider_content`` only for ``/history`` UI rehydration
  and gets stripped from the wire by the existing
  ``sanitize_messages`` underscore-prefix strip on every request.

What this change does

* ``ModelCapabilities.supports_reasoning_replay: bool = False`` added
  to the dataclass.  Set True on every OpenAI reasoning model
  (gpt-5* + o-series via the Responses API) and every Anthropic
  Claude entry (default + 6 model-specific).  Path-2 wire-build does
  ``replay_active = bool(replay_reasoning_to_model and caps.supports_
  reasoning_replay)`` so an operator who flips the flag on a
  non-reasoning model (gpt-4o via Responses) silently no-ops rather
  than emit a malformed ``include=`` request.

* ``OpenAIResponsesProvider`` gains:
  - ``_build_kwargs`` accepts ``replay_reasoning_to_model: bool``
    (threaded from ``create_streaming``/``create_completion``);
    adds ``include=["reasoning.encrypted_content"]`` when active.
  - ``_convert_messages`` accepts the same flag, captures
    ``_provider_content`` reasoning items pre-sanitization, and
    emits them as input items immediately before the assistant
    message they belong to.  Position is tracked by ASSISTANT
    ORDINAL (not raw index) — ``sanitize_messages`` drops orphan
    tool results and inserts synthesized error tool messages, but
    NEVER drops or duplicates assistant messages, so the n-th
    assistant in the original list is invariably the n-th in the
    sanitized list.  Index-based lookup would have silently
    misrouted reasoning attachments after any tool-message repair.
  - ``extract_reasoning_text`` walks ``type=="reasoning"`` items and
    returns ``summary[*].text`` + ``content[*].text`` concatenation.
  - ``_reasoning_item_for_input`` projects a stored item into
    ``ResponseReasoningItemParam`` shape (drops server-only
    ``status``).  Returns ``None`` when ``id`` is missing or non-
    string per the SDK ``Required[str]`` schema; caller skips
    appending, preventing malformed input items from reaching the API.

* ``OpenAIChatCompletionsProvider`` gains:
  - ``extract_reasoning_text`` walks synthetic
    ``type=="reasoning_text"`` blocks and returns the concatenated
    text directly (no underlying provider semantics — the synth
    block IS the surface).

* ``ChatSession`` gains:
  - ``_resolve_server_type(alias)`` reads ``server_compat.server_type``
    from the active model's capabilities dict.
  - ``_maybe_synth_reasoning_block(provider_blocks, reasoning_parts)``
    creates the synthetic ``reasoning_text`` block when no native
    blocks were emitted but reasoning was captured.  Wired at the
    end of ``_stream_response`` immediately before the
    ``_provider_content`` stamp.

* ``history_decoration.py`` dispatcher collapses three near-identical
  lazy-init singleton getters (one per recognised block type) into a
  single ``_BLOCK_TYPE_PROVIDER_FACTORY`` dict + helper.  Adding a
  fourth provider becomes a one-line dict entry.

* Constants hoist: ``MAX_REASONING_DISPLAY_BYTES = 64 * 1024`` moved
  from three sibling provider modules into ``_protocol.py`` so a
  tuning change propagates uniformly to every provider's display path.

Cross-provider safety

The synthetic ``reasoning_text`` block type is intentionally NOT in
``ANTHROPIC_VALID_BLOCK_TYPES`` (Phase 2 constant).  Cross-model
resumption (operator switches from a local model to Anthropic mid-
workstream) falls through Phase 2's shape filter cleanly to the
text+tool_calls rebuild path rather than reaching Anthropic with a
malformed block.  Pinned by ``test_synthetic_block_falls_through_
anthropic_shape_filter``.

Same protection applies in reverse: OpenAI Responses
``type=="reasoning"`` items reaching Anthropic mid-workstream fail
the shape filter and rebuild from text+tool_calls.

Tests (49 net new tests)

* ``tests/test_provider_openai_responses_reasoning.py`` (21 tests):
  - Extractor unit tests: empty/none/no-reasoning/single/mixed/
    truncation/malformed/non-list (8).
  - ``_reasoning_item_for_input`` projection (4 tests including the
    new None-on-missing-id guard).
  - ``_build_kwargs`` include= gating: flag+capability/flag-false/
    capability-false/default-omits (4).
  - ``_convert_messages`` reasoning round-trip: emit-before-assistant/
    drop-on-replay-false/foreign-shape-skipped/default-replay-false (5).

* ``tests/test_session_synth_reasoning_block.py`` (23 tests):
  - ``_maybe_synth_reasoning_block`` direct unit tests (6).
  - Cross-provider safety regression — synthetic block falls through
    Anthropic shape filter (2).
  - ``OpenAIChatCompletionsProvider.extract_reasoning_text`` for the
    new synthetic block type (6).
  - ``_resolve_server_type`` direct unit tests (5).
  - ``_stream_response`` integration tests driving fake reasoning-
    emitting streams through the actual session method (3 tests
    — added in response to a code-review finding that pinned the
    wire-up at session.py needs an integration test).

* ``tests/test_session_replay_reasoning.py`` extended with 4
  ``TestSessionToOpenAIResponsesBoundaryIntegration`` tests driving
  ``session._try_stream`` -> real ``OpenAIResponsesProvider`` ->
  captured ``client.responses.create`` SDK boundary call.  Negative-
  tested: temporarily reverting the ``include=`` step in
  ``_build_kwargs`` makes ``test_replay_true_adds_include_to_
  responses_request`` fail; restoring makes it pass.

* ``tests/test_history_decoration.py`` extended with the new
  ``reasoning_text`` dispatcher branch test, and the Phase 1 stub
  test for the OpenAI Responses dispatcher branch was tightened
  (it now asserts real text extraction instead of the empty-string
  stub).

* ``tests/test_provider_anthropic_reasoning.py`` had its Phase 1
  ``OpenAIResponses returns "" for reasoning blocks`` stub test
  retitled and updated to assert the real Phase 3 behaviour.

Code-review pass

Multi-stage ``/review`` pipeline (4 finders + verify + dedupe) ran
on this diff.  6 findings (1 major, 3 minor, 2 nit), 0 critical, 0
security, 0 performance.  All applied:

* MAJOR (bug-1+bug-4+q-1): ``_convert_messages`` enumerate-index
  lookup was unsound under ``sanitize_messages`` length changes.
  Fixed by switching to assistant-ordinal-keyed lookup.
* MINOR (q-2+q-3): ``_MAX_REASONING_DISPLAY_BYTES`` duplicated
  across three provider modules + declared after first use.
  Fixed by hoisting to ``_protocol.py``.
* MINOR (q-4): three near-identical singleton getters in dispatcher.
  Fixed by collapsing to ``_BLOCK_TYPE_PROVIDER_FACTORY`` dict.
* MINOR (q-5): ``_maybe_synth_reasoning_block`` wire-up not pinned
  by integration test.  Fixed by adding three
  ``TestStreamResponseSynthBlockIntegration`` tests.
* NIT (bug-2): ``_reasoning_item_for_input`` fell back to ``id=""``;
  fixed to return ``None`` on missing/non-string id.
* NIT (q-6): four naming variants for the same concept; renamed
  ``_convert_messages`` kwarg to match the operator-flag name.
* REFUTED (bug-3): SDK distinguishes summary vs content as separate
  fields; no double-counting concern.

Briefing departures

The briefing's Phase 3 plan grouped Gemini with OpenAI Responses on
the assumption that Gemini reasoning had its own native shape (like
Anthropic's ``thinking``).  The spike confirmed Gemini-via-OpenAI-
compat is path-3 (Chat Completions shape, no native reasoning
items).  Phase 3+4 merger handles Gemini for free via the synthetic
``reasoning_text`` block — same mechanism used for vLLM and
llama.cpp.  Whether Gemini's specific endpoint actually emits
``reasoning_content`` deltas is server-dependent and not yet
empirically verified; capture is best-effort (server-emission-driven,
no flag gate).

The briefing's Phase 4 plan stamped reasoning as Anthropic-shaped
``thinking`` blocks ``{type: "thinking", thinking: <text>}``.  This
PR uses a distinct ``{type: "reasoning_text", text, source?}`` shape
to avoid a cross-model resumption hazard the briefing missed: an
unsigned synthetic Anthropic-shape block reaching Anthropic's wire
would 400 the API.  The distinct shape falls through Phase 2's shape
filter cleanly without needing signature validation in the filter.

Lint + test gate

* ruff check + ruff format -- clean.
* mypy -- no issues across all 191 source files.
* pytest -m 'not live' -- 6110 passed (3 deselected).  Phase 3+4
  added 49 net new tests.
This commit is contained in:
Patrick Buckley
2026-05-08 21:31:10 -07:00
parent 00bd80a658
commit b477c85ddc
12 changed files with 1276 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_history_decoration.py
+27 -7
View File
@@ -518,12 +518,12 @@ class TestExtractReasoningForHistory:
extract_reasoning_for_history(messages, persist_reasoning_flag=True)
assert messages[0]["reasoning"] == "first"
def test_first_block_reasoning_dispatches_to_openai_phase3_stub(self) -> None:
# OpenAI Responses extractor returns "" until Phase 3 wires
# ``include=["reasoning.encrypted_content"]``. The dispatcher
# must still route to it (not silently fall through to "").
# We assert the dispatcher routed by checking the strip happens
# AND no reasoning field is added (because the stub returns "").
def test_first_block_reasoning_dispatches_to_openai_responses(self) -> None:
# Phase 3: dispatcher routes type=="reasoning" to the
# OpenAI Responses extractor, which now returns the
# summary[*].text concatenation. Pre-Phase-3 this asserted
# "" (the stub); the assertion was tightened once the wire
# path landed.
from turnstone.core.history_decoration import extract_reasoning_for_history
messages = [
@@ -536,7 +536,7 @@ class TestExtractReasoningForHistory:
}
]
extract_reasoning_for_history(messages, persist_reasoning_flag=True)
assert "reasoning" not in messages[0]
assert messages[0]["reasoning"] == "s"
assert "_provider_content" not in messages[0]
def test_unknown_first_block_type_no_op(self) -> None:
@@ -596,3 +596,23 @@ class TestExtractReasoningForHistory:
extract_reasoning_for_history(messages, persist_reasoning_flag=True)
assert "reasoning" not in messages[0]
assert "_provider_content" not in messages[0]
def test_first_block_reasoning_text_dispatches_to_openai_chat(self) -> None:
# Phase 3 path 3: synthetic ``reasoning_text`` blocks (stamped
# by ChatSession._maybe_synth_reasoning_block for vLLM /
# llama.cpp / Gemini-compat conversations) dispatch to
# OpenAIChatCompletionsProvider.extract_reasoning_text.
from turnstone.core.history_decoration import extract_reasoning_for_history
messages = [
{
"role": "assistant",
"content": "answer",
"_provider_content": [
{"type": "reasoning_text", "text": "synth thought", "source": "vllm"},
],
}
]
extract_reasoning_for_history(messages, persist_reasoning_flag=True)
assert messages[0]["reasoning"] == "synth thought"
assert "_provider_content" not in messages[0]
tests/test_provider_anthropic_reasoning.py
+10 -8
View File
@@ -17,12 +17,12 @@ from __future__ import annotations
import pytest
from turnstone.core.providers._anthropic import (
_MAX_REASONING_DISPLAY_BYTES,
AnthropicProvider,
)
from turnstone.core.providers._anthropic import AnthropicProvider
from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
from turnstone.core.providers._protocol import (
MAX_REASONING_DISPLAY_BYTES as _MAX_REASONING_DISPLAY_BYTES,
)
@pytest.fixture
@@ -115,9 +115,11 @@ class TestOtherProvidersDefault:
blocks = [{"type": "thinking", "thinking": "would-be-text", "signature": "s"}]
assert provider.extract_reasoning_text(blocks) == ""
def test_openai_responses_returns_empty_on_reasoning_shaped_blocks(self) -> None:
# Phase 3 stub — reasoning items exist but extractor returns ""
# until ``include=["reasoning.encrypted_content"]`` is wired.
def test_openai_responses_extracts_reasoning_summary(self) -> None:
# Phase 3: extractor now walks reasoning items captured via
# include=["reasoning.encrypted_content"] and returns the
# summary[*].text concatenation. Pre-Phase-3 this returned
# "" — the stub was replaced once the wire path landed.
provider = OpenAIResponsesProvider()
blocks = [{"type": "reasoning", "summary": [{"type": "summary_text", "text": "x"}]}]
assert provider.extract_reasoning_text(blocks) == ""
assert provider.extract_reasoning_text(blocks) == "x"
tests/test_provider_openai_responses_reasoning.py
+355
View File
@@ -0,0 +1,355 @@
"""Tests for OpenAI Responses reasoning capture + replay (Phase 3 path 2).
Phase 3 wires:
1. ``include=["reasoning.encrypted_content"]`` on the request when
the operator flag AND the model capability both allow.
2. ``_convert_messages`` round-tripping stored reasoning items as
``ResponseReasoningItemParam`` input items on subsequent turns.
3. ``OpenAIResponsesProvider.extract_reasoning_text`` walking
reasoning items and returning concatenated summary + content text.
All tests drive through the real ``OpenAIResponsesProvider`` — no
mocks of the converter/build_kwargs themselves; only the SDK boundary
is mocked where relevant.
"""
from __future__ import annotations
import pytest
from turnstone.core.providers._openai_responses import (
OpenAIResponsesProvider,
_reasoning_item_for_input,
)
from turnstone.core.providers._protocol import (
MAX_REASONING_DISPLAY_BYTES as _MAX_REASONING_DISPLAY_BYTES,
)
from turnstone.core.providers._protocol import ModelCapabilities
@pytest.fixture
def provider() -> OpenAIResponsesProvider:
return OpenAIResponsesProvider()
def _capable_caps() -> ModelCapabilities:
"""Capability fixture for a reasoning-replay-capable model."""
return ModelCapabilities(
context_window=400000,
max_output_tokens=128000,
supports_temperature=False,
reasoning_effort_values=("low", "medium", "high"),
default_reasoning_effort="medium",
supports_reasoning_replay=True,
)
def _incapable_caps() -> ModelCapabilities:
return ModelCapabilities(
context_window=128000,
supports_reasoning_replay=False,
)
class TestExtractReasoningText:
def test_none_returns_empty(self, provider: OpenAIResponsesProvider) -> None:
assert provider.extract_reasoning_text(None) == ""
def test_empty_list_returns_empty(self, provider: OpenAIResponsesProvider) -> None:
assert provider.extract_reasoning_text([]) == ""
def test_no_reasoning_items_returns_empty(self, provider: OpenAIResponsesProvider) -> None:
blocks = [
{"type": "message", "role": "assistant", "content": "hi"},
{"type": "function_call", "call_id": "c1", "name": "x", "arguments": "{}"},
]
assert provider.extract_reasoning_text(blocks) == ""
def test_summary_text_extracted(self, provider: OpenAIResponsesProvider) -> None:
# Per ResponseReasoningItem (response_reasoning_item.py:31-62):
# summary is always present; content is optional.
blocks = [
{
"type": "reasoning",
"id": "r_1",
"summary": [
{"type": "summary_text", "text": "I considered X"},
{"type": "summary_text", "text": "then Y"},
],
}
]
assert provider.extract_reasoning_text(blocks) == "I considered X\nthen Y"
def test_content_text_extracted_alongside_summary(
self, provider: OpenAIResponsesProvider
) -> None:
blocks = [
{
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "summary line"}],
"content": [{"type": "reasoning_text", "text": "raw reasoning"}],
}
]
# Order: summary first, then content (matches the order the SDK
# surfaces them via streaming events).
result = provider.extract_reasoning_text(blocks)
assert "summary line" in result
assert "raw reasoning" in result
def test_truncation_at_64kib_cap(self, provider: OpenAIResponsesProvider) -> None:
long_text = "x" * (_MAX_REASONING_DISPLAY_BYTES + 1024)
blocks = [
{
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": long_text}],
}
]
result = provider.extract_reasoning_text(blocks)
assert len(result) == _MAX_REASONING_DISPLAY_BYTES
def test_malformed_summary_entry_skipped(self, provider: OpenAIResponsesProvider) -> None:
blocks = [
{
"type": "reasoning",
"id": "r_1",
"summary": [
"not a dict",
{"type": "summary_text"}, # missing text
{"type": "summary_text", "text": ""}, # empty text
{"type": "summary_text", "text": "good"},
],
}
]
assert provider.extract_reasoning_text(blocks) == "good"
def test_non_list_input_returns_empty(self, provider: OpenAIResponsesProvider) -> None:
assert provider.extract_reasoning_text("not a list") == "" # type: ignore[arg-type]
def test_other_block_types_skipped_in_walk(self, provider: OpenAIResponsesProvider) -> None:
# Mixed payload: only the reasoning block contributes.
blocks = [
{"type": "message", "role": "assistant", "content": "hi"},
{
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "thought"}],
},
{"type": "function_call", "call_id": "c1", "name": "x", "arguments": "{}"},
]
assert provider.extract_reasoning_text(blocks) == "thought"
class TestReasoningItemForInput:
"""``_reasoning_item_for_input`` projects a stored ``ResponseReasoningItem``
dict into ``ResponseReasoningItemParam`` shape (drops server-only
``status``)."""
def test_minimal_item_round_trip(self) -> None:
stored = {
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "x"}],
"status": "completed",
}
result = _reasoning_item_for_input(stored)
assert result["type"] == "reasoning"
assert result["id"] == "r_1"
assert result["summary"] == [{"type": "summary_text", "text": "x"}]
# status NOT round-tripped (server-only field per
# ResponseReasoningItemParam at response_reasoning_item_param.py).
assert "status" not in result
def test_encrypted_content_round_trips_when_present(self) -> None:
stored = {
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "x"}],
"encrypted_content": "opaque-blob",
}
result = _reasoning_item_for_input(stored)
assert result["encrypted_content"] == "opaque-blob"
def test_encrypted_content_omitted_when_absent(self) -> None:
stored = {
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "x"}],
}
result = _reasoning_item_for_input(stored)
assert "encrypted_content" not in result
def test_content_round_trips_when_present(self) -> None:
stored = {
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "s"}],
"content": [{"type": "reasoning_text", "text": "raw"}],
}
result = _reasoning_item_for_input(stored)
assert result["content"] == [{"type": "reasoning_text", "text": "raw"}]
class TestBuildKwargsInclude:
"""``_build_kwargs`` adds ``include=["reasoning.encrypted_content"]``
only when the operator flag AND the model capability both allow."""
def test_include_added_when_flag_and_capability_true(
self, provider: OpenAIResponsesProvider
) -> None:
kwargs = provider._build_kwargs(
model="gpt-5",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
temperature=0.5,
reasoning_effort="medium",
deferred_names=None,
capabilities=_capable_caps(),
replay_reasoning_to_model=True,
)
assert kwargs.get("include") == ["reasoning.encrypted_content"]
def test_include_omitted_when_flag_false(self, provider: OpenAIResponsesProvider) -> None:
kwargs = provider._build_kwargs(
model="gpt-5",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
temperature=0.5,
reasoning_effort="medium",
deferred_names=None,
capabilities=_capable_caps(),
replay_reasoning_to_model=False,
)
assert "include" not in kwargs
def test_include_omitted_when_capability_false(self, provider: OpenAIResponsesProvider) -> None:
# Defends against operator flipping the flag on a non-reasoning
# model — the capability gate prevents the include= from being
# sent (silently no-op'd).
kwargs = provider._build_kwargs(
model="gpt-4o",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
temperature=0.5,
reasoning_effort="medium",
deferred_names=None,
capabilities=_incapable_caps(),
replay_reasoning_to_model=True,
)
assert "include" not in kwargs
def test_include_omitted_by_default(self, provider: OpenAIResponsesProvider) -> None:
# When neither flag nor capability is passed, replay defaults
# True (kwarg) but capability defaults False — net: no include.
kwargs = provider._build_kwargs(
model="gpt-4o",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
temperature=0.5,
reasoning_effort="medium",
deferred_names=None,
)
assert "include" not in kwargs
class TestConvertMessagesReasoningReplay:
"""``_convert_messages`` round-trips stored reasoning items as input."""
def test_reasoning_item_emitted_before_assistant_when_replay_true(
self, provider: OpenAIResponsesProvider
) -> None:
messages = [
{"role": "user", "content": "explain"},
{
"role": "assistant",
"content": "Final answer.",
"_provider_content": [
{
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "I thought"}],
"encrypted_content": "abc",
}
],
},
{"role": "user", "content": "follow up"},
]
_, items = provider._convert_messages(messages, replay_reasoning_to_model=True)
# Find the reasoning input item.
types = [it.get("type") for it in items]
# Expected: user, reasoning, message (assistant), user.
assert types == ["message", "reasoning", "message", "message"]
reasoning_idx = types.index("reasoning")
r_item = items[reasoning_idx]
assert r_item["id"] == "r_1"
assert r_item["encrypted_content"] == "abc"
# And the reasoning item appears immediately BEFORE the
# assistant message it belongs to.
assert items[reasoning_idx + 1]["role"] == "assistant"
def test_reasoning_item_dropped_when_replay_false(
self, provider: OpenAIResponsesProvider
) -> None:
messages = [
{
"role": "assistant",
"content": "Answer.",
"_provider_content": [
{
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "thought"}],
}
],
},
]
_, items = provider._convert_messages(messages, replay_reasoning_to_model=False)
types = [it.get("type") for it in items]
assert "reasoning" not in types
def test_no_reasoning_items_when_provider_content_lacks_reasoning(
self, provider: OpenAIResponsesProvider
) -> None:
# Anthropic-shaped _provider_content reaching OpenAI Responses
# (cross-provider — operator switch from Anthropic to GPT-5):
# no type=="reasoning" items, so nothing emitted.
messages = [
{
"role": "assistant",
"content": "x",
"_provider_content": [
{"type": "thinking", "thinking": "anth", "signature": "s"},
],
},
]
_, items = provider._convert_messages(messages, replay_reasoning_to_model=True)
types = [it.get("type") for it in items]
assert "reasoning" not in types
def test_default_replay_reasoning_false_omits_reasoning(
self, provider: OpenAIResponsesProvider
) -> None:
# Pre-Phase-3 callers (no kwarg) get the back-compat behaviour:
# reasoning items are silently dropped (sanitize_messages was
# already stripping _provider_content anyway).
messages = [
{
"role": "assistant",
"content": "x",
"_provider_content": [
{
"type": "reasoning",
"id": "r_1",
"summary": [{"type": "summary_text", "text": "x"}],
}
],
},
]
_, items = provider._convert_messages(messages) # no kwarg
types = [it.get("type") for it in items]
assert "reasoning" not in types
tests/test_session_replay_reasoning.py
+175
View File
@@ -357,6 +357,181 @@ class TestSessionToWireBoundaryIntegration:
)
class TestSessionToOpenAIResponsesBoundaryIntegration:
"""End-to-end integration: session._try_stream -> real
OpenAIResponsesProvider.create_streaming -> captured Responses
SDK boundary call. Mirrors the AnthropicProvider test above
but for the path-2 (Responses API) replay flow.
Pins the include= request kwarg + reasoning input-item emission
actually fire at the wire boundary when the operator flag and
model capability both allow.
"""
def _stub_responses_client(self) -> tuple[MagicMock, dict[str, object]]:
"""Mock OpenAI Responses client. ``client.responses.create``
captures kwargs and returns an empty stream iterator."""
captured: dict[str, object] = {}
def create(**kwargs: object) -> object:
captured.update(kwargs)
return iter([])
client = MagicMock()
client.responses.create = create
return client, captured
def _registry_with_reasoning_capability(
self, replay: bool = True, supports_replay: bool = True
) -> Any:
from turnstone.core.providers._protocol import ModelCapabilities
return SimpleNamespace(
get_config=lambda alias: SimpleNamespace(
replay_reasoning_to_model=replay,
capabilities={}, # no overrides
),
_caps=ModelCapabilities(
context_window=400000,
supports_temperature=False,
reasoning_effort_values=("low", "medium", "high"),
default_reasoning_effort="medium",
supports_reasoning_replay=supports_replay,
),
)
def test_replay_true_adds_include_to_responses_request(self) -> None:
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
registry = self._registry_with_reasoning_capability(replay=True, supports_replay=True)
session = _make_session()
session._registry = registry
session._model_alias = "gpt-5"
client, captured = self._stub_responses_client()
real_provider = OpenAIResponsesProvider()
with (
patch.object(session, "_get_active_tools", return_value=None),
patch.object(session, "_provider_extra_params", return_value=None),
patch.object(session, "_get_deferred_names", return_value=frozenset()),
patch.object(session, "_check_cancelled"),
):
stream = session._try_stream(
client=client,
model="gpt-5",
msgs=[{"role": "user", "content": "hi"}],
provider=real_provider,
capabilities=registry._caps,
model_alias="gpt-5",
)
list(stream)
assert captured.get("include") == ["reasoning.encrypted_content"]
def test_replay_false_omits_include(self) -> None:
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
registry = self._registry_with_reasoning_capability(replay=False, supports_replay=True)
session = _make_session()
session._registry = registry
session._model_alias = "gpt-5"
client, captured = self._stub_responses_client()
real_provider = OpenAIResponsesProvider()
with (
patch.object(session, "_get_active_tools", return_value=None),
patch.object(session, "_provider_extra_params", return_value=None),
patch.object(session, "_get_deferred_names", return_value=frozenset()),
patch.object(session, "_check_cancelled"),
):
stream = session._try_stream(
client=client,
model="gpt-5",
msgs=[{"role": "user", "content": "hi"}],
provider=real_provider,
capabilities=registry._caps,
model_alias="gpt-5",
)
list(stream)
assert "include" not in captured
def test_capability_false_omits_include_even_when_flag_true(self) -> None:
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
# Operator flips replay=True but the model has
# supports_reasoning_replay=False (e.g. gpt-4o via Responses).
# Capability gate prevents the include= from being sent.
registry = self._registry_with_reasoning_capability(replay=True, supports_replay=False)
session = _make_session()
session._registry = registry
session._model_alias = "gpt-4o"
client, captured = self._stub_responses_client()
real_provider = OpenAIResponsesProvider()
with (
patch.object(session, "_get_active_tools", return_value=None),
patch.object(session, "_provider_extra_params", return_value=None),
patch.object(session, "_get_deferred_names", return_value=frozenset()),
patch.object(session, "_check_cancelled"),
):
stream = session._try_stream(
client=client,
model="gpt-4o",
msgs=[{"role": "user", "content": "hi"}],
provider=real_provider,
capabilities=registry._caps,
model_alias="gpt-4o",
)
list(stream)
assert "include" not in captured
def test_replay_true_emits_reasoning_input_item(self) -> None:
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
registry = self._registry_with_reasoning_capability(replay=True, supports_replay=True)
session = _make_session()
session._registry = registry
session._model_alias = "gpt-5"
client, captured = self._stub_responses_client()
real_provider = OpenAIResponsesProvider()
# Multi-turn conversation with stored reasoning on assistant turn.
msgs: list[dict[str, object]] = [
{"role": "user", "content": "explain"},
{
"role": "assistant",
"content": "Final answer.",
"_provider_content": [
{
"type": "reasoning",
"id": "r_xyz",
"summary": [{"type": "summary_text", "text": "I thought"}],
"encrypted_content": "blob",
}
],
},
{"role": "user", "content": "follow-up"},
]
with (
patch.object(session, "_get_active_tools", return_value=None),
patch.object(session, "_provider_extra_params", return_value=None),
patch.object(session, "_get_deferred_names", return_value=frozenset()),
patch.object(session, "_check_cancelled"),
):
stream = session._try_stream(
client=client,
model="gpt-5",
msgs=msgs,
provider=real_provider,
capabilities=registry._caps,
model_alias="gpt-5",
)
list(stream)
# Walk the wire input items — one of them must be the reasoning
# round-trip (id matches what we stored).
wire_input = captured.get("input")
assert isinstance(wire_input, list)
reasoning_items = [it for it in wire_input if it.get("type") == "reasoning"]
assert len(reasoning_items) == 1
assert reasoning_items[0]["id"] == "r_xyz"
assert reasoning_items[0]["encrypted_content"] == "blob"
class TestUtilityCompletionPassesFlag:
"""Non-streaming utility path (title gen, compaction, extraction) —
same plumbing requirement as streaming."""
tests/test_session_synth_reasoning_block.py
+332
View File
@@ -0,0 +1,332 @@
"""Tests for ChatSession synthetic ``reasoning_text`` block stamping (Phase 3 path 3).
Path 3 covers OpenAI Chat Completions endpoints — vLLM with
``--reasoning-parser``, llama.cpp with ``reasoning_format``, Gemini's
``/v1beta/openai/`` endpoint, and any other server that surfaces
``delta.reasoning_content`` Pydantic extras. These have no native
provider_blocks shape on the wire, so ``ChatSession._stream_response``
captures the streamed reasoning text into ``reasoning_parts`` and
``_maybe_synth_reasoning_block`` stamps it onto ``_provider_content``
as a synthetic ``{type: "reasoning_text"}`` block at the end of the
turn.
These tests pin:
1. The synthesizer fires only when no native blocks were emitted AND
reasoning was captured (Anthropic + OpenAI Responses bypass it).
2. ``source`` field is tagged with the active model's server_type
(informational; pulled from ``server_compat.server_type``).
3. ``OpenAIChatCompletionsProvider.extract_reasoning_text`` round-trips
the synthetic block on history rehydration.
4. The synthetic shape is NOT in ``ANTHROPIC_VALID_BLOCK_TYPES`` so
cross-model resumption (local-model → Anthropic) falls through
cleanly to the text+tool_calls rebuild path.
"""
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock
from turnstone.core.providers._anthropic import (
ANTHROPIC_VALID_BLOCK_TYPES,
AnthropicProvider,
)
from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
from turnstone.core.session import ChatSession
from turnstone.core.session_ui_base import SessionUIBase
class _NullUI(SessionUIBase):
def __init__(self) -> None:
super().__init__()
def _make_session(**kwargs: Any) -> ChatSession:
defaults: dict[str, Any] = {
"client": MagicMock(),
"model": "test-model",
"ui": _NullUI(),
"instructions": None,
"temperature": 0.5,
"max_tokens": 4096,
"tool_timeout": 30,
}
defaults.update(kwargs)
return ChatSession(**defaults)
class TestMaybeSynthReasoningBlock:
"""Direct unit tests for ``ChatSession._maybe_synth_reasoning_block``."""
def test_no_synth_when_provider_blocks_present(self) -> None:
# Anthropic / OpenAI Responses path — native blocks already
# carry the reasoning, no synth needed.
session = _make_session()
existing = [{"type": "thinking", "thinking": "x"}]
out = session._maybe_synth_reasoning_block(existing, ["should not be added"])
assert out is existing
def test_no_synth_when_reasoning_parts_empty(self) -> None:
session = _make_session()
out = session._maybe_synth_reasoning_block([], [])
assert out == []
def test_no_synth_when_reasoning_parts_only_whitespace(self) -> None:
session = _make_session()
out = session._maybe_synth_reasoning_block([], [" ", "\n\t"])
assert out == []
def test_synth_creates_reasoning_text_block(self) -> None:
session = _make_session()
out = session._maybe_synth_reasoning_block([], ["thought ", "process"])
assert len(out) == 1
assert out[0]["type"] == "reasoning_text"
assert out[0]["text"] == "thought process"
def test_synth_omits_source_when_no_server_type(self) -> None:
session = _make_session()
# No registry / no server_compat → source field omitted.
out = session._maybe_synth_reasoning_block([], ["text"])
assert "source" not in out[0]
def test_synth_includes_source_when_server_type_resolvable(self) -> None:
session = _make_session()
session._registry = SimpleNamespace(
get_config=lambda alias: SimpleNamespace(
capabilities={"server_compat": {"server_type": "vllm"}},
)
)
session._model_alias = "qwen3-32b"
out = session._maybe_synth_reasoning_block([], ["text"])
assert out[0]["source"] == "vllm"
def test_synth_handles_registry_exception(self) -> None:
# _resolve_server_type silently returns "" on any lookup error
# — synth still fires but omits the source field.
class BrokenRegistry:
def get_config(self, alias: str) -> Any:
raise KeyError(alias)
session = _make_session()
session._registry = BrokenRegistry()
session._model_alias = "missing"
out = session._maybe_synth_reasoning_block([], ["text"])
assert out[0]["text"] == "text"
assert "source" not in out[0]
class TestSyntheticBlockShapeContract:
"""The synthetic block shape MUST stay outside Anthropic's valid
block types so cross-model resumption falls through cleanly."""
def test_reasoning_text_not_in_anthropic_valid_types(self) -> None:
# If this assertion ever fails, the cross-model resumption
# safety story breaks: a synthetic block from a local-model
# session would reach Anthropic's wire as a malformed block.
assert "reasoning_text" not in ANTHROPIC_VALID_BLOCK_TYPES
def test_synthetic_block_falls_through_anthropic_shape_filter(self) -> None:
# Cross-model resumption regression: turn 1 was on a local
# model (synthetic block stamped), then the operator switched
# to Anthropic. The shape filter must reject the synthetic
# block and fall through to text+tool_calls rebuild.
provider = AnthropicProvider()
msg = {
"role": "assistant",
"content": "spoken answer",
"_provider_content": [
{"type": "reasoning_text", "text": "synth thought", "source": "vllm"},
],
}
_, converted = provider._convert_messages([msg])
assistant = next(m for m in converted if m["role"] == "assistant")
block_types = [b.get("type") for b in assistant["content"] if isinstance(b, dict)]
# Foreign block did NOT reach Anthropic's wire. Rebuilt from
# text only.
assert "reasoning_text" not in block_types
assert assistant["content"] == [{"type": "text", "text": "spoken answer"}]
class TestOpenAIChatExtractReasoningText:
"""``OpenAIChatCompletionsProvider.extract_reasoning_text`` reads
the synthetic block back out for UI rehydration."""
def test_reads_synthetic_reasoning_text_block(self) -> None:
provider = OpenAIChatCompletionsProvider()
blocks = [{"type": "reasoning_text", "text": "captured thought"}]
assert provider.extract_reasoning_text(blocks) == "captured thought"
def test_concatenates_multiple_blocks(self) -> None:
provider = OpenAIChatCompletionsProvider()
blocks = [
{"type": "reasoning_text", "text": "first"},
{"type": "reasoning_text", "text": "second"},
]
assert provider.extract_reasoning_text(blocks) == "first\nsecond"
def test_skips_other_block_types(self) -> None:
provider = OpenAIChatCompletionsProvider()
blocks = [
{"type": "thinking", "thinking": "anth"},
{"type": "reasoning", "summary": [{"text": "openai"}]},
{"type": "reasoning_text", "text": "chat"},
]
assert provider.extract_reasoning_text(blocks) == "chat"
def test_handles_empty_text_field(self) -> None:
provider = OpenAIChatCompletionsProvider()
blocks = [
{"type": "reasoning_text", "text": ""},
{"type": "reasoning_text", "text": "kept"},
]
assert provider.extract_reasoning_text(blocks) == "kept"
def test_handles_missing_text_field(self) -> None:
provider = OpenAIChatCompletionsProvider()
blocks = [
{"type": "reasoning_text"}, # no text
{"type": "reasoning_text", "text": "kept"},
]
assert provider.extract_reasoning_text(blocks) == "kept"
def test_returns_empty_for_no_synth_blocks(self) -> None:
provider = OpenAIChatCompletionsProvider()
blocks = [{"type": "thinking", "thinking": "x"}]
assert provider.extract_reasoning_text(blocks) == ""
class TestStreamResponseSynthBlockIntegration:
"""Integration test: drives a fake reasoning-emitting stream
through ``ChatSession._stream_response`` and asserts the
synthesizer wires up correctly. Pins the call site at
``session.py`` (where ``_maybe_synth_reasoning_block`` is invoked
on the assembled provider_blocks before stamping ``_provider_content``)
— without this, a future refactor that drops the synthesizer call
would silently break path-3 capture (vLLM/llama.cpp/Gemini-compat
reasoning would be visible live but invisible on history reload).
"""
def _make_stream(self, content: str, reasoning: str) -> Any:
"""Build an iterator of StreamChunks that mimic a path-3
capture (reasoning_delta chunks, content chunks, no
provider_blocks emitted).
"""
from turnstone.core.providers._protocol import StreamChunk, UsageInfo
chunks = []
# Reasoning first (matches live SSE order).
if reasoning:
chunks.append(StreamChunk(reasoning_delta=reasoning, is_first=True))
# Content next.
if content:
chunks.append(
StreamChunk(
content_delta=content,
is_first=not reasoning,
)
)
# Final chunk with finish_reason + usage.
chunks.append(
StreamChunk(
finish_reason="stop",
usage=UsageInfo(prompt_tokens=10, completion_tokens=20, total_tokens=30),
)
)
return iter(chunks)
def test_stream_response_stamps_synth_block_when_path3_reasoning_captured(
self,
) -> None:
"""Drive a fake stream emitting reasoning_delta chunks (no
native provider_blocks) through ``_stream_response``; assert
the resulting assistant_msg carries a synthetic reasoning_text
block stamped onto ``_provider_content``."""
session = _make_session()
# No registry → source field omitted from synth block.
stream = self._make_stream(content="Final answer.", reasoning="path-3 reasoning")
msg = session._stream_response(stream)
assert msg["role"] == "assistant"
assert msg["content"] == "Final answer."
# Synthetic block should be stamped onto _provider_content.
provider_content = msg.get("_provider_content")
assert isinstance(provider_content, list)
assert len(provider_content) == 1
assert provider_content[0]["type"] == "reasoning_text"
assert provider_content[0]["text"] == "path-3 reasoning"
def test_stream_response_no_synth_when_no_reasoning_captured(self) -> None:
"""Stream emits only content (no reasoning_delta). No synth
block stamped — _provider_content key absent on assistant_msg."""
session = _make_session()
stream = self._make_stream(content="just content", reasoning="")
msg = session._stream_response(stream)
assert msg["content"] == "just content"
# No synth block (and no native blocks either) → key absent.
assert "_provider_content" not in msg
def test_stream_response_synth_block_carries_source_when_server_type_resolvable(
self,
) -> None:
"""When the active model has server_compat.server_type set,
the synth block carries it as the ``source`` field."""
session = _make_session()
session._registry = SimpleNamespace(
get_config=lambda alias: SimpleNamespace(
capabilities={"server_compat": {"server_type": "vllm"}},
)
)
session._model_alias = "qwen3-32b"
stream = self._make_stream(content="answer", reasoning="reasoning text")
msg = session._stream_response(stream)
provider_content = msg.get("_provider_content")
assert isinstance(provider_content, list)
assert provider_content[0]["source"] == "vllm"
class TestResolveServerType:
"""Direct unit tests for the helper that pulls server_type from
the active model's capabilities dict."""
def test_returns_empty_when_no_registry(self) -> None:
session = _make_session()
session._registry = None
assert session._resolve_server_type() == ""
def test_returns_empty_when_no_alias(self) -> None:
session = _make_session()
session._registry = SimpleNamespace(
get_config=lambda alias: SimpleNamespace(capabilities={})
)
session._model_alias = ""
assert session._resolve_server_type() == ""
def test_returns_server_type_when_present(self) -> None:
session = _make_session()
session._registry = SimpleNamespace(
get_config=lambda alias: SimpleNamespace(
capabilities={"server_compat": {"server_type": "llama.cpp"}}
)
)
session._model_alias = "local-model"
assert session._resolve_server_type() == "llama.cpp"
def test_returns_empty_when_server_compat_missing(self) -> None:
session = _make_session()
session._registry = SimpleNamespace(
get_config=lambda alias: SimpleNamespace(
capabilities={"context_window": 32768},
)
)
session._model_alias = "local-model"
assert session._resolve_server_type() == ""
def test_returns_empty_on_exception(self) -> None:
class BrokenRegistry:
def get_config(self, alias: str) -> Any:
raise RuntimeError("boom")
session = _make_session()
session._registry = BrokenRegistry()
session._model_alias = "x"
assert session._resolve_server_type() == ""
turnstone/core/history_decoration.py
+51 -31
View File
@@ -269,46 +269,67 @@ def extract_advisories_from_tool_envelope(
if TYPE_CHECKING:
from turnstone.core.providers._anthropic import AnthropicProvider
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
from collections.abc import Callable
_anthropic_provider_singleton: AnthropicProvider | None = None
_openai_responses_provider_singleton: OpenAIResponsesProvider | None = None
from turnstone.core.providers._protocol import LLMProvider
def _get_anthropic_provider() -> AnthropicProvider:
global _anthropic_provider_singleton
if _anthropic_provider_singleton is None:
from turnstone.core.providers._anthropic import AnthropicProvider as _Anthropic
def _make_provider_factory(module_path: str, class_name: str) -> Callable[[], LLMProvider]:
"""Build a thread-unsafe lazy-init factory for a provider singleton.
_anthropic_provider_singleton = _Anthropic()
return _anthropic_provider_singleton
Single source of truth for the dispatcher's per-provider lazy-load
pattern — each block-type entry in ``_BLOCK_TYPE_PROVIDER_FACTORY``
closes over this with its own (module_path, class_name) pair.
Adding a fourth provider is a single tuple in the dict, not a
new 9-line getter.
"""
cached: dict[str, LLMProvider] = {}
def factory() -> LLMProvider:
if "instance" not in cached:
import importlib
module = importlib.import_module(module_path)
cached["instance"] = getattr(module, class_name)()
return cached["instance"]
return factory
def _get_openai_responses_provider() -> OpenAIResponsesProvider:
global _openai_responses_provider_singleton
if _openai_responses_provider_singleton is None:
from turnstone.core.providers._openai_responses import (
OpenAIResponsesProvider as _OpenAIResp,
)
_openai_responses_provider_singleton = _OpenAIResp()
return _openai_responses_provider_singleton
# Block-type → provider factory. Routing is structural — block shape
# is non-overlapping across providers by API design. Three block
# types are recognised today:
#
# * ``"thinking"`` — Anthropic native (Phase 1). Walks the
# ``thinking`` field on each block.
# * ``"reasoning"`` — OpenAI Responses native (Phase 3). Walks
# ``summary[*].text`` (always present) and ``content[*].text``
# (present when ``include=["reasoning.encrypted_content"]`` is
# requested AND the response carries raw reasoning text).
# * ``"reasoning_text"`` — synthetic, stamped by
# ``ChatSession._maybe_synth_reasoning_block`` for Chat Completions
# paths (vLLM, llama.cpp, Gemini-compat) where reasoning surfaces
# only as ``reasoning_delta`` chunks with no native block shape.
_BLOCK_TYPE_PROVIDER_FACTORY: dict[str, Callable[[], LLMProvider]] = {
"thinking": _make_provider_factory("turnstone.core.providers._anthropic", "AnthropicProvider"),
"reasoning": _make_provider_factory(
"turnstone.core.providers._openai_responses", "OpenAIResponsesProvider"
),
"reasoning_text": _make_provider_factory(
"turnstone.core.providers._openai_chat", "OpenAIChatCompletionsProvider"
),
}
def extract_reasoning_text_from_provider_content(provider_content: Any) -> str:
"""Dispatch reasoning extraction by first-block ``type`` field.
Routing is structural — block shape is non-overlapping across
providers by API design (Anthropic ``thinking``, OpenAI Responses
``reasoning``, Gemini ``thought``). Phase 1 wires Anthropic
extraction; OpenAI Responses returns ``""`` until Phase 3 adds the
``include=["reasoning.encrypted_content"]`` request flag. Returns
``""`` for empty / missing / non-list / unknown-type input.
Returns ``""`` for empty / missing / non-list / unknown-type input.
Pure transform — safe from any thread. Both history surfaces
(interactive ``_build_history`` and lifted ``make_history_handler``)
call this directly.
call this directly. See ``_BLOCK_TYPE_PROVIDER_FACTORY`` above
for the recognised block types and the providers that own them.
"""
if not isinstance(provider_content, list) or not provider_content:
return ""
@@ -318,11 +339,10 @@ def extract_reasoning_text_from_provider_content(provider_content: Any) -> str:
block_type = first_block.get("type")
if not isinstance(block_type, str):
return ""
if block_type == "thinking":
return _get_anthropic_provider().extract_reasoning_text(provider_content)
if block_type == "reasoning":
return _get_openai_responses_provider().extract_reasoning_text(provider_content)
return ""
factory = _BLOCK_TYPE_PROVIDER_FACTORY.get(block_type)
if factory is None:
return ""
return factory().extract_reasoning_text(provider_content)
def extract_reasoning_for_history(
turnstone/core/providers/_anthropic.py
+10 -8
View File
@@ -12,6 +12,7 @@ import sys
from typing import TYPE_CHECKING, Any
from turnstone.core.providers._protocol import (
MAX_REASONING_DISPLAY_BYTES,
CompletionResult,
ModelCapabilities,
StreamChunk,
@@ -80,6 +81,7 @@ _ANTHROPIC_DEFAULT = ModelCapabilities(
thinking_mode="manual",
supports_web_search=True,
supports_vision=True,
supports_reasoning_replay=True,
)
_ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
@@ -95,6 +97,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
supports_vision=True,
supports_temperature=False,
thinking_display="summarized",
supports_reasoning_replay=True,
),
"claude-opus-4-6": ModelCapabilities(
context_window=1000000,
@@ -106,6 +109,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
supports_web_search=True,
supports_tool_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
"claude-sonnet-4-6": ModelCapabilities(
context_window=1000000,
@@ -117,6 +121,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
supports_web_search=True,
supports_tool_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
"claude-haiku-4-5": ModelCapabilities(
context_window=200000,
@@ -125,6 +130,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
thinking_mode="manual",
supports_web_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
"claude-sonnet-4-5": ModelCapabilities(
context_window=200000,
@@ -133,6 +139,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
thinking_mode="manual",
supports_web_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
"claude-opus-4-5": ModelCapabilities(
context_window=200000,
@@ -143,6 +150,7 @@ _ANTHROPIC_CAPABILITIES: dict[str, ModelCapabilities] = {
effort_levels=("low", "medium", "high"),
supports_web_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
}
@@ -1040,17 +1048,11 @@ class AnthropicProvider:
if not parts:
return ""
joined = "\n".join(parts)
# Operator-friendly UI cap. Larger reasoning bodies are still
# stored verbatim in provider_data; only the rehydrated UI
# display payload is truncated.
if len(joined) > _MAX_REASONING_DISPLAY_BYTES:
return joined[:_MAX_REASONING_DISPLAY_BYTES]
if len(joined) > MAX_REASONING_DISPLAY_BYTES:
return joined[:MAX_REASONING_DISPLAY_BYTES]
return joined
_MAX_REASONING_DISPLAY_BYTES = 64 * 1024
def _normalize_finish_reason(reason: str) -> str:
"""Normalize Anthropic stop reasons to OpenAI-compatible strings."""
if reason == "end_turn":
turnstone/core/providers/_openai_chat.py
+30 -6
View File
@@ -24,6 +24,7 @@ from turnstone.core.providers._openai_common import (
sanitize_messages,
)
from turnstone.core.providers._protocol import (
MAX_REASONING_DISPLAY_BYTES,
CompletionResult,
ModelCapabilities,
StreamChunk,
@@ -389,9 +390,32 @@ class OpenAIChatCompletionsProvider:
self,
provider_blocks: list[dict[str, Any]] | None,
) -> str:
# OpenAI Chat (and the local-model server flavours that route
# through this adapter) have no first-class reasoning shape.
# Chat-template ``<think>`` content is captured via the inflight
# buffer for live UI but not persisted to ``provider_blocks``;
# Phase 4 may revisit.
return ""
"""Walk synthetic ``reasoning_text`` blocks (Phase 3 path-3
capture) and return the concatenated reasoning text.
OpenAI Chat Completions has no native reasoning shape on the
wire — vLLM ``--reasoning-parser``, llama.cpp
``reasoning_format``, and Gemini's OpenAI-compat endpoint all
surface reasoning as non-canonical ``delta.reasoning_content``
Pydantic extras. ``ChatSession._maybe_synth_reasoning_block``
captures these into a single ``{type: "reasoning_text", text,
source?}`` block when no native ``provider_blocks`` were
emitted. This extractor unwraps those for UI rehydration.
"""
if not isinstance(provider_blocks, list):
return ""
parts: list[str] = []
for block in provider_blocks:
if not isinstance(block, dict):
continue
if block.get("type") != "reasoning_text":
continue
text = block.get("text")
if isinstance(text, str) and text:
parts.append(text)
if not parts:
return ""
joined = "\n".join(parts)
if len(joined) > MAX_REASONING_DISPLAY_BYTES:
return joined[:MAX_REASONING_DISPLAY_BYTES]
return joined
turnstone/core/providers/_openai_common.py
+18
View File
@@ -33,6 +33,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("minimal", "low", "medium", "high"),
default_reasoning_effort="medium",
supports_vision=True,
supports_reasoning_replay=True,
),
"gpt-5-mini": ModelCapabilities(
context_window=400000,
@@ -41,6 +42,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("minimal", "low", "medium", "high"),
default_reasoning_effort="medium",
supports_vision=True,
supports_reasoning_replay=True,
),
"gpt-5-nano": ModelCapabilities(
context_window=400000,
@@ -49,6 +51,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("minimal", "low", "medium", "high"),
default_reasoning_effort="medium",
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5 pro — high reasoning only, extended output
"gpt-5-pro": ModelCapabilities(
@@ -58,6 +61,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("high",),
default_reasoning_effort="high",
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.1 — temperature OK when reasoning_effort=none (default)
"gpt-5.1": ModelCapabilities(
@@ -66,6 +70,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("none", "low", "medium", "high"),
default_reasoning_effort="none",
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.2 — adds xhigh
"gpt-5.2": ModelCapabilities(
@@ -74,6 +79,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("none", "low", "medium", "high", "xhigh"),
default_reasoning_effort="none",
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.2 pro — always-reasoning variant
"gpt-5.2-pro": ModelCapabilities(
@@ -83,6 +89,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("medium", "high", "xhigh"),
default_reasoning_effort="medium",
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.3 — same capabilities as 5.2 (matches gpt-5.3-chat-latest, codex)
"gpt-5.3": ModelCapabilities(
@@ -91,6 +98,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
reasoning_effort_values=("none", "low", "medium", "high", "xhigh"),
default_reasoning_effort="none",
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.4 — 1M context window, native tool search
"gpt-5.4": ModelCapabilities(
@@ -100,6 +108,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
default_reasoning_effort="none",
supports_tool_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.4 pro — always-reasoning, 1M context, native tool search
"gpt-5.4-pro": ModelCapabilities(
@@ -110,6 +119,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
default_reasoning_effort="medium",
supports_tool_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.5 — 1M context, native tool search, stronger agentic/tool use
"gpt-5.5": ModelCapabilities(
@@ -119,6 +129,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
default_reasoning_effort="none",
supports_tool_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
# GPT-5.5 pro — always-reasoning, 1M context, native tool search
"gpt-5.5-pro": ModelCapabilities(
@@ -129,6 +140,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
default_reasoning_effort="medium",
supports_tool_search=True,
supports_vision=True,
supports_reasoning_replay=True,
),
# O-series reasoning models
"o1": ModelCapabilities(
@@ -137,6 +149,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
supports_temperature=False,
supports_streaming=False,
supports_vision=True,
supports_reasoning_replay=True,
),
"o1-mini": ModelCapabilities(
context_window=128000,
@@ -144,18 +157,21 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
supports_temperature=False,
supports_streaming=False,
supports_vision=True,
supports_reasoning_replay=True,
),
"o3": ModelCapabilities(
context_window=200000,
max_output_tokens=100000,
supports_temperature=False,
supports_vision=True,
supports_reasoning_replay=True,
),
"o3-mini": ModelCapabilities(
context_window=200000,
max_output_tokens=100000,
supports_temperature=False,
supports_vision=True,
supports_reasoning_replay=True,
),
"o3-pro": ModelCapabilities(
context_window=200000,
@@ -163,12 +179,14 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
supports_temperature=False,
supports_streaming=False,
supports_vision=True,
supports_reasoning_replay=True,
),
"o4-mini": ModelCapabilities(
context_window=200000,
max_output_tokens=100000,
supports_temperature=False,
supports_vision=True,
supports_reasoning_replay=True,
),
# Search models — always search on every request, no reasoning_effort
"gpt-5-search-api": ModelCapabilities(
turnstone/core/providers/_openai_responses.py
+174 -17
View File
@@ -28,6 +28,7 @@ from turnstone.core.providers._openai_common import (
sanitize_messages,
)
from turnstone.core.providers._protocol import (
MAX_REASONING_DISPLAY_BYTES,
CompletionResult,
ModelCapabilities,
StreamChunk,
@@ -92,16 +93,68 @@ class OpenAIResponsesProvider:
@staticmethod
def _convert_messages(
messages: list[dict[str, Any]],
*,
replay_reasoning_to_model: bool = False,
) -> tuple[str | None, list[dict[str, Any]]]:
"""Convert Chat Completions messages to Responses API input items.
Returns ``(instructions, input_items)`` where *instructions* is the
concatenated system/developer messages (or ``None``) and *input_items*
is the Responses API ``input`` array.
When *replay_reasoning_to_model* is True, stored ``_provider_content``
reasoning items (``type=="reasoning"``, captured via
``include=["reasoning.encrypted_content"]`` on a prior turn)
are emitted as ``ResponseReasoningItemParam`` input items
immediately before the assistant message they belong to. The
SDK explicitly documents this round-trip pattern at
``response_reasoning_item_param.py:33-37``: "Be sure to include
these items in your ``input`` to the Responses API for
subsequent turns of a conversation if you are manually managing
context". Even with ``store=False``, ``encrypted_content``
round-trips correctly per ``response_create_params.py:70-74``.
When *replay_reasoning_to_model* is False, reasoning items are silently
dropped (they were stripped from the wire by ``sanitize_messages``
anyway, but we also skip the input-item emission step).
"""
# Capture ``_provider_content`` reasoning items per ASSISTANT
# ORDINAL (not raw message index) BEFORE sanitization strips
# the underscore-prefixed key. Position-by-index would be
# unsafe: ``sanitize_messages`` drops orphan tool results
# (``_openai_common.py:489-498`` / ``:521-535``) and inserts
# synthesized error tool messages for orphaned tool_calls
# (``:510-517``). Either operation shifts subsequent message
# indices, so a pre-vs-post-sanitize index match would
# silently miss reasoning attachments after any tool-message
# repair. Assistant messages themselves are never dropped or
# duplicated by sanitize_messages — only tool messages — so
# the n-th assistant in the original list is invariably the
# n-th assistant in the sanitized list. Ordinal-keyed lookup
# survives any tool-message length change.
reasoning_by_assistant_ordinal: dict[int, list[dict[str, Any]]] = {}
if replay_reasoning_to_model:
ord_pre = 0
for raw_msg in messages:
if raw_msg.get("role") != "assistant":
continue
pc = raw_msg.get("_provider_content")
if isinstance(pc, list):
items_to_replay = [
b for b in pc if isinstance(b, dict) and b.get("type") == "reasoning"
]
if items_to_replay:
reasoning_by_assistant_ordinal[ord_pre] = items_to_replay
ord_pre += 1
messages = sanitize_messages(messages)
instructions_parts: list[str] = []
items: list[dict[str, Any]] = []
# Track assistant ordinal in the SANITIZED list so the lookup
# into reasoning_by_assistant_ordinal stays aligned with the
# original-list ordinal. See the long comment above for why
# ordinal is invariant under sanitization.
assistant_ordinal_post = 0
for msg in messages:
role = msg.get("role", "")
@@ -129,9 +182,15 @@ class OpenAIResponsesProvider:
items.append(item)
elif role == "assistant":
# With store=False, provider_blocks cannot be replayed as input
# (output format != input format, and IDs aren't persisted).
# Rebuild from the normalized content/tool_calls instead.
# Phase 3 reasoning replay: emit stored reasoning items
# BEFORE the assistant message they belong to. The SDK
# expects reasoning items to appear in input order
# alongside the assistant turn that produced them.
for r_item in reasoning_by_assistant_ordinal.get(assistant_ordinal_post, []):
item_for_input = _reasoning_item_for_input(r_item)
if item_for_input is not None:
items.append(item_for_input)
assistant_ordinal_post += 1
# Text content → assistant message (plain string for input)
if content:
@@ -239,11 +298,33 @@ class OpenAIResponsesProvider:
reasoning_effort: str,
deferred_names: frozenset[str] | None,
capabilities: ModelCapabilities | None = None,
replay_reasoning_to_model: bool = True,
) -> dict[str, Any]:
"""Build the kwargs dict for ``client.responses.create/stream``."""
"""Build the kwargs dict for ``client.responses.create/stream``.
``replay_reasoning_to_model`` (Phase 3 of the reasoning-
persistence feature) gates two things together:
1. ``include=["reasoning.encrypted_content"]`` on the request
(so the API surfaces ``encrypted_content`` on reasoning
items in ``provider_blocks``).
2. ``_convert_messages`` round-tripping stored reasoning items
from ``_provider_content`` as ``input`` items on subsequent
turns (the SDK's ``ResponseReasoningItemParam`` shape).
Both are also gated by ``caps.supports_reasoning_replay`` —
models without a reasoning lane (gpt-4o, etc.) silently skip
replay even if the operator flag is set.
"""
caps = capabilities or self.get_capabilities(model)
instructions, input_items = self._convert_messages(messages)
# The two replay gates collapse to a single boolean: replay is
# active only when both the operator flag AND the model's
# capability allow it. Threaded into ``_convert_messages`` so
# stored reasoning items become input items on the next call.
replay_active = bool(replay_reasoning_to_model and caps.supports_reasoning_replay)
instructions, input_items = self._convert_messages(
messages, replay_reasoning_to_model=replay_active
)
tools = apply_tool_search(caps, tools, deferred_names)
converted_tools = self._convert_tools(tools, caps)
@@ -261,6 +342,13 @@ class OpenAIResponsesProvider:
"store": False,
}
if replay_active:
# SDK doc (response_create_params.py:70-74): with
# ``include=["reasoning.encrypted_content"]`` the API
# surfaces opaque ``encrypted_content`` on reasoning
# items, enabling stateless replay even with ``store=False``.
kwargs["include"] = ["reasoning.encrypted_content"]
if instructions:
kwargs["instructions"] = instructions
@@ -293,12 +381,11 @@ class OpenAIResponsesProvider:
deferred_names: frozenset[str] | None = None,
cancel_ref: list[Any] | None = None,
capabilities: ModelCapabilities | None = None,
# Phase 2 reasoning-persistence kwarg — accepted for Protocol
# conformance. Phase 3 will gate ``include=
# ["reasoning.encrypted_content"]`` on this flag once OpenAI
# Responses captures replayable reasoning items. Today the
# request kwargs don't carry reasoning, so the flag is unused
# but threaded for forward-compat.
# Phase 3 reasoning-persistence kwarg — gates
# ``include=["reasoning.encrypted_content"]`` on the request
# AND ``_convert_messages`` round-tripping stored reasoning
# items as input. Both are also gated by
# ``caps.supports_reasoning_replay`` inside ``_build_kwargs``.
replay_reasoning_to_model: bool = True,
) -> Iterator[StreamChunk]:
if extra_params:
@@ -312,6 +399,7 @@ class OpenAIResponsesProvider:
reasoning_effort,
deferred_names,
capabilities=capabilities,
replay_reasoning_to_model=replay_reasoning_to_model,
)
kwargs["stream"] = True
@@ -481,7 +569,7 @@ class OpenAIResponsesProvider:
extra_params: dict[str, Any] | None = None,
deferred_names: frozenset[str] | None = None,
capabilities: ModelCapabilities | None = None,
# See create_streaming above for the Phase 2 reasoning-persistence rationale.
# See create_streaming above for the Phase 3 reasoning-persistence rationale.
replay_reasoning_to_model: bool = True,
) -> CompletionResult:
if extra_params:
@@ -495,6 +583,7 @@ class OpenAIResponsesProvider:
reasoning_effort,
deferred_names,
capabilities=capabilities,
replay_reasoning_to_model=replay_reasoning_to_model,
)
log.debug(
@@ -592,8 +681,76 @@ class OpenAIResponsesProvider:
self,
provider_blocks: list[dict[str, Any]] | None,
) -> str:
# Phase 3 will wire this to walk ``type=="reasoning"`` items
# captured via ``include=["reasoning.encrypted_content"]``.
# Today the request kwargs don't pass ``include`` so reasoning
# items in ``provider_blocks`` carry no replayable text.
return ""
if not isinstance(provider_blocks, list):
return ""
parts: list[str] = []
for block in provider_blocks:
if not isinstance(block, dict):
continue
if block.get("type") != "reasoning":
continue
# Per ``ResponseReasoningItem`` (response_reasoning_item.py:31-62):
# ``summary`` is the human-readable summary list (always
# present), ``content`` is the raw reasoning text list
# (optional). We surface both — summary is what the model
# produces by default; content is only present on certain
# configurations.
for s in block.get("summary") or []:
if isinstance(s, dict) and s.get("type") == "summary_text":
text = s.get("text")
if isinstance(text, str) and text:
parts.append(text)
for c in block.get("content") or []:
if isinstance(c, dict) and c.get("type") == "reasoning_text":
text = c.get("text")
if isinstance(text, str) and text:
parts.append(text)
if not parts:
return ""
joined = "\n".join(parts)
if len(joined) > MAX_REASONING_DISPLAY_BYTES:
return joined[:MAX_REASONING_DISPLAY_BYTES]
return joined
def _reasoning_item_for_input(stored: dict[str, Any]) -> dict[str, Any] | None:
"""Project a stored reasoning item into ``ResponseReasoningItemParam`` shape.
The output of a Responses API call carries reasoning items shaped
like ``ResponseReasoningItem`` (response_reasoning_item.py:31-62);
we stored those verbatim into ``provider_blocks`` via
``item.model_dump()`` (``_iter_stream`` line 415-420 captures all
output items). To replay them as input on the next turn, the
Responses API expects ``ResponseReasoningItemParam``
(response_reasoning_item_param.py:31-62) which has the same shape
minus ``status`` (a server-only field).
The ``id``, ``summary``, ``content``, ``encrypted_content``, and
``type`` fields all round-trip directly. We project explicitly
rather than ``del stored["status"]; return stored`` so callers
aren't surprised by mutation of the source dict.
Returns ``None`` when ``id`` is missing or non-string — per the
SDK schema (``response_reasoning_item_param.py:39``) ``id`` is
``Required[str]``; sending an empty string would emit a malformed
input item that the API may either reject (4xx) or silently
misroute. Caller skips appending when None is returned. Items
captured via the streaming layer always have ``id`` populated, so
this guard is defensive against manually-constructed or migrated
storage rows.
"""
item_id = stored.get("id")
if not isinstance(item_id, str) or not item_id:
return None
out: dict[str, Any] = {
"type": "reasoning",
"id": item_id,
"summary": stored.get("summary") or [],
}
content = stored.get("content")
if content:
out["content"] = content
encrypted = stored.get("encrypted_content")
if encrypted:
out["encrypted_content"] = encrypted
return out
turnstone/core/providers/_protocol.py
+21
View File
@@ -87,6 +87,27 @@ class ModelCapabilities:
supports_tool_search: bool = False
supports_vision: bool = False
thinking_display: str = "" # "summarized" for models that omit thinking by default
# Phase 3 reasoning-persistence: gate the per-model
# ``replay_reasoning_to_model`` flag. When False, the wire-build
# path skips replay regardless of the operator flag (defends
# against operators flipping the flag on a model whose API has
# no reasoning-replay shape — e.g. OpenAI Chat Completions, where
# reasoning is purely server-side and never round-trips). Set
# True for: Anthropic models with ``thinking_mode != "none"``,
# OpenAI Responses o-series + GPT-5+ (``include=
# ["reasoning.encrypted_content"]`` round-trip). Path-3 capture
# (Chat Completions / vLLM / llama.cpp / Gemini-compat) is
# persist-only and doesn't gate on this flag.
supports_reasoning_replay: bool = False
# Operator-friendly UI cap on reasoning text returned from
# ``LLMProvider.extract_reasoning_text``. Single source of truth so a
# tuning change propagates to every provider's display path uniformly.
# Larger reasoning bodies are still stored verbatim in
# ``provider_data``; only the rehydrated UI display payload is
# truncated. 64 KiB matches the briefing's recommendation.
MAX_REASONING_DISPLAY_BYTES = 64 * 1024
def _lookup_capabilities(
turnstone/core/session.py
+73 -1
View File
@@ -1096,6 +1096,73 @@ class ChatSession:
return self._cached_capabilities
return self._resolve_capabilities(p, m, "")
def _resolve_server_type(self, alias: str | None = None) -> str:
"""Read ``server_compat.server_type`` for an alias from the registry.
Used by :meth:`_maybe_synth_reasoning_block` to tag synthetic
path-3 reasoning blocks with their origin server (vllm,
llama.cpp, sglang, etc.) — informational today, useful for
future per-server replay paths. Returns ``""`` on any lookup
miss; the synthetic block then omits the ``source`` field.
"""
target_alias = alias or self._model_alias or ""
if not self._registry or not target_alias:
return ""
try:
cfg: ModelConfig = self._registry.get_config(target_alias)
sc = (
cfg.capabilities.get("server_compat")
if isinstance(cfg.capabilities, dict)
else None
)
if isinstance(sc, dict):
return str(sc.get("server_type") or "")
except Exception:
pass
return ""
def _maybe_synth_reasoning_block(
self,
provider_blocks: list[dict[str, Any]],
reasoning_parts: list[str],
) -> list[dict[str, Any]]:
"""Stamp captured ``reasoning_parts`` as a synthetic ``reasoning_text``
block when no native ``provider_blocks`` were emitted.
Anthropic emits native ``thinking`` blocks; OpenAI Responses
emits native ``reasoning`` items via ``output_item.done``.
Both populate ``provider_blocks`` directly during streaming
and need no synthesis here.
OpenAI Chat Completions (with vLLM ``--reasoning-parser``,
llama.cpp ``reasoning_format``, or Gemini's ``/v1beta/openai/``
endpoint if it surfaces ``reasoning_content``) streams
reasoning as ``reasoning_delta`` chunks but never produces a
provider_blocks item. Without this synthesis the captured
text would be dropped at the end of the stream — visible live,
invisible on page reload.
The synthetic block uses ``type="reasoning_text"`` (NOT
``"thinking"``) so it falls through Phase 2's
``ANTHROPIC_VALID_BLOCK_TYPES`` shape filter on cross-model
resumption — protecting against operator-switches from a
local-model session to Anthropic, which would otherwise hit
Anthropic's input boundary with an unsigned ``thinking`` block.
"""
if provider_blocks:
return provider_blocks
text = "".join(reasoning_parts)
if not text.strip():
return provider_blocks
block: dict[str, Any] = {
"type": "reasoning_text",
"text": text,
}
server_type = self._resolve_server_type()
if server_type:
block["source"] = server_type
return [block]
def _resolve_replay_reasoning_to_model(self, alias: str | None = None) -> bool:
"""Read ``ModelConfig.replay_reasoning_to_model`` for an alias.
@@ -3793,7 +3860,12 @@ class ChatSession:
)
# Store raw provider content blocks for multi-turn preservation
# (e.g. Anthropic web_search_tool_result with encrypted_content)
# (e.g. Anthropic web_search_tool_result with encrypted_content).
# Phase 3 path-3 capture: when no native blocks were emitted but
# ``reasoning_delta`` chunks accumulated text, synthesize a
# ``reasoning_text`` block so the captured reasoning survives
# past the live stream and surfaces on history reload.
provider_blocks = self._maybe_synth_reasoning_block(provider_blocks, reasoning_parts)
if provider_blocks:
msg["_provider_content"] = provider_blocks