fix(relay): prompt-lifecycle acks are fire-and-forget — awaiting them ON the read loop self-deadlocked the transport

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

@@ -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.

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, (