ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
unsloth/studio/backend/tests/test_openai_tool_passthrough.py
Daniel Han a35fbe22ea
 Studio: serialize non-streaming responses once and pool the proxy client (#6393) 
* Studio: serialize non-streaming responses once and pool the proxy client

Two safe latency wins on the OpenAI/Anthropic-compatible endpoints that leave
the streaming generation paths untouched (they keep Connection: close and
max_keepalive_connections=0 so a client disconnect still stops GPU decode).

1. Non-streaming responses used JSONResponse(content=model.model_dump()), which
   builds a dict and then re-runs json.dumps. Serialize once with
   model.model_dump_json() via a small _model_json_response helper. The body is
   byte-identical (nulls preserved), about 3x faster to encode in a microbench.

2. The non-streaming completions and embeddings proxies built a fresh
   httpx.AsyncClient per request. Route them through one pooled client
   (core/inference/llama_http) closed on shutdown; streaming generation keeps
   its own per-request close-only client. About 5x faster per call to the
   local llama-server in a microbench.

The existing API-monitor tests for the non-streaming completions, embeddings
and passthrough paths now patch nonstreaming_client instead of httpx.AsyncClient
to match the pooled client, so they stay deterministic.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: make the pooled non-streaming client per event loop

Review follow-up on the shared httpx client. It was a single module-global
instance, which has two lifecycle problems the per-request client did not:

1. After aclose() in lifespan shutdown, nonstreaming_client() kept handing back
   the closed client, so a second lifespan in the same process (repeated
   TestClient, embedded restart) failed with "client has been closed".

2. An httpx client binds its transport to the loop it first runs on, so reuse
   from another loop could raise "Event loop is closed".

Hold one client per running loop in a WeakKeyDictionary, recreate when missing
or closed, and close all on shutdown. Single-loop production is unchanged.

* [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>
2026-06-17 22:38:02 -07:00

2823 lines
102 KiB
Python
Raw Permalink Blame History

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
"""Tests for the OpenAI /v1/chat/completions client-side tool pass-through."""
import os
import sys
import asyncio
import json
import threading
from types import SimpleNamespace
_backend = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, _backend)
import httpx
import pytest
from fastapi import HTTPException
from pydantic import ValidationError
from models.inference import (
ChatCompletionRequest,
ChatMessage,
CompletionChoice,
CompletionMessage,
ResponsesRequest,
)
from core.inference.anthropic_compat import (
anthropic_tool_choice_to_openai,
)
from core.inference.api_monitor import ApiMonitor
from routes.inference import (
_build_chat_request,
_build_openai_passthrough_body,
_build_passthrough_payload,
_clamp_finish_reason,
_cmpl_stream_event_out,
_coalesce_consecutive_user_turns,
_drop_empty_assistant_sentinels,
_effective_max_tokens,
_extract_content_parts,
_friendly_error,
_merge_user_content,
_monitor_openai_chunk,
_monitor_openai_sse_event,
_openai_messages_for_gguf_chat,
_openai_passthrough_non_streaming,
_openai_passthrough_stream,
_openai_stream_usage_chunk,
_proxy_to_external_provider,
_set_or_prepend_system_message,
openai_completions,
openai_embeddings,
openai_chat_completions,
)
from state.tool_policy import reset_tool_policy
# =====================================================================
# ChatMessage — tool role, tool_calls, optional content
# =====================================================================
class TestChatMessageToolRoles:
def test_tool_role_with_tool_call_id(self):
msg = ChatMessage(
role = "tool",
tool_call_id = "call_abc123",
content = '{"temperature": 72}',
)
assert msg.role == "tool"
assert msg.tool_call_id == "call_abc123"
assert msg.content == '{"temperature": 72}'
def test_tool_role_with_name(self):
msg = ChatMessage(
role = "tool",
tool_call_id = "call_abc123",
name = "get_weather",
content = '{"temperature": 72}',
)
assert msg.name == "get_weather"
def test_assistant_with_tool_calls_no_content(self):
msg = ChatMessage(
role = "assistant",
content = None,
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
)
assert msg.role == "assistant"
assert msg.content is None
assert msg.tool_calls is not None
assert len(msg.tool_calls) == 1
assert msg.tool_calls[0]["function"]["name"] == "get_weather"
def test_assistant_with_content_and_tool_calls(self):
msg = ChatMessage(
role = "assistant",
content = "Let me check the weather.",
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{}"},
}
],
)
assert msg.content == "Let me check the weather."
assert msg.tool_calls[0]["id"] == "call_1"
def test_plain_user_message_still_works(self):
msg = ChatMessage(role = "user", content = "Hello")
assert msg.role == "user"
assert msg.tool_call_id is None
assert msg.tool_calls is None
assert msg.name is None
def test_invalid_role_rejected(self):
with pytest.raises(ValidationError):
ChatMessage(role = "function", content = "x")
def test_content_absent_on_assistant_tool_call_defaults_to_none(self):
# Assistant messages carrying only tool_calls are the one documented
# case where `content=None` is permitted.
msg = ChatMessage(
role = "assistant",
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {"name": "f", "arguments": "{}"},
}
],
)
assert msg.content is None
def test_tool_role_missing_tool_call_id_left_for_request_validator(self):
# Per-message: missing tool_call_id is now allowed at this layer.
# ChatCompletionRequest's walkback fills it from the prior assistant
# tool_calls; see test_inference_model_validation.py for resolution
# coverage.
msg = ChatMessage(role = "tool", content = '{"temperature": 72}')
assert msg.tool_call_id is None
assert msg.content == '{"temperature": 72}'
def test_tool_role_empty_tool_call_id_left_for_request_validator(self):
msg = ChatMessage(
role = "tool",
tool_call_id = "",
content = '{"temperature": 72}',
)
# Empty-string is treated the same as missing by the walkback.
assert msg.tool_call_id in (None, "")
# ── Role-aware content requirements ────────────────────────────
@pytest.mark.parametrize("role", ["user", "system"])
def test_empty_string_content_allowed(self, role):
msg = ChatMessage(role = role, content = "")
assert msg.content == ""
def test_user_missing_content_rejected(self):
with pytest.raises(ValidationError):
ChatMessage(role = "user")
def test_user_empty_list_content_rejected(self):
with pytest.raises(ValidationError):
ChatMessage(role = "user", content = [])
def test_tool_empty_content_accepted(self):
# Empty tool output (mkdir, git add, ...) is routine in agentic loops;
# OpenAI and llama-server both accept it, so Studio must not 400.
msg = ChatMessage(role = "tool", tool_call_id = "call_1", content = "")
assert msg.content == ""
def test_assistant_without_content_or_tool_calls_tolerated(self):
# Stop-button leaves an empty assistant turn; tolerate for replay.
msg = ChatMessage(role = "assistant")
assert msg.content is None
assert msg.tool_calls is None
def test_assistant_empty_string_content_normalised_to_none(self):
msg = ChatMessage(role = "assistant", content = "")
assert msg.content is None
def test_assistant_empty_list_content_normalised_to_none(self):
msg = ChatMessage(role = "assistant", content = [])
assert msg.content is None
# ── Role-constrained tool-call metadata ────────────────────────
def test_tool_calls_on_user_rejected(self):
with pytest.raises(ValidationError) as exc_info:
ChatMessage(
role = "user",
content = "Hi",
tool_calls = [
{
"id": "c1",
"type": "function",
"function": {"name": "f", "arguments": "{}"},
}
],
)
assert "tool_calls" in str(exc_info.value)
def test_tool_call_id_on_user_rejected(self):
with pytest.raises(ValidationError) as exc_info:
ChatMessage(role = "user", content = "Hi", tool_call_id = "call_1")
assert "tool_call_id" in str(exc_info.value)
def test_name_on_user_rejected(self):
with pytest.raises(ValidationError) as exc_info:
ChatMessage(role = "user", content = "Hi", name = "get_weather")
assert "name" in str(exc_info.value)
# =====================================================================
# ChatCompletionRequest — standard OpenAI tool fields
# =====================================================================
class TestChatCompletionRequestToolFields:
def _make(self, **kwargs):
base = {"messages": [{"role": "user", "content": "Hi"}]}
base.update(kwargs)
return ChatCompletionRequest(**base)
def test_tools_parses(self):
req = self._make(
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Return the weather in a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
}
],
)
assert req.tools is not None
assert len(req.tools) == 1
assert req.tools[0]["function"]["name"] == "get_weather"
def test_image_base64_allows_empty_user_text(self):
req = ChatCompletionRequest(
messages = [{"role": "user", "content": ""}],
image_base64 = "aW1hZ2U=",
)
assert req.messages[0].content == ""
assert req.image_base64 == "aW1hZ2U="
def test_tool_choice_string_auto(self):
assert self._make(tool_choice = "auto").tool_choice == "auto"
def test_tool_choice_string_required(self):
assert self._make(tool_choice = "required").tool_choice == "required"
def test_tool_choice_string_none(self):
assert self._make(tool_choice = "none").tool_choice == "none"
def test_tool_choice_named_function(self):
tc = {"type": "function", "function": {"name": "get_weather"}}
assert self._make(tool_choice = tc).tool_choice == tc
def test_stop_string(self):
assert self._make(stop = "\nUser:").stop == "\nUser:"
def test_stop_list(self):
assert self._make(stop = ["\nUser:", "\nAssistant:"]).stop == ["\nUser:", "\nAssistant:"]
def test_tools_default_none(self):
req = self._make()
assert req.tools is None
assert req.tool_choice is None
assert req.stop is None
def test_extra_fields_accepted(self):
# `frequency_penalty` and `response_format` are not yet explicitly
# declared but must survive Pydantic parsing now that extra="allow" is
# set. `seed` is declared and should land on the typed field instead.
req = self._make(
frequency_penalty = 0.5,
seed = 42,
response_format = {"type": "json_object"},
)
assert req.seed == 42
# Extras land in model_extra
assert req.model_extra is not None
assert req.model_extra.get("frequency_penalty") == 0.5
assert "seed" not in req.model_extra
assert req.model_extra.get("response_format") == {"type": "json_object"}
def test_unsloth_extensions_still_work(self):
req = self._make(
enable_tools = True,
enabled_tools = ["web_search", "python"],
session_id = "abc",
)
assert req.enable_tools is True
assert req.enabled_tools == ["web_search", "python"]
assert req.session_id == "abc"
def test_stream_defaults_false_matching_openai_spec(self):
# OpenAI defaults `stream` to false. Studio used to default true,
# breaking naive curl/.NET clients (#5047) that omit it. Pin the fix.
req = self._make()
assert req.stream is False
def test_post_without_stream_field_decodes_to_stream_false_over_http(self, monkeypatch):
# Wire-level guard: a POST body omitting `stream` must deserialise to
# stream=False and return application/json, never text/event-stream.
# Mounts the real router to catch middleware/aliasing regressions;
# backends are bypassed via provider_type + a stubbed proxy.
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
import routes.inference as inference_route
from auth.authentication import get_current_subject
captured = {}
async def _fake_proxy(payload, request, current_subject):
assert current_subject == "test-user"
captured["stream"] = payload.stream
return JSONResponse({"choices": [], "object": "chat.completion"})
monkeypatch.setattr(inference_route, "_proxy_to_external_provider", _fake_proxy)
app = FastAPI()
app.include_router(inference_route.router)
app.dependency_overrides[get_current_subject] = lambda: "test-user"
client = TestClient(app)
resp = client.post(
"/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
},
)
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("application/json")
assert "text/event-stream" not in resp.headers["content-type"]
assert captured["stream"] is False
def _v1_client(
self,
monkeypatch,
llama_backend,
inference_backend = None,
):
from fastapi import FastAPI
from fastapi.testclient import TestClient
import routes.inference as inference_route
from auth.authentication import get_current_subject
from utils.api_errors import install_api_error_handlers
monkeypatch.setattr(inference_route, "get_llama_cpp_backend", lambda: llama_backend)
if inference_backend is not None:
monkeypatch.setattr(inference_route, "get_inference_backend", lambda: inference_backend)
app = FastAPI()
app.include_router(inference_route.router, prefix = "/v1")
install_api_error_handlers(app)
app.dependency_overrides[get_current_subject] = lambda: "test-user"
return TestClient(app)
def _assert_unsupported_param(self, response, param):
assert response.status_code == 400
body = response.json()
assert body["error"]["param"] == param
assert body["error"]["code"] == "unsupported_parameter"
def _assert_unsupported_n(self, response):
self._assert_unsupported_param(response, "n")
def test_n_allows_openai_chat_completion_range(self):
req = self._make(n = 128)
assert req.n == 128
with pytest.raises(ValidationError):
self._make(n = 129)
def test_n_rejected_for_external_provider_path(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"n": 2,
},
)
self._assert_unsupported_n(resp)
def test_confirm_tool_calls_rejected_for_provider_tools(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"external_model": "gpt-4.1",
"enable_tools": True,
"enabled_tools": ["web_search"],
"confirm_tool_calls": True,
},
)
assert resp.status_code == 400
body = resp.json()
assert body["error"]["param"] == "confirm_tool_calls"
assert "only supported for local streaming tools" in body["error"]["message"]
def test_logprobs_rejected_until_supported(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"logprobs": True,
},
)
self._assert_unsupported_param(resp, "logprobs")
def test_top_logprobs_rejected_until_supported(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"top_logprobs": 3,
},
)
self._assert_unsupported_param(resp, "top_logprobs")
def test_n_rejected_for_gguf_streaming_path(self, monkeypatch):
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
is_vision = False
_is_audio = False
context_length = 4096
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"stream": True,
"n": 2,
},
)
self._assert_unsupported_n(resp)
def test_n_rejected_for_gguf_tools_passthrough_path(self, monkeypatch):
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = True
is_vision = False
_is_audio = False
context_length = 4096
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object"},
},
}
],
"n": 2,
},
)
self._assert_unsupported_n(resp)
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "n > 1 is not supported" in entry["error"]
assert monitor.active_count() == 0
def test_n_rejected_for_non_gguf_path(self, monkeypatch):
class _NoGGUFBackend:
is_loaded = False
supports_tools = False
class _InferenceBackend:
active_model_name = "test-model"
models = {"test-model": {}}
client = self._v1_client(monkeypatch, _NoGGUFBackend(), _InferenceBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"n": 2,
},
)
self._assert_unsupported_n(resp)
def test_confirm_tool_calls_requires_streaming_for_safetensors_tools(self, monkeypatch):
import routes.inference as inference_route
class _NoGGUFBackend:
is_loaded = False
supports_tools = False
class _InferenceBackend:
active_model_name = "test-model"
models = {"test-model": {"chat_template_info": {"template": "chatml"}}}
def generate_chat_completion_with_tools(self, **kwargs):
raise AssertionError("tool loop should be rejected before starting")
def generate_chat_completion(self, **kwargs):
raise AssertionError("plain path should not be used")
monkeypatch.setattr(
inference_route,
"_detect_safetensors_features",
lambda backend, chat_template: {"supports_tools": True},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _NoGGUFBackend(), _InferenceBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"enable_tools": True,
"enabled_tools": ["web_search"],
"confirm_tool_calls": True,
"stream": False,
},
)
assert resp.status_code == 400
body = resp.json()
assert body["error"]["param"] == "confirm_tool_calls"
assert "requires stream=true" in body["error"]["message"]
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "confirm_tool_calls requires stream=true" in entry["error"]
assert monitor.active_count() == 0
def test_multiturn_tool_loop_messages(self):
req = ChatCompletionRequest(
messages = [
{"role": "user", "content": "What's the weather in Paris?"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"temperature": 14, "unit": "celsius"}',
},
],
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"parameters": {"type": "object"},
},
}
],
)
assert len(req.messages) == 3
assert req.messages[1].role == "assistant"
assert req.messages[1].content is None
assert req.messages[1].tool_calls[0]["id"] == "call_1"
assert req.messages[2].role == "tool"
assert req.messages[2].tool_call_id == "call_1"
# =====================================================================
# anthropic_tool_choice_to_openai — pure translation helper
# =====================================================================
class TestAnthropicToolChoiceToOpenAI:
def test_auto(self):
assert anthropic_tool_choice_to_openai({"type": "auto"}) == "auto"
def test_any_becomes_required(self):
assert anthropic_tool_choice_to_openai({"type": "any"}) == "required"
def test_none(self):
assert anthropic_tool_choice_to_openai({"type": "none"}) == "none"
def test_tool_named(self):
result = anthropic_tool_choice_to_openai({"type": "tool", "name": "get_weather"})
assert result == {"type": "function", "function": {"name": "get_weather"}}
def test_tool_missing_name_returns_none(self):
assert anthropic_tool_choice_to_openai({"type": "tool"}) is None
def test_none_input_returns_none(self):
assert anthropic_tool_choice_to_openai(None) is None
def test_unrecognized_shape_returns_none(self):
assert anthropic_tool_choice_to_openai({"type": "wibble"}) is None
assert anthropic_tool_choice_to_openai("auto") is None
assert anthropic_tool_choice_to_openai(42) is None
# =====================================================================
# _build_passthrough_payload — tool_choice propagation
# =====================================================================
class TestBuildPassthroughPayloadToolChoice:
def _args(self):
return dict(
openai_messages = [{"role": "user", "content": "Hi"}],
openai_tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
temperature = 0.6,
top_p = 0.95,
top_k = 20,
max_tokens = 128,
stream = False,
)
def test_default_tool_choice_is_auto(self):
body = _build_passthrough_payload(**self._args())
assert body["tool_choice"] == "auto"
def test_override_tool_choice_required(self):
body = _build_passthrough_payload(**self._args(), tool_choice = "required")
assert body["tool_choice"] == "required"
def test_override_tool_choice_none(self):
body = _build_passthrough_payload(**self._args(), tool_choice = "none")
assert body["tool_choice"] == "none"
def test_override_tool_choice_named_function(self):
tc = {"type": "function", "function": {"name": "f"}}
body = _build_passthrough_payload(**self._args(), tool_choice = tc)
assert body["tool_choice"] == tc
def test_stream_omits_usage_options_when_client_did_not_request_them(self):
args = self._args()
args["stream"] = True
body = _build_passthrough_payload(**args)
assert "stream_options" not in body
def test_stream_forwards_include_usage_when_client_requests_it(self):
args = self._args()
args["stream"] = True
body = _build_passthrough_payload(
**args,
stream_options = {"include_usage": True},
)
assert body.get("stream_options") == {"include_usage": True}
def test_stream_forwards_include_usage_false_when_client_requests_it(self):
args = self._args()
args["stream"] = True
body = _build_passthrough_payload(
**args,
stream_options = {"include_usage": False},
)
assert body.get("stream_options") == {"include_usage": False}
def test_repetition_penalty_renamed(self):
body = _build_passthrough_payload(**self._args(), repetition_penalty = 1.1)
assert body.get("repeat_penalty") == 1.1
assert "repetition_penalty" not in body
def test_passthrough_body_merges_system_and_developer_messages(self):
payload = ChatCompletionRequest(
model = "default",
messages = [
{"role": "system", "content": "original system"},
{"role": "developer", "content": "developer rules"},
{"role": "user", "content": "hi"},
],
tools = self._args()["openai_tools"],
)
body = _build_openai_passthrough_body(payload, backend_ctx = 4096)
assert body["messages"] == [
{"role": "system", "content": "original system\n\ndeveloper rules"},
{"role": "user", "content": "hi"},
]
# =====================================================================
# Passthrough reasoning kwargs — enable_thinking / reasoning_effort /
# preserve_thinking must reach llama-server via chat_template_kwargs,
# gated on template capabilities like the non-passthrough paths.
# =====================================================================
def _reasoning_backend(
supports_reasoning = True,
reasoning_style = "enable_thinking",
reasoning_always_on = False,
supports_preserve_thinking = False,
):
"""Bare LlamaCppBackend with just the reasoning capability flags set,
so _build_openai_passthrough_body exercises the real
_request_reasoning_kwargs gating."""
from core.inference.llama_cpp import LlamaCppBackend
backend = LlamaCppBackend.__new__(LlamaCppBackend)
backend._supports_reasoning = supports_reasoning
backend._reasoning_style = reasoning_style
backend._reasoning_always_on = reasoning_always_on
backend._supports_preserve_thinking = supports_preserve_thinking
return backend
class TestPassthroughReasoningKwargs:
def _payload(self, **fields):
return ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
**fields,
)
def test_enable_thinking_forwarded(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False),
backend_ctx = 4096,
llama_backend = _reasoning_backend(),
)
assert body["chat_template_kwargs"] == {"enable_thinking": False}
def test_preserve_thinking_forwarded_when_template_supports_it(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = True, preserve_thinking = True),
backend_ctx = 4096,
llama_backend = _reasoning_backend(supports_preserve_thinking = True),
)
assert body["chat_template_kwargs"] == {
"enable_thinking": True,
"preserve_thinking": True,
}
def test_preserve_thinking_dropped_when_template_lacks_it(self):
body = _build_openai_passthrough_body(
self._payload(preserve_thinking = True),
backend_ctx = 4096,
llama_backend = _reasoning_backend(supports_preserve_thinking = False),
)
assert "chat_template_kwargs" not in body
def test_reasoning_effort_forwarded_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(reasoning_effort = "high"),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "high"}
def test_reasoning_effort_none_forwarded_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False, reasoning_effort = "none"),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "none"}
def test_reasoning_effort_minimal_maps_to_low_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = True, reasoning_effort = "minimal"),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "low"}
def test_enable_thinking_maps_to_effort_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "low"}
def test_always_on_reasoning_skips_thinking_kwargs(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_always_on = True),
)
assert "chat_template_kwargs" not in body
def test_no_reasoning_fields_omits_chat_template_kwargs(self):
body = _build_openai_passthrough_body(
self._payload(),
backend_ctx = 4096,
llama_backend = _reasoning_backend(supports_preserve_thinking = True),
)
assert "chat_template_kwargs" not in body
# =====================================================================
# OpenAI API compatibility helpers — verified spec edge cases
# =====================================================================
class TestOpenAICompatibilityHelpers:
def test_max_completion_tokens_wins_over_deprecated_max_tokens(self):
payload = SimpleNamespace(max_tokens = 128, max_completion_tokens = 64)
assert _effective_max_tokens(payload) == 64
@pytest.mark.parametrize(
"finish_reason",
["stop", "length", "tool_calls", "content_filter", "function_call"],
)
def test_clamp_finish_reason_preserves_openai_finish_reasons(self, finish_reason):
assert _clamp_finish_reason(finish_reason) == finish_reason
def test_clamp_finish_reason_defaults_unknown_to_stop(self):
assert _clamp_finish_reason(None) == "stop"
assert _clamp_finish_reason("unexpected") == "stop"
def test_non_streaming_completion_choice_accepts_tool_calls_finish_reason(self):
choice = CompletionChoice(
index = 0,
message = CompletionMessage(content = ""),
finish_reason = "tool_calls",
)
assert choice.finish_reason == "tool_calls"
def test_stream_usage_chunk_requires_include_usage(self):
usage = {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}
payload = SimpleNamespace(stream_options = None)
assert (
_openai_stream_usage_chunk(payload, "chatcmpl-test", 123, "model", usage, None) is None
)
payload.stream_options = {"include_usage": True}
line = _openai_stream_usage_chunk(payload, "chatcmpl-test", 123, "model", usage, None)
assert line is not None
assert '"choices":[]' in line
assert '"usage"' in line
def test_stream_usage_chunk_coerces_nullable_counts(self):
payload = SimpleNamespace(stream_options = {"include_usage": True})
line = _openai_stream_usage_chunk(
payload,
"chatcmpl-test",
123,
"model",
{"prompt_tokens": None, "completion_tokens": 7, "total_tokens": None},
None,
)
assert line is not None
parsed = json.loads(line.removeprefix("data: "))
usage = parsed["usage"]
assert usage["prompt_tokens"] == 0
assert usage["completion_tokens"] == 7
assert usage["total_tokens"] == 7
def test_completion_stream_monitor_reads_usage_before_client_strip(self, monkeypatch):
import routes.inference as inf_mod
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/completions",
method = "POST",
model = "m",
prompt = "hi",
context_length = 100,
)
event = (
b'data: {"id":"chatcmpl-test","choices":[{"text":"done","finish_reason":"stop"}],'
b'"usage":{"prompt_tokens":4,"completion_tokens":6,"total_tokens":10}}\n'
)
_monitor_openai_sse_event(monitor_id, event, context_length = 100)
out = _cmpl_stream_event_out(event, include_usage = False)
assert out is not None
assert b'"usage"' not in out
[entry] = monitor.snapshot()
assert entry["reply"] == "done"
assert entry["prompt_tokens"] == 4
assert entry["completion_tokens"] == 6
assert entry["total_tokens"] == 10
assert entry["context_usage"] == 0.1
def test_developer_message_preserves_existing_system_prompt(self):
payload = ChatCompletionRequest(
messages = [
{"role": "system", "content": "original system"},
{"role": "developer", "content": "developer rules"},
{"role": "user", "content": "hi"},
]
)
for message in payload.messages:
if message.role == "developer":
message.role = "system"
system_prompt, chat_messages, image_b64 = _extract_content_parts(payload.messages)
assert system_prompt == "original system\n\ndeveloper rules"
assert chat_messages == [{"role": "user", "content": "hi"}]
assert image_b64 is None
# =====================================================================
# _friendly_error — httpx transport failures
# =====================================================================
class TestFriendlyErrorHttpx:
def _req(self):
return httpx.Request("POST", "http://127.0.0.1:65535/v1/chat/completions")
def test_connect_error_mapped(self):
exc = httpx.ConnectError("All connection attempts failed", request = self._req())
assert "Lost connection" in _friendly_error(exc)
def test_read_error_mapped(self):
exc = httpx.ReadError("EOF", request = self._req())
assert "Lost connection" in _friendly_error(exc)
def test_remote_protocol_error_mapped(self):
exc = httpx.RemoteProtocolError("peer closed", request = self._req())
assert "Lost connection" in _friendly_error(exc)
def test_read_timeout_mapped(self):
exc = httpx.ReadTimeout("timed out", request = self._req())
assert "first token within 20 minutes" in _friendly_error(exc)
def test_non_httpx_unchanged(self):
# Non-httpx exceptions still fall through to the substring heuristics
# — a context-size message must still produce "Message too long".
ctx_msg = "request (4096 tokens) exceeds the available context size (2048 tokens)"
assert "Message too long" in _friendly_error(ValueError(ctx_msg))
def test_generic_exception_returns_generic_message(self):
assert _friendly_error(RuntimeError("unrelated")) == "An internal error occurred"
from routes.inference import ( # noqa: E402
_drop_empty_assistant_sentinels,
_openai_messages_for_gguf_chat,
_openai_messages_for_passthrough,
)
class TestDropEmptyAssistantSentinels:
def test_drops_empty_assistant_between_real_turns(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": ""},
{"role": "user", "content": "again"},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == [{"role": "user", "content": "hi"}, {"role": "user", "content": "again"}]
def test_drops_assistant_with_no_content_key(self):
# exclude_none=True strips the content key entirely; filter must catch it.
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant"},
{"role": "user", "content": "ok"},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == [{"role": "user", "content": "hi"}, {"role": "user", "content": "ok"}]
def test_preserves_assistant_with_text(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello back"},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == msgs
def test_preserves_assistant_with_tool_calls_only(self):
msgs = [
{"role": "user", "content": "weather?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{}"},
},
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"t": 72}',
},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == msgs
def test_preserves_user_and_system_with_empty_content(self):
# Filter scoped to role="assistant" only.
msgs = [
{"role": "system", "content": ""},
{"role": "user", "content": ""},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == msgs
def test_openai_messages_for_passthrough_drops_sentinel(self):
"""End-to-end: Stop-sentinel must not reach the wire."""
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
out = _openai_messages_for_passthrough(req)
roles = [m["role"] for m in out]
assert roles == ["user", "user"]
for m in out:
assert m.get("content"), m
class TestGgufVisionMessages:
_PNG_B64 = (
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADUlEQVR42mNk"
"+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
)
def test_preserves_multiturn_image_parts_on_original_turns(self):
req = ChatCompletionRequest(
model = "default",
image_base64 = self._PNG_B64,
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "describe image one"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
{"role": "assistant", "content": "first answer"},
{
"role": "user",
"content": [
{"type": "text", "text": "describe image two"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
],
)
messages, has_image = _openai_messages_for_gguf_chat(req, is_vision = True)
assert has_image is True
assert messages[0]["content"][0] == {"type": "text", "text": "describe image one"}
assert messages[0]["content"][1]["type"] == "image_url"
assert len(messages[0]["content"]) == 2
assert messages[2]["content"][0] == {"type": "text", "text": "describe image two"}
assert messages[2]["content"][1]["type"] == "image_url"
assert len(messages[2]["content"]) == 2
assert isinstance(messages[1]["content"], str)
# Legacy top-level image_base64 must be ignored when a message-level
# image exists; otherwise turn 2 ends up with two image parts.
for msg in messages:
content = msg.get("content")
if isinstance(content, list):
image_parts = [p for p in content if p.get("type") == "image_url"]
assert len(image_parts) == 1, msg
def test_legacy_image_base64_is_injected_when_messages_are_text_only(self):
req = ChatCompletionRequest(
model = "default",
image_base64 = self._PNG_B64,
messages = [{"role": "user", "content": "describe this image"}],
)
messages, has_image = _openai_messages_for_gguf_chat(req, is_vision = True)
assert has_image is True
assert messages[0]["content"][0] == {"type": "text", "text": "describe this image"}
assert messages[0]["content"][1]["type"] == "image_url"
assert messages[0]["content"][1]["image_url"]["url"].startswith("data:image/png;base64,")
def test_rejects_image_parts_for_text_only_gguf(self):
req = ChatCompletionRequest(
model = "default",
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "look"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
],
)
with pytest.raises(HTTPException) as exc_info:
_openai_messages_for_gguf_chat(req, is_vision = False)
assert "does not support vision" in str(exc_info.value)
def test_tool_nudge_system_update_preserves_image_parts(self):
messages = [
{"role": "system", "content": "Base instructions."},
{
"role": "user",
"content": [
{"type": "text", "text": "describe this"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
]
updated = _set_or_prepend_system_message(
messages, "Base instructions.\n\nUse tools when appropriate."
)
assert updated[0] == {
"role": "system",
"content": "Base instructions.\n\nUse tools when appropriate.",
}
assert updated[1]["content"][1]["type"] == "image_url"
assert messages[1]["content"][1]["type"] == "image_url"
def test_tool_nudge_system_update_handles_none_messages(self):
assert _set_or_prepend_system_message(None, "") == []
assert _set_or_prepend_system_message(None, "Use tools.") == [
{"role": "system", "content": "Use tools."}
]
def test_tool_nudge_system_update_dedupes_non_leading_system(self):
messages = [
{"role": "user", "content": "earlier"},
{"role": "system", "content": "Mid instructions."},
{"role": "user", "content": "now"},
]
updated = _set_or_prepend_system_message(messages, "Mid instructions.\n\nUse tools.")
assert [m["role"] for m in updated] == ["system", "user", "user"]
assert updated[0]["content"] == "Mid instructions.\n\nUse tools."
class TestGgufVisionToolRouting:
class _Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
async def is_disconnected(self):
return False
@staticmethod
def _drive(coro):
return asyncio.run(coro)
@staticmethod
def _consume_response(response):
async def _consume():
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
return chunks
return TestGgufVisionToolRouting._drive(_consume())
def test_image_request_with_enabled_tools_enters_gguf_tool_loop(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
captured = {}
def _plain(**kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**kwargs):
captured["kwargs"] = kwargs
yield {"type": "content", "text": "done"}
backend = SimpleNamespace(
is_loaded = True,
is_vision = True,
supports_tools = True,
model_identifier = "gemma-4-12b-it-GGUF",
context_length = 4096,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
enable_tools = True,
enabled_tools = ["web_search"],
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What is in this image?"},
{
"type": "image_url",
"image_url": {
"url": (f"data:image/png;base64,{TestGgufVisionMessages._PNG_B64}"),
},
},
],
},
],
)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
self._consume_response(response)
assert "kwargs" in captured
assert captured["kwargs"]["tools"]
tool_messages = captured["kwargs"]["messages"]
assert tool_messages[0]["role"] == "system"
assert tool_messages[1]["role"] == "user"
assert tool_messages[1]["content"][1]["type"] == "image_url"
def test_parallel_tool_calls_false_reaches_gguf_tool_loop(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
captured = {}
def _plain(**kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**kwargs):
captured["kwargs"] = kwargs
yield {"type": "content", "text": "done"}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
enable_tools = True,
enabled_tools = ["web_search"],
parallel_tool_calls = False,
messages = [{"role": "user", "content": "search once"}],
)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
self._consume_response(response)
assert captured["kwargs"]["disable_parallel_tool_use"] is True
def test_confirm_tool_calls_requires_streaming_for_gguf_tools(self, monkeypatch):
import routes.inference as inf_mod
def _plain(**kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**kwargs):
raise AssertionError("tool loop should be rejected before starting")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
enable_tools = True,
enabled_tools = ["web_search"],
confirm_tool_calls = True,
stream = False,
messages = [{"role": "user", "content": "search once"}],
)
with pytest.raises(HTTPException) as exc:
self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert exc.value.status_code == 400
assert "requires stream=true" in exc.value.detail["error"]["message"]
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "confirm_tool_calls requires stream=true" in entry["error"]
assert monitor.active_count() == 0
def test_non_streaming_gguf_n_records_all_monitor_replies(self, monkeypatch):
import routes.inference as inf_mod
calls = {"count": 0}
def _generate(**_kwargs):
calls["count"] += 1
text = f"reply {calls['count']}"
yield text
yield {
"type": "metadata",
"usage": {
"prompt_tokens": 3,
"completion_tokens": calls["count"],
"total_tokens": 3 + calls["count"],
},
}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
n = 2,
messages = [{"role": "user", "content": "two please"}],
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
body = json.loads(response.body)
assert [c["message"]["content"] for c in body["choices"]] == ["reply 1", "reply 2"]
[entry] = monitor.snapshot()
assert entry["reply"] == "Choice 1:\nreply 1\n\nChoice 2:\nreply 2"
assert entry["completion_tokens"] == 3
assert monitor.active_count() == 0
def test_standard_gguf_merges_system_and_developer_messages(self, monkeypatch):
import routes.inference as inf_mod
captured = {}
def _generate(**kwargs):
captured["messages"] = kwargs["messages"]
yield "done"
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
"finish_reason": "stop",
}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [
{"role": "system", "content": "original system"},
{"role": "developer", "content": "developer rules"},
{"role": "user", "content": "hi"},
],
)
self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
assert captured["messages"] == [
{"role": "system", "content": "original system\n\ndeveloper rules"},
{"role": "user", "content": "hi"},
]
@pytest.mark.parametrize(
("seed", "expected"),
[
(41, [41, 42, 43]),
(-1, [-1, -1, -1]),
],
)
def test_gguf_n_choices_vary_explicit_non_negative_seed(self, monkeypatch, seed, expected):
import routes.inference as inf_mod
seen_seeds = []
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
def _generate(**kwargs):
seen_seeds.append(kwargs.get("seed"))
yield f"choice-{len(seen_seeds)}"
yield {
"type": "metadata",
"usage": {
"prompt_tokens": 5,
"completion_tokens": 7,
"total_tokens": 12,
},
"finish_reason": "stop",
}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
n = 3,
seed = seed,
)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
body = json.loads(response.body)
assert seen_seeds == expected
assert [choice["index"] for choice in body["choices"]] == [0, 1, 2]
assert body["usage"]["prompt_tokens"] == 5
assert body["usage"]["completion_tokens"] == 21
[entry] = monitor.snapshot()
assert entry["prompt_tokens"] == 5
assert entry["completion_tokens"] == 21
assert entry["total_tokens"] == 26
class TestApiMonitorProviderAndCompletionStreams:
class _Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
async def is_disconnected(self):
return False
def test_external_non_streaming_json_updates_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyExternalClient:
def __init__(self, **_kwargs):
pass
async def stream_chat_completion(self, **kwargs):
assert kwargs["stream"] is False
yield json.dumps(
{
"choices": [{"message": {"content": "provider [DONE] reply"}}],
"usage": {
"prompt_tokens": 3,
"completion_tokens": 4,
"total_tokens": 7,
},
}
)
async def close(self):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "ExternalProviderClient", DummyExternalClient)
payload = ChatCompletionRequest(
model = "default",
external_model = "gpt-test",
provider_type = "openai",
provider_base_url = "https://api.openai.com/v1",
messages = [ChatMessage(role = "user", content = "hi")],
)
response = await _proxy_to_external_provider(payload, self._Request())
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
assert chunks[-1] == "data: [DONE]\n\n"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "provider [DONE] reply"
assert entry["prompt_tokens"] == 3
assert entry["completion_tokens"] == 4
assert entry["total_tokens"] == 7
asyncio.run(_run())
def test_external_stream_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyExternalClient:
def __init__(self, **_kwargs):
pass
async def stream_chat_completion(self, **_kwargs):
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}'
await asyncio.sleep(3600)
async def close(self):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "ExternalProviderClient", DummyExternalClient)
payload = ChatCompletionRequest(
model = "default",
external_model = "gpt-test",
provider_type = "openai",
provider_base_url = "https://api.openai.com/v1",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await _proxy_to_external_provider(payload, self._Request())
iterator = response.body_iterator
first = await anext(iterator)
assert "hello" in first
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_completions_preheader_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": True}
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return None
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
response = await openai_completions(Request(), current_subject = "test")
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
assert chunks == []
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_stream_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": True}
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield b'data: {"choices":[{"text":"hello"}]}\n\n'
await asyncio.sleep(3600)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
response = await openai_completions(Request(), current_subject = "test")
iterator = response.body_iterator
first = await anext(iterator)
assert b"hello" in first
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_completions_non_streaming_post_error_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": False}
class FailingAsyncClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def post(self, *_args, **_kwargs):
raise httpx.ConnectError("llama down")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"nonstreaming_client",
lambda: FailingAsyncClient(),
)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
with pytest.raises(httpx.ConnectError):
await openai_completions(Request(), current_subject = "test")
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "Lost connection to the model server" in entry["error"]
assert monitor.active_count() == 0
asyncio.run(_run())
def test_monitor_openai_chunk_records_all_choice_replies(self, monkeypatch):
import routes.inference as inf_mod
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
_monitor_openai_chunk(
monitor_id,
{
"choices": [
{"text": "first"},
{"text": "second"},
],
"usage": {
"prompt_tokens": 2,
"completion_tokens": 5,
"total_tokens": 7,
},
},
4096,
)
entry = monitor.get(monitor_id)
assert entry["reply"] == "Choice 1:\nfirst\n\nChoice 2:\nsecond"
assert entry["prompt_tokens"] == 2
assert entry["completion_tokens"] == 5
assert entry["context_length"] == 4096
def test_monitor_openai_chunk_records_tool_call_reply(self, monkeypatch):
import routes.inference as inf_mod
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
_monitor_openai_chunk(
monitor_id,
{
"choices": [
{
"message": {
"tool_calls": [
{
"type": "function",
"function": {
"name": "lookup",
"arguments": '{"query":"weather"}',
},
}
]
}
}
]
},
4096,
)
entry = monitor.get(monitor_id)
assert entry["reply"] == 'Tool call: lookup({"query":"weather"})'
def test_embeddings_request_is_counted_active_and_completed(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/embeddings")
method = "POST"
async def json(self):
return {"input": ["alpha", "beta"], "model": "embed"}
class FakeAsyncClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def post(self, *_args, **_kwargs):
assert monitor.active_count() == 1
return httpx.Response(
200,
json = {
"data": [{"embedding": [0.1]}],
"usage": {"prompt_tokens": 4, "total_tokens": 4},
},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"nonstreaming_client",
lambda: FakeAsyncClient(),
)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
response = await openai_embeddings(Request(), current_subject = "test")
assert response.status_code == 200
[entry] = monitor.snapshot()
assert entry["endpoint"] == "/v1/embeddings"
assert entry["status"] == "completed"
assert entry["prompt_preview"] == "alpha\nbeta"
assert entry["prompt_tokens"] == 4
assert entry["total_tokens"] == 4
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_task_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
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)
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/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
iterator = response.body_iterator
first = await anext(iterator)
assert "hello" in first
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_passthrough_non_streaming_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class CancellingAsyncClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def post(self, *_args, **_kwargs):
raise asyncio.CancelledError()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"nonstreaming_client",
lambda: CancellingAsyncClient(),
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
with pytest.raises(asyncio.CancelledError):
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
)
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_clean_eof_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}'
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/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
assert chunks == ['data: {"choices":[{"delta":{"content":"hello"}}]}\n\n']
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "hello"
assert monitor.active_count() == 0
asyncio.run(_run())
class TestApiMonitorSafetensorsUsage:
class _Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
def test_non_streaming_safetensors_records_usage(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyBackend:
active_model_name = "safe-model"
models = {"safe-model": {"context_length": 2048}}
def generate_chat_response(self, *, stats_holder, **_kwargs):
stats_holder["stats"] = {
"usage": {
"prompt_tokens": 8,
"completion_tokens": 5,
"total_tokens": 13,
}
}
yield "safe reply"
def reset_generation_state(self):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = False,
supports_tools = False,
is_vision = False,
context_length = None,
),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyBackend())
monkeypatch.setattr(
inf_mod,
"_detect_safetensors_features",
lambda *_args, **_kwargs: {"supports_tools": False},
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
body = json.loads(response.body)
assert body["choices"][0]["message"]["content"] == "safe reply"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "safe reply"
assert entry["prompt_tokens"] == 8
assert entry["completion_tokens"] == 5
assert entry["total_tokens"] == 13
assert entry["context_length"] == 2048
asyncio.run(_run())
def test_non_streaming_safetensors_tool_cancel_records_cancelled(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
reset_tool_policy()
class DummyBackend:
active_model_name = "safe-model"
models = {"safe-model": {"context_length": 2048}}
def generate_chat_response(self, **_kwargs):
raise AssertionError("plain safetensors path should not be used")
def generate_chat_completion_with_tools(
self, *, cancel_event, stats_holder, **_kwargs
):
stats_holder["stats"] = {
"usage": {
"prompt_tokens": 8,
"completion_tokens": 5,
"total_tokens": 13,
}
}
yield {"type": "content", "text": "partial"}
cancel_event.set()
yield {"type": "content", "text": "ignored"}
def reset_generation_state(self):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = False,
supports_tools = False,
is_vision = False,
context_length = None,
),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyBackend())
monkeypatch.setattr(
inf_mod,
"_detect_safetensors_features",
lambda *_args, **_kwargs: {"supports_tools": True},
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
enable_tools = True,
enabled_tools = ["web_search"],
cancel_id = "safe-cancel",
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
body = json.loads(response.body)
assert body["choices"][0]["message"]["content"] == "partial"
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["reply"] == "partial"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_non_streaming_safetensors_tool_task_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
reset_tool_policy()
reset_called = False
class DummyBackend:
active_model_name = "safe-model"
models = {"safe-model": {"context_length": 2048}}
def generate_chat_response(self, **_kwargs):
raise AssertionError("plain safetensors path should not be used")
def generate_chat_completion_with_tools(self, **_kwargs):
yield {"type": "content", "text": "unused"}
def reset_generation_state(self):
nonlocal reset_called
reset_called = True
async def fake_to_thread(*_args, **_kwargs):
raise asyncio.CancelledError()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod.asyncio, "to_thread", fake_to_thread)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = False,
supports_tools = False,
is_vision = False,
context_length = None,
),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyBackend())
monkeypatch.setattr(
inf_mod,
"_detect_safetensors_features",
lambda *_args, **_kwargs: {"supports_tools": True},
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
enable_tools = True,
enabled_tools = ["web_search"],
cancel_id = "safe-cancel",
)
with pytest.raises(asyncio.CancelledError):
await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
assert reset_called is True
asyncio.run(_run())
class TestApiMonitorAudioInput:
def _patch_audio_backend(self, monkeypatch, chunks):
import routes.inference as inf_mod
class DummyAudioBackend:
active_model_name = "audio-model"
models = {
"audio-model": {
"has_audio_input": True,
"audio_type": "audio-input",
}
}
def generate_audio_input_response(self, **_kwargs):
yield from chunks
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(is_loaded = False),
)
monkeypatch.setattr(
inf_mod,
"get_inference_backend",
lambda: DummyAudioBackend(),
)
monkeypatch.setattr(
inf_mod,
"_decode_audio_base64",
lambda _payload: object(),
)
return inf_mod
def test_audio_input_non_streaming_records_active_monitor(self, monkeypatch):
async def _run():
inf_mod = self._patch_audio_backend(monkeypatch, ["hello", " world"])
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "describe this audio")],
audio_base64 = "ZmFrZQ==",
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
response = await openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
body = json.loads(response.body)
assert body["choices"][0]["message"]["content"] == "hello world"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "hello world"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_audio_input_streaming_records_monitor_reply(self, monkeypatch):
async def _run():
inf_mod = self._patch_audio_backend(monkeypatch, ["hello", " world"])
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
async def is_disconnected():
return False
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "describe this audio")],
audio_base64 = "ZmFrZQ==",
stream = True,
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
is_disconnected = is_disconnected,
)
response = await openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk)
assert chunks[-1] == "data: [DONE]\n\n"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "hello world"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_non_gguf_tts_auto_route_records_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyTtsBackend:
active_model_name = "tts-model"
models = {
"tts-model": {
"is_audio": True,
"audio_type": "snac",
}
}
async def fake_generate_audio(
_payload,
_request,
current_subject = None,
):
return inf_mod.JSONResponse(
content = {
"choices": [
{
"message": {
"content": "[Generated audio]",
}
}
]
}
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(is_loaded = False),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyTtsBackend())
monkeypatch.setattr(inf_mod, "generate_audio", fake_generate_audio)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "say hello")],
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
response = await inf_mod.openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
assert json.loads(response.body)["choices"][0]["message"]["content"] == (
"[Generated audio]"
)
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["model"] == "tts-model"
assert entry["reply"] == "[Generated audio]"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_non_gguf_tts_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyTtsBackend:
active_model_name = "tts-model"
models = {
"tts-model": {
"is_audio": True,
"audio_type": "snac",
}
}
async def fake_generate_audio(
_payload,
_request,
current_subject = None,
):
raise asyncio.CancelledError()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(is_loaded = False),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyTtsBackend())
monkeypatch.setattr(inf_mod, "generate_audio", fake_generate_audio)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "say hello")],
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
with pytest.raises(asyncio.CancelledError):
await inf_mod.openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["model"] == "tts-model"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_gguf_tts_auto_route_records_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_generate_audio(
_payload,
_request,
current_subject = None,
):
return inf_mod.JSONResponse(
content = {
"choices": [
{
"message": {
"content": "[Generated audio]",
}
}
]
}
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
_is_audio = True,
model_identifier = "gguf-tts",
context_length = 2048,
),
)
monkeypatch.setattr(inf_mod, "generate_audio", fake_generate_audio)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "say hello")],
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
await inf_mod.openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["model"] == "gguf-tts"
assert entry["context_length"] == 2048
assert entry["reply"] == "[Generated audio]"
assert monitor.active_count() == 0
asyncio.run(_run())
# =====================================================================
# Responses API -> Chat Completions translation: chat_template_kwargs
# (e.g. {"enable_thinking": true}) sent via the Responses extra-body must
# reach the built ChatCompletionRequest's typed ``enable_thinking`` field,
# otherwise /v1/responses silently ignores reasoning control (issue #6198).
# =====================================================================
class TestResponsesChatTemplateKwargs:
_messages = [ChatMessage(role = "user", content = "What is 100 - 67?")]
def test_enable_thinking_lifted_from_extra_body(self):
payload = ResponsesRequest(
model = "qwen-local",
input = "What is 100 - 67?",
chat_template_kwargs = {"enable_thinking": True},
)
chat_req = _build_chat_request(payload, self._messages, stream = False)
assert chat_req.enable_thinking is True
def test_enable_thinking_false_lifted_from_extra_body(self):
payload = ResponsesRequest(
model = "qwen-local",
input = "hi",
chat_template_kwargs = {"enable_thinking": False},
)
chat_req = _build_chat_request(payload, self._messages, stream = True)
assert chat_req.enable_thinking is False
def test_no_chat_template_kwargs_leaves_enable_thinking_unset(self):
payload = ResponsesRequest(model = "qwen-local", input = "hi")
chat_req = _build_chat_request(payload, self._messages, stream = False)
assert chat_req.enable_thinking is None
def test_chat_template_kwargs_without_enable_thinking_is_ignored(self):
payload = ResponsesRequest(
model = "qwen-local",
input = "hi",
chat_template_kwargs = {"some_other_flag": True},
)
chat_req = _build_chat_request(payload, self._messages, stream = False)
assert chat_req.enable_thinking is None
# =====================================================================
# GGUF chat-template role alternation: coalesce orphaned user turns left
# behind when an empty assistant turn is dropped, so strict templates
# (Gemma 3, ...) do not 400 on a role-parity break.
# =====================================================================
class TestMergeUserContent:
def test_strings_join_with_blank_line(self):
assert _merge_user_content("hi", "again") == "hi\n\nagain"
def test_empty_sides_passthrough(self):
assert _merge_user_content("", "again") == "again"
assert _merge_user_content("hi", "") == "hi"
def test_multimodal_parts_concatenate(self):
img = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}
out = _merge_user_content([{"type": "text", "text": "look"}, img], "and this?")
assert out == [
{"type": "text", "text": "look"},
img,
{"type": "text", "text": "and this?"},
]
class TestCoalesceConsecutiveUserTurns:
def test_merges_two_string_user_turns(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "user", "content": "again"},
]
assert _coalesce_consecutive_user_turns(msgs) == [
{"role": "user", "content": "hi\n\nagain"},
]
def test_merges_three_consecutive_user_turns(self):
msgs = [
{"role": "user", "content": "a"},
{"role": "user", "content": "b"},
{"role": "user", "content": "c"},
]
assert _coalesce_consecutive_user_turns(msgs) == [
{"role": "user", "content": "a\n\nb\n\nc"},
]
def test_alternating_history_is_unchanged(self):
msgs = [
{"role": "system", "content": "sys"},
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello"},
{"role": "user", "content": "bye"},
]
assert _coalesce_consecutive_user_turns(msgs) == msgs
def test_assistant_and_tool_turns_untouched(self):
msgs = [
{"role": "user", "content": "weather?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "{}"},
]
assert _coalesce_consecutive_user_turns(msgs) == msgs
def test_multimodal_parts_survive_merge(self):
img = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}
msgs = [
{"role": "user", "content": [{"type": "text", "text": "look"}, img]},
{"role": "user", "content": "and this?"},
]
out = _coalesce_consecutive_user_turns(msgs)
assert len(out) == 1
assert out[0]["content"] == [
{"type": "text", "text": "look"},
img,
{"type": "text", "text": "and this?"},
]
def test_does_not_mutate_input(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "user", "content": "again"},
]
_coalesce_consecutive_user_turns(msgs)
assert msgs[0]["content"] == "hi"
class TestGgufChatHistoryAlternation:
def test_empty_assistant_turn_dropped_then_users_coalesced(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
roles = [m["role"] for m in out]
assert roles == ["user"]
assert out[0]["content"] == "hi\n\nagain"
def test_bare_stop_sentinel_also_coalesced(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant"),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
roles = [m["role"] for m in out]
assert all(roles[i] != roles[i + 1] for i in range(len(roles) - 1)), roles
assert roles == ["user"]
def test_system_prompt_preserved(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "system", content = "be brief"),
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
assert [m["role"] for m in out] == ["system", "user"]
assert out[1]["content"] == "hi\n\nagain"
def test_normal_history_unchanged(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = "hello"),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
assert [m["role"] for m in out] == ["user", "assistant", "user"]
def test_tool_path_rebuild_stays_alternating(self):
# Tool path rebuilds via _set_or_prepend_system_message over the coalesced
# history, so it stays alternating too.
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
normalized, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
rebuilt = _set_or_prepend_system_message(normalized, "You have access to tools.")
roles = [m["role"] for m in rebuilt]
assert roles == ["system", "user"]
assert all(roles[i] != roles[i + 1] for i in range(len(roles) - 1)), roles
Reference in a new issue View git blame Copy permalink