fix(responses): label a structured output turn with its returned message's phase and keep lone surrogates from failing the checksum

This commit is contained in:
mateo-berri 2026-10-28 18:51:17 -07:00
parent 2ec0722c17
commit 10adc6359c
3 changed files with 47 additions and 10 deletions

View file

@ -222,7 +222,7 @@ def _checksum(message: _ChatMessage, entries: Sequence[_MessageEntry | _IndexedE
"items": [entry.model_dump(mode="json", exclude_none=True) for entry in entries],
}
encoded: Final = json.dumps(bound, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
return hashlib.sha256(encoded.encode()).hexdigest()
return hashlib.sha256(encoded.encode("utf-8", "surrogatepass")).hexdigest()
def _kind(item: _OutputItem) -> str:
@ -263,9 +263,11 @@ def _matches(items: Sequence[_OutputItem], message: _ChatMessage) -> bool:
)
def _last_phase(items: Sequence[_OutputItem | None]) -> Phase | None:
def _content_phase(items: Sequence[_OutputItem | None], *, first_message_only: bool) -> Phase | None:
messages: Final = tuple(item for item in items if isinstance(item, _OutputMessage))
return messages[-1].phase if messages else None
if not messages:
return None
return messages[0].phase if first_message_only else messages[-1].phase
def _with_phase(message: _ChatMessage, phase: Phase | None) -> _ChatMessage:
@ -275,9 +277,11 @@ def _with_phase(message: _ChatMessage, phase: Phase | None) -> _ChatMessage:
return message.model_copy(update={"provider_specific_fields": fields})
def replay_fields(output_items: Sequence[object], message: object, *, completed: bool) -> Mapping[str, object]:
def replay_fields(
output_items: Sequence[object], message: object, *, completed: bool, first_message_only: bool
) -> Mapping[str, object]:
parsed: Final = tuple(map(_parsed_output_item, output_items))
phase: Final = _last_phase(parsed)
phase: Final = _content_phase(parsed, first_message_only=first_message_only)
phase_fields: Final[Mapping[str, object]] = MappingProxyType({PHASE_FIELD: phase} if phase else {})
items: Final = tuple(item for item in parsed if item is not None)
parsed_message: Final = _chat_message(message)

View file

@ -535,10 +535,12 @@ def _fold_choices_into_one(
)
def _with_replay_fields(choice: "Choices", output_items: Sequence[object], *, completed: bool) -> "Choices":
def _with_replay_fields(
choice: "Choices", output_items: Sequence[object], *, completed: bool, structured_output: bool
) -> "Choices":
fields: Final = {
**(choice.message.provider_specific_fields or {}),
**replay_fields(output_items, choice.message, completed=completed),
**replay_fields(output_items, choice.message, completed=completed, first_message_only=structured_output),
}
message: Final = choice.message.model_copy(update={"provider_specific_fields": fields or None})
return choice.model_copy(update={"message": message})
@ -565,7 +567,10 @@ def completed_stream_history(
"reasoning_items": list(_reasoning_items_from_output_items(items)),
"provider_specific_fields": tool_search_fields,
}
fields: Final = {**(tool_search_fields or {}), **replay_fields(items, streamed_message, completed=completed)}
fields: Final = {
**(tool_search_fields or {}),
**replay_fields(items, streamed_message, completed=completed, first_message_only=structured_output),
}
return fields or None
@ -1385,7 +1390,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
tool_search_items=tool_search_items,
structured_output=structured_output,
)
return [_with_replay_fields(folded, output_items, completed=completed) if replay_openai_history else folded]
if not replay_openai_history:
return [folded]
return [_with_replay_fields(folded, output_items, completed=completed, structured_output=structured_output)]
@staticmethod
def _build_empty_choice(

View file

@ -190,10 +190,14 @@ def _assistant_turn(
model: str = f"openai/responses/{MODEL}",
terminal_event: str | None = "response.completed",
status: str = "completed",
response_format: Mapping[str, object] | None = None,
) -> dict[str, object]:
reply: Final = httpx.Response(
200, content=json.dumps(_response(output, status)), headers={"content-type": "application/json"}
)
with respx.mock(assert_all_called=True) as router:
router.post(url__startswith=OPENAI_RESPONSES_URL if model.startswith("openai/") else AZURE_API_BASE).mock(
return_value=_sse(output, terminal_event) if stream else httpx.Response(200, json=_response(output, status))
return_value=_sse(output, terminal_event) if stream else reply
)
response: Final = litellm.completion(
model=model,
@ -203,6 +207,7 @@ def _assistant_turn(
api_base=None if model.startswith("openai/") else AZURE_API_BASE,
api_version=None if model.startswith("openai/") else "2025-04-01-preview",
stream=stream,
response_format=response_format,
)
message: Final = (
(litellm.stream_chunk_builder(list(response), messages=[USER]) if stream else response).choices[0].message
@ -341,6 +346,27 @@ def test_an_unphased_answer_after_commentary_is_never_labeled_commentary(stream:
assert "phase" not in sent[1]
def _with_text(item: Mapping[str, object], text: str) -> Mapping[str, object]:
return {**item, "content": [{"type": "output_text", "text": text, "annotations": []}]}
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_a_structured_output_turn_carries_the_phase_of_the_message_it_returns(stream: bool) -> None:
output: Final = (_with_text(COMMENTARY, '{"city":"Paris"}'), _with_text(FINAL_ANSWER, '{"city":"Lyon"}'))
assistant: Final = _assistant_turn(output, stream=stream, response_format={"type": "json_object"})
assert assistant["content"] == '{"city":"Paris"}'
assert _replayed_fields(assistant)["phase"] == "commentary"
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_a_lone_surrogate_in_the_reply_still_gets_a_record(stream: bool) -> None:
assistant: Final = _assistant_turn((_with_text(COMMENTARY, "Checking \ud83d"), WEATHER_CALL), stream=stream)
assert "openai_responses_replay" in _replayed_fields(assistant)
def test_a_phase_openai_does_not_define_is_dropped_under_drop_params() -> None:
assistant: Final = _assistant_turn(OUTPUT, stream=False)