From 87fe9af443f89c0dad793f3819e34e71aa05ad02 Mon Sep 17 00:00:00 2001 From: 0xallam Date: Fri, 28 Aug 2026 20:58:45 +0300 Subject: [PATCH] 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) --- strix/core/agents.py | 45 +++++ strix/core/execution.py | 57 ++++++ strix/interface/tui/backend/live_view.py | 6 + strix/interface/tui/live_view.py | 8 + strix/interface/tui/runtime.py | 66 +++++-- tests/test_execution_provider_credit.py | 223 +++++++++++++++++++++++ tests/test_go_tui_runtime.py | 93 ++++++++++ 7 files changed, 478 insertions(+), 20 deletions(-) create mode 100644 tests/test_execution_provider_credit.py diff --git a/strix/core/agents.py b/strix/core/agents.py index c96204df..93851c59 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -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, *, diff --git a/strix/core/execution.py b/strix/core/execution.py index 9f2674d2..5c642a9b 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -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) diff --git a/strix/interface/tui/backend/live_view.py b/strix/interface/tui/backend/live_view.py index bc12c034..e86dcfde 100644 --- a/strix/interface/tui/backend/live_view.py +++ b/strix/interface/tui/backend/live_view.py @@ -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 diff --git a/strix/interface/tui/live_view.py b/strix/interface/tui/live_view.py index 7e8f2534..a4f97ee8 100644 --- a/strix/interface/tui/live_view.py +++ b/strix/interface/tui/live_view.py @@ -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: diff --git a/strix/interface/tui/runtime.py b/strix/interface/tui/runtime.py index 451e6ddc..85505742 100644 --- a/strix/interface/tui/runtime.py +++ b/strix/interface/tui/runtime.py @@ -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] = [] diff --git a/tests/test_execution_provider_credit.py b/tests/test_execution_provider_credit.py new file mode 100644 index 00000000..ec402104 --- /dev/null +++ b/tests/test_execution_provider_credit.py @@ -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" diff --git a/tests/test_go_tui_runtime.py b/tests/test_go_tui_runtime.py index ee5c0a23..a02f543c 100644 --- a/tests/test_go_tui_runtime.py +++ b/tests/test_go_tui_runtime.py @@ -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