fix(task-agent): move the blank-id gate into the shared native-lane builder

The blank-provider-id gate lived only at the _run_agent call site while
the main-loop stream accumulator has the identical back-fill-then-carry
seam — and it over-dropped, discarding the reasoning lane for exactly
the servers that emit blank ids. The gate now lives in
_finalize_provider_blocks as a had_blank_ids parameter both harnesses
thread: client tool blocks (which keep the blank id the mirror back-fill
never reached) are stripped, and when any were present the remaining
Messages-shaped blocks go with them (a surviving native lane REPLACES
the rebuilt content on the Anthropic translator, so a lane missing its
tool_use would orphan every mirrored call) — while shape-invalid
reasoning residuals (reasoning_text, Responses reasoning items) are
kept. This also closes the pre-existing main-loop case: a Gemini
openai-compat turn with a blank tool id no longer persists a raw
fidelity dict whose blank id the swap would resurrect on every replay.

The Google fidelity-swap legalization now reuses the canonical
lowering.legalized_arguments (made public) instead of a hand-rolled
narrower copy: dict-shaped arguments are serialized rather than
collapsed to {}, the standard wire.tool_args_legalized breadcrumb is
logged, and a degenerate non-dict function entry passes through
untouched instead of raising.

(cherry picked from commit 98cefc3660)

tests/test_providers.py

@@ -1953,6 +1953,58 @@ class TestGoogleProviderFidelity:
         assert tcs[1]["function"]["arguments"] == '{"ok": 1}'
         assert tcs[1]["thought_signature"] == "sig456"
 
+    def test_prepare_messages_swap_serializes_dict_arguments(self) -> None:
+        # The internal-shape case the shared legalize helper handles: a raw
+        # fidelity dict whose arguments landed as an unserialized dict is
+        # json.dumps'd — content preserved, not collapsed to "{}".
+        from turnstone.core.providers._google import GoogleProvider
+
+        prov = GoogleProvider()
+        msgs = [
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {"id": "c1", "type": "function", "function": {"name": "f", "arguments": "{}"}},
+                ],
+                "_provider_content": [
+                    {
+                        "id": "c1",
+                        "type": "function",
+                        "function": {"name": "f", "arguments": {"path": "/tmp/x"}},
+                        "thought_signature": "sig1",
+                    },
+                ],
+            },
+            {"role": "tool", "tool_call_id": "c1", "content": "ok"},
+        ]
+        cleaned = prov._prepare_messages(msgs)
+        tc = cleaned[0]["tool_calls"][0]
+        assert json.loads(tc["function"]["arguments"]) == {"path": "/tmp/x"}
+        assert tc["thought_signature"] == "sig1"
+
+    def test_prepare_messages_swap_passes_non_dict_function_through(self) -> None:
+        # A degenerate fidelity block with function=None must pass through
+        # untouched (the prior behaviour), not raise.
+        from turnstone.core.providers._google import GoogleProvider
+
+        prov = GoogleProvider()
+        msgs = [
+            {
+                "role": "assistant",
+                "content": "x",
+                "tool_calls": [
+                    {"id": "c1", "type": "function", "function": {"name": "f", "arguments": "{}"}},
+                ],
+                "_provider_content": [
+                    {"id": "c1", "type": "function", "function": None},
+                ],
+            },
+            {"role": "tool", "tool_call_id": "c1", "content": "ok"},
+        ]
+        cleaned = prov._prepare_messages(msgs)
+        assert cleaned[0]["tool_calls"][0]["function"] is None
+
     def test_non_streaming_captures_provider_blocks(self) -> None:
         from turnstone.core.providers._google import GoogleProvider
 

tests/test_session.py

@@ -2770,10 +2770,11 @@ class TestAgentChildRegistration:
         # the tool_calls mirror (_ensure_tool_call_ids) — but the native
         # tool_use block keeps the blank id verbatim.  Carrying the lane for
         # that turn would replay a native tool_use whose id matches no
-        # tool_result (Anthropic orphans the result and 400s).  The seam must
-        # skip the native lane for exactly that turn and fall back to the
-        # rebuild path, where every wire representation uses the back-filled
-        # id consistently.
+        # tool_result (Anthropic orphans the result and 400s).  The shared
+        # builder must drop the whole Messages-shaped lane for exactly that
+        # turn (a residual thinking block would REPLACE the rebuilt content
+        # and lose the tool_use) and fall back to the rebuild path, where
+        # every wire representation uses the back-filled id consistently.
         from turnstone.core.providers._anthropic import AnthropicProvider
 
         class _Block:
@@ -2859,6 +2860,82 @@ class TestAgentChildRegistration:
         assert tool_uses[0]["id"]  # non-blank (uuid back-fill, restored)
         assert tool_results[0]["tool_use_id"] == tool_uses[0]["id"]
 
+    def test_agent_blank_provider_id_keeps_synthesized_reasoning(self):
+        # The over-drop guard: a Chat-Completions server that BOTH leaves
+        # tool-call ids blank AND surfaces reasoning_content (llama.cpp,
+        # older vLLM) must still get its reasoning carried — the blank-id
+        # gate drops only the blocks a back-fill desyncs, and the
+        # synthesized reasoning_text lane has no client tool blocks at all.
+        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"
+        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()
+
+        call_count = [0]
+
+        def fake_create(**kwargs):
+            call_count[0] += 1
+            resp = MagicMock()
+            choice = MagicMock()
+            if call_count[0] == 1:
+                choice.finish_reason = "tool_calls"
+                tc = MagicMock()
+                tc.id = ""  # blank — the back-fill case
+                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 = "work it out"
+            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=1, completion_tokens=1)
+            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",
+            )
+
+        # Reasoning survives the blank-id turn.
+        assert turns[1].native is not None
+        assert [b["type"] for b in turns[1].native.blocks] == ["reasoning_text"]
+        assert turns[1].native.blocks[0]["text"] == "work it out"
+
     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

tests/test_session_synth_reasoning_block.py

@@ -361,3 +361,69 @@ class TestResolveServerType:
         session._registry = BrokenRegistry()
         session._model_alias = "x"
         assert session._resolve_server_type() == ""
+
+
+class TestFinalizeProviderBlocks:
+    """Direct unit tests for the shared native-lane builder
+    ``ChatSession._finalize_provider_blocks`` — in particular the
+    ``had_blank_ids`` gate (a uuid back-fill reaches only the tool_calls
+    mirror, so blocks that would replay the blank id must be dropped while
+    the reasoning lane survives)."""
+
+    def test_passthrough_without_blank_ids(self) -> None:
+        session = _make_session()
+        blocks = [
+            {"type": "thinking", "thinking": "x", "signature": "s"},
+            {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {}},
+        ]
+        out = session._finalize_provider_blocks(blocks, [], has_tool_calls=True)
+        assert out is blocks
+
+    def test_no_tool_calls_strips_orphan_client_blocks(self) -> None:
+        session = _make_session()
+        blocks = [
+            {"type": "thinking", "thinking": "x", "signature": "s"},
+            {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {}},
+        ]
+        out = session._finalize_provider_blocks(blocks, [], has_tool_calls=False)
+        assert [b["type"] for b in out] == ["thinking"]
+
+    def test_blank_ids_drop_messages_shaped_lane_entirely(self) -> None:
+        # Anthropic-shaped lane with a blank-id tool_use: the client block is
+        # stripped, and the surviving thinking/text blocks must go with it —
+        # on the Messages translator a native lane REPLACES the rebuilt
+        # content, so a lane missing its tool_use would orphan the mirror's
+        # calls.
+        session = _make_session()
+        blocks = [
+            {"type": "thinking", "thinking": "x", "signature": "s"},
+            {"type": "text", "text": "using f"},
+            {"type": "tool_use", "id": "", "name": "f", "input": {}},
+        ]
+        out = session._finalize_provider_blocks(blocks, [], has_tool_calls=True, had_blank_ids=True)
+        assert out == []
+
+    def test_blank_ids_keep_shape_invalid_reasoning_residuals(self) -> None:
+        # Google-shaped lane: the raw function dict (blank id) is stripped;
+        # the synthesized reasoning_text block survives — it is shape-invalid
+        # on the Messages translator by design, and the Google swap simply
+        # finds no function blocks and keeps the sanitized mirror.
+        session = _make_session()
+        blocks = [
+            {"id": "", "type": "function", "function": {"name": "f", "arguments": "{}"}},
+        ]
+        out = session._finalize_provider_blocks(
+            blocks, ["thinking text"], has_tool_calls=True, had_blank_ids=True
+        )
+        assert [b["type"] for b in out] == ["reasoning_text"]
+        assert out[0]["text"] == "thinking text"
+
+    def test_blank_ids_without_client_blocks_keep_the_lane(self) -> None:
+        # llama.cpp / older vLLM: blank tool ids AND loose reasoning text,
+        # but no client tool blocks at all — nothing can desync, so the
+        # synthesized reasoning lane must be kept (the over-drop case).
+        session = _make_session()
+        out = session._finalize_provider_blocks(
+            [], ["step by step"], has_tool_calls=True, had_blank_ids=True
+        )
+        assert [b["type"] for b in out] == ["reasoning_text"]

turnstone/core/lowering.py

@@ -232,7 +232,7 @@ def tool_args_preview(arguments: Any) -> str:
     return _ARGS_PREVIEW_CONTROL_RE.sub(" ", redact_credentials(text))[:120]
 
 
-def _legalized_arguments(arguments: Any) -> str | None:
+def legalized_arguments(arguments: Any) -> str | None:
     """A wire-valid replacement for *arguments*, or ``None`` if already valid.
 
     A raw ``dict`` (an internal shape that reached the wire seat) is serialized;
@@ -276,7 +276,7 @@ def sanitize_tool_call_arguments(messages: list[dict[str, Any]]) -> list[dict[st
             fn = tc.get("function")
             if not isinstance(fn, dict):
                 continue
-            replacement = _legalized_arguments(fn.get("arguments"))
+            replacement = legalized_arguments(fn.get("arguments"))
             if replacement is None:
                 continue  # already wire-valid — leave byte-for-byte untouched
             if repaired is None:

turnstone/core/providers/_google.py

@@ -23,11 +23,14 @@ from typing import TYPE_CHECKING, Any
 if TYPE_CHECKING:
     from collections.abc import Iterator
 
-from turnstone.core.lowering import wire_valid_arguments
+from turnstone.core.log import get_logger
+from turnstone.core.lowering import legalized_arguments, tool_args_preview
 from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 from turnstone.core.providers._openai_common import sanitize_messages
 from turnstone.core.providers._protocol import ModelCapabilities, StreamChunk
 
+log = get_logger(__name__)
+
 # Default endpoint used when no base_url is configured.
 GOOGLE_DEFAULT_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/openai/"
 
@@ -101,21 +104,37 @@ class GoogleProvider(OpenAIChatCompletionsProvider):
                 # their own round-trip handling here.
                 raw_tcs = [b for b in pc if b.get("type") == "function"]
                 if raw_tcs:
-                    # The raw dicts carry the model's ORIGINAL arguments
-                    # string; the top-level mirror this swap replaces may
-                    # have been legalized upstream
-                    # (lowering.sanitize_tool_call_arguments), so re-apply
-                    # the same validity floor here — otherwise the fidelity
-                    # swap resurrects a malformed arguments string on every
-                    # replay.  Copy-on-write per offending entry; ids and
+                    # The raw dicts carry the model's ORIGINAL arguments;
+                    # the top-level mirror this swap replaces may have been
+                    # legalized upstream (lowering.sanitize_tool_call_arguments),
+                    # so re-apply the SAME legalization (shared helper — a
+                    # dict-shaped value is serialized, anything else invalid
+                    # collapses to "{}", and the standard breadcrumb is
+                    # logged) — otherwise the fidelity swap resurrects a
+                    # malformed arguments value on every replay.  A non-dict
+                    # ``function`` passes through untouched (someone else's
+                    # malformation, exactly like the sanitize pass).
+                    # Copy-on-write per offending entry; ids and
                     # ``thought_signature`` stay untouched.
-                    raw_tcs = [
-                        b
-                        if wire_valid_arguments(b.get("function", {}).get("arguments"))
-                        else {**b, "function": {**b.get("function", {}), "arguments": "{}"}}
-                        for b in raw_tcs
-                    ]
-                    msg["tool_calls"] = raw_tcs
+                    fixed: list[dict[str, Any]] = []
+                    for b in raw_tcs:
+                        fn = b.get("function")
+                        replacement = (
+                            legalized_arguments(fn.get("arguments"))
+                            if isinstance(fn, dict)
+                            else None
+                        )
+                        if replacement is None:
+                            fixed.append(b)
+                            continue
+                        log.debug(
+                            "wire.tool_args_legalized",
+                            tool=fn.get("name", "?"),
+                            call_id=b.get("id", ""),
+                            raw_preview=tool_args_preview(fn.get("arguments")),
+                        )
+                        fixed.append({**b, "function": {**fn, "arguments": replacement}})
+                    msg["tool_calls"] = fixed
             cleaned.append(msg)
         return sanitize_messages(cleaned)
 

turnstone/core/session.py

@@ -2199,6 +2199,7 @@ class ChatSession:
         reasoning_parts: list[str],
         *,
         has_tool_calls: bool,
+        had_blank_ids: bool = False,
         alias: str | None = None,
     ) -> list[dict[str, Any]]:
         """Finalize an assistant turn's provider-native block lane: synthesize
@@ -2210,16 +2211,44 @@ class ChatSession:
         ``tool_result`` (see ``storage._utils.normalize_native_for_save``, the
         save-time chokepoint with the same gate).
 
+        *had_blank_ids* is the OTHER direction of that mirror: the caller's
+        ``_ensure_tool_call_ids`` back-fill reaches only the ``tool_calls``
+        mirror, so a client tool block in the lane still carries its blank
+        provider id verbatim and any replay of it desyncs from the mirror and
+        the results (Anthropic orphans the result and 400s; the Google swap
+        re-fills a fresh id and drops the real result).  The client tool
+        blocks are therefore stripped — and when any were present, the lane's
+        remaining Messages-shaped blocks (``thinking`` / ``text``) go with
+        them, because on the Anthropic translator a surviving native lane
+        REPLACES the rebuilt content wholesale and a lane missing its
+        ``tool_use`` would orphan every mirrored call.  Shape-invalid
+        residuals (the ``reasoning_text`` synth block, Responses ``reasoning``
+        items) are kept: they fall through that translator's per-block filter
+        by design, and their own translators pair them by ordinal, not id.
+
         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.
         """
+        from turnstone.core.providers._anthropic import ANTHROPIC_VALID_BLOCK_TYPES
+
         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)
+        if not provider_blocks:
+            return provider_blocks
+        if not has_tool_calls:
+            return strip_orphan_client_tool_blocks(provider_blocks)
+        if had_blank_ids:
+            stripped = strip_orphan_client_tool_blocks(provider_blocks)
+            if len(stripped) != len(provider_blocks):
+                stripped = [
+                    b
+                    for b in stripped
+                    if not (isinstance(b, dict) and b.get("type") in ANTHROPIC_VALID_BLOCK_TYPES)
+                ]
+            return stripped
         return provider_blocks
 
     def _resolve_replay_reasoning_to_model(
@@ -7031,7 +7060,13 @@ class ChatSession:
         content = "".join(content_parts)
         msg["content"] = content or ""
 
+        had_blank_ids = False
         if tool_calls_acc:
+            # Record blanks BEFORE the uuid back-fill (the back-fill reaches
+            # only this mirror; the native blocks keep the blank id verbatim)
+            # — threaded to _finalize_provider_blocks, which drops the blocks
+            # a back-filled id would desync.
+            had_blank_ids = any(not tc.get("id") for tc in tool_calls_acc.values())
             self._ensure_tool_call_ids(tool_calls_acc)
             ordered = [tool_calls_acc[i] for i in sorted(tool_calls_acc)]
             msg["tool_calls"] = ordered
@@ -7064,7 +7099,10 @@ class ChatSession:
         # 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"))
+            provider_blocks,
+            reasoning_parts,
+            has_tool_calls=bool(msg.get("tool_calls")),
+            had_blank_ids=had_blank_ids,
         )
         if provider_blocks:
             msg["_provider_content"] = provider_blocks
@@ -15426,8 +15464,8 @@ class ChatSession:
                 # Record blanks BEFORE the uuid back-fill: a back-filled id
                 # exists only in the tool_calls mirror — the native blocks
                 # keep the blank provider id verbatim (they are never
-                # rewritten), so a turn with a back-fill must not carry its
-                # native lane (see the gate below).
+                # rewritten), so the shared builder must drop the blocks the
+                # back-fill desyncs (see _finalize_provider_blocks).
                 had_blank_ids = any(not tc.get("id") for tc in result.tool_calls)
                 self._ensure_tool_call_ids(result.tool_calls)
                 # Mint each sub-agent tool id session-unique:
@@ -15472,22 +15510,16 @@ class ChatSession:
             # the backend that produced them (and the translators' per-block
             # shape filters drop anything foreign).
             #
-            # SKIPPED when a blank provider id was uuid-back-filled above:
-            # the back-fill reaches only the mirror, so the native tool_use
-            # block still carries the blank id and replaying it would desync
-            # from the restored tool_result (Anthropic orphans the result and
-            # 400s; Google re-fills a fresh uuid and drops the result).  The
-            # rebuild path keeps every wire representation on the back-filled
-            # id — the pre-native behaviour, for exactly the degenerate case.
-            native_blocks = (
-                []
-                if had_blank_ids
-                else self._finalize_provider_blocks(
-                    result.provider_blocks,
-                    [result.reasoning],
-                    has_tool_calls=bool(result.tool_calls),
-                    alias=agent_alias,
-                )
+            # ``had_blank_ids`` lets the shared builder drop exactly the
+            # blocks a uuid-back-fill desyncs (client tool blocks + their
+            # Messages-shaped siblings) while keeping the reasoning lane —
+            # see _finalize_provider_blocks.
+            native_blocks = self._finalize_provider_blocks(
+                result.provider_blocks,
+                [result.reasoning],
+                has_tool_calls=bool(result.tool_calls),
+                had_blank_ids=had_blank_ids,
+                alias=agent_alias,
             )
             native = (
                 ProviderNative(producer=agent_provider.provider_name, blocks=tuple(native_blocks))