diff --git a/strix/core/agents.py b/strix/core/agents.py index 93851c59..bf993f73 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -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): diff --git a/strix/core/execution.py b/strix/core/execution.py index 5c642a9b..4e801811 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -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) diff --git a/tests/test_execution_agent_revival.py b/tests/test_execution_agent_revival.py new file mode 100644 index 00000000..03504180 --- /dev/null +++ b/tests/test_execution_agent_revival.py @@ -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 diff --git a/tests/test_execution_provider_credit.py b/tests/test_execution_provider_credit.py deleted file mode 100644 index ec402104..00000000 --- a/tests/test_execution_provider_credit.py +++ /dev/null @@ -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"