fix(background_process): do not consume the delivery budget on transient wake-up refusals
start_session_turn() answers 409 retryable=True while the Agent runtime is stale, 409 process_wakeup_paused while wake-ups are paused, and 409 with active_stream_id while the session is busy with another turn. The bridge released the durable claim as a plain failure each time, so a burst of refusals within seconds terminally dropped a completion whose origin session was alive and waiting. release_async_delegation_delivery(..., retryable=True) now forwards the refund to an Agent core whose release_event_delivery() accepts the `retryable` keyword (NousResearch/hermes-agent#120028), detected through inspect.signature, and falls back to the historical plain release on an older core. Hard rejections keep the bounded budget.
This commit is contained in:
1 parent
b1774b45a9
commit
9b6851eaf1
3 files changed
+228
-3
No files matched your search
@@ -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: "
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user