diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index c2bbb2bb1e..17d875682f 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -4138,21 +4138,30 @@ class BasePlatformAdapter(ABC): # base path instead (returned False, or raised before storing it) and nothing is # queued, start this event. if session_key not in self._active_sessions: - # Busy admission may queue through the runner's delivery adapter. That can be a - # replacement instance after reconnect, so recover from the adapter that owns the - # event rather than reading this intake adapter's slot (and do it exactly once). - delivery_adapter = self + # Busy admission queues through the delivery adapter resolved when it stored the + # event; a reconnect during the handler's later awaits can replace that adapter. + # Recover from whichever slot actually holds it — this one, or the replacement — + # and start it on that slot's owner. A replacement with a live guard drains its + # own slot, so leave that one alone. + owners = [self] runner = getattr(self, "gateway_runner", None) if runner is not None: try: - delivery_adapter = runner._delivery_adapter_for(event.source) or self + delivery_adapter = runner._delivery_adapter_for(event.source) except Exception: + delivery_adapter = None logger.debug("[%s] Delivery-adapter lookup failed during pending recovery", self.name, exc_info=True) - orphan = delivery_adapter.get_pending_message(session_key) - if orphan is not None: - self._start_session_processing(orphan, session_key) - elif not handled: + if (delivery_adapter is not None and delivery_adapter is not self + and session_key not in delivery_adapter._active_sessions): + owners.append(delivery_adapter) + orphan = None + for owner in owners: + orphan = owner.get_pending_message(session_key) + if orphan is not None: + owner._start_session_processing(orphan, session_key) + break + if orphan is None and not handled: event._gateway_accepted = self._start_session_processing(event, session_key) return if handled: diff --git a/tests/gateway/test_busy_followup_guard_release.py b/tests/gateway/test_busy_followup_guard_release.py index 721a100c80..a61683c990 100644 --- a/tests/gateway/test_busy_followup_guard_release.py +++ b/tests/gateway/test_busy_followup_guard_release.py @@ -172,6 +172,7 @@ def _gateway(tmp_path, monkeypatch, route, held): runner.session_store._db = SessionDB(db_path=tmp_path / "state.db") route(runner, monkeypatch, held) runner._wire_adapter_handlers(adapter, message_handler=turns) + adapter.gateway_runner = runner # production _create_adapter binds this key = adapter._event_session_key(_event("probe")) # A returning user: the session exists, so the compression check reads its lock. runner.session_store.get_or_create_session(_event("probe").source) @@ -265,50 +266,63 @@ async def test_followup_stored_before_the_busy_handler_raised_runs_once( await adapter.cancel_background_tasks() +@pytest.mark.parametrize("swap", ["before_enqueue", "after_enqueue"]) @pytest.mark.asyncio -async def test_guard_release_recovers_pending_event_from_replacement_delivery_adapter(): - """A reconnect handoff must drain the delivery adapter's slot, not the intake adapter twice.""" - intake = _Adapter() +async def test_reconnect_during_busy_admission_runs_accepted_followup_once( + tmp_path, monkeypatch, swap +): + """A reconnect can replace the delivery adapter while the busy handler awaits, either + before it stores the follow-up (it lands in the replacement's slot) or after (it stays in + the old adapter's slot). Either way the accepted follow-up must run exactly once.""" + monkeypatch.setenv("HERMES_GATEWAY_BUSY_ACK_ENABLED", "true") + held = _HeldRead(asyncio.get_running_loop()) + adapter, turns, key = _gateway(tmp_path, monkeypatch, _stock_interrupt, held) + runner = adapter.gateway_runner replacement = _Adapter() - pending = _event("M2") - incoming = _event("M1") - session_key = intake._event_session_key(incoming) - started = [] - get_calls = [] + replacement.gateway_runner = runner + runner._wire_adapter_handlers(replacement, message_handler=turns) + ack_entered, ack_release = asyncio.Event(), asyncio.Event() - class _Runner: - def _delivery_adapter_for(self, source): - return replacement + async def _ack(*_args, **_kwargs): + ack_entered.set() + await asyncio.wait_for(ack_release.wait(), timeout=5) - intake.gateway_runner = _Runner() - original_get = replacement.get_pending_message + monkeypatch.setattr(runner, "_send_busy_ack_reply", _ack) + # Publication syncs voice state; this bare runner has none to sync. + monkeypatch.setattr(runner, "_sync_voice_mode_state_to_adapter", lambda _adapter: None) + monkeypatch.setattr(runner, "_bind_voice_input_callback", lambda _adapter: None) + followup = None + try: + adapter.hold_reply = "reply to M1" + await adapter.handle_message(_event("M1")) + await asyncio.wait_for(adapter.reply_held.wait(), timeout=5) + first_turn = adapter._session_tasks[key] + followup = asyncio.create_task(adapter.handle_message(_event("M2"))) + await asyncio.wait_for(held.entered.wait(), timeout=5) + adapter.release_reply.set() + await asyncio.wait_for(asyncio.shield(first_turn), timeout=5) + if swap == "before_enqueue": + runner._publish_primary_adapter(Platform.SIGNAL, replacement) + held.release() + await asyncio.wait_for(ack_entered.wait(), timeout=5) + if swap == "after_enqueue": + assert key in adapter._pending_messages # accepted into the old adapter's slot + runner._publish_primary_adapter(Platform.SIGNAL, replacement) + else: + assert key in replacement._pending_messages + ack_release.set() + await asyncio.wait_for(followup, timeout=5) + await _run_until_idle(adapter, key) + await _run_until_idle(replacement, key) - def _get_pending(key): - get_calls.append(key) - return original_get(key) - - replacement.get_pending_message = _get_pending - - def _start(event, key): - started.append((event, key)) - return True - - intake._start_session_processing = _start - intake._active_sessions[session_key] = asyncio.Event() - - async def _busy_handler(event, key): - replacement._pending_messages[key] = pending - intake._active_sessions.pop(key, None) - return True - - async def _message_handler(event): - return "" - - intake.set_busy_session_handler(_busy_handler) - intake.set_message_handler(_message_handler) - await intake.handle_message(incoming) - - assert started == [(pending, session_key)] - assert get_calls == [session_key] - assert session_key not in replacement._pending_messages - assert session_key not in intake._pending_messages + assert turns.order == ["M1", "M2"] + assert not _orphaned(adapter, key) and not _orphaned(replacement, key) + finally: + held.release() + adapter.release_reply.set() + ack_release.set() + if followup is not None and not followup.done(): + followup.cancel() + await asyncio.gather(followup, return_exceptions=True) + await adapter.cancel_background_tasks() + await replacement.cancel_background_tasks()