feat(task-agent): carry the provider-native reasoning lane in the sub-harness

A task agent's replayed turns now carry the native reasoning lane the
model produced (Anthropic thinking blocks + signatures, OpenAI Responses
reasoning items, Gemini thought_signature blocks, vLLM/llama.cpp parsed
reasoning text) instead of being rebuilt from content + tool_calls with
the reasoning dropped — restoring reasoning continuity across the
agent's own multi-turn tool loop on every provider lane.

The prerequisite is the id half: replace legalize_tool_call_ids with
restore_provider_tool_ids, a lowering pass that maps the session-minted
sub-tool ids back to the provider's own ids on the transient wire copy
(from the per-run mint map, never by string-splitting). The native
tool_use block is replayed verbatim — its id and signature untouched —
and the top-level mirror and tool_result agree with it on every request.
The minted id stays the sole internal key (registry, DOM, recall,
cancel ledger), #820 unchanged.

Chat-Completions lane: non-streaming create_completion now surfaces
reasoning/reasoning_content as CompletionResult.reasoning (the twin of
the streaming reasoning_delta extraction), and the agent seam runs the
Phase 5 vLLM reasoning-field replay against the agent's own provider
and alias. The native lane is finalized by a shared helper
(_finalize_provider_blocks) so the main loop and the sub-harness cannot
drift; replay honors the per-model replay_reasoning_to_model flag on
every lane, and llama.cpp stays capture-only, matching the main loop.

(cherry picked from commit d660819142)

tests/test_lowering.py

@@ -12,17 +12,15 @@ this pins the behaviour the old ``_anthropic`` ``pc_tool_ids`` /
 from __future__ import annotations
 
 import json
-import re
 from typing import Any
 
 from turnstone.core.lowering import (
     CANCELLED_TOOL_RESULT,
     _find_orphaned_tool_calls,
-    legalize_tool_call_ids,
     repair_wire_messages,
+    restore_provider_tool_ids,
     sanitize_tool_call_arguments,
     tool_args_preview,
-    wire_safe_tool_call_id,
     wire_valid_arguments,
 )
 
@@ -346,65 +344,79 @@ def test_pipeline_every_emitted_arguments_is_a_json_object() -> None:
 
 
 # --------------------------------------------------------------------------- #
-# legalize_tool_call_ids — the agent-wire id projection (defensive hardening).
+# restore_provider_tool_ids — the agent-wire id map (minted → provider-original).
 #
 # Sub-agent tool ids are minted "{parent}::r{run}s{step}::{provider_id}" for
-# session-unique correlation (registry / DOM / recall).  These replay fine on
-# the lenient anthropic-compatible deployment, but the pass projects the long,
-# "::"-containing ids to a deterministic wire-safe token on the transient wire
-# copy so an agent's self-built history stays valid on a hypothetically
-# stricter backend, keeping assistant call / tool result pairing intact.
+# session-unique correlation (registry / DOM / recall).  On the wire the pass
+# maps them BACK to the provider's own ids from the per-run mint map, so the
+# provider-native tool_use block (replayed verbatim, id never rewritten), the
+# top-level tool_calls mirror, and the tool_result all agree on every request.
 # --------------------------------------------------------------------------- #
-def test_wire_safe_tool_call_id_passthrough_when_legal() -> None:
-    # Provider-issued ids (their own echo) and uuid-filled ids are untouched.
-    for tc_id in ("call_9dSVOCr8sPbk3f0oJU8IruBP", "toolu_01ABCdef", "call_" + "a" * 32):
-        assert wire_safe_tool_call_id(tc_id) is tc_id
+def test_restore_ids_identity_on_empty_map() -> None:
+    msgs = [_assistant_calls(_call("task-1::r1s1::call_0", "{}")), _tool("task-1::r1s1::call_0")]
+    assert restore_provider_tool_ids(msgs, {}) is msgs
 
 
-def test_wire_safe_tool_call_id_rewrites_illegal_charset_and_length() -> None:
-    minted = "call_0::r1s1::call_0"
-    long_id = "call_" + "a" * 60
-    for bad in (minted, long_id):
-        safe = wire_safe_tool_call_id(bad)
-        assert safe != bad
-        assert re.fullmatch(r"[a-zA-Z0-9_-]{1,40}", safe)
-    # Deterministic (pairing across messages and across requests), distinct inputs distinct.
-    assert wire_safe_tool_call_id(minted) == wire_safe_tool_call_id(minted)
-    assert wire_safe_tool_call_id(minted) != wire_safe_tool_call_id(long_id)
-
-
-def test_legalize_ids_identity_when_all_legal() -> None:
+def test_restore_ids_identity_when_nothing_matches() -> None:
     msgs = [_assistant_calls(_call("call_1", "{}")), _tool("call_1")]
-    assert legalize_tool_call_ids(msgs) is msgs
+    assert restore_provider_tool_ids(msgs, {"task-1::r1s1::call_0": "call_0"}) is msgs
 
 
-def test_legalize_ids_rewrites_call_and_result_consistently() -> None:
-    minted = "task-1::r1s1::call_0"
+def test_restore_ids_maps_call_and_result_to_provider_original() -> None:
+    minted = "task-1::r1s1::toolu_01AB"
     msgs = [_assistant_calls(_call(minted, "{}")), _tool(minted)]
-    out = legalize_tool_call_ids(msgs)
-    wire_id = out[0]["tool_calls"][0]["id"]
-    assert re.fullmatch(r"[a-zA-Z0-9_-]{1,40}", wire_id)
-    assert out[1]["tool_call_id"] == wire_id  # pairing survives the projection
+    out = restore_provider_tool_ids(msgs, {minted: "toolu_01AB"})
+    assert out[0]["tool_calls"][0]["id"] == "toolu_01AB"
+    assert out[1]["tool_call_id"] == "toolu_01AB"  # pairing restored on both sides
     # Copy-on-write: the input messages (the canonical-adjacent dicts) are unmutated.
     assert msgs[0]["tool_calls"][0]["id"] == minted
     assert msgs[1]["tool_call_id"] == minted
 
 
-def test_legalize_ids_leaves_legal_siblings_untouched() -> None:
+def test_restore_ids_recovers_originals_containing_the_mint_delimiter() -> None:
+    # Recovery is by MAP, not by string-splitting the mint suffix: a provider
+    # id that itself contains "::" round-trips exactly.
+    original = "srv::call::0"
+    minted = f"task-1::r1s1::{original}"
+    msgs = [_assistant_calls(_call(minted, "{}")), _tool(minted)]
+    out = restore_provider_tool_ids(msgs, {minted: original})
+    assert out[0]["tool_calls"][0]["id"] == original
+    assert out[1]["tool_call_id"] == original
+
+
+def test_restore_ids_duplicate_originals_across_turns() -> None:
+    # A local server reissuing "call_0" every turn: two distinct minted ids
+    # both restore to "call_0" — the proven prior wire shape, each round
+    # pairing with its adjacent result.
+    m1, m2 = "task-1::r1s1::call_0", "task-1::r1s2::call_0"
+    msgs = [
+        _assistant_calls(_call(m1, "{}")),
+        _tool(m1),
+        _assistant_calls(_call(m2, "{}")),
+        _tool(m2),
+    ]
+    out = restore_provider_tool_ids(msgs, {m1: "call_0", m2: "call_0"})
+    assert out[0]["tool_calls"][0]["id"] == "call_0"
+    assert out[1]["tool_call_id"] == "call_0"
+    assert out[2]["tool_calls"][0]["id"] == "call_0"
+    assert out[3]["tool_call_id"] == "call_0"
+
+
+def test_restore_ids_leaves_unmapped_siblings_untouched() -> None:
     minted = "task-1::r1s2::call_1"
     msgs = [
         _assistant_calls(_call("call_ok", "{}"), _call(minted, "{}")),
         _tool("call_ok"),
         _tool(minted),
     ]
-    out = legalize_tool_call_ids(msgs)
+    out = restore_provider_tool_ids(msgs, {minted: "call_1"})
     assert out[0]["tool_calls"][0]["id"] == "call_ok"
     assert out[1]["tool_call_id"] == "call_ok"
-    assert out[0]["tool_calls"][1]["id"] == out[2]["tool_call_id"]
-    assert "::" not in out[0]["tool_calls"][1]["id"]
+    assert out[0]["tool_calls"][1]["id"] == "call_1"
+    assert out[2]["tool_call_id"] == "call_1"
 
 
-def test_legalize_ids_skips_empty_and_non_string() -> None:
+def test_restore_ids_skips_empty_and_non_string() -> None:
     # Empty ids belong to repair_wire_messages' back-fill; non-strings are
     # someone else's malformation — neither is this pass's to invent.
     msgs = [
@@ -413,6 +425,6 @@ def test_legalize_ids_skips_empty_and_non_string() -> None:
         ),
         {"role": "tool", "tool_call_id": None, "content": "x"},
     ]
-    out = legalize_tool_call_ids(msgs)
+    out = restore_provider_tool_ids(msgs, {"task-1::r1s1::x": "x"})
     assert out[0]["tool_calls"][0]["id"] == ""
     assert out[1]["tool_call_id"] is None

tests/test_provider_openai_responses_reasoning.py

@@ -277,6 +277,41 @@ class TestConvertMessagesReasoningReplay:
         types = [it.get("type") for it in items]
         assert "reasoning" not in types
 
+    def test_agent_shaped_turn_pairs_reasoning_with_restored_call_ids(
+        self, provider: OpenAIResponsesProvider
+    ) -> None:
+        # The sub-agent wire shape (native lane carried, minted ids already
+        # restored to the provider originals by the lowering map): the stored
+        # reasoning item rides immediately before the function_call rebuilt
+        # from the SAME original call id, and the function_call_output pairs
+        # to it — the ordering + id agreement the Responses API requires when
+        # replaying reasoning across an agent's own tool loop.
+        messages = [
+            {"role": "user", "content": "go"},
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [{"id": "call_orig1", "function": {"name": "f", "arguments": "{}"}}],
+                "_provider_content": [
+                    {"type": "reasoning", "id": "rs_1", "summary": [], "encrypted_content": "enc"},
+                    {
+                        "type": "function_call",
+                        "call_id": "call_orig1",
+                        "name": "f",
+                        "arguments": "{}",
+                    },
+                ],
+            },
+            {"role": "tool", "tool_call_id": "call_orig1", "content": "out"},
+        ]
+        _, items = provider._convert_messages(messages, replay_reasoning_to_model=True)
+        types = [it.get("type") for it in items]
+        assert types == ["message", "reasoning", "function_call", "function_call_output"]
+        assert items[1]["id"] == "rs_1"
+        assert items[1]["encrypted_content"] == "enc"
+        assert items[2]["call_id"] == "call_orig1"
+        assert items[3]["call_id"] == "call_orig1"
+
     def test_no_reasoning_items_when_provider_content_lacks_reasoning(
         self, provider: OpenAIResponsesProvider
     ) -> None:

tests/test_providers.py

@@ -3174,6 +3174,80 @@ class TestAnthropicProviderBlocks:
         assert assistant_msg["role"] == "assistant"
         assert assistant_msg["content"] == [{"type": "text", "text": "Hi there"}]
 
+    def test_agent_native_lane_with_restore_map_is_wire_consistent(self) -> None:
+        """The sub-agent wire shape: an assistant Turn carrying the provider-
+        native lane, its minted tool id restored to the provider original by
+        the lowering map.  The native blocks replay verbatim (thinking +
+        signature untouched) and the native tool_use id, the top-level
+        mirror, and the tool_result all agree."""
+        from turnstone.core.lowering import restore_provider_tool_ids
+        from turnstone.core.trajectory import (
+            ProviderNative,
+            ToolCall,
+            Turn,
+            dicts_from_turns,
+        )
+
+        thinking = {"type": "thinking", "thinking": "look first", "signature": "sig_1"}
+        tool_use = {"type": "tool_use", "id": "toolu_01X", "name": "f", "input": {}}
+        minted = "task-1::r1s1::toolu_01X"
+        turns = [
+            Turn.user("go"),
+            Turn.assistant(
+                "using f",
+                tool_calls=(ToolCall(id=minted, name="f", arguments="{}"),),
+                native=ProviderNative(
+                    producer="anthropic",
+                    blocks=(thinking, {"type": "text", "text": "using f"}, tool_use),
+                ),
+            ),
+            Turn.tool(minted, "out"),
+        ]
+        wire = restore_provider_tool_ids(dicts_from_turns(turns), {minted: "toolu_01X"})
+        _, converted = self.provider._convert_messages(wire, replay_reasoning_to_model=True)
+        assistant = converted[1]
+        assert [b["type"] for b in assistant["content"]] == ["thinking", "text", "tool_use"]
+        assert assistant["content"][0]["signature"] == "sig_1"
+        assert assistant["content"][2]["id"] == "toolu_01X"
+        tool_results = [b for b in converted[2]["content"] if b.get("type") == "tool_result"]
+        assert tool_results and tool_results[0]["tool_use_id"] == "toolu_01X"
+
+    def test_agent_native_lane_without_restore_map_orphans_the_result(self) -> None:
+        """Documents why the id map is a PREREQUISITE of carrying the native
+        lane, not hygiene: without it the tool_result arrives with the minted
+        id, matches no native tool_use, and the converter drops it as an
+        orphan — leaving an unanswered tool_use on the wire (a provider
+        rejection)."""
+        from turnstone.core.trajectory import (
+            ProviderNative,
+            ToolCall,
+            Turn,
+            dicts_from_turns,
+        )
+
+        tool_use = {"type": "tool_use", "id": "toolu_01X", "name": "f", "input": {}}
+        minted = "task-1::r1s1::toolu_01X"
+        turns = [
+            Turn.user("go"),
+            Turn.assistant(
+                "",
+                tool_calls=(ToolCall(id=minted, name="f", arguments="{}"),),
+                native=ProviderNative(producer="anthropic", blocks=(tool_use,)),
+            ),
+            Turn.tool(minted, "out"),
+        ]
+        _, converted = self.provider._convert_messages(
+            dicts_from_turns(turns), replay_reasoning_to_model=True
+        )
+        all_results = [
+            b
+            for m in converted
+            if isinstance(m.get("content"), list)
+            for b in m["content"]
+            if isinstance(b, dict) and b.get("type") == "tool_result"
+        ]
+        assert all_results == []
+
     def test_block_to_dict_with_model_dump(self) -> None:
         """_block_to_dict uses model_dump(exclude_none=True) when available."""
         from turnstone.core.providers._anthropic import _block_to_dict
@@ -4235,6 +4309,56 @@ class TestOpenAIResponsesProvider:
         assert caps.supports_tool_search is True
 
 
+class TestOpenAIChatReasoningCapture:
+    """Non-streaming ``create_completion`` surfaces the Chat-Completions
+    lane's non-canonical reasoning (vLLM ``--reasoning-parser``, llama.cpp
+    ``reasoning_format``) as ``CompletionResult.reasoning`` — the twin of the
+    streaming path's ``reasoning_delta`` extraction, same attribute pair and
+    precedence."""
+
+    @staticmethod
+    def _client(*, reasoning: Any = None, reasoning_content: Any = None) -> MagicMock:
+        msg = MagicMock()
+        msg.content = "ok"
+        msg.tool_calls = None
+        msg.annotations = None
+        msg.reasoning = reasoning
+        msg.reasoning_content = reasoning_content
+        choice = MagicMock()
+        choice.message = msg
+        choice.finish_reason = "stop"
+        resp = MagicMock()
+        resp.choices = [choice]
+        resp.usage = None
+        client = MagicMock()
+        client.chat.completions.create.return_value = resp
+        return client
+
+    def _complete(self, client: MagicMock):
+        provider = OpenAIChatCompletionsProvider()
+        return provider.create_completion(
+            client=client, model="m", messages=[{"role": "user", "content": "hi"}]
+        )
+
+    def test_reasoning_content_captured(self) -> None:
+        result = self._complete(self._client(reasoning_content="thought text"))
+        assert result.reasoning == "thought text"
+
+    def test_reasoning_attribute_takes_precedence(self) -> None:
+        result = self._complete(self._client(reasoning="direct", reasoning_content="parsed"))
+        assert result.reasoning == "direct"
+
+    def test_absent_reasoning_is_empty(self) -> None:
+        result = self._complete(self._client())
+        assert result.reasoning == ""
+
+    def test_non_string_reasoning_collapses_to_empty(self) -> None:
+        # A server surfacing a structured reasoning object (not text) must not
+        # leak a non-str into the result.
+        result = self._complete(self._client(reasoning={"odd": True}))
+        assert result.reasoning == ""
+
+
 class TestResponsesMessageConversion:
     """Tests for _convert_messages — Chat Completions format to Responses API."""
 

tests/test_session.py

@@ -2567,7 +2567,7 @@ class TestAgentChildRegistration:
         # _tool_status, the card's own data-call-id row — so two runs with the
         # same parent id still alias at the card level.  Parent ids are
         # main-loop ids; de-colliding them is the main-loop id-hygiene
-        # follow-up (see legalize_tool_call_ids' docstring), not this change.
+        # follow-up, not this change.
         from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 
         session = _make_session()
@@ -2597,17 +2597,16 @@ class TestAgentChildRegistration:
         assert minted == ["call_0::r1s1::call_0", "call_0::r2s1::call_0"]
         assert len(set(minted)) == 2
 
-    def test_agent_wire_is_validity_legalized(self):
+    def test_agent_wire_restores_provider_ids_and_sanitizes_args(self):
         # The agent seam bypasses the main-loop wire prep and builds its own
-        # history, so it runs the same two validity passes itself.  Drive one
-        # tool turn whose call carries both a minted "::" id (projected as
-        # defensive hardening) and malformed non-object arguments (a strict
+        # history, so it runs its own validity passes.  Drive one tool turn
+        # whose call carries a minted "::" id (mapped back to the provider's
+        # own id on the wire) and malformed non-object arguments (a strict
         # renderer json.loads and 400s them), then assert the REPLAY request
-        # the second _api_call sends carries the projected id (call/result
-        # pairing intact) and object-shaped arguments.  The internal id keeps
-        # the "::" form.
+        # the second _api_call sends carries the PROVIDER-ORIGINAL id on both
+        # the call and its result, and object-shaped arguments.  The internal
+        # id keeps the minted "::" form.
         import json as _json
-        import re as _re
 
         from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 
@@ -2662,19 +2661,201 @@ class TestAgentChildRegistration:
         # Internal id (registry) keeps the minted "::" form.
         internal = session.ui.note_agent_child.call_args.args[0]
         assert internal == "task-1::r1s1::call_0"
-        # The SECOND request replays the tool turn: wire id is the legal
-        # projection, consistent between the call and its result; arguments
-        # are legalized to a JSON object.
+        # The SECOND request replays the tool turn: the wire carries the
+        # provider's own id, consistent between the call and its result (the
+        # shape the provider-native tool_use block also holds, so a native
+        # replay and a rebuild agree); arguments are legalized to a JSON
+        # object.
         replay = seen_messages[1]
         wire_calls = [tc for m in replay if m.get("tool_calls") for tc in m["tool_calls"]]
         wire_results = [m for m in replay if m.get("role") == "tool"]
         assert wire_calls and wire_results
-        wire_id = wire_calls[0]["id"]
-        assert _re.fullmatch(r"[a-zA-Z0-9_-]{1,40}", wire_id), wire_id
-        assert wire_results[0]["tool_call_id"] == wire_id
-        assert wire_id != internal
+        assert wire_calls[0]["id"] == "call_0"
+        assert wire_results[0]["tool_call_id"] == "call_0"
         assert isinstance(_json.loads(wire_calls[0]["function"]["arguments"]), dict)
 
+    def test_agent_carries_native_lane_and_replays_thinking_anthropic(self):
+        # The load-bearing fidelity pin: a thinking-model agent's SECOND
+        # request must carry the prior assistant turn's native lane verbatim
+        # — thinking block and signature untouched — with the provider's own
+        # tool_use id agreeing across the native block, the restored
+        # top-level mirror, and the tool_result.  Pre-native-lane, the seam
+        # rebuilt the turn from content + tool_calls and the model re-reasoned
+        # from scratch every tool turn (and commercial Anthropic rejects a
+        # thinking-enabled tool_use turn without its thinking block).
+        from turnstone.core.providers._anthropic import AnthropicProvider
+
+        class _Block:
+            def __init__(self, **d):
+                self._d = d
+                for k, v in d.items():
+                    setattr(self, k, v)
+
+            def model_dump(self, **_kw):
+                return dict(self._d)
+
+        session = _make_session()
+        session._provider = AnthropicProvider()
+        session.ui.note_agent_child = MagicMock()
+
+        seen: list[dict] = []
+        call_count = [0]
+
+        def fake_stream(**kwargs):
+            seen.append(kwargs)
+            call_count[0] += 1
+            resp = MagicMock()
+            if call_count[0] == 1:
+                resp.content = [
+                    _Block(type="thinking", thinking="check the file first", signature="sig_v1"),
+                    _Block(type="text", text="reading"),
+                    _Block(type="tool_use", id="toolu_01AB", name="read_file", input={"path": "x"}),
+                ]
+                resp.stop_reason = "tool_use"
+            else:
+                resp.content = [_Block(type="text", text="done")]
+                resp.stop_reason = "end_turn"
+            resp.usage = None
+            mgr = MagicMock()
+            mgr.__enter__ = MagicMock(
+                return_value=MagicMock(get_final_message=MagicMock(return_value=resp))
+            )
+            mgr.__exit__ = MagicMock(return_value=False)
+            return mgr
+
+        session.client.messages.stream = fake_stream
+
+        def fake_prepare(tc_dict, **_kwargs):
+            return {
+                "call_id": tc_dict["id"],
+                "func_name": "read_file",
+                "needs_approval": False,
+                "execute": lambda p: (p["call_id"], "contents"),
+            }
+
+        with (
+            patch.object(session, "_prepare_tool", side_effect=fake_prepare),
+            patch.object(session, "_resolve_replay_reasoning_to_model", return_value=True),
+        ):
+            session._run_agent(
+                [Turn.user("x")],
+                tools=[{"type": "function", "function": {"name": "read_file", "parameters": {}}}],
+                label="task",
+                parent_call_id="task-1",
+            )
+
+        # Internal key stays minted — the nesting registry saw the "::" id.
+        assert session.ui.note_agent_child.call_args.args[0] == "task-1::r1s1::toolu_01AB"
+        # Second request: the assistant wire turn IS the native lane.
+        replay = seen[1]["messages"]
+        assistant = next(
+            m for m in replay if m["role"] == "assistant" and isinstance(m.get("content"), list)
+        )
+        kinds = [b.get("type") for b in assistant["content"]]
+        assert kinds == ["thinking", "text", "tool_use"]
+        assert assistant["content"][0]["thinking"] == "check the file first"
+        assert assistant["content"][0]["signature"] == "sig_v1"  # byte-untouched
+        assert assistant["content"][2]["id"] == "toolu_01AB"  # provider-original
+        tool_results = [
+            b
+            for m in replay
+            if m["role"] == "user" and isinstance(m.get("content"), list)
+            for b in m["content"]
+            if isinstance(b, dict) and b.get("type") == "tool_result"
+        ]
+        assert tool_results and tool_results[0]["tool_use_id"] == "toolu_01AB"
+
+    def test_agent_synthesizes_reasoning_and_attaches_vllm_replay_field(self):
+        # Chat-Completions lane (vLLM): non-streaming ``reasoning_content`` is
+        # captured into CompletionResult.reasoning, synthesized into the agent
+        # turn's native lane as a ``reasoning_text`` block by the SAME
+        # finalize helper the main loop uses — source-tagged from the AGENT
+        # alias — and replayed on the next request as vLLM's non-standard
+        # ``reasoning`` field (Phase 5 at the agent seam; the internal
+        # ``_provider_content`` key itself never reaches the wire).
+        from types import SimpleNamespace
+
+        from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
+        from turnstone.core.providers._openai_common import OPENAI_COMPAT_DEFAULT
+
+        session = _make_session()
+        session._provider = OpenAIChatCompletionsProvider()
+        session._model_alias = "loc-qwen"
+        session._registry = MagicMock()
+        session._registry.resolve_agent_alias.return_value = None
+        session._registry.resolve_agent_effort.return_value = None
+        session._registry.get_config.return_value = SimpleNamespace(
+            server_compat={"server_type": "vllm"}, replay_reasoning_to_model=True
+        )
+        session.ui.note_agent_child = MagicMock()
+
+        seen_messages: list[list[dict]] = []
+        call_count = [0]
+
+        def fake_create(**kwargs):
+            seen_messages.append(kwargs.get("messages") or [])
+            call_count[0] += 1
+            resp = MagicMock()
+            choice = MagicMock()
+            if call_count[0] == 1:
+                choice.finish_reason = "tool_calls"
+                tc = MagicMock()
+                tc.id = "call_0"
+                tc.function.name = "read_file"
+                tc.function.arguments = '{"path": "x"}'
+                choice.message.tool_calls = [tc]
+                choice.message.content = None
+                choice.message.reasoning = None
+                choice.message.reasoning_content = "scan the repo first"
+            else:
+                choice.finish_reason = "stop"
+                choice.message.tool_calls = None
+                choice.message.content = "done"
+                choice.message.reasoning = None
+                choice.message.reasoning_content = None
+            resp.choices = [choice]
+            resp.usage = MagicMock(prompt_tokens=10, completion_tokens=5)
+            return resp
+
+        session.client.chat.completions.create = fake_create
+
+        def fake_prepare(tc_dict, **_kwargs):
+            return {
+                "call_id": tc_dict["id"],
+                "func_name": "read_file",
+                "needs_approval": False,
+                "execute": lambda p: (p["call_id"], "contents"),
+            }
+
+        turns = [Turn.user("x")]
+        with (
+            patch.object(session, "_prepare_tool", side_effect=fake_prepare),
+            patch.object(session, "_resolve_capabilities", return_value=OPENAI_COMPAT_DEFAULT),
+            patch.object(session, "_provider_extra_params", return_value={}),
+        ):
+            session._run_agent(
+                turns,
+                tools=[{"type": "function", "function": {"name": "read_file"}}],
+                label="task",
+                parent_call_id="task-1",
+            )
+
+        # The agent Turn carries the synthesized native lane, source-tagged
+        # via the agent alias (alias threading through the shared helper).
+        assistant_turn = turns[1]
+        assert assistant_turn.native is not None
+        assert assistant_turn.native.producer == "openai-compatible"
+        assert assistant_turn.native.blocks == (
+            {"type": "reasoning_text", "text": "scan the repo first", "source": "vllm"},
+        )
+        # The replay request carries the vLLM ``reasoning`` field on the
+        # assistant turn; the internal ``_provider_content`` key is stripped
+        # by the provider's sanitize before the wire.
+        replay = seen_messages[1]
+        assistant_wire = next(m for m in replay if m.get("role") == "assistant")
+        assert assistant_wire.get("reasoning") == "scan the repo first"
+        assert "_provider_content" not in assistant_wire
+
 
 class TestRunAgentDenialMessage:
     """A denied sub-tool must surface the SPECIFIC denial reason that

turnstone/core/lowering.py

@@ -19,14 +19,14 @@ This module owns the three provider-neutral lowering passes:
   can't reject the whole request.  Mutates the transient wire copy only — the
   canonical trajectory keeps the raw output.  See
   :func:`sanitize_tool_call_arguments`.  The id sibling,
-  :func:`legalize_tool_call_ids`, projects session-minted sub-agent tool ids
-  (``{parent}::r{run}s{step}::{provider_id}`` — long and ``::``-containing) to
-  deterministic safe tokens as DEFENSIVE hardening (no backend turnstone
-  targets is known to reject them; the prior ``::`` id format ran fine); it
-  runs at the AGENT wire seam (``ChatSession._run_agent``'s ``_api_call``)
-  only — main-loop ids are provider-issued or uuid-filled, and main-loop
-  assistant turns
-  carry a provider-native lane whose block ids must stay untouched.
+  :func:`restore_provider_tool_ids`, maps session-minted sub-agent tool ids
+  (``{parent}::r{run}s{step}::{provider_id}``) back to the provider's OWN ids
+  on the wire copy, so the provider-native block lane — whose ``tool_use``
+  blocks carry the provider id verbatim, under a reasoning signature that must
+  never be touched — stays id-consistent with the top-level mirror and the
+  tool results.  It runs at the AGENT wire seam (``ChatSession._run_agent``'s
+  ``_api_call``) only — main-loop ids are provider-issued or uuid-filled and
+  already consistent with their native lane.
 * **repair** (validity) — synthesizing cancellation results for orphaned client
   tool calls.  See :func:`repair_wire_messages`.
 
@@ -60,7 +60,6 @@ their own.
 
 from __future__ import annotations
 
-import hashlib
 import json
 import re
 from typing import Any
@@ -296,61 +295,40 @@ def sanitize_tool_call_arguments(messages: list[dict[str, Any]]) -> list[dict[st
     return messages if out is None else out
 
 
-# A conservative wire-legal shape for tool-call ids: alphanumerics, ``_`` and
-# ``-``, up to 40 chars — the tightest charset/length a provider wire is
-# plausibly strict about.  DEFENSIVE only: no backend turnstone targets is
-# known to require it — the deployment is lenient anthropic-compatible vLLM,
-# and the prior ``::``-containing id format replayed fine — so an id already
-# matching this passes through untouched and the projection is belt-and-braces.
-_WIRE_TOOL_ID_RE = re.compile(r"[a-zA-Z0-9_-]{1,40}")
+def restore_provider_tool_ids(
+    messages: list[dict[str, Any]], id_map: dict[str, str]
+) -> list[dict[str, Any]]:
+    """Return *messages* with session-minted sub-agent tool ids mapped back to
+    the provider's own ids — the id half of the legalize pass, applied at the
+    AGENT wire seam.
 
+    *id_map* is the per-``_run_agent`` ``{minted_id: provider_original_id}``
+    record built at the mint site.  Rewriting assistant ``tool_calls[*].id``
+    and tool ``tool_call_id`` back to the provider originals makes every wire
+    representation agree: the provider-native ``tool_use`` block (which holds
+    the provider id verbatim and must never be rewritten — its bytes sit under
+    the turn's reasoning signature), the top-level ``tool_calls`` mirror, and
+    the ``tool_result``.  A translator that replays the native lane and one
+    that rebuilds from ``tool_calls`` therefore emit the same ids, so the
+    pairing holds on both paths.  The minted id stays the internal key
+    (registry / DOM / recall / cancel ledger) untouched — only the transient
+    wire copy is mapped.
 
-def wire_safe_tool_call_id(tc_id: str) -> str:
-    """Return *tc_id* unchanged if it already matches the conservative
-    wire-legal shape (:data:`_WIRE_TOOL_ID_RE`), else a deterministic safe
-    token (``tid_`` + 32 hex chars of its SHA-256, 36 chars total).
-
-    Determinism is the contract: the same original id maps to the same token
-    in the assistant ``tool_use`` and its ``tool_result`` (intra-request
-    pairing) and across successive requests that replay the same turns.
-    Provider-issued ids (their own echo) and ``_ensure_tool_call_ids``'s
-    uuid fills already match, so they pass through unchanged; only
-    session-minted composite ids (``{parent}::r{run}s{step}::{provider_id}``)
-    are projected.
-    """
-    if _WIRE_TOOL_ID_RE.fullmatch(tc_id):
-        return tc_id
-    return "tid_" + hashlib.sha256(tc_id.encode("utf-8", "surrogatepass")).hexdigest()[:32]
-
-
-def legalize_tool_call_ids(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Return *messages* with every tool-call id made wire-legal — the id
-    half of the legalize pass, applied at the AGENT wire seam.
-
-    Rewrites assistant ``tool_calls[*].id`` and tool ``tool_call_id`` through
-    :func:`wire_safe_tool_call_id`, so a minted ``::`` id is projected to a
-    plain token before the wire (call/result pairing survives via
-    determinism).  This is DEFENSIVE hardening — the ids replay fine on the
-    lenient anthropic-compatible deployment today; the projection just keeps
-    an agent's self-built history valid on a hypothetically stricter backend.
-    Empty and non-string ids are left alone — the empty back-fill belongs to
-    :func:`repair_wire_messages`'s domain, and a non-string is someone else's
-    malformation to surface, not silently rename.
+    Replaying the provider's own ids to the producing provider is the proven
+    prior behaviour, including the duplicate-ish ids a local server that
+    reissues per-response ids ("call_0") produces — an agent run is pinned to
+    one provider, so the ids always return to the backend that issued them.
+    Ids not in the map (uuid back-fills, an unparented run that never minted)
+    pass through untouched; recovery is by MAP ONLY, never by string-splitting
+    the mint suffix (provider ids can contain surprising characters,
+    including the mint's own delimiter).
 
     Copy-on-write + identity-preserving, exactly like
-    :func:`sanitize_tool_call_arguments`: a conversation whose ids are all
-    legal returns the same object.  Applied at the AGENT seam only, where the
-    ``::`` mint is the sole illegal-id source; NOT wired into the main-loop
-    wire prep, because assistant turns there can carry a provider-native block
-    lane whose ids must stay byte-identical to the mirrored ``tool_calls``,
-    and projecting them would desync the two.  If the main loop ever needs the
-    same hygiene (a mid-session ``/model`` switch replays the prior backend's
-    ids — a vLLM ``chatcmpl-tool-`` + 32-hex id is 46 chars, long enough that a
-    backend with a short ``tool_call_id`` cap could reject it), the fix is
-    de-colliding + legalizing at ``_ensure_tool_call_ids`` in a way that also
-    rewrites the native lane — the broader main-loop id-hygiene follow-up, out
-    of scope here.
+    :func:`sanitize_tool_call_arguments`: an empty map or a conversation with
+    no minted id returns the same object.
     """
+    if not id_map:
+        return messages
     out: list[dict[str, Any]] | None = None  # copy-on-write: None until first fix
     for idx, msg in enumerate(messages):
         role = msg.get("role")
@@ -358,29 +336,23 @@ def legalize_tool_call_ids(messages: list[dict[str, Any]]) -> list[dict[str, Any
             repaired: list[dict[str, Any]] | None = None
             for ci, tc in enumerate(msg["tool_calls"]):
                 tc_id = tc.get("id")
-                if not isinstance(tc_id, str) or not tc_id:
-                    continue
-                safe = wire_safe_tool_call_id(tc_id)
-                if safe is tc_id:
+                original = id_map.get(tc_id) if isinstance(tc_id, str) else None
+                if original is None or original == tc_id:
                     continue
                 if repaired is None:
                     repaired = list(msg["tool_calls"])
-                log.debug("wire.tool_id_legalized", call_id=tc_id, wire_id=safe)
-                repaired[ci] = {**tc, "id": safe}
+                repaired[ci] = {**tc, "id": original}
             if repaired is not None:
                 if out is None:
                     out = list(messages)
                 out[idx] = {**msg, "tool_calls": repaired}
         elif role == "tool":
             tc_id = msg.get("tool_call_id")
-            if not isinstance(tc_id, str) or not tc_id:
-                continue
-            safe = wire_safe_tool_call_id(tc_id)
-            if safe is tc_id:
-                continue
-            if out is None:
-                out = list(messages)
-            out[idx] = {**msg, "tool_call_id": safe}
+            original = id_map.get(tc_id) if isinstance(tc_id, str) else None
+            if original is not None and original != tc_id:
+                if out is None:
+                    out = list(messages)
+                out[idx] = {**msg, "tool_call_id": original}
     return messages if out is None else out
 
 

turnstone/core/providers/_openai_chat.py

@@ -344,6 +344,11 @@ class OpenAIChatCompletionsProvider:
         if annotations:
             content = format_citations(content, annotations)
 
+        # Non-canonical reasoning text (vLLM ``--reasoning-parser``, llama.cpp
+        # ``reasoning_format``) — same attribute pair, same precedence, as the
+        # streaming delta extraction above, so the two paths can't drift.
+        reasoning = getattr(msg, "reasoning", None) or getattr(msg, "reasoning_content", None)
+
         usage = extract_usage(getattr(response, "usage", None))
 
         result = CompletionResult(
@@ -352,6 +357,7 @@ class OpenAIChatCompletionsProvider:
             finish_reason=choice.finish_reason or "stop",
             usage=usage,
             provider_blocks=provider_blocks,
+            reasoning=reasoning if isinstance(reasoning, str) else "",
         )
         log.debug(
             "openai.chat.response",

turnstone/core/providers/_protocol.py

@@ -59,6 +59,12 @@ class CompletionResult:
     finish_reason: str = "stop"
     usage: UsageInfo | None = None
     provider_blocks: list[dict[str, Any]] = field(default_factory=list)
+    # Non-canonical reasoning text surfaced by Chat-Completions-lane servers
+    # (vLLM ``--reasoning-parser``, llama.cpp ``reasoning_format``) — the
+    # non-streaming twin of ``StreamChunk.reasoning_delta``.  Lanes whose
+    # reasoning rides ``provider_blocks`` natively (Anthropic ``thinking``,
+    # OpenAI Responses ``reasoning`` items) leave it empty.
+    reasoning: str = ""
 
 
 @dataclass(frozen=True)

turnstone/core/session.py

@@ -69,8 +69,8 @@ from turnstone.core.lowering import (
     UNOBSERVED_OUTCOME_CLAUSE,
     drop_empty_user_turns,
     fold_system_turns,
-    legalize_tool_call_ids,
     repair_wire_messages,
+    restore_provider_tool_ids,
     sanitize_tool_call_arguments,
     tool_args_preview,
     wire_valid_arguments,
@@ -185,6 +185,7 @@ from turnstone.core.tools import (
 )
 from turnstone.core.trajectory import (
     EffectStatus,
+    ProviderNative,
     Role,
     TextBlock,
     ToolCall,
@@ -2119,6 +2120,7 @@ class ChatSession:
         self,
         provider_blocks: list[dict[str, Any]],
         reasoning_parts: list[str],
+        alias: str | None = None,
     ) -> list[dict[str, Any]]:
         """Stamp captured ``reasoning_parts`` as a synthetic ``reasoning_text``
         block when no reasoning-bearing block already appears in
@@ -2165,6 +2167,11 @@ class ChatSession:
         (:meth:`_maybe_attach_vllm_chat_reasoning`) reads
         ``cfg.server_compat`` directly rather than the synthetic
         block's tag.
+
+        *alias* names the model whose server produced the reasoning —
+        the sub-agent loop passes its own agent alias so the source tag
+        names the agent's server, not the session primary's.  ``None``
+        (the main-loop caller) keeps the primary-alias resolution.
         """
         text = "".join(reasoning_parts)
         if not text.strip():
@@ -2179,13 +2186,42 @@ class ChatSession:
             "type": "reasoning_text",
             "text": text,
         }
-        server_type = self._resolve_server_type()
+        server_type = self._resolve_server_type(alias)
         if server_type:
             block["source"] = server_type
         # Append rather than replace so non-reasoning fidelity blocks
         # (e.g. Google tool_calls with thought_signature) survive.
         return [*provider_blocks, block]
 
+    def _finalize_provider_blocks(
+        self,
+        provider_blocks: list[dict[str, Any]],
+        reasoning_parts: list[str],
+        *,
+        has_tool_calls: bool,
+        alias: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """Finalize an assistant turn's provider-native block lane: synthesize
+        the path-3 ``reasoning_text`` block when reasoning arrived only as
+        loose text (:meth:`_maybe_synth_reasoning_block`), then enforce the
+        native↔tool_calls mirror in memory — a truncation that cleared
+        ``tool_calls`` can leave an orphan client ``tool_use`` in the captured
+        blocks, which a same-provider replay would send with no matching
+        ``tool_result`` (see ``storage._utils.normalize_native_for_save``, the
+        save-time chokepoint with the same gate).
+
+        The ONE builder both harnesses share: the main-loop stream accumulator
+        and the sub-agent loop (``_run_agent``) finalize their captured blocks
+        here, so how a native lane is assembled cannot drift between them.
+        Returns a possibly-empty list; callers attach it only when non-empty.
+        """
+        provider_blocks = self._maybe_synth_reasoning_block(
+            provider_blocks, reasoning_parts, alias=alias
+        )
+        if provider_blocks and not has_tool_calls:
+            provider_blocks = strip_orphan_client_tool_blocks(provider_blocks)
+        return provider_blocks
+
     def _resolve_replay_reasoning_to_model(
         self,
         alias: str | None = None,
@@ -7024,18 +7060,12 @@ class ChatSession:
 
         # Store raw provider content blocks for multi-turn preservation
         # (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)
-        # Enforce the native↔tool_calls mirror in memory too.  A truncation that cleared
-        # tool_calls (finish_reason="length") can leave an orphan tool_use in the captured
-        # blocks; the save-time chokepoint fixes the persisted row, but a same-session
-        # continuation reads this in-memory copy, so strip the orphan here as well.
-        # See storage._utils.normalize_native_for_save.
-        if provider_blocks and not msg.get("tool_calls"):
-            provider_blocks = strip_orphan_client_tool_blocks(provider_blocks)
+        # Finalization (path-3 reasoning synthesis + the in-memory
+        # native↔tool_calls mirror gate) is shared with the sub-agent loop —
+        # see _finalize_provider_blocks.
+        provider_blocks = self._finalize_provider_blocks(
+            provider_blocks, reasoning_parts, has_tool_calls=bool(msg.get("tool_calls"))
+        )
         if provider_blocks:
             msg["_provider_content"] = provider_blocks
 
@@ -15264,32 +15294,38 @@ class ChatSession:
             turns: list[Turn],
             _tools: list[dict[str, Any]] | None = tools,
         ) -> CompletionResult:
-            # NOTE: Phase 5 vLLM ``reasoning`` field replay is intentionally
-            # NOT wired here.  Agent assistant messages are built from
-            # ``CompletionResult.content + tool_calls`` only (no
-            # ``_provider_content`` carried), so the helper would no-op
-            # every turn anyway.  Task agents are excluded from the
-            # persistence/replay contract — their conversation history
-            # is in-memory and rebuilt per ``_run_agent`` invocation.
             # Lower the trajectory once, not once per retry attempt — ``turns``
             # is invariant across attempts (the retry path only sleeps and
-            # re-sends the same messages).  Two validity passes, the same two
-            # the main-loop wire prep runs, both applied here because agent
-            # calls bypass that prep and build their own history:
+            # re-sends the same messages).  Agent calls bypass the main-loop
+            # wire prep and build their own history, so the seam runs its own
+            # passes:
             #   * ``sanitize_tool_call_arguments`` — a local model can emit an
             #     unterminated / non-object ``arguments`` with a non-``length``
             #     finish reason; a strict renderer (vLLM ``deepseek_v4``) then
             #     ``json.loads`` it and 400s every request that replays it.
             #     (Documented in-tree for the main loop; agents hit the same
             #     backends, so the same guard applies.)
-            #   * ``legalize_tool_call_ids`` — project the session-minted
-            #     ``::`` sub-tool ids to plain tokens, DEFENSIVE only: they
-            #     replay fine on the lenient anthropic-compatible deployment,
-            #     this just keeps the history valid on a hypothetically
-            #     stricter backend.  Deterministic, so call/result pairing
-            #     holds within and across requests; the minted id stays the
-            #     internal key (registry / DOM / recall) untouched.
-            wire = legalize_tool_call_ids(sanitize_tool_call_arguments(dicts_from_turns(turns)))
+            #   * ``restore_provider_tool_ids`` — map the session-minted ``::``
+            #     sub-tool ids back to the provider's own ids, so the
+            #     provider-native ``tool_use`` block (replayed verbatim below,
+            #     under a reasoning signature that must not be touched), the
+            #     ``tool_calls`` mirror, and the ``tool_result`` all agree on
+            #     the wire.  The minted id stays the internal key (registry /
+            #     DOM / recall / cancel ledger) untouched.
+            #   * ``_maybe_attach_vllm_chat_reasoning`` — Phase 5 replay for
+            #     the agent's own turns, live here since agent turns carry
+            #     ``_provider_content`` (reasoning included); the helper's
+            #     three gates (Chat-Completions provider, server_type vllm,
+            #     operator flag) all resolve against the AGENT's provider and
+            #     alias, exactly like the main loop's send paths.
+            # Agent trajectories stay excluded from the persistence/replay
+            # contract — history is in-memory, rebuilt per ``_run_agent``
+            # invocation; the native lane carried here serves the WITHIN-RUN
+            # reasoning continuity of the agent's own tool loop.
+            wire = restore_provider_tool_ids(
+                sanitize_tool_call_arguments(dicts_from_turns(turns)), wire_id_map
+            )
+            wire = self._maybe_attach_vllm_chat_reasoning(wire, agent_provider, agent_alias)
             last_err: Exception | None = None
             for attempt in range(self._MAX_RETRIES + 1):
                 try:
@@ -15332,10 +15368,14 @@ class ChatSession:
         # for the PARENT task_agent call too.  ``sub_step_seq`` is monotonic
         # across the WHOLE run, so ids stay distinct across turns even when
         # the provider reuses per-response ids ("call_0") for sub-tools.
+        # ``wire_id_map`` records minted → provider-original for every mint,
+        # read by ``restore_provider_tool_ids`` in ``_api_call`` (the id map
+        # IS the recovery path — never string-split the mint suffix).
         with self._agent_run_seq_lock:
             self._agent_run_seq += 1
             run_seq = self._agent_run_seq
         sub_step_seq = 0
+        wire_id_map: dict[str, str] = {}
         while max_tool_turns < 0 or turn < max_tool_turns:
             self._check_cancelled()
             try:
@@ -15387,15 +15427,19 @@ class ChatSession:
                 # id traceable and is what the frontend's "::" child checks
                 # key off.  Every downstream consumer (nesting registry,
                 # error-flags, DOM data-call-id, recall, cancel ledger) keys
-                # on this ONE id; the wire alone sees a deterministic legal
-                # projection instead (``legalize_tool_call_ids`` in
-                # ``_api_call`` — strict providers reject ``::`` ids), which
-                # preserves intra-request call/result pairing.  Skipped for a
-                # top-level run (no parent → no nesting).
+                # on this ONE id; the wire alone sees the provider's original
+                # ids restored from ``wire_id_map`` instead
+                # (``restore_provider_tool_ids`` in ``_api_call``), so the
+                # top-level mirror, the ``tool_result``, and the native
+                # ``tool_use`` block — which keeps the provider id verbatim
+                # and is never rewritten — agree on every request.  Skipped
+                # for a top-level run (no parent → no nesting).
                 if parent_call_id:
                     for tc in result.tool_calls:
                         sub_step_seq += 1
-                        tc["id"] = f"{parent_call_id}::r{run_seq}s{sub_step_seq}::{tc['id']}"
+                        original_id = tc["id"]
+                        tc["id"] = f"{parent_call_id}::r{run_seq}s{sub_step_seq}::{original_id}"
+                        wire_id_map[tc["id"]] = original_id
                 agent_tool_calls = tuple(
                     ToolCall(
                         id=tc["id"],
@@ -15404,7 +15448,29 @@ class ChatSession:
                     )
                     for tc in result.tool_calls
                 )
-            agent_turns.append(Turn.assistant(result.content or "", tool_calls=agent_tool_calls))
+            # Carry the provider-native lane (thinking blocks, signatures,
+            # Responses reasoning items, synthesized ``reasoning_text``) so
+            # the agent's own multi-turn tool loop keeps its reasoning
+            # continuity instead of re-reasoning from scratch each turn —
+            # the same fidelity the main loop keeps, finalized by the same
+            # shared builder.  ``producer`` is the agent's own provider: a
+            # run is pinned to one provider, so the blocks always replay to
+            # the backend that produced them (and the translators' per-block
+            # shape filters drop anything foreign).
+            native_blocks = self._finalize_provider_blocks(
+                result.provider_blocks,
+                [result.reasoning] if result.reasoning else [],
+                has_tool_calls=bool(result.tool_calls),
+                alias=agent_alias,
+            )
+            native = (
+                ProviderNative(producer=agent_provider.provider_name, blocks=tuple(native_blocks))
+                if native_blocks
+                else None
+            )
+            agent_turns.append(
+                Turn.assistant(result.content or "", tool_calls=agent_tool_calls, native=native)
+            )
 
             if not result.tool_calls:
                 content = result.content or "(no output)"