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