37166efcfc* Fix Gemma 4 GGUF OpenAI API streams * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Avoid duplicate Responses stream disconnect watcher * Keep reasoning-only Responses output hidden * Address Gemma stream review comments * Avoid Responses stream task-group cleanup * Harden OpenAI chat completion streams * Address OpenAI stream review issues * Clean up Studio OpenAI stream helpers * Fix Studio passthrough cold stream timeout * Fix tool parser compatibility exports lint * Preserve audio stream disconnect cancellation * Avoid synthetic finish after passthrough errors * Address stream cleanup and Gemma parser reviews * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Gemma 4: parse bare-string tool args and keep safetensors tools for native <|tool_call> - Quote bare unquoted string values in Gemma native tool-call args (e.g. {location:Tokyo,unit:celsius}) so they parse; JSON scalars stay typed. - Stop _detect_safetensors_features from suppressing supports_tools for templates that emit Gemma native <|tool_call>, which the shared parser now reads. - Add tests for both. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Harden Gemma tool-call parsing and stream-error detection Address three issues in the Gemma-native tool-call path: - _quote_gemma_object_keys stopped a bare (unquoted) string value at the first comma, so an argument like `location:New York, NY` was split mid-value and the synthesized JSON failed to parse, dropping the whole tool call. A bare value now ends only at `}` or a comma that begins the next `key:` pair. - parse_tool_calls_from_text scanned the entire response for Gemma markers even inside a tool call already parsed from a `<tool_call>{...}` JSON block, so a marker-like string inside an argument (data) was promoted to a second, unintended tool call. Matches inside an already-consumed call span are now skipped. - _openai_passthrough_stream relied on _monitor_openai_sse_line to flag a stream error, which returns early when monitor_id is None (skip_api_monitor), so an upstream error chunk left saw_stream_error unset and the synthetic-finish guard emitted a successful finish_reason after a failed stream. Error chunks are now detected independently of API monitoring. Adds tests/test_gemma_tool_parse_edge_cases.py covering the comma and marker-injection cases. * Emit the terminal finish_reason chunk in GGUF streams The OpenAI chat-completions GGUF tool stream and plain stream both built a final ChatCompletionChunk carrying finish_reason but never yielded it, so clients received the optional usage chunk and [DONE] with no chunk carrying finish_reason. OpenAI-compatible consumers rely on that terminal choice to distinguish stop/length/tool_calls. Yield it before the usage chunk and [DONE], matching the other streaming paths. * Parse tool calls in document order and skip nested markers both ways Unify the JSON- and Gemma-format tool-call passes into a single position-ordered scan: - Calls are now emitted in byte order across both formats, so a mixed output like `<|tool_call>call:create{...}<tool_call|> ... <tool_call> {"name":"read",...}</tool_call>` executes create before read, matching the order they appear in (tools run in returned order). - A candidate that starts inside an already-accepted call's span is skipped, in both directions: a JSON marker inside a Gemma argument and a Gemma marker inside a JSON argument are treated as data, not promoted to a second executable tool call. Extends tests/test_gemma_tool_parse_edge_cases.py with the ordering and JSON-in-Gemma nesting cases. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Quote bare Gemma array elements; order finish before trailing usage - _quote_gemma_object_keys skipped array values, so a Gemma call with a bare-string array argument like labels:[bug,ui] produced invalid JSON and the whole tool call was dropped. Array values are now scanned and bare string elements quoted, while numbers, quoted strings, and JSON literals are preserved. - In the OpenAI passthrough stream, a trailing usage-only chunk (stream_options.include_usage) that arrived before any finish chunk was relayed before the synthetic finish, producing usage -> finish -> [DONE]. Emit the synthetic finish before that usage chunk so the order matches the other streams (finish -> usage -> [DONE]). Extends tests/test_gemma_tool_parse_edge_cases.py with the bare-array cases. * Harden Gemma array parsing, XML-parameter guard, and stream teardown Address five review findings on the Gemma tool-call and OpenAI passthrough streaming paths: - parse_tool_calls_from_text collected JSON and Gemma markers without the _inside_open_parameter guard, so a marker embedded in an existing <function=...><parameter=...> value was promoted to a separate tool call. Candidates that start inside an open XML parameter are now skipped, matching the guard the XML-style parser already applies. - _quote_gemma_array_elements preserved array elements starting with { or [ verbatim, so an array of objects (items:[{path:a}]) or a nested array failed json.loads and the whole call was dropped. Object and nested-array elements are now normalised recursively. - _openai_passthrough_stream synthesized a finish chunk before a trailing usage-only chunk and set saw_finish_reason, which made the EOF guard skip the [DONE] sentinel. The EOF path now emits [DONE] whenever the upstream omitted it, even after a finish chunk was already synthesized. - /generate/stream drove generation through asyncio.to_thread with no disconnect watcher, so a client disconnect during a long generation went unnoticed until the next send. It now runs _await_disconnect_then_cancel against the request, matching the other local streaming endpoints. - _SameTaskStreamingResponse closed the body iterator with aclose() on a send-side disconnect, raising GeneratorExit so the generators' cancellation handlers (which finish the api_monitor entry) never ran. It now throws CancelledError, falling back to aclose() when athrow is unavailable. Extends tests/test_gemma_tool_parse_edge_cases.py with array-of-objects, nested-array, and marker-inside-XML-parameter cases. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Watch disconnects on Anthropic streams; keep timestamps in Gemma values Two follow-ups on the streaming and tool-parse paths: - _anthropic_tool_stream and _anthropic_plain_stream drove generation through asyncio.to_thread(next, gen, ...) and only polled is_disconnected() between events, so a client disconnect during prefill or a long generation/tool step held the decode slot until the next event or a failed send. Both now run the _await_disconnect_then_cancel watcher used by the other local streams, stop it in finally, and break promptly when cancel_event is set. - _GEMMA_NEXT_KEY_RE treated any comma followed by word-chars-then-colon as the next key, so a bare value such as "meet at 10:00, 11:00 tomorrow" was split into bogus keys. The next-key token must now be identifier-shaped (start with a letter or underscore), so a comma before a timestamp, ratio, or other numeric-then-colon text stays part of the value. Adds a timestamp-in-bare-value regression test. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Guard nested markers, reset on disconnect, clean unstarted streams Three follow-ups on the tool-parse and streaming paths: - parse_tool_calls_from_text only skipped markers that fell inside a span it had already parsed successfully, so when an unquoted Gemma argument contained a literal marker (code:<|tool_call>call:terminal{...}<tool_call|>) the outer object failed to normalize, its span was never recorded, and the inner marker was promoted to a standalone terminal call. Candidates nested inside any other candidate's brace span are now skipped regardless of whether the enclosing candidate parsed, so a marker in malformed outer data is never executed. - /generate/stream skipped backend.reset_generation_state() when the disconnect watcher set cancel_event between chunks: the loop broke and the finally's reset is guarded on cancel_event being unset. A subprocess backend kept decoding after the client left. The cancel-break path now resets the backend. - _SameTaskStreamingResponse threw CancelledError / called aclose() on the body iterator on a send-side disconnect, but neither runs the try/finally of a generator that never started (early disconnect on http.response.start), so the passthrough's eagerly-opened upstream httpx stream and cancel-registry entry leaked. It now tracks whether the body started and, when it did not, runs an optional unstarted_cleanup hook; the OpenAI passthrough wires it to close the upstream resp/client and exit the cancel tracker. Adds a nested-unquoted-marker regression test. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Daniel Han <danielhanchen@gmail.com>
1988 lines
75 KiB
Python
1988 lines
75 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
|
|
|
|
"""
|
|
Tests for the OpenAI /v1/responses client-side function-calling pass-through.
|
|
|
|
Covers:
|
|
- ResponsesRequest accepts Responses-shape `tools`, `tool_choice`,
|
|
`parallel_tool_calls`, and `function_call` / `function_call_output`
|
|
input items for multi-turn tool loops.
|
|
- _translate_responses_tools_to_chat(): flat Responses tool shape ->
|
|
nested Chat Completions shape, drops non-function built-in tools,
|
|
returns None for empty lists.
|
|
- _translate_responses_tool_choice_to_chat(): passes string choices
|
|
through, converts {type:function,name:X} to the nested shape.
|
|
- _normalise_responses_input(): maps function_call_output items to
|
|
role="tool" ChatMessages with tool_call_id, and function_call items to
|
|
assistant messages with tool_calls.
|
|
- _chat_tool_calls_to_responses_output(): keeps call_id, drops
|
|
non-function tool calls.
|
|
- ResponsesOutputFunctionCall / ResponsesResponse round-trip tool-call
|
|
outputs without losing fields.
|
|
|
|
No running server or GPU required.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
_backend = os.path.join(os.path.dirname(__file__), "..")
|
|
sys.path.insert(0, _backend)
|
|
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import ValidationError
|
|
|
|
from core.inference.api_monitor import ApiMonitor
|
|
from models.inference import (
|
|
ChatMessage,
|
|
ResponsesFunctionCallInputItem,
|
|
ResponsesFunctionCallOutputInputItem,
|
|
ResponsesFunctionTool,
|
|
ResponsesInputMessage,
|
|
ResponsesOutputFunctionCall,
|
|
ResponsesOutputMessage,
|
|
ResponsesOutputReasoning,
|
|
ResponsesOutputTextContent,
|
|
ResponsesOutputTextPart,
|
|
ResponsesRequest,
|
|
ResponsesResponse,
|
|
ResponsesUnknownContentPart,
|
|
ResponsesUnknownInputItem,
|
|
ResponsesUsage,
|
|
)
|
|
from routes.inference import (
|
|
_SameTaskStreamingResponse,
|
|
_build_chat_request,
|
|
_chat_tool_calls_to_responses_output,
|
|
_extract_responses_reasoning,
|
|
_normalise_responses_input,
|
|
_responses_tool_output_content,
|
|
_responses_non_streaming,
|
|
_responses_stream,
|
|
_translate_responses_tool_choice_to_chat,
|
|
_translate_responses_tools_to_chat,
|
|
)
|
|
|
|
|
|
# =====================================================================
|
|
# Request model — tools / tool_choice / parallel_tool_calls
|
|
# =====================================================================
|
|
|
|
|
|
class TestResponsesRequestTools:
|
|
def test_flat_function_tool_accepted(self):
|
|
req = ResponsesRequest(
|
|
input = "hi",
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"name": "get_weather",
|
|
"description": "Get the weather for a city.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"city": {"type": "string"}},
|
|
"required": ["city"],
|
|
},
|
|
"strict": True,
|
|
}
|
|
],
|
|
)
|
|
assert req.tools is not None
|
|
assert req.tools[0]["name"] == "get_weather"
|
|
assert req.tools[0]["type"] == "function"
|
|
assert req.tools[0]["strict"] is True
|
|
|
|
def test_tool_choice_string_values(self):
|
|
for choice in ("auto", "required", "none"):
|
|
req = ResponsesRequest(input = "hi", tool_choice = choice)
|
|
assert req.tool_choice == choice
|
|
|
|
def test_tool_choice_forcing_object(self):
|
|
req = ResponsesRequest(
|
|
input = "hi",
|
|
tool_choice = {"type": "function", "name": "get_weather"},
|
|
)
|
|
assert req.tool_choice == {"type": "function", "name": "get_weather"}
|
|
|
|
def test_parallel_tool_calls(self):
|
|
req = ResponsesRequest(input = "hi", parallel_tool_calls = True)
|
|
assert req.parallel_tool_calls is True
|
|
|
|
def test_builtin_tool_type_passes_validation(self):
|
|
"""Non-function built-in tools (web_search, file_search, mcp, ...)
|
|
must not raise at validation so SDKs that default to them don't
|
|
fail on Studio; they're filtered out during translation."""
|
|
req = ResponsesRequest(
|
|
input = "hi",
|
|
tools = [{"type": "web_search_preview"}],
|
|
)
|
|
assert req.tools == [{"type": "web_search_preview"}]
|
|
|
|
def test_function_tool_model_direct(self):
|
|
tool = ResponsesFunctionTool(
|
|
type = "function",
|
|
name = "send_email",
|
|
parameters = {"type": "object", "properties": {}},
|
|
)
|
|
assert tool.name == "send_email"
|
|
assert tool.description is None
|
|
|
|
def test_function_tool_rejects_other_type(self):
|
|
with pytest.raises(ValidationError):
|
|
ResponsesFunctionTool(type = "web_search", name = "x")
|
|
|
|
|
|
# =====================================================================
|
|
# Request model — function_call / function_call_output input items
|
|
# =====================================================================
|
|
|
|
|
|
class TestResponsesMultiTurnInput:
|
|
def test_function_call_input_item(self):
|
|
req = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Weather in Paris?"},
|
|
{
|
|
"type": "function_call",
|
|
"id": "fc_abc",
|
|
"call_id": "call_abc",
|
|
"name": "get_weather",
|
|
"arguments": '{"city": "Paris"}',
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_abc",
|
|
"output": '{"temp": 12}',
|
|
},
|
|
],
|
|
)
|
|
assert len(req.input) == 3
|
|
assert isinstance(req.input[1], ResponsesFunctionCallInputItem)
|
|
assert req.input[1].call_id == "call_abc"
|
|
assert isinstance(req.input[2], ResponsesFunctionCallOutputInputItem)
|
|
assert req.input[2].call_id == "call_abc"
|
|
assert req.input[2].output == '{"temp": 12}'
|
|
|
|
def test_function_call_output_missing_call_id_rejected(self):
|
|
with pytest.raises(ValidationError):
|
|
ResponsesFunctionCallOutputInputItem(type = "function_call_output", output = "x")
|
|
|
|
def test_function_call_output_accepts_content_array(self):
|
|
item = ResponsesFunctionCallOutputInputItem(
|
|
type = "function_call_output",
|
|
call_id = "call_1",
|
|
output = [{"type": "output_text", "text": "done"}],
|
|
)
|
|
assert isinstance(item.output, list)
|
|
|
|
|
|
# =====================================================================
|
|
# Translators — tools, tool_choice
|
|
# =====================================================================
|
|
|
|
|
|
class TestToolsTranslation:
|
|
def test_flat_to_nested(self):
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"name": "get_weather",
|
|
"description": "Returns weather.",
|
|
"parameters": {"type": "object"},
|
|
"strict": True,
|
|
}
|
|
]
|
|
out = _translate_responses_tools_to_chat(tools)
|
|
assert out == [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_weather",
|
|
"description": "Returns weather.",
|
|
"parameters": {"type": "object"},
|
|
"strict": True,
|
|
},
|
|
}
|
|
]
|
|
|
|
def test_builtin_tools_dropped(self):
|
|
out = _translate_responses_tools_to_chat(
|
|
[
|
|
{"type": "web_search_preview"},
|
|
{"type": "file_search"},
|
|
{
|
|
"type": "function",
|
|
"name": "search",
|
|
"parameters": {"type": "object"},
|
|
},
|
|
]
|
|
)
|
|
assert len(out) == 1
|
|
assert out[0]["function"]["name"] == "search"
|
|
|
|
def test_empty_returns_none(self):
|
|
assert _translate_responses_tools_to_chat(None) is None
|
|
assert _translate_responses_tools_to_chat([]) is None
|
|
|
|
def test_only_builtin_tools_returns_none(self):
|
|
assert _translate_responses_tools_to_chat([{"type": "web_search_preview"}]) is None
|
|
|
|
def test_description_optional(self):
|
|
out = _translate_responses_tools_to_chat(
|
|
[
|
|
{
|
|
"type": "function",
|
|
"name": "noop",
|
|
"parameters": {"type": "object"},
|
|
}
|
|
]
|
|
)
|
|
assert "description" not in out[0]["function"]
|
|
|
|
|
|
class TestToolChoiceTranslation:
|
|
def test_string_passthrough(self):
|
|
for v in ("auto", "required", "none"):
|
|
assert _translate_responses_tool_choice_to_chat(v) == v
|
|
|
|
def test_none_passthrough(self):
|
|
assert _translate_responses_tool_choice_to_chat(None) is None
|
|
|
|
def test_forcing_object_converted(self):
|
|
assert _translate_responses_tool_choice_to_chat(
|
|
{"type": "function", "name": "get_weather"}
|
|
) == {"type": "function", "function": {"name": "get_weather"}}
|
|
|
|
def test_already_chat_nested_shape_passes_through(self):
|
|
"""A client sending the Chat Completions nested shape isn't
|
|
double-wrapped."""
|
|
already_nested = {"type": "function", "function": {"name": "get_weather"}}
|
|
assert _translate_responses_tool_choice_to_chat(already_nested) == already_nested
|
|
|
|
def test_unknown_shape_passes_through(self):
|
|
obj = {"type": "allowed_tools", "tools": [{"type": "function", "name": "x"}]}
|
|
assert _translate_responses_tool_choice_to_chat(obj) == obj
|
|
|
|
|
|
class TestBuildChatRequest:
|
|
def test_parallel_tool_calls_false_is_preserved_for_passthrough_caps(self):
|
|
payload = ResponsesRequest(
|
|
input = "hi",
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"name": "lookup",
|
|
"parameters": {"type": "object"},
|
|
}
|
|
],
|
|
parallel_tool_calls = False,
|
|
)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
chat_req = _build_chat_request(payload, messages, stream = True)
|
|
|
|
assert chat_req.parallel_tool_calls is False
|
|
|
|
def test_chat_template_kwargs_enable_thinking_true_is_lifted(self):
|
|
payload = ResponsesRequest(
|
|
input = "hi",
|
|
chat_template_kwargs = {"enable_thinking": True},
|
|
)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
chat_req = _build_chat_request(payload, messages, stream = False)
|
|
|
|
assert chat_req.enable_thinking is True
|
|
|
|
def test_chat_template_kwargs_enable_thinking_false_is_lifted(self):
|
|
payload = ResponsesRequest(
|
|
input = "hi",
|
|
chat_template_kwargs = {"enable_thinking": False},
|
|
)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
chat_req = _build_chat_request(payload, messages, stream = False)
|
|
|
|
assert chat_req.enable_thinking is False
|
|
|
|
def test_reasoning_effort_high_enables_local_thinking(self):
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
chat_req = _build_chat_request(payload, messages, stream = False)
|
|
|
|
assert chat_req.reasoning_effort == "high"
|
|
assert chat_req.enable_thinking is True
|
|
|
|
def test_reasoning_effort_none_disables_local_thinking(self):
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "none"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
chat_req = _build_chat_request(payload, messages, stream = False)
|
|
|
|
assert chat_req.reasoning_effort == "none"
|
|
assert chat_req.enable_thinking is False
|
|
|
|
def test_explicit_enable_thinking_false_disables_reasoning_effort(self):
|
|
payload = ResponsesRequest(
|
|
input = "hi",
|
|
reasoning = {"effort": "high"},
|
|
chat_template_kwargs = {"enable_thinking": False},
|
|
)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
chat_req = _build_chat_request(payload, messages, stream = False)
|
|
|
|
assert chat_req.reasoning_effort == "none"
|
|
assert chat_req.enable_thinking is False
|
|
|
|
|
|
# =====================================================================
|
|
# _normalise_responses_input — multi-turn tool mapping
|
|
# =====================================================================
|
|
|
|
|
|
class TestNormaliseResponsesInputWithTools:
|
|
def test_function_call_output_maps_to_tool_role(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Weather?"},
|
|
{
|
|
"type": "function_call",
|
|
"call_id": "call_1",
|
|
"name": "get_weather",
|
|
"arguments": "{}",
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": '{"temp": 20}',
|
|
},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert len(msgs) == 3
|
|
assert msgs[0].role == "user"
|
|
|
|
assert msgs[1].role == "assistant"
|
|
assert msgs[1].tool_calls is not None
|
|
assert msgs[1].tool_calls[0]["id"] == "call_1"
|
|
assert msgs[1].tool_calls[0]["function"]["name"] == "get_weather"
|
|
|
|
assert msgs[2].role == "tool"
|
|
assert msgs[2].tool_call_id == "call_1"
|
|
assert msgs[2].content == '{"temp": 20}'
|
|
|
|
def test_instructions_plus_developer_message_are_merged(self):
|
|
"""Codex CLI sends `instructions` (system prompt) AND a developer
|
|
message in `input`. Strict chat templates (harmony / gpt-oss,
|
|
Qwen3, ...) raise "System message must be at the beginning" on two
|
|
separate system-role messages, so we emit exactly one merged
|
|
system message at the top.
|
|
"""
|
|
payload = ResponsesRequest(
|
|
instructions = "Base instructions.",
|
|
input = [
|
|
{"role": "developer", "content": "Developer override."},
|
|
{"role": "user", "content": "Hi"},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
system_roles = [m for m in msgs if m.role == "system"]
|
|
assert len(system_roles) == 1
|
|
assert "Base instructions." in system_roles[0].content
|
|
assert "Developer override." in system_roles[0].content
|
|
# System must be the first message for strict templates.
|
|
assert msgs[0].role == "system"
|
|
assert msgs[1].role == "user"
|
|
|
|
def test_developer_message_after_user_is_still_hoisted(self):
|
|
"""A developer message appearing after user turns must still
|
|
produce a single leading system message, not a mid-conversation
|
|
system that strict templates reject."""
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Hello"},
|
|
{"role": "assistant", "content": "Hi!"},
|
|
{"role": "developer", "content": "Updated rules."},
|
|
{"role": "user", "content": "Continue"},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "system"
|
|
assert "Updated rules." in msgs[0].content
|
|
for m in msgs[1:]:
|
|
assert m.role != "system", "no trailing system message permitted"
|
|
|
|
def test_no_system_output_when_no_system_input(self):
|
|
payload = ResponsesRequest(input = "Hi")
|
|
msgs = _normalise_responses_input(payload)
|
|
assert all(m.role != "system" for m in msgs)
|
|
|
|
def test_multiple_system_messages_in_input_are_merged(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{"role": "system", "content": "A"},
|
|
{"role": "system", "content": "B"},
|
|
{"role": "user", "content": "Hi"},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert sum(1 for m in msgs if m.role == "system") == 1
|
|
assert "A" in msgs[0].content and "B" in msgs[0].content
|
|
|
|
def test_content_array_text_output_flattens_to_tool_text(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [{"type": "input_text", "text": "ok"}],
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "tool"
|
|
assert msgs[0].content == "ok"
|
|
|
|
def test_content_array_image_output_becomes_multimodal_tool_content(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [
|
|
{"type": "input_text", "text": "see image"},
|
|
{
|
|
"type": "input_image",
|
|
"image_url": "data:image/png;base64,AAA",
|
|
"detail": "high",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "tool"
|
|
assert msgs[0].tool_call_id == "call_1"
|
|
assert msgs[0].model_dump(exclude_none = True)["content"] == [
|
|
{"type": "text", "text": "see image"},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {
|
|
"url": "data:image/png;base64,AAA",
|
|
"detail": "high",
|
|
},
|
|
},
|
|
]
|
|
|
|
chat_req = _build_chat_request(payload, msgs, stream = False)
|
|
assert chat_req.model_dump(exclude_none = True)["messages"][0]["content"] == [
|
|
{"type": "text", "text": "see image"},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {
|
|
"url": "data:image/png;base64,AAA",
|
|
"detail": "high",
|
|
},
|
|
},
|
|
]
|
|
|
|
def test_content_array_image_output_allows_original_detail(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [
|
|
{
|
|
"type": "input_image",
|
|
"image_url": "https://example.com/screenshot.png",
|
|
"detail": "original",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].model_dump(exclude_none = True)["content"] == [
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {
|
|
"url": "https://example.com/screenshot.png",
|
|
"detail": "original",
|
|
},
|
|
},
|
|
]
|
|
|
|
def test_content_array_file_id_image_output_rejected_clearly(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [
|
|
{"type": "input_text", "text": "see image"},
|
|
{"type": "input_image", "file_id": "file_abc"},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
_normalise_responses_input(payload)
|
|
assert exc.value.status_code == 400
|
|
assert "file_id" in str(exc.value.detail)
|
|
|
|
def test_content_array_file_output_rejected_clearly(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [
|
|
{"type": "input_text", "text": "see file"},
|
|
{
|
|
"type": "input_file",
|
|
"file_data": "data:application/pdf;base64,AAA",
|
|
"filename": "report.pdf",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
_normalise_responses_input(payload)
|
|
assert exc.value.status_code == 400
|
|
assert "input_file" in str(exc.value.detail)
|
|
|
|
def test_content_array_malformed_image_output_rejected_clearly(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [{"type": "input_image", "detail": "high"}],
|
|
}
|
|
],
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
_normalise_responses_input(payload)
|
|
assert exc.value.status_code == 400
|
|
assert "image_url" in str(exc.value.detail)
|
|
|
|
def test_empty_function_call_output_gets_no_output_sentinel(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": "",
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "tool"
|
|
assert msgs[0].tool_call_id == "call_1"
|
|
assert msgs[0].content == "(no output)"
|
|
ChatMessage(**msgs[0].model_dump(exclude_none = True))
|
|
|
|
def test_whitespace_function_call_output_gets_no_output_sentinel(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": " \n\t",
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].content == "(no output)"
|
|
|
|
def test_empty_content_array_output_gets_no_output_sentinel(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [],
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].content == "(no output)"
|
|
|
|
def test_image_content_array_tool_output_is_serialised(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": "iVBORw0KGgo=",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "tool"
|
|
assert json.loads(msgs[0].content)[0]["type"] == "image"
|
|
|
|
def test_image_payload_outside_output_gets_no_output_sentinel(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": "",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": "iVBORw0KGgo=",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "tool"
|
|
assert msgs[0].tool_call_id == "call_1"
|
|
assert msgs[0].content == "(no output)"
|
|
|
|
def test_tool_output_serializer_preserves_non_empty_text(self):
|
|
assert _responses_tool_output_content("done") == "done"
|
|
assert _responses_tool_output_content(" done ") == " done "
|
|
|
|
|
|
# =====================================================================
|
|
# Response mapping — tool_calls → function_call output items
|
|
# =====================================================================
|
|
|
|
|
|
class TestChatToolCallsToResponsesOutput:
|
|
def test_basic_mapping(self):
|
|
items = _chat_tool_calls_to_responses_output(
|
|
[
|
|
{
|
|
"id": "call_abc",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_weather",
|
|
"arguments": '{"city":"Paris"}',
|
|
},
|
|
}
|
|
]
|
|
)
|
|
assert len(items) == 1
|
|
assert items[0]["type"] == "function_call"
|
|
assert items[0]["call_id"] == "call_abc"
|
|
assert items[0]["name"] == "get_weather"
|
|
assert items[0]["arguments"] == '{"city":"Paris"}'
|
|
assert items[0]["status"] == "completed"
|
|
assert items[0]["id"].startswith("fc_")
|
|
|
|
def test_multiple_tool_calls_preserved(self):
|
|
items = _chat_tool_calls_to_responses_output(
|
|
[
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "a", "arguments": "{}"},
|
|
},
|
|
{
|
|
"id": "call_2",
|
|
"type": "function",
|
|
"function": {"name": "b", "arguments": "{}"},
|
|
},
|
|
]
|
|
)
|
|
assert [it["call_id"] for it in items] == ["call_1", "call_2"]
|
|
|
|
def test_non_function_tool_call_dropped(self):
|
|
items = _chat_tool_calls_to_responses_output([{"id": "x", "type": "retrieval"}])
|
|
assert items == []
|
|
|
|
def test_missing_arguments_coerced_to_empty_string(self):
|
|
items = _chat_tool_calls_to_responses_output(
|
|
[{"id": "call_1", "type": "function", "function": {"name": "x"}}]
|
|
)
|
|
assert items[0]["arguments"] == ""
|
|
|
|
|
|
# =====================================================================
|
|
# Non-streaming Responses adapter
|
|
# =====================================================================
|
|
|
|
|
|
class TestResponsesNonStreamingAdapter:
|
|
class _Request:
|
|
pass
|
|
|
|
@staticmethod
|
|
def _run_with_message(
|
|
monkeypatch,
|
|
message,
|
|
payload = None,
|
|
llama_backend = None,
|
|
):
|
|
import routes.inference as inf_mod
|
|
|
|
async def fake_chat_completions(chat_req, request):
|
|
return JSONResponse(
|
|
content = {
|
|
"model": "test-model",
|
|
"choices": [{"message": message}],
|
|
"usage": {"prompt_tokens": 2, "completion_tokens": 3},
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(inf_mod, "openai_chat_completions", fake_chat_completions)
|
|
if llama_backend is not None:
|
|
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: llama_backend)
|
|
payload = payload or ResponsesRequest(input = "hi")
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_non_streaming(
|
|
payload, messages, TestResponsesNonStreamingAdapter._Request()
|
|
)
|
|
return json.loads(response.body.decode())
|
|
|
|
return asyncio.run(run())
|
|
|
|
def test_think_block_becomes_reasoning_item_before_message(self, monkeypatch):
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"})
|
|
body = self._run_with_message(
|
|
monkeypatch,
|
|
{"content": "<think>plan</think>33"},
|
|
payload = payload,
|
|
)
|
|
|
|
assert [item["type"] for item in body["output"]] == ["reasoning", "message"]
|
|
assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "plan"}]
|
|
assert body["output"][0]["summary"] == []
|
|
assert body["output"][1]["content"][0]["text"] == "33"
|
|
assert "<think>" not in body["output"][1]["content"][0]["text"]
|
|
assert "</think>" not in body["output"][1]["content"][0]["text"]
|
|
|
|
def test_unclosed_think_block_extracts_as_reasoning(self):
|
|
reasoning, visible = _extract_responses_reasoning(
|
|
"<think>partial plan",
|
|
parse_think_markers = True,
|
|
)
|
|
|
|
assert reasoning == "partial plan"
|
|
assert visible == ""
|
|
|
|
def test_monitor_records_translated_visible_text(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
async def fake_chat_completions(chat_req, request):
|
|
assert request.state.skip_api_monitor is True
|
|
return JSONResponse(
|
|
content = {
|
|
"model": "test-model",
|
|
"choices": [{"message": {"content": "<think>plan</think>answer"}}],
|
|
"usage": {"prompt_tokens": 2, "completion_tokens": 3},
|
|
}
|
|
)
|
|
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monkeypatch.setattr(inf_mod, "openai_chat_completions", fake_chat_completions)
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
request = SimpleNamespace(
|
|
state = SimpleNamespace(),
|
|
url = SimpleNamespace(path = "/v1/responses"),
|
|
method = "POST",
|
|
)
|
|
|
|
async def run():
|
|
response = await _responses_non_streaming(payload, messages, request)
|
|
return json.loads(response.body.decode())
|
|
|
|
body = asyncio.run(run())
|
|
|
|
assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "plan"}]
|
|
assert body["output"][1]["content"][0]["text"] == "answer"
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "completed"
|
|
assert entry["reply"] == "answer"
|
|
assert entry["prompt_tokens"] == 2
|
|
assert entry["completion_tokens"] == 3
|
|
assert request.state.skip_api_monitor is False
|
|
|
|
def test_monitor_records_tool_only_reply(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
async def fake_chat_completions(chat_req, request):
|
|
assert request.state.skip_api_monitor is True
|
|
return JSONResponse(
|
|
content = {
|
|
"model": "test-model",
|
|
"choices": [
|
|
{
|
|
"message": {
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "lookup",
|
|
"arguments": '{"query":"weather"}',
|
|
},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
],
|
|
"usage": {"prompt_tokens": 2, "completion_tokens": 3},
|
|
}
|
|
)
|
|
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monkeypatch.setattr(inf_mod, "openai_chat_completions", fake_chat_completions)
|
|
payload = ResponsesRequest(
|
|
input = "hi",
|
|
tools = [{"type": "function", "name": "lookup"}],
|
|
)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
request = SimpleNamespace(
|
|
state = SimpleNamespace(),
|
|
url = SimpleNamespace(path = "/v1/responses"),
|
|
method = "POST",
|
|
)
|
|
|
|
async def run():
|
|
response = await _responses_non_streaming(payload, messages, request)
|
|
return json.loads(response.body.decode())
|
|
|
|
body = asyncio.run(run())
|
|
|
|
assert body["output"][0]["type"] == "function_call"
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "completed"
|
|
assert entry["reply"] == 'Tool call: lookup({"query":"weather"})'
|
|
assert request.state.skip_api_monitor is False
|
|
|
|
def test_cancelled_chat_completion_finalizes_monitor(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
async def fake_chat_completions(chat_req, request):
|
|
assert request.state.skip_api_monitor is True
|
|
raise asyncio.CancelledError()
|
|
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monkeypatch.setattr(inf_mod, "openai_chat_completions", fake_chat_completions)
|
|
payload = ResponsesRequest(input = "hi")
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
request = SimpleNamespace(
|
|
state = SimpleNamespace(),
|
|
url = SimpleNamespace(path = "/v1/responses"),
|
|
method = "POST",
|
|
)
|
|
|
|
async def run():
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await _responses_non_streaming(payload, messages, request)
|
|
|
|
asyncio.run(run())
|
|
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "cancelled"
|
|
assert monitor.active_count() == 0
|
|
assert request.state.skip_api_monitor is False
|
|
|
|
def test_literal_think_tags_remain_visible_without_reasoning_request(self, monkeypatch):
|
|
body = self._run_with_message(monkeypatch, {"content": "show <think>x</think> tags"})
|
|
|
|
assert [item["type"] for item in body["output"]] == ["message"]
|
|
assert body["output"][0]["content"][0]["text"] == "show <think>x</think> tags"
|
|
|
|
def test_non_reasoning_gguf_keeps_literal_think_tags_visible(self, monkeypatch):
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"})
|
|
body = self._run_with_message(
|
|
monkeypatch,
|
|
{"content": "show <think>x</think> tags"},
|
|
payload = payload,
|
|
llama_backend = SimpleNamespace(
|
|
is_loaded = True,
|
|
reasoning_always_on = False,
|
|
supports_reasoning = False,
|
|
),
|
|
)
|
|
|
|
assert [item["type"] for item in body["output"]] == ["message"]
|
|
assert body["output"][0]["content"][0]["text"] == "show <think>x</think> tags"
|
|
|
|
def test_reasoning_capable_gguf_parses_think_tags_by_default(self, monkeypatch):
|
|
body = self._run_with_message(
|
|
monkeypatch,
|
|
{"content": "<think>plan</think>answer"},
|
|
llama_backend = SimpleNamespace(
|
|
is_loaded = True,
|
|
reasoning_always_on = False,
|
|
supports_reasoning = True,
|
|
),
|
|
)
|
|
|
|
assert [item["type"] for item in body["output"]] == ["reasoning", "message"]
|
|
assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "plan"}]
|
|
assert body["output"][1]["content"][0]["text"] == "answer"
|
|
|
|
def test_reasoning_capable_gguf_sanitizes_think_tags_when_disabled(self, monkeypatch):
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "none"})
|
|
body = self._run_with_message(
|
|
monkeypatch,
|
|
{"content": "<think>leaked</think>answer"},
|
|
payload = payload,
|
|
llama_backend = SimpleNamespace(
|
|
is_loaded = True,
|
|
reasoning_always_on = False,
|
|
supports_reasoning = True,
|
|
),
|
|
)
|
|
|
|
assert [item["type"] for item in body["output"]] == ["reasoning", "message"]
|
|
assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "leaked"}]
|
|
assert body["output"][1]["content"][0]["text"] == "answer"
|
|
|
|
def test_structured_reasoning_content_extracts_text_parts(self, monkeypatch):
|
|
body = self._run_with_message(
|
|
monkeypatch,
|
|
{
|
|
"content": "33",
|
|
"reasoning_content": [
|
|
{"type": "reasoning_text", "text": "plan"},
|
|
{"type": "reasoning_text", "text": " next"},
|
|
],
|
|
},
|
|
)
|
|
|
|
assert [item["type"] for item in body["output"]] == ["reasoning", "message"]
|
|
assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "plan next"}]
|
|
assert body["output"][1]["content"][0]["text"] == "33"
|
|
|
|
def test_plain_content_remains_message_only(self, monkeypatch):
|
|
body = self._run_with_message(monkeypatch, {"content": "33"})
|
|
|
|
assert [item["type"] for item in body["output"]] == ["message"]
|
|
assert body["output"][0]["content"][0]["text"] == "33"
|
|
|
|
def test_reasoning_only_stays_out_of_visible_message_text(self, monkeypatch):
|
|
payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"})
|
|
body = self._run_with_message(
|
|
monkeypatch,
|
|
{"content": "<think>plan</think>"},
|
|
payload = payload,
|
|
)
|
|
|
|
assert [item["type"] for item in body["output"]] == ["reasoning"]
|
|
assert body["output"][0]["content"][0]["text"] == "plan"
|
|
|
|
|
|
# =====================================================================
|
|
# Streaming Responses adapter
|
|
# =====================================================================
|
|
|
|
|
|
class TestResponsesStreamAdapter:
|
|
class _Request:
|
|
async def is_disconnected(self):
|
|
return False
|
|
|
|
@staticmethod
|
|
async def _collect(response):
|
|
chunks = []
|
|
async for chunk in response.body_iterator:
|
|
chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk)
|
|
return chunks
|
|
|
|
@staticmethod
|
|
def _payloads(lines, event_name):
|
|
prefix = f"event: {event_name}\n"
|
|
return [
|
|
json.loads(line.split("data: ", 1)[1].strip())
|
|
for line in lines
|
|
if line.startswith(prefix)
|
|
]
|
|
|
|
@staticmethod
|
|
def _install_stream_mock(
|
|
monkeypatch,
|
|
chunks,
|
|
*,
|
|
supports_reasoning = True,
|
|
reasoning_always_on = False,
|
|
):
|
|
import routes.inference as inf_mod
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
content = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks)
|
|
content += "data: [DONE]\n\n"
|
|
return httpx.Response(
|
|
200,
|
|
content = content.encode(),
|
|
headers = {"content-type": "text/event-stream"},
|
|
)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
real_async_client = httpx.AsyncClient
|
|
|
|
def _client(*args, **kwargs):
|
|
return real_async_client(
|
|
transport = transport,
|
|
timeout = kwargs.get("timeout", 600),
|
|
)
|
|
|
|
monkeypatch.setattr(inf_mod.httpx, "AsyncClient", _client)
|
|
monkeypatch.setattr(
|
|
inf_mod,
|
|
"get_llama_cpp_backend",
|
|
lambda: SimpleNamespace(
|
|
is_loaded = True,
|
|
is_vision = False,
|
|
context_length = 4096,
|
|
base_url = "http://llama.test",
|
|
supports_reasoning = supports_reasoning,
|
|
reasoning_always_on = reasoning_always_on,
|
|
_request_reasoning_kwargs = (
|
|
lambda enable_thinking = None, reasoning_effort = None, preserve_thinking = None: None
|
|
),
|
|
),
|
|
)
|
|
|
|
def test_stream_response_avoids_legacy_receive_watcher(self, monkeypatch):
|
|
self._install_stream_mock(
|
|
monkeypatch,
|
|
[{"choices": [{"delta": {"content": "33"}}]}],
|
|
)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
assert isinstance(response, _SameTaskStreamingResponse)
|
|
|
|
sent = []
|
|
|
|
async def receive():
|
|
raise AssertionError("Responses streams poll disconnects in the generator")
|
|
|
|
async def send(message):
|
|
sent.append(message)
|
|
|
|
await response({"type": "http", "asgi": {"spec_version": "2.3"}}, receive, send)
|
|
return sent
|
|
|
|
sent = asyncio.run(run())
|
|
|
|
assert sent[0]["type"] == "http.response.start"
|
|
body = b"".join(message.get("body", b"") for message in sent).decode()
|
|
assert "response.output_text.delta" in body
|
|
assert '"delta":"33"' in body.replace(" ", "")
|
|
|
|
def test_split_think_markers_stream_as_reasoning_and_visible_text(self, monkeypatch):
|
|
chunks = [
|
|
{"choices": [{"delta": {"content": "<thi"}}]},
|
|
{"choices": [{"delta": {"content": "nk>pla"}}]},
|
|
{"choices": [{"delta": {"content": "n</th"}}]},
|
|
{"choices": [{"delta": {"content": "ink>33"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert "".join(event["delta"] for event in reasoning_deltas) == "plan"
|
|
assert "".join(event["delta"] for event in text_deltas) == "33"
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert [item["type"] for item in completed["response"]["output"]] == [
|
|
"reasoning",
|
|
"message",
|
|
]
|
|
assert completed["response"]["output"][0]["content"][0]["text"] == "plan"
|
|
assert completed["response"]["output"][1]["content"][0]["text"] == "33"
|
|
|
|
def test_usage_only_chunk_updates_monitor(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
chunks = [
|
|
{"choices": [{"delta": {"content": "33"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monitor_id = monitor.start(
|
|
endpoint = "/v1/responses",
|
|
method = "POST",
|
|
model = "m",
|
|
prompt = "hi",
|
|
)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(
|
|
payload,
|
|
messages,
|
|
self._Request(),
|
|
monitor_id = monitor_id,
|
|
)
|
|
return await self._collect(response)
|
|
|
|
asyncio.run(run())
|
|
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "completed"
|
|
assert entry["reply"] == "33"
|
|
assert entry["prompt_tokens"] == 2
|
|
assert entry["completion_tokens"] == 3
|
|
assert entry["total_tokens"] == 5
|
|
assert entry["context_length"] == 4096
|
|
|
|
def test_function_call_chunk_updates_monitor_reply(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
chunks = [
|
|
{
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_1",
|
|
"function": {
|
|
"name": "lookup",
|
|
"arguments": '{"query":"weather"}',
|
|
},
|
|
}
|
|
]
|
|
}
|
|
}
|
|
]
|
|
}
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monitor_id = monitor.start(
|
|
endpoint = "/v1/responses",
|
|
method = "POST",
|
|
model = "m",
|
|
prompt = "hi",
|
|
)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(
|
|
payload,
|
|
messages,
|
|
self._Request(),
|
|
monitor_id = monitor_id,
|
|
)
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
assert self._payloads(lines, "response.output_item.done")[-1]["item"]["name"] == "lookup"
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "completed"
|
|
assert entry["reply"] == 'Tool call: lookup({"query":"weather"})'
|
|
|
|
def test_preheader_cancel_finalizes_monitor(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
self._install_stream_mock(monkeypatch, [])
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monitor_id = monitor.start(
|
|
endpoint = "/v1/responses",
|
|
method = "POST",
|
|
model = "m",
|
|
prompt = "hi",
|
|
)
|
|
|
|
async def fake_send(*_args, **_kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(
|
|
payload,
|
|
messages,
|
|
self._Request(),
|
|
monitor_id = monitor_id,
|
|
)
|
|
return await self._collect(response)
|
|
|
|
asyncio.run(run())
|
|
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "cancelled"
|
|
assert monitor.active_count() == 0
|
|
|
|
def test_stream_task_cancel_finalizes_monitor(self, monkeypatch):
|
|
async def _run():
|
|
import routes.inference as inf_mod
|
|
|
|
async def fake_send(*_args, **_kwargs):
|
|
return httpx.Response(200, content = b"")
|
|
|
|
async def fake_items(*_args, **_kwargs):
|
|
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}'
|
|
await asyncio.sleep(3600)
|
|
|
|
self._install_stream_mock(monkeypatch, [])
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
|
|
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
|
|
monitor_id = monitor.start(
|
|
endpoint = "/v1/responses",
|
|
method = "POST",
|
|
model = "m",
|
|
prompt = "hi",
|
|
)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
response = await _responses_stream(
|
|
payload,
|
|
messages,
|
|
self._Request(),
|
|
monitor_id = monitor_id,
|
|
)
|
|
iterator = response.body_iterator
|
|
first = ""
|
|
for _ in range(8):
|
|
first = await anext(iterator)
|
|
if "hello" in first:
|
|
break
|
|
else:
|
|
pytest.fail("stream did not emit text delta")
|
|
|
|
pending = asyncio.create_task(anext(iterator))
|
|
await asyncio.sleep(0)
|
|
pending.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await pending
|
|
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "cancelled"
|
|
assert entry["reply"] == "hello"
|
|
assert monitor.active_count() == 0
|
|
|
|
asyncio.run(_run())
|
|
|
|
def test_final_visible_text_updates_monitor(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
class FakeExtractor:
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
def feed(
|
|
self,
|
|
_content,
|
|
_reasoning_content = None,
|
|
):
|
|
return "", ""
|
|
|
|
def finish(self):
|
|
return "", "tail"
|
|
|
|
self._install_stream_mock(monkeypatch, [{"choices": [{"delta": {"content": "<tai"}}]}])
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monkeypatch.setattr(inf_mod, "_ResponsesReasoningExtractor", FakeExtractor)
|
|
monitor_id = monitor.start(
|
|
endpoint = "/v1/responses",
|
|
method = "POST",
|
|
model = "m",
|
|
prompt = "hi",
|
|
)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(
|
|
payload,
|
|
messages,
|
|
self._Request(),
|
|
monitor_id = monitor_id,
|
|
)
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
assert self._payloads(lines, "response.output_text.delta")[-1]["delta"] == "tail"
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "completed"
|
|
assert entry["reply"] == "tail"
|
|
|
|
def test_reasoning_only_stream_does_not_update_visible_monitor_reply(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
class FakeExtractor:
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
def feed(
|
|
self,
|
|
_content,
|
|
_reasoning_content = None,
|
|
):
|
|
return "", ""
|
|
|
|
def finish(self):
|
|
return "plan", ""
|
|
|
|
self._install_stream_mock(monkeypatch, [{"choices": [{"delta": {"content": "<think>"}}]}])
|
|
monitor = ApiMonitor(max_entries = 3)
|
|
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
|
|
monkeypatch.setattr(inf_mod, "_ResponsesReasoningExtractor", FakeExtractor)
|
|
monitor_id = monitor.start(
|
|
endpoint = "/v1/responses",
|
|
method = "POST",
|
|
model = "m",
|
|
prompt = "hi",
|
|
)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(
|
|
payload,
|
|
messages,
|
|
self._Request(),
|
|
monitor_id = monitor_id,
|
|
)
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
assert self._payloads(lines, "response.output_text.delta") == []
|
|
assert self._payloads(lines, "response.reasoning_text.delta")[-1]["delta"] == "plan"
|
|
[entry] = monitor.snapshot()
|
|
assert entry["status"] == "completed"
|
|
assert entry["reply"] == ""
|
|
|
|
def test_reasoning_capable_gguf_stream_parses_think_tags_by_default(self, monkeypatch):
|
|
chunks = [
|
|
{"choices": [{"delta": {"content": "<thi"}}]},
|
|
{"choices": [{"delta": {"content": "nk>plan</think>answer"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert "".join(event["delta"] for event in reasoning_deltas) == "plan"
|
|
assert "".join(event["delta"] for event in text_deltas) == "answer"
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert [item["type"] for item in completed["response"]["output"]] == [
|
|
"reasoning",
|
|
"message",
|
|
]
|
|
assert completed["response"]["output"][0]["content"][0]["text"] == "plan"
|
|
assert completed["response"]["output"][1]["content"][0]["text"] == "answer"
|
|
|
|
def test_non_reasoning_gguf_stream_keeps_literal_think_tags_visible(self, monkeypatch):
|
|
chunks = [
|
|
{"choices": [{"delta": {"content": "show <thi"}}]},
|
|
{"choices": [{"delta": {"content": "nk>x</think> tags"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks, supports_reasoning = False)
|
|
payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert reasoning_deltas == []
|
|
assert "".join(event["delta"] for event in text_deltas) == "show <think>x</think> tags"
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert [item["type"] for item in completed["response"]["output"]] == ["message"]
|
|
assert completed["response"]["output"][0]["content"][0]["text"] == (
|
|
"show <think>x</think> tags"
|
|
)
|
|
|
|
def test_reasoning_only_stream_stays_out_of_visible_message_text(self, monkeypatch):
|
|
chunks = [
|
|
{"choices": [{"delta": {"content": "<think>plan</think>"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert "".join(event["delta"] for event in reasoning_deltas) == "plan"
|
|
assert text_deltas == []
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert [item["type"] for item in completed["response"]["output"]] == ["reasoning"]
|
|
assert completed["response"]["output"][0]["content"][0]["text"] == "plan"
|
|
|
|
def test_unclosed_think_stream_stays_out_of_visible_message_text(self, monkeypatch):
|
|
chunks = [
|
|
{"choices": [{"delta": {"content": "<thi"}}]},
|
|
{"choices": [{"delta": {"content": "nk>plan"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"})
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert "".join(event["delta"] for event in reasoning_deltas) == "plan"
|
|
assert text_deltas == []
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert [item["type"] for item in completed["response"]["output"]] == ["reasoning"]
|
|
assert completed["response"]["output"][0]["content"][0]["text"] == "plan"
|
|
|
|
def test_structured_reasoning_content_streams_as_reasoning(self, monkeypatch):
|
|
chunks = [
|
|
{"choices": [{"delta": {"reasoning_content": "plan"}}]},
|
|
{"choices": [{"delta": {"content": "33"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert "".join(event["delta"] for event in reasoning_deltas) == "plan"
|
|
assert "".join(event["delta"] for event in text_deltas) == "33"
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert completed["response"]["output"][0]["type"] == "reasoning"
|
|
assert completed["response"]["output"][1]["type"] == "message"
|
|
|
|
def test_structured_reasoning_content_parts_stream_as_reasoning(self, monkeypatch):
|
|
chunks = [
|
|
{
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"reasoning_content": {
|
|
"content": [
|
|
{"type": "reasoning_text", "text": "plan"},
|
|
{"type": "reasoning_text", "text": " next"},
|
|
]
|
|
}
|
|
}
|
|
}
|
|
]
|
|
},
|
|
{"choices": [{"delta": {"content": "33"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta")
|
|
text_deltas = self._payloads(lines, "response.output_text.delta")
|
|
assert "".join(event["delta"] for event in reasoning_deltas) == "plan next"
|
|
assert "".join(event["delta"] for event in text_deltas) == "33"
|
|
assert "reasoning_text" not in "".join(event["delta"] for event in reasoning_deltas)
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert completed["response"]["output"][0]["content"][0]["text"] == "plan next"
|
|
assert completed["response"]["output"][1]["content"][0]["text"] == "33"
|
|
|
|
def test_tool_first_stream_closes_items_in_output_index_order(self, monkeypatch):
|
|
chunks = [
|
|
{
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_0",
|
|
"type": "function",
|
|
"function": {"name": "lookup", "arguments": "{}"},
|
|
}
|
|
]
|
|
}
|
|
}
|
|
]
|
|
},
|
|
{"choices": [{"delta": {"content": "done"}}]},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
self._install_stream_mock(monkeypatch, chunks)
|
|
payload = ResponsesRequest(input = "hi", stream = True)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
done_events = self._payloads(lines, "response.output_item.done")
|
|
assert [event["output_index"] for event in done_events] == [0, 1]
|
|
assert [event["item"]["type"] for event in done_events] == ["function_call", "message"]
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert [item["type"] for item in completed["response"]["output"]] == [
|
|
"function_call",
|
|
"message",
|
|
]
|
|
|
|
def test_requests_usage_and_caps_parallel_tool_calls(self, monkeypatch):
|
|
import routes.inference as inf_mod
|
|
|
|
captured = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured["body"] = json.loads(request.content.decode())
|
|
chunks = [
|
|
{
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_0",
|
|
"type": "function",
|
|
"function": {"name": "first", "arguments": "{}"},
|
|
},
|
|
{
|
|
"index": 1,
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "second", "arguments": "{}"},
|
|
},
|
|
]
|
|
}
|
|
}
|
|
]
|
|
},
|
|
{"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}},
|
|
]
|
|
content = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks)
|
|
content += "data: [DONE]\n\n"
|
|
return httpx.Response(
|
|
200,
|
|
content = content.encode(),
|
|
headers = {"content-type": "text/event-stream"},
|
|
)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
real_async_client = httpx.AsyncClient
|
|
|
|
def _client(*args, **kwargs):
|
|
return real_async_client(
|
|
transport = transport,
|
|
timeout = kwargs.get("timeout", 600),
|
|
)
|
|
|
|
monkeypatch.setattr(inf_mod.httpx, "AsyncClient", _client)
|
|
monkeypatch.setattr(
|
|
inf_mod,
|
|
"get_llama_cpp_backend",
|
|
lambda: SimpleNamespace(
|
|
is_loaded = True,
|
|
is_vision = False,
|
|
context_length = 4096,
|
|
base_url = "http://llama.test",
|
|
# Non-reasoning template: the real backend returns None here.
|
|
_request_reasoning_kwargs = (
|
|
lambda enable_thinking = None, reasoning_effort = None, preserve_thinking = None: None
|
|
),
|
|
),
|
|
)
|
|
|
|
payload = ResponsesRequest(
|
|
input = "hi",
|
|
stream = True,
|
|
parallel_tool_calls = False,
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"name": "first",
|
|
"parameters": {"type": "object"},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"name": "second",
|
|
"parameters": {"type": "object"},
|
|
},
|
|
],
|
|
)
|
|
messages = [ChatMessage(role = "user", content = "hi")]
|
|
|
|
async def run():
|
|
response = await _responses_stream(payload, messages, self._Request())
|
|
return await self._collect(response)
|
|
|
|
lines = asyncio.run(run())
|
|
|
|
assert captured["body"]["stream_options"] == {"include_usage": True}
|
|
joined = "".join(lines)
|
|
assert "call_0" in joined
|
|
assert "call_1" not in joined
|
|
completed = self._payloads(lines, "response.completed")[0]
|
|
assert completed["response"]["usage"] == {
|
|
"input_tokens": 2,
|
|
"output_tokens": 3,
|
|
"total_tokens": 5,
|
|
}
|
|
|
|
|
|
# =====================================================================
|
|
# Response model — ResponsesOutputFunctionCall / mixed output
|
|
# =====================================================================
|
|
|
|
|
|
class TestResponsesOutputFunctionCall:
|
|
def test_reasoning_output_item_serialises_full_reasoning_content(self):
|
|
item = ResponsesOutputReasoning(content = [{"type": "reasoning_text", "text": "plan"}])
|
|
d = item.model_dump()
|
|
assert d["type"] == "reasoning"
|
|
assert d["id"].startswith("rs_")
|
|
assert d["status"] == "completed"
|
|
assert d["summary"] == []
|
|
assert d["content"] == [{"type": "reasoning_text", "text": "plan"}]
|
|
|
|
def test_direct_construction(self):
|
|
fc = ResponsesOutputFunctionCall(
|
|
call_id = "call_1",
|
|
name = "get_weather",
|
|
arguments = '{"city":"Paris"}',
|
|
)
|
|
d = fc.model_dump()
|
|
assert d["type"] == "function_call"
|
|
assert d["call_id"] == "call_1"
|
|
assert d["status"] == "completed"
|
|
assert d["id"].startswith("fc_")
|
|
|
|
def test_response_with_tool_call_output(self):
|
|
resp = ResponsesResponse(
|
|
model = "test",
|
|
output = [
|
|
ResponsesOutputFunctionCall(
|
|
call_id = "call_1",
|
|
name = "get_weather",
|
|
arguments = "{}",
|
|
)
|
|
],
|
|
usage = ResponsesUsage(input_tokens = 1, output_tokens = 1, total_tokens = 2),
|
|
)
|
|
d = json.loads(resp.model_dump_json())
|
|
assert d["output"][0]["type"] == "function_call"
|
|
assert d["output"][0]["call_id"] == "call_1"
|
|
|
|
def test_response_with_mixed_output(self):
|
|
resp = ResponsesResponse(
|
|
model = "test",
|
|
output = [
|
|
ResponsesOutputMessage(
|
|
content = [ResponsesOutputTextContent(text = "Calling...")],
|
|
),
|
|
ResponsesOutputFunctionCall(
|
|
call_id = "call_1",
|
|
name = "get_weather",
|
|
arguments = '{"city":"Paris"}',
|
|
),
|
|
],
|
|
)
|
|
d = resp.model_dump()
|
|
assert d["output"][0]["type"] == "message"
|
|
assert d["output"][1]["type"] == "function_call"
|
|
|
|
|
|
# =====================================================================
|
|
# Regression: ChatMessage validator still accepts mapped tool messages
|
|
# =====================================================================
|
|
|
|
|
|
class TestCodexStyleRequestShapes:
|
|
"""Regression tests for the request shapes OpenAI Codex CLI sends."""
|
|
|
|
def test_assistant_replay_output_text_accepted(self):
|
|
"""Codex replays prior assistant turns with `output_text` content;
|
|
this used to 422 on every turn after the first."""
|
|
req = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Hi"},
|
|
{
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": "Hello!",
|
|
"annotations": [],
|
|
"logprobs": [],
|
|
}
|
|
],
|
|
},
|
|
{"role": "user", "content": "Continue"},
|
|
],
|
|
)
|
|
assert len(req.input) == 3
|
|
parts = req.input[1].content
|
|
assert isinstance(parts, list)
|
|
assert isinstance(parts[0], ResponsesOutputTextPart)
|
|
assert parts[0].text == "Hello!"
|
|
|
|
def test_reasoning_item_accepted_as_unknown(self):
|
|
"""`reasoning` items replayed from prior o-series turns must not
|
|
fail validation — Codex keeps them in multi-turn."""
|
|
req = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Hi"},
|
|
{
|
|
"type": "reasoning",
|
|
"id": "rs_1",
|
|
"summary": [],
|
|
"encrypted_content": "opaque",
|
|
},
|
|
{"role": "assistant", "content": "Hello!"},
|
|
],
|
|
)
|
|
assert len(req.input) == 3
|
|
assert isinstance(req.input[1], ResponsesUnknownInputItem)
|
|
|
|
def test_emitted_reasoning_item_replay_is_dropped_for_local_chat(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Hi"},
|
|
{
|
|
"type": "reasoning",
|
|
"id": "rs_1",
|
|
"summary": [],
|
|
"content": [{"type": "reasoning_text", "text": "plan"}],
|
|
},
|
|
{"role": "assistant", "content": "33"},
|
|
{"role": "user", "content": "Continue"},
|
|
],
|
|
)
|
|
|
|
msgs = _normalise_responses_input(payload)
|
|
|
|
assert [m.role for m in msgs] == ["user", "assistant", "user"]
|
|
assert all("plan" not in (m.content or "") for m in msgs if isinstance(m.content, str))
|
|
|
|
def test_unknown_content_part_type_accepted(self):
|
|
"""Unknown content-part types (e.g. future input_audio) validate as
|
|
ResponsesUnknownContentPart so the request doesn't 422."""
|
|
req = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "input_text", "text": "See:"},
|
|
{"type": "input_audio", "audio": {"data": "..."}},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
parts = req.input[0].content
|
|
assert isinstance(parts[1], ResponsesUnknownContentPart)
|
|
assert parts[1].type == "input_audio"
|
|
|
|
def test_codex_full_shape_roundtrip(self):
|
|
"""End-to-end: developer + user + assistant(output_text) +
|
|
function_call + function_call_output + reasoning in one request."""
|
|
payload = ResponsesRequest(
|
|
instructions = "Base instructions.",
|
|
input = [
|
|
{
|
|
"type": "message",
|
|
"role": "developer",
|
|
"content": [{"type": "input_text", "text": "Dev override."}],
|
|
},
|
|
{
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": "Weather?"}],
|
|
},
|
|
{
|
|
"type": "reasoning",
|
|
"id": "rs_1",
|
|
"summary": [],
|
|
},
|
|
{
|
|
"type": "function_call",
|
|
"call_id": "call_1",
|
|
"name": "get_weather",
|
|
"arguments": "{}",
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": '{"temp":20}',
|
|
},
|
|
{
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": "It's 20°C.",
|
|
"annotations": [],
|
|
"logprobs": [],
|
|
}
|
|
],
|
|
},
|
|
{"role": "user", "content": "And tomorrow?"},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
# One leading merged system; no mid-conversation system.
|
|
assert msgs[0].role == "system"
|
|
assert sum(1 for m in msgs if m.role == "system") == 1
|
|
assert "Base instructions." in msgs[0].content
|
|
assert "Dev override." in msgs[0].content
|
|
|
|
roles = [m.role for m in msgs[1:]]
|
|
# Reasoning dropped. Order: user, assistant(tool_calls), tool,
|
|
# assistant(text), user.
|
|
assert roles == ["user", "assistant", "tool", "assistant", "user"]
|
|
assert msgs[2].tool_calls is not None
|
|
assert msgs[3].role == "tool"
|
|
assert msgs[3].tool_call_id == "call_1"
|
|
assert msgs[4].content == "It's 20°C."
|
|
|
|
def test_single_output_text_part_flattens_to_string(self):
|
|
"""ChatMessage assistant role prefers plain string content — we
|
|
don't forward a single-part array that would force legacy chat
|
|
templates into multimodal handling."""
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"role": "assistant",
|
|
"content": [{"type": "output_text", "text": "ok", "annotations": []}],
|
|
},
|
|
{"role": "user", "content": "next"},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
assert msgs[0].role == "assistant"
|
|
assert msgs[0].content == "ok"
|
|
|
|
|
|
class TestTranslatedMessagesValidate:
|
|
"""Messages from _normalise_responses_input satisfy ChatMessage's
|
|
role-shape validator so the downstream /v1/chat/completions
|
|
pass-through doesn't reject them."""
|
|
|
|
def test_round_trip_multi_turn(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{"role": "user", "content": "Weather in Paris?"},
|
|
{
|
|
"type": "function_call",
|
|
"call_id": "call_1",
|
|
"name": "get_weather",
|
|
"arguments": '{"city": "Paris"}',
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": '{"temp": 20}',
|
|
},
|
|
{"role": "user", "content": "Thanks!"},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
for m in msgs:
|
|
# Building a fresh ChatMessage from the dump round-trips the
|
|
# role-shape validator — the passthrough's key invariant.
|
|
ChatMessage(**m.model_dump(exclude_none = True))
|
|
|
|
def test_empty_tool_output_round_trips_through_chat_message_validator(self):
|
|
payload = ResponsesRequest(
|
|
input = [
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_empty",
|
|
"output": "",
|
|
},
|
|
],
|
|
)
|
|
msgs = _normalise_responses_input(payload)
|
|
for m in msgs:
|
|
ChatMessage(**m.model_dump(exclude_none = True))
|