fix(core): recover the agent fleet and the UI from a provider credit outage

An exhausted provider balance fails every in-flight agent within seconds, and
a failed agent is gated behind a user message. The user can only realistically
message the root, so the rest of the fleet stayed dead for the life of the scan
even after credits were topped up, and the header kept reporting a failure that
had already resolved.

Treat a billing rejection as its own class of failure:

- Detect it from the response body across the whole exception chain. The status
  code is not a signal: the same rejection arrives as a 400 pre-stream and as a
  LiteLLM-rewrapped 500 mid-stream, and the 500 shape was being replayed through
  the full transient-retry budget on a failure no retry can fix.
- Tag the runtimes it stops, so the first call that succeeds again lifts their
  human-wake gate and hands each agent a note explaining the gap. Their sessions
  and tasks are intact throughout - only the gate kept them parked.
- Report it as one actionable sentence and one log line per agent rather than a
  LiteLLM traceback per agent.

Unlatch the TUI alongside it. `scan_state` had no path back from "failed" and
`controller.error` was never cleared, so a resolved error stayed on screen for
the rest of a working scan; an agent's `error_message` was likewise write-only,
pinning it to a healthy agent. An unchanged error is now announced once rather
than once per retry, since a revived agent reads as "running" the instant it is
woken and again the instant it dies.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam 2026-08-28 20:58:45 +03:00
parent cf179d564e
commit 87fe9af443
7 changed files with 478 additions and 20 deletions

View file

@ -39,6 +39,7 @@ class AgentRuntime:
wake: asyncio.Event = field(default_factory=asyncio.Event)
mailbox: list[dict[str, Any]] = field(default_factory=list)
user_wake_required: bool = False
provider_outage: bool = False
class AgentCoordinator:
@ -120,6 +121,50 @@ class AgentCoordinator:
},
)
async def mark_provider_outage(self, agent_id: str) -> None:
"""Record that the provider, not the agent, is what stopped ``agent_id``."""
async with self._lock:
self.runtimes.setdefault(agent_id, AgentRuntime()).provider_outage = True
async def revive_after_provider_outage(self, reason: str) -> list[str]:
"""Wake every agent a provider outage killed, once a call succeeds again.
An outage on the provider side - no credits, a hard quota - fails every
in-flight agent within seconds, and a failed agent is gated behind a user
message. The user can only realistically message the root, so without this
the rest of the fleet stays dead for the life of the scan even though
nothing is wrong with it. The first call that gets through proves the
outage is over, so lift the gate and tell each agent why it lost time.
"""
async with self._lock:
revived = [
agent_id
for agent_id, runtime in self.runtimes.items()
if runtime.provider_outage and self.statuses.get(agent_id) in {"failed", "crashed"}
]
for runtime in self.runtimes.values():
runtime.provider_outage = False
for agent_id in revived:
# ``set_status`` only retires an error on the way to "running", and
# these go to "waiting" first; the outage they name is over.
self.errors.pop(agent_id, None)
for agent_id in revived:
# "waiting" clears the human-wake gate; the message then wakes the
# loop that is parked in ``wait_for_message``.
await self.set_status(agent_id, "waiting")
await self.send(
agent_id,
{"from": "system", "type": "provider_recovered", "content": reason},
interrupt=False,
)
if revived:
logger.info("provider outage cleared; revived %d agent(s)", len(revived))
return revived
@property
def provider_outage_active(self) -> bool:
return any(runtime.provider_outage for runtime in self.runtimes.values())
async def reset_budget_stops(
self,
*,

View file

@ -129,9 +129,53 @@ def _model_error_status_code(exc: BaseException) -> int | None:
return code if isinstance(code, int) else None
# Providers report an empty account under whatever status code the failing layer
# happens to use - Anthropic returns a 400 pre-stream and LiteLLM re-wraps the
# same body as a 500 mid-stream - so the body text is the only stable signal.
_PROVIDER_CREDIT_MARKERS = (
"credit balance is too low",
"insufficient credits",
"insufficient_quota",
"exceeded your current quota",
"requires more credits",
"billing_hard_limit_reached",
)
PROVIDER_CREDIT_MESSAGE = (
"The model provider rejected the request: the account is out of credits. "
"Add credits, then send any message to resume - the whole agent fleet picks "
"up where it left off."
)
PROVIDER_RECOVERED_MESSAGE = (
"[Provider] Calls were failing because the account was out of credits. "
"Credits are available again - continue your current task from where it stopped."
)
def _is_provider_credit_error(exc: BaseException) -> bool:
"""True when the whole exception chain points at an empty provider account.
Billing rejections are permanent until a human pays, so they must never be
retried like a blip; but they are also not the agent's fault, so they must
not read as a crash either.
"""
seen: set[int] = set()
current: BaseException | None = exc
while current is not None and id(current) not in seen:
seen.add(id(current))
text = str(current).lower()
if any(marker in text for marker in _PROVIDER_CREDIT_MARKERS):
return True
current = current.__cause__ or current.__context__
return False
def _is_transient_model_error(exc: BaseException) -> bool:
if codex.is_content_guardrail_error(exc):
return False
if _is_provider_credit_error(exc):
return False
if isinstance(
exc, APITimeoutError | APIConnectionError | TimeoutError | ConnectionError | OSError
):
@ -789,6 +833,17 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
continue
if session is not None:
await _salvage_stream_to_session(session, pre_run_items, stream, agent_id)
if _is_provider_credit_error(exc):
# Every agent in flight hits this within seconds of each other, so
# log one line rather than a traceback per agent, and tag the runtime
# so the first call that succeeds again can revive the whole fleet.
logger.warning("agent %s stopped: provider account is out of credits", agent_id)
await coordinator.mark_provider_outage(agent_id)
await coordinator.set_status(agent_id, "failed", error=PROVIDER_CREDIT_MESSAGE)
await notify_parent_on_terminal(coordinator, agent_id, "failed")
if not interactive:
raise
return None
if isinstance(exc, ProviderRefusalError):
logger.warning("agent %s refused by the model provider: %s", agent_id, exc)
await coordinator.set_status(agent_id, "failed", error=str(exc))
@ -811,6 +866,8 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
raise
return None
else:
if coordinator.provider_outage_active:
await coordinator.revive_after_provider_outage(PROVIDER_RECOVERED_MESSAGE)
return cast("RunResultBase | None", stream)

View file

@ -5,6 +5,7 @@ from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from strix.interface.tui.live_view import ERROR_BEARING_STATUSES
from strix.interface.tui.live_view import TuiLiveView as BaseLiveView
@ -60,6 +61,11 @@ class TuiLiveView(BaseLiveView):
if error_message and current.get("error_message") != error_message:
current["error_message"] = error_message
changed = True
elif not error_message and status is not None and status not in ERROR_BEARING_STATUSES:
# An error belongs to the failure that produced it. Leaving it attached
# once the agent runs again pins a resolved error to a healthy agent.
if current.pop("error_message", None) is not None:
changed = True
if changed:
current["updated_at"] = now
return changed

View file

@ -17,6 +17,10 @@ from strix.interface.tui.history import load_session_history
from strix.tools.mcp import resolve_mcp_call
# The only statuses an error message can legitimately be attached to.
ERROR_BEARING_STATUSES = frozenset({"failed", "crashed"})
class TuiLiveView:
def __init__(self) -> None:
self.agents: dict[str, dict[str, Any]] = {}
@ -181,6 +185,10 @@ class TuiLiveView:
current["status"] = status
if error_message is not None:
current["error_message"] = error_message
elif status is not None and status not in ERROR_BEARING_STATUSES:
# An error belongs to the failure that produced it. Leaving it attached
# once the agent runs again pins a resolved error to a healthy agent.
current.pop("error_message", None)
current["updated_at"] = now
def record_agent_error(self, agent_id: str, error: str) -> None:

View file

@ -62,7 +62,10 @@ class GoTuiRuntime:
self.scan_task: asyncio.Task[None] | None = None
self.scan_error: BaseException | None = None
self._last_sync_fingerprint = ""
self._error_noted_agents: set[str] = set()
# agent id -> the error already announced for it, so a failure that keeps
# recurring while the user retries is reported once rather than per revive.
self._error_noted_agents: dict[str, str] = {}
self._root_failure_latched = False
self.controller = TuiController(
args,
live_view=self.live_view,
@ -236,12 +239,16 @@ class GoTuiRuntime:
or changed
)
if status in {"failed", "crashed"} and error:
if agent_id not in self._error_noted_agents:
self._error_noted_agents.add(agent_id)
if self._error_noted_agents.get(agent_id) != error:
self._error_noted_agents[agent_id] = error
self.live_view.record_agent_error(agent_id, error)
changed = True
else:
self._error_noted_agents.discard(agent_id)
elif status in {"waiting", "completed"}:
# Only a turn that actually landed retires the note. A revived agent
# reads as "running" the instant it is woken and again the instant it
# dies, so clearing on "running" re-announces an unchanged error once
# per retry - the loop a user sees while waiting out a provider outage.
self._error_noted_agents.pop(agent_id, None)
# The user's opening message waits for the root agent to exist, which is
# the first thing this sync learns about.
@ -250,27 +257,46 @@ class GoTuiRuntime:
roots = [agent_id for agent_id, parent_id in parent_of.items() if parent_id is None]
root_id = roots[0] if roots else None
root_status = statuses.get(root_id) if root_id is not None else None
report_status = (
self.report_state.run_record.get("status") if self.report_state is not None else None
scan_state = self._resolve_scan_state(
root_status, errors.get(root_id) if root_id is not None else None
)
scan_state = self.controller.scan_state
if root_status in {"failed", "crashed"}:
scan_state = "failed"
if root_id is not None and errors.get(root_id):
self.controller.error = errors[root_id]
elif scan_state != "failed":
if report_status == "completed":
scan_state = "completed"
elif root_status == "stopped":
scan_state = "stopped"
elif root_status == "completed":
scan_state = "failed"
self.controller.error = "Scan ended without a completed report"
if scan_state != self.controller.scan_state:
self.controller.scan_state = scan_state
changed = True
return changed
def _resolve_scan_state(self, root_status: str | None, root_error: str | None) -> str:
"""Fold the root agent's status into the banner the scan header shows."""
report_status = (
self.report_state.run_record.get("status") if self.report_state is not None else None
)
if root_status in {"failed", "crashed"}:
self._root_failure_latched = True
if root_error:
self.controller.error = root_error
return "failed"
scan_state = self.controller.scan_state
if self._root_failure_latched:
# The root only runs again because something revived it, so the banner
# its last failure left behind is stale. Nothing else ever lowers
# "failed", which otherwise pins a resolved error - an exhausted API
# balance the user has since topped up, say - on the screen for the rest
# of a scan that is working fine.
self._root_failure_latched = False
self.controller.error = None
scan_state = "running"
if scan_state == "failed":
return scan_state
if report_status == "completed":
return "completed"
if root_status == "stopped":
return "stopped"
if root_status == "completed":
self.controller.error = "Scan ended without a completed report"
return "failed"
return scan_state
def _runtime_sync_fingerprint(self) -> str:
usage: dict[str, Any] = {}
vulnerabilities: list[object] = []

View file

@ -0,0 +1,223 @@
"""Provider-billing failures: classification, fallout, and fleet revival.
Modelled on a real run in which an Anthropic balance ran dry mid-scan and took
eighteen agents down in thirteen seconds.
"""
from __future__ import annotations
from typing import Any, cast
import httpx
import pytest
from agents import RunConfig, Runner
from openai import BadRequestError, InternalServerError
from strix.core import execution
from strix.core.agents import AgentCoordinator
# The body Anthropic actually returns, as it reached the logs.
_CREDIT_BODY = (
'b\'{"type":"error","error":{"type":"invalid_request_error","message":"Your credit '
"balance is too low to access the Anthropic API. Please go to Plans & Billing to "
'upgrade or purchase credits."},"request_id":"req_011CeVWdCD5de48p3k3MFHc1"}\''
)
def _request() -> httpx.Request:
return httpx.Request("POST", "https://api.anthropic.com/v1/messages")
def _credit_bad_request() -> BadRequestError:
"""The pre-stream shape: a clean 400 straight from the provider."""
return BadRequestError(
f"litellm.BadRequestError: AnthropicException - {_CREDIT_BODY}",
response=httpx.Response(400, request=_request()),
body=None,
)
def _credit_midstream() -> InternalServerError:
"""The mid-stream shape: LiteLLM re-wraps the same 400 body as a 500.
Nothing on the surface of this one says "billing": the status code is
retryable and the message names a fallback. Only the body text gives it away.
"""
inner = Exception(f"AnthropicError: {_CREDIT_BODY}")
outer = InternalServerError(
"litellm.MidStreamFallbackError: litellm.InternalServerError: AnthropicException - "
"Your credit balance is too low to access the Anthropic API.",
response=httpx.Response(500, request=_request()),
body=None,
)
outer.__cause__ = inner
return outer
def test_credit_error_detected_in_both_wire_shapes() -> None:
assert execution._is_provider_credit_error(_credit_bad_request()) is True
assert execution._is_provider_credit_error(_credit_midstream()) is True
def test_credit_error_detected_through_a_wrapped_chain() -> None:
root_cause = Exception(_CREDIT_BODY)
middle = RuntimeError("during handling of the above exception")
middle.__cause__ = root_cause
top = RuntimeError("agent run failed")
top.__context__ = middle
assert execution._is_provider_credit_error(top) is True
def test_unrelated_errors_are_not_credit_errors() -> None:
assert execution._is_provider_credit_error(ValueError("nope")) is False
assert (
execution._is_provider_credit_error(
BadRequestError("bad", response=httpx.Response(400, request=_request()), body=None)
)
is False
)
def test_credit_error_is_never_retried_as_transient() -> None:
"""A 500-shaped billing rejection must not read as a server blip."""
assert execution._is_transient_model_error(_credit_midstream()) is False
assert execution._is_transient_model_error(_credit_bad_request()) is False
def test_a_cyclic_exception_chain_terminates() -> None:
first = RuntimeError("a")
second = RuntimeError("b")
first.__cause__ = second
second.__cause__ = first
assert execution._is_provider_credit_error(first) is False
class _FakeStream:
def __init__(self, exc: BaseException | None = None) -> None:
self._exc = exc
self._events: list[Any] = []
self.run_loop_exception: BaseException | None = None
async def stream_events(self) -> Any:
if self._exc is not None:
raise self._exc
for event in self._events:
yield event
async def _run_cycle(
monkeypatch: pytest.MonkeyPatch,
coordinator: AgentCoordinator,
agent_id: str,
streams: list[_FakeStream],
) -> tuple[Any, int]:
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0)
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0)
calls = {"n": 0}
def _fake_run_streamed(*_args: Any, **_kwargs: Any) -> _FakeStream:
stream = streams[calls["n"]]
calls["n"] += 1
return stream
monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed)
result = await execution._run_cycle(
object(),
coordinator,
agent_id,
input_data="task",
run_config=cast("RunConfig", object()),
context={},
max_turns=5,
session=None,
interactive=True,
event_sink=None,
hooks=None,
)
return result, calls["n"]
@pytest.mark.asyncio
async def test_credit_error_fails_once_without_burning_retries(
monkeypatch: pytest.MonkeyPatch,
) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
result, attempts = await _run_cycle(
monkeypatch, coordinator, "root", [_FakeStream(exc=_credit_midstream())]
)
assert result is None
assert attempts == 1, "a billing rejection must not be replayed"
assert coordinator.statuses["root"] == "failed"
assert coordinator.errors["root"] == execution.PROVIDER_CREDIT_MESSAGE
assert coordinator.provider_outage_active is True
@pytest.mark.asyncio
async def test_a_successful_call_revives_the_agents_the_outage_killed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The scenario the user hit: credits are topped up and only the root is reachable."""
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
for index in range(3):
await coordinator.register(f"child{index}", f"hunter {index}", parent_id="root")
# The outage takes the whole fleet down within seconds of itself.
for agent_id in ("root", "child0", "child1", "child2"):
await _run_cycle(
monkeypatch, coordinator, agent_id, [_FakeStream(exc=_credit_bad_request())]
)
assert all(
coordinator.statuses[agent_id] == "failed"
for agent_id in ("root", "child0", "child1", "child2")
)
# Every one of them is gated behind a user message it will never get.
assert all(
coordinator.runtimes[agent_id].user_wake_required
for agent_id in ("child0", "child1", "child2")
)
# The user adds credits and messages the root, which is all they can reach.
await _run_cycle(monkeypatch, coordinator, "root", [_FakeStream()])
for child in ("child0", "child1", "child2"):
assert coordinator.statuses[child] == "waiting", f"{child} was left for dead"
assert coordinator.runtimes[child].user_wake_required is False
assert coordinator.pending_counts[child] == 1
assert child not in coordinator.errors
assert coordinator.runtimes[child].mailbox[0]["content"] == (
execution.PROVIDER_RECOVERED_MESSAGE
)
assert coordinator.provider_outage_active is False
@pytest.mark.asyncio
async def test_revival_leaves_agents_that_died_for_other_reasons_alone(
monkeypatch: pytest.MonkeyPatch,
) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "hunter", parent_id="root")
await _run_cycle(monkeypatch, coordinator, "root", [_FakeStream(exc=_credit_bad_request())])
await coordinator.set_status("child", "failed", error="something else broke")
await _run_cycle(monkeypatch, coordinator, "root", [_FakeStream()])
assert coordinator.statuses["child"] == "failed"
assert coordinator.errors["child"] == "something else broke"
@pytest.mark.asyncio
async def test_revival_is_a_no_op_without_an_outage() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.set_status("root", "failed", error="unrelated")
assert coordinator.provider_outage_active is False
assert await coordinator.revive_after_provider_outage("recovered") == []
assert coordinator.statuses["root"] == "failed"

View file

@ -857,6 +857,99 @@ async def test_agent_state_sync_does_not_mask_root_failure_with_completed_report
assert runtime.controller.error == "finalization failed"
@pytest.mark.asyncio
async def test_root_recovery_clears_the_failure_banner() -> None:
"""A resolved failure must not stay on screen for the rest of the scan."""
runtime = GoTuiRuntime(args())
await runtime.coordinator.register("root", "Strix", parent_id=None)
await runtime.coordinator.set_status("root", "failed", error="out of credits")
await runtime._sync_agent_state()
assert runtime.controller.scan_state == "failed"
# The user tops up their balance and messages the root, which runs again.
await runtime.coordinator.mark_running("root")
await runtime._sync_agent_state()
assert runtime.controller.scan_state == "running"
assert runtime.controller.error is None
assert "error_message" not in runtime.live_view.agents["root"]
@pytest.mark.asyncio
async def test_recovered_scan_can_still_reach_a_terminal_state() -> None:
runtime = GoTuiRuntime(args())
runtime.report_state = cast("Any", SimpleNamespace(run_record={"status": "running"}))
await runtime.coordinator.register("root", "Strix", parent_id=None)
await runtime.coordinator.set_status("root", "failed", error="out of credits")
await runtime._sync_agent_state()
await runtime.coordinator.mark_running("root")
await runtime._sync_agent_state()
assert runtime.controller.scan_state == "running"
runtime.report_state = cast("Any", SimpleNamespace(run_record={"status": "completed"}))
await runtime.coordinator.set_status("root", "completed")
await runtime._sync_agent_state()
assert runtime.controller.scan_state == "completed"
@pytest.mark.asyncio
async def test_a_terminal_scan_failure_survives_a_later_sync() -> None:
"""Un-latching must only undo a root failure, never a scan-level one."""
runtime = GoTuiRuntime(args())
await runtime.coordinator.register("root", "Strix", parent_id=None)
runtime.controller.scan_state = "failed"
runtime.controller.error = "sandbox teardown exploded"
await runtime._sync_agent_state()
assert runtime.controller.scan_state == "failed"
assert runtime.controller.error == "sandbox teardown exploded"
@pytest.mark.asyncio
async def test_an_unchanged_error_is_announced_once_across_retries() -> None:
"""Reviving into the same failure must not repost the same message each time."""
runtime = GoTuiRuntime(args())
await runtime.coordinator.register("root", "Strix", parent_id=None)
for _ in range(3):
await runtime.coordinator.set_status("root", "failed", error="out of credits")
await runtime._sync_agent_state()
# The user messages the root; it starts, then dies on the same error.
await runtime.coordinator.mark_running("root")
await runtime._sync_agent_state()
announced = [
event
for event in runtime.live_view.events
if event.get("data", {}).get("metadata", {}).get("source") == "agent_error"
]
assert len(announced) == 1
@pytest.mark.asyncio
async def test_a_new_error_after_a_landed_turn_is_announced() -> None:
runtime = GoTuiRuntime(args())
await runtime.coordinator.register("root", "Strix", parent_id=None)
await runtime.coordinator.set_status("root", "failed", error="out of credits")
await runtime._sync_agent_state()
# A turn that actually lands settles the agent into "waiting".
await runtime.coordinator.set_status("root", "waiting")
await runtime._sync_agent_state()
await runtime.coordinator.set_status("root", "failed", error="out of credits")
await runtime._sync_agent_state()
announced = [
event
for event in runtime.live_view.events
if event.get("data", {}).get("metadata", {}).get("source") == "agent_error"
]
assert len(announced) == 2
def _direct_launch_args() -> argparse.Namespace:
launch_args = args()
launch_args.needs_setup = False