feat(gateway): add active-wake operator receipts

gateway/kanban_watchers.py

@@ -18,6 +18,13 @@
 from pathlib import Path
 from typing import Any, Optional
 
+from tools.send_message_tool import (
+    _sanitize_active_wake_text,
+    _sanitize_error_text,
+    _stable_correlation_id,
+    _trigger_adapter_active_wake,
+)
+
 # Match the logger run.py uses (logging.getLogger(__name__) where __name__ ==
 # "gateway.run") so extracted log records keep their original logger name.
 logger = logging.getLogger("gateway.run")
@@ -352,13 +359,30 @@ def _collect():
                             sub["chat_id"], sub.get("thread_id") or "",
                         )
                         try:
-                            await adapter.send(
+                            send_result = await adapter.send(
                                 sub["chat_id"], msg, metadata=metadata,
                             )
                             logger.debug(
                                 "kanban notifier: delivered %s event for %s to %s/%s on board %s",
                                 kind, sub["task_id"], platform_str, sub["chat_id"], board_slug,
                             )
+                            if bool(sub.get("trigger_agent")):
+                                receipt = await self._kanban_active_wake_receipt(
+                                    send_result=send_result,
+                                    platform=plat,
+                                    platform_name=platform_str,
+                                    chat_id=str(sub["chat_id"]),
+                                    thread_id=(sub.get("thread_id") or None),
+                                    message=msg,
+                                    adapter=adapter,
+                                )
+                                await asyncio.to_thread(
+                                    self._kanban_record_notify_receipt,
+                                    sub,
+                                    kind,
+                                    receipt,
+                                    board_slug,
+                                )
                             # After delivering the text notification, surface
                             # any artifact paths the worker referenced in
                             # ``kanban_complete(summary=..., artifacts=[...])``
@@ -440,6 +464,160 @@ def _collect():
                     return
                 await asyncio.sleep(1)
 
+    async def _kanban_active_wake_receipt(
+        self,
+        *,
+        send_result: Any,
+        platform: Any,
+        platform_name: str,
+        chat_id: str,
+        thread_id: Optional[str],
+        message: str,
+        adapter: Any,
+    ) -> dict[str, Any]:
+        """Return a passive-send + active-wake receipt for a notifier event.
+
+        The notifier already performed the visible send before this helper is
+        called. Active wake is a separate synthetic inbound event; its status is
+        reported independently so a successful chat send never masquerades as a
+        successful agent wake.
+        """
+        receipt = self._kanban_normalize_send_result(send_result)
+        receipt.setdefault("success", True)
+        receipt["receipt_correlation"] = _stable_correlation_id(
+            platform_name, chat_id, thread_id, message
+        )
+        receipt["platform"] = platform_name
+        receipt["chat_id"] = str(chat_id)
+        if thread_id:
+            receipt["thread_id"] = str(thread_id)
+        receipt["active_wake_required"] = True
+
+        if not bool(receipt.get("success")):
+            receipt["scheduled_agent"] = False
+            receipt["triggered_agent"] = False
+            receipt["trigger_error"] = "SEND_FAILED"
+            return receipt
+
+        loop = getattr(self, "_gateway_loop", None)
+        wake_text = _sanitize_active_wake_text(message)
+        trigger_result = _trigger_adapter_active_wake(
+            platform=platform,
+            adapter=adapter,
+            loop=loop,
+            platform_name=platform_name,
+            chat_id=chat_id,
+            thread_id=thread_id,
+            message=wake_text,
+            runner=self,
+        )
+        acceptance = trigger_result.get("_acceptance") if isinstance(trigger_result, dict) else None
+        receipt.update({k: v for k, v in trigger_result.items() if not str(k).startswith("_")})
+        if trigger_result.get("scheduled_agent"):
+            # If scheduled onto this running loop, yield briefly so the gateway
+            # can resolve/claim the real operator session before the receipt is
+            # persisted.  Do not wait for the model turn itself; acceptance is a
+            # pre-turn state transition that mutates ``acceptance`` quickly.
+            if isinstance(acceptance, dict):
+                try:
+                    if asyncio.get_running_loop() is loop:
+                        for _ in range(10):
+                            if acceptance.get("active_wake_status") != "scheduled":
+                                break
+                            await asyncio.sleep(0)
+                except RuntimeError:
+                    pass
+                receipt.update(acceptance)
+        return receipt
+
+    @staticmethod
+    def _kanban_normalize_send_result(send_result: Any) -> dict[str, Any]:
+        if isinstance(send_result, dict):
+            receipt = dict(send_result)
+            if "success" not in receipt:
+                receipt["success"] = True
+        else:
+            receipt = {"success": bool(getattr(send_result, "success", True))}
+            message_id = getattr(send_result, "message_id", None)
+            if message_id:
+                receipt["message_id"] = str(message_id)
+            error = getattr(send_result, "error", None)
+            if error:
+                receipt["error"] = _sanitize_error_text(str(error))
+        if "error" in receipt and receipt["error"]:
+            receipt["error"] = _sanitize_error_text(str(receipt["error"]))
+        return receipt
+
+    def _kanban_record_notify_receipt(
+        self,
+        sub: dict,
+        event_kind: str,
+        receipt: dict[str, Any],
+        board: Optional[str] = None,
+    ) -> None:
+        """Persist a sanitized notifier active-wake receipt event."""
+        from hermes_cli import kanban_db as _kb
+
+        allowed = {
+            "success",
+            "message_id",
+            "receipt_correlation",
+            "scheduled_agent",
+            "triggered_agent",
+            "trigger_error",
+            "platform",
+            "chat_id",
+            "thread_id",
+            "active_wake_required",
+            "active_wake_status",
+            "accepted_by_session",
+            "started_by_session",
+            "target_session_key",
+        }
+        payload = {key: receipt[key] for key in allowed if key in receipt}
+        payload["notified_event_kind"] = event_kind
+        payload.setdefault("platform", sub.get("platform"))
+        payload.setdefault("chat_id", sub.get("chat_id"))
+        if sub.get("thread_id"):
+            payload.setdefault("thread_id", sub.get("thread_id"))
+        if payload.get("trigger_error"):
+            payload["trigger_error"] = _sanitize_error_text(str(payload["trigger_error"]))
+        conn = _kb.connect(board=board)
+        try:
+            _kb._append_event(
+                conn,
+                sub["task_id"],
+                "notify_active_wake_receipt",
+                payload,
+            )
+            try:
+                from hermes_cli import kanban_db_ack_ledger as _ack
+                _ack.record_ack_active_wake(
+                    conn,
+                    task_id=sub["task_id"],
+                    triggered_agent=bool(payload.get("scheduled_agent")),
+                    trigger_error=payload.get("trigger_error"),
+                    correlation_id=payload.get("receipt_correlation"),
+                    status=str(payload.get("active_wake_status") or ("scheduled" if payload.get("scheduled_agent") else "failed")),
+                    accepted_by_session=bool(payload.get("accepted_by_session")),
+                    started_by_session=bool(payload.get("started_by_session")),
+                    target_session_key=payload.get("target_session_key"),
+                )
+                if payload.get("accepted_by_session") or payload.get("started_by_session"):
+                    _ack.record_ack_operator_receipt(
+                        conn,
+                        task_id=sub["task_id"],
+                        status="observed",
+                        actor="gateway",
+                        actor_ref=payload.get("target_session_key"),
+                        correlation_id=payload.get("receipt_correlation"),
+                    )
+            except Exception as ledger_exc:
+                logger.debug("kanban notifier: ack ledger receipt shadow-write failed: %s", ledger_exc)
+            conn.commit()
+        finally:
+            conn.close()
+
     def _kanban_advance(
         self, sub: dict, cursor: int, board: Optional[str] = None,
     ) -> None:

gateway/platforms/api_server.py

@@ -1118,6 +1118,182 @@ async def _handle_health_detailed(self, request: "web.Request") -> "web.Response
             "pid": os.getpid(),
         })
 
+    def _is_loopback_request(self, request: "web.Request") -> bool:
+        """Return True only for direct localhost requests."""
+        candidates = {"127.0.0.1", "::1", "localhost"}
+        remote = getattr(request, "remote", "") or ""
+        if remote in candidates:
+            return True
+        try:
+            peer = request.transport.get_extra_info("peername") if request.transport else None
+            if isinstance(peer, (tuple, list)) and peer and str(peer[0]) in candidates:
+                return True
+        except Exception:
+            pass
+        return False
+
+    async def _handle_active_wake_smoke(self, request: "web.Request") -> "web.Response":
+        """POST /api/debug/active-wake-smoke — gateway-in-process wake smoke.
+
+        Narrow localhost-only debug hook for proving the live gateway boundary
+        that standalone CLI/cron calls cannot prove: visible send followed by a
+        synthetic internal MessageEvent scheduled on the target adapter.
+        """
+        auth_err = self._check_auth(request)
+        if auth_err:
+            return auth_err
+        if not self._is_loopback_request(request):
+            return web.json_response(
+                {"error": {"message": "active_wake smoke is loopback-only", "code": "loopback_only"}},
+                status=403,
+            )
+
+        try:
+            payload = await request.json()
+        except Exception:
+            payload = {}
+        if not isinstance(payload, dict):
+            payload = {}
+
+        target = str(payload.get("target") or "").strip()
+        if not target or ":" not in target:
+            return web.json_response(
+                {"error": {"message": "target is required, e.g. discord:<channel_id>", "code": "missing_target"}},
+                status=400,
+            )
+        platform_name, target_ref = target.split(":", 1)
+        platform_name = platform_name.strip().lower()
+        target_ref = target_ref.strip()
+
+        try:
+            from gateway.config import Platform
+            platform = Platform(platform_name)
+        except Exception:
+            return web.json_response(
+                {"error": {"message": f"unknown platform: {platform_name}", "code": "unknown_platform"}},
+                status=400,
+            )
+
+        try:
+            from tools.send_message_tool import (
+                _parse_target_ref,
+                _sanitize_active_wake_text,
+                _sanitize_error_text,
+            )
+        except Exception as exc:
+            return web.json_response(
+                {
+                    "success": False,
+                    "scheduled_agent": False,
+                    "triggered_agent": False,
+                    "trigger_error": f"IMPORT_FAILED:{type(exc).__name__}",
+                },
+                status=500,
+            )
+
+        chat_id, thread_id, is_explicit = _parse_target_ref(platform_name, target_ref)
+        if not chat_id or not is_explicit:
+            return web.json_response(
+                {"error": {"message": "target must be an explicit platform id", "code": "non_explicit_target"}},
+                status=400,
+            )
+
+        nonce = str(payload.get("nonce") or f"AW-{int(time.time())}").strip()[:120]
+        correlation_id = str(payload.get("correlation_id") or f"activewake-smoke-{nonce}").strip()[:200]
+        message = str(payload.get("message") or "").strip()
+        if not message:
+            message = (
+                "@agent ACTIVE_WAKE_SMOKE\n"
+                f"Nonce: {nonce}\n"
+                "Task: If this message arrived through active_wake synthetic inbound, "
+                "reply in this channel with exactly one line:\n\n"
+                f"SMOKE_ACK {nonce}\n\n"
+                "No research, no routing, no follow-up work."
+            )
+
+        try:
+            from gateway.run import _gateway_runner_ref
+            runner = _gateway_runner_ref()
+        except Exception:
+            runner = None
+        adapter = None
+        if runner is not None:
+            try:
+                adapter = runner.adapters.get(platform)
+            except Exception:
+                adapter = None
+        if runner is None or adapter is None:
+            return web.json_response({
+                "success": False,
+                "receipt_correlation": correlation_id,
+                "scheduled_agent": False,
+                "triggered_agent": False,
+                "trigger_error": "NOT_WIRED",
+            })
+
+        try:
+            metadata = {"thread_id": thread_id} if thread_id else None
+            send_result = await adapter.send(chat_id=str(chat_id), content=message, metadata=metadata)
+        except Exception as exc:
+            return web.json_response({
+                "success": False,
+                "receipt_correlation": correlation_id,
+                "scheduled_agent": False,
+                "triggered_agent": False,
+                "trigger_error": _sanitize_error_text(str(exc)) or "SEND_FAILED",
+            })
+
+        receipt = {
+            "success": bool(getattr(send_result, "success", False)),
+            "message_id": getattr(send_result, "message_id", None),
+            "receipt_correlation": correlation_id,
+        }
+        if not receipt["success"]:
+            receipt.update({
+                "scheduled_agent": False,
+                "triggered_agent": False,
+                "trigger_error": _sanitize_error_text(str(getattr(send_result, "error", ""))) or "SEND_FAILED",
+            })
+            return web.json_response(receipt)
+
+        try:
+            from gateway.session import SessionSource
+            from gateway.platforms.base import MessageEvent, MessageType
+
+            wake_text = _sanitize_active_wake_text(message)
+            source = SessionSource(
+                platform=platform,
+                chat_id=str(chat_id),
+                chat_type="group",
+                thread_id=str(thread_id) if thread_id else None,
+                # Match production active-wake routing: do not append a
+                # synthetic participant id that would create a ghost session.
+                user_id=None,
+                user_name="Hermes Active Wake Smoke",
+                is_bot=False,
+                message_id=None,
+            )
+            wake_event = MessageEvent(
+                text=wake_text,
+                message_type=MessageType.TEXT,
+                source=source,
+                internal=True,
+            )
+            task = asyncio.create_task(adapter.handle_message(wake_event))
+            self._background_tasks.add(task)
+            task.add_done_callback(self._background_tasks.discard)
+            # Scheduling-only receipt; not proof of operator-session acceptance.
+            receipt["scheduled_agent"] = True
+            receipt["triggered_agent"] = True
+            return web.json_response(receipt)
+        except Exception as exc:
+            receipt.update({
+                "scheduled_agent": False,
+                "triggered_agent": False,
+                "trigger_error": _sanitize_error_text(str(exc)) or "ACTIVE_WAKE_FAILED",
+            })
+            return web.json_response(receipt)
+
     async def _handle_models(self, request: "web.Request") -> "web.Response":
         """GET /v1/models — return hermes-agent as an available model."""
         auth_err = self._check_auth(request)
@@ -4247,6 +4423,7 @@ async def connect(self) -> bool:
             assert self._app is not None
             self._app.router.add_get("/health", self._handle_health)
             self._app.router.add_get("/health/detailed", self._handle_health_detailed)
+            self._app.router.add_post("/api/debug/active-wake-smoke", self._handle_active_wake_smoke)
             self._app.router.add_get("/v1/health", self._handle_health)
             self._app.router.add_get("/v1/models", self._handle_models)
             self._app.router.add_get("/v1/capabilities", self._handle_capabilities)