fix: extend provider error boundary to streaming and batch paths

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)

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(

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 <id> streaming
+   request failed` / `provider <id> 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
 

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",

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):

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