From f77b9b6e9109a7d339b03c6dd73c90f87e9b42c7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 18 Aug 2026 16:44:33 -0700 Subject: [PATCH 1/3] fix: fail over passthrough requests across providers --- CHANGELOG.md | 9 ++ contextual_orchestrator/__init__.py | 3 +- contextual_orchestrator/__main__.py | 3 +- .../resilient_orchestrator.py | 45 ++++++++ tests/test_passthrough_failover.py | 102 ++++++++++++++++++ 5 files changed, 160 insertions(+), 2 deletions(-) create mode 100644 CHANGELOG.md create mode 100644 contextual_orchestrator/resilient_orchestrator.py create mode 100644 tests/test_passthrough_failover.py diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..e43552e35 --- /dev/null +++ b/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, so a rate-limited primary model can hand the unchanged request to the next eligible model while preserving the full provider response shape. diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py index 70dbd71c6..d6e1752b4 100644 --- a/contextual_orchestrator/__init__.py +++ b/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__ = [ diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index 5f68c3b74..6a74a4340 100644 --- a/contextual_orchestrator/__main__.py +++ b/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 diff --git a/contextual_orchestrator/resilient_orchestrator.py b/contextual_orchestrator/resilient_orchestrator.py new file mode 100644 index 000000000..a8891b205 --- /dev/null +++ b/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 diff --git a/tests/test_passthrough_failover.py b/tests/test_passthrough_failover.py new file mode 100644 index 000000000..a017eec61 --- /dev/null +++ b/tests/test_passthrough_failover.py @@ -0,0 +1,102 @@ +"""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 hand the unchanged request to 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 "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"] From 260b311dda856c93dcc86c692ee492f175f441c4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 18 Aug 2026 16:55:04 -0700 Subject: [PATCH 2/3] docs: clarify passthrough fallback mutation boundary --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e43552e35..fe3f07591 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,4 +6,4 @@ All notable changes introduced after the initial repository bootstrap are record ### Fixed -- Added cross-agent provider fallback for OpenAI-compatible passthrough requests, including tool-calling and Responses API traffic, so a rate-limited primary model can hand the unchanged request to the next eligible model while preserving the full provider response shape. +- 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. From 46e964594a2403d918895b0bb272bdfc3f0d0682 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 18 Aug 2026 16:55:24 -0700 Subject: [PATCH 3/3] test: cover responses passthrough fallback --- tests/test_passthrough_failover.py | 42 +++++++++++++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/tests/test_passthrough_failover.py b/tests/test_passthrough_failover.py index a017eec61..3af64e529 100644 --- a/tests/test_passthrough_failover.py +++ b/tests/test_passthrough_failover.py @@ -50,7 +50,7 @@ def _orchestrator(client: SequencedProxyClient) -> TaskOrchestrator: def test_proxy_completion_fails_over_after_primary_rate_limit_and_preserves_tools() -> None: - """A rate-limited tool-call provider must hand the unchanged request to fallback.""" + """A rate-limited tool-call provider must preserve fields for fallback.""" client = SequencedProxyClient( { "primary_agent": RuntimeError("429 rate limit"), @@ -77,6 +77,46 @@ def test_proxy_completion_fails_over_after_primary_rate_limit_and_preserves_tool 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