diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx
index f1542b75..a92b9003 100644
--- a/docs/advanced/configuration.mdx
+++ b/docs/advanced/configuration.mdx
@@ -35,6 +35,14 @@ Configure Strix using environment variables or a config file.
Maximum number of retries for LLM API calls on transient failures.
+
+ Optional fallback model used after repeated content-guardrail denials within one agent's lifecycle. Unset disables this behavior.
+
+
+
+ Number of content-guardrail denials before the agent switches to `STRIX_LLM_FALLBACK` for the rest of its lifecycle.
+
+
Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode.
diff --git a/strix/config/settings.py b/strix/config/settings.py
index 42a2c97e..190271b6 100644
--- a/strix/config/settings.py
+++ b/strix/config/settings.py
@@ -48,6 +48,12 @@ class LlmSettings(BaseSettings):
default=False,
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
)
+ # A model to fall back to for the rest of an agent's lifecycle once it has
+ # been content-denied ``denied_retries`` times (e.g. a ChatGPT-subscription
+ # cyber-risk guardrail block). Unset disables the fallback: a denial stays
+ # terminal for that agent as before.
+ fallback_model: str | None = Field(default=None, alias="STRIX_LLM_FALLBACK")
+ denied_retries: int = Field(default=3, ge=0, alias="STRIX_LLM_DENIED_RETRIES")
prompt_cache: bool = Field(
default=True,
alias="STRIX_PROMPT_CACHE",
diff --git a/strix/core/agents.py b/strix/core/agents.py
index c96204df..fc98c076 100644
--- a/strix/core/agents.py
+++ b/strix/core/agents.py
@@ -53,6 +53,10 @@ class AgentCoordinator:
self.errors: dict[str, str] = {}
self.recovery_counts: dict[str, int] = {}
self.idle_resume_counts: dict[str, int] = {}
+ self.denial_counts: dict[str, int] = {}
+ self.denial_fallback: set[str] = set()
+ self._denial_fallback_model: str | None = None
+ self._denied_retries: int = 3
self.wait_kinds: dict[str, WaitKind] = {}
self.runtimes: dict[str, AgentRuntime] = {}
self._parent_notified: set[str] = set()
@@ -67,6 +71,19 @@ class AgentCoordinator:
def set_snapshot_path(self, path: Path) -> None:
self._snapshot_path = path
+ def configure_denial_fallback(self, model: str | None, denied_retries: int) -> None:
+ """Configure the per-agent content denial fallback."""
+ self._denial_fallback_model = model
+ self._denied_retries = denied_retries
+
+ @property
+ def denial_fallback_model(self) -> str | None:
+ return self._denial_fallback_model
+
+ @property
+ def denied_retries(self) -> int:
+ return self._denied_retries
+
def mark_shutting_down(self) -> None:
self.is_shutting_down = True
@@ -243,6 +260,27 @@ class AgentCoordinator:
return
await self._maybe_snapshot()
+ async def record_denial(self, agent_id: str) -> int:
+ """Count a content denial; return the new total."""
+ async with self._lock:
+ count = self.denial_counts.get(agent_id, 0) + 1
+ self.denial_counts[agent_id] = count
+ await self._maybe_snapshot()
+ return count
+
+ async def mark_denial_fallback(self, agent_id: str) -> None:
+ """Mark an agent as using its content denial fallback."""
+ async with self._lock:
+ if agent_id in self.denial_fallback:
+ return
+ self.denial_fallback.add(agent_id)
+ await self._maybe_snapshot()
+
+ async def is_on_denial_fallback(self, agent_id: str) -> bool:
+ """Return whether an agent is using its content denial fallback."""
+ async with self._lock:
+ return agent_id in self.denial_fallback
+
async def set_status(
self, agent_id: str, status: Status | str, *, error: str | None = None
) -> None:
@@ -473,6 +511,8 @@ class AgentCoordinator:
"pending_counts": dict(self.pending_counts),
"recovery_counts": dict(self.recovery_counts),
"idle_resume_counts": dict(self.idle_resume_counts),
+ "denial_counts": dict(self.denial_counts),
+ "denial_fallback": sorted(self.denial_fallback),
"wait_kinds": dict(self.wait_kinds),
"mailboxes": {
aid: [dict(m) for m in runtime.mailbox]
@@ -495,6 +535,8 @@ class AgentCoordinator:
self.errors = dict(snap.get("errors", {}))
self.recovery_counts = dict(snap.get("recovery_counts", {}))
self.idle_resume_counts = dict(snap.get("idle_resume_counts", {}))
+ self.denial_counts = dict(snap.get("denial_counts", {}))
+ self.denial_fallback = set(snap.get("denial_fallback", []))
self.wait_kinds = dict(snap.get("wait_kinds", {}))
mailboxes = snap.get("mailboxes", {})
if isinstance(mailboxes, dict):
diff --git a/strix/core/execution.py b/strix/core/execution.py
index bd99e7c3..09bb5559 100644
--- a/strix/core/execution.py
+++ b/strix/core/execution.py
@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
import contextlib
+import dataclasses
import logging
import uuid
from collections.abc import Callable
@@ -644,6 +645,11 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
compactions = 0
model_retries = 0
while True:
+ active_run_config = run_config
+ if coordinator.denial_fallback_model and await coordinator.is_on_denial_fallback(agent_id):
+ active_run_config = dataclasses.replace(
+ run_config, model=coordinator.denial_fallback_model
+ )
stream: Any = None
pre_run_items: list[Any] = []
try:
@@ -656,7 +662,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
except Exception:
logger.exception("image-budget enforcement failed for %s", agent_id)
try:
- await _compact_session(agent, session, run_config, force=False)
+ await _compact_session(agent, session, active_run_config, force=False)
except Exception:
logger.exception("proactive compaction failed for %s", agent_id)
with contextlib.suppress(Exception):
@@ -664,7 +670,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
stream = Runner.run_streamed(
agent,
input=input_data,
- run_config=run_config,
+ run_config=active_run_config,
context=context,
max_turns=max_turns,
session=session,
@@ -744,7 +750,9 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
and is_context_overflow(exc)
):
try:
- compacted = await _compact_session(agent, session, run_config, force=True)
+ compacted = await _compact_session(
+ agent, session, active_run_config, force=True
+ )
except Exception:
logger.exception("overflow compaction recovery failed for %s", agent_id)
compacted = False
@@ -757,6 +765,31 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
)
input_data = []
continue
+ if (
+ coordinator.denial_fallback_model
+ and codex.is_content_guardrail_error(exc)
+ and not await coordinator.is_on_denial_fallback(agent_id)
+ ):
+ denials = await coordinator.record_denial(agent_id)
+ if denials >= coordinator.denied_retries:
+ await coordinator.mark_denial_fallback(agent_id)
+ logger.warning(
+ "agent %s hit %d content denial(s); falling back to %s for the rest "
+ "of its lifecycle",
+ agent_id,
+ denials,
+ coordinator.denial_fallback_model,
+ )
+ else:
+ logger.warning(
+ "agent %s content-denied (%d/%d); replaying the turn",
+ agent_id,
+ denials,
+ coordinator.denied_retries,
+ )
+ if session is not None:
+ input_data = []
+ continue
if model_retries < _MAX_TRANSIENT_MODEL_RETRIES and _is_transient_model_error(exc):
model_retries += 1
delay = _transient_model_retry_delay(model_retries)
diff --git a/strix/core/runner.py b/strix/core/runner.py
index 8726f819..38ba6522 100644
--- a/strix/core/runner.py
+++ b/strix/core/runner.py
@@ -175,6 +175,10 @@ async def run_strix_scan(
if coordinator is None:
coordinator = AgentCoordinator()
coordinator.set_snapshot_path(agents_path)
+ coordinator.configure_denial_fallback(
+ getattr(settings.llm, "fallback_model", None),
+ getattr(settings.llm, "denied_retries", 3),
+ )
from strix.tools.notes.tools import hydrate_notes_from_disk
from strix.tools.todo.tools import hydrate_todos_from_disk
diff --git a/tests/test_denied_retry_fallback.py b/tests/test_denied_retry_fallback.py
new file mode 100644
index 00000000..084d5678
--- /dev/null
+++ b/tests/test_denied_retry_fallback.py
@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from agents import RunConfig, Runner
+
+from strix.config import codex
+from strix.core import execution
+from strix.core.agents import AgentCoordinator
+
+
+class _FakeStream:
+ def __init__(self, exc: BaseException | None = None) -> None:
+ self._exc = exc
+ self.run_loop_exception: BaseException | None = None
+
+ async def stream_events(self) -> Any:
+ if self._exc is not None:
+ raise self._exc
+ events: list[Any] = []
+ for event in events:
+ yield event
+
+
+def _patch_fast_backoff(monkeypatch: pytest.MonkeyPatch) -> None:
+ monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0)
+ monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0)
+
+
+def _guardrail_stream() -> _FakeStream:
+ return _FakeStream(codex.CodexContentGuardrailError("gpt-5.6-sol"))
+
+
+async def _run_once(
+ monkeypatch: pytest.MonkeyPatch,
+ streams: list[_FakeStream],
+ *,
+ fallback_model: str | None = None,
+ denied_retries: int = 3,
+ primary_model: str = "openai/gpt-5.6-sol",
+) -> tuple[Any, list[str | None], AgentCoordinator]:
+ _patch_fast_backoff(monkeypatch)
+ calls: list[str | None] = []
+
+ def _fake_run_streamed(*_args: Any, **kwargs: Any) -> _FakeStream:
+ run_config = kwargs["run_config"]
+ calls.append(run_config.model)
+ return streams[len(calls) - 1]
+
+ monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed)
+
+ coordinator = AgentCoordinator()
+ await coordinator.register("root", "strix", parent_id=None)
+ if fallback_model is not None:
+ coordinator.configure_denial_fallback(fallback_model, denied_retries)
+
+ result = await execution._run_cycle(
+ object(),
+ coordinator,
+ "root",
+ input_data="task",
+ run_config=RunConfig(model=primary_model),
+ context={},
+ max_turns=5,
+ session=None,
+ interactive=False,
+ event_sink=None,
+ hooks=None,
+ )
+ return result, calls, coordinator
+
+
+@pytest.mark.asyncio
+async def test_run_cycle_falls_back_after_repeated_content_denials(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ streams = [_guardrail_stream() for _ in range(3)] + [_FakeStream()]
+ result, models, coordinator = await _run_once(
+ monkeypatch,
+ streams,
+ fallback_model="openai/gpt-5.4",
+ )
+
+ assert result is streams[3]
+ assert models == ["openai/gpt-5.6-sol"] * 3 + ["openai/gpt-5.4"]
+ assert await coordinator.is_on_denial_fallback("root") is True
+
+
+@pytest.mark.asyncio
+async def test_run_cycle_does_not_retry_guardrail_without_fallback(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ guardrail = codex.CodexContentGuardrailError("gpt-5.6-sol")
+ with pytest.raises(codex.CodexContentGuardrailError):
+ await _run_once(monkeypatch, [_FakeStream(guardrail), _FakeStream()])
+
+
+@pytest.mark.asyncio
+async def test_run_cycle_switches_on_first_denial_at_boundary(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ streams = [_guardrail_stream(), _FakeStream()]
+ result, models, coordinator = await _run_once(
+ monkeypatch,
+ streams,
+ fallback_model="openai/gpt-5.4",
+ denied_retries=1,
+ )
+
+ assert result is streams[1]
+ assert models == ["openai/gpt-5.6-sol", "openai/gpt-5.4"]
+ assert await coordinator.is_on_denial_fallback("root") is True
+
+
+@pytest.mark.asyncio
+async def test_denial_fallback_state_round_trips_through_snapshot() -> None:
+ coordinator = AgentCoordinator()
+ await coordinator.register("root", "strix", parent_id=None)
+ coordinator.configure_denial_fallback("openai/gpt-5.4", 3)
+ await coordinator.record_denial("root")
+ await coordinator.mark_denial_fallback("root")
+
+ restored = AgentCoordinator()
+ await restored.restore(await coordinator.snapshot())
+
+ assert restored.denial_counts == {"root": 1}
+ assert await restored.is_on_denial_fallback("root") is True