docs(gateway): say which admission awaits the drain re-check covers

The re-check covers the awaits before the store (Telegram topic recovery,
the busy handler's profile scope), not "the busy handler's later awaits";
the shared-runtime busy path can still await after it stores a wake, and a
drain seen then still dooms the stored copy (the drain gate refuses a
claimed turn, the pending discard drops a queued follow-up). The
WakeNotAccepted docstring lists runtime_draining from the adapter path. The
busy-handler test is documented as a guard for a post-store await, asserts
that the post-admission re-check is the one that fired, and loses a no-op
sleep(0).
This commit is contained in:
kshitijk4poor authored and kshitij committed 2026-10-26 13:08:08 +05:30
1 parent c33731aacb
commit e82bf01eec
2 files changed
+12 -10

No files matched your search

+6 -5
View File
@@ -47,7 +47,8 @@ PERMANENT_REFUSAL_REASONS = frozenset({'not_found', 'invalid_params'})
class WakeNotAccepted(RuntimeError):
"""No adapter admission: retry without treating a healthy chat as dead. ``reason`` is the
store's refusal code when the session authority refused it (None: the adapter did not accept);
store's refusal code when the session authority refused it, ``runtime_draining`` when the
adapter path refused it during a drain (None: the adapter did not accept);
``permanent`` is set once the refusal is proven final (``refusal_proven_permanent``)."""
def __init__(self, message: str = "internal wake not accepted by adapter", *, reason: Optional[str] = None):
@@ -107,10 +108,10 @@ async def admit_internal_event(adapter: Any, event: Any) -> None:
await adapter.handle_message(event)
if event._gateway_accepted is not True:
raise WakeNotAccepted()
# handle_message can await before it claims or queues the event (Telegram topic recovery)
# and after (the busy handler's later awaits). A drain seen here, whenever it began, still
# dooms the stored event: the drain gate refuses the claimed turn and the queued follow-up
# is discarded. Refuse it so the producer retries it after the restart.
# handle_message can await before it claims or queues the event (Telegram topic recovery, the
# busy handler's profile scope). A drain seen here dooms the stored event: the drain gate refuses
# a claimed turn and the pending discard drops a queued follow-up. Refuse it so the producer
# retries it after the restart.
if _runner_draining(adapter):
raise WakeNotAccepted("gateway began draining during admission; retry after restart",
reason="runtime_draining")
@@ -139,8 +139,9 @@ async def test_wake_is_refused_when_drain_starts_during_admission():
@pytest.mark.asyncio
async def test_wake_is_refused_when_drain_starts_after_the_busy_handler_queued_it():
"""The busy handler can await after it stores the wake: a drain that begins there dooms the
queued follow-up, so the wake must be refused (and retried), not receipted and discarded."""
"""Guard for a busy handler that awaits after it stores the wake (no in-tree handler does today):
a drain that begins there dooms the queued follow-up, so the wake must be refused and retried,
not receipted and discarded."""
runner, adapter, source, _turns = wake_route()
key = adapter._event_session_key(MessageEvent(text="", source=source))
adapter._active_sessions[key] = asyncio.Event() # a turn is still running in this chat
@@ -148,13 +149,13 @@ async def test_wake_is_refused_when_drain_starts_after_the_busy_handler_queued_i
async def store_then_drain(event, session_key):
handled = await store(event, session_key) # the wake is now the pending follow-up
runner._draining = True # /restart begins during the handler's later awaits
await asyncio.sleep(0)
runner._draining = True # /restart begins after the store, before the handler returns
return handled
adapter._busy_session_handler = store_then_drain
with pytest.raises(WakeNotAccepted):
with pytest.raises(WakeNotAccepted) as refused:
await deliver_wake(adapter, text="[kanban] Task t_1 completed.", source=source)
assert "during admission" in str(refused.value) # the post-admission re-check fired
event, text = await runner._run_agent_drain_pending(
{"final_response": "done"}, adapter, source, key
)