mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
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.
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
+81
-4
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
+53
-21
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user