fix(gateway): fail fast on undeliverable approval notifications

gateway/run.py

@@ -15605,6 +15605,12 @@ def _approval_notify_sync(approval_data: dict) -> None:
                     f"Reply `/approve` to execute, `/approve session` to approve this pattern "
                     f"for the session, `/approve always` to approve permanently, or `/deny` to cancel."
                 )
+                # Propagate delivery failure to the approval system so it fails
+                # fast with BLOCKED instead of waiting out the gateway_timeout
+                # (default 5 min). Adapters that don't support push delivery
+                # (e.g. APIServerAdapter.send returns SendResult(success=False))
+                # would otherwise leave the agent thread blocked on
+                # entry.event.wait() with no path to ever resolve. See #19731.
                 try:
                     _approval_send_fut = safe_schedule_threadsafe(
                         _status_adapter.send(
@@ -15616,10 +15622,23 @@ def _approval_notify_sync(approval_data: dict) -> None:
                         logger=logger,
                         log_message="Approval text-send scheduling error",
                     )
-                    if _approval_send_fut is not None:
-                        _approval_send_fut.result(timeout=15)
+                    if _approval_send_fut is None:
+                        raise RuntimeError(
+                            "Approval text-send scheduling failed: loop unavailable"
+                        )
+                    _send_result = _approval_send_fut.result(timeout=15)
                 except Exception as _e:
                     logger.error("Failed to send approval request: %s", _e)
+                    raise
+                if not getattr(_send_result, "success", False):
+                    _err = getattr(_send_result, "error", None) or "send returned success=False"
+                    logger.error(
+                        "Approval delivery rejected by %s adapter (%s); denying instead of hanging.",
+                        type(_status_adapter).__name__, _err,
+                    )
+                    raise RuntimeError(
+                        f"approval delivery failed via {type(_status_adapter).__name__}: {_err}"
+                    )
 
             # Prepend pending model switch note so the model knows about the switch
             _pending_notes = getattr(self, '_pending_model_notes', {})

tests/tools/test_approval_plugin_hooks.py

@@ -142,4 +142,54 @@ class TestGatewayPathFiresHooks:
     approval event until resolve_gateway_approval() is called from another
     thread."""
 
+    def test_notify_cb_failure_blocks_fast_without_waiting_for_timeout(
+        self, isolated_session, monkeypatch
+    ):
+        """Regression for #19731: when the gateway can't deliver the approval
+        request (e.g. APIServerAdapter.send returns SendResult(success=False)
+        because it has no push channel), the gateway-side notify_cb raises and
+        the approval flow must return BLOCKED immediately rather than blocking
+        the agent thread on event.wait() for the full gateway_timeout (5 min).
+        """
+        import threading
+
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+        monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual")
+        # Use a deliberately long gateway_timeout: the test should finish in
+        # well under a second because notify_cb raises, not because the
+        # timeout fires. If the test takes anywhere near gateway_timeout,
+        # the fail-fast path is broken.
+        monkeypatch.setattr(
+            approval_module, "_get_approval_config", lambda: {"gateway_timeout": 60}
+        )
+
+        def failing_notify_cb(approval_data):
+            raise RuntimeError(
+                "approval delivery failed via APIServerAdapter: "
+                "API server uses HTTP request/response, not send()"
+            )
 
+        register_gateway_notify(isolated_session, failing_notify_cb)
+        result_holder = {}
+
+        def run_guard():
+            with patch("hermes_cli.plugins.invoke_hook", return_value=[]):
+                result_holder["result"] = check_all_command_guards(
+                    "rm -rf /tmp/test-notify-fail", "local",
+                )
+
+        t = threading.Thread(target=run_guard, daemon=True)
+        t.start()
+        # Generous join cap; the actual return path is synchronous after
+        # notify_cb raises, so this should complete in milliseconds.
+        t.join(timeout=5)
+        assert not t.is_alive(), "Agent thread hung after notify_cb raised"
+        unregister_gateway_notify(isolated_session)
+
+        result = result_holder["result"]
+        assert result["approved"] is False
+        assert "BLOCKED" in result["message"]
+        # Queue should be drained — no orphaned approval entries left behind.
+        assert isolated_session not in approval_module._gateway_queues