fix(responses): keep tool-call-only stream content empty and drop unreachable replay guards

This commit is contained in:
mateo-berri 2026-10-28 04:35:56 -07:00
parent c8d7cdad97
commit 71fc3e91ce
3 changed files with 58 additions and 35 deletions

View file

@ -281,14 +281,8 @@ def _stored_replay(message: _ChatMessage) -> _Replay | None:
return None
def _covers(entries: Sequence[_IndexedEntry], count: int) -> bool:
return sorted(entry.index for entry in entries) == list(range(count))
def _message_texts(entries: Sequence[_MessageEntry], content: str) -> tuple[tuple[str, ...], ...] | None:
def _message_texts(entries: Sequence[_MessageEntry], content: str) -> tuple[tuple[str, ...], ...]:
lengths: Final = tuple(chain.from_iterable(entry.parts for entry in entries))
if sum(lengths) != len(content):
return None
bounds: Final = tuple(accumulate(lengths, initial=0))
parts: Final = tuple(content[start:end] for start, end in pairwise(bounds))
groups: Final = tuple(accumulate((len(entry.parts) for entry in entries), initial=0))
@ -355,27 +349,17 @@ def _input_item(resolved: _Resolved, position: int, entry: _MessageEntry | _Inde
def _resolve(
message: _ChatMessage, entries: Sequence[_MessageEntry | _IndexedEntry], replay_tool_search: bool
) -> ReplayedTurn | None:
messages: Final = tuple(entry for entry in entries if isinstance(entry, _MessageEntry))
indexed: Final = tuple(entry for entry in entries if isinstance(entry, _IndexedEntry))
reasoning: Final = tuple(entry for entry in indexed if entry.type == "reasoning")
calls: Final = tuple(entry for entry in indexed if entry.type in ("function_call", "custom_tool_call"))
searches: Final = tuple(entry for entry in indexed if entry.type in ("tool_search_call", "tool_search_output"))
tool_searches: Final = _tool_search_entries(message)
texts: Final = _message_texts(messages, message.content or "")
if (
texts is None
or (searches and not replay_tool_search)
or not _covers(reasoning, len(message.reasoning_items or ()))
or not _covers(calls, len(message.tool_calls or ()))
or not _covers(searches, len(tool_searches))
or any(tool_searches[entry.index].get("type") != entry.type for entry in searches)
):
if not replay_tool_search and any(entry.type in ("tool_search_call", "tool_search_output") for entry in indexed):
return None
messages: Final = tuple(entry for entry in entries if isinstance(entry, _MessageEntry))
calls: Final = tuple(entry for entry in indexed if entry.type in ("function_call", "custom_tool_call"))
texts: Final = _message_texts(messages, message.content or "")
positions: Final = tuple(position for position, entry in enumerate(entries) if isinstance(entry, _MessageEntry))
resolved: Final = _Resolved(
message=message,
texts=MappingProxyType(dict(zip(positions, texts, strict=True))),
tool_searches=tool_searches,
tool_searches=_tool_search_entries(message),
)
tool_calls: Final = message.tool_calls or ()
return ReplayedTurn(

View file

@ -535,12 +535,11 @@ def _fold_choices_into_one(
def _with_replay_fields(choice: "Choices", output_items: Sequence[object], *, completed: bool) -> "Choices":
fields: Final = replay_fields(output_items, choice.message, completed=completed)
if not fields:
return choice
message: Final = choice.message.model_copy(
update={"provider_specific_fields": {**(choice.message.provider_specific_fields or {}), **fields}}
)
fields: Final = {
**(choice.message.provider_specific_fields or {}),
**replay_fields(output_items, choice.message, completed=completed),
}
message: Final = choice.message.model_copy(update={"provider_specific_fields": fields or None})
return choice.model_copy(update={"message": message})
@ -556,10 +555,7 @@ def completed_stream_history(
completed: bool,
structured_output: bool,
) -> dict[str, object] | None: # mutable-ok: Delta.provider_specific_fields is typed as a dict
try:
items: Final = _STREAMED_OUTPUT_ITEMS.validate_python(output_payload)
except ValidationError:
return dict(tool_search_fields) if tool_search_fields else None
items: Final = _STREAMED_OUTPUT_ITEMS.validate_python(output_payload)
calls: Final = tuple(item for item in items if item.get("type") in ("function_call", "custom_tool_call"))
streamed_message: Final = {
"role": "assistant",
@ -2261,7 +2257,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
StreamingChoices(
index=0,
delta=Delta(
content="",
content=None,
reasoning_items=terminal_reasoning_items_typed,
provider_specific_fields=(
completed_stream_history(

View file

@ -205,7 +205,9 @@ def _assistant_turn(
def _next_turn_input(
assistant: Mapping[str, object], model: str = f"openai/responses/{MODEL}"
assistant: Mapping[str, object],
model: str = f"openai/responses/{MODEL}",
tools: Sequence[Mapping[str, object]] = TOOLS,
) -> list[Mapping[str, object]]:
with respx.mock(assert_all_called=True) as router:
upstream: Final = router.post(
@ -214,7 +216,7 @@ def _next_turn_input(
litellm.completion(
model=model,
messages=[USER, dict(assistant), TOOL_RESULT],
tools=list(TOOLS),
tools=list(tools),
api_key="sk-test",
api_base=None if model.startswith("openai/") else AZURE_API_BASE,
api_version=None if model.startswith("openai/") else "2025-04-01-preview",
@ -319,6 +321,47 @@ def test_missing_or_corrupt_replay_metadata_falls_back_to_the_chat_fields(
assert sent[3]["phase"] == "final_answer"
def test_a_phase_openai_does_not_define_is_left_off_the_fallback() -> None:
assistant: Final = _assistant_turn(OUTPUT, stream=False)
sent: Final = _next_turn_input({**assistant, "provider_specific_fields": {"phase": "analysis"}})
assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"]
assert "phase" not in sent[3]
SEARCH_CALL: Final = {
"type": "tool_search_call",
"id": "tsc_weather",
"arguments": {"paths": ["get_weather"]},
"status": "completed",
}
SEARCH_OUTPUT: Final = {
"type": "tool_search_output",
"id": "tso_weather",
"status": "completed",
"tools": [{"name": "get_weather"}],
}
@pytest.mark.parametrize(
("tools", "expected"),
[
((*TOOLS, {"type": "tool_search"}), ["tool_search_call", "tool_search_output", *_types(REPLAYED)]),
(TOOLS, ["reasoning", "reasoning", "message", "function_call"]),
],
ids=["searching", "not-searching"],
)
def test_tool_search_items_replay_only_when_the_next_request_searches(
tools: Sequence[Mapping[str, object]], expected: Sequence[str]
) -> None:
assistant: Final = _assistant_turn((SEARCH_CALL, SEARCH_OUTPUT, *OUTPUT), stream=False)
sent: Final = _next_turn_input(assistant, tools=tools)
assert _types(sent[1:-1]) == list(expected)
WEB_SEARCH: Final = {"type": "web_search_call", "id": "ws_1", "status": "completed", "action": {"type": "search"}}