diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index fc91c5256..fcaa7d81d 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -179,6 +179,13 @@ def _request_endpoint_partition() -> str: MAX_LOCAL_CONCURRENCY = 64 MAX_PROVIDER_RESPONSE_BYTES = 8 * 1024 * 1024 _PASSTHROUGH_UNAVAILABLE_STATUS = frozenset({404, 410, 413}) +# RFC 9110 section 9.2.2 permits an automatic non-idempotent retry only when +# the client knows the original request was never applied. These statuses +# describe a request rejection; generic origin/gateway failures (500/502/504) +# and the non-standard 529 do not provide that evidence and stay sticky. +_PASSTHROUGH_REJECTED_STATUS = _PASSTHROUGH_UNAVAILABLE_STATUS | frozenset( + {408, 409, 425, 429, 503} +) _PROVIDER_ERROR_CHAIN_LIMIT = 8 _PROVIDER_TOOL_DESCRIPTION_LIMIT_MESSAGE = ( "each tool.function.description must be at most 1024 characters" @@ -2067,13 +2074,11 @@ def _is_passthrough_failover_error(exc: BaseException) -> bool: return False seen.add(id(current)) if isinstance(current, ProviderUpstreamError): - if current.provider_status in ( - _PASSTHROUGH_UNAVAILABLE_STATUS | TRANSIENT_HTTP_STATUS - ): + if current.provider_status in _PASSTHROUGH_REJECTED_STATUS: return True if ( isinstance(current, urllib.error.HTTPError) - and current.code in (_PASSTHROUGH_UNAVAILABLE_STATUS | TRANSIENT_HTTP_STATUS) + and current.code in _PASSTHROUGH_REJECTED_STATUS ): return True if ( @@ -2203,6 +2208,54 @@ def _is_capability_mismatch_failover_error(exc: BaseException) -> bool: return False +def _passthrough_failure_phase(exc: ProviderUpstreamError) -> str: + """Map one classified passthrough failure onto a bounded lifecycle phase.""" + if exc.error_code in {"tls_failure", "tls_verification_failed"}: + return "connecting" + if exc.error_code == "provider_connection_error": + return "transport" + if exc.error_code == "request_too_large": + return "request_validation" + return "provider_response" + + +def _passthrough_attempt_record( + exc: ProviderUpstreamError, + *, + provider_name: str, + attempt_number: int, + failover_decision: str, +) -> dict[str, Any]: + """Return the public attempt receipt for one classified passthrough failure.""" + return { + "agent_id": exc.agent_id, + "model": exc.model, + "provider_name": provider_name, + "attempt_number": attempt_number, + "error_code": exc.error_code, + "client_status": exc.client_status, + "provider_status": exc.provider_status, + "retryable": exc.retryable, + "transport": exc.transport, + "phase": _passthrough_failure_phase(exc), + "failover_decision": failover_decision, + } + + +def _set_passthrough_attempt_evidence( + exc: ProviderUpstreamError, + *, + selected_candidate_ids: list[str], + attempts: list[dict[str, Any]], + terminal_reason: str, +) -> ProviderUpstreamError: + """Attach bounded request-scoped passthrough evidence to one final error.""" + exc.selected_candidate_ids = tuple(selected_candidate_ids) + exc.attempts = tuple(dict(item) for item in attempts) + exc.terminal_reason = terminal_reason + return exc + + def _assistant_message_extras(data: dict[str, Any]) -> dict[str, Any] | None: """Keep provider tool_calls/finish_reason beside the text-only chat() result.""" choices = data.get("choices") @@ -3460,7 +3513,15 @@ def _proxy_send( ) if agent.base_url.startswith("mock://"): return self._mock_raw(agent, normalized_endpoint, payload) - destination = self._validate_provider(agent) # pragma: no cover + try: + destination = self._validate_provider(agent) # pragma: no cover + except ProviderUpstreamError as exc: + raise classify_provider_failure( + exc, + agent_id=agent.id, + model=agent.model, + transport="passthrough", + ) from None parsed_provider = urlparse(agent.base_url) operation_name = { "chat/completions": "chat", @@ -3883,7 +3944,16 @@ def _validate_provider(self, agent: ModelAgent) -> ProviderDestination: raise RuntimeError(f"{agent.id} base_url must not contain credentials, query data, or fragments") hostname = parsed.hostname.lower() if self.allowed_provider_hosts and hostname not in self.allowed_provider_hosts: - raise RuntimeError(f"{agent.id} provider host is not allowlisted") + raise ProviderUpstreamError( + agent_id=agent.id, + model=agent.model, + error_code="provider_connection_error", + message=f"{agent.id} provider host is not allowlisted", + client_status=502, + provider_status=None, + retryable=False, + transport="chat", + ) addresses = self._resolve_addresses(hostname, parsed.port or 443) for _family, sockaddr in addresses: ip_address = ipaddress.ip_address(sockaddr[0]) @@ -6052,7 +6122,7 @@ def proxy_completion( self._record_failure(agent.id) if agent.group_name: self._group_router.observe_failure(agent.id) - raise ProviderUpstreamError( + unknown = ProviderUpstreamError( agent_id=agent.id, model=agent.model, error_code=PROVIDER_OUTCOME_UNKNOWN_CODE, @@ -6063,10 +6133,37 @@ def proxy_completion( client_status=502, retryable=False, transport="passthrough", + ) + raise _set_passthrough_attempt_evidence( + unknown, + selected_candidate_ids=[agent.id], + attempts=[ + _passthrough_attempt_record( + unknown, + provider_name=agent.provider_name.strip() or "unreported", + attempt_number=1, + failover_decision="sticky_candidate_failure", + ) + ], + terminal_reason="terminal_provider_failure", ) from None request_too_large = _is_request_too_large_error(exc) if measured and not request_too_large: self._group_router.observe_failure(agent.id) + if isinstance(exc, ProviderUpstreamError): + raise _set_passthrough_attempt_evidence( + exc, + selected_candidate_ids=[agent.id], + attempts=[ + _passthrough_attempt_record( + exc, + provider_name=agent.provider_name.strip() or "unreported", + attempt_number=1, + failover_decision="sticky_candidate_failure", + ) + ], + terminal_reason="terminal_provider_failure", + ) from None raise if measured: self._group_router.observe_success( @@ -6152,6 +6249,8 @@ def proxy_completion( candidates, tool_loop_evidence = self._apply_tool_loop_route(candidates, messages) last_failure: tuple[Exception, ModelAgent] | None = None every_failure_was_request_too_large = True + selected_candidate_ids = [candidate.id for candidate in candidates] + attempt_receipts: list[dict[str, Any]] = [] # Rate-limit-storm admission (evidence: noema run 34758641142, strix # run 34758679736 -- every candidate returned 429 within ~50ms; # ContextualWisdomLab/.github#2148, #2165). ``wait_deadline`` is @@ -6192,7 +6291,19 @@ def proxy_completion( record_initial_selection([candidate.id], "automatic_proxy") result = send_once(candidate, endpoint, candidate_payload) except Exception as exc: # noqa: BLE001 - provider trust boundary - if not _is_passthrough_failover_error(exc): + classified = classify_provider_failure( + exc, + agent_id=candidate.id, + model=candidate.model, + transport="passthrough", + ) + failover_eligible = _is_passthrough_failover_error(exc) + prior_attempted = {item["agent_id"] for item in attempt_receipts} + has_remaining_candidates = any( + other.id != candidate.id and other.id not in prior_attempted + for other in candidates + ) + if not failover_eligible: if _is_ambiguous_passthrough_transport_failure(exc): # This candidate's own outcome is unknown (the timeout # or reset may follow provider acceptance), so it is @@ -6216,28 +6327,74 @@ def proxy_completion( self._record_failure(candidate.id) if candidate.group_name: self._group_router.observe_failure(candidate.id) + attempt_receipts.append( + _passthrough_attempt_record( + classified, + provider_name=( + candidate.provider_name.strip() or "unreported" + ), + attempt_number=len(attempt_receipts) + 1, + failover_decision=( + "advance_to_next_candidate" + if virtual_selector and has_remaining_candidates + else ( + "eligible_candidates_exhausted" + if virtual_selector + else "sticky_candidate_failure" + ) + ), + ) + ) if virtual_selector: - last_failure = (exc, candidate) + last_failure = (classified, candidate) every_failure_was_request_too_large = False continue - raise ProviderUpstreamError( - agent_id=candidate.id, - model=candidate.model, - error_code=PROVIDER_OUTCOME_UNKNOWN_CODE, - message="the provider request outcome is unknown; automatic replay is unsafe", - client_status=502, - retryable=False, - transport="passthrough", - ) from None - if isinstance(exc, (urllib.error.HTTPError, ProviderUpstreamError)): - raise classify_provider_failure( - exc, - agent_id=candidate.id, - model=candidate.model, - transport="passthrough", + raise _set_passthrough_attempt_evidence( + ProviderUpstreamError( + agent_id=candidate.id, + model=candidate.model, + error_code=PROVIDER_OUTCOME_UNKNOWN_CODE, + message=( + "the provider request outcome is unknown; " + "automatic replay is unsafe" + ), + client_status=502, + retryable=False, + transport="passthrough", + ), + selected_candidate_ids=selected_candidate_ids, + attempts=attempt_receipts, + terminal_reason="terminal_provider_failure", ) from None - raise - last_failure = (exc, candidate) + attempt_receipts.append( + _passthrough_attempt_record( + classified, + provider_name=( + candidate.provider_name.strip() or "unreported" + ), + attempt_number=len(attempt_receipts) + 1, + failover_decision="sticky_candidate_failure", + ) + ) + raise _set_passthrough_attempt_evidence( + classified, + selected_candidate_ids=selected_candidate_ids, + attempts=attempt_receipts, + terminal_reason="terminal_provider_failure", + ) from None + attempt_receipts.append( + _passthrough_attempt_record( + classified, + provider_name=candidate.provider_name.strip() or "unreported", + attempt_number=len(attempt_receipts) + 1, + failover_decision=( + "advance_to_next_candidate" + if has_remaining_candidates + else "eligible_candidates_exhausted" + ), + ) + ) + last_failure = (classified, candidate) request_too_large = _is_request_too_large_error(exc) every_failure_was_request_too_large = ( every_failure_was_request_too_large @@ -6320,16 +6477,26 @@ def proxy_completion( ): break if last_failure is not None and every_failure_was_request_too_large: - raise ProviderRequestTooLargeError( - "request body exceeds every eligible provider limit" + raise _set_passthrough_attempt_evidence( + ProviderRequestTooLargeError( + "request body exceeds every eligible provider limit" + ), + selected_candidate_ids=selected_candidate_ids, + attempts=attempt_receipts, + terminal_reason="request_too_large_exhausted", ) from None if last_failure is not None: last_error, failed_candidate = last_failure - raise classify_provider_failure( - last_error, - agent_id=failed_candidate.id, - model=failed_candidate.model, - transport="passthrough", + raise _set_passthrough_attempt_evidence( + classify_provider_failure( + last_error, + agent_id=failed_candidate.id, + model=failed_candidate.model, + transport="passthrough", + ), + selected_candidate_ids=selected_candidate_ids, + attempts=attempt_receipts, + terminal_reason="eligible_candidates_exhausted", ) from None raise RuntimeError("passthrough has no eligible provider candidate") @@ -19656,4 +19823,3 @@ def sse_stream_body(chunks: list[dict[str, Any]]) -> str: frames = [f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n" for chunk in chunks] frames.append("data: [DONE]\n\n") return "".join(frames) - diff --git a/contextual_orchestrator/provider_errors.py b/contextual_orchestrator/provider_errors.py index 26a1bdfb4..3e6c8d4d3 100644 --- a/contextual_orchestrator/provider_errors.py +++ b/contextual_orchestrator/provider_errors.py @@ -343,6 +343,15 @@ def detail(self) -> dict[str, Any]: "retryable": self.retryable, "transport": self.transport, } + selected_candidate_ids = getattr(self, "selected_candidate_ids", None) + if isinstance(selected_candidate_ids, (list, tuple)) and selected_candidate_ids: + payload["selected_candidate_ids"] = list(selected_candidate_ids) + attempts = getattr(self, "attempts", None) + if isinstance(attempts, (list, tuple)) and attempts: + payload["attempts"] = [dict(item) for item in attempts if isinstance(item, dict)] + terminal_reason = getattr(self, "terminal_reason", None) + if isinstance(terminal_reason, str) and terminal_reason: + payload["terminal_reason"] = terminal_reason payload.update(self.extra_detail) return payload diff --git a/docs/doctoring/2026-09-04-passthrough-failure-evidence.md b/docs/doctoring/2026-09-04-passthrough-failure-evidence.md new file mode 100644 index 000000000..1ca01f1bf --- /dev/null +++ b/docs/doctoring/2026-09-04-passthrough-failure-evidence.md @@ -0,0 +1,112 @@ +# PR #1049 passthrough failure evidence + +Relocated on 2026-09-10 from the PR's cumulative baseline and changelog to +preserve current main's records without changing any production or test blob. +The dated observations and earlier validation below are historical evidence, +not a fresh integrated-head, release, or deployment claim. The complete source +history remains on PR #1049, including head +`e2641c16a82816e15f12013efad7fe50e94a3333`. + +## Changelog fragment — Unreleased + +- Passthrough attempt receipts no longer infer public provider labels from + endpoint hostnames. Ambiguous timeout/connection failures now report the + neutral `transport` phase and remain sticky even for `orchestrator/free`, + preventing duplicate completion and unreported upstream usage. Generic + HTTP 500/502/504 and non-standard 529 responses are sticky as well: an HTTP + retry classification alone does not prove that a non-idempotent completion + request was never applied. + +## 2026-09-04 Autonomous Commercialization Loop: issue #1045 root-cause fix for orchestrator/free passthrough 502 evidence + +Observation time: 2026-09-04 Asia/Seoul. + +GitHub authentication was re-verified first with `gh api user`. The primary +checkout was dirty, so work continued in a clean linked worktree at +`.worktrees/commercial-loop-20260904-issue1045`. Open PR heads and prior +`commercial-loop-*` worktrees were re-fetched before editing. No existing +open PR head covered this exact contract: PR [#1046](https://github.com/ContextualWisdomLab/contextual-orchestrator/pull/1046) +was only queued behind hosted checks for an unrelated EgressWeave SSRF change, +while the active `orchestrator/free` queue items [#1028](https://github.com/ContextualWisdomLab/contextual-orchestrator/pull/1028) +and [#993](https://github.com/ContextualWisdomLab/contextual-orchestrator/pull/993) +were preserved as distinct in-flight contracts. The highest-leverage +independent unit was therefore issue +[#1045](https://github.com/ContextualWisdomLab/contextual-orchestrator/issues/1045): +typed attempt evidence and bounded failover for long `orchestrator/free` +tool-loop transport failures. + +### Root cause confirmed on current `main` + +The live Noema review incidents in issue `#1045` reproduced the single-agent +tool-loop path, not `route_once()`. `/v1/chat/completions` tool-bearing +requests call `proxy_completion(..., single_agent=True)`, which performs +virtual-model passthrough failover inside `TaskOrchestrator.proxy_completion`. + +Current-head RCA: + +- raw HTTP passthrough failures were already classified and could participate + in bounded failover, but raw provider transport exceptions such as + `TimeoutError`, `urllib.error.URLError`, `ConnectionError`, and wrapped DNS + failures were not classified inside the multi-candidate passthrough loop; +- `classify_provider_failure()` already mapped those raw exceptions to bounded + typed 502 surfaces, but `proxy_completion()` only invoked the classifier for + `HTTPError` and already-classified `ProviderUpstreamError` instances; +- raw transport failures had no request-scoped attempt receipt explaining why + the gateway stopped. They remain non-replayable because a timeout or generic + connection failure does not prove that the provider rejected the request + before accepting work or usage. + +### Local fix completed + +The worktree change makes one surgical contract extension: + +- `proxy_completion()` now classifies every caught passthrough provider + exception before deciding whether failover is permitted; +- `orchestrator/free` virtual passthrough advances only after evidence that + proves non-acceptance, such as an RFC-defined request rejection or temporary + pre-request DNS failure. RFC 9110 section 9.2.2 does not authorize automatic + replay of a non-idempotent request from retryability alone. Generic + 500/502/504 responses and the non-standard 529 therefore remain sticky, + alongside raw timeout and generic transport failures, to prevent duplicate + completion and unreported usage; +- sticky failures now record the distinct failover decision + `sticky_candidate_failure` instead of incorrectly reusing + `eligible_candidates_exhausted`, while explicit concrete-model requests + remain single-provider sticky. + +The public error detail remains bounded and secret-safe: no credentials, raw +provider bodies, prompt text, or inferred endpoint hostnames are emitted. +Unclassified connection/timeout outcomes use lifecycle phase `transport`; +only explicit TLS failures use `connecting`. + +### Exact local verification + +The 2026-09-08 current-head review repair added three explicit acceptance +boundaries: no replay after an ambiguous transport outcome, no inferred +endpoint hostname in public attempt evidence, and no invented `connecting` +phase for an outcome whose lifecycle stage is unknown. The revised tests first +failed as expected (`3 failed, 61 passed`) and then passed after the minimal +owner fix. + +- Added focused regressions proving ambiguous raw/classified transport errors + remain sticky in both the in-process free-model loop and real + `/v1/chat/completions` HTTP path. A follow-up RED contract found that HTTP + 500 still replayed; the repair also keeps 502, 504, and 529 sticky while + retaining bounded failover for explicit rejection and temporary pre-request + DNS evidence. +- `.venv/bin/python -m pytest tests/test_passthrough_provider_failover.py tests/test_openai_passthrough.py -q` + -> `98 passed in 10.64s` +- `.venv/bin/python -m pytest tests/test_provider_error_taxonomy.py tests/test_passthrough_provider_failover.py tests/test_openai_passthrough.py -q` + -> `119 passed in 11.61s` +- `uvx ruff check --select E4,E7,E9,F contextual_orchestrator/orchestrator.py tests/test_passthrough_provider_failover.py tests/test_openai_passthrough.py` + and `git diff --check` -> success. + +### Branch-local quality note + +- `uv run ruff check contextual_orchestrator/orchestrator.py tests/test_passthrough_provider_failover.py tests/test_openai_passthrough.py` + could not run in this worktree because the pinned environment does not + currently expose a `ruff` executable (`No such file or directory`). + +Hosted exact-head checks, protected merge, and the unchanged LifeOS/Noema +consumer canary remain future steps because this invocation stopped at one +completed local root-cause work unit, per the hourly-loop boundary. diff --git a/tests/test_openai_passthrough.py b/tests/test_openai_passthrough.py index fd12a786c..4e783f26c 100644 --- a/tests/test_openai_passthrough.py +++ b/tests/test_openai_passthrough.py @@ -27,6 +27,7 @@ ModelClient, _responses_to_chat_payload, ) +from contextual_orchestrator.provider_errors import ProviderUpstreamError # noqa: E402 from contextual_orchestrator.server import ( # noqa: E402 SecurityConfig, build_server, @@ -510,6 +511,152 @@ def proxy_send_once(self, agent, endpoint, payload): assert "synthetic provider outage" not in json.dumps(body) +def test_http_named_tool_passthrough_exposes_bounded_attempt_evidence_on_502() -> None: + """Named tool passthrough keeps request-scoped candidate evidence on the wire. + + Virtual ``orchestrator/free`` + tools now take the Fugu route/conduct path on + main (``named_tool_passthrough`` requires a concrete model). Free-pool + multi-candidate sticky HTTP coverage remains in + ``tests/test_passthrough_provider_failover.py``. + """ + + class ConnectingNamedPool(ModelClient): + def proxy_send_once(self, agent, endpoint, payload): + del endpoint, payload + raise ProviderUpstreamError( + agent_id=agent.id, + model=agent.model, + error_code="provider_connection_error", + message=f"the provider {agent.id} connection failed or did not finish in time", + client_status=502, + provider_status=None, + retryable=True, + transport="passthrough", + ) + + proxy_send = proxy_send_once + + token = "passthrough_token" + orchestrator = TaskOrchestrator( + [ + ModelAgent( + "free_primary", + "free-primary-model", + tags=("cost:free",), + provider_name="free-primary", + priority=2, + ), + ModelAgent( + "free_backup", + "free-backup-model", + tags=("cost:free",), + provider_name="free-backup", + priority=1, + ), + ], + client=ConnectingNamedPool(), # type: ignore[arg-type] + ) + server = build_server( + orchestrator, port=0, security=SecurityConfig(auth_token=token) + ) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + status, body = _post( + f"http://127.0.0.1:{server.server_address[1]}/v1/chat/completions", + { + "model": "free-primary-model", + "messages": [{"role": "user", "content": "use the tool"}], + "tools": [{"type": "function", "function": {"name": "inspect"}}], + }, + token, + ) + finally: + server.shutdown() + server.server_close() + + assert status == 502 + assert body["error"]["code"] == "provider_connection_error" + assert body["error"]["detail"]["selected_candidate_ids"] == ["free_primary"] + assert body["error"]["detail"]["terminal_reason"] == "terminal_provider_failure" + assert body["error"]["detail"]["attempts"] == [ + { + "agent_id": "free_primary", + "model": "free-primary-model", + "provider_name": "free-primary", + "attempt_number": 1, + "error_code": "provider_connection_error", + "client_status": 502, + "provider_status": None, + "retryable": True, + "transport": "passthrough", + "phase": "transport", + "failover_decision": "sticky_candidate_failure", + }, + ] + assert "use the tool" not in json.dumps(body) + + +def test_http_named_tool_passthrough_raw_timeout_does_not_replay() -> None: + """Named tool passthrough keeps an ambiguous timeout on its selected candidate.""" + + class TimeoutNamedPool(ModelClient): + def proxy_send_once(self, agent, endpoint, payload): + del endpoint, payload + if agent.id == "free_primary": + raise TimeoutError("provider timed out") + return {"id": "chatcmpl-free", "object": "chat.completion", "model": agent.model, "choices": []} + + proxy_send = proxy_send_once + + token = "passthrough_token" + orchestrator = TaskOrchestrator( + [ + ModelAgent( + "free_primary", + "free-primary-model", + tags=("cost:free",), + provider_name="free-primary", + priority=2, + ), + ModelAgent( + "free_backup", + "free-backup-model", + tags=("cost:free",), + provider_name="free-backup", + priority=1, + ), + ], + client=TimeoutNamedPool(), # type: ignore[arg-type] + ) + server = build_server( + orchestrator, port=0, security=SecurityConfig(auth_token=token) + ) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + status, body = _post( + f"http://127.0.0.1:{server.server_address[1]}/v1/chat/completions", + { + "model": "free-primary-model", + "messages": [{"role": "user", "content": "use the tool"}], + "tools": [{"type": "function", "function": {"name": "inspect"}}], + }, + token, + ) + finally: + server.shutdown() + server.server_close() + + assert status == 502 + assert body["error"]["code"] == "provider_outcome_unknown" + assert body["error"]["detail"]["retryable"] is False + assert body["error"]["detail"]["terminal_reason"] == "terminal_provider_failure" + assert [item["agent_id"] for item in body["error"]["detail"]["attempts"]] == [ + "free_primary" + ] + assert "provider timed out" not in json.dumps(body) + + + def test_http_chat_completions_accepts_response_format_and_passes_through() -> None: server, port, token = _serve() url = f"http://127.0.0.1:{port}/v1/chat/completions" diff --git a/tests/test_passthrough_provider_failover.py b/tests/test_passthrough_provider_failover.py index a8257381f..fb599c804 100644 --- a/tests/test_passthrough_provider_failover.py +++ b/tests/test_passthrough_provider_failover.py @@ -271,6 +271,35 @@ def test_virtual_passthrough_all_oversized_tool_errors_preserve_size_contract() assert orchestrator._circuit == {} +def test_free_passthrough_raw_timeout_advances_with_attempt_evidence() -> None: + """Virtual free selectors may advance past one ambiguous timeout (#1166).""" + client = SequencedProxyClient( + { + "primary_agent": TimeoutError("provider timed out"), + "fallback_agent": {"model": "fallback-model", "choices": []}, + } + ) + orchestrator = _build(client) + orchestrator.agents = [ + replace(agent, tags=(*agent.tags, "cost:free")) for agent in orchestrator.agents + ] + + result = orchestrator.proxy_completion( + { + "model": TaskOrchestrator.FREE_MODEL, + "messages": [{"role": "user", "content": "use the tool"}], + "tools": [{"type": "function", "function": {"name": "inspect"}}], + } + ) + + assert result["model"] == "fallback-model" + assert [agent_id for agent_id, _ in client.calls] == [ + "primary_agent", + "fallback_agent", + ] + assert "primary_agent" in orchestrator._circuit + + def test_virtual_passthrough_keeps_non_size_tool_errors_sticky() -> None: """A generic provider invalid_tools response must not hide a bad request.""" failure = _invalid_tools_error() @@ -1314,6 +1343,164 @@ def test_classified_ambiguous_connection_error_does_not_fail_over() -> None: assert [agent_id for agent_id, _ in client.calls] == ["primary_agent"] +def test_free_virtual_model_keeps_ambiguous_transport_502_sticky() -> None: + """A classified ambiguous transport failure must not replay a free request.""" + failure = ProviderUpstreamError( + agent_id="primary_agent", + model="primary-model", + error_code="provider_connection_error", + message="the provider primary_agent connection failed or did not finish in time", + client_status=502, + provider_status=None, + retryable=True, + transport="passthrough", + ) + client = SequencedProxyClient( + { + "primary_agent": failure, + "fallback_agent": {"model": "fallback-model"}, + } + ) + orchestrator = _build(client) + orchestrator.agents = [ + replace(agent, tags=(*agent.tags, "cost:free")) for agent in orchestrator.agents + ] + + with pytest.raises(ProviderUpstreamError) as caught: + orchestrator.proxy_completion( + { + "model": TaskOrchestrator.FREE_MODEL, + "messages": [{"role": "user", "content": "use the tool"}], + "tools": [{"type": "function", "function": {"name": "inspect"}}], + } + ) + + assert caught.value.detail["attempts"][0]["phase"] == "transport" + assert [agent_id for agent_id, _ in client.calls] == ["primary_agent"] + + + +@pytest.mark.parametrize("status", [500, 502, 504, 529]) +def test_free_virtual_model_keeps_ambiguous_http_failure_sticky(status: int) -> None: + """Ambiguous HTTP failures do not prove non-acceptance and cannot replay.""" + client = SequencedProxyClient( + { + "primary_agent": _http_error(status), + "fallback_agent": {"model": "fallback-model"}, + } + ) + orchestrator = _build(client) + orchestrator.agents = [ + replace(agent, tags=(*agent.tags, "cost:free")) for agent in orchestrator.agents + ] + + with pytest.raises(ProviderUpstreamError) as caught: + orchestrator.proxy_completion( + { + "model": TaskOrchestrator.FREE_MODEL, + "messages": [{"role": "user", "content": "use the tool"}], + "tools": [{"type": "function", "function": {"name": "inspect"}}], + } + ) + + assert caught.value.provider_status == status + assert caught.value.detail["terminal_reason"] == "terminal_provider_failure" + assert caught.value.detail["attempts"][0]["failover_decision"] == ( + "sticky_candidate_failure" + ) + assert [agent_id for agent_id, _ in client.calls] == ["primary_agent"] + + +def test_free_virtual_model_stops_after_a_rejection_then_ambiguous_failure() -> None: + """One proved rejection may advance, but an ambiguous next failure is sticky.""" + client = SequencedProxyClient( + {"primary_agent": _http_error(429), "fallback_agent": _http_error(500)} + ) + orchestrator = _build(client) + orchestrator.agents = [ + replace( + agent, + base_url=f"mock://{agent.id}", + provider_name="", + tags=(*agent.tags, "cost:free"), + ) + for agent in orchestrator.agents + ] + + with pytest.raises(ProviderUpstreamError) as caught: + orchestrator.proxy_completion( + { + "model": TaskOrchestrator.FREE_MODEL, + "messages": [{"role": "user", "content": "use the tool"}], + "tools": [{"type": "function", "function": {"name": "inspect"}}], + } + ) + + assert caught.value.agent_id == "fallback_agent" + assert caught.value.detail["terminal_reason"] == "terminal_provider_failure" + assert caught.value.detail["selected_candidate_ids"] == [ + "primary_agent", + "fallback_agent", + ] + assert caught.value.detail["attempts"] == [ + { + "agent_id": "primary_agent", + "model": "primary-model", + "provider_name": "unreported", + "attempt_number": 1, + "error_code": "rate_limit_exceeded", + "client_status": 429, + "provider_status": 429, + "retryable": True, + "transport": "passthrough", + "phase": "provider_response", + "failover_decision": "advance_to_next_candidate", + }, + { + "agent_id": "fallback_agent", + "model": "fallback-model", + "provider_name": "unreported", + "attempt_number": 2, + "error_code": "api_error", + "client_status": 502, + "provider_status": 500, + "retryable": True, + "transport": "passthrough", + "phase": "provider_response", + "failover_decision": "sticky_candidate_failure", + }, + ] + assert [agent_id for agent_id, _ in client.calls] == ["primary_agent", "fallback_agent"] + + +def test_explicit_model_classified_transport_502_remains_sticky() -> None: + """A concrete model id keeps single-provider stickiness for retryable transport failures.""" + failure = ProviderUpstreamError( + agent_id="primary_agent", + model="primary-model", + error_code="provider_connection_error", + message="the provider primary_agent connection failed or did not finish in time", + client_status=502, + provider_status=None, + retryable=True, + transport="passthrough", + ) + client = SequencedProxyClient( + { + "primary_agent": failure, + "fallback_agent": {"model": "fallback-model"}, + } + ) + + with pytest.raises(ProviderUpstreamError) as caught: + _build(client).proxy_completion( + {"model": "primary-model", "messages": [{"role": "user", "content": "x"}]} + ) + + assert caught.value is failure + assert [agent_id for agent_id, _ in client.calls] == ["primary_agent"] + + def test_suppressed_transient_context_does_not_authorize_failover() -> None: """A deliberately hidden exception context cannot become a routing signal.""" try: @@ -1330,23 +1517,38 @@ def test_suppressed_transient_context_does_not_authorize_failover() -> None: } ) - with pytest.raises(RuntimeError, match="terminal wrapper") as caught: + with pytest.raises(ProviderUpstreamError) as caught: _build(client).proxy_completion({"messages": [{"role": "user", "content": "x"}]}) - assert caught.value is failure + assert caught.value.error_code == "api_error" + assert caught.value.detail["terminal_reason"] == "terminal_provider_failure" + assert caught.value.detail["attempts"] == [ + { + "agent_id": "primary_agent", + "model": "primary-model", + "provider_name": "primary", + "attempt_number": 1, + "error_code": "api_error", + "client_status": 502, + "provider_status": None, + "retryable": False, + "transport": "passthrough", + "phase": "provider_response", + "failover_decision": "sticky_candidate_failure", + } + ] assert [agent_id for agent_id, _ in client.calls] == ["primary_agent"] def test_all_candidates_chain_the_last_failure() -> None: """Exhaustion reports one stable gateway error with the final provider cause.""" final = _http_error(503) - # 500 here (not 429): this test is about which classified failure - # survives exhaustion, not rate-limiting -- a bare 429 with no - # Retry-After now assumes a short quota cooldown and waits, which would - # route this candidate's identity through the rate-limit-storm path - # instead of the plain exhaustion path this test actually exercises. + # 408 here (not 429 or 500): 429 can enter the rate-limit-storm wait path, + # and 500/502/504/529 stay sticky under the RFC-bounded rejection set from + # #1049. A proved request rejection (408) may advance, then the final 503 + # is reported on exhaustion without waiting. orchestrator = _build( - SequencedProxyClient({"primary_agent": _http_error(500), "fallback_agent": final}) + SequencedProxyClient({"primary_agent": _http_error(408), "fallback_agent": final}) ) with pytest.raises(ProviderUpstreamError) as caught: @@ -1451,10 +1653,25 @@ def test_only_temporary_dns_failures_advance( {"messages": [{"role": "user", "content": "x"}]} )["model"] == "fallback-model" else: - with pytest.raises(RuntimeError, match="provider resolution failed"): - _build(client).proxy_completion( - {"messages": [{"role": "user", "content": "x"}]} - ) + with pytest.raises(ProviderUpstreamError) as caught: + _build(client).proxy_completion({"messages": [{"role": "user", "content": "x"}]}) + assert caught.value.error_code == "provider_connection_error" + assert caught.value.detail["terminal_reason"] == "terminal_provider_failure" + assert caught.value.detail["attempts"] == [ + { + "agent_id": "primary_agent", + "model": "primary-model", + "provider_name": "primary", + "attempt_number": 1, + "error_code": "provider_connection_error", + "client_status": 502, + "provider_status": None, + "retryable": False, + "transport": "passthrough", + "phase": "transport", + "failover_decision": "sticky_candidate_failure", + } + ] @pytest.mark.parametrize("error_type", [TimeoutError, ConnectionError]) diff --git a/tests/test_provider_reliability.py b/tests/test_provider_reliability.py index 15028e2bf..78fd023ce 100644 --- a/tests/test_provider_reliability.py +++ b/tests/test_provider_reliability.py @@ -22,6 +22,10 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.credentials import ( # noqa: E402 + InMemoryCredentialBackend, + set_backend, +) from contextual_orchestrator.orchestrator import ( # noqa: E402 TRANSIENT_HTTP_STATUS, ModelClient, @@ -913,6 +917,152 @@ def chat(self, agent: ModelAgent, messages: list, temperature: float = 0.2) -> s assert "priced_worker" not in client.calls +def test_unallowlisted_provider_host_is_classified_upstream_error() -> None: + """Live orchestrator/free 500s were raw RuntimeError from host allowlisting.""" + client = ModelClient(allowed_provider_hosts={"ok.example"}) + agent = ModelAgent( + "blocked_agent", + "blocked-model", + base_url="https://blocked.example/v1", + credential_key="MODEL_KEY", + ) + backend = InMemoryCredentialBackend() + backend.set("MODEL_KEY", "sk-host-check") + set_backend(backend) + try: + with pytest.raises(ProviderUpstreamError) as excinfo: + client._validate_provider(agent) + finally: + set_backend(None) + assert excinfo.value.client_status == 502 + assert excinfo.value.retryable is False + assert excinfo.value.error_code == "provider_connection_error" + assert "allowlisted" in str(excinfo.value) + assert "blocked.example" not in str(excinfo.value) + + +def test_passthrough_allowlist_failure_reports_passthrough_transport() -> None: + """Passthrough validation evidence must name the surface that invoked it.""" + client = ModelClient(allowed_provider_hosts={"ok.example"}) + agent = ModelAgent( + "blocked_agent", + "blocked-model", + base_url="https://blocked.example/v1", + credential_key="MODEL_KEY", + ) + backend = InMemoryCredentialBackend() + backend.set("MODEL_KEY", "sk-host-check") + set_backend(backend) + try: + with pytest.raises(ProviderUpstreamError) as excinfo: + client.proxy_send_once( + agent, + "chat/completions", + {"messages": [{"role": "user", "content": "hello"}]}, + ) + finally: + set_backend(None) + + assert excinfo.value.transport == "passthrough" + + +def test_free_model_advances_past_unallowlisted_provider_host() -> None: + """A host-allowlist miss must skip to the next free candidate, not 500.""" + calls: list[str] = [] + blocked = ModelAgent( + "free_route_a", + "free_route_a-model", + base_url="https://blocked.example/v1", + credential_key="MODEL_KEY", + tags=("reasoning", "cost:free"), + ) + + class AllowlistThenOk(ModelClient): + def __init__(self) -> None: + super().__init__(allowed_provider_hosts={"ok.example"}) + + def chat(self, agent: ModelAgent, messages: list, temperature: float = 0.2) -> str: # type: ignore[override] + del messages, temperature + calls.append(agent.id) + if agent.id == "free_route_a": + self._validate_provider(blocked) + return f"[{agent.id}] answer" + + backend = InMemoryCredentialBackend() + backend.set("MODEL_KEY", "sk-host-check") + set_backend(backend) + try: + orchestrator = _free_pool_orchestrator( + AllowlistThenOk(), free_ids=("free_route_a", "free_route_b") + ) + orchestrator.tool_retry_attempts = 0 + result = orchestrator.route_once( + [{"role": "user", "content": "route this"}], + model_name=TaskOrchestrator.FREE_MODEL, + ) + finally: + set_backend(None) + + assert result["answer"] == "[free_route_b] answer" + assert result["trace"][0]["served_agent_id"] == "free_route_b" + assert calls == ["free_route_a", "free_route_b"] + assert "priced_worker" not in calls + + +def test_free_model_exhausted_allowlist_pool_fails_closed_as_502() -> None: + """Every free candidate missing the host allowlist must not collapse to HTTP 500.""" + blocked_by_id = { + "free_route_a": ModelAgent( + "free_route_a", + "free_route_a-model", + base_url="https://blocked-a.example/v1", + credential_key="MODEL_KEY", + tags=("reasoning", "cost:free"), + ), + "free_route_b": ModelAgent( + "free_route_b", + "free_route_b-model", + base_url="https://blocked-b.example/v1", + credential_key="MODEL_KEY", + tags=("reasoning", "cost:free"), + ), + } + + class AllBlocked(ModelClient): + def __init__(self) -> None: + super().__init__(allowed_provider_hosts={"ok.example"}) + self.calls: list[str] = [] + + def chat(self, agent: ModelAgent, messages: list, temperature: float = 0.2) -> str: # type: ignore[override] + del messages, temperature + self.calls.append(agent.id) + self._validate_provider(blocked_by_id[agent.id]) + raise AssertionError("allowlisted host was not supposed to be reached") + + backend = InMemoryCredentialBackend() + backend.set("MODEL_KEY", "sk-host-check") + set_backend(backend) + client = AllBlocked() + try: + orchestrator = _free_pool_orchestrator( + client, free_ids=("free_route_a", "free_route_b") + ) + orchestrator.tool_retry_attempts = 0 + with pytest.raises(ProviderUpstreamError) as excinfo: + orchestrator.route_once( + [{"role": "user", "content": "route this"}], + model_name=TaskOrchestrator.FREE_MODEL, + ) + finally: + set_backend(None) + + assert excinfo.value.client_status == 502 + assert "all " not in str(excinfo.value) + assert "role=" not in str(excinfo.value) + assert client.calls == ["free_route_a", "free_route_b"] + assert "priced_worker" not in client.calls + + def test_free_model_failover_survives_a_tool_shaped_provider_message() -> None: """A 500 whose body happens to mention "tool"/"invalid arguments" still fails over.