Add review tests for PR #5061

tests/test_pr5061_r2_anthropic_nonstream_502.py

@@ -0,0 +1,79 @@
+import asyncio
+import os, sys
+from unittest.mock import patch
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+import httpx
+import pytest
+from fastapi import HTTPException
+
+
+class _Llama:
+    base_url = "http://127.0.0.1:0"
+    _api_key = None
+
+
+def test_httpx_connect_error_mapped_to_502():
+    from routes import inference as inf_mod
+
+    class _BadClient:
+        def __init__(self, *a, **kw):
+            pass
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *a):
+            return False
+        async def post(self, *a, **kw):
+            raise httpx.ConnectError("refused", request=httpx.Request("POST", "http://x"))
+
+    with patch.object(inf_mod.httpx, "AsyncClient", _BadClient):
+        with pytest.raises(HTTPException) as ei:
+            asyncio.run(
+                inf_mod._anthropic_passthrough_non_streaming(
+                    _Llama(),
+                    openai_messages=[{"role": "user", "content": "q"}],
+                    openai_tools=[{"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}],
+                    temperature=0.6,
+                    top_p=0.95,
+                    top_k=20,
+                    max_tokens=None,
+                    message_id="msg_1",
+                    model_name="stub",
+                    tool_choice="auto",
+                )
+            )
+    assert ei.value.status_code == 502
+    assert "Lost connection" in ei.value.detail
+
+
+def test_httpx_read_error_mapped_to_502():
+    from routes import inference as inf_mod
+
+    class _BadClient:
+        def __init__(self, *a, **kw):
+            pass
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *a):
+            return False
+        async def post(self, *a, **kw):
+            raise httpx.ReadError("eof", request=httpx.Request("POST", "http://x"))
+
+    with patch.object(inf_mod.httpx, "AsyncClient", _BadClient):
+        with pytest.raises(HTTPException) as ei:
+            asyncio.run(
+                inf_mod._anthropic_passthrough_non_streaming(
+                    _Llama(),
+                    openai_messages=[{"role": "user", "content": "q"}],
+                    openai_tools=[{"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}],
+                    temperature=0.6,
+                    top_p=0.95,
+                    top_k=20,
+                    max_tokens=None,
+                    message_id="msg_1",
+                    model_name="stub",
+                )
+            )
+    assert ei.value.status_code == 502

tests/test_pr5061_r2_anthropic_tool_choice_warn.py

@@ -0,0 +1,49 @@
+import os, sys
+from unittest.mock import patch, MagicMock
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+from routes import inference as inf_mod
+
+
+def _simulate_coerce(tool_choice):
+    """Mimic the exact snippet that lives at the top of anthropic_messages()."""
+    translated = inf_mod.anthropic_tool_choice_to_openai(tool_choice)
+    if translated is None:
+        if tool_choice is not None:
+            inf_mod.logger.warning(
+                "anthropic_messages.tool_choice_unrecognized",
+                tool_choice=tool_choice,
+            )
+        translated = "auto"
+    return translated
+
+
+def test_unrecognized_dict_warns_and_falls_back_to_auto():
+    with patch.object(inf_mod.logger, "warning") as mock_warn:
+        result = _simulate_coerce({"type": "wibble"})
+    assert result == "auto"
+    mock_warn.assert_called_once()
+    assert mock_warn.call_args.args[0] == "anthropic_messages.tool_choice_unrecognized"
+
+
+def test_non_dict_input_warns_and_falls_back():
+    with patch.object(inf_mod.logger, "warning") as mock_warn:
+        result = _simulate_coerce("auto_string")
+    assert result == "auto"
+    mock_warn.assert_called_once()
+
+
+def test_none_input_no_warning():
+    with patch.object(inf_mod.logger, "warning") as mock_warn:
+        result = _simulate_coerce(None)
+    assert result == "auto"
+    mock_warn.assert_not_called()
+
+
+def test_recognized_input_no_warning():
+    with patch.object(inf_mod.logger, "warning") as mock_warn:
+        result = _simulate_coerce({"type": "any"})
+    assert result == "required"
+    mock_warn.assert_not_called()

tests/test_pr5061_r2_auth_headers.py

@@ -0,0 +1,40 @@
+import os, sys
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+from routes.inference import _llama_auth_headers
+
+
+class _WithKey:
+    _api_key = "k_secret_123"
+
+
+class _NoKey:
+    _api_key = None
+
+
+class _Missing:
+    pass
+
+
+class _EmptyKey:
+    _api_key = ""
+
+
+def test_returns_bearer_header_when_api_key_set():
+    headers = _llama_auth_headers(_WithKey())
+    assert headers == {"Authorization": "Bearer k_secret_123"}
+
+
+def test_returns_none_when_api_key_none():
+    assert _llama_auth_headers(_NoKey()) is None
+
+
+def test_returns_none_when_api_key_attribute_missing():
+    assert _llama_auth_headers(_Missing()) is None
+
+
+def test_returns_none_when_api_key_empty_string():
+    # Empty string is falsy; the helper should treat it as absent.
+    assert _llama_auth_headers(_EmptyKey()) is None

tests/test_pr5061_r2_enable_tools_tool_choice_400.py

@@ -0,0 +1,75 @@
+import asyncio
+import os, sys
+from unittest.mock import patch
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+import pytest
+from fastapi import HTTPException
+
+
+class _Llama:
+    is_loaded = True
+    supports_tools = True
+    is_vision = False
+    _is_audio = False
+    model_identifier = "stub"
+    base_url = "http://127.0.0.1:0"
+    _api_key = None
+
+
+class _Req:
+    async def is_disconnected(self):
+        return False
+
+
+def _build_payload(enable_tools, tool_choice):
+    from models.inference import AnthropicMessagesRequest
+    return AnthropicMessagesRequest(
+        model="default",
+        max_tokens=64,
+        messages=[{"role": "user", "content": "q"}],
+        tool_choice=tool_choice,
+        enable_tools=enable_tools,
+    )
+
+
+def test_enable_tools_true_with_tool_choice_raises_400():
+    from routes import inference as inf_mod
+    payload = _build_payload(enable_tools=True, tool_choice={"type": "any"})
+    with patch.object(inf_mod, "get_llama_cpp_backend", return_value=_Llama()):
+        with pytest.raises(HTTPException) as ei:
+            asyncio.run(inf_mod.anthropic_messages(payload, _Req(), current_subject="u"))
+    assert ei.value.status_code == 400
+    assert "tool_choice" in ei.value.detail
+    assert "enable_tools" in ei.value.detail
+
+
+def test_guard_logic_unit():
+    # Re-implement the guard predicate exactly and pin it.
+    def would_raise(enable_tools, supports_tools, tool_choice):
+        server_tools = enable_tools and supports_tools
+        return bool(server_tools and tool_choice is not None)
+
+    assert would_raise(True, True, {"type": "any"}) is True
+    assert would_raise(True, True, None) is False
+    assert would_raise(False, True, {"type": "any"}) is False
+    assert would_raise(True, False, {"type": "any"}) is False
+
+
+def test_enable_tools_false_with_tool_choice_does_not_raise_the_guard():
+    from routes import inference as inf_mod
+    payload = _build_payload(enable_tools=False, tool_choice={"type": "any"})
+    with patch.object(inf_mod, "get_llama_cpp_backend", return_value=_Llama()):
+        try:
+            asyncio.run(inf_mod.anthropic_messages(payload, _Req(), current_subject="u"))
+        except HTTPException as e:
+            assert not (
+                e.status_code == 400
+                and "tool_choice is not honored" in (e.detail or "")
+            )
+        except Exception:
+            # Any other failure downstream is fine; we only pin that the
+            # enable_tools+tool_choice guard does NOT fire when enable_tools is False.
+            pass

tests/test_pr5061_r2_image_dedup_last_user.py

@@ -0,0 +1,74 @@
+import os, sys
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+from models.inference import ChatMessage
+from routes.inference import _openai_messages_for_passthrough
+
+
+_TINY = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
+
+
+class _P:
+    def __init__(self, messages, image_base64=None):
+        self.messages = messages
+        self.image_base64 = image_base64
+
+
+def _img_url_part(b64=_TINY):
+    return {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
+
+
+def test_prior_user_image_does_not_block_new_base64():
+    p = _P(
+        messages=[
+            ChatMessage(role="user", content=[{"type": "text", "text": "q1"}, _img_url_part()]),
+            ChatMessage(role="assistant", content="a1"),
+            ChatMessage(role="user", content="q2 no inline"),
+        ],
+        image_base64=_TINY,
+    )
+    out = _openai_messages_for_passthrough(p)
+    # last user must receive spliced image
+    assert isinstance(out[-1]["content"], list)
+    assert any(x.get("type") == "image_url" for x in out[-1]["content"])
+
+
+def test_last_user_with_inline_image_skips_splice():
+    p = _P(
+        messages=[
+            ChatMessage(role="user", content="earlier"),
+            ChatMessage(role="user", content=[{"type": "text", "text": "last"}, _img_url_part()]),
+        ],
+        image_base64=_TINY,
+    )
+    out = _openai_messages_for_passthrough(p)
+    last_parts = out[-1]["content"]
+    assert sum(1 for x in last_parts if x.get("type") == "image_url") == 1
+
+
+def test_no_user_messages_appends_trailing_user():
+    p = _P(
+        messages=[ChatMessage(role="system", content="sys")],
+        image_base64=_TINY,
+    )
+    out = _openai_messages_for_passthrough(p)
+    assert out[-1]["role"] == "user"
+    assert any(x.get("type") == "image_url" for x in out[-1]["content"])
+
+
+def test_three_user_turns_only_last_receives_splice():
+    p = _P(
+        messages=[
+            ChatMessage(role="user", content="u1"),
+            ChatMessage(role="assistant", content="a1"),
+            ChatMessage(role="user", content="u2"),
+            ChatMessage(role="assistant", content="a2"),
+            ChatMessage(role="user", content="u3"),
+        ],
+        image_base64=_TINY,
+    )
+    out = _openai_messages_for_passthrough(p)
+    assert isinstance(out[0]["content"], str)
+    assert isinstance(out[2]["content"], str)
+    assert isinstance(out[4]["content"], list)

tests/test_pr5061_r2_nonstream_success_shape.py

@@ -0,0 +1,92 @@
+import asyncio
+import os, sys
+from unittest.mock import patch, MagicMock
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+from models.inference import ChatCompletionRequest
+
+
+class _Llama:
+    base_url = "http://127.0.0.1:0"
+    _api_key = None
+
+
+def _payload():
+    return ChatCompletionRequest(
+        messages=[{"role": "user", "content": "q"}],
+        tools=[{"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}],
+    )
+
+
+def _build_mock_client(status, payload_json):
+    class _Client:
+        def __init__(self, *a, **kw):
+            pass
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *a):
+            return False
+        async def post(self, *a, **kw):
+            m = MagicMock()
+            m.status_code = status
+            m.text = ""
+            m.json = lambda: payload_json
+            return m
+    return _Client
+
+
+def test_verbatim_json_body_returned():
+    from routes import inference as inf_mod
+
+    native = {
+        "id": "chatcmpl-foo",
+        "object": "chat.completion",
+        "model": "qwen-native",
+        "choices": [
+            {
+                "index": 0,
+                "finish_reason": "tool_calls",
+                "message": {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": [
+                        {
+                            "id": "call_1",
+                            "type": "function",
+                            "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'},
+                        }
+                    ],
+                },
+            }
+        ],
+        "usage": {"prompt_tokens": 42, "completion_tokens": 7, "total_tokens": 49},
+    }
+
+    with patch.object(inf_mod.httpx, "AsyncClient", _build_mock_client(200, native)):
+        resp = asyncio.run(inf_mod._openai_passthrough_non_streaming(_Llama(), _payload()))
+
+    import json
+    body = json.loads(resp.body.decode("utf-8"))
+    assert body == native  # verbatim
+    assert body["choices"][0]["finish_reason"] == "tool_calls"
+    assert body["usage"]["prompt_tokens"] == 42
+
+
+def test_preserves_native_id_and_model_fields():
+    from routes import inference as inf_mod
+
+    native = {
+        "id": "chatcmpl-native-xyz",
+        "model": "llama-native",
+        "choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "ok"}}],
+        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+    }
+    with patch.object(inf_mod.httpx, "AsyncClient", _build_mock_client(200, native)):
+        resp = asyncio.run(inf_mod._openai_passthrough_non_streaming(_Llama(), _payload()))
+
+    import json
+    body = json.loads(resp.body.decode("utf-8"))
+    assert body["id"] == "chatcmpl-native-xyz"
+    assert body["model"] == "llama-native"

tests/test_pr5061_r2_nonstream_upstream_log.py

@@ -0,0 +1,74 @@
+import asyncio
+import os, sys
+from unittest.mock import patch, MagicMock
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+import pytest
+from fastapi import HTTPException
+
+from models.inference import ChatCompletionRequest
+
+
+class _Llama:
+    base_url = "http://127.0.0.1:0"
+    _api_key = None
+
+
+def _payload():
+    return ChatCompletionRequest(
+        messages=[{"role": "user", "content": "q"}],
+        tools=[{"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}],
+    )
+
+
+def _mk_client(status, text):
+    class _Client:
+        def __init__(self, *a, **kw):
+            pass
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *a):
+            return False
+        async def post(self, *a, **kw):
+            m = MagicMock()
+            m.status_code = status
+            m.text = text
+            m.json = lambda: {}
+            return m
+    return _Client
+
+
+def test_nonstream_non_200_calls_logger_error_with_status_and_body():
+    from routes import inference as inf_mod
+
+    with patch.object(inf_mod.httpx, "AsyncClient", _mk_client(503, "backend overloaded detail")), \
+         patch.object(inf_mod.logger, "error") as mock_error:
+        with pytest.raises(HTTPException) as ei:
+            asyncio.run(inf_mod._openai_passthrough_non_streaming(_Llama(), _payload()))
+
+    assert ei.value.status_code == 503
+    mock_error.assert_called()
+    # Check any error call mentions upstream status and body
+    found = False
+    for call in mock_error.call_args_list:
+        msg = call.args[0] if call.args else ""
+        combined = f"{msg} {call.args} {call.kwargs}"
+        if "upstream error" in combined and "503" in combined and "backend overloaded" in combined:
+            found = True
+            break
+    assert found, f"expected upstream error log with status and body, got {mock_error.call_args_list}"
+
+
+def test_nonstream_200_does_not_call_logger_error():
+    from routes import inference as inf_mod
+
+    with patch.object(inf_mod.httpx, "AsyncClient", _mk_client(200, "{}")), \
+         patch.object(inf_mod.logger, "error") as mock_error:
+        asyncio.run(inf_mod._openai_passthrough_non_streaming(_Llama(), _payload()))
+
+    # no upstream-error log on 200
+    for call in mock_error.call_args_list:
+        msg = call.args[0] if call.args else ""
+        assert "upstream error" not in msg

tests/test_pr5061_r2_role_tool_nonpass_guard.py

@@ -0,0 +1,123 @@
+import asyncio
+import os, sys
+from unittest.mock import patch
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+import pytest
+from fastapi import HTTPException
+
+from models.inference import ChatCompletionRequest
+
+
+class _Llama:
+    def __init__(self, is_loaded=True, supports_tools=True, is_vision=False):
+        self.is_loaded = is_loaded
+        self.supports_tools = supports_tools
+        self.is_vision = is_vision
+        self._is_audio = False
+        self.model_identifier = "stub"
+        self.base_url = "http://127.0.0.1:0"
+        self._api_key = None
+
+
+class _Inf:
+    active_model_name = "hf"
+    models = {"hf": {}}
+
+
+class _Req:
+    async def is_disconnected(self):
+        return False
+
+
+async def _marker_stream(*a, **kw):
+    return ("passthrough_stream", None)
+
+
+async def _marker_nonstream(*a, **kw):
+    return ("passthrough_nonstream", None)
+
+
+async def _call(payload, llama):
+    from routes import inference as inf_mod
+    with patch.object(inf_mod, "get_llama_cpp_backend", return_value=llama), \
+         patch.object(inf_mod, "get_inference_backend", return_value=_Inf()), \
+         patch.object(inf_mod, "_openai_passthrough_stream", new=_marker_stream), \
+         patch.object(inf_mod, "_openai_passthrough_non_streaming", new=_marker_nonstream):
+        return await inf_mod.openai_chat_completions(payload, _Req(), current_subject="u")
+
+
+def test_role_tool_on_non_gguf_rejected():
+    payload = ChatCompletionRequest(
+        messages=[
+            {"role": "user", "content": "q"},
+            {"role": "tool", "tool_call_id": "c1", "content": "r"},
+        ],
+    )
+    llama = _Llama(is_loaded=False)
+    with pytest.raises(HTTPException) as ei:
+        asyncio.run(_call(payload, llama))
+    assert ei.value.status_code == 400
+    assert "role='tool'" in ei.value.detail or "tool" in ei.value.detail.lower()
+
+
+def test_role_tool_on_gguf_without_tool_support_rejected():
+    payload = ChatCompletionRequest(
+        messages=[
+            {"role": "user", "content": "q"},
+            {"role": "tool", "tool_call_id": "c1", "content": "r"},
+        ],
+    )
+    llama = _Llama(supports_tools=False)
+    with pytest.raises(HTTPException) as ei:
+        asyncio.run(_call(payload, llama))
+    assert ei.value.status_code == 400
+
+
+def test_role_tool_with_enable_tools_true_rejected():
+    payload = ChatCompletionRequest(
+        messages=[
+            {"role": "user", "content": "q"},
+            {"role": "tool", "tool_call_id": "c1", "content": "r"},
+        ],
+        enable_tools=True,
+    )
+    llama = _Llama()
+    with pytest.raises(HTTPException) as ei:
+        asyncio.run(_call(payload, llama))
+    assert ei.value.status_code == 400
+
+
+def test_assistant_tool_only_still_rejected_on_non_passthrough():
+    payload = ChatCompletionRequest(
+        messages=[
+            {"role": "user", "content": "q"},
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [
+                    {"id": "c1", "type": "function", "function": {"name": "f", "arguments": "{}"}}
+                ],
+            },
+        ],
+    )
+    llama = _Llama(supports_tools=False)
+    with pytest.raises(HTTPException) as ei:
+        asyncio.run(_call(payload, llama))
+    assert ei.value.status_code == 400
+
+
+def test_no_tool_messages_passes_through():
+    payload = ChatCompletionRequest(
+        messages=[{"role": "user", "content": "plain"}],
+    )
+    llama = _Llama(supports_tools=False)
+    # Should not raise the tool-shape guard; may raise later for non-GGUF/inf path,
+    # but specifically this guard must not fire.
+    try:
+        asyncio.run(_call(payload, llama))
+    except HTTPException as e:
+        assert "role='tool'" not in (e.detail or "")
+        assert "tool_calls-only" not in (e.detail or "")

tests/test_pr5061_r2_sse_done_on_error.py

@@ -0,0 +1,117 @@
+import asyncio
+import os, sys, threading
+from unittest.mock import patch, MagicMock, AsyncMock
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+import httpx
+
+from models.inference import ChatCompletionRequest
+
+
+class _Llama:
+    base_url = "http://127.0.0.1:0"
+    _api_key = None
+
+
+class _Req:
+    async def is_disconnected(self):
+        return False
+
+
+def _collect(stream_response):
+    async def _run():
+        out = []
+        async for chunk in stream_response.body_iterator:
+            out.append(chunk if isinstance(chunk, str) else chunk.decode("utf-8"))
+        return out
+    return asyncio.run(_run())
+
+
+def _make_payload():
+    return ChatCompletionRequest(
+        messages=[{"role": "user", "content": "q"}],
+        tools=[{"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}],
+        stream=True,
+    )
+
+
+def test_done_emitted_after_non_200_error():
+    from routes import inference as inf_mod
+
+    class _Resp:
+        status_code = 500
+        async def aread(self):
+            return b"server oops"
+        async def aclose(self):
+            pass
+
+    async def _send(req, stream):
+        return _Resp()
+
+    with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
+        inst = MagicMock()
+        inst.build_request = MagicMock(return_value=MagicMock())
+        inst.send = _send
+        inst.aclose = AsyncMock()
+        mock_cls.return_value = inst
+
+        resp = asyncio.run(
+            inf_mod._openai_passthrough_stream(
+                _Req(), threading.Event(), _Llama(), _make_payload()
+            )
+        )
+        chunks = _collect(resp)
+
+    assert any('"error"' in c for c in chunks)
+    assert any(c.strip() == "data: [DONE]" for c in chunks)
+
+
+def test_done_emitted_after_exception():
+    from routes import inference as inf_mod
+
+    async def _send(req, stream):
+        raise httpx.ConnectError("boom", request=httpx.Request("POST", "http://x"))
+
+    with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
+        inst = MagicMock()
+        inst.build_request = MagicMock(return_value=MagicMock())
+        inst.send = _send
+        inst.aclose = AsyncMock()
+        mock_cls.return_value = inst
+
+        resp = asyncio.run(
+            inf_mod._openai_passthrough_stream(
+                _Req(), threading.Event(), _Llama(), _make_payload()
+            )
+        )
+        chunks = _collect(resp)
+
+    assert any('"error"' in c for c in chunks)
+    assert any(c.strip() == "data: [DONE]" for c in chunks)
+
+
+def test_done_comes_after_error_chunk():
+    from routes import inference as inf_mod
+
+    async def _send(req, stream):
+        raise httpx.ReadError("reset", request=httpx.Request("POST", "http://x"))
+
+    with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
+        inst = MagicMock()
+        inst.build_request = MagicMock(return_value=MagicMock())
+        inst.send = _send
+        inst.aclose = AsyncMock()
+        mock_cls.return_value = inst
+
+        resp = asyncio.run(
+            inf_mod._openai_passthrough_stream(
+                _Req(), threading.Event(), _Llama(), _make_payload()
+            )
+        )
+        chunks = _collect(resp)
+
+    err_idx = next(i for i, c in enumerate(chunks) if '"error"' in c)
+    done_idx = next(i for i, c in enumerate(chunks) if c.strip() == "data: [DONE]")
+    assert done_idx > err_idx

tests/test_pr5061_r2_verbatim_stream_chunks.py

@@ -0,0 +1,122 @@
+import asyncio
+import os, sys, threading
+from unittest.mock import patch, MagicMock
+
+_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
+sys.path.insert(0, _backend)
+
+from models.inference import ChatCompletionRequest
+
+
+class _Llama:
+    base_url = "http://127.0.0.1:0"
+    _api_key = None
+
+
+class _Req:
+    async def is_disconnected(self):
+        return False
+
+
+def _payload():
+    return ChatCompletionRequest(
+        messages=[{"role": "user", "content": "q"}],
+        tools=[{"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}],
+        stream=True,
+    )
+
+
+class _FakeResp:
+    def __init__(self, lines):
+        self.status_code = 200
+        self._lines = list(lines)
+
+    def aiter_lines(self):
+        parent = self
+
+        class _It:
+            def __init__(self):
+                self._idx = 0
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                if self._idx >= len(parent._lines):
+                    raise StopAsyncIteration
+                v = parent._lines[self._idx]
+                self._idx += 1
+                return v
+
+            async def aclose(self):
+                pass
+
+        return _It()
+
+    async def aread(self):
+        return b""
+
+    async def aclose(self):
+        pass
+
+
+def _run_stream(fake_lines):
+    from routes import inference as inf_mod
+
+    async def _send(req, stream):
+        return _FakeResp(fake_lines)
+
+    async def _aclose_noop():
+        return None
+
+    with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
+        inst = MagicMock()
+        inst.build_request = MagicMock(return_value=MagicMock())
+        inst.send = _send
+        inst.aclose = _aclose_noop
+        mock_cls.return_value = inst
+
+        resp = asyncio.run(
+            inf_mod._openai_passthrough_stream(
+                _Req(), threading.Event(), _Llama(), _payload()
+            )
+        )
+
+        async def _collect():
+            return [
+                c if isinstance(c, str) else c.decode("utf-8")
+                async for c in resp.body_iterator
+            ]
+
+        return asyncio.run(_collect())
+
+
+def test_passthrough_relays_data_lines_verbatim():
+    chunks = _run_stream([
+        'data: {"id":"abc","choices":[{"delta":{"content":"hi"}}]}',
+        'data: [DONE]',
+    ])
+    assert any('"id":"abc"' in c for c in chunks)
+    assert any("data: [DONE]" in c for c in chunks)
+
+
+def test_passthrough_ignores_blank_and_non_data_lines():
+    chunks = _run_stream([
+        "",
+        ": heartbeat",
+        'data: {"x":1}',
+        'data: [DONE]',
+    ])
+    # Only data: lines propagate.
+    for c in chunks:
+        assert c.startswith("data: ") or c == ""
+    assert any('"x":1' in c for c in chunks)
+
+
+def test_passthrough_breaks_on_done():
+    chunks = _run_stream([
+        'data: {"a":1}',
+        'data: [DONE]',
+        'data: {"should_not_appear":true}',
+    ])
+    assert not any("should_not_appear" in c for c in chunks)