mirror of
https://github.com/usestrix/strix.git
synced 2026-10-14 04:08:06 +00:00
refactor(core): revive any failed agent, without inspecting why it died
Recovery was keyed on recognising a provider billing rejection by its response body. That is a guess about one provider's prose: reword the message, add a provider, and detection silently fails back to the original bug. It also had to be right about the cause, when the cause was never the point. The stranding is general. A failed agent is gated behind a user message and the user can only realistically reach the root, so any child that dies stays dead for the rest of the scan - however it died. Credit exhaustion was only conspicuous because it took eighteen agents down at once. So revive on evidence the scan works rather than on a theory of what broke: a turn that lands anywhere buys every casualty another attempt, bounded by a budget the agent refunds by landing a turn of its own. That budget is what a cause analysis was standing in for, and it holds for causes nobody enumerated. It joins the bounded-recovery limits already governing retries, compaction and tool nudges, all of which have run by the time an agent reaches this state. Reviving off live status rather than a marker set at failure time also removes a race: a marker could be cleared by a recovery that overtook the agent recording its own status, stranding it for good. A billing 400 that LiteLLM re-wraps as a 500 mid-stream now spends the transient retry budget before failing. That is the honest reading - a 500 is retryable by contract, and only the prose said otherwise - and it costs bounded wall-clock, no tokens, and self-heals through revival. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
87fe9af443
commit
f0727b9653
4 changed files with 258 additions and 309 deletions
|
|
@ -39,7 +39,6 @@ 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:
|
||||
|
|
@ -54,6 +53,7 @@ class AgentCoordinator:
|
|||
self.errors: dict[str, str] = {}
|
||||
self.recovery_counts: dict[str, int] = {}
|
||||
self.idle_resume_counts: dict[str, int] = {}
|
||||
self.revival_counts: dict[str, int] = {}
|
||||
self.wait_kinds: dict[str, WaitKind] = {}
|
||||
self.runtimes: dict[str, AgentRuntime] = {}
|
||||
self._parent_notified: set[str] = set()
|
||||
|
|
@ -121,49 +121,54 @@ 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_failed_agents(self, reason: str, *, limit: int) -> list[str]:
|
||||
"""Give agents that died a bounded second chance while the scan is healthy.
|
||||
|
||||
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.
|
||||
A failed agent is gated behind a user message, and the user can only
|
||||
realistically reach the root, so a child that dies stays dead for the rest
|
||||
of the scan - however it died, and however brief the cause was. A turn that
|
||||
lands anywhere is the scan's own evidence that whatever broke may be over,
|
||||
which is worth another attempt. ``limit`` is what separates that from a
|
||||
livelock: an agent that is genuinely broken drains its budget and stays
|
||||
down. The self-recovery layers - retries, compaction, tool nudges - have all
|
||||
already run by the time an agent reaches this state, so a revival is a last
|
||||
chance rather than a first one.
|
||||
"""
|
||||
revived: list[str] = []
|
||||
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:
|
||||
for agent_id, status in self.statuses.items():
|
||||
if status not in {"failed", "crashed"}:
|
||||
continue
|
||||
if self.revival_counts.get(agent_id, 0) >= limit:
|
||||
continue
|
||||
self.revival_counts[agent_id] = self.revival_counts.get(agent_id, 0) + 1
|
||||
# ``set_status`` only retires an error on the way to "running", and
|
||||
# these go to "waiting" first; the outage they name is over.
|
||||
# these go to "waiting" first.
|
||||
self.errors.pop(agent_id, None)
|
||||
revived.append(agent_id)
|
||||
for agent_id in revived:
|
||||
# "waiting" clears the human-wake gate; the message then wakes the
|
||||
# loop that is parked in ``wait_for_message``.
|
||||
# "waiting" clears the human-wake gate; the message then wakes the loop
|
||||
# parked in ``wait_for_message``.
|
||||
await self.set_status(agent_id, "waiting")
|
||||
await self.send(
|
||||
agent_id,
|
||||
{"from": "system", "type": "provider_recovered", "content": reason},
|
||||
{"from": "system", "type": "revived", "content": reason},
|
||||
interrupt=False,
|
||||
)
|
||||
if revived:
|
||||
logger.info("provider outage cleared; revived %d agent(s)", len(revived))
|
||||
logger.info("scan recovered; revived %d failed agent(s)", len(revived))
|
||||
return revived
|
||||
|
||||
async def reset_revivals(self, agent_id: str) -> None:
|
||||
"""Retire an agent's revival budget once it completes a turn of its own."""
|
||||
async with self._lock:
|
||||
if self.revival_counts.pop(agent_id, None) is None:
|
||||
return
|
||||
await self._maybe_snapshot()
|
||||
|
||||
@property
|
||||
def provider_outage_active(self) -> bool:
|
||||
return any(runtime.provider_outage for runtime in self.runtimes.values())
|
||||
def has_failed_agents(self) -> bool:
|
||||
return any(status in {"failed", "crashed"} for status in self.statuses.values())
|
||||
|
||||
async def reset_budget_stops(
|
||||
self,
|
||||
|
|
@ -518,6 +523,7 @@ class AgentCoordinator:
|
|||
"pending_counts": dict(self.pending_counts),
|
||||
"recovery_counts": dict(self.recovery_counts),
|
||||
"idle_resume_counts": dict(self.idle_resume_counts),
|
||||
"revival_counts": dict(self.revival_counts),
|
||||
"wait_kinds": dict(self.wait_kinds),
|
||||
"mailboxes": {
|
||||
aid: [dict(m) for m in runtime.mailbox]
|
||||
|
|
@ -540,6 +546,7 @@ 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.revival_counts = dict(snap.get("revival_counts", {}))
|
||||
self.wait_kinds = dict(snap.get("wait_kinds", {}))
|
||||
mailboxes = snap.get("mailboxes", {})
|
||||
if isinstance(mailboxes, dict):
|
||||
|
|
|
|||
|
|
@ -119,6 +119,15 @@ async def _compact_session(
|
|||
)
|
||||
|
||||
|
||||
# Revivals granted to an agent that died, spent whenever the scan proves healthy
|
||||
# again and refunded once the agent lands a turn of its own.
|
||||
_MAX_AGENT_REVIVALS = 3
|
||||
|
||||
AGENT_REVIVED_MESSAGE = (
|
||||
"[Recovery] Your last turn failed and the scan has since recovered. Continue "
|
||||
"your current task from where it stopped."
|
||||
)
|
||||
|
||||
_MAX_TRANSIENT_MODEL_RETRIES = 5
|
||||
_TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0
|
||||
_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 90.0
|
||||
|
|
@ -129,53 +138,9 @@ 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
|
||||
):
|
||||
|
|
@ -833,17 +798,6 @@ 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))
|
||||
|
|
@ -866,8 +820,13 @@ 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)
|
||||
# A turn that lands proves the scan is working, which retires this
|
||||
# agent's own revival budget and buys one for anything still down.
|
||||
await coordinator.reset_revivals(agent_id)
|
||||
if coordinator.has_failed_agents:
|
||||
await coordinator.revive_failed_agents(
|
||||
AGENT_REVIVED_MESSAGE, limit=_MAX_AGENT_REVIVALS
|
||||
)
|
||||
return cast("RunResultBase | None", stream)
|
||||
|
||||
|
||||
|
|
|
|||
206
tests/test_execution_agent_revival.py
Normal file
206
tests/test_execution_agent_revival.py
Normal file
|
|
@ -0,0 +1,206 @@
|
|||
"""Bounded revival of agents that died, driven by the scan recovering around them.
|
||||
|
||||
A failed agent is gated behind a user message and the user can only realistically
|
||||
reach the root, so without this a child that dies stays dead for the rest of the
|
||||
scan. Drawn from a run where a provider outage failed eighteen agents in thirteen
|
||||
seconds and only the root ever came back.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, cast
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from agents import RunConfig, Runner
|
||||
from agents.exceptions import AgentsException
|
||||
from openai import APIError, BadRequestError
|
||||
|
||||
from strix.core import execution
|
||||
from strix.core.agents import AgentCoordinator
|
||||
|
||||
|
||||
def _request() -> httpx.Request:
|
||||
return httpx.Request("POST", "https://api.anthropic.com/v1/messages")
|
||||
|
||||
|
||||
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,
|
||||
stream: _FakeStream,
|
||||
) -> Any:
|
||||
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0)
|
||||
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0)
|
||||
monkeypatch.setattr(Runner, "run_streamed", lambda *_a, **_k: stream)
|
||||
return 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,
|
||||
)
|
||||
|
||||
|
||||
async def _fleet(children: int = 3) -> AgentCoordinator:
|
||||
coordinator = AgentCoordinator()
|
||||
await coordinator.register("root", "strix", parent_id=None)
|
||||
for index in range(children):
|
||||
await coordinator.register(f"child{index}", f"hunter {index}", parent_id="root")
|
||||
return coordinator
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_landed_turn_revives_the_agents_that_died(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""The reported scenario: the whole fleet is down and only the root is reachable."""
|
||||
coordinator = await _fleet()
|
||||
for agent_id in ("root", "child0", "child1", "child2"):
|
||||
await _run_cycle(
|
||||
monkeypatch, coordinator, agent_id, _FakeStream(exc=AgentsException("provider said no"))
|
||||
)
|
||||
# Every one of them is gated behind a user message it will never get.
|
||||
assert all(coordinator.runtimes[f"child{index}"].user_wake_required for index in range(3))
|
||||
|
||||
# Whatever broke is fixed and the user messages the root, all they can reach.
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
|
||||
for index in range(3):
|
||||
child = f"child{index}"
|
||||
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.AGENT_REVIVED_MESSAGE
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc",
|
||||
[
|
||||
AgentsException("sdk gave up"),
|
||||
APIError("provider rejected the request", _request(), body=None),
|
||||
BadRequestError("bad", response=httpx.Response(400, request=_request()), body=None),
|
||||
RuntimeError("something nobody anticipated"),
|
||||
],
|
||||
ids=["agents", "api", "bad-request", "unknown"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_revival_does_not_care_how_the_agent_died(
|
||||
monkeypatch: pytest.MonkeyPatch, exc: BaseException
|
||||
) -> None:
|
||||
"""Cause of death is never inspected - only whether the scan works now."""
|
||||
coordinator = await _fleet(children=1)
|
||||
await _run_cycle(monkeypatch, coordinator, "child0", _FakeStream(exc=exc))
|
||||
assert coordinator.statuses["child0"] in {"failed", "crashed"}
|
||||
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
|
||||
assert coordinator.statuses["child0"] == "waiting"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_agent_that_keeps_dying_drains_its_budget(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A broken agent must not fail and revive forever."""
|
||||
coordinator = await _fleet(children=1)
|
||||
|
||||
for attempt in range(execution._MAX_AGENT_REVIVALS):
|
||||
await coordinator.set_status("child0", "failed", error="down again")
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
assert coordinator.statuses["child0"] == "waiting", f"revival {attempt} was refused"
|
||||
|
||||
await coordinator.set_status("child0", "failed", error="down again")
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
|
||||
assert coordinator.statuses["child0"] == "failed"
|
||||
assert coordinator.revival_counts["child0"] == execution._MAX_AGENT_REVIVALS
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_turn_of_its_own_refunds_the_budget(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
coordinator = await _fleet(children=1)
|
||||
await coordinator.set_status("child0", "failed", error="down")
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
assert coordinator.revival_counts["child0"] == 1
|
||||
|
||||
# The revived agent picks its task back up and completes a turn.
|
||||
await _run_cycle(monkeypatch, coordinator, "child0", _FakeStream())
|
||||
|
||||
assert "child0" not in coordinator.revival_counts
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_agent_that_dies_after_the_recovery_is_still_revived(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Ordering must not strand anyone: revival is driven by live status, not a flag.
|
||||
|
||||
A marker set at failure time can be cleared by a recovery that raced ahead of
|
||||
the agent recording its own status, leaving it permanently unrevivable.
|
||||
"""
|
||||
coordinator = await _fleet(children=1)
|
||||
|
||||
# A turn lands while nothing is down at all.
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
# Only afterwards does the child die.
|
||||
await _run_cycle(monkeypatch, coordinator, "child0", _FakeStream(exc=AgentsException("late")))
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
|
||||
assert coordinator.statuses["child0"] == "waiting"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_healthy_agents_are_left_alone(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
coordinator = await _fleet(children=2)
|
||||
await coordinator.set_status("child0", "waiting")
|
||||
await coordinator.set_status("child1", "completed")
|
||||
|
||||
await _run_cycle(monkeypatch, coordinator, "root", _FakeStream())
|
||||
|
||||
assert coordinator.statuses["child0"] == "waiting"
|
||||
assert coordinator.statuses["child1"] == "completed"
|
||||
assert coordinator.pending_counts["child0"] == 0
|
||||
assert coordinator.revival_counts == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revival_is_a_no_op_with_nothing_down() -> None:
|
||||
coordinator = await _fleet(children=1)
|
||||
|
||||
assert coordinator.has_failed_agents is False
|
||||
assert await coordinator.revive_failed_agents("recovered", limit=3) == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_budget_survives_a_resume() -> None:
|
||||
"""A snapshot that drops the budget hands a broken agent unlimited revivals."""
|
||||
coordinator = await _fleet(children=1)
|
||||
await coordinator.set_status("child0", "failed", error="down")
|
||||
await coordinator.revive_failed_agents("recovered", limit=3)
|
||||
snapshot = await coordinator.snapshot()
|
||||
|
||||
restored = AgentCoordinator()
|
||||
await restored.restore(snapshot)
|
||||
|
||||
assert restored.revival_counts["child0"] == 1
|
||||
|
|
@ -1,223 +0,0 @@
|
|||
"""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"
|
||||
Loading…
Add table
Reference in a new issue