Merge master (exp-v0.52.354) into stage/7761-watcher-fence

# Conflicts:
#	CHANGELOG.md

CHANGELOG.md

@@ -32,6 +32,12 @@
   leaves.** Final unsubscribe and queue eviction now invalidate the cache and fence projections that
   were already in flight, without holding the lock across database reads or SSE writes, so a client
   that re-subscribes gets a fresh snapshot instead of a stale one. (#7761 by @ruizanthony)
+- **A burst of "session busy" refusals no longer drops a background-task completion.** When
+  `start_session_turn()` refused a completion wake-up with a transient 409 (Agent runtime stale,
+  process wake-ups paused, or the session busy with another turn), the bridge released the durable
+  claim as a plain failure, so a few refusals in a row could terminally drop a completion whose
+  session was alive and waiting. Transient refusals now return the claim as retryable, while hard
+  failures still use up the attempt budget. (#7758 by @ruizanthony)
 - **A first visit now uses the browser's language.** The server stores "no preference" as `null`
   instead of defaulting to `"en"`, so a first-time visitor gets `navigator.language` while an
   explicitly saved language (including English) still wins, and legacy `settings.json` files that

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,6 +1082,10 @@ def _start_async_delegation_wakeup_turn(
                 )
                 return
 
+            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: "

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,15 +452,45 @@ 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
 
+            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(

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