diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index 7c9b372fd..da94efc79 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -38,6 +38,10 @@ def _provider_discovery_error_code(exc: Exception) -> str: return "timeout" if isinstance(exc, urllib.error.URLError): return "transport_error" + if isinstance(exc, (ConnectionError, OSError)): + # Raw connection resets/failures that are not URLError still count as + # transport failures so callers see one stable code, never provider text. + return "transport_error" if isinstance(exc, ValueError): return "invalid_response" return "provider_error" @@ -212,7 +216,10 @@ def discover_provider_models( url = f"{url}?task={source.task_filter}" try: payload = _fetch_json(url, api_key=api_key, auth_scheme=source.auth_scheme, timeout=timeout) - except (urllib.error.URLError, TimeoutError, ValueError) as exc: # pragma: no cover - network path + except (urllib.error.URLError, TimeoutError, ValueError, OSError) as exc: + # OSError covers ConnectionError/reset failures that are not URLError + # subclasses, so a raw provider transport failure can never escape the + # discovery boundary with provider text attached. raise ProviderDiscoveryError(source.provider_name, _provider_discovery_error_code(exc)) from None if source.style == "bytez": return _parse_bytez(payload, source) diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 61b72a650..661885b6d 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -1206,6 +1206,7 @@ def _stream_send( headers=headers, method="POST", ) + stream_error: RuntimeError | None = None try: with self._open_provider(request, destination) as response: for raw in response: @@ -1219,13 +1220,25 @@ def _stream_send( chunk = json.loads(data) except json.JSONDecodeError: continue - delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content") + choices = chunk.get("choices") or [{}] + delta = (choices[0] or {}).get("delta", {}).get("content") if delta: yield delta - except urllib.error.HTTPError as exc: + except Exception as exc: # noqa: BLE001 - provider error boundary (CWE-209) + # The gateway's own terminal tool-stop contract must survive the + # boundary: convert the provider HTTP shape into the package-owned + # stop error so callers keep the 409 semantics they rely on. if _is_tool_execution_stopped(exc): raise _provider_tool_execution_stopped(agent) from None - raise + if isinstance(exc, ToolFallbackStoppedError): + raise + # A stream may already have emitted bytes, so it can neither be retried + # nor failed over to another provider. Keep the provider status, body, + # and exception cause inside the gateway; callers get one stable, + # package-owned error instead of raw provider diagnostics. + stream_error = RuntimeError(f"provider {agent.id} streaming request failed") + if stream_error is not None: + raise stream_error # -- Full OpenAI passthrough (transport) ------------------------------------ # Requests that carry provider features the multi-agent verifier cannot merge @@ -1436,9 +1449,18 @@ def batch_chat( results = self._local_batch_chat(agent, requests, temperature, effort_profile) else: destination = self._validate_provider(agent) # pragma: no cover - results = self._batch_run( # pragma: no cover - agent, requests, temperature, poll_interval, poll_timeout, destination, effort_profile - ) + batch_error: RuntimeError | None = None + try: + results = self._batch_run( # pragma: no cover + agent, requests, temperature, poll_interval, poll_timeout, destination, effort_profile + ) + except Exception: # noqa: BLE001 - provider batch boundary (CWE-209) + # Batch upload, polling, and output retrieval all cross the same + # public gateway boundary; provider bodies and exception text stay + # inside the authorized provider observability system. + batch_error = RuntimeError(f"provider {agent.id} batch request failed") + if batch_error is not None: + raise batch_error return _validate_batch_results(requests, results) def _local_batch_chat( diff --git a/docs/planning/adrs/0011-provider-error-boundary.md b/docs/planning/adrs/0011-provider-error-boundary.md index ab71103f0..4c6885412 100644 --- a/docs/planning/adrs/0011-provider-error-boundary.md +++ b/docs/planning/adrs/0011-provider-error-boundary.md @@ -24,6 +24,11 @@ available through a public gateway error or an exception cause. 5. Provider diagnostics may be counted by allowlisted type/code in internal telemetry, but raw bodies, exception text, credentials, and prompts are not persisted or returned. +6. Streaming and Batch API paths follow the same boundary: a mid-stream or + batch failure surfaces one package-owned error (`provider streaming + request failed` / `provider batch request failed`) because a stream may + already have emitted bytes (no retry, no failover), and raw connection + resets outside ``URLError`` map to the same stable ``transport_error`` code. ## Consequences @@ -35,10 +40,10 @@ responses have a wider audience than provider credentials and request data. ## Verification -`tests/test_model_discovery.py`, `tests/test_provider_reliability.py`, and -`tests/test_model_judge.py` assert that provider response text is absent from -public messages and causes, while the full suite must remain green before -merge. +`tests/test_model_discovery.py`, `tests/test_provider_reliability.py`, +`tests/test_true_streaming.py`, and `tests/test_model_judge.py` assert that +provider response text is absent from public messages and causes, while the +full suite must remain green before merge. ## References diff --git a/tests/test_model_discovery.py b/tests/test_model_discovery.py index 26c509ebf..ee8f91f9b 100644 --- a/tests/test_model_discovery.py +++ b/tests/test_model_discovery.py @@ -194,6 +194,30 @@ def urlopen(request, timeout=None): assert errors[0].__cause__ is None +def test_discovery_boundary_contains_raw_connection_reset() -> None: + """A raw ConnectionResetError (not a URLError) still fails inside the boundary. + + Regression: ``ConnectionError``/``OSError`` subclasses that are not + ``URLError`` used to escape ``discover_provider_models`` uncaught, leaking + provider transport diagnostics to discovery callers. + """ + register_credential("OPENAI_API_KEY", "sk-openai") + + def urlopen(request, timeout=None): + raise ConnectionResetError(104, "Connection reset by peer") + + with patch("contextual_orchestrator.model_discovery.urllib.request.urlopen", side_effect=urlopen): + try: + discover_provider_models(OPENAI_SOURCE) + except ProviderDiscoveryError as error: + assert error.provider_name == "openai" + assert error.error_code == "transport_error" + assert "reset" not in str(error) + assert error.__cause__ is None + else: # pragma: no cover + raise AssertionError("a raw connection reset must become a ProviderDiscoveryError") + + def test_agent_id_for_is_two_word_snake_case() -> None: discovered = DiscoveredModel( provider_name="openrouter", diff --git a/tests/test_provider_reliability.py b/tests/test_provider_reliability.py index 2b149977a..63c3105f9 100644 --- a/tests/test_provider_reliability.py +++ b/tests/test_provider_reliability.py @@ -431,6 +431,41 @@ def test_mock_path_is_unchanged_no_failover_no_circuit_state() -> None: assert orchestrator._circuit == {} +def test_batch_boundary_hides_raw_upload_error_text_and_cause() -> None: + """Provider Batch API failures surface one package-owned error (CWE-209). + + Regression: ``_batch_run`` (upload, poll, download) propagated raw + ``urllib`` errors carrying the provider URL and response body out of + ``batch_chat``. + """ + class RawBatchFailureClient(ModelClient): + def _validate_provider(self, agent): # type: ignore[override] + return None # skip URL/credential validation; only the boundary matters here + + def _batch_run(self, agent, requests, temperature, poll_interval, poll_timeout, destination=None): # type: ignore[override] + raise RuntimeError("provider-secret-batch-body") + + client = RawBatchFailureClient() + agent = ModelAgent("batch_worker", "gpt", base_url="https://provider.example/v1") + try: + client.batch_chat(agent, {"task_0": [{"role": "user", "content": "ping"}]}) + except RuntimeError as error: + assert "batch request failed" in str(error) + assert "provider-secret-batch-body" not in str(error) + assert error.__cause__ is None + else: # pragma: no cover + raise AssertionError("a failed provider batch must raise a package-owned error") + + +def test_mock_batch_path_is_not_wrapped() -> None: + """The mock batch path keeps its plain results — no boundary wrapping.""" + client = ModelClient() + agent = ModelAgent("solo_worker", "mock") + results = client.batch_chat(agent, {"task_0": [{"role": "user", "content": "hello"}]}) + assert set(results) == {"task_0"} + assert results["task_0"]["content"] + + if __name__ == "__main__": for name, fn in sorted(globals().items()): if name.startswith("test_") and callable(fn): diff --git a/tests/test_true_streaming.py b/tests/test_true_streaming.py index 7220def82..d8f99e264 100644 --- a/tests/test_true_streaming.py +++ b/tests/test_true_streaming.py @@ -58,6 +58,15 @@ def _delta(content: str) -> str: return 'data: ' + json.dumps({"choices": [{"delta": {"content": content}}]}) + "\n\n" +def _usage_frame(choices: list[dict] | None = None, usage: dict | None = None) -> str: + payload: dict[str, object] = {} + if choices is not None: + payload["choices"] = choices + if usage is not None: + payload["usage"] = usage + return 'data: ' + json.dumps(payload) + "\n\n" + + def test_stream_send_parses_real_provider_sse() -> None: frames = [ 'data: {"choices":[{"delta":{"role":"assistant"}}]}\n\n', # role delta, no content @@ -74,6 +83,65 @@ def test_stream_send_parses_real_provider_sse() -> None: assert "".join(deltas) == "Hello streamed world" +def test_stream_send_ignores_empty_and_missing_choices() -> None: + """Usage-only or malformed frames must not abort a valid stream (IndexError).""" + frames = [ + _delta("before"), + _usage_frame(choices=[], usage={"prompt_tokens": 3}), + _usage_frame(usage={"completion_tokens": 7}), + _delta("after"), + "data: [DONE]\n\n", + ] + with _FakeSSEProvider(frames) as provider: + client = ModelClient() + agent = ModelAgent("worker_agent", "gpt-x", base_url=provider.base_url, api_key_env="UNSET_KEY_ENV") + deltas = list(client._stream_send(agent, {"model": "gpt-x", "stream": True})) + assert deltas == ["before", "after"] + + +def test_stream_send_hides_raw_provider_error_text_and_cause() -> None: + """A mid-stream provider failure surfaces one package-owned error (CWE-209). + + Regression: the raw ``HTTPError`` (provider URL, reason phrase, and body) + used to re-raise out of the streaming generator, leaking upstream + diagnostics to gateway callers and logs. + """ + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers.get("content-length", 0)) + self.rfile.read(length) + self.send_response(500) + self.end_headers() + self.wfile.write(b"upstream-secret-diagnostic http://10.0.0.9/internal") + + def log_message(self, *args: object) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + client = ModelClient() + agent = ModelAgent( + "worker_agent", + "gpt-x", + base_url=f"http://127.0.0.1:{server.server_address[1]}", + api_key_env="UNSET_KEY_ENV", + ) + try: + list(client._stream_send(agent, {"model": "gpt-x", "stream": True})) + except RuntimeError as error: + assert "streaming request failed" in str(error) + assert "http://" not in str(error) # provider URL stays inside + assert "upstream-secret-diagnostic" not in str(error) + assert error.__cause__ is None + else: # pragma: no cover + raise AssertionError("a failed stream must raise a package-owned error") + finally: + server.shutdown() + + def test_stream_chat_mock_yields_chunks() -> None: client = ModelClient() agent = ModelAgent("general_agent", "mock-model") # base_url defaults to mock://local