diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..fe3f07591 --- /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. 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. 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..3af64e529 --- /dev/null +++ b/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"]