fix: preserve provider tool-stop failures

contextual_orchestrator/orchestrator.py

@@ -423,7 +423,8 @@ def as_dict(self) -> dict[str, Any]:
 
 def _is_tool_execution_stopped(error: urllib.error.HTTPError) -> bool:
     """Return whether an HTTP error carries the terminal tool-stop contract."""
-    cached = getattr(error, "_contextual_tool_execution_stopped", None)
+    cache_key = "_contextual_orchestrator_tool_execution_stopped"
+    cached = getattr(error, cache_key, None)
     if isinstance(cached, bool):
         return cached
     try:
@@ -434,23 +435,23 @@ def _is_tool_execution_stopped(error: urllib.error.HTTPError) -> bool:
         details = payload.get("error") if isinstance(payload, dict) else None
         result = isinstance(details, dict) and details.get("code") == "tool_execution_stopped"
     try:
-        error._contextual_tool_execution_stopped = result
-    except AttributeError:  # pragma: no cover - HTTPError permits attributes
+        setattr(error, cache_key, result)
+    except (AttributeError, TypeError):  # pragma: no cover - HTTPError is mutable
         pass
     return result
 
 
-def _provider_tool_execution_stopped(agent_id: str) -> ToolFallbackStoppedError:
-    """Convert the provider's terminal tool-stop response to the safe gateway contract."""
+def _provider_tool_execution_stopped(agent: ModelAgent) -> ToolFallbackStoppedError:
+    """Convert the provider's terminal tool-stop contract to the public safe error."""
     decision = classify_tool_failure(
         ToolExecutionError(
-            "provider reported tool_execution_stopped",
-            tool_name="provider",
+            "provider reported terminal tool execution state",
+            tool_name="provider_tool_runtime",
             kind=ToolFailureKind.TRANSPORT_ERROR,
             outcome_unknown=True,
         )
     )
-    return ToolFallbackStoppedError(agent_id, decision)
+    return ToolFallbackStoppedError(agent.id, decision)
 
 
 def _is_local_provider_url(base_url: str) -> bool:
@@ -958,7 +959,7 @@ def _send_with_retry(
                     break
                 self._sleep(self._backoff_delay(attempt))
         if isinstance(last_error, urllib.error.HTTPError) and _is_tool_execution_stopped(last_error):
-            raise _provider_tool_execution_stopped(agent.id) from None
+            raise _provider_tool_execution_stopped(agent) from None
         if isinstance(last_error, ProviderResponseError):
             raise last_error
         raise RuntimeError(f"provider {agent.id} request failed") from None
@@ -1154,21 +1155,26 @@ def _stream_send(
             headers=headers,
             method="POST",
         )
-        with self._open_provider(request, destination) as response:
-            for raw in response:
-                line = raw.decode("utf-8").strip()
-                if not line.startswith("data:"):
-                    continue
-                data = line[len("data:") :].strip()
-                if data == "[DONE]":
-                    break
-                try:
-                    chunk = json.loads(data)
-                except json.JSONDecodeError:
-                    continue
-                delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content")
-                if delta:
-                    yield delta
+        try:
+            with self._open_provider(request, destination) as response:
+                for raw in response:
+                    line = raw.decode("utf-8").strip()
+                    if not line.startswith("data:"):
+                        continue
+                    data = line[len("data:") :].strip()
+                    if data == "[DONE]":
+                        break
+                    try:
+                        chunk = json.loads(data)
+                    except json.JSONDecodeError:
+                        continue
+                    delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content")
+                    if delta:
+                        yield delta
+        except urllib.error.HTTPError as exc:
+            if _is_tool_execution_stopped(exc):
+                raise _provider_tool_execution_stopped(agent) from None
+            raise
 
     # -- Full OpenAI passthrough (transport) ------------------------------------
     # Requests that carry provider features the multi-agent verifier cannot merge
@@ -1215,7 +1221,7 @@ def _send_raw_with_retry(
                     break
                 self._sleep(self._backoff_delay(attempt))
         if isinstance(last_error, urllib.error.HTTPError) and _is_tool_execution_stopped(last_error):
-            raise _provider_tool_execution_stopped(agent.id) from None
+            raise _provider_tool_execution_stopped(agent) from None
         raise RuntimeError(f"provider {agent.id} passthrough request failed") from None
 
     def _send_raw(
@@ -2617,7 +2623,7 @@ def _invoke(
             while True:
                 try:
                     output = self.client.chat(agent, messages)
-                except Exception as exc:  # noqa: BLE001 - classify before fallback
+                except Exception as exc:
                     if isinstance(exc, (ProviderResponseError, ToolFallbackStoppedError)):
                         raise
                     last_error = exc

tests/test_provider_reliability.py

@@ -16,6 +16,8 @@
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
+import pytest
+
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
 from contextual_orchestrator import ModelAgent, TaskOrchestrator  # noqa: E402
@@ -25,20 +27,20 @@
     ProviderResponseError,
     is_transient_error,
 )
-from contextual_orchestrator.tool_fallback import ToolFallbackStoppedError  # noqa: E402
+from contextual_orchestrator.tool_fallback import ToolFallbackStoppedError
 
 
 def _http_error(code: int) -> urllib.error.HTTPError:
     return urllib.error.HTTPError("https://provider.example/chat/completions", code, "err", None, None)
 
 
-def _tool_execution_stopped_error() -> urllib.error.HTTPError:
+def _stopped_http_error() -> urllib.error.HTTPError:
     return urllib.error.HTTPError(
         "https://provider.example/chat/completions",
         409,
-        "tool stopped",
+        "Conflict",
         None,
-        io.BytesIO(b'{"error":{"code":"tool_execution_stopped"}}'),
+        io.BytesIO(json.dumps({"error": {"code": "tool_execution_stopped"}}).encode()),
     )
 
 
@@ -67,12 +69,12 @@ def _validate_provider(self, agent: ModelAgent):  # type: ignore[override]
         def _send(self, agent, payload, destination=None, *, timeout=None):  # type: ignore[override]
             del agent, payload, destination, timeout
             self.chat_calls += 1
-            raise _tool_execution_stopped_error()
+            raise _stopped_http_error()
 
         def _send_raw(self, agent, endpoint, payload, destination=None):  # type: ignore[override]
             del agent, endpoint, payload, destination
             self.raw_calls += 1
-            raise _tool_execution_stopped_error()
+            raise _stopped_http_error()
 
     client = StoppedProviderClient()
     agent = ModelAgent(
@@ -105,17 +107,42 @@ def _send_raw(self, agent, endpoint, payload, destination=None):  # type: ignore
 
 
 def test_tool_execution_stopped_409_is_terminal_but_generic_conflict_retries() -> None:
-    stopped = urllib.error.HTTPError(
-        "https://provider.example/chat/completions",
-        409,
-        "Conflict",
-        None,
-        io.BytesIO(json.dumps({"error": {"code": "tool_execution_stopped"}}).encode()),
-    )
+    stopped = _stopped_http_error()
     assert not is_transient_error(stopped)
     assert is_transient_error(_http_error(409))
 
 
+def test_terminal_tool_stop_is_preserved_by_chat_and_passthrough_transport() -> None:
+    class StoppedClient(ModelClient):
+        def __init__(self) -> None:
+            super().__init__(max_retries=3, retry_backoff=0.0)
+            self.attempts = 0
+
+        def _send(self, agent: ModelAgent, payload: dict, destination=None) -> str:  # type: ignore[override]
+            del agent, payload, destination
+            self.attempts += 1
+            raise _stopped_http_error()
+
+        def _send_raw(self, agent: ModelAgent, endpoint: str, payload: dict, destination=None) -> dict:  # type: ignore[override]
+            del agent, endpoint, payload, destination
+            self.attempts += 1
+            raise _stopped_http_error()
+
+    client = StoppedClient()
+    agent = ModelAgent("worker_agent", "gpt", base_url="https://provider.example/v1")
+
+    with pytest.raises(ToolFallbackStoppedError) as chat_error:
+        client._send_with_retry(agent, {"model": agent.model})
+    assert chat_error.value.decision.kind.value == "ambiguous_outcome"
+    assert chat_error.value.decision.observed_kind.value == "transport_error"
+    assert client.attempts == 1
+
+    with pytest.raises(ToolFallbackStoppedError) as passthrough_error:
+        client._send_raw_with_retry(agent, "chat/completions", {"model": agent.model})
+    assert passthrough_error.value.decision.reason_code == "tool_failure.ambiguous_outcome.fail_closed"
+    assert client.attempts == 2
+
+
 def test_retry_recovers_from_transient_failures_with_backoff() -> None:
     delays: list[float] = []
 

tests/test_tool_execution_fallback.py

@@ -21,9 +21,9 @@
 from contextual_orchestrator.tool_fallback import (
     MAX_TOOL_RETRY_ATTEMPTS,
     ToolExecutionError,
+    ToolFailureKind,
     ToolFallbackAction,
     ToolFallbackStoppedError,
-    ToolFailureKind,
     classify_tool_failure,
 )
 
@@ -641,6 +641,46 @@ def _post_fallback_json(
         return error.code, json.loads(error.read().decode("utf-8"))
 
 
+def _provider_tool_stop_http_error() -> urllib.error.HTTPError:
+    """Build the provider-originated terminal 409 without retaining provider text."""
+    return urllib.error.HTTPError(
+        "https://provider.example/chat/completions",
+        409,
+        "Conflict",
+        None,
+        io.BytesIO(json.dumps({"error": {"code": "tool_execution_stopped"}}).encode()),
+    )
+
+
+class _ProviderStoppedChatClient(ModelClient):
+    """Return the provider terminal contract through the real chat transport path."""
+
+    def __init__(self) -> None:
+        super().__init__(max_retries=3, retry_backoff=0.0)
+        self.calls: list[str] = []
+
+    def _validate_provider(self, agent: ModelAgent):  # type: ignore[override]
+        del agent
+        return None
+
+    def _send(self, agent: ModelAgent, payload: dict, destination=None) -> str:  # type: ignore[override]
+        del payload, destination
+        self.calls.append(agent.id)
+        raise _provider_tool_stop_http_error()
+
+
+class _ProviderStoppedStreamingClient(ModelClient):
+    """Return the provider terminal contract while opening a streaming response."""
+
+    def _validate_provider(self, agent: ModelAgent):  # type: ignore[override]
+        del agent
+        return None
+
+    def _open_provider(self, request, destination=None, *, timeout=None):  # type: ignore[override]
+        del request, destination, timeout
+        raise _provider_tool_stop_http_error()
+
+
 def test_http_fail_closed_tool_error_has_dedicated_contract() -> None:
     error = ToolExecutionError(
         "request may have completed token=must-not-leak",
@@ -683,39 +723,24 @@ def test_http_fail_closed_tool_error_has_dedicated_contract() -> None:
     assert "must-not-leak" not in json.dumps(body)
 
 
-class _ProviderStoppedClient(ModelClient):
-    """Expose a provider's terminal 409 without making a network request."""
-
-    def _validate_provider(self, agent: ModelAgent):  # type: ignore[override]
-        del agent
-        return
-
-    def _send(self, agent, payload, destination=None, *, timeout=None):  # type: ignore[override]
-        del agent, payload, destination, timeout
-        raise urllib.error.HTTPError(
-            "https://provider.example/chat/completions",
-            409,
-            "tool stopped",
-            None,
-            io.BytesIO(b'{"error":{"code":"tool_execution_stopped"}}'),
-        )
-
-
-def test_provider_tool_stop_preserves_http_contract() -> None:
-    orchestrator = TaskOrchestrator(
-        [
-            ModelAgent(
-                "provider_worker",
-                "provider-model",
-                base_url="https://provider.example/v1",
-                api_key_env="",
-                credential_key="",
-            )
-        ],
-        client=_ProviderStoppedClient(max_retries=2, retry_backoff=0.0),
-    )
+def test_provider_http_tool_stop_preserves_409_and_does_not_fail_over() -> None:
+    agents = [
+        ModelAgent(
+            "primary_worker",
+            "provider-model",
+            base_url="https://provider.example/v1",
+            credential_key="",
+        ),
+        ModelAgent(
+            "backup_worker",
+            "provider-model",
+            base_url="https://provider.example/v1",
+            credential_key="",
+        ),
+    ]
+    client = _ProviderStoppedChatClient()
     server = build_server(
-        orchestrator,
+        TaskOrchestrator(agents, client=client),
         port=0,
         security=SecurityConfig(auth_token="secret_token"),
     )
@@ -737,9 +762,55 @@ def test_provider_tool_stop_preserves_http_contract() -> None:
     assert status == 409
     assert body["error"]["code"] == "tool_execution_stopped"
     assert body["error"]["detail"]["failure_kind"] == "ambiguous_outcome"
+    assert client.calls == ["primary_worker"]
     assert "provider.example" not in json.dumps(body)
 
 
+def test_provider_http_tool_stop_preserves_streaming_sse_contract() -> None:
+    agent = ModelAgent(
+        "primary_worker",
+        "provider-model",
+        base_url="https://provider.example/v1",
+        credential_key="",
+    )
+    server = build_server(
+        TaskOrchestrator([agent], client=_ProviderStoppedStreamingClient(max_retries=0)),
+        port=0,
+        security=SecurityConfig(auth_token="secret_token"),
+    )
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    request = urllib.request.Request(
+        f"http://127.0.0.1:{server.server_address[1]}/v1/chat/completions",
+        data=json.dumps(
+            {
+                "model": "provider-model",
+                "mode": "route",
+                "stream": True,
+                "messages": [{"role": "user", "content": "send this message"}],
+            }
+        ).encode("utf-8"),
+        headers={
+            "authorization": "Bearer secret_token",
+            "content-type": "application/json",
+            "connection": "close",
+        },
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=5) as response:
+            status = response.status
+            body = response.read().decode("utf-8")
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+    assert status == 200
+    assert '"code": "tool_execution_stopped"' in body
+    assert '"finish_reason": "error"' in body
+    assert body.endswith("data: [DONE]\n\n")
+
+
 class _StoppedStreamingClient(ModelClient):
     """Raise a structured fail-closed decision from the live stream path."""