diff --git a/agent/managed_worker.py b/agent/managed_worker.py
index 5939edb57624..3ddfc32d36e8 100644
--- a/agent/managed_worker.py
+++ b/agent/managed_worker.py
@@ -204,7 +204,8 @@ def read(self):
                     self.finish.set()
                     return
                 if (set(frame) != {'type', 'prompt_id', 'value'} or frame['type'] not in {'approval', 'clarify'}
-                        or not isinstance(frame['prompt_id'], str) or not isinstance(frame['value'], str)
+                        or not isinstance(frame['prompt_id'], str) or not frame['prompt_id']
+                        or not isinstance(frame['value'], str)
                         or len(frame['value']) > 16384):
                     raise ValueError('invalid_managed_control')
                 if frame['type'] == 'approval':
diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py
index 0b7711963d5d..7f933aef7e07 100644
--- a/gateway/relay/adapter.py
+++ b/gateway/relay/adapter.py
@@ -1940,7 +1940,7 @@ async def _send_exec_approval_prompt(self, prompt: ExecApprovalPrompt) -> SendRe
         metadata = dict(prompt.metadata or {})
         metadata.pop("_approval_request_id", None)
         result = await self._mint_and_send_prompt(
-            "exec_approval", {"session_key": prompt.session_key}, prompt.chat_id, prompt_kind="approval",
+            "exec_approval", {"session_key": prompt.session_key, "request_id": prompt.request_id}, prompt.chat_id, prompt_kind="approval",
             text=prompt.text, options=options, metadata=metadata,
         )
         return result if result is not None else self._PROMPT_UNAVAILABLE
@@ -2055,7 +2055,9 @@ async def _resolve_exec_approval(self, state, option_id, chat_id, ack_meta) -> N
         from tools.approval import resolve_gateway_approval
 
         choice = option_id if option_id in _EXEC_APPROVAL_LABELS else "deny"
-        count = resolve_gateway_approval(str(state.get("session_key") or ""), choice)
+        request_id = state.get("request_id")
+        count = resolve_gateway_approval(str(state.get("session_key") or ""), choice,
+                                         request_id=request_id) if request_id else 0
         label = _EXEC_APPROVAL_LABELS[choice] if count else "⌛ Approval expired — no command was waiting."
         # In-channel ack preserves the audit trail the native edit gives (the
         # connector's prompt message can't be edited cross-platform yet).
diff --git a/gateway/session_pending_controls.py b/gateway/session_pending_controls.py
index 4ff987908058..6dea006149c7 100644
--- a/gateway/session_pending_controls.py
+++ b/gateway/session_pending_controls.py
@@ -28,7 +28,7 @@ def register(self, session_id, route, generation, data, *, timeout_s=None):
                       'execution_generation': generation,
                       'command': _redact_approval_command(data.get('command', '')),
                       'description': _redact_approval_command(data.get('description', '')),
-                      'choices': choices}
+                      'choices': choices, 'smart_denied': bool(data.get('smart_denied'))}
             if 'edit' in data:
                 prompt['edit'] = deepcopy(data['edit'])
                 prompt['choices'] = ['once', 'deny']
diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py
index 6e2a2a620998..4e3e3d7d6a19 100644
--- a/gateway/slash_commands.py
+++ b/gateway/slash_commands.py
@@ -1194,7 +1194,7 @@ def _blocking_approval_or_stale(self, event: MessageEvent, stale_key: str, none_
     async def _handle_approve_command(self, event: MessageEvent) -> Optional[str]:
         """Handle /approve — unblock waiting agent thread(s). They block inside tools/approval.py;
         signalling the event resumes them so the command executes inline (same flow as the CLI)."""
-        from tools.approval import resolve_gateway_approval
+        from tools.approval import list_gateway_approvals, resolve_gateway_approval
         session_key, stale = self._blocking_approval_or_stale(event, "gateway.approval_expired",
                                                               "gateway.approve.no_pending")
         if stale:
@@ -1205,8 +1205,12 @@ async def _handle_approve_command(self, event: MessageEvent) -> Optional[str]:
         choice = "always" if "always" in choices else "session" if "session" in choices else "once"
         count = resolve_gateway_approval(session_key, choice, resolve_all="all" in args)
         if not count:
+            if any(entry.get("smart_denied") for entry in list_gateway_approvals(session_key)):
+                return t("gateway.approve.own_card")
             return t("gateway.approve.no_pending")
         confirmation_text = t(f"gateway.approve.{choice}_{'plural' if count > 1 else 'singular'}", count=count)
+        if "all" in args and any(entry.get("smart_denied") for entry in list_gateway_approvals(session_key)):
+            confirmation_text += "\n" + t("gateway.approve.own_card")
         logger.info("User approved %d dangerous command(s) via /approve (%s)", count, choice)
         return await self._deliver_approval_confirmation(event, confirmation_text, "approve")
 
diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py
index 684d29fc793c..9c69ce906cad 100644
--- a/hermes_cli/cli_modal_mixin.py
+++ b/hermes_cli/cli_modal_mixin.py
@@ -1102,6 +1102,7 @@ def _approval_callback(self, command: str, description: str,
                     allow_session=allow_session,
                     smart_denied=smart_denied),
                 "selected": 0,
+                "smart_denied": smart_denied,
                 "response_queue": response_queue}
             self._approval_deadline = _time.monotonic() + timeout
             self._ring_bell(prompt=True, context=t("cli.approval.bell_context"), detail=command)
diff --git a/hermes_cli/cli_tui_mixin.py b/hermes_cli/cli_tui_mixin.py
index a456e92ac6b1..0d5896a0f7a9 100644
--- a/hermes_cli/cli_tui_mixin.py
+++ b/hermes_cli/cli_tui_mixin.py
@@ -193,12 +193,23 @@ def _get_approval_display_fragments(self):
         and is truncated to the remaining row budget, so HSplit never clips approve/deny off-screen.
         """
         from cli import _panel_box_width, _wrap_panel_text_keep_ws
+        from tools.approval import REVIEWER_DENY_NOTE
         state = self._approval_state
         if not state:
             return []
         wrap = _wrap_panel_text_keep_ws
         command = state["command"]
         description = state["description"]
+        reviewer_warning = None
+        if state.get("smart_denied"):
+            description_lines = description.splitlines()
+            warning_index = next((
+                i for i, line in enumerate(description_lines)
+                if line.startswith(REVIEWER_DENY_NOTE)
+            ), None)
+            if warning_index is not None:
+                reviewer_warning = description_lines.pop(warning_index)
+                description = "\n".join(description_lines)
         choices = state["choices"]
         selected = state.get("selected", 0)
         show_full = state.get("show_full", False)
@@ -213,7 +224,7 @@ def _get_approval_display_fragments(self):
         box_width = _panel_box_width(title, preview_lines)
         inner_text_width = max(8, box_width - 2)
 
-        # Pre-wrap the mandatory content — command + choices must always render.
+        # Wrap content before allocating the available rows.
         cmd_wrapped = wrap(command, inner_text_width)
         if not show_full and "view" in choices and len(cmd_wrapped) > 4:
             cmd_wrapped = cmd_wrapped[:3] + wrap(
@@ -222,37 +233,46 @@ def _get_approval_display_fragments(self):
             f"{'❯' if i == selected else ' '} {_num_prefix(i)}. {_approval_choice_label(choice)}"
             for i, choice in enumerate(choices)]
         choice_wrapped = _wrap_rows(wrap, choice_labels, inner_text_width, "    ")
+        warning_wrapped = wrap(reviewer_warning, inner_text_width) if reviewer_warning else []
 
-        # Row budget so HSplit never clips the command or choices. Full chrome = top border +
-        # title + blank + blank-between-cmd/choices + bottom border (5); when that doesn't fit,
-        # drop the separator blanks (3) so every choice stays on-screen in compact terminals.
         available = max(0, _term_rows() - _PANEL_RESERVED_BELOW)
-        use_compact_chrome = 5 + len(cmd_wrapped) + len(choice_wrapped) > available
-        chrome_rows = 3 if use_compact_chrome else 5
-
-        # A command too long to leave room for the choices (e.g. "view" on a multi-hundred-char
-        # command) is truncated so approve/deny still render; keep at least 1 command row.
-        max_cmd_rows = max(1, available - chrome_rows - len(choice_wrapped))
+        # Borders take two rows; full chrome adds a title and two separators.
+        # Compact warning panels place the title in the border.
+        use_compact_chrome = 5 + len(cmd_wrapped) + len(choice_wrapped) + len(warning_wrapped) > available
+        inline_title = use_compact_chrome and bool(warning_wrapped)
+        omit_bottom_border = bool(warning_wrapped) and 2 + len(choice_wrapped) + 2 > available
+        chrome_rows = 2 - int(omit_bottom_border) + (0 if inline_title else 1) + (0 if use_compact_chrome else 2)
+        warning_budget = max(1, available - chrome_rows - len(choice_wrapped) - 1)
+        if len(warning_wrapped) > warning_budget:
+            warning_wrapped = warning_wrapped[:warning_budget]
+            if warning_wrapped:
+                warning_wrapped[-1] = warning_wrapped[-1][:max(0, inner_text_width - 2)] + " …"
+        max_cmd_rows = max(1, available - chrome_rows - len(choice_wrapped) - len(warning_wrapped))
         if len(cmd_wrapped) > max_cmd_rows:
-            keep = max(1, max_cmd_rows - 1) if max_cmd_rows > 1 else 1
-            cmd_wrapped = cmd_wrapped[:keep] + wrap(
-                t("cli.tui.approval_command_truncated"), inner_text_width)
+            if max_cmd_rows == 1:
+                cmd_wrapped = [cmd_wrapped[0]]
+            else:
+                cmd_wrapped = cmd_wrapped[:max_cmd_rows - 1] + [
+                    t("cli.tui.approval_command_truncated")]
 
-        # Remaining rows go to the description (minus the blank separator in full mode), capped
-        # at 10 so the panel stays compact even on huge terminals.
-        mandatory_no_desc = chrome_rows + len(cmd_wrapped) + len(choice_wrapped)
+        # Remaining rows go to the description (minus its blank separator), capped at 10.
+        mandatory_no_desc = chrome_rows + len(warning_wrapped) + len(cmd_wrapped) + len(choice_wrapped)
         available_for_desc = available - mandatory_no_desc - (0 if use_compact_chrome else 1)
         available_for_desc = max(0, min(available_for_desc, 10))
         desc_wrapped = wrap(description, inner_text_width) if description else []
         if available_for_desc < 1 or not desc_wrapped:
             desc_wrapped = []
         elif len(desc_wrapped) > available_for_desc:
-            desc_wrapped = desc_wrapped[:max(1, available_for_desc - 1)] + [t("cli.tui.approval_description_truncated")]
-
-        # Render title → command → choices → description; description last so any overflow
-        # clips the least-critical content, never the command or choices.
-        panel = _Panel('class:approval-border', box_width)
-        panel.row('class:approval-title', title)
+            desc_wrapped = desc_wrapped[:available_for_desc - 1] + [
+                t("cli.tui.approval_description_truncated")]
+        # Render the warning before the command; compact chrome integrates the title into its border.
+        panel = _Panel('class:approval-border', box_width,
+                       title if inline_title else "",
+                       'class:approval-title')
+        if not inline_title:
+            panel.row('class:approval-title', title)
+        for wrapped in warning_wrapped:
+            panel.row('class:approval-desc', wrapped)
         if not use_compact_chrome:
             panel.blank()
         for wrapped in cmd_wrapped:
@@ -267,7 +287,7 @@ def _get_approval_display_fragments(self):
                 panel.blank()
             for wrapped in desc_wrapped:
                 panel.row('class:approval-desc', wrapped)
-        return panel.close()
+        return panel.lines if omit_bottom_border else panel.close()
 
     def _get_tui_prompt_symbols(self) -> tuple[str, str]:
         """Return ``(normal_prompt, state_suffix)`` for the active skin.
diff --git a/hermes_cli/gateway_agent.py b/hermes_cli/gateway_agent.py
index 39dc0190e333..55c3168e8825 100644
--- a/hermes_cli/gateway_agent.py
+++ b/hermes_cli/gateway_agent.py
@@ -988,16 +988,17 @@ def _emit_print(self, text):
         (self._print_fn or print)(text)
 
     def _on_approval(self, payload, stream_callback, starts):
+        from tools.approval_prompt import callback_accepts
         from tools.terminal_tool import _get_approval_callback
         callback = _get_approval_callback()
         choices = list(payload.get("choices") or [])
         choice = "deny"
         if callback is not None:
-            try:
+            if callback_accepts(callback, "smart_denied"):
                 choice = callback(payload.get("command") or "", payload.get("description") or "",
                                   allow_permanent="always" in choices, allow_session="session" in choices,
-                                  smart_denied=False)
-            except TypeError:
+                                  smart_denied=bool(payload.get("smart_denied")))
+            else:
                 choice = callback(payload.get("command") or "", payload.get("description") or "")
         if choice not in choices:
             choice = "deny"
diff --git a/locales/af.yaml b/locales/af.yaml
index c83eee703f8c..29b8dcdd08d8 100644
--- a/locales/af.yaml
+++ b/locales/af.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - kind {index}: {calls} API-oproepe · {doing}"
     active_ago: " · {seconds}s gelede aktief"
   approve:
+    own_card: "Versoeke wat die beoordelaar afgekeur het, moet op hul eie kaart beantwoord word."
     no_pending: "Geen hangende opdrag om goed te keur nie."
     once_singular: "✅ Opdrag goedgekeur. Die agent gaan voort..."
     once_plural: "✅ Opdragte goedgekeur ({count} opdragte). Die agent gaan voort..."
diff --git a/locales/ar.yaml b/locales/ar.yaml
index 0ba1fc9f1985..05dbd5300cec 100644
--- a/locales/ar.yaml
+++ b/locales/ar.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - الوكيل الفرعي {index}: {calls} استدعاء API · {doing}"
     active_ago: " · كان نشطًا قبل {seconds} ث"
   approve:
+    own_card: "يجب الرد على الطلبات التي رفضها المراجع في بطاقتها الخاصة."
     no_pending: "لا يوجد أمر معلّق للموافقة عليه."
     once_singular: "✅ تمت الموافقة على الأمر. الوكيل يستأنف..."
     once_plural: "✅ تمت الموافقة على الأوامر ({count} أوامر). الوكيل يستأنف..."
diff --git a/locales/de.yaml b/locales/de.yaml
index 96a2f9bf1057..a2b1cf5ccc69 100644
--- a/locales/de.yaml
+++ b/locales/de.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - Kind {index}: {calls} API-Aufrufe · {doing}"
     active_ago: " · vor {seconds}s aktiv"
   approve:
+    own_card: "Vom Prüfer abgelehnte Anfragen müssen auf ihrer eigenen Karte beantwortet werden."
     no_pending: "Kein ausstehender Befehl zum Genehmigen."
     once_singular: "✅ Befehl genehmigt. Der Agent wird fortgesetzt..."
     once_plural: "✅ Befehle genehmigt ({count} Befehle). Der Agent wird fortgesetzt..."
diff --git a/locales/en.yaml b/locales/en.yaml
index 1a773037e498..da49806b9c47 100644
--- a/locales/en.yaml
+++ b/locales/en.yaml
@@ -95,6 +95,7 @@ gateway:
     child_row: "  - child {index}: {calls} api calls · {doing}"
     active_ago: " · active {seconds}s ago"
   approve:
+    own_card: "Reviewer DENY requests must be answered on their own card."
     no_pending: "No pending command to approve."
     once_singular: "✅ Command approved. The agent is resuming..."
     once_plural: "✅ Commands approved ({count} commands). The agent is resuming..."
diff --git a/locales/es.yaml b/locales/es.yaml
index 25835352ec18..dd7f1d794ba9 100644
--- a/locales/es.yaml
+++ b/locales/es.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - hijo {index}: {calls} llamadas a la API · {doing}"
     active_ago: " · activo hace {seconds}s"
   approve:
+    own_card: "Las solicitudes rechazadas por el revisor deben responderse en su propia tarjeta."
     no_pending: "No hay ningún comando pendiente que aprobar."
     once_singular: "✅ Comando aprobado. El agente se está reanudando..."
     once_plural: "✅ Comandos aprobados ({count} comandos). El agente se está reanudando..."
diff --git a/locales/fr.yaml b/locales/fr.yaml
index 13b40dd94556..336224cbe586 100644
--- a/locales/fr.yaml
+++ b/locales/fr.yaml
@@ -97,6 +97,7 @@ gateway:
     child_row: "  - enfant {index} : {calls} appels API · {doing}"
     active_ago: " · actif il y a {seconds}s"
   approve:
+    own_card: "Les demandes refusées par le réviseur doivent être traitées sur leur propre carte."
     no_pending: "Aucune commande en attente d'approbation."
     once_singular: "✅ Commande approuvée. L'agent reprend..."
     once_plural: "✅ Commandes approuvées ({count} commandes). L'agent reprend..."
diff --git a/locales/ga.yaml b/locales/ga.yaml
index 85f5164f5c2f..eedbaa5c3d0e 100644
--- a/locales/ga.yaml
+++ b/locales/ga.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - fo-ghníomhaire {index}: {calls} glao API · {doing}"
     active_ago: " · gníomhach {seconds}s ó shin"
   approve:
+    own_card: "Ní mór iarratais a dhiúltaigh an t-athbhreithneoir a fhreagairt ar a gcárta féin."
     no_pending: "Níl aon ordú ag fanacht le ceadú."
     once_singular: "✅ Ordú ceadaithe. Tá an gníomhaire ag atosú..."
     once_plural: "✅ Orduithe ceadaithe ({count} ordú). Tá an gníomhaire ag atosú..."
diff --git a/locales/hu.yaml b/locales/hu.yaml
index d73ed002d515..35a707b88363 100644
--- a/locales/hu.yaml
+++ b/locales/hu.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - gyerek {index}: {calls} API-hívás · {doing}"
     active_ago: " · {seconds} mp-e volt aktív"
   approve:
+    own_card: "Az ellenőrző által elutasított kéréseket a saját kártyájukon kell megválaszolni."
     no_pending: "Nincs jóváhagyásra váró parancs."
     once_singular: "✅ Parancs jóváhagyva. Az ügynök folytatja..."
     once_plural: "✅ Parancsok jóváhagyva ({count} parancs). Az ügynök folytatja..."
diff --git a/locales/it.yaml b/locales/it.yaml
index 23ed9d57f78d..c2e2568306e8 100644
--- a/locales/it.yaml
+++ b/locales/it.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - figlio {index}: {calls} chiamate API · {doing}"
     active_ago: " · attivo {seconds}s fa"
   approve:
+    own_card: "Le richieste rifiutate dal revisore devono essere gestite sulla propria scheda."
     no_pending: "Nessun comando in attesa di approvazione."
     once_singular: "✅ Comando approvato. L'agente sta riprendendo..."
     once_plural: "✅ Comandi approvati ({count} comandi). L'agente sta riprendendo..."
diff --git a/locales/ja.yaml b/locales/ja.yaml
index 992246f9d74f..78105e53d666 100644
--- a/locales/ja.yaml
+++ b/locales/ja.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - 子 {index}: API 呼び出し {calls} 回 · {doing}"
     active_ago: " · {seconds} 秒前に活動"
   approve:
+    own_card: "レビュー担当が拒否したリクエストには、それぞれのカードで回答してください。"
     no_pending: "承認待ちのコマンドはありません。"
     once_singular: "✅ コマンドを承認しました。エージェントを再開しています..."
     once_plural: "✅ コマンドを承認しました ({count} 件)。エージェントを再開しています..."
diff --git a/locales/ko.yaml b/locales/ko.yaml
index a76da7f4ff28..ea27c36b6388 100644
--- a/locales/ko.yaml
+++ b/locales/ko.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - 자식 {index}: API 호출 {calls}회 · {doing}"
     active_ago: " · {seconds}초 전 활동"
   approve:
+    own_card: "검토자가 거부한 요청은 해당 카드에서 응답해야 합니다."
     no_pending: "승인 대기 중인 명령이 없습니다."
     once_singular: "✅ 명령이 승인되었습니다. 에이전트가 재개됩니다..."
     once_plural: "✅ 명령이 승인되었습니다 ({count}개). 에이전트가 재개됩니다..."
diff --git a/locales/pt.yaml b/locales/pt.yaml
index 0656f7b0b418..e68b311b4c1e 100644
--- a/locales/pt.yaml
+++ b/locales/pt.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - filho {index}: {calls} chamadas de API · {doing}"
     active_ago: " · ativo há {seconds}s"
   approve:
+    own_card: "Os pedidos rejeitados pelo revisor devem ser respondidos no respetivo cartão."
     no_pending: "Não há nenhum comando pendente para aprovar."
     once_singular: "✅ Comando aprovado. O agente está a retomar..."
     once_plural: "✅ Comandos aprovados ({count} comandos). O agente está a retomar..."
diff --git a/locales/ru.yaml b/locales/ru.yaml
index 8002a2140f5a..3b0dc67b11fa 100644
--- a/locales/ru.yaml
+++ b/locales/ru.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - потомок {index}: {calls} вызовов API · {doing}"
     active_ago: " · активен {seconds}с назад"
   approve:
+    own_card: "На запросы, отклонённые проверяющим, нужно отвечать в их собственной карточке."
     no_pending: "Нет команды, ожидающей одобрения."
     once_singular: "✅ Команда одобрена. Агент возобновляет работу..."
     once_plural: "✅ Команды одобрены ({count} команд). Агент возобновляет работу..."
diff --git a/locales/tr.yaml b/locales/tr.yaml
index 6a6473c74298..54826ade6eea 100644
--- a/locales/tr.yaml
+++ b/locales/tr.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - alt ajan {index}: {calls} API çağrısı · {doing}"
     active_ago: " · {seconds} sn önce aktifti"
   approve:
+    own_card: "İnceleyicinin reddettiği istekler kendi kartlarında yanıtlanmalıdır."
     no_pending: "Onaylanacak bekleyen komut yok."
     once_singular: "✅ Komut onaylandı. Ajan devam ediyor..."
     once_plural: "✅ Komutlar onaylandı ({count} komut). Ajan devam ediyor..."
diff --git a/locales/uk.yaml b/locales/uk.yaml
index da3207f340b0..34839f782b7e 100644
--- a/locales/uk.yaml
+++ b/locales/uk.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - дочірній {index}: {calls} api-викликів · {doing}"
     active_ago: " · активний {seconds}с тому"
   approve:
+    own_card: "На запити, відхилені перевіряльником, потрібно відповідати на їхній власній картці."
     no_pending: "Немає команди на схвалення."
     once_singular: "✅ Команду схвалено. Агент відновлює роботу…"
     once_plural: "✅ Команди схвалено ({count} команд). Агент відновлює роботу…"
diff --git a/locales/zh-hant.yaml b/locales/zh-hant.yaml
index d24097161399..66ba8cc42db9 100644
--- a/locales/zh-hant.yaml
+++ b/locales/zh-hant.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - 子代理 {index}：{calls} 次 API 呼叫 · {doing}"
     active_ago: " · {seconds} 秒前活動"
   approve:
+    own_card: "審查者拒絕的請求必須在各自的卡片上回覆。"
     no_pending: "沒有待批准的指令。"
     once_singular: "✅ 指令已批准。代理正在恢復…"
     once_plural: "✅ 指令已批准（{count} 條指令）。代理正在恢復…"
diff --git a/locales/zh.yaml b/locales/zh.yaml
index ecb2348e2cfc..02101549d190 100644
--- a/locales/zh.yaml
+++ b/locales/zh.yaml
@@ -92,6 +92,7 @@ gateway:
     child_row: "  - 子代理 {index}：{calls} 次 API 调用 · {doing}"
     active_ago: " · {seconds} 秒前活跃"
   approve:
+    own_card: "审核者拒绝的请求必须在各自的卡片上回复。"
     no_pending: "没有待批准的命令。"
     once_singular: "✅ 命令已批准。代理正在恢复…"
     once_plural: "✅ 命令已批准（{count} 条命令）。代理正在恢复…"
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 7df8153d75c0..1176b2b21dc7 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -701,7 +701,7 @@ def __init__(self, config: PlatformConfig):
         self._max_doc_bytes: int = 2 * 1024 * 1024 * 1024 if extra.get("base_url") else 20 * 1024 * 1024
         self._model_picker_state: dict[str, dict] = {}  # per-chat interactive picker state
         self._choice_picker_state: dict[str, dict] = {}
-        self._approval_state: dict[int, str] = {}  # message_id → session_key
+        self._approval_state: dict[int, tuple[str, str]] = {}  # message_id → (session_key, request_id)
         self._slash_confirm_state: dict[str, str] = {}  # confirm_id → session_key
         self._clarify_state: dict[str, str] = {}  # clarify_id → session_key
         # "important" (default): only final responses, approvals and slash confirmations notify;
@@ -4240,7 +4240,8 @@ def build():
             buttons = [InlineKeyboardButton(label, callback_data=f"ea:{choice}:{approval_id}")
                        for label, choice, _ in prompt.actions]
             return prompt.text, InlineKeyboardMarkup(self._rows_of_two(buttons)), (
-                lambda msg: self._approval_state.__setitem__(approval_id, prompt.session_key))
+                lambda msg: self._approval_state.__setitem__(
+                    approval_id, (prompt.session_key, prompt.request_id)))
         return await self._send_prompt(
             "send_exec_approval", prompt.chat_id, prompt.metadata, build, parse_mode=ParseMode.HTML,
             thread_id=self._metadata_thread_id(prompt.metadata), reply_to_mode=self._reply_to_mode)
@@ -4726,11 +4727,12 @@ async def _handle_exec_approval_callback(self, query, data: str, cb: dict[str, A
         except (ValueError, IndexError):
             await query.answer(text=_toast("platform.telegram.approval.toast_invalid_data"))
             return
-        session_key = await self._claim_callback_state(
+        state = await self._claim_callback_state(
             query, cb, self._approval_state, approval_id, _unauthorized(),
             _toast("platform.telegram.approval.toast_already_resolved"))
-        if not session_key:
+        if not state:
             return
+        session_key, request_id = state
         user_display = getattr(query.from_user, "first_name", None) or t("platform.telegram.user_fallback")
         # Resolve FIRST (unblocks the agent thread), render after: a tap landing after the wait timed out
         # (count == 0) must NOT claim "Approved" — the command was already denied.
@@ -4739,7 +4741,7 @@ async def _handle_exec_approval_callback(self, query, data: str, cb: dict[str, A
             # the approval wait timed out (count == 0) must NOT claim "Approved" — the command was already
             # denied and will not run (#63501 regression follow-up: 60s waits made stale taps common).
             from tools.approval import resolve_gateway_approval
-            count = resolve_gateway_approval(session_key, choice)
+            count = resolve_gateway_approval(session_key, choice, request_id=request_id) if request_id else 0
             logger.info(
                 "Telegram button resolved %d approval(s) for session %s (choice=%s, user=%s)", count, session_key, choice, user_display)
         except Exception as exc:
diff --git a/tests/gateway/fixtures/plugin_origin_approval_peer.py b/tests/gateway/fixtures/plugin_origin_approval_peer.py
index fca9d138eafb..42cc062588fe 100644
--- a/tests/gateway/fixtures/plugin_origin_approval_peer.py
+++ b/tests/gateway/fixtures/plugin_origin_approval_peer.py
@@ -13,8 +13,9 @@
 PLUGIN_KEYS = ("allow_gateway_injection", "allow_origin_sessions")
 CASES = {
     "review-approve": {"hook": True, "review": True, "answer": "once"},
+    "review-mixed-queue": {"hook": True, "review": True, "answer": "once"},
     "review-escalate": {"hook": True, "review": True, "verdict": "escalate", "answer": "once"},
-    "review-deny": {"hook": True, "review": True, "verdict": "deny", "answer": None},
+    "review-deny": {"hook": True, "review": True, "verdict": "deny", "answer": "once"},
     "human-deny": {"hook": True, "review": True, "answer": "deny"},
     "timeout": {"hook": True, "review": True, "answer": "timeout"},
     "stale-always": {"hook": True, "review": True, "answer": "always"},
@@ -203,9 +204,19 @@ async def answer_cards(*, answer, expected, before, settled):
                 assert set(prompt.choices) == {"once", "deny"}, prompt.choices
                 assert "wants to run" in prompt.text.lower(), prompt.text
                 assert adapter._approval_mention_content() == "<@12345>"
+                if case_name == "review-mixed-queue":
+                    from tools.approval_gateway_wait import _ApprovalEntry
+                    denied = _ApprovalEntry({"request_id": "earlier-denial", "smart_denied": True})
+                    with approval._lock:
+                        approval._gateway_queues[session_key].insert(0, denied)
+                if case_name == "review-deny" and seen == before + 1:
+                    assert "flagged as dangerous" in prompt.text.lower(), prompt.text
                 if answer not in {None, "timeout"}:
                     assert answer in {"once", "deny", "always"}
-                    assert resolve_gateway_approval(session_key, answer) == 1
+                    assert resolve_gateway_approval(
+                        session_key, answer, request_id=prompt.request_id) == 1
+                    if case_name == "review-mixed-queue":
+                        assert denied.result is None and not denied.event.is_set()
                 continue
             try:
                 await asyncio.wait_for(adapter.card_ready.wait(), timeout=0.2)
diff --git a/tests/gateway/relay/test_relay_interactive.py b/tests/gateway/relay/test_relay_interactive.py
index 9749877558ca..43053ae49a44 100644
--- a/tests/gateway/relay/test_relay_interactive.py
+++ b/tests/gateway/relay/test_relay_interactive.py
@@ -111,10 +111,28 @@ async def test_exec_approval_renders_full_option_set():
 async def test_exec_approval_does_not_send_internal_request_id_in_metadata():
     adapter, stub = _adapter()
     await adapter.send_exec_approval(
-        "c1", "rm -rf /tmp/x", "sess:1", metadata={"_approval_request_id": "internal"},
+        "c1", "rm -rf /tmp/x", "sess:1", metadata={"_approval_request_id": "internal"}, request_id="internal",
     )
 
     assert "_approval_request_id" not in stub.sent[-1].get("metadata", {})
+    state = adapter._pending_prompts[stub.sent[-1]["prompt_id"]]
+    assert state["request_id"] == "internal"
+
+
+@pytest.mark.asyncio
+async def test_relay_card_targets_ordinary_request_ahead_of_denial(monkeypatch):
+    from tools import approval
+    from tools.approval_gateway_wait import _ApprovalEntry
+
+    adapter, stub = _adapter()
+    denied = _ApprovalEntry({"request_id": "deny", "smart_denied": True})
+    ordinary = _ApprovalEntry({"request_id": "ordinary"})
+    monkeypatch.setitem(approval._gateway_queues, "s", [denied, ordinary])
+    await adapter.send_exec_approval("c1", "cmd", "s", request_id="ordinary")
+    state = adapter._pending_prompts[stub.sent[-1]["prompt_id"]]
+    await adapter._resolve_exec_approval(state, "once", "c1", {})
+    assert ordinary.result == "once"
+    assert denied.result is None
 
 
 @pytest.mark.asyncio
diff --git a/tests/gateway/test_approve_deny_commands.py b/tests/gateway/test_approve_deny_commands.py
index 178e66d21050..69354c1493f4 100644
--- a/tests/gateway/test_approve_deny_commands.py
+++ b/tests/gateway/test_approve_deny_commands.py
@@ -162,6 +162,34 @@ class TestApproveCommand:
     def setup_method(self):
         _clear_approval_state()
 
+    @pytest.mark.asyncio
+    async def test_approve_stops_at_denial_and_reports_card(self):
+        from tools.approval import _gateway_queues
+        from tools.approval_gateway_wait import _ApprovalEntry
+
+        runner = _make_runner()
+        session_key = runner._session_key_for_source(_make_source())
+        denied = _ApprovalEntry({"request_id": "denied", "smart_denied": True})
+        ordinary = _ApprovalEntry({"request_id": "ordinary"})
+        _gateway_queues[session_key] = [denied, ordinary]
+        response = await runner._handle_approve_command(_make_event("/approve"))
+        assert "own card" in response
+        assert ordinary.result is None and denied.result is None
+
+    @pytest.mark.asyncio
+    async def test_approve_all_reports_remaining_denial(self):
+        from tools.approval import _gateway_queues
+        from tools.approval_gateway_wait import _ApprovalEntry
+
+        runner = _make_runner()
+        session_key = runner._session_key_for_source(_make_source())
+        denied = _ApprovalEntry({"request_id": "denied", "smart_denied": True})
+        ordinary = _ApprovalEntry({"request_id": "ordinary"})
+        _gateway_queues[session_key] = [denied, ordinary]
+        response = await runner._handle_approve_command(_make_event("/approve all"))
+        assert ordinary.result == "once" and denied.result is None
+        assert "own card" in response
+
 
     @pytest.mark.asyncio
     async def test_approve_all_resolves_multiple(self):
diff --git a/tests/gateway/test_managed_worker_clarify_timeout.py b/tests/gateway/test_managed_worker_clarify_timeout.py
index 90e8fd6f721c..14e452eeb8a0 100644
--- a/tests/gateway/test_managed_worker_clarify_timeout.py
+++ b/tests/gateway/test_managed_worker_clarify_timeout.py
@@ -52,3 +52,22 @@ def wait(self, timeout=None):
     monkeypatch.setattr(worker.threading, 'Event', _Event)
     assert _controls()._ask('q?', [], False) is None
     assert waits == [None]
+
+
+@pytest.mark.parametrize('kind', ['approval', 'clarify'])
+def test_control_reader_rejects_empty_prompt_id(monkeypatch, kind):
+    from agent import managed_worker as worker
+    from tools import approval
+
+    controls = _controls()
+    controls.finish = threading.Event()
+    frames = iter([{'type': kind, 'prompt_id': '', 'value': 'once'}, {'type': 'finish'}])
+    resolved = []
+    monkeypatch.setattr(worker, 'read_frame', lambda _: next(frames))
+    monkeypatch.setattr(approval, 'resolve_gateway_approval', lambda *args, **kwargs: resolved.append(args))
+
+    controls.read()
+
+    assert controls.stopped.is_set()
+    assert controls.finish.is_set()
+    assert resolved == []
diff --git a/tests/gateway/test_plugin_origin_approval.py b/tests/gateway/test_plugin_origin_approval.py
index 4a3570cefc6d..0fe0d9bd6d2b 100644
--- a/tests/gateway/test_plugin_origin_approval.py
+++ b/tests/gateway/test_plugin_origin_approval.py
@@ -11,7 +11,7 @@
 
 @pytest.mark.platforms("linux")
 @pytest.mark.parametrize("case", [
-    "review-approve", "review-escalate", "review-deny", "human-deny", "timeout", "stale-always",
+    "review-approve", "review-escalate", "review-deny", "review-mixed-queue", "human-deny", "timeout", "stale-always",
     "terminal", "execute-code", "yolo", "session-yolo", "mode-off", "grants", "allowlist",
     "smart-approve", "smart-deny", "cotal-hook-deny", "cotal-command-deny", "cotal-code-deny",
     "operator-bypasses",
diff --git a/tests/gateway/test_telegram_approval_buttons.py b/tests/gateway/test_telegram_approval_buttons.py
index 197d71c7f11b..0072e8044268 100644
--- a/tests/gateway/test_telegram_approval_buttons.py
+++ b/tests/gateway/test_telegram_approval_buttons.py
@@ -202,6 +202,26 @@ async def mock_send_message(**kwargs):
 class TestTelegramApprovalCallback:
     """Test the approval callback handling in _handle_callback_query."""
 
+    @pytest.mark.asyncio
+    async def test_tap_targets_card_and_stale_tap_cannot_release_denial(self, monkeypatch):
+        from tools import approval
+        from tools.approval_gateway_wait import _ApprovalEntry
+
+        adapter = _make_adapter()
+        denied = _ApprovalEntry({"request_id": "a", "smart_denied": True})
+        ordinary = _ApprovalEntry({"request_id": "b"})
+        monkeypatch.setitem(approval._gateway_queues, "session", [denied, ordinary])
+        adapter._approval_state[5] = ("session", "b")
+        adapter._approval_state[6] = ("session", "old")
+        query = AsyncMock()
+        query.from_user.first_name = "owner"
+        adapter._claim_callback_state = AsyncMock(side_effect=[("session", "b"), ("session", "old")])
+        adapter._edit_md_quiet = AsyncMock()
+        await adapter._handle_exec_approval_callback(query, "ea:once:5", {"chat_id": "1"})
+        await adapter._handle_exec_approval_callback(query, "ea:once:6", {"chat_id": "1"})
+        assert ordinary.result == "once"
+        assert denied.result is None
+
 
     @pytest.mark.asyncio
     async def test_resume_typing_after_inline_approval(self):
@@ -212,7 +232,7 @@ async def test_resume_typing_after_inline_approval(self):
         rest of a long-running turn after a button click.
         """
         adapter = _make_adapter()
-        adapter._approval_state[5] = "agent:main:telegram:group:12345:99"
+        adapter._approval_state[5] = ("agent:main:telegram:group:12345:99", "request-5")
         adapter.pause_typing_for_chat("12345")
         assert "12345" in adapter._typing_paused
 
@@ -240,7 +260,7 @@ async def test_resume_typing_after_inline_approval(self):
     @pytest.mark.asyncio
     async def test_approval_callback_escapes_dynamic_user_name(self):
         adapter = _make_adapter()
-        adapter._approval_state[3] = "agent:main:telegram:group:12345:99"
+        adapter._approval_state[3] = ("agent:main:telegram:group:12345:99", "request-3")
 
         query = AsyncMock()
         query.data = "ea:once:3"
diff --git a/tests/hermes_cli/test_cli_approval_ui.py b/tests/hermes_cli/test_cli_approval_ui.py
index feec4091d651..cc70156b60bd 100644
--- a/tests/hermes_cli/test_cli_approval_ui.py
+++ b/tests/hermes_cli/test_cli_approval_ui.py
@@ -4,6 +4,8 @@
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 import cli as cli_module
 from agent.i18n import t
 from cli import HermesCLI
@@ -176,10 +178,6 @@ def test_handle_approval_selection_view_expands_in_place(self):
         assert cli._approval_state["response_queue"].empty()
 
 
-
-
-
-
     def test_approval_display_truncates_giant_command_in_view_mode(self):
         """If the user hits /view on a massive command, choices still render.
 
@@ -214,6 +212,34 @@ def test_approval_display_truncates_giant_command_in_view_mode(self):
         # Command got truncated with a marker.
         assert t("cli.tui.approval_command_truncated") in rendered
 
+
+    @pytest.mark.parametrize("height", [11, 12, 14])
+    def test_reviewer_deny_warning_is_visible_with_long_reason_and_short_terminal(self, height):
+        cli = _make_cli_stub()
+        cli._approval_state = {
+            "command": "guarded action",
+            "description": "plugin reason " + ("x" * 1000)
+            + "\n**Reviewer:** flagged as dangerous: reviewer reason " + ("y" * 1000),
+            "choices": ["once", "deny"],
+            "selected": 0,
+            "smart_denied": True,
+            "response_queue": queue.Queue(),
+        }
+
+        import shutil as _shutil
+
+        with patch("cli.shutil.get_terminal_size",
+                   return_value=_shutil.os.terminal_size((60, height))):
+            fragments = cli._get_approval_display_fragments()
+
+        rendered = "".join(text for _style, text in fragments)
+        assert "flagged as dangerous" in rendered
+        assert "guarded action" in rendered
+        assert rendered.index("flagged as dangerous") < rendered.index("guarded action")
+        assert "…" in rendered
+        from hermes_cli.cli_tui_mixin import _PANEL_RESERVED_BELOW
+        assert rendered.count("\n") <= height - _PANEL_RESERVED_BELOW
+
     def test_background_task_registers_thread_local_approval_callbacks(self):
         """Background /btw tasks must use the prompt_toolkit approval UI.
 
diff --git a/tests/hermes_cli/test_reviewer_gateway_propagation.py b/tests/hermes_cli/test_reviewer_gateway_propagation.py
new file mode 100644
index 000000000000..6b40b13b7ad2
--- /dev/null
+++ b/tests/hermes_cli/test_reviewer_gateway_propagation.py
@@ -0,0 +1,35 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+import threading
+
+import pytest
+from gateway.session_pending_controls import PendingControls
+from hermes_cli.gateway_agent import GatewayAgent
+from tools import terminal_tool
+
+
+def test_reviewer_denial_reaches_cli_callback(monkeypatch):
+    events = SimpleNamespace(lock=threading.RLock(), publish=Mock())
+    controls = PendingControls(events)
+    controls.register("session", "route", 1, {"request_id": "r", "smart_denied": True})
+    payload = events.publish.call_args.args[1]
+    callback = Mock(return_value="deny")
+    monkeypatch.setattr(terminal_tool, "_get_approval_callback", lambda: callback)
+    agent = SimpleNamespace(connection=SimpleNamespace(call=Mock()), session_id="session", _generation=1)
+    GatewayAgent._on_approval(agent, payload, None, None)
+    assert callback.call_args.kwargs["smart_denied"] is True
+
+
+def test_callback_type_error_does_not_retry_without_deny_warning(monkeypatch):
+    calls = []
+
+    def callback(command, description, **kwargs):
+        calls.append(kwargs)
+        raise TypeError("callback bug")
+
+    monkeypatch.setattr(terminal_tool, "_get_approval_callback", lambda: callback)
+    agent = SimpleNamespace(connection=SimpleNamespace(call=Mock()), session_id="session", _generation=1)
+    with pytest.raises(TypeError, match="callback bug"):
+        GatewayAgent._on_approval(agent, {"choices": ["once", "deny"], "smart_denied": True}, None, None)
+    assert len(calls) == 1
+    assert calls[0]["smart_denied"] is True
diff --git a/tests/tools/test_approval_gateway_wait_coalesce_race.py b/tests/tools/test_approval_gateway_wait_coalesce_race.py
index a49a76ec5efe..526565a89cee 100644
--- a/tests/tools/test_approval_gateway_wait_coalesce_race.py
+++ b/tests/tools/test_approval_gateway_wait_coalesce_race.py
@@ -203,3 +203,128 @@ def run():
             thread.join(0.05)
     assert (prompts, queued) == (2, 1)
     assert [r["choice"] for r in results] == ["once"] * 3
+
+
+def test_smart_denied_follower_gets_its_own_prompt(monkeypatch):
+    _clear()
+    monkeypatch.setattr(wait_mod._ctx, "_fire_approval_hook", lambda name, **kwargs: None)
+    monkeypatch.setattr(wait_mod._ctx, "_get_approval_timeout", lambda: APPROVAL_TIMEOUT)
+    leader_notified = threading.Event()
+    follower_notified = threading.Event()
+    results = {}
+
+    def notify_leader(data):
+        results["leader_id"] = data["request_id"]
+        leader_notified.set()
+
+    def notify_follower(data):
+        results["follower_id"] = data["request_id"]
+        follower_notified.set()
+
+    def run(name, notify, data):
+        results[name] = wait_mod._await_gateway_decision(SESSION_KEY, notify, data)
+
+    leader = threading.Thread(target=run, args=("leader", notify_leader, dict(APPROVAL)), daemon=True)
+    denied_data = {**APPROVAL, "description": "reviewer flagged the request", "smart_denied": True}
+    follower = threading.Thread(target=run, args=("follower", notify_follower, denied_data), daemon=True)
+    leader.start()
+    assert leader_notified.wait(10), "ordinary leader prompt was not sent"
+    follower.start()
+    try:
+        assert follower_notified.wait(10), "DENY follower did not send its own prompt"
+        with mod._lock:
+            pending = list(mod._gateway_queues[SESSION_KEY])
+        assert len(pending) == 2
+        assert pending[1].data["smart_denied"] is True
+
+        assert mod.resolve_gateway_approval(SESSION_KEY, "session", request_id=results["leader_id"]) == 1
+        leader.join(10)
+        assert not leader.is_alive()
+        assert results["leader"]["choice"] == "session"
+        assert follower.is_alive(), "DENY follower adopted the leader's broad approval"
+        with mod._lock:
+            (waiting_deny,) = mod._gateway_queues[SESSION_KEY]
+        assert waiting_deny.data["request_id"] == results["follower_id"]
+
+        assert mod.resolve_gateway_approval(SESSION_KEY, "once", request_id=results["follower_id"]) == 1
+        follower.join(10)
+        assert not follower.is_alive()
+        assert results["follower"]["choice"] == "once"
+    finally:
+        mod.resolve_gateway_approval(SESSION_KEY, "deny", resolve_all=True)
+        leader.join(10)
+        follower.join(10)
+
+
+def test_smart_denied_broad_gateway_reply_is_normalized_to_once(monkeypatch):
+    _clear()
+    monkeypatch.setattr(wait_mod._ctx, "_fire_approval_hook", lambda name, **kwargs: None)
+    result = {}
+
+    def notify(data):
+        assert data["smart_denied"] is True
+        assert mod.resolve_gateway_approval(SESSION_KEY, "always", request_id=data["request_id"]) == 1
+
+    thread = threading.Thread(
+        target=lambda: result.update(wait_mod._await_gateway_decision(
+            SESSION_KEY, notify, {**APPROVAL, "smart_denied": True})),
+        daemon=True,
+    )
+    thread.start()
+    thread.join(10)
+
+    assert not thread.is_alive()
+    assert result["resolved"] is True
+    assert result["choice"] == "once"
+
+
+def test_smart_denied_bulk_approval_waits_for_bulk_deny():
+    _clear()
+    entry = wait_mod._ApprovalEntry({**APPROVAL, "smart_denied": True})
+    with mod._lock:
+        mod._gateway_queues[SESSION_KEY] = [entry]
+
+    assert mod.resolve_gateway_approval(SESSION_KEY, "session", resolve_all=True) == 0
+    assert not entry.event.is_set()
+    assert mod.resolve_gateway_approval(SESSION_KEY, "deny", resolve_all=True) == 1
+    assert entry.result == "deny"
+    assert entry.event.is_set()
+
+
+def test_bulk_ordinary_approval_does_not_resolve_smart_denied():
+    _clear()
+    ordinary = wait_mod._ApprovalEntry(dict(APPROVAL))
+    denied = wait_mod._ApprovalEntry({**APPROVAL, "smart_denied": True, "request_id": "deny"})
+    with mod._lock:
+        mod._gateway_queues[SESSION_KEY] = [ordinary, denied]
+
+    assert mod.resolve_gateway_approval(SESSION_KEY, "session", resolve_all=True) == 1
+    assert ordinary.result == "session"
+    assert not denied.event.is_set()
+    assert mod.resolve_gateway_approval(SESSION_KEY, "once", request_id="deny") == 1
+    assert denied.result == "once"
+
+
+def test_empty_request_id_does_not_resolve_queue_head():
+    _clear()
+    ordinary = wait_mod._ApprovalEntry({**APPROVAL, "request_id": "ordinary"})
+    with mod._lock:
+        mod._gateway_queues[SESSION_KEY] = [ordinary]
+
+    assert mod.resolve_gateway_approval(SESSION_KEY, "once", request_id="") == 0
+    assert mod._gateway_queues[SESSION_KEY] == [ordinary]
+    assert not ordinary.event.is_set()
+    mod._gateway_queues.pop(SESSION_KEY)
+
+
+def test_unbound_once_does_not_skip_smart_denied_queue_head():
+    _clear()
+    denied = wait_mod._ApprovalEntry({**APPROVAL, "smart_denied": True, "request_id": "deny"})
+    ordinary = wait_mod._ApprovalEntry({**APPROVAL, "request_id": "ordinary"})
+    with mod._lock:
+        mod._gateway_queues[SESSION_KEY] = [denied, ordinary]
+
+    assert mod.resolve_gateway_approval(SESSION_KEY, "once") == 0
+    assert not denied.event.is_set()
+    assert not ordinary.event.is_set()
+    assert mod.resolve_gateway_approval(SESSION_KEY, "deny", resolve_all=True) == 2
diff --git a/tests/tools/test_delegate_reviewer_signal.py b/tests/tools/test_delegate_reviewer_signal.py
new file mode 100644
index 000000000000..5134df35ed23
--- /dev/null
+++ b/tests/tools/test_delegate_reviewer_signal.py
@@ -0,0 +1,6 @@
+from tools.delegate_tool_config import _subagent_auto_approve
+
+
+def test_reviewer_text_alone_does_not_override_ordinary_prompt():
+    assert _subagent_auto_approve("command", "**Reviewer:** flagged as dangerous") == "once"
+    assert _subagent_auto_approve("command", "other wording", smart_denied=True) == "deny"
diff --git a/tests/tools/test_plugin_approval_review.py b/tests/tools/test_plugin_approval_review.py
index 4b2f17496d88..3ef2f5eb88a3 100644
--- a/tests/tools/test_plugin_approval_review.py
+++ b/tests/tools/test_plugin_approval_review.py
@@ -21,7 +21,10 @@ def _gateway_session(monkeypatch):
     monkeypatch.setattr(tools_approval_context, "_get_approval_mode", lambda: "manual")
     monkeypatch.setattr(approval, "_presence", lambda cb=None: (None, False, True, False))
     approval._denial_tally.pop("review-session", None)
-    yield
+    try:
+        yield
+    finally:
+        approval._denial_tally.pop("review-session", None)
 
 
 def _capture_human(monkeypatch, choice="once"):
@@ -85,6 +88,18 @@ def test_reviewer_approve_still_asks_the_human(monkeypatch):
     assert "passed" in seen["description"]
 
 
+@pytest.mark.parametrize("verdict, prompts", [("approve", False), ("escalate", False), ("deny", True)])
+def test_review_first_honours_cached_grant_except_deny(monkeypatch, verdict, prompts):
+    monkeypatch.setattr(approval, "is_approved", lambda *_: True)
+    seen = _capture_human(monkeypatch)
+    calls = _verdict(monkeypatch, verdict)
+    result = request_tool_approval("terminal", "why", rule_key="k", display="cmd", review=True)
+    assert calls == ["cmd"]
+    assert result["approved"] is True
+    assert bool(seen) is prompts
+    if prompts:
+        assert seen["smart_denied"] is True
+
 def test_vault_metadata_is_in_reviewer_display_and_human_prompt(monkeypatch):
 
     monkeypatch.setattr("agent.vault_backends.describe_ref", lambda item, field=None: {
@@ -212,6 +227,89 @@ def _approve(command, description, **kwargs):
         assert "**Reviewer:** flagged as dangerous: Pipes the key to a remote host." in result["message"]
 
 
+def test_reviewer_deny_with_auto_approve_callback_is_blocked(monkeypatch):
+    from tools.delegate_tool import _subagent_auto_approve
+
+    monkeypatch.setattr(approval, "_presence", lambda cb=None: (cb, True, False, False))
+    _verdict_with_note(monkeypatch, "deny", "Pipes the key to a remote host.")
+
+    result = request_tool_approval(
+        "vault_secret", "why", rule_key="k", display="cmd", review=True,
+        approval_callback=_subagent_auto_approve,
+    )
+
+    assert result["approved"] is False
+    assert result["outcome"] == "denied"
+    assert "**Reviewer:** flagged as dangerous" in result["message"]
+    assert approval._denial_tally["review-session"] == 1
+
+
+def test_reviewer_deny_once_resets_tally(monkeypatch):
+    monkeypatch.setattr(approval, "_presence", lambda cb=None: (cb, True, False, False))
+    choices = iter(("once", "deny"))
+    monkeypatch.setattr(approval, "_get_denial_breaker_threshold", lambda: 2)
+    _verdict_with_note(monkeypatch, "deny", "Pipes the key to a remote host.")
+
+    def answer(*args, **kwargs):
+        return next(choices)
+
+    first = request_tool_approval(
+        "vault_secret", "why", rule_key="first", display="cmd", review=True, approval_callback=answer,
+    )
+    second = request_tool_approval(
+        "vault_secret", "why", rule_key="second", display="cmd", review=True, approval_callback=answer,
+    )
+
+    assert first["approved"] is True
+    assert second["approved"] is False
+    assert "CIRCUIT BREAKER:" not in second["message"]
+    assert approval._denial_tally["review-session"] == 1
+
+
+@pytest.mark.parametrize("choice", ["deny", "timeout"])
+def test_reviewer_deny_refusal_includes_breaker_addendum(monkeypatch, choice):
+    monkeypatch.setattr(approval, "_presence", lambda cb=None: (cb, True, False, False))
+    monkeypatch.setattr(approval, "_get_denial_breaker_threshold", lambda: 2)
+    approval._denial_tally["review-session"] = 1
+    _verdict_with_note(monkeypatch, "deny", "Pipes the key to a remote host.")
+    result = request_tool_approval(
+        "vault_secret", "why", rule_key="k", display="cmd", review=True,
+        approval_callback=lambda *args, **kwargs: choice,
+    )
+
+    assert "CIRCUIT BREAKER:" in result["message"]
+
+
+def test_reviewer_deny_broad_gateway_reply_is_once_and_not_saved(monkeypatch):
+    from tools import approval_context
+
+    monkeypatch.setattr(approval, "_presence", lambda cb=None: (cb, False, True, False))
+    _verdict_with_note(monkeypatch, "deny", "Pipes the key to a remote host.")
+    committed_choices = []
+    persisted = []
+    monkeypatch.setattr(approval_context, "_fire_approval_hook",
+                        lambda name, **kwargs: committed_choices.append(kwargs["choice"])
+                        if name == "post_approval_response" else None)
+    monkeypatch.setattr(approval, "_persist_choice", lambda *args: persisted.append(args))
+
+    def reply(data):
+        assert data["smart_denied"] is True
+        assert data["allow_session"] is False
+        assert data["allow_permanent"] is False
+        assert approval.resolve_gateway_approval("review-session", "session", request_id=data["request_id"]) == 1
+
+    approval.register_gateway_notify("review-session", reply)
+    try:
+        result = request_tool_approval("vault_secret", "why", rule_key="k", display="cmd", review=True)
+    finally:
+        approval.unregister_gateway_notify("review-session")
+
+    assert result["approved"] is True
+    assert committed_choices == ["once"]
+    assert persisted == []
+
+
+
 def test_a_stale_reason_does_not_leak_into_the_next_prompt(monkeypatch):
     from tools import approval_smart
 
diff --git a/tests/tools/test_reviewer_fifo_guard.py b/tests/tools/test_reviewer_fifo_guard.py
new file mode 100644
index 000000000000..462a92d38ea5
--- /dev/null
+++ b/tests/tools/test_reviewer_fifo_guard.py
@@ -0,0 +1,16 @@
+from tools import approval
+
+
+def test_review_first_ignores_cached_approval(monkeypatch):
+    monkeypatch.setattr(approval.approval_context, "is_plugin_origin_turn", lambda: False)
+    monkeypatch.setattr(approval, "_yolo_active", lambda: False)
+    monkeypatch.setattr(approval.approval_context, "_get_approval_mode", lambda: "ask")
+    monkeypatch.setattr(approval, "is_approved", lambda *args: True)
+    monkeypatch.setattr(approval, "_presence", lambda cb: (cb, True, False, False))
+    monkeypatch.setattr(approval, "_smart_verdict", lambda *args: "deny")
+    monkeypatch.setattr(approval, "_record_denial", lambda *args: None)
+    seen = {}
+    monkeypatch.setattr(approval, "_human_decision", lambda *args, **kwargs: seen.update(kwargs) or {})
+    approval._run_approval_gate(pattern_key="key", description="reason", display_target="action",
+                                autoapprove_log_prefix="test", review_first=True)
+    assert seen["smart_denied"] is True
diff --git a/tools/approval.py b/tools/approval.py
index 2d38c8c42e0f..e170f27cd36a 100644
--- a/tools/approval.py
+++ b/tools/approval.py
@@ -155,23 +155,35 @@ def resolve_gateway_approval(session_key: str, choice: str,
         queue = _gateway_queues.get(session_key)
         if not queue:
             return 0
-        if request_id:
+        if request_id is not None:
             targets = [entry for entry in queue if entry.data.get("request_id") == request_id]
             if not targets:
                 return 0
             queue[:] = [entry for entry in queue if entry not in targets]
         elif resolve_all:
-            targets = list(queue)
-            queue.clear()
+            if choice == "deny":
+                targets = list(queue)
+                queue.clear()
+            else:
+                # Bulk approval cannot establish consent for an individual warning.
+                targets = [entry for entry in queue if not entry.data.get("smart_denied")]
+                queue[:] = [entry for entry in queue if entry.data.get("smart_denied")]
         else:
-            targets = [queue.pop(0)]
+            target = queue[0]
+            if choice != "deny" and target.data.get("smart_denied"):
+                return 0
+            targets = [target]
+            queue.pop(0)
         if not queue:
             _gateway_queues.pop(session_key, None)
         # Popping the entry and committing its outcome are ONE critical section: the waiter's
         # ``_drop_entry`` reads ``entry.result`` under this same lock after its deadline check, so a
         # choice acked to the client here can never be popped-and-lost as a timeout (#112548).
         for entry in targets:
-            entry.result = choice
+            effective_choice = choice
+            if entry.data.get("smart_denied") and choice in {"session", "always"}:
+                effective_choice = "once"
+            entry.result = effective_choice
             if reason:
                 entry.reason = reason
             entry.event.set()
@@ -836,13 +848,14 @@ def _human_decision(spec: _GateSpec, *, command: str, description: str,
     allow_permanent = not one_shot
 
     def deny(template: str, outcome: str, **fmt) -> dict:
-        breaker = ""
-        if "{breaker}" in template:
-            breaker = _denial_breaker_addendum(session_key)
+        has_breaker = "{breaker}" in template
+        breaker = _denial_breaker_addendum(session_key) if has_breaker or smart_denied else ""
         deny_reason = fmt.pop("deny_reason", None)
         extra = {"deny_reason": deny_reason} if "reason" in fmt else {}
-        return _denied(template.format(description=description, breaker=breaker, **fmt),
-                       pattern_key=pattern_key, description=description,
+        message = template.format(description=description, breaker=breaker, **fmt)
+        if smart_denied and not has_breaker:
+            message += breaker
+        return _denied(message, pattern_key=pattern_key, description=description,
                        outcome=outcome, noun=spec.noun, **extra)
 
     def grant(choice: str) -> dict:
@@ -850,6 +863,8 @@ def grant(choice: str) -> dict:
             _persist_choice(session_key, choice, pattern_keys)
         if spec.user_approved:
             return _user_approved(session_key, description)
+        if smart_denied:
+            _reset_denials(session_key)
         return _approved()
 
     if spec.transport:
@@ -984,10 +999,11 @@ def _presence(approval_callback=None) -> tuple:
 
 
 # Reviewer verdict appended to a plugin approval's reason when the plugin asked for review first.
+REVIEWER_DENY_NOTE = "**Reviewer:** flagged as dangerous"
 _REVIEW_NOTES = {
     "approve": "**Reviewer:** passed",
     "escalate": "**Reviewer:** unsure, check it carefully",
-    "deny": "**Reviewer:** flagged as dangerous",
+    "deny": REVIEWER_DENY_NOTE,
 }
 
 
@@ -1019,14 +1035,14 @@ def _run_approval_gate(
     Unattended deny text is ``ctx.block_message(subject, noun, advice)`` unless the caller passes
     an explicit ``*_deny_message`` (the file-tool write gates word their own).
     ``review_first``: run the smart reviewer before the human, whatever ``approvals.mode`` is.
-    Every verdict asks the human; a reviewer DENY is a one-operation owner override, so the
-    reviewer can never approve or block the request on its own.
+    A reviewer DENY is a one-operation owner override and always asks the human; other verdicts
+    honour cached human grants.
     """
     plugin_origin = approval_context.is_plugin_origin_turn()
     session_key = get_current_session_key()
     if not plugin_origin and (_yolo_active() or approval_context._get_approval_mode() == "off"):
         return _approved()
-    if not plugin_origin and is_approved(session_key, pattern_key):
+    if not plugin_origin and not review_first and is_approved(session_key, pattern_key):
         return _approved()
     approval_callback, is_cli, is_gateway, is_ask = _presence(approval_callback)
     if not is_cli and not is_gateway and not is_ask:
@@ -1075,6 +1091,8 @@ def _run_approval_gate(
             reviewer_denied = True
         else:
             reviewer_denied = False
+        if not reviewer_denied and not plugin_origin and is_approved(session_key, pattern_key):
+            return _approved()
         description = f"{description}\n{_review_line(verdict, note)}"
     else:
         reviewer_denied = False
diff --git a/tools/approval_gateway_wait.py b/tools/approval_gateway_wait.py
index 1265a445cb69..621d21233a68 100644
--- a/tools/approval_gateway_wait.py
+++ b/tools/approval_gateway_wait.py
@@ -169,16 +169,15 @@ def _enqueue() -> None:
         register_prepared_approval(session_key, entry)
         _approval._gateway_queues.setdefault(session_key, []).append(entry)
 
-    # The leader lookup and our enqueue are ONE critical section: split, two identical requests both
-    # miss each other, both enqueue and the user gets duplicate prompts. A ``once`` releases the
-    # followers to prompt for themselves; they re-coalesce behind whichever re-prompts first, so the
-    # user still sees one prompt at a time (the answered leader is already off the queue).
+    # Lookup and enqueue must be atomic to coalesce concurrent identical requests.
+    # When the leader finishes, followers re-enter and prompt again. Reviewer DENY never shares consent.
     while True:
         with _approval._lock:
             leader = None
-            if not preparing_terminal_approval():
+            if not preparing_terminal_approval() and not approval_data.get("smart_denied"):
                 leader = next((e for e in _approval._gateway_queues.get(session_key, [])
-                               if e.data.get("command") == approval_data.get("command")
+                               if not e.data.get("smart_denied")
+                               and e.data.get("command") == approval_data.get("command")
                                and list(e.data.get("pattern_keys") or []) == keys), None)
             if leader is None:
                 _enqueue()
diff --git a/tools/delegate_tool_config.py b/tools/delegate_tool_config.py
index 357d1ba43a8f..29b3bb377bc9 100644
--- a/tools/delegate_tool_config.py
+++ b/tools/delegate_tool_config.py
@@ -33,12 +33,7 @@ def _cfg() -> dict:
 
 
 # ── Subagent approval callbacks ─────────────────────────────────────────────
-# Subagent worker threads don't inherit the CLI's threading.local approval
-# callback, so prompt_dangerous_approval() would fall back to input() and
-# deadlock the parent's prompt_toolkit TUI. Every worker gets a non-interactive
-# callback via ThreadPoolExecutor(initializer=...): deny by default, approve when
-# delegation.subagent_auto_approve is true. Both warn for audit. Gateway sessions
-# are unaffected (they resolve approvals via tools/approval.py's per-session queue).
+# Callbacks are non-interactive because input() would deadlock the parent TUI.
 def _subagent_auto_deny(command: str, description: str, **kwargs) -> str:
     """Auto-deny (safe default): returns 'deny' so the child sees a recoverable refusal."""
     logger.warning(
@@ -48,7 +43,11 @@ def _subagent_auto_deny(command: str, description: str, **kwargs) -> str:
     return "deny"
 
 def _subagent_auto_approve(command: str, description: str, **kwargs) -> str:
-    """Auto-approve (opt-in YOLO via delegation.subagent_auto_approve): returns 'once'."""
+    """Auto-approve ordinary prompts; reviewer DENY is refused, never auto-approved."""
+    if kwargs.get("smart_denied"):
+        logger.warning("Subagent blocked reviewer-denied command without human approval: %s (%s)",
+                       command, description)
+        return "deny"
     logger.warning("Subagent auto-approved dangerous command: %s (%s)", command, description)
     return "once"