diff --git a/api/background_process.py b/api/background_process.py index 2a3c581f..6e6323af 100644 --- a/api/background_process.py +++ b/api/background_process.py @@ -1020,6 +1020,29 @@ def _record_async_delegation_accepted( ) +def _wakeup_refusal_is_transient(resp: dict | None, status: int) -> bool: + """True when the target refused the wake-up for a reason it will recover from. + + Such refusals are not delivery failures of the completion itself, so they + must not consume its bounded durable delivery budget. Otherwise a burst of + refusals (e.g. ``agent_runtime_stale`` answers while the Agent runtime is + being replaced) terminally drops a completion whose origin session is alive: + + * any payload that explicitly says ``retryable: True`` (stale runtime); + * the paused-wakeup contract (``process_wakeup_paused``); + * a session busy with another turn (``active_stream_id`` on a 409). + """ + if not isinstance(resp, dict): + return False + if resp.get("retryable") is True: + return True + if status != 409: + return False + if resp.get("error") == "process_wakeup_paused": + return True + return bool(resp.get("active_stream_id")) + + def _start_async_delegation_wakeup_turn( session_id: str, wakeup_prompt: str, @@ -1059,7 +1082,11 @@ def _start_async_delegation_wakeup_turn( ) return - release_async_delegation_delivery(evt, claim) + transient = _wakeup_refusal_is_transient(resp, status) + if transient: + release_async_delegation_delivery(evt, claim, retryable=True) + else: + release_async_delegation_delivery(evt, claim) _retry_unclaimed_async_delegation_event( process_registry, evt, keep_legacy_retrying=True ) @@ -1068,6 +1095,14 @@ def _start_async_delegation_wakeup_turn( "async delegation wakeup paused for session %s; delivery remains retryable", session_id, ) + elif transient: + logger.info( + "async delegation wakeup refused transiently for session %s " + "(status=%s err=%r); attempt refunded, durable retry scheduled", + session_id, + status, + (resp or {}).get("error"), + ) else: logger.debug( "async delegation wakeup not accepted for session %s: " diff --git a/api/process_event_utils.py b/api/process_event_utils.py index 2d9872f6..f6970d6d 100644 --- a/api/process_event_utils.py +++ b/api/process_event_utils.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections import OrderedDict from dataclasses import dataclass +import inspect import logging import math import re @@ -451,16 +452,46 @@ def complete_async_delegation_delivery( _mark_legacy_async_delivery_complete(claim.delegation_id) +def _release_accepts_retryable(release_fn: Any) -> bool: + """Whether the core ``release_event_delivery`` supports ``retryable=``. + + Older Agent cores predate the keyword; calling them with it would raise + ``TypeError`` and skip the release entirely. + """ + try: + params = inspect.signature(release_fn).parameters + except (TypeError, ValueError): + return False + if "retryable" in params: + return params["retryable"].kind in ( + inspect.Parameter.KEYWORD_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ) + return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()) + + def release_async_delegation_delivery( evt: Any, claim: AsyncDelegationDeliveryClaim, + *, + retryable: bool = False, ) -> None: - """Release a failed claim so a later WebUI consumer can retry it.""" + """Release a failed claim so a later WebUI consumer can retry it. + + ``retryable=True`` tells the core the refusal was transient, so the claimed + attempt is refunded instead of counting against the bounded delivery + budget (``tools.async_delegation.release_event_delivery(..., + retryable=True)``). A core that predates the keyword gets a plain release: + the attempt is then consumed as before, never an exception. + """ try: if claim.durable: from tools.async_delegation import release_event_delivery - release_event_delivery(evt, claim.claim_id) + if retryable and _release_accepts_retryable(release_event_delivery): + release_event_delivery(evt, claim.claim_id, retryable=True) + else: + release_event_delivery(evt, claim.claim_id) except Exception: logger.warning( "Failed to release durable async delegation delivery for %s", diff --git a/tests/test_async_delegation_retryable_refusal.py b/tests/test_async_delegation_retryable_refusal.py new file mode 100644 index 00000000..17be2e52 --- /dev/null +++ b/tests/test_async_delegation_retryable_refusal.py @@ -0,0 +1,159 @@ +"""A wake-up refused for a transient reason the WebUI did not cause must not +consume the durable delivery budget. + +Scenario: while the Agent runtime is being replaced, the still-running WebUI +answers every server-side wake-up with ``409 agent_runtime_stale +retryable=True``. Releasing the claim as a plain failure each time exhausts the +bounded budget within seconds and terminally drops a completion whose origin +session is alive and waiting. + +Contract under test (``api/background_process._start_async_delegation_wakeup_turn``): + +* a refusal whose payload says ``retryable: True`` (stale runtime), a paused + wake-up or a session busy with another stream releases the claim with + ``retryable=True`` so the core refunds the attempt; +* every other rejection keeps the historical bounded behaviour; +* an older core whose ``release_event_delivery`` has no ``retryable`` keyword + still gets a plain release (compatibility, never an exception). +""" +from __future__ import annotations + +import sys +import time +import types + +import pytest + +from api import background_process as bp +from api import config as cfg +from api import process_event_utils as peu + + +def _wait_until(predicate, timeout=3.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return bool(predicate()) + + +def _reset() -> None: + cfg.PROCESS_SESSION_INDEX.clear() + cfg.PENDING_BG_TASK_COMPLETIONS.clear() + cfg.BG_TASK_COMPLETE_EVENTS_SEEN.clear() + cfg.DEFERRED_PROCESS_WAKEUPS.clear() + reset = getattr(peu, "_reset_legacy_async_delivery_dedupe_for_tests", None) + if reset is not None: + reset() + + +def _event(): + return { + "type": "async_delegation", + "delegation_id": "deleg_retryable", + "session_key": "webui-session-1", + "status": "completed", + "is_batch": True, + "results": [{"task_index": 0, "status": "completed", "summary": "verdict"}], + "dispatched_at": time.time() - 2, + "completed_at": time.time(), + } + + +class _Registry: + def __init__(self): + import queue + + self.completion_queue = queue.Queue() + + +def _install_core(monkeypatch, *, supports_retryable: bool): + calls = {"release": [], "complete": []} + mod = types.ModuleType("tools.async_delegation") + mod.claim_event_delivery = lambda evt, consumer: f"claim:{consumer}" + mod.complete_event_delivery = lambda evt, claim_id: calls["complete"].append(claim_id) + mod.get_durable_delegation = lambda delegation_id: {"delivery_state": "pending", "delivery_attempts": 1} + mod.restore_undelivered_completions = lambda queue_: 0 + if supports_retryable: + def _release(evt, claim_id, *, retryable=False): + calls["release"].append((claim_id, retryable)) + else: + def _release(evt, claim_id): # pre-refund core: no keyword at all + calls["release"].append((claim_id, None)) + mod.release_event_delivery = _release + pkg = sys.modules.get("tools") or types.ModuleType("tools") + monkeypatch.setitem(sys.modules, "tools", pkg) + monkeypatch.setitem(sys.modules, "tools.async_delegation", mod) + monkeypatch.setattr(pkg, "async_delegation", mod, raising=False) + return calls + + +def _run(monkeypatch, response, *, supports_retryable=True): + _reset() + calls = _install_core(monkeypatch, supports_retryable=supports_retryable) + from api import routes + + monkeypatch.setattr(routes, "start_session_turn", lambda *a, **k: response) + monkeypatch.setattr(bp, "ASYNC_DELIVERY_ROUTING_RETRY_SECONDS", 10.0) + registry = _Registry() + evt = _event() + claim = peu.claim_async_delegation_delivery(evt, "webui-background") + assert claim is not None + try: + bp._start_async_delegation_wakeup_turn( + "webui-session-1", "verdict", delegation_id="deleg_retryable", + evt=evt, claim=claim, process_registry=registry, + ) + assert _wait_until(lambda: len(calls["release"]) == 1) + assert calls["complete"] == [] + return calls["release"][0] + finally: + _reset() + + +@pytest.mark.parametrize("response", [ + {"_status": 409, "error": "runtime changed", "type": "agent_runtime_stale", "retryable": True}, + {"_status": 503, "error": "temporarily unavailable", "retryable": True}, + {"_status": 409, "error": "process_wakeup_paused"}, + {"_status": 409, "error": "session already has an active stream", "active_stream_id": "s1"}, +]) +def test_transient_refusal_releases_with_refund(monkeypatch, response): + _claim, retryable = _run(monkeypatch, response) + assert retryable is True + + +@pytest.mark.parametrize("response", [ + {"_status": 500, "error": "boom"}, + {"_status": 404, "error": "Session not found"}, + {"_status": 400, "error": "message is required"}, + {"_status": 409, "error": "conflict", "retryable": False}, + None, +]) +def test_hard_rejection_keeps_bounded_budget(monkeypatch, response): + _claim, retryable = _run(monkeypatch, response) + assert retryable is False + + +def test_old_core_without_retryable_keyword_still_releases(monkeypatch): + _claim, retryable = _run( + monkeypatch, + {"_status": 409, "error": "x", "type": "agent_runtime_stale", "retryable": True}, + supports_retryable=False, + ) + assert retryable is None # plain release happened, no TypeError escaped + + +def test_real_core_signature_is_detected(): + def new_core(evt, claim_id, *, retryable=False): + return None + + def kwargs_core(evt, claim_id, **kwargs): + return None + + def old_core(evt, claim_id): + return None + + assert peu._release_accepts_retryable(new_core) is True + assert peu._release_accepts_retryable(kwargs_core) is True + assert peu._release_accepts_retryable(old_core) is False