From 424d07edac932ed7b365fa325705f179b1e2ba1d Mon Sep 17 00:00:00 2001 From: Victor Kyriazakos Date: Thu, 20 Aug 2026 20:21:09 +0000 Subject: [PATCH] =?UTF-8?q?fix(relay):=20prompt-lifecycle=20acks=20are=20f?= =?UTF-8?q?ire-and-forget=20=E2=80=94=20awaiting=20them=20ON=20the=20read?= =?UTF-8?q?=20loop=20self-deadlocked=20the=20transport?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 of the approval-turn stuck-stream hunt. Round 1 (interim-marked acks) fixed the draft-hijack-by-matching path — live logs confirm the absorption fallback no longer fires — but the freeze persisted because of a second, deeper defect on the same codepath: _consume_prompt_response executes ON the transport read loop (inbound frame -> _handle_frame -> _inbound handler). The handler awaited self.send() for its '✅ Approved once' ack — but send() blocks on an outbound_result future that ONLY the read loop can resolve, and the read loop is blocked inside this very handler. Guaranteed self-deadlock for the full outbound timeout (30s) on EVERY button tap. While wedged, everything on the transport starved: draft appends (the frozen stream right after approving), sibling approval-card sends (timed out into 'possibly-delivered' — the observed double-approval ambiguity), and the turn's seal (timed out ambiguous -> plain-send fallback -> duplicate final). Log signature was the tell: card-send timeout at tap time, no absorption INFO, no seal-failed WARNING, no suppression line. Fix: _send_lifecycle_ack() — acks ride a background task with strong ref retention; the handler returns immediately and the read loop keeps consuming, so the ack's own result frame resolves normally. Applied to all six lifecycle sends (approval ack, slash-confirm ack + result text, clarify acks, expiry notice). Acks are cosmetic by contract; failure logs at debug and never breaks the reader. Tests: new deadlock-shape test (gated transport send; handler must return within 1s and the ack must still egress afterwards — RED on the awaited version via TimeoutError at the exact deadlock), prior 3 tests green with a yield for the background task. Targeted sweep 203/203. --- gateway/relay/adapter.py | 78 ++++++++++++++----- .../test_relay_prompt_ack_stream_isolation.py | 41 ++++++++++ 2 files changed, 100 insertions(+), 19 deletions(-) diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index 403f1f58f5..d579c243d1 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -107,6 +107,9 @@ class RelayAdapter(BasePlatformAdapter): # per-chat keying collided three concurrent turns: merged task # cards, clobbered seal state, 3x duplicate finals). self._open_draft_by_chat: Dict[str, int] = {} + # Strong refs for in-flight fire-and-forget lifecycle acks (asyncio + # holds tasks weakly; unreferenced tasks can be GC'd mid-flight). + self._lifecycle_ack_tasks: set = set() # Draft keys whose post-seal tombstone swallow has been logged once # (observability for the hijacked-live-stream class; bounded FIFO # like the sibling caches). @@ -2954,8 +2957,11 @@ class RelayAdapter(BasePlatformAdapter): # Acknowledge in-channel (the connector's prompt message can't # be edited cross-platform yet — edit support varies; a short # confirmation preserves the audit trail the native edit gives). - await self.send( - chat_id, label, metadata=self._prompt_reply_metadata(event) + # Fire-and-forget: we are ON the read loop here (see + # _send_lifecycle_ack) — awaiting the send self-deadlocks the + # transport for the full outbound timeout. + self._send_lifecycle_ack( + chat_id, label, self._prompt_reply_metadata(event) ) if count: self.resume_typing_for_chat(chat_id) @@ -2973,14 +2979,15 @@ class RelayAdapter(BasePlatformAdapter): "always": "🔒 Always approve", "cancel": "❌ Cancelled", }.get(choice, "Resolved") - await self.send( - chat_id, label, metadata=self._prompt_reply_metadata(event) + # Fire-and-forget (read-loop context — see _send_lifecycle_ack). + self._send_lifecycle_ack( + chat_id, label, self._prompt_reply_metadata(event) ) if result_text: - await self.send( + self._send_lifecycle_ack( chat_id, str(result_text), - metadata=self._prompt_reply_metadata(event), + self._prompt_reply_metadata(event), ) elif kind == "clarify": from tools.clarify_gateway import ( @@ -2991,10 +2998,10 @@ class RelayAdapter(BasePlatformAdapter): clarify_id = str(state.get("clarify_id") or "") if option_id == "other": mark_awaiting_text(clarify_id) - await self.send( + self._send_lifecycle_ack( chat_id, "✏️ Type your answer:", - metadata=self._prompt_reply_metadata(event), + self._prompt_reply_metadata(event), ) else: choices = state.get("choices") or [] @@ -3004,10 +3011,10 @@ class RelayAdapter(BasePlatformAdapter): idx = -1 if 0 <= idx < len(choices): resolve_gateway_clarify(clarify_id, str(choices[idx])) - await self.send( + self._send_lifecycle_ack( chat_id, f"✅ {choices[idx]}", - metadata=self._prompt_reply_metadata(event), + self._prompt_reply_metadata(event), ) else: # Unmappable option: flip to text capture so the user @@ -3019,6 +3026,40 @@ class RelayAdapter(BasePlatformAdapter): logger.warning("relay prompt_response resolution failed", exc_info=True) return True + def _send_lifecycle_ack( + self, chat_id: str, text: str, metadata: Dict[str, Any] + ) -> None: + """Fire-and-forget a prompt-lifecycle ack from read-loop context. + + Live finding round 2 (rc.4): _consume_prompt_response executes ON + the transport read loop (inbound frame -> _handle_frame -> the + _inbound handler). ``await self.send(...)`` there is a + SELF-DEADLOCK: send() blocks on an outbound_result future that only + the read loop can resolve — and the read loop is blocked inside + this very handler. Every button tap wedged the transport for the + full outbound timeout: draft appends starved (the observed frozen + stream right after approving), sibling approval-card sends timed + out into 'possibly-delivered' ambiguity, and the turn's seal timed + out ambiguous -> plain-send fallback (the duplicate final). + + Acks are cosmetic by contract (the audit trail), so they ride a + background task: the handler returns immediately, the read loop + keeps consuming, and the ack's own result frame resolves normally. + Failures are logged at debug — an undelivered ack must never break + the reader or the turn. The task ref is retained (asyncio only + weakly references tasks) and dropped on completion. + """ + + async def _ack() -> None: + try: + await self.send(chat_id, text, metadata=metadata) + except Exception: # noqa: BLE001 - ack is best-effort + logger.debug("relay lifecycle ack failed", exc_info=True) + + task = asyncio.create_task(_ack(), name="relay-lifecycle-ack") + self._lifecycle_ack_tasks.add(task) + task.add_done_callback(self._lifecycle_ack_tasks.discard) + async def _notify_prompt_expired(self, event) -> None: """Tell the presser their prompt is no longer waiting. @@ -3029,15 +3070,14 @@ class RelayAdapter(BasePlatformAdapter): chat_id = str(getattr(event.source, "chat_id", "") or "") if not chat_id: return - try: - await self.send( - chat_id, - "⌛ That prompt is no longer waiting for an answer. " - "Send your reply as a normal message.", - metadata=self._prompt_reply_metadata(event), - ) - except Exception: # noqa: BLE001 - notification is best-effort - logger.debug("relay expired-prompt notice failed", exc_info=True) + # Fire-and-forget (read-loop context — see _send_lifecycle_ack): + # _notify_prompt_expired is called from _consume_prompt_response too. + self._send_lifecycle_ack( + chat_id, + "⌛ That prompt is no longer waiting for an answer. " + "Send your reply as a normal message.", + self._prompt_reply_metadata(event), + ) def _prompt_reply_metadata(self, event) -> Dict[str, Any]: """Thread/topic metadata so prompt acks land where the prompt lives. diff --git a/tests/relay/test_relay_prompt_ack_stream_isolation.py b/tests/relay/test_relay_prompt_ack_stream_isolation.py index d18971855e..515ee39f31 100644 --- a/tests/relay/test_relay_prompt_ack_stream_isolation.py +++ b/tests/relay/test_relay_prompt_ack_stream_isolation.py @@ -21,6 +21,7 @@ matching entirely. These tests pin the whole class, plus the regression contract that real turn-finals still absorb into their stream. """ +import asyncio import json import pytest @@ -86,6 +87,44 @@ async def _open_turn_draft(a, chat_id="D01", draft_id=7): class TestPromptAckDoesNotSealDraft: + @pytest.mark.asyncio + async def test_prompt_response_handler_does_not_block_on_ack_send(self): + """Live finding round 2 (rc.4): _consume_prompt_response runs ON the + transport read loop (inbound frame -> _handle_frame -> _inbound). + Awaiting the ack send there is a self-deadlock: the outbound_result + that resolves the send's future arrives on the SAME read loop, which + is blocked inside the handler. Every tap wedged the transport for + the full outbound timeout — draft appends starved (frozen stream), + a second approval card's ack couldn't be read (send timed out, + 'possibly-delivered'), and the seal timed out ambiguous (plain-send + duplicate). The handler must RETURN without awaiting ack delivery; + the ack is best-effort and rides a background task.""" + a = _adapter() + gate = asyncio.Event() + orig = a._transport.send_outbound + + async def gated_send(frame, platform=None): + if frame.get("op") == "send": + await gate.wait() # simulate outbound_result not readable yet + return await orig(frame, platform=platform) + + a._transport.send_outbound = gated_send + prompt_id = a._mint_prompt( + "exec_approval", {"session_key": "sess-dl", "chat_id": "D01"} + ) + # Pre-fix this hangs until the gate opens (deadlock shape) and the + # wait_for trips. Post-fix it returns promptly. + consumed = await asyncio.wait_for( + a._consume_prompt_response(FakeEvent(prompt_id, "once")), + timeout=1.0, + ) + assert consumed is True + # Release the gate; the background ack must still go out. + gate.set() + await asyncio.sleep(0.05) + ack_frames = [f for f, _ in a._transport.frames if f["op"] == "send"] + assert ack_frames, "background ack was never sent" + @pytest.mark.asyncio async def test_approval_ack_leaves_open_draft_untouched(self): """The exact live failure: exec-approval tap resolves while the @@ -104,6 +143,8 @@ class TestPromptAckDoesNotSealDraft: FakeEvent(prompt_id, "once") ) assert consumed is True + # The ack rides a background task now (deadlock fix): yield so it runs. + await asyncio.sleep(0.05) # The draft interception must still be armed for the real turn-final. assert draft_key in a._open_draft_by_chat, (