fix(providers): fall back to Responses API for compatible endpoints

coworker/providers/openai_provider.py

@@ -1,11 +1,12 @@
 """OpenAI Chat Completions provider — the compat workhorse.
 
-Uses the OpenAI Python SDK `chat.completions` API only, which is what the entire
+Uses the OpenAI Python SDK `chat.completions` API, which is what the entire
 OpenAI-compatible world implements: the compat vendors (DeepSeek, Z AI, Kimi, …),
 resellers, Ollama, custom endpoints (Azure OpenAI, vLLM), and the Bedrock/Vertex MaaS
 paths. Native OpenAI models (the `openai` provider with no custom endpoint) route to
-`openai_responses.OpenAIResponsesProvider` instead — Chat Completions rejects function
-tools combined with reasoning on GPT-5.6+, so reasoning + tools needs `/v1/responses`.
+`openai_responses.OpenAIResponsesProvider` instead. A custom endpoint stays on Chat
+Completions unless it explicitly rejects a tool-enabled request and directs us to
+`/v1/responses`; that model then uses the Responses API for subsequent turns.
 """
 
 from __future__ import annotations
@@ -53,6 +54,20 @@ def resolve_api_key(secrets: Any = None) -> Optional[str]:
 _EFFORT_ERROR = "function tools with reasoning_effort are not supported"
 
 
+def _requires_responses_api(exc: Exception) -> bool:
+    """Whether a Chat Completions error explicitly directs this request to Responses.
+
+    OpenAI-compatible servers vary widely, so a custom endpoint must remain on the
+    Chat Completions path unless it advertises this exact compatibility requirement.
+    """
+    message = str(exc).lower()
+    return (
+        _EFFORT_ERROR in message
+        and "/v1/chat/completions" in message
+        and "/v1/responses" in message
+    )
+
+
 def _pin_reasoning_effort(kwargs: dict[str, Any]) -> None:
     if kwargs.get("tools") and str(kwargs.get("model", "")).startswith("gpt-5.6"):
         kwargs.setdefault("reasoning_effort", "none")
@@ -146,6 +161,8 @@ def __init__(
         self._base_url = base_url
         self._secrets = secrets
         self.default_model = default_model
+        self._responses_models: set[str] = set()
+        self._responses_fallback: Any = None
 
     def _ensure_client(self) -> Any:
         if self._client is None:
@@ -164,13 +181,64 @@ def _ensure_client(self) -> Any:
             self._client = OpenAI(**kwargs)
         return self._client
 
+    def _responses_provider(self) -> Any:
+        """Build the lazy Responses fallback around this provider's existing SDK client."""
+        if self._responses_fallback is None:
+            # Imported lazily to avoid the response provider's resolve_api_key import
+            # forming a module cycle during provider registry initialization.
+            from .openai_responses import OpenAIResponsesProvider
+
+            self._responses_fallback = OpenAIResponsesProvider(
+                client=self._ensure_client()
+            )
+        return self._responses_fallback
+
+    def _complete_with_responses(
+        self,
+        *,
+        model: str,
+        messages: list[dict[str, Any]],
+        tools: Optional[list[dict[str, Any]]],
+        settings: dict[str, Any],
+    ) -> AssistantTurn:
+        turn = self._responses_provider().complete(
+            model=model, messages=messages, tools=tools, **settings
+        )
+        # Cache only after a complete Responses request succeeds. A gateway that
+        # advertises the route but does not implement it should keep its old behavior.
+        self._responses_models.add(model)
+        return turn
+
     def complete(
         self,
         *,
         model: str,
         messages: list[dict[str, Any]],
         tools: Optional[list[dict[str, Any]]] = None,
         **settings: Any,
+    ) -> AssistantTurn:
+        if model in self._responses_models:
+            return self._complete_with_responses(
+                model=model, messages=messages, tools=tools, settings=settings
+            )
+        try:
+            return self._complete_with_chat(
+                model=model, messages=messages, tools=tools, settings=settings
+            )
+        except Exception as exc:
+            if not tools or not _requires_responses_api(exc):
+                raise
+            return self._complete_with_responses(
+                model=model, messages=messages, tools=tools, settings=settings
+            )
+
+    def _complete_with_chat(
+        self,
+        *,
+        model: str,
+        messages: list[dict[str, Any]],
+        tools: Optional[list[dict[str, Any]]],
+        settings: dict[str, Any],
     ) -> AssistantTurn:
         kwargs: dict[str, Any] = {
             "model": model,
@@ -215,6 +283,51 @@ def stream(
         messages: list[dict[str, Any]],
         tools: Optional[list[dict[str, Any]]] = None,
         **settings: Any,
+    ):
+        if model in self._responses_models:
+            yield from self._stream_with_responses(
+                model=model, messages=messages, tools=tools, settings=settings
+            )
+            return
+
+        yielded = False
+        try:
+            for chunk in self._stream_with_chat(
+                model=model, messages=messages, tools=tools, settings=settings
+            ):
+                yielded = True
+                yield chunk
+        except Exception as exc:
+            # Falling back after emitting a Chat Completions delta would duplicate
+            # visible output. The route negotiation error is returned before any
+            # stream event, so only retry when nothing reached the caller.
+            if yielded or not tools or not _requires_responses_api(exc):
+                raise
+            yield from self._stream_with_responses(
+                model=model, messages=messages, tools=tools, settings=settings
+            )
+
+    def _stream_with_responses(
+        self,
+        *,
+        model: str,
+        messages: list[dict[str, Any]],
+        tools: Optional[list[dict[str, Any]]],
+        settings: dict[str, Any],
+    ):
+        for chunk in self._responses_provider().stream(
+            model=model, messages=messages, tools=tools, **settings
+        ):
+            yield chunk
+        self._responses_models.add(model)
+
+    def _stream_with_chat(
+        self,
+        *,
+        model: str,
+        messages: list[dict[str, Any]],
+        tools: Optional[list[dict[str, Any]]],
+        settings: dict[str, Any],
     ):
         kwargs: dict[str, Any] = {
             "model": model,

tests/test_providers.py

@@ -186,6 +186,145 @@ def test_effort_400_from_an_unpinned_model_retries_once_at_none():
     assert out[-1].turn.text == "ok" and len(client.chat.completions.calls) == 4
 
 
+_RESPONSES_REQUIRED_ERROR = (
+    "Error code: 400 - {'error': {'message': \"Function tools with reasoning_effort "
+    "are not supported for gpt-5.6-terra in /v1/chat/completions. Please use "
+    "/v1/responses instead.\", 'type': 'invalid_request_error'}}"
+)
+
+
+def _responses_result(text: str):
+    return SimpleNamespace(
+        output=[
+            SimpleNamespace(
+                type="message",
+                content=[SimpleNamespace(type="output_text", text=text)],
+            )
+        ]
+    )
+
+
+def _responses_tool_result():
+    return SimpleNamespace(
+        output=[
+            SimpleNamespace(
+                type="function_call",
+                call_id="response-call-1",
+                name="read_file",
+                arguments='{"path": "a.py"}',
+            )
+        ]
+    )
+
+
+class _ResponsesRequiredCompletions:
+    def __init__(self):
+        self.calls: list[dict] = []
+
+    def create(self, **kwargs):
+        self.calls.append(kwargs)
+        raise RuntimeError(_RESPONSES_REQUIRED_ERROR)
+
+
+class _FallbackResponses:
+    def __init__(self, response, events=()):
+        self._response = response
+        self._events = events
+        self.calls: list[dict] = []
+
+    def create(self, **kwargs):
+        self.calls.append(kwargs)
+        return iter(self._events) if kwargs.get("stream") else self._response
+
+
+class _FallbackClient:
+    def __init__(self, response, events=()):
+        self.chat = SimpleNamespace(completions=_ResponsesRequiredCompletions())
+        self.responses = _FallbackResponses(response, events)
+
+
+def test_custom_endpoint_falls_back_to_responses_and_caches_the_model():
+    client = _FallbackClient(_responses_result("from responses"))
+    provider = OpenAIProvider(client=client)
+
+    turn = provider.complete(
+        model="gpt-5.6-terra",
+        messages=[{"role": "user", "content": "read the file"}],
+        tools=_TOOLS,
+        temperature=0.2,
+    )
+
+    assert turn.text == "from responses"
+    assert len(client.chat.completions.calls) == 1
+    request = client.responses.calls[0]
+    assert request["input"] == [{"role": "user", "content": "read the file"}]
+    assert request["tools"] == [{"type": "function", "name": "read_file"}]
+    assert request["temperature"] == 0.2
+    assert "reasoning_effort" not in request
+
+    # The negotiated protocol is per-model, so the next turn avoids a second 400.
+    provider.complete(model="gpt-5.6-terra", messages=[])
+    assert len(client.chat.completions.calls) == 1
+    assert len(client.responses.calls) == 2
+
+
+def test_custom_endpoint_fallback_preserves_response_call_ids():
+    client = _FallbackClient(_responses_tool_result())
+    provider = OpenAIProvider(client=client)
+
+    turn = provider.complete(model="gpt-5.6-terra", messages=[], tools=_TOOLS)
+
+    assert turn.tool_calls == [
+        ToolCall(
+            id="response-call-1", name="read_file", arguments={"path": "a.py"}
+        )
+    ]
+    assert turn.finish_reason == "tool_calls"
+
+
+def test_custom_endpoint_stream_falls_back_to_responses_before_emitting_deltas():
+    result = _responses_result("from responses")
+    events = [
+        SimpleNamespace(type="response.output_text.delta", delta="from responses"),
+        SimpleNamespace(type="response.completed", response=result),
+    ]
+    client = _FallbackClient(result, events)
+    provider = OpenAIProvider(client=client)
+
+    chunks = list(
+        provider.stream(model="gpt-5.6-terra", messages=[], tools=_TOOLS)
+    )
+
+    assert [chunk.text_delta for chunk in chunks if chunk.text_delta] == ["from responses"]
+    assert chunks[-1].turn.text == "from responses"
+    assert len(client.chat.completions.calls) == 1
+    assert client.responses.calls[0]["stream"] is True
+
+    list(provider.stream(model="gpt-5.6-terra", messages=[]))
+    assert len(client.chat.completions.calls) == 1
+    assert len(client.responses.calls) == 2
+
+
+def test_custom_endpoint_only_falls_back_when_it_advertises_responses():
+    client = _FallbackClient(_responses_result("unused"))
+
+    class _OtherCompletions:
+        def create(self, **kwargs):
+            raise RuntimeError(
+                "Function tools with reasoning_effort are not supported for this model"
+            )
+
+    client.chat.completions = _OtherCompletions()
+    provider = OpenAIProvider(client=client)
+
+    try:
+        provider.complete(model="gpt-5.6-terra", messages=[], tools=_TOOLS)
+        raise AssertionError("should have raised")
+    except RuntimeError as exc:
+        assert "reasoning_effort" in str(exc)
+    assert client.responses.calls == []
+
+
 def test_max_tokens_rejection_retries_as_max_completion_tokens():
     """Reasoning-routed models 400 on max_tokens (want max_completion_tokens); compat
     servers know only max_tokens — so the swap happens on rejection, never up front.