fix: fail over tool-calling passthrough across model providers

CHANGELOG.md

@@ -0,0 +1,9 @@
+# Changelog
+
+All notable changes introduced after the initial repository bootstrap are recorded here.
+
+## [Unreleased]
+
+### Fixed
+
+- Added cross-agent provider fallback for OpenAI-compatible passthrough requests, including tool-calling and Responses API traffic. Upstream request fields are preserved while each candidate receives its own model and non-streaming is enforced so a rate-limited primary can hand the request to the next eligible model without changing the provider response shape.

contextual_orchestrator/__init__.py

@@ -37,7 +37,8 @@
 from .cost_router import CostRoutingCoordinator
 from .credentials import NotConfigured, get_credential, register_credential
 from .kv_config import InMemoryConfigStore, get_config_store
-from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents
+from .orchestrator import ModelAgent, WorkflowStep, load_agents
+from .resilient_orchestrator import TaskOrchestrator
 from .token_counting import HeuristicTokenCounter, build_token_counter
 
 __all__ = [

contextual_orchestrator/__main__.py

@@ -8,7 +8,8 @@
 import sys
 
 from .credentials import register_credential
-from .orchestrator import ModelClient, TaskOrchestrator, load_agents
+from .orchestrator import ModelClient, load_agents
+from .resilient_orchestrator import TaskOrchestrator
 from .server import SecurityConfig, serve
 
 

contextual_orchestrator/resilient_orchestrator.py

@@ -0,0 +1,45 @@
+"""Resilient orchestration entrypoint for provider-neutral passthrough requests."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .orchestrator import TaskOrchestrator as BaseTaskOrchestrator
+from .orchestrator import _coerce_input_text
+
+
+class TaskOrchestrator(BaseTaskOrchestrator):
+    """Add cross-agent failover to full-shape OpenAI passthrough requests."""
+
+    def proxy_completion(
+        self, body: dict[str, Any], *, endpoint: str = "chat/completions"
+    ) -> dict[str, Any]:
+        """Preserve provider response shape while failing over across ranked agents."""
+        messages = body.get("messages")
+        if isinstance(messages, list):
+            text = self._latest_user_text(messages)
+        else:
+            text = _coerce_input_text(body.get("input"))
+        primary = self._select_agent(text, "worker")
+        upstream = {
+            key: value
+            for key, value in body.items()
+            if key not in self._ORCHESTRATION_ONLY_KEYS
+        }
+        candidates = self._failover_candidates(primary, text, "worker")
+        last_error: Exception | None = None
+        for agent in candidates:
+            candidate_request = dict(upstream)
+            candidate_request["model"] = agent.model
+            candidate_request["stream"] = False
+            try:
+                response = self.client.proxy_send(agent, endpoint, candidate_request)
+            except Exception as exc:  # noqa: BLE001 - one provider failure routes to the next
+                last_error = exc
+                self._record_failure(agent.id)
+                continue
+            self._record_success(agent.id)
+            return response
+        raise RuntimeError(
+            f"all {len(candidates)} candidate agents failed for passthrough endpoint={endpoint}"
+        ) from last_error

tests/test_passthrough_failover.py

@@ -0,0 +1,142 @@
+"""Regression tests for provider failover on full-shape OpenAI passthrough."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+import pytest
+
+from contextual_orchestrator import ModelAgent, TaskOrchestrator
+
+
+class SequencedProxyClient:
+    """Return configured raw responses or raise configured errors by agent ID."""
+
+    def __init__(self, outcomes: dict[str, dict[str, Any] | BaseException]) -> None:
+        self.outcomes = outcomes
+        self.calls: list[tuple[str, str, dict[str, Any]]] = []
+
+    def proxy_send(
+        self, agent: ModelAgent, endpoint: str, payload: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Record one call and return or raise the configured outcome."""
+        self.calls.append((agent.id, endpoint, deepcopy(payload)))
+        outcome = self.outcomes[agent.id]
+        if isinstance(outcome, BaseException):
+            raise outcome
+        return deepcopy(outcome)
+
+
+def _orchestrator(client: SequencedProxyClient) -> TaskOrchestrator:
+    """Build a deterministic two-provider passthrough pool."""
+    return TaskOrchestrator(
+        agents=[
+            ModelAgent(
+                "primary_agent",
+                "primary-model",
+                tags=("coding", "implementation", "security", "review"),
+                priority=10,
+            ),
+            ModelAgent(
+                "fallback_agent",
+                "fallback-model",
+                tags=("coding", "implementation", "security", "review"),
+                priority=1,
+            ),
+        ],
+        client=client,
+    )
+
+
+def test_proxy_completion_fails_over_after_primary_rate_limit_and_preserves_tools() -> None:
+    """A rate-limited tool-call provider must preserve fields for fallback."""
+    client = SequencedProxyClient(
+        {
+            "primary_agent": RuntimeError("429 rate limit"),
+            "fallback_agent": {
+                "object": "chat.completion",
+                "model": "fallback-model",
+                "choices": [],
+            },
+        }
+    )
+    orchestrator = _orchestrator(client)
+    tools = [{"type": "function", "function": {"name": "inspect", "parameters": {}}}]
+    body = {
+        "messages": [{"role": "user", "content": "review this security-sensitive code"}],
+        "tools": tools,
+        "mode": "auto",
+    }
+    original = deepcopy(body)
+
+    result = orchestrator.proxy_completion(body)
+
+    assert result["model"] == "fallback-model"
+    assert [call[0] for call in client.calls] == ["primary_agent", "fallback_agent"]
+    assert client.calls[0][2]["model"] == "primary-model"
+    assert client.calls[1][2]["model"] == "fallback-model"
+    assert client.calls[1][2]["tools"] == tools
+    assert client.calls[1][2]["stream"] is False
+    assert "mode" not in client.calls[1][2]
+    assert body == original
+
+
+def test_proxy_completion_fails_over_responses_input_and_forces_non_streaming() -> None:
+    """Responses API input must use the same fallback and non-streaming contract."""
+    client = SequencedProxyClient(
+        {
+            "primary_agent": RuntimeError("429 rate limit"),
+            "fallback_agent": {
+                "object": "response",
+                "model": "fallback-model",
+                "output": [],
+            },
+        }
+    )
+    orchestrator = _orchestrator(client)
+    body = {
+        "input": [
+            {
+                "role": "user",
+                "content": [{"type": "input_text", "text": "inspect this change"}],
+            }
+        ],
+        "tools": [{"type": "function", "name": "inspect", "parameters": {}}],
+        "stream": True,
+        "mode": "auto",
+    }
+    original = deepcopy(body)
+
+    result = orchestrator.proxy_completion(body, endpoint="responses")
+
+    assert result["model"] == "fallback-model"
+    assert [call[0] for call in client.calls] == ["primary_agent", "fallback_agent"]
+    assert [call[1] for call in client.calls] == ["responses", "responses"]
+    assert client.calls[1][2]["input"] == original["input"]
+    assert client.calls[1][2]["tools"] == original["tools"]
+    assert client.calls[1][2]["model"] == "fallback-model"
+    assert client.calls[1][2]["stream"] is False
+    assert "mode" not in client.calls[1][2]
+    assert body == original
+
+
+def test_proxy_completion_reports_all_candidate_failures() -> None:
+    """The gateway must fail closed after every eligible passthrough provider fails."""
+    first = RuntimeError("primary unavailable")
+    final = RuntimeError("fallback unavailable")
+    client = SequencedProxyClient(
+        {"primary_agent": first, "fallback_agent": final}
+    )
+    orchestrator = _orchestrator(client)
+
+    with pytest.raises(RuntimeError, match="all 2 candidate agents failed") as caught:
+        orchestrator.proxy_completion(
+            {
+                "messages": [{"role": "user", "content": "review code"}],
+                "tools": [],
+            }
+        )
+
+    assert caught.value.__cause__ is final
+    assert [call[0] for call in client.calls] == ["primary_agent", "fallback_agent"]