fix(gateway): preserve detached approvals and bind Telegram cards

gateway/platforms/base.py

@@ -1582,7 +1582,8 @@ class ExecApprovalPrompt:
     ``BasePlatformAdapter.send_exec_approval``). ``actions`` rows are ``(label, choice, style)``
     with ``choice`` in ``once`` / ``session`` / ``always`` / ``deny`` — the vocabulary
     ``tools.approval.resolve_gateway_approval`` accepts — and ``style`` in ``primary`` /
-    ``danger`` / ``""``."""
+    ``danger`` / ``""``. ``request_id`` correlates a control with its exact pending
+    operation; native controls must not substitute session-level FIFO resolution."""
     chat_id: str
     session_key: str
     text: str
@@ -1591,6 +1592,7 @@ class ExecApprovalPrompt:
     description: str
     smart_denied: bool
     metadata: Optional[Dict[str, Any]] = None
+    request_id: Optional[str] = None
 
     @property
     def choices(self) -> List[str]:
@@ -1605,6 +1607,8 @@ class SendResult:
     error: Optional[str] = None
     # Adapter-specific metadata. Contract: Telegram edit-overflow partials set
     # raw_response["partial_overflow"] so the stream consumer sends the missing tail.
+    # raw_response["exec_approval_settlement"] = True means a native card already owns the
+    # core settlement hook (including in-flight sends); the runner must not overwrite it.
     raw_response: Any = None
     retryable: bool = False  # transient connection error — base retries automatically
     retry_after: Optional[float] = None  # server-requested delay (Telegram FloodWait) beats our backoff
@@ -2737,14 +2741,14 @@ def supports_exec_approval_buttons(cls) -> bool:
     async def send_exec_approval(
         self, chat_id: str, command: str, session_key: str, description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None, allow_permanent: bool = True, allow_session: bool = True,
-        smart_denied: bool = False,
+        smart_denied: bool = False, *, request_id: Optional[str] = None,
     ) -> SendResult:
         """Interactive exec-approval prompt; a press resolves via
         ``tools.approval.resolve_gateway_approval``. Text and choice set are shared; adapters
         render them natively in ``_send_exec_approval_prompt``."""
         prompt = ExecApprovalPrompt(
             chat_id=chat_id, session_key=session_key, metadata=metadata, command=str(command or ""),
-            description=description, smart_denied=smart_denied,
+            description=description, smart_denied=smart_denied, request_id=request_id,
             text=self._format_exec_approval(command, description, smart_denied),
             actions=self._exec_approval_actions(
                 allow_permanent=allow_permanent, allow_session=allow_session, smart_denied=smart_denied))

gateway/run_turn_runner.py

@@ -1452,20 +1452,26 @@ def _approval_notify_sync(self, approval_data: dict) -> None:
         # Slack's assistant_threads_setStatus disables the compose box, so the user can't type
         # /approve while "is thinking..." shows. Pausing stops _keep_typing re-setting it; resumed
         # in approve/deny.
-        adapter.pause_typing_for_chat(ctx._status_chat_id)
-        self._close_native_stream_boundary("Approval")
+        # A retained child route may outlive this parent turn. Do not pause a newer turn
+        # or re-open the old stream merely to deliver that child's independent prompt.
+        still_current = getattr(ctx, "_run_still_current", None)
+        if not callable(still_current) or still_current():
+            adapter.pause_typing_for_chat(ctx._status_chat_id)
+            self._close_native_stream_boundary("Approval")
         # Redact credentials before display: Tirith's findings are already redacted, but the raw
         # command string still leaks secrets. Both the button and plain-text paths use this value.
         cmd = _redact_approval_command(approval_data.get("command", ""))
         desc = approval_data.get("description", "dangerous command")
         flags = {k: approval_data.get(k, d) for k, d in (("allow_permanent", True), ("allow_session", True), ("smart_denied", False))}
         # Check the *class*, not the instance — MagicMock auto-creates attributes in tests.
         if _renders_exec_approval_buttons(type(adapter)):
+            binding = ({"request_id": approval_data.get("request_id")}
+                       if _accepts_keyword(adapter.send_exec_approval, "request_id") else {})
             try:
                 fut = self._schedule(
                     adapter.send_exec_approval(
                         chat_id=ctx._status_chat_id, command=cmd, session_key=ctx.session_key or "",
-                        description=desc, metadata=ctx._status_thread_metadata, **flags,
+                        description=desc, metadata=ctx._status_thread_metadata, **flags, **binding,
                     ),
                     "send_exec_approval scheduling error",
                 )
@@ -1475,9 +1481,12 @@ def _approval_notify_sync(self, approval_data: dict) -> None:
                 if outcome == "sent":
                     # Without this, a card whose timer runs out keeps live buttons and nobody
                     # learns the command did NOT run (only the TUI registered a settle hook).
-                    register_timeout_notice(
-                        self, approval_data, command=cmd,
-                        card_message_id=getattr(fut.result(timeout=0), "message_id", None))
+                    sent = fut.result(timeout=0)
+                    metadata = getattr(sent, "raw_response", None)
+                    if not (isinstance(metadata, dict) and metadata.get("exec_approval_settlement") is True):
+                        register_timeout_notice(
+                            self, approval_data, command=cmd,
+                            card_message_id=getattr(sent, "message_id", None))
                     return
                 if outcome == "ambiguous":
                     # Timeout ≠ failure: the card may have posted with a late ack. The prompt
@@ -1673,15 +1682,16 @@ def _native_image_run_message(self):
 
     def _run_conversation_with_approval(self, agent, agent_history, observed_group_context,
                                         persist_user_message_override, persist_user_timestamp_override):
-        """Run the turn with the per-session gateway approval callback registered: dangerous-command
-        approval blocks the agent thread (mirrors CLI input()); the callback bridges sync→async."""
+        """Own this turn's approval waits while detached workers retain their own delivery.
+        Approval blocks the agent thread (mirrors CLI input()); the callback bridges sync→async."""
         from gateway.run import _wrap_current_message_with_observed_context
-        from tools.approval import register_gateway_notify, unregister_gateway_notify
+        from tools.approval_ownership import gateway_approval_owner, register_gateway_approval_owner
         from tools.approval_context import reset_current_session_key, set_current_session_key
         ctx = self._ctx
         session_key = ctx.session_key or ""
         token = set_current_session_key(session_key)
-        register_gateway_notify(session_key, self._approval_notify_sync)
+        owner = register_gateway_approval_owner(session_key, self._approval_notify_sync)
+        owner_token = gateway_approval_owner.set(owner)
         try:
             api_message = _wrap_current_message_with_observed_context(self._native_image_run_message(), observed_group_context)
             kwargs = {"conversation_history": agent_history, "task_id": ctx.session_id}
@@ -1711,7 +1721,8 @@ def _run_conversation_with_approval(self, agent, agent_history, observed_group_c
             with notification_turn(agent, muted=ctx.mute_notification_reply, session_id=ctx.session_id or ""):
                 return agent.run_conversation(api_message, **kwargs)
         finally:
-            unregister_gateway_notify(session_key)
+            owner.close("the turn ended before the prompt was answered")
+            gateway_approval_owner.reset(owner_token)
             # Cancel pending clarify entries so blocked agent threads don't hang past the end of the
             # run (interrupt, completion, gateway shutdown). Idempotent.
             with suppress(Exception):

gateway/run_turn_runner_approval_settle.py

@@ -27,8 +27,8 @@ def register_timeout_notice(
     delivered BUTTON card's id when the adapter returned one, so the card itself is edited in place
     (which also drops its buttons). The plain-text prompt passes ``None``: it has no buttons to
     drop and rewriting it would erase the record of what was asked. ``command`` is the
-    already-redacted command shown to the user. The notice is skipped when the run is no longer
-    current (``ctx._run_still_current``).
+    already-redacted command shown to the user. The core request owns this notice, not the
+    parent turn: a detached child may still be waiting after that turn returns.
     """
     from tools.approval import register_gateway_settle
 
@@ -41,11 +41,8 @@ def register_timeout_notice(
     def settle(reason: str) -> None:
         if reason != "timeout":
             return  # answered / interrupted / notify_failed already produced their own feedback
-        # Same guard as every other late notice in TurnRunner: after /stop, /new or a restart the
-        # turn is over and this chat belongs to a newer run — do not edit or post into it.
-        still_current = getattr(runner._ctx, "_run_still_current", None)
-        if callable(still_current) and not still_current():
-            return
+        # The exact request timed out. Parent handoff does not invalidate its card or route;
+        # explicit owner/session cancellation instead settles with a non-timeout reason.
         runner._schedule(
             _post_timeout_notice(runner._ctx, command, card_message_id, timeout_s),
             "Approval timeout notice scheduling error")