feat(budget): per-model budget fallback chains

docs/advanced/configuration.mdx

@@ -74,6 +74,45 @@ skill. An explicit model passed to `create_agent` takes precedence over routes.
 }
 ```
 
+### Per-model budgets and fallback chains
+
+Set `llm.model_budgets_usd` and `llm.model_fallbacks` to move every agent using
+an exhausted model to its next configured model. Budgets are cumulative per
+model across the whole scan, so the same rule covers the orchestrator, default
+children, explicit child routes, and skill routes. Fallback targets can have
+their own budget and fallback, creating any number of layers.
+
+```json
+{
+  "llm": {
+    "model": "openai/gpt-5.4",
+    "subagent_model": "z-ai/glm-4.7",
+    "skill_model_routes": [
+      {"skill": "xss", "model": "openai/gpt-5.4"}
+    ],
+    "model_budgets_usd": {
+      "openai/gpt-5.4": 20.0,
+      "z-ai/glm-4.7": 10.0
+    },
+    "model_fallbacks": {
+      "openai/gpt-5.4": "z-ai/glm-4.7",
+      "z-ai/glm-4.7": "deepseek/deepseek-chat"
+    }
+  }
+}
+```
+
+A fallback is checked after a response returns, so a call can slightly
+overshoot its model budget. Strix discards that boundary response before any of
+its tool calls execute, updates the agent's persisted active model, and replays
+the unchanged session with the fallback. This avoids duplicate side effects and
+keeps scan resume on the selected fallback layer. A model budget without a
+fallback only tracks usage; it does not stop the scan. The scan-wide
+`--max-budget-usd` remains a hard stop and takes precedence.
+
+The TUI's bottom-right panel displays cumulative tokens and estimated cost for
+each model actually used, including fallback models.
+
 ## Optional Features
 
 <ParamField path="PERPLEXITY_API_KEY" type="string">

strix/agents/factory.py

@@ -17,8 +17,12 @@
 from pydantic import ValidationError
 
 from strix.agents.prompt import render_system_prompt
-from strix.config.models import _same_provider, uses_chat_completions_tool_schema
 from strix.core.inputs import make_model_settings
+from strix.core.model_routing import (
+    chain_uses_chat_completions_tools,
+    resolve_budget_model,
+    resolve_route_api_key,
+)
 from strix.tools.agents_graph.tools import (
     agent_finish,
     create_agent,
@@ -63,7 +67,7 @@
     from agents import RunContextWrapper
     from agents.tool import FunctionToolResult
 
-    from strix.config.settings import LlmSettings, ReasoningEffort, Settings, SkillModelRoute
+    from strix.config.settings import ReasoningEffort, Settings, SkillModelRoute
 
 
 logger = logging.getLogger(__name__)
@@ -482,22 +486,6 @@ def build_strix_agent(
     )
 
 
-def _resolve_child_api_key(
-    llm: LlmSettings, resolved_model: str, default_model: str
-) -> str | None:
-    """Pick the credential a child should send as a per-call ``api_key``.
-
-    Prefer an explicit ``SUBAGENT_LLM_API_KEY``. Otherwise reuse the
-    orchestrator key only when the child shares the orchestrator's provider;
-    cross-provider children fall back to their provider's ambient env/auth.
-    """
-    if llm.subagent_api_key and llm.subagent_api_key.strip():
-        return llm.subagent_api_key.strip()
-    if llm.api_key and _same_provider(resolved_model, default_model):
-        return llm.api_key
-    return None
-
-
 def make_child_factory(
     *,
     settings: Settings,
@@ -526,20 +514,21 @@ def _factory(
         model: str | None = None,
     ) -> SandboxAgent[Any]:
         route = _matching_route(skills) if model is None else None
-        resolved_model = (
+        configured_model = (
             model
             or (route.model if route is not None else None)
             or llm.subagent_model
             or default_model
         ).strip()
+        resolved_model = resolve_budget_model(configured_model, llm)
         reasoning: ReasoningEffort | None = (
             route.reasoning_effort
             if route is not None and route.reasoning_effort is not None
             else llm.subagent_reasoning_effort
         )
         if reasoning is None:
             reasoning = llm.reasoning_effort
-        child_api_key = _resolve_child_api_key(llm, resolved_model, default_model)
+        child_api_key = resolve_route_api_key(llm, resolved_model, default_model)
         child_model_settings = make_model_settings(
             reasoning,
             model_name=resolved_model,
@@ -553,8 +542,8 @@ def _factory(
             scan_mode=scan_mode,
             is_whitebox=is_whitebox,
             interactive=interactive,
-            chat_completions_tools=uses_chat_completions_tool_schema(
-                resolved_model,
+            chat_completions_tools=chain_uses_chat_completions_tools(
+                configured_model,
                 settings,
             ),
             model=resolved_model,

strix/config/__init__.py

@@ -19,6 +19,7 @@
 from strix.config.settings import (
     IntegrationSettings,
     LlmSettings,
+    ReasoningEffort,
     RuntimeSettings,
     Settings,
     SkillModelRoute,
@@ -29,6 +30,7 @@
 __all__ = [
     "IntegrationSettings",
     "LlmSettings",
+    "ReasoningEffort",
     "RuntimeSettings",
     "Settings",
     "SkillModelRoute",

strix/config/models.py

@@ -125,7 +125,7 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
         # orchestrator credential into differently-routed child providers.
         # Provider-specific environment variables keep concurrent routes apart.
         _mirror_api_key_to_provider_env(llm.model, llm.api_key)
-    for model_name in _subagent_model_names(settings):
+    for model_name in _configured_model_names(settings):
         key = llm.subagent_api_key or (
             llm.api_key if _same_provider(model_name, llm.model) else None
         )
@@ -152,14 +152,24 @@ def _same_provider(first: str | None, second: str | None) -> bool:
     return _provider_prefix(first) == _provider_prefix(second)
 
 
-def _subagent_model_names(settings: Settings) -> set[str]:
+def _configured_model_names(settings: Settings) -> set[str]:
     llm = settings.llm
     names = {route.model for route in llm.skill_model_routes if route.model}
+    names.update(llm.model_budgets_usd)
+    names.update(llm.model_fallbacks)
+    names.update(llm.model_fallbacks.values())
+    if llm.model:
+        names.add(llm.model)
     if llm.subagent_model:
         names.add(llm.subagent_model)
     return names
 
 
+# Kept as a private compatibility alias for callers/tests from earlier releases.
+def _subagent_model_names(settings: Settings) -> set[str]:
+    return _configured_model_names(settings)
+
+
 def _mirror_api_key_to_provider_env(model_name: str | None, api_key: str) -> None:
     if not model_name:
         return

strix/config/settings.py

@@ -2,9 +2,10 @@
 
 from __future__ import annotations
 
+import math
 from typing import Literal
 
-from pydantic import AliasChoices, BaseModel, Field, model_validator
+from pydantic import AliasChoices, BaseModel, Field, field_validator, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -73,6 +74,14 @@ class LlmSettings(BaseSettings):
     )
     subagent_model: str | None = Field(default=None, alias="STRIX_SUBAGENT_MODEL")
     skill_model_routes: list[SkillModelRoute] = Field(default_factory=list)
+    model_budgets_usd: dict[str, float] = Field(
+        default_factory=dict,
+        alias="STRIX_MODEL_BUDGETS_USD",
+    )
+    model_fallbacks: dict[str, str] = Field(
+        default_factory=dict,
+        alias="STRIX_MODEL_FALLBACKS",
+    )
     api_key: str | None = Field(
         default=None,
         validation_alias=AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"),
@@ -99,6 +108,95 @@ class LlmSettings(BaseSettings):
     )
     timeout: int = Field(default=300, alias="LLM_TIMEOUT")
 
+    @field_validator("model_budgets_usd", mode="before")
+    @classmethod
+    def validate_model_budgets(cls, value: object) -> object:
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            raise TypeError("model_budgets_usd must be a model-to-USD mapping")
+        normalized: dict[str, float] = {}
+        for raw_model, raw_budget in value.items():
+            model = str(raw_model).strip()
+            if not model:
+                raise ValueError("model_budgets_usd contains an empty model name")
+            try:
+                budget = float(raw_budget)
+            except (TypeError, ValueError) as exc:
+                raise ValueError(f"budget for {model!r} must be a number") from exc
+            if not math.isfinite(budget) or budget <= 0:
+                raise ValueError(f"budget for {model!r} must be finite and greater than 0")
+            normalized[model] = budget
+        return normalized
+
+    @field_validator("model_fallbacks", mode="before")
+    @classmethod
+    def validate_model_fallbacks(cls, value: object) -> object:
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            raise TypeError("model_fallbacks must be a model-to-model mapping")
+        normalized: dict[str, str] = {}
+        for raw_model, raw_fallback in value.items():
+            model = str(raw_model).strip()
+            fallback = str(raw_fallback).strip()
+            if not model or not fallback:
+                raise ValueError("model_fallbacks cannot contain empty model names")
+            if model.lower() == fallback.lower():
+                raise ValueError(f"model {model!r} cannot fall back to itself")
+            normalized[model] = fallback
+        return normalized
+
+    @model_validator(mode="after")
+    def validate_fallback_graph(self) -> LlmSettings:
+        budget_keys = {name.lower() for name in self.model_budgets_usd}
+        fallbacks = {name.lower(): target.lower() for name, target in self.model_fallbacks.items()}
+        display_names = {name.lower(): name for name in self.model_fallbacks}
+        for source in fallbacks:
+            if source not in budget_keys:
+                raise ValueError(
+                    f"model_fallbacks source {display_names[source]!r} requires a "
+                    "model_budgets_usd entry"
+                )
+            seen: set[str] = set()
+            current = source
+            while current in fallbacks:
+                if current in seen:
+                    raise ValueError("model_fallbacks contains a cycle")
+                seen.add(current)
+                current = fallbacks[current]
+        return self
+
+    def budget_for(self, model: str) -> float | None:
+        normalized = model.strip().lower()
+        return next(
+            (
+                budget
+                for name, budget in self.model_budgets_usd.items()
+                if name.lower() == normalized
+            ),
+            None,
+        )
+
+    def fallback_for(self, model: str) -> str | None:
+        normalized = model.strip().lower()
+        return next(
+            (
+                fallback
+                for name, fallback in self.model_fallbacks.items()
+                if name.lower() == normalized
+            ),
+            None,
+        )
+
+    def model_chain(self, model: str) -> list[str]:
+        chain: list[str] = []
+        current: str | None = model.strip()
+        while current:
+            chain.append(current)
+            current = self.fallback_for(current)
+        return chain
+
 
 class RuntimeSettings(BaseSettings):
     model_config = _BASE_CONFIG

strix/core/agents.py

@@ -10,6 +10,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, cast
 
+from strix.core.inputs import make_model_settings
+from strix.core.model_routing import resolve_route_api_key
 from strix.core.sessions import session_write_lock
 
 
@@ -25,6 +27,7 @@
 
 @dataclass(slots=True)
 class AgentRuntime:
+    agent: Any | None = None
     session: Session | None = None
     task: asyncio.Task[Any] | None = None
     stream: Any | None = None
@@ -92,12 +95,15 @@ async def attach_runtime(
         self,
         agent_id: str,
         *,
+        agent: Any | None = None,
         session: Session | None = None,
         task: asyncio.Task[Any] | None = None,
         interrupt_on_message: bool | None = None,
     ) -> None:
         async with self._lock:
             runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            if agent is not None:
+                runtime.agent = agent
             if session is not None:
                 runtime.session = session
             if task is not None:
@@ -111,6 +117,52 @@ async def mark_running(self, agent_id: str) -> None:
                 self.statuses[agent_id] = "running"
         await self._maybe_snapshot()
 
+    async def transition_model(
+        self,
+        previous_model: str,
+        next_model: str,
+        *,
+        llm_settings: Any | None = None,
+    ) -> None:
+        """Persist and apply a scan-wide model fallback to every matching agent."""
+        previous = previous_model.strip().lower()
+        changed: list[str] = []
+        async with self._lock:
+            for agent_id, metadata in self.metadata.items():
+                current = metadata.get("model")
+                if not isinstance(current, str) or current.strip().lower() != previous:
+                    continue
+                metadata["model"] = next_model
+                runtime_agent = self.runtimes.setdefault(agent_id, AgentRuntime()).agent
+                if runtime_agent is not None:
+                    runtime_agent.model = next_model
+                    if llm_settings is not None:
+                        reasoning = (
+                            llm_settings.reasoning_effort
+                            if self.parent_of.get(agent_id) is None
+                            else (
+                                llm_settings.subagent_reasoning_effort
+                                or llm_settings.reasoning_effort
+                            )
+                        )
+                        runtime_agent.model_settings = make_model_settings(
+                            reasoning,
+                            model_name=next_model,
+                            force_required_tool_choice=llm_settings.force_required_tool_choice,
+                            api_key=resolve_route_api_key(
+                                llm_settings, next_model, llm_settings.model or next_model
+                            ),
+                        )
+                changed.append(agent_id)
+        if changed:
+            logger.info(
+                "agent.model fallback %s -> %s (%d agent(s))",
+                previous_model,
+                next_model,
+                len(changed),
+            )
+            await self._maybe_snapshot()
+
     async def park_waiting(self, agent_id: str) -> None:
         await self.set_status(agent_id, "waiting")