fix(llm): surface structured provider refusals

strix/core/execution.py

@@ -55,6 +55,23 @@ _INPUT_REJECTION_CODES = frozenset({400, 404, 422})
 _MAX_COMPACTIONS_PER_CYCLE = 2
 
 
+class ProviderRefusalError(AgentsException):
+    """Raised when a provider returns a structured refusal instead of an exception."""
+
+
+def _structured_provider_refusal(result: Any) -> str | None:
+    for item in getattr(result, "new_items", ()) or ():
+        raw_item = getattr(item, "raw_item", None)
+        for content in getattr(raw_item, "content", ()) or ():
+            if getattr(content, "type", None) != "refusal":
+                continue
+            refusal = getattr(content, "refusal", None)
+            if isinstance(refusal, str) and refusal.strip():
+                return refusal.strip()
+            return "The model provider refused this request."
+    return None
+
+
 def _run_config_model(run_config: RunConfig) -> str | None:
     return run_config.model if isinstance(run_config.model, str) else None
 
@@ -490,6 +507,8 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                                 logger.exception("stream event sink failed for %s", agent_id)
                     if stream.run_loop_exception is not None:
                         raise stream.run_loop_exception
+                    if refusal := _structured_provider_refusal(stream):
+                        raise ProviderRefusalError(refusal)
                 except (BudgetExceededError, BudgetPausedError, SubagentBudgetReservedError):
                     raise
                 except RuntimeError as stream_exc:

tests/test_execution.py

@@ -9,8 +9,10 @@ from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
+from agents.items import MessageOutputItem
 from agents.memory import SQLiteSession
 from agents.tool_context import ToolContext
+from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal
 
 from strix.config import codex
 from strix.core import execution
@@ -24,6 +26,30 @@ from strix.core.execution import (
 from strix.tools.finish.tool import finish_scan
 
 
+class _StructuredRefusalStream:
+    def __init__(self, refusal: str) -> None:
+        self.run_loop_exception: BaseException | None = None
+        self.new_items = [
+            MessageOutputItem(
+                agent=MagicMock(),
+                raw_item=ResponseOutputMessage(
+                    id="msg-refusal",
+                    content=[ResponseOutputRefusal(type="refusal", refusal=refusal)],
+                    role="assistant",
+                    status="completed",
+                    type="message",
+                ),
+            )
+        ]
+
+    async def stream_events(self) -> Any:
+        if False:
+            yield None
+
+    def cancel(self, mode: str = "immediate") -> None:  # noqa: ARG002
+        return
+
+
 async def _call_finish_scan(
     coordinator: AgentCoordinator, agent_id: str, parent_id: str | None
 ) -> dict[str, Any]:
@@ -544,6 +570,35 @@ async def test_guardrail_noninteractive_fails_only_blocked_agent(tmp_path: Any)
     session.close()
 
 
+@pytest.mark.asyncio
+async def test_structured_provider_refusal_fails_interactive_agent(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    refusal = "This request was blocked under the provider's usage policy."
+    stream = _StructuredRefusalStream(refusal)
+    monkeypatch.setattr(execution.Runner, "run_streamed", lambda *_args, **_kwargs: stream)
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+
+    result = await execution._run_cycle(
+        MagicMock(),
+        coordinator,
+        "root",
+        input_data="task",
+        run_config=MagicMock(),
+        context={},
+        max_turns=5,
+        session=None,
+        interactive=True,
+        event_sink=None,
+        hooks=None,
+    )
+
+    assert result is None
+    assert coordinator.statuses["root"] == "failed"
+    assert coordinator.errors["root"] == refusal
+
+
 @pytest.mark.asyncio
 async def test_resume_revives_guardrail_parked_child_but_not_plain_waiting(
     tmp_path: Any, monkeypatch: pytest.MonkeyPatch