diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 21753054f018..461372083ff0 100644 --- a/gateway/kanban_watchers.py +++ b/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: diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 09d0dc227a25..82c5c61cc395 100644 --- a/gateway/platforms/api_server.py +++ b/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:", "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) diff --git a/gateway/run.py b/gateway/run.py index f5a411244aaa..31bf6de5b8eb 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4125,7 +4125,7 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session # creating a session. The busy path must enforce the same check; # otherwise unauthorized users in shared threads (Slack/Telegram/Discord) # can inject messages into an active session they don't own. - if not self._is_user_authorized(event.source): + if not event.internal and not self._is_user_authorized(event.source): logger.warning( "Dropping message from unauthorized user in active session: " "user=%s (%s), platform=%s, session=%s", @@ -7236,6 +7236,21 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: # Otherwise control/session commands like /new or /help get silently # consumed as update answers instead of being dispatched normally. _quick_key = self._session_key_for_source(source) + _active_wake_acceptance = getattr(event, "_hermes_active_wake_acceptance", None) + if isinstance(_active_wake_acceptance, dict) and is_internal: + # This internal wake has already been routed through the session-key + # builder with no synthetic participant id. Mark acceptance only + # after the live gateway resolves the concrete operator/channel key. + _target_key = getattr(event, "_hermes_active_wake_target_session_key", None) + if isinstance(_target_key, str) and _target_key and _target_key != _quick_key: + logger.warning( + "active_wake target key mismatch: requested=%s resolved=%s", + _target_key, + _quick_key, + ) + _active_wake_acceptance["target_session_key"] = _quick_key + _active_wake_acceptance["accepted_by_session"] = True + _active_wake_acceptance["active_wake_status"] = "accepted" _update_prompts = getattr(self, "_update_prompt_pending", {}) if _update_prompts.get(_quick_key): raw = (event.text or "").strip() @@ -8309,6 +8324,9 @@ async def _do_undo(): self._active_session_leases[_quick_key] = _active_session_lease self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL self._running_agents_ts[_quick_key] = time.time() + if isinstance(_active_wake_acceptance, dict) and is_internal: + _active_wake_acceptance["started_by_session"] = True + _active_wake_acceptance["active_wake_status"] = "started" _run_generation = self._begin_session_run_generation(_quick_key) try: diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index dbfd778daf9b..68282665c870 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -341,6 +341,49 @@ async def _handle_kanban_command(self, event: MessageEvent) -> str: is_create = action == "create" + # Gateway-originated creates should use the same explicit origin + # contract as terminal-created tasks instead of relying only on a + # post-create notify-subscribe side effect. Inject the supported CLI + # flags when the message source has platform/chat metadata and the user + # did not already provide an origin target. The legacy post-create + # subscribe below remains as a harmless idempotent backstop for old + # output shapes and board parsing edge cases. + if is_create: + has_origin_flags = any( + tok == "--origin-platform" + or tok.startswith("--origin-platform=") + or tok == "--origin-chat-id" + or tok.startswith("--origin-chat-id=") + for tok in tokens + ) + if not has_origin_flags: + source = event.source + platform = getattr(source, "platform", None) + platform_value = getattr(platform, "value", None) + platform_str = (platform_value if platform_value is not None else str(platform or "")).lower() + chat_id = str(getattr(source, "chat_id", "") or "") + thread_id = str(getattr(source, "thread_id", "") or "") + user_id = str(getattr(source, "user_id", "") or "") + if platform_str and chat_id: + origin_args = [ + "--origin-platform", platform_str, + "--origin-chat-id", chat_id, + ] + if thread_id: + origin_args.extend(["--origin-thread-id", thread_id]) + if user_id: + origin_args.extend(["--origin-user-id", user_id]) + try: + profile_getter = getattr(self, "_active_profile_name", None) + profile_name = getattr(self, "_kanban_notifier_profile", None) or ( + profile_getter() if callable(profile_getter) else None + ) + except Exception: + profile_name = None + if profile_name: + origin_args.extend(["--notifier-profile", profile_name]) + text = (text.rstrip() + " " + shlex.join(origin_args)).strip() + try: output = await asyncio.to_thread(run_slash, text) except Exception as exc: # pragma: no cover - defensive diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 31c4bf68ae85..ca8c7cf148d2 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -362,6 +362,18 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu "that require immediate human ops (R3 gate) " "to skip the brief running-to-blocked transition.") p_create.add_argument("--json", action="store_true", help="Emit JSON output") + p_create.add_argument("--origin-platform", default=None, + help="Gateway platform to notify for terminal ACKs") + p_create.add_argument("--origin-chat-id", default=None, + help="Gateway chat/channel id to notify for terminal ACKs") + p_create.add_argument("--origin-thread-id", default=None, + help="Optional gateway thread/topic id for terminal ACKs") + p_create.add_argument("--origin-user-id", default=None, + help="Optional originating user id for notification records") + p_create.add_argument("--notifier-profile", default=None, + help="Profile gateway that owns/delivers the ACK subscription") + p_create.add_argument("--ack-trigger-agent", action="store_true", + help="Mark the ACK subscription as active-wake capable") # --- swarm --- p_swarm = sub.add_parser( @@ -690,6 +702,8 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu "--notifier-profile", default=None, help="Profile gateway that owns/delivers this subscription (default: active profile)", ) + p_nsub.add_argument("--trigger-agent", action="store_true", + help="Mark this subscription as active-wake capable") p_nlist = sub.add_parser( "notify-list", @@ -987,7 +1001,6 @@ def _profile_author() -> str: except Exception: return "user" - # --------------------------------------------------------------------------- # Boards management (hermes kanban boards …) # --------------------------------------------------------------------------- @@ -1303,6 +1316,20 @@ def _cmd_assignees(args: argparse.Namespace) -> int: def _cmd_create(args: argparse.Namespace) -> int: + origin_platform = (getattr(args, "origin_platform", None) or "").strip() + origin_chat_id = (getattr(args, "origin_chat_id", None) or "").strip() + # Do not infer durable subscriptions from body prose. A body line such as + # "Origin/return_to: ..." is useful context for humans/notifiers, but it is + # not an explicit request to persist a notify subscription or to mark an + # active wake as required. Durable ACK subscriptions are created only from + # the explicit origin flags below. + if bool(origin_platform) ^ bool(origin_chat_id): + print( + "kanban: pass both --origin-platform and --origin-chat-id " + "to wire terminal ACKs", + file=sys.stderr, + ) + return 2 try: ws_kind, ws_path = _parse_workspace_flag(args.workspace) branch_name = _parse_branch_flag(getattr(args, "branch", None)) @@ -1325,6 +1352,7 @@ def _cmd_create(args: argparse.Namespace) -> int: file=sys.stderr, ) return 2 + ack_subscription: dict[str, Any] | None = None with kb.connect_closing() as conn: task_id = kb.create_task( conn, @@ -1348,18 +1376,42 @@ def _cmd_create(args: argparse.Namespace) -> int: initial_status=getattr(args, "initial_status", "running"), ) task = kb.get_task(conn, task_id) + if task is None: + print(f"kanban: created task {task_id} but could not read it back", file=sys.stderr) + return 1 + if origin_platform and origin_chat_id: + thread_id = getattr(args, "origin_thread_id", None) + # Passive delivery is the safe default. `trigger_agent` is an + # active-wake claim and must be set only by the explicit flag; + # body prose alone is not proof that the notifier/gateway can + # wake an agent session. + trigger_agent = bool(getattr(args, "ack_trigger_agent", False)) + notifier_profile = getattr(args, "notifier_profile", None) or _profile_author() + kb.add_notify_sub( + conn, + task_id=task_id, + platform=origin_platform, + chat_id=origin_chat_id, + thread_id=thread_id, + user_id=getattr(args, "origin_user_id", None), + notifier_profile=notifier_profile, + trigger_agent=trigger_agent, + ) + ack_subscription = { + "task_id": task_id, + "platform": origin_platform, + "chat_id": origin_chat_id, + "thread_id": thread_id or "", + "trigger_agent": trigger_agent, + "notifier_profile": notifier_profile, + } if getattr(args, "json", False): - print(json.dumps(_task_to_dict(task), indent=2, ensure_ascii=False)) + payload = _task_to_dict(task) + if ack_subscription is not None: + payload["ack_subscription"] = ack_subscription + print(json.dumps(payload, indent=2, ensure_ascii=False)) else: print(f"Created {task_id} ({task.status}, assignee={task.assignee or '-'})") - - # Warn when the task would sit in `ready` because no dispatcher is - # present. Only warn on ready+assigned tasks — triage/todo are - # expected to sit idle until promoted, and unassigned tasks - # can't be dispatched. Skipped in --json mode so the stdout - # stream stays strictly machine-parseable for callers (the JSON - # response itself carries enough info for them to decide if - # they want to check dispatcher presence separately). if task.status == "ready" and task.assignee: running, message = _check_dispatcher_presence() if not running and message: @@ -2426,6 +2478,7 @@ def _cmd_notify_subscribe(args: argparse.Namespace) -> int: platform=args.platform, chat_id=args.chat_id, thread_id=args.thread_id, user_id=args.user_id, notifier_profile=args.notifier_profile or _profile_author(), + trigger_agent=bool(getattr(args, "trigger_agent", False)), ) print(f"Subscribed {args.platform}:{args.chat_id}" + (f":{args.thread_id}" if args.thread_id else "") diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c82d762d5924..a959e93f118f 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1130,6 +1130,7 @@ class Event: thread_id TEXT NOT NULL DEFAULT '', user_id TEXT, notifier_profile TEXT, + trigger_agent INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, last_event_id INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (task_id, platform, chat_id, thread_id) @@ -1145,6 +1146,87 @@ class Event: CREATE INDEX IF NOT EXISTS idx_runs_status ON task_runs(status); CREATE INDEX IF NOT EXISTS idx_attachments_task ON task_attachments(task_id, created_at); CREATE INDEX IF NOT EXISTS idx_notify_task ON kanban_notify_subs(task_id); + +-- --------------------------------------------------------------------------- +-- Shadow ACK delivery ledger (M1 root ACK). +-- Durable, queryable records for terminal-state delivery without changing +-- live notifier behavior. All writes are shadow-only; readers can join to +-- kanban_notify_subs and task_events for a complete picture. +-- --------------------------------------------------------------------------- + +-- Terminal work verdict recorded at completion time. Independent of whether +-- the origin ACK/wake actually reached a target. +CREATE TABLE IF NOT EXISTS ack_task_verdict ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + run_id INTEGER, + event_id INTEGER, + verdict TEXT, + status TEXT, + summary_ref TEXT, + summary_safe TEXT, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_ack_task_verdict_task ON ack_task_verdict(task_id); + +-- Explicit subscription / origin_return path for a task. This shadows the +-- existing kanban_notify_subs row at terminal time; it must NOT be inferred +-- from prose body text. +CREATE TABLE IF NOT EXISTS ack_subscription ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + subscription_id INTEGER, + platform TEXT, + chat_id TEXT, + thread_id TEXT, + notifier_profile TEXT, + desired_delivery_mode TEXT, + active_wake_required INTEGER NOT NULL DEFAULT 0, + operator_receipt_required INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_ack_subscription_task ON ack_subscription(task_id); + +-- Passive terminal delivery attempt (gateway notifier text/artifact send). +CREATE TABLE IF NOT EXISTS ack_passive_delivery ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + subscription_id INTEGER, + message_id TEXT, + status TEXT, + error_safe TEXT, + correlation_id TEXT, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_ack_passive_delivery_task ON ack_passive_delivery(task_id); + +-- Active wake attempt (synthetic inbound triggered by terminal state). +CREATE TABLE IF NOT EXISTS ack_active_wake ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + subscription_id INTEGER, + triggered_agent INTEGER NOT NULL DEFAULT 0, + trigger_error TEXT, + correlation_id TEXT, + status TEXT, + accepted_by_session INTEGER NOT NULL DEFAULT 0, + started_by_session INTEGER NOT NULL DEFAULT 0, + target_session_key TEXT, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_ack_active_wake_task ON ack_active_wake(task_id); + +-- Operator receipt state for human-in-the-loop ACK tracking. +CREATE TABLE IF NOT EXISTS ack_operator_receipt ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + status TEXT NOT NULL, + actor TEXT, + actor_ref TEXT, + correlation_id TEXT, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_ack_operator_receipt_task ON ack_operator_receipt(task_id); """ @@ -1703,6 +1785,21 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: conn, "tasks", "session_id", "session_id TEXT" ) + notify_table_exists = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='kanban_notify_subs'" + ).fetchone() is not None + if notify_table_exists: + notify_cols = { + row["name"] for row in conn.execute("PRAGMA table_info(kanban_notify_subs)") + } + if "trigger_agent" not in notify_cols: + _add_column_if_missing( + conn, + "kanban_notify_subs", + "trigger_agent", + "trigger_agent INTEGER NOT NULL DEFAULT 0", + ) + # Indexes over additive ``tasks`` columns must be created after the # columns exist. Keeping them in SCHEMA_SQL breaks legacy boards: SQLite # parses each statement in ``executescript`` against the live schema, so a @@ -1744,6 +1841,22 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: conn, "kanban_notify_subs", "notifier_profile", "notifier_profile TEXT" ) + active_wake_table_exists = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='ack_active_wake'" + ).fetchone() is not None + if active_wake_table_exists: + wake_cols = { + row["name"] for row in conn.execute("PRAGMA table_info(ack_active_wake)") + } + for col, ddl in ( + ("status", "status TEXT"), + ("accepted_by_session", "accepted_by_session INTEGER NOT NULL DEFAULT 0"), + ("started_by_session", "started_by_session INTEGER NOT NULL DEFAULT 0"), + ("target_session_key", "target_session_key TEXT"), + ): + if col not in wake_cols: + _add_column_if_missing(conn, "ack_active_wake", col, ddl) + # One-shot backfill: any task that is 'running' before runs existed # had its claim_lock / claim_expires / worker_pid on the task row. # Synthesize a matching task_runs row so subsequent end-run / heartbeat @@ -2276,6 +2389,54 @@ def create_task( "INSERT OR IGNORE INTO task_links (parent_id, child_id) VALUES (?, ?)", (pid, task_id), ) + if parents: + # ACK-edge inheritance: if a parent/root task is already + # wired for terminal notifications, every child in the + # durable graph must inherit that return path. Otherwise a + # child can BLOCK while the origin lane never hears it. + placeholders = ",".join("?" * len(parents)) + parent_subs = conn.execute( + "SELECT * FROM kanban_notify_subs " + f"WHERE task_id IN ({placeholders}) " + "ORDER BY created_at ASC", + parents, + ).fetchall() + for sub in parent_subs: + conn.execute( + """ + INSERT OR IGNORE INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, user_id, + notifier_profile, trigger_agent, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, + sub["platform"], + sub["chat_id"], + sub["thread_id"] or "", + sub["user_id"], + sub["notifier_profile"], + 1 if sub["trigger_agent"] else 0, + now, + ), + ) + if sub["trigger_agent"]: + conn.execute( + """ + UPDATE kanban_notify_subs + SET trigger_agent = 1 + WHERE task_id = ? + AND platform = ? + AND chat_id = ? + AND thread_id = ? + """, + ( + task_id, + sub["platform"], + sub["chat_id"], + sub["thread_id"] or "", + ), + ) _append_event( conn, task_id, @@ -3559,6 +3720,128 @@ def _scan_prose_for_phantom_ids( return [m for m in unique if m not in existing] +# Substrings that, when present in a terminal task's summary / result / +# metadata, mean the origin wake/ACK relay did NOT reach a live target even +# though the task itself reached a terminal verdict. Worker CLI sessions +# frequently cannot send messages, so the gateway relay path is the only way +# the origin learns the task finished — when that path emits one of these +# strings the completion is durable but the ACK is missing. Matched +# case-insensitively. See :func:`classify_ack_relay`. +ACK_RELAY_FAILURE_PATTERNS: tuple[str, ...] = ( + "no messaging targets", + "origin relay could not be sent", + "no live gateway runner", +) + + +def _parse_task_verdict(text: str) -> Optional[str]: + """Extract a ``Verdict: `` token from terminal handoff text. + + Review / fan-in tasks encode their work decision as a ``Verdict:`` + line (e.g. ``Verdict: GO`` / ``Verdict: BLOCK``). Returns the + normalised upper-case token (``-`` folded to ``_``) or ``None`` when + no verdict line is present. Deliberately verdict-only: it says + nothing about whether the origin was woken — that is ``ack_status``. + """ + if not text: + return None + m = re.search(r"verdict\s*[:=]\s*([A-Za-z][A-Za-z_-]*)", text, re.IGNORECASE) + if not m: + return None + return m.group(1).strip().upper().replace("-", "_") + + +def _has_origin_return_intent(text: Optional[str]) -> bool: + """Return True when task prose declares explicit origin/return ACK intent. + + A generic prose mention of the word ``origin`` is not enough: ordinary + tasks may ask to research the origin of a bug/topic without asking the + control plane to wake a return target. Treat only labelled return-target + declarations and structured origin fields as ACK intent. + """ + if not text: + return False + for line in str(text).splitlines(): + if re.match( + r"^\s*(?:origin(?:\s*/\s*return[_-]?to)?|return[_-]?to|return\s+to)\s*[:=]", + line, + re.IGNORECASE, + ): + return True + if re.match( + r"^\s*origin_(?:platform|chat_id|thread_id|user_id)\s*[:=]", + line, + re.IGNORECASE, + ): + return True + return False + + +def classify_ack_relay( + summary: Optional[str] = None, + result: Optional[str] = None, + metadata: Optional[dict] = None, + *, + notify_subs: Optional[list] = None, + task_body: Optional[str] = None, +) -> dict: + """Classify a terminal task's handoff into separate verdict + ack signals. + + Returns a dict with: + + * ``task_verdict`` — the work decision (``"GO"`` / ``"BLOCK"`` / …) parsed + from a ``Verdict:`` line, or ``None``. + * ``ack_status`` — one of: + - ``"failed"``: a relay-failure pattern was found in the text/metadata + (the origin ACK demonstrably could not be delivered). + - ``"missing_subscription"``: task body declares Origin/return_to + intent, ``notify_subs`` was supplied, and zero subscription rows + existed at terminal time. This is a typed control-plane delivery + problem, not a successful ACK. + - ``"ambiguous"``: no failure string, but a verdict is present and + ``notify_subs`` was supplied and empty — the origin relay had no + target at terminal time, but no explicit origin intent was found. + - ``"unknown"``: no positive or negative ACK signal. + * ``relay_failure`` — ``True`` iff a failure pattern matched. + * ``matched`` — the failure substrings found (lower-cased), in pattern order. + + ``task_verdict`` and ``ack_status`` are intentionally independent so a + done BLOCK/GO task is never treated as proof the origin was woken. + """ + haystack_parts: list[str] = [] + for part in (summary, result): + if part: + haystack_parts.append(str(part)) + if isinstance(metadata, dict): + try: + haystack_parts.append(json.dumps(metadata, ensure_ascii=False)) + except Exception: + haystack_parts.append(str(metadata)) + haystack = "\n".join(haystack_parts) + hay_lower = haystack.lower() + + matched = [p for p in ACK_RELAY_FAILURE_PATTERNS if p in hay_lower] + relay_failure = bool(matched) + + verdict = _parse_task_verdict(haystack) + + if relay_failure: + ack_status = "failed" + elif notify_subs is not None and len(notify_subs) == 0 and _has_origin_return_intent(task_body): + ack_status = "missing_subscription" + elif verdict is not None and notify_subs is not None and len(notify_subs) == 0: + ack_status = "ambiguous" + else: + ack_status = "unknown" + + return { + "task_verdict": verdict, + "ack_status": ack_status, + "relay_failure": relay_failure, + "matched": matched, + } + + class HallucinatedCardsError(ValueError): """Raised by ``complete_task`` when ``created_cards`` contains ids that don't exist or weren't created by the completing worker. @@ -3644,6 +3927,11 @@ def complete_task( else: verified_cards = [] + task_row = conn.execute( + "SELECT body FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + task_body = task_row["body"] if task_row else None + with write_txn(conn): if expected_run_id is None: cur = conn.execute( @@ -3747,6 +4035,58 @@ def complete_task( }, run_id=run_id, ) + # Durable ACK/relay classification. The work verdict can be terminal + # (done, BLOCK/GO) while the origin wake/ACK relay never reached a + # target — worker CLI sessions often cannot send messages, so a relay + # failure string in the handoff (or an empty notify list at terminal + # time) means the origin may be silently waiting. Record a durable + # ``ack_relay_status`` event so diagnostics can surface it; keep + # ``task_verdict`` and ``ack_status`` as distinct payload fields so a + # done verdict is never mistaken for a delivered ACK. Emitted at most + # once per terminal transition (this block only runs when the status + # update above flipped exactly one row), so it cannot storm. + ack = classify_ack_relay( + summary=summary, + result=result, + metadata=metadata, + notify_subs=list_notify_subs(conn, task_id), + task_body=task_body, + ) + if ack["ack_status"] in ("failed", "ambiguous", "missing_subscription"): + event_kind = ( + "delivery_problem" + if ack["ack_status"] == "missing_subscription" + else "ack_relay_status" + ) + payload = { + "ack_status": ack["ack_status"], + "task_verdict": ack["task_verdict"], + "matched": ack["matched"], + "source": "completion", + } + if ack["ack_status"] == "missing_subscription": + payload["problem_type"] = "missing_subscription" + with write_txn(conn): + _append_event( + conn, + task_id, + event_kind, + payload, + run_id=run_id, + ) + # Shadow ACK ledger: record the terminal verdict and any explicit + # subscription snapshot. This is intentionally independent of live + # notifier behavior and independent of the ack_relay_status event. + # Only explicit kanban_notify_subs rows are shadow-copied; prose + # Origin/return_to bodies are NOT treated as subscriptions. + _shadow_write_ack_ledger_on_complete( + conn, + task_id=task_id, + run_id=run_id, + event_id=None, + summary=summary, + result=result, + ) # Successful completion — wipe the consecutive-failures counter. # Failure history stays on the event log for audit; the counter # just tracks "is there a current pathology the breaker should @@ -3759,6 +4099,54 @@ def complete_task( return True +def _shadow_write_ack_ledger_on_complete( + conn: sqlite3.Connection, + *, + task_id: str, + run_id: Optional[int], + event_id: Optional[int], + summary: Optional[str], + result: Optional[str], +) -> None: + """Write shadow ACK ledger rows for a completed task. + + Records ``ack_task_verdict`` (work decision) and, iff an explicit + ``kanban_notify_subs`` row exists, an ``ack_subscription`` snapshot. + Does nothing for tasks without an explicit subscription, so a task + with only a prose ``Origin/return_to`` body gets a verdict row but no + invented subscription row. + """ + from hermes_cli import kanban_db_ack_ledger as _ack + + verdict = _parse_task_verdict(" ".join(filter(None, [summary, result]))) + _ack.record_ack_task_verdict( + conn, + task_id=task_id, + run_id=run_id, + event_id=event_id, + verdict=verdict, + status="done", + summary_ref=None, + summary_safe=summary if summary is not None else result, + ) + + subs = list_notify_subs(conn, task_id) + if not subs: + return + for sub in subs: + _ack.record_ack_subscription( + conn, + task_id=task_id, + subscription_id=None, + platform=sub.get("platform"), + chat_id=sub.get("chat_id"), + thread_id=sub.get("thread_id") or "", + notifier_profile=sub.get("notifier_profile"), + desired_delivery_mode="passive", + active_wake_required=bool(sub.get("trigger_agent")), + operator_receipt_required=False, + ) + # --------------------------------------------------------------------------- # Workspace / tmux cleanup # --------------------------------------------------------------------------- @@ -4135,6 +4523,10 @@ def block_task( expected_run_id: Optional[int] = None, ) -> bool: """Transition ``running -> blocked``.""" + task_row = conn.execute( + "SELECT body FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + task_body = task_row["body"] if task_row else None with write_txn(conn): if expected_run_id is None: cur = conn.execute( @@ -4179,7 +4571,27 @@ def block_task( summary=reason, ) _append_event(conn, task_id, "blocked", {"reason": reason}, run_id=run_id) - return True + ack = classify_ack_relay( + summary=reason, + notify_subs=list_notify_subs(conn, task_id), + task_body=task_body, + ) + if ack["ack_status"] == "missing_subscription": + with write_txn(conn): + _append_event( + conn, + task_id, + "delivery_problem", + { + "ack_status": "missing_subscription", + "problem_type": "missing_subscription", + "task_verdict": ack["task_verdict"], + "matched": ack["matched"], + "source": "block", + }, + run_id=run_id, + ) + return True @@ -7359,31 +7771,59 @@ def add_notify_sub( thread_id: Optional[str] = None, user_id: Optional[str] = None, notifier_profile: Optional[str] = None, + trigger_agent: bool = False, ) -> None: - """Register a gateway source that wants terminal-state notifications - for ``task_id``. Idempotent on (task, platform, chat, thread).""" + """Register a gateway source that wants terminal-state notifications. + + Idempotent on (task, platform, chat, thread). Repeated calls preserve the + original notifier owner while allowing a later caller to upgrade the edge + to active-wake semantics. + """ now = int(time.time()) with write_txn(conn): conn.execute( """ INSERT OR IGNORE INTO kanban_notify_subs - (task_id, platform, chat_id, thread_id, user_id, notifier_profile, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?) + (task_id, platform, chat_id, thread_id, user_id, + notifier_profile, trigger_agent, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, - (task_id, platform, chat_id, thread_id or "", user_id, notifier_profile, now), + ( + task_id, + platform, + chat_id, + thread_id or "", + user_id, + notifier_profile, + 1 if trigger_agent else 0, + now, + ), ) if notifier_profile: - # Self-heal legacy rows that predate notifier ownership by - # backfilling only when the existing value is unset. conn.execute( """ UPDATE kanban_notify_subs SET notifier_profile = ? - WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? + WHERE task_id = ? + AND platform = ? + AND chat_id = ? + AND thread_id = ? AND (notifier_profile IS NULL OR notifier_profile = '') """, (notifier_profile, task_id, platform, chat_id, thread_id or ""), ) + if trigger_agent: + conn.execute( + """ + UPDATE kanban_notify_subs + SET trigger_agent = 1 + WHERE task_id = ? + AND platform = ? + AND chat_id = ? + AND thread_id = ? + """, + (task_id, platform, chat_id, thread_id or ""), + ) def list_notify_subs( diff --git a/hermes_cli/kanban_db_ack_ledger.py b/hermes_cli/kanban_db_ack_ledger.py new file mode 100644 index 000000000000..86bacd0f8e57 --- /dev/null +++ b/hermes_cli/kanban_db_ack_ledger.py @@ -0,0 +1,286 @@ +"""Shadow ACK delivery ledger helpers for Hermes Kanban. + +This module is split out of ``kanban_db.py`` to keep the core DB file from +growing further and to make the ledger surface easy to test in isolation. +All helpers operate on an existing kanban SQLite connection and write to the +``ack_*`` tables defined in ``kanban_db.SCHEMA_SQL``. +""" + +from __future__ import annotations + +import hashlib +import sqlite3 +import time +from typing import Optional + +from hermes_cli.kanban_db import write_txn + + +def _safe_summary(text: Optional[str], max_len: int = 400) -> Optional[str]: + """Return a single-line, length-capped summary safe for ledger storage. + + Strips surrounding whitespace, collapses newlines, and truncates. Does not + redact secrets — callers should run sensitive text through + ``_sanitize_error_text`` / ``_sanitize_active_wake_text`` before storing + error or wake payloads. + """ + if not text: + return None + s = str(text).strip().replace("\r\n", " ").replace("\n", " ").strip() + if not s: + return None + return s[:max_len] + + +def _ack_correlation_id( + kind: str, + task_id: str, + platform: Optional[str], + chat_id: Optional[str], + thread_id: Optional[str], + now: int, +) -> str: + """Build a short, non-secret correlation id for an ACK delivery row. + + The id is stable for the same (kind, task, subscription target, second) + tuple so idempotent callers do not create duplicate rows. Raw message text + is NOT included in the correlation value. + """ + key = f"{kind}:{task_id}:{platform or ''}:{chat_id or ''}:{thread_id or ''}:{now}" + return f"ack_{hashlib.sha256(key.encode('utf-8')).hexdigest()[:20]}" + + +def record_ack_task_verdict( + conn: sqlite3.Connection, + *, + task_id: str, + run_id: Optional[int] = None, + event_id: Optional[int] = None, + verdict: Optional[str] = None, + status: Optional[str] = None, + summary_ref: Optional[str] = None, + summary_safe: Optional[str] = None, + created_at: Optional[int] = None, +) -> int: + """Shadow-write a durable task verdict record. Returns the new row id.""" + now = created_at if created_at is not None else int(time.time()) + with write_txn(conn): + cur = conn.execute( + """ + INSERT INTO ack_task_verdict + (task_id, run_id, event_id, verdict, status, + summary_ref, summary_safe, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, + run_id, + event_id, + verdict, + status, + summary_ref, + _safe_summary(summary_safe), + now, + ), + ) + return int(cur.lastrowid) + + +def record_ack_subscription( + conn: sqlite3.Connection, + *, + task_id: str, + subscription_id: Optional[int] = None, + platform: Optional[str] = None, + chat_id: Optional[str] = None, + thread_id: Optional[str] = None, + notifier_profile: Optional[str] = None, + desired_delivery_mode: Optional[str] = None, + active_wake_required: bool = False, + operator_receipt_required: bool = False, + created_at: Optional[int] = None, +) -> int: + """Shadow-write a snapshot of the explicit subscription used for ACKs.""" + now = created_at if created_at is not None else int(time.time()) + with write_txn(conn): + cur = conn.execute( + """ + INSERT INTO ack_subscription + (task_id, subscription_id, platform, chat_id, thread_id, + notifier_profile, desired_delivery_mode, active_wake_required, + operator_receipt_required, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, + subscription_id, + platform, + chat_id, + thread_id or "", + notifier_profile, + desired_delivery_mode or "passive", + 1 if active_wake_required else 0, + 1 if operator_receipt_required else 0, + now, + ), + ) + return int(cur.lastrowid) + + +def record_ack_passive_delivery( + conn: sqlite3.Connection, + *, + task_id: str, + subscription_id: Optional[int] = None, + message_id: Optional[str] = None, + status: Optional[str] = None, + error: Optional[str] = None, + correlation_id: Optional[str] = None, + created_at: Optional[int] = None, +) -> int: + """Shadow-write a passive delivery attempt. ``error`` is sanitized.""" + from tools.send_message_tool import _sanitize_error_text + + now = created_at if created_at is not None else int(time.time()) + safe_error = _sanitize_error_text(error) if error else None + corr = correlation_id or _ack_correlation_id( + "passive", task_id, None, None, None, now + ) + with write_txn(conn): + cur = conn.execute( + """ + INSERT INTO ack_passive_delivery + (task_id, subscription_id, message_id, status, error_safe, + correlation_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, + subscription_id, + message_id, + status, + safe_error, + corr, + now, + ), + ) + return int(cur.lastrowid) + + +def record_ack_active_wake( + conn: sqlite3.Connection, + *, + task_id: str, + subscription_id: Optional[int] = None, + triggered_agent: bool = False, + trigger_error: Optional[str] = None, + correlation_id: Optional[str] = None, + status: Optional[str] = None, + accepted_by_session: bool = False, + started_by_session: bool = False, + target_session_key: Optional[str] = None, + created_at: Optional[int] = None, +) -> int: + """Shadow-write an active wake attempt. ``trigger_error`` is sanitized.""" + from tools.send_message_tool import _sanitize_error_text + + now = created_at if created_at is not None else int(time.time()) + safe_error = _sanitize_error_text(trigger_error) if trigger_error else None + corr = correlation_id or _ack_correlation_id( + "active_wake", task_id, None, None, None, now + ) + with write_txn(conn): + cur = conn.execute( + """ + INSERT INTO ack_active_wake + (task_id, subscription_id, triggered_agent, trigger_error, + correlation_id, status, accepted_by_session, + started_by_session, target_session_key, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, + subscription_id, + 1 if triggered_agent else 0, + safe_error, + corr, + status, + 1 if accepted_by_session else 0, + 1 if started_by_session else 0, + target_session_key, + now, + ), + ) + return int(cur.lastrowid) + + +def record_ack_operator_receipt( + conn: sqlite3.Connection, + *, + task_id: str, + status: str, + actor: Optional[str] = None, + actor_ref: Optional[str] = None, + correlation_id: Optional[str] = None, + created_at: Optional[int] = None, +) -> int: + """Shadow-write an operator receipt state change. + + ``status`` must be one of: pending, observed, timed_out, escalated. + """ + if status not in {"pending", "observed", "timed_out", "escalated"}: + raise ValueError(f"invalid operator_receipt status: {status!r}") + now = created_at if created_at is not None else int(time.time()) + corr = correlation_id or _ack_correlation_id( + "operator_receipt", task_id, None, None, None, now + ) + with write_txn(conn): + cur = conn.execute( + """ + INSERT INTO ack_operator_receipt + (task_id, status, actor, actor_ref, correlation_id, created_at) + VALUES (?, ?, ?, ?, ?, ?) + """, + (task_id, status, actor, actor_ref, corr, now), + ) + return int(cur.lastrowid) + + +def list_ack_task_verdicts(conn: sqlite3.Connection, task_id: str) -> list[dict]: + rows = conn.execute( + "SELECT * FROM ack_task_verdict WHERE task_id = ? ORDER BY created_at ASC, id ASC", + (task_id,), + ).fetchall() + return [dict(r) for r in rows] + + +def list_ack_subscriptions(conn: sqlite3.Connection, task_id: str) -> list[dict]: + rows = conn.execute( + "SELECT * FROM ack_subscription WHERE task_id = ? ORDER BY created_at ASC, id ASC", + (task_id,), + ).fetchall() + return [dict(r) for r in rows] + + +def list_ack_passive_deliveries(conn: sqlite3.Connection, task_id: str) -> list[dict]: + rows = conn.execute( + "SELECT * FROM ack_passive_delivery WHERE task_id = ? ORDER BY created_at ASC, id ASC", + (task_id,), + ).fetchall() + return [dict(r) for r in rows] + + +def list_ack_active_wakes(conn: sqlite3.Connection, task_id: str) -> list[dict]: + rows = conn.execute( + "SELECT * FROM ack_active_wake WHERE task_id = ? ORDER BY created_at ASC, id ASC", + (task_id,), + ).fetchall() + return [dict(r) for r in rows] + + +def list_ack_operator_receipts(conn: sqlite3.Connection, task_id: str) -> list[dict]: + rows = conn.execute( + "SELECT * FROM ack_operator_receipt WHERE task_id = ? ORDER BY created_at ASC, id ASC", + (task_id,), + ).fetchall() + return [dict(r) for r in rows] diff --git a/hermes_cli/kanban_diagnostics.py b/hermes_cli/kanban_diagnostics.py index bef9bc8a97e2..52a73ae41438 100644 --- a/hermes_cli/kanban_diagnostics.py +++ b/hermes_cli/kanban_diagnostics.py @@ -974,6 +974,90 @@ def _rule_stranded_in_ready(task, events, runs, now, cfg) -> list[Diagnostic]: )] +def _rule_missing_ack_relay(task, events, runs, now, cfg) -> list[Diagnostic]: + """A terminal task reached a work verdict but the origin ACK/relay + could not be delivered (or had no target). + + Fires on active ``ack_relay_status`` events — those not superseded by a + later clean ``completed`` / ``edited`` (a retry that relayed fine clears + the stale signal). ``ack_status="failed"`` is an error (the relay + demonstrably failed); ``ack_status="ambiguous"`` is a warning (no target + at terminal time — could be delivered-then-removed or never subscribed). + + Keeps ``task_verdict`` and ``ack_status`` separate so a done BLOCK/GO + task is not silently treated as fully delivered. The recovery hint is + deliberately conservative: surface the gap and let the operator relay / + re-trigger, rather than auto-firing a wake. + """ + hits = _active_hallucination_events(events, "ack_relay_status") + if not hits: + return [] + # The latest active event wins for status/verdict; "failed" anywhere in + # the active run dominates "ambiguous". + statuses = [(_parse_payload(ev).get("ack_status") or "") for ev in hits] + failed = "failed" in statuses + ack_status = "failed" if failed else (statuses[-1] or "ambiguous") + matched: list[str] = [] + task_verdict = None + for ev in hits: + p = _parse_payload(ev) + if p.get("task_verdict"): + task_verdict = p.get("task_verdict") + for m in p.get("matched", []) or []: + if m not in matched: + matched.append(m) + severity = "error" if failed else "warning" + task_id = _task_field(task, "id") + actions: list[DiagnosticAction] = [ + DiagnosticAction( + kind="comment", + label="Relay the verdict to the origin and comment NEED_ACK_RELAY", + suggested=True, + ), + ] + if task_id: + actions.append(DiagnosticAction( + kind="cli_hint", + label=f"Inspect relay context: hermes kanban show {task_id}", + payload={"command": f"hermes kanban show {task_id}"}, + )) + verdict_str = task_verdict or "(none)" + if failed: + detail = ( + f"This task reached a terminal work verdict ({verdict_str}) but the " + f"origin ACK/relay could not be delivered " + f"({', '.join(matched) or 'relay failure'}). The worker could not " + f"message the origin and the gateway relay had no live target, so " + f"the origin may be silently waiting. The task is done, but the " + f"ACK is not — relay the result or re-trigger the origin wake." + ) + else: + detail = ( + f"This task reached a terminal work verdict ({verdict_str}) but no " + f"notify target was registered at completion time, so whether the " + f"origin was woken is ambiguous. Confirm the origin received the " + f"verdict; relay it manually if not." + ) + return [Diagnostic( + kind="missing_ack_relay", + severity=severity, + title=( + f"Verdict {verdict_str} done but origin ACK " + f"{'failed' if failed else 'unconfirmed'}" + ), + detail=detail, + actions=actions, + first_seen_at=_event_ts(hits[0]), + last_seen_at=_event_ts(hits[-1]), + count=len(hits), + data={ + "task_verdict": task_verdict, + "ack_status": ack_status, + "matched": matched, + }, + )] + + # Registry — order matters: rules higher on the list render first when # severity ties. Add new rules here. _RULES: list[RuleFn] = [ @@ -985,6 +1069,7 @@ def _rule_stranded_in_ready(task, events, runs, now, cfg) -> list[Diagnostic]: _rule_stuck_in_blocked, _rule_block_unblock_cycling, _rule_stranded_in_ready, + _rule_missing_ack_relay, ] @@ -999,6 +1084,7 @@ def _rule_stranded_in_ready(task, events, runs, now, cfg) -> list[Diagnostic]: "stuck_in_blocked", "block_unblock_cycling", "stranded_in_ready", + "missing_ack_relay", ) diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py index d6e1e588506a..f66f39a67c38 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/tests/gateway/test_api_server_runs.py @@ -10,13 +10,14 @@ import asyncio import threading +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from aiohttp import web from aiohttp.test_utils import TestClient, TestServer -from gateway.config import PlatformConfig +from gateway.config import PlatformConfig, Platform from gateway.platforms.api_server import ( APIServerAdapter, cors_middleware, @@ -52,6 +53,15 @@ def _create_runs_app(adapter: APIServerAdapter) -> web.Application: return app +def _create_debug_app(adapter: APIServerAdapter) -> web.Application: + """Create an aiohttp app with the local active-wake smoke route.""" + mws = [mw for mw in (cors_middleware, security_headers_middleware) if mw is not None] + app = web.Application(middlewares=mws) + app["api_server_adapter"] = adapter + app.router.add_post("/api/debug/active-wake-smoke", adapter._handle_active_wake_smoke) + return app + + def _make_slow_agent(**kwargs): """Create a mock agent that blocks in run_conversation until interrupted. @@ -527,3 +537,86 @@ async def test_stop_sends_sentinel_to_events_stream(self, adapter): body = await events_resp.text() # Stream should have received run.failed and closed assert "run.failed" in body or "stream closed" in body + + +class _FakeWakeAdapter: + def __init__(self): + self.sent = [] + self.events = [] + + async def send(self, *, chat_id, content, metadata=None): + self.sent.append({"chat_id": chat_id, "content": content, "metadata": metadata}) + return SimpleNamespace(success=True, message_id="msg-smoke-1", error=None) + + async def handle_message(self, event): + self.events.append(event) + + +class TestActiveWakeSmoke: + @pytest.mark.asyncio + async def test_smoke_sends_and_schedules_internal_event(self, auth_adapter): + app = _create_debug_app(auth_adapter) + fake_adapter = _FakeWakeAdapter() + fake_runner = SimpleNamespace(adapters={Platform.DISCORD: fake_adapter}) + + async with TestClient(TestServer(app)) as cli: + with patch("gateway.run._gateway_runner_ref", return_value=fake_runner): + resp = await cli.post( + "/api/debug/active-wake-smoke", + headers={"Authorization": "Bearer sk-secret"}, + json={ + "target": "discord:12345", + "nonce": "AW-TEST", + "correlation_id": "corr-test", + }, + ) + assert resp.status == 200 + data = await resp.json() + + assert data["success"] is True + assert data["message_id"] == "msg-smoke-1" + assert data["receipt_correlation"] == "corr-test" + assert data["scheduled_agent"] is True + assert data["triggered_agent"] is True + assert "accepted_by_session" not in data + assert "operator_receipt" not in data + assert fake_adapter.sent[0]["chat_id"] == "12345" + await asyncio.sleep(0) + assert len(fake_adapter.events) == 1 + event = fake_adapter.events[0] + assert event.internal is True + assert event.source.chat_id == "12345" + assert event.source.user_id is None + assert event.source.user_name == "Hermes Active Wake Smoke" + assert "SMOKE_ACK AW-TEST" in event.text + + @pytest.mark.asyncio + async def test_smoke_reports_not_wired_without_live_runner(self, auth_adapter): + app = _create_debug_app(auth_adapter) + async with TestClient(TestServer(app)) as cli: + with patch("gateway.run._gateway_runner_ref", return_value=None): + resp = await cli.post( + "/api/debug/active-wake-smoke", + headers={"Authorization": "Bearer sk-secret"}, + json={"target": "discord:12345", "correlation_id": "corr-nowire"}, + ) + assert resp.status == 200 + data = await resp.json() + + assert data == { + "success": False, + "receipt_correlation": "corr-nowire", + "scheduled_agent": False, + "triggered_agent": False, + "trigger_error": "NOT_WIRED", + } + + @pytest.mark.asyncio + async def test_smoke_requires_auth(self, auth_adapter): + app = _create_debug_app(auth_adapter) + async with TestClient(TestServer(app)) as cli: + resp = await cli.post( + "/api/debug/active-wake-smoke", + json={"target": "discord:12345"}, + ) + assert resp.status == 401 diff --git a/tests/gateway/test_busy_session_auth_bypass.py b/tests/gateway/test_busy_session_auth_bypass.py index b1c25a12d876..bc7f6247704b 100644 --- a/tests/gateway/test_busy_session_auth_bypass.py +++ b/tests/gateway/test_busy_session_auth_bypass.py @@ -38,7 +38,7 @@ # --------------------------------------------------------------------------- def _make_event(text="hello", chat_id="123", user_id="user1", user_name="TestUser", - platform_val="slack", thread_id="thread-abc"): + platform_val="slack", thread_id="thread-abc", internal=False): """Build a MessageEvent for a shared thread.""" source = SessionSource( platform=MagicMock(value=platform_val), @@ -53,6 +53,7 @@ def _make_event(text="hello", chat_id="123", user_id="user1", user_name="TestUse message_type=MessageType.TEXT, source=source, message_id="msg1", + internal=internal, ) return evt @@ -218,3 +219,73 @@ async def test_unauthorized_user_cannot_steer_active_agent(self): running_agent.steer.assert_not_called() # Nothing queued assert sk not in adapter._pending_messages + + @pytest.mark.asyncio + async def test_internal_synthetic_event_bypasses_busy_auth_drop(self): + """Internal active-wake events are gateway-owned and bypass human allowlist checks.""" + from gateway.run import GatewayRunner + + runner, _sentinel = _make_runner(authorized_users={"operator"}) + runner._busy_input_mode = "queue" + runner._busy_text_mode = "interrupt" + adapter = _make_adapter() + + event = _make_event( + text="wake operator", + user_id="hermes-active-wake", + user_name="Hermes Active Wake", + internal=True, + ) + sk = build_session_key(event.source) + + running_agent = MagicMock() + running_agent.get_activity_summary.return_value = {} + runner._running_agents[sk] = running_agent + runner._running_agents_ts[sk] = time.time() + runner.adapters[event.source.platform] = adapter + + result = await GatewayRunner._handle_active_session_busy_message( + runner, event, sk + ) + + # Upstream #49738 intentionally routes internal completion/wake events + # through the adapter fallthrough path while preserving the auth bypass: + # no unauthorized drop, no interrupt/steer, and no direct busy-path queue. + assert result is False + assert sk not in adapter._pending_messages + running_agent.steer.assert_not_called() + + @pytest.mark.asyncio + async def test_internal_active_wake_records_acceptance_and_started_status(self): + """Live gateway marks internal wake acceptance only after resolving/starting target session.""" + from gateway.run import GatewayRunner + + runner, _sentinel = _make_runner(authorized_users={"operator"}) + runner._startup_restore_in_progress = False + runner._update_prompt_pending = {} + runner._running_agents = {} + runner._running_agents_ts = {} + runner._draining = False + runner._busy_input_mode = "queue" + runner._is_telegram_topic_root_lobby = lambda source: False + runner._claim_active_session_slot = lambda session_key, source: (None, None) + runner._begin_session_run_generation = lambda session_key: 7 + runner._release_running_agent_state = MagicMock() + runner._handle_message_with_agent = AsyncMock(return_value="ok") + target_key = "agent:main:slack:channel:123:thread-abc" + runner._session_key_for_source = lambda source: target_key + + acceptance = {} + event = _make_event(text="wake operator", user_id="", user_name="Hermes Active Wake", internal=True) + event.source.user_id = None + setattr(event, "_hermes_active_wake_acceptance", acceptance) + setattr(event, "_hermes_active_wake_target_session_key", target_key) + + result = await GatewayRunner._handle_message(runner, event) + + assert result == "ok" + assert acceptance["target_session_key"] == target_key + assert acceptance["accepted_by_session"] is True + assert acceptance["started_by_session"] is True + assert acceptance["active_wake_status"] == "started" + runner._handle_message_with_agent.assert_awaited_once() diff --git a/tests/hermes_cli/test_kanban_ack_ledger.py b/tests/hermes_cli/test_kanban_ack_ledger.py new file mode 100644 index 000000000000..ccfd3d389e2d --- /dev/null +++ b/tests/hermes_cli/test_kanban_ack_ledger.py @@ -0,0 +1,309 @@ +"""Tests for the shadow ACK delivery ledger (M1 root ACK).""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb +from hermes_cli import kanban_db_ack_ledger as ack + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + """Isolated HERMES_HOME with an empty kanban DB.""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +# --------------------------------------------------------------------------- +# Schema / round-trip +# --------------------------------------------------------------------------- + + +def test_init_creates_ack_ledger_tables(kanban_home): + with kb.connect() as conn: + rows = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" + ).fetchall() + names = {r["name"] for r in rows} + assert { + "ack_task_verdict", + "ack_subscription", + "ack_passive_delivery", + "ack_active_wake", + "ack_operator_receipt", + } <= names + + +def test_task_verdict_round_trip(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="verdict task", assignee="worker") + rid = ack.record_ack_task_verdict( + conn, + task_id=tid, + run_id=7, + event_id=42, + verdict="GO", + status="done", + summary_ref="run_7_summary", + summary_safe="All good.\nSecond line.", + ) + rows = ack.list_ack_task_verdicts(conn, tid) + + assert len(rows) == 1 + row = rows[0] + assert row["id"] == rid + assert row["task_id"] == tid + assert row["run_id"] == 7 + assert row["event_id"] == 42 + assert row["verdict"] == "GO" + assert row["status"] == "done" + assert row["summary_ref"] == "run_7_summary" + assert row["summary_safe"] == "All good. Second line." + + +def test_subscription_round_trip(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="sub task", assignee="worker") + rid = ack.record_ack_subscription( + conn, + task_id=tid, + subscription_id=3, + platform="discord", + chat_id="1499390151393284106", + thread_id="123", + notifier_profile="gateway-a", + desired_delivery_mode="passive", + active_wake_required=True, + operator_receipt_required=True, + ) + rows = ack.list_ack_subscriptions(conn, tid) + + assert len(rows) == 1 + row = rows[0] + assert row["id"] == rid + assert row["task_id"] == tid + assert row["subscription_id"] == 3 + assert row["platform"] == "discord" + assert row["chat_id"] == "1499390151393284106" + assert row["thread_id"] == "123" + assert row["notifier_profile"] == "gateway-a" + assert row["desired_delivery_mode"] == "passive" + assert int(row["active_wake_required"]) == 1 + assert int(row["operator_receipt_required"]) == 1 + + +def test_passive_delivery_sanitizes_error(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="passive task", assignee="worker") + rid = ack.record_ack_passive_delivery( + conn, + task_id=tid, + subscription_id=5, + message_id="msg-1", + status="failed", + error="send failed: access_token=secret123 chat_id=1499390151393284106", + ) + rows = ack.list_ack_passive_deliveries(conn, tid) + + assert len(rows) == 1 + row = rows[0] + assert row["id"] == rid + assert row["status"] == "failed" + assert "secret123" not in row["error_safe"] + assert "access_token=***" in row["error_safe"] + assert row["correlation_id"].startswith("ack_") + + +def test_active_wake_sanitizes_trigger_error(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="wake task", assignee="worker") + rid = ack.record_ack_active_wake( + conn, + task_id=tid, + subscription_id=6, + triggered_agent=False, + trigger_error="wake refused: api_key=super_secret", + correlation_id="corr-1", + status="started", + accepted_by_session=True, + started_by_session=True, + target_session_key="agent:main:discord:group:1497895797579190357", + ) + rows = ack.list_ack_active_wakes(conn, tid) + + assert len(rows) == 1 + row = rows[0] + assert row["id"] == rid + assert int(row["triggered_agent"]) == 0 + assert "super_secret" not in row["trigger_error"] + assert "api_key=***" in row["trigger_error"] + assert row["correlation_id"] == "corr-1" + assert row["status"] == "started" + assert int(row["accepted_by_session"]) == 1 + assert int(row["started_by_session"]) == 1 + assert row["target_session_key"] == "agent:main:discord:group:1497895797579190357" + + +def test_operator_receipt_status_validation(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="receipt task", assignee="worker") + with pytest.raises(ValueError): + ack.record_ack_operator_receipt(conn, task_id=tid, status="bad_status") + rid = ack.record_ack_operator_receipt( + conn, + task_id=tid, + status="pending", + actor="operator-x", + actor_ref="discord:u:123", + ) + rows = ack.list_ack_operator_receipts(conn, tid) + + assert len(rows) == 1 + row = rows[0] + assert row["id"] == rid + assert row["status"] == "pending" + assert row["actor"] == "operator-x" + assert row["actor_ref"] == "discord:u:123" + + +# --------------------------------------------------------------------------- +# Completion shadow-write behavior +# --------------------------------------------------------------------------- + + +def test_complete_task_with_explicit_subscription_shadow_writes_ledger(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="wired task", assignee="worker") + kb.add_notify_sub( + conn, + task_id=tid, + platform="discord", + chat_id="1499390151393284106", + thread_id="123", + notifier_profile="gateway-a", + trigger_agent=True, + ) + kb.complete_task( + conn, + tid, + summary="Verdict: GO\nShipped.", + result="done", + ) + verdicts = ack.list_ack_task_verdicts(conn, tid) + subs = ack.list_ack_subscriptions(conn, tid) + + assert len(verdicts) == 1 + assert verdicts[0]["verdict"] == "GO" + assert verdicts[0]["status"] == "done" + assert verdicts[0]["summary_safe"].startswith("Verdict: GO") + assert "Shipped." in verdicts[0]["summary_safe"] + + assert len(subs) == 1 + assert subs[0]["platform"] == "discord" + assert subs[0]["chat_id"] == "1499390151393284106" + assert subs[0]["thread_id"] == "123" + assert subs[0]["notifier_profile"] == "gateway-a" + assert int(subs[0]["active_wake_required"]) == 1 + + +def test_complete_task_snapshots_all_explicit_subscriptions(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="multi sub task", assignee="worker") + kb.add_notify_sub( + conn, + task_id=tid, + platform="discord", + chat_id="1499390151393284106", + thread_id="123", + notifier_profile="gateway-a", + trigger_agent=True, + ) + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="987654321", + notifier_profile="gateway-b", + trigger_agent=False, + ) + kb.complete_task(conn, tid, summary="Verdict: GO\nDone.") + subs = ack.list_ack_subscriptions(conn, tid) + + assert len(subs) == 2 + by_target = {(row["platform"], row["chat_id"]): row for row in subs} + assert set(by_target) == { + ("discord", "1499390151393284106"), + ("telegram", "987654321"), + } + assert int(by_target[("discord", "1499390151393284106")]["active_wake_required"]) == 1 + assert int(by_target[("telegram", "987654321")]["active_wake_required"]) == 0 + + +def test_complete_task_without_subscription_does_not_invent_one(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="unwired task", assignee="worker") + kb.complete_task(conn, tid, summary="Verdict: BLOCK\nNo origin.") + verdicts = ack.list_ack_task_verdicts(conn, tid) + subs = ack.list_ack_subscriptions(conn, tid) + + assert len(verdicts) == 1 + assert verdicts[0]["verdict"] == "BLOCK" + assert subs == [] + + +def test_complete_task_with_origin_body_only_classified_missing_subscription(kanban_home): + """Fixture reproducing t_a11319e9 shape. + + The task is created directly via ``create_task`` with a body that contains + ``Origin/return_to:`` prose but no explicit subscription. Completion must + record a verdict but must NOT infer a subscription from the body text. + """ + body = "Origin/return_to: Discord Devhub #research (<#1499390151393284106>)\nDo the work." + with kb.connect() as conn: + tid = kb.create_task( + conn, + title="origin prose task", + body=body, + assignee="ccreviewer", + ) + kb.complete_task(conn, tid, summary="Verdict: GO\nDone.") + verdicts = ack.list_ack_task_verdicts(conn, tid) + subs = ack.list_ack_subscriptions(conn, tid) + notify_subs = kb.list_notify_subs(conn, tid) + + assert notify_subs == [], "prose body must not create a real subscription" + assert len(verdicts) == 1 + assert verdicts[0]["verdict"] == "GO" + assert subs == [], "prose body must not be shadow-copied as subscription" + + +# --------------------------------------------------------------------------- +# Helper sanity checks +# --------------------------------------------------------------------------- + + +def test_safe_summary_collapses_and_caps_text(): + assert ack._safe_summary(None) is None + assert ack._safe_summary(" \n ") is None + assert ack._safe_summary("line1\nline2") == "line1 line2" + long_text = "x" * 500 + assert ack._safe_summary(long_text) == "x" * 400 + + +def test_correlation_id_is_stable_and_non_secret(): + c1 = ack._ack_correlation_id("passive", "t_a", "discord", "c1", "th1", 12345) + c2 = ack._ack_correlation_id("passive", "t_a", "discord", "c1", "th1", 12345) + c3 = ack._ack_correlation_id("passive", "t_b", "discord", "c1", "th1", 12345) + assert c1 == c2 + assert c1 != c3 + assert c1.startswith("ack_") + assert "t_a" not in c1 diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index 2762e220e79a..926016c79d56 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -3147,6 +3147,12 @@ def test_legacy_db_without_skills_column_migrates(tmp_path): after = {r[1] for r in conn.execute("PRAGMA table_info(tasks)")} assert "skills" in after, f"migration did not add skills column: {after}" + # The legacy DB deliberately has no notify table; adding optional + # trigger_agent must be skipped rather than crashing on a missing table. + assert conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='kanban_notify_subs'" + ).fetchone() is None + # Idempotent: running again must not raise. kb._migrate_add_optional_columns(conn) diff --git a/tests/hermes_cli/test_kanban_diagnostics.py b/tests/hermes_cli/test_kanban_diagnostics.py index 2de4933dc634..54e427c06cb8 100644 --- a/tests/hermes_cli/test_kanban_diagnostics.py +++ b/tests/hermes_cli/test_kanban_diagnostics.py @@ -748,3 +748,72 @@ def test_severity_at_or_above_uses_threshold_semantics(): assert kd.severity_at_or_above("error", "critical") is False assert kd.severity_at_or_above("mystery", "warning") is False assert kd.severity_at_or_above("warning", None) is True + + +# --------------------------------------------------------------------------- +# missing_ack_relay — origin ACK/relay failed even though the work verdict +# is terminal. task_verdict and ack_status are tracked separately so a +# done BLOCK/GO task does not silently look fully delivered. +# --------------------------------------------------------------------------- + + +def test_missing_ack_relay_fires_on_failed_event(): + task = _task(status="done") + events = [ + _event("completed", ts=100, summary="Verdict: BLOCK"), + _event("ack_relay_status", ts=101, ack_status="failed", + task_verdict="BLOCK", matched=["no messaging targets"]), + ] + diags = kd.compute_task_diagnostics(task, events, [], now=300) + hits = [d for d in diags if d.kind == "missing_ack_relay"] + assert len(hits) == 1 + d = hits[0] + assert d.severity == "error" + # Verdict is reported but kept distinct from the ack status. + assert d.data["task_verdict"] == "BLOCK" + assert d.data["ack_status"] == "failed" + assert "no messaging targets" in d.data["matched"] + + +def test_missing_ack_relay_ambiguous_is_warning(): + task = _task(status="done") + events = [ + _event("completed", ts=100, summary="Verdict: GO"), + _event("ack_relay_status", ts=101, ack_status="ambiguous", + task_verdict="GO", matched=[]), + ] + diags = kd.compute_task_diagnostics(task, events, [], now=300) + hits = [d for d in diags if d.kind == "missing_ack_relay"] + assert len(hits) == 1 + assert hits[0].severity == "warning" + assert hits[0].data["ack_status"] == "ambiguous" + + +def test_missing_ack_relay_clears_after_clean_recompletion(): + task = _task(status="done") + events = [ + _event("completed", ts=100, summary="Verdict: BLOCK"), + _event("ack_relay_status", ts=101, ack_status="failed", + task_verdict="BLOCK", matched=["origin relay could not be sent"]), + # A later clean completion (e.g. after a reclaim+retry that relayed + # fine) supersedes the stale failure. + _event("completed", ts=200, summary="Verdict: GO"), + ] + diags = kd.compute_task_diagnostics(task, events, [], now=300) + assert [d for d in diags if d.kind == "missing_ack_relay"] == [] + + +def test_missing_ack_relay_does_not_storm_on_repeated_events(): + task = _task(status="done") + events = [ + _event("completed", ts=100, summary="Verdict: BLOCK"), + _event("ack_relay_status", ts=101, ack_status="failed", + task_verdict="BLOCK", matched=["no live gateway runner"]), + _event("ack_relay_status", ts=102, ack_status="failed", + task_verdict="BLOCK", matched=["no live gateway runner"]), + ] + diags = kd.compute_task_diagnostics(task, events, [], now=300) + hits = [d for d in diags if d.kind == "missing_ack_relay"] + # Coalesced into a single diagnostic, not one per event. + assert len(hits) == 1 + assert hits[0].count == 2 diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index f8109416cb5a..a79b7db1764c 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -1,4 +1,5 @@ import asyncio +import json import pytest from pathlib import Path @@ -78,6 +79,242 @@ async def _fast_sleep(_): assert subs == [], "Subscription should be unsub after completed event" + + +def test_notify_sub_trigger_agent_persists_and_preserves_owner(kanban_home): + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="trigger sub task", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + notifier_profile="owner-a", + ) + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + notifier_profile="owner-b", + trigger_agent=True, + ) + subs = kb.list_notify_subs(conn, tid) + finally: + conn.close() + + assert len(subs) == 1 + assert subs[0]["notifier_profile"] == "owner-a" + assert int(subs[0]["trigger_agent"]) == 1 + + +def test_child_task_inherits_parent_notify_subscription(kanban_home): + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + parent = kb.create_task(conn, title="root", assignee=None) + kb.add_notify_sub( + conn, + task_id=parent, + platform="discord", + chat_id="1499390151393284106", + thread_id="123", + user_id="u1", + notifier_profile="default", + trigger_agent=True, + ) + child = kb.create_task( + conn, + title="review child", + assignee="ccreviewer", + parents=[parent], + ) + subs = kb.list_notify_subs(conn, child) + finally: + conn.close() + + assert len(subs) == 1 + assert subs[0]["platform"] == "discord" + assert subs[0]["chat_id"] == "1499390151393284106" + assert subs[0]["thread_id"] == "123" + assert subs[0]["user_id"] == "u1" + assert subs[0]["notifier_profile"] == "default" + assert int(subs[0]["trigger_agent"]) == 1 + + +def test_cli_create_body_origin_return_to_does_not_create_notify_subscription(kanban_home): + import hermes_cli.kanban as kc + import hermes_cli.kanban_db as kb + + body = "Origin/return_to: Discord Devhub #research (<#1499390151393284106>)\nDo the work." + out = kc.run_slash( + "create 'origin prose task' " + f"--body {json.dumps(body)} " + "--assignee ccreviewer --json" + ) + payload = json.loads(out) + task_id = payload["id"] + + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, task_id) + finally: + conn.close() + + assert "ack_subscription" not in payload + assert subs == [] + + +def test_cli_create_explicit_origin_flags_create_notify_subscription(kanban_home): + import hermes_cli.kanban as kc + import hermes_cli.kanban_db as kb + + out = kc.run_slash( + "create 'explicit origin task' " + "--body 'Do the work.' " + "--assignee ccreviewer " + "--origin-platform discord " + "--origin-chat-id 1499390151393284106 " + "--origin-thread-id 123 " + "--notifier-profile gateway-a " + "--ack-trigger-agent " + "--json" + ) + payload = json.loads(out) + task_id = payload["id"] + + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, task_id) + finally: + conn.close() + + assert payload["ack_subscription"] == { + "task_id": task_id, + "platform": "discord", + "chat_id": "1499390151393284106", + "thread_id": "123", + "trigger_agent": True, + "notifier_profile": "gateway-a", + } + assert len(subs) == 1 + assert subs[0]["platform"] == "discord" + assert subs[0]["chat_id"] == "1499390151393284106" + assert subs[0]["thread_id"] == "123" + assert subs[0]["notifier_profile"] == "gateway-a" + assert int(subs[0]["trigger_agent"]) == 1 + + +def test_cli_create_explicit_ack_trigger_agent_upgrades_idempotent_subscription(kanban_home): + import hermes_cli.kanban as kc + import hermes_cli.kanban_db as kb + + cmd = ( + "create 'origin flags task' --assignee ccreviewer " + "--idempotency-key ack-replay-1 " + "--origin-platform discord --origin-chat-id 1499390151393284106 " + "--origin-thread-id 987654321 --origin-user-id u1 " + "--notifier-profile gateway-owner --json" + ) + first = json.loads(kc.run_slash(cmd)) + replay = json.loads(kc.run_slash(cmd + " --ack-trigger-agent")) + assert replay["id"] == first["id"] + + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, first["id"]) + finally: + conn.close() + + assert len(subs) == 1 + assert subs[0]["platform"] == "discord" + assert subs[0]["chat_id"] == "1499390151393284106" + assert subs[0]["thread_id"] == "987654321" + assert subs[0]["user_id"] == "u1" + assert subs[0]["notifier_profile"] == "gateway-owner" + assert int(subs[0]["trigger_agent"]) == 1 + + +@pytest.mark.parametrize( + "body", + [ + "Origin/return_to: Discord Devhub #research (<#1499390151393284106>)\nDo the work.", + "return-to: discord:#hermes-main:1497895797579190357\nDo the work.", + ], +) +def test_completion_origin_intent_without_subscription_emits_missing_subscription(kanban_home, body): + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="origin missing sub", body=body, assignee="worker1") + assert kb.list_notify_subs(conn, tid) == [] + assert kb.complete_task(conn, tid, summary="Verdict: GO\nEvidence: done") is True + events = conn.execute( + "SELECT kind, payload FROM task_events WHERE task_id = ? ORDER BY id", + (tid,), + ).fetchall() + finally: + conn.close() + + delivery = [e for e in events if e["kind"] == "delivery_problem"] + assert len(delivery) == 1 + payload = json.loads(delivery[0]["payload"]) + assert payload["problem_type"] == "missing_subscription" + assert payload["ack_status"] == "missing_subscription" + assert payload["task_verdict"] == "GO" + + +def test_completion_generic_origin_prose_without_subscription_is_not_missing_subscription(kanban_home): + import hermes_cli.kanban_db as kb + + body = "Research the origin of this bug. No return target here." + conn = kb.connect() + try: + tid = kb.create_task(conn, title="ordinary origin prose", body=body, assignee="worker1") + assert kb.list_notify_subs(conn, tid) == [] + assert kb.complete_task(conn, tid, summary="Verdict: GO\nEvidence: done") is True + events = conn.execute( + "SELECT kind, payload FROM task_events WHERE task_id = ? ORDER BY id", + (tid,), + ).fetchall() + finally: + conn.close() + + assert [e for e in events if e["kind"] == "delivery_problem"] == [] + relay = [e for e in events if e["kind"] == "ack_relay_status"] + assert len(relay) == 1 + payload = json.loads(relay[0]["payload"]) + assert payload["ack_status"] == "ambiguous" + assert payload.get("problem_type") != "missing_subscription" + + +def test_completion_structured_origin_fields_without_subscription_emit_missing_subscription(kanban_home): + import hermes_cli.kanban_db as kb + + body = "origin_platform: discord\norigin_chat_id: 1499390151393284106\nDo the work." + conn = kb.connect() + try: + tid = kb.create_task(conn, title="structured origin missing sub", body=body, assignee="worker1") + assert kb.list_notify_subs(conn, tid) == [] + assert kb.complete_task(conn, tid, summary="Verdict: GO\nEvidence: done") is True + events = conn.execute( + "SELECT kind, payload FROM task_events WHERE task_id = ? ORDER BY id", + (tid,), + ).fetchall() + finally: + conn.close() + + delivery = [e for e in events if e["kind"] == "delivery_problem"] + assert len(delivery) == 1 + payload = json.loads(delivery[0]["payload"]) + assert payload["problem_type"] == "missing_subscription" + assert payload["ack_status"] == "missing_subscription" + @pytest.mark.asyncio @pytest.mark.parametrize('kind', ["gave_up", "crashed", "timed_out"]) async def test_notifier_unsubs_after_abnormal_events(kind, kanban_home): @@ -477,6 +714,8 @@ async def test_gateway_create_autosubscribes_on_explicit_board(kanban_home): assert len(subs) == 1 assert subs[0]["chat_id"] == "chat1" assert subs[0]["thread_id"] == "th1" + assert subs[0]["user_id"] == "u1" + assert int(subs[0]["trigger_agent"]) == 0 conn = kb.connect(board="default") try: @@ -653,3 +892,471 @@ async def _fast_sleep(_): # Only the real file was uploaded. assert len(documents_uploaded) == 1 assert "real.pdf" in documents_uploaded[0] + + +# --------------------------------------------------------------------------- +# ACK/relay status classification + durable event on terminal completion. +# task_verdict (the work decision) is tracked separately from ack_status +# (whether the origin wake/ACK relay actually reached a target). +# --------------------------------------------------------------------------- + + +def test_classify_ack_relay_detects_no_messaging_targets(): + res = kb.classify_ack_relay( + summary="Verdict: BLOCK\nReview failed. no messaging targets", + ) + assert res["task_verdict"] == "BLOCK" + assert res["ack_status"] == "failed" + assert res["relay_failure"] is True + assert "no messaging targets" in res["matched"] + + +def test_classify_ack_relay_parses_go_verdict_without_relay_failure(): + res = kb.classify_ack_relay(summary="Verdict: GO\nAll good, merged.") + assert res["task_verdict"] == "GO" + # No relay-failure string and no notify context -> not a hard failure. + assert res["relay_failure"] is False + assert res["ack_status"] == "unknown" + + +def test_classify_ack_relay_empty_notify_after_terminal_is_ambiguous(): + # A done task carrying a verdict but with zero notify subscriptions is + # ambiguous: the subs may have been delivered+removed, or there may + # never have been an origin target. Not a confirmed failure. + res = kb.classify_ack_relay(summary="Verdict: GO", notify_subs=[]) + assert res["task_verdict"] == "GO" + assert res["relay_failure"] is False + assert res["ack_status"] == "ambiguous" + + +def test_classify_ack_relay_scans_result_and_metadata(): + res = kb.classify_ack_relay( + result="done", + metadata={"relay": "trigger_error no live gateway runner"}, + ) + assert res["relay_failure"] is True + assert res["ack_status"] == "failed" + + +def test_complete_task_emits_ack_relay_status_event_on_relay_failure(kanban_home): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="review task", assignee="reviewer") + kb.complete_task( + conn, tid, + summary="Verdict: BLOCK\norigin relay could not be sent", + ) + events = kb.list_events(conn, tid) + finally: + conn.close() + ack = [e for e in events if e.kind == "ack_relay_status"] + assert len(ack) == 1, "exactly one durable ack_relay_status event" + payload = ack[0].payload + assert payload["ack_status"] == "failed" + assert payload["task_verdict"] == "BLOCK" + assert "origin relay could not be sent" in payload["matched"] + + +def test_complete_task_clean_summary_emits_no_ack_relay_event(kanban_home): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="leaf task", assignee="worker") + # Has a live notify sub, so the empty-notify ambiguity does not apply. + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + kb.complete_task(conn, tid, summary="Verdict: GO\nshipped.") + events = kb.list_events(conn, tid) + finally: + conn.close() + assert [e for e in events if e.kind == "ack_relay_status"] == [] + + +def test_complete_task_ack_relay_event_emitted_once_no_storm(kanban_home): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="fanin task", assignee="reviewer") + kb.complete_task(conn, tid, summary="Verdict: BLOCK no messaging targets") + # A second completion call on an already-done task is a no-op and + # must not append another ack_relay_status event. + kb.complete_task(conn, tid, summary="Verdict: BLOCK no messaging targets") + events = kb.list_events(conn, tid) + finally: + conn.close() + assert len([e for e in events if e.kind == "ack_relay_status"]) == 1 + +# --------------------------------------------------------------------------- +# Active-wake notifier receipts +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_notifier_trigger_agent_false_sends_passive_only(kanban_home): + """If subscription trigger_agent is false, only passive adapter.send runs.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="passive task", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + trigger_agent=False, + ) + kb.complete_task(conn, tid, result="done") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + fake_adapter.handle_message.assert_not_called() + + +@pytest.mark.asyncio +async def test_notifier_trigger_agent_true_invokes_active_wake_and_records_receipt(kanban_home): + """Active-wake subscriptions call adapter.handle_message and persist a receipt event.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + from gateway.platforms.base import MessageEvent + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="wake task", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + trigger_agent=True, + ) + kb.complete_task(conn, tid, result="done") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner._gateway_loop = asyncio.get_running_loop() + + wake_events: list[MessageEvent] = [] + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + async def _capture_handle_message(event): + wake_events.append(event) + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + fake_adapter.handle_message = AsyncMock(side_effect=_capture_handle_message) + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + fake_adapter.handle_message.assert_awaited_once() + assert len(wake_events) == 1 + wake = wake_events[0] + assert wake.internal is True + assert wake.source.platform == Platform.TELEGRAM + assert wake.source.chat_id == "chat1" + assert wake.source.user_id is None + assert wake.source.user_name == "Hermes Active Wake" + + conn = kb.connect() + try: + events = kb.list_events(conn, tid) + from hermes_cli import kanban_db_ack_ledger as ack + wake_rows = ack.list_ack_active_wakes(conn, tid) + operator_rows = ack.list_ack_operator_receipts(conn, tid) + finally: + conn.close() + receipt_events = [e for e in events if e.kind == "notify_active_wake_receipt"] + assert len(receipt_events) == 1 + payload = receipt_events[0].payload + assert payload is not None + assert payload["scheduled_agent"] is True + assert payload["triggered_agent"] is True # legacy scheduling alias only + assert payload["accepted_by_session"] is False + assert payload["started_by_session"] is False + assert payload["active_wake_status"] == "scheduled" + assert payload["target_session_key"] == "agent:main:telegram:group:chat1" + assert "hermes-active-wake" not in payload["target_session_key"] + assert "operator_receipt" not in payload + assert payload["platform"] == "telegram" + assert payload["chat_id"] == "chat1" + assert "receipt_correlation" in payload + assert len(wake_rows) == 1 + assert wake_rows[0]["status"] == "scheduled" + assert wake_rows[0]["accepted_by_session"] == 0 + assert wake_rows[0]["started_by_session"] == 0 + assert wake_rows[0]["target_session_key"] == "agent:main:telegram:group:chat1" + assert operator_rows == [] + + +@pytest.mark.asyncio +async def test_notifier_active_wake_not_wired_records_failure(kanban_home): + """If gateway runner lacks a usable adapter/loop, active wake records NOT_WIRED.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="wake not wired", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + trigger_agent=True, + ) + kb.complete_task(conn, tid, result="done") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + # Passive delivery is wired, but no _gateway_loop exists for active wake. + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + conn = kb.connect() + try: + events = kb.list_events(conn, tid) + finally: + conn.close() + receipt_events = [e for e in events if e.kind == "notify_active_wake_receipt"] + assert len(receipt_events) == 1 + payload = receipt_events[0].payload + assert payload is not None + assert payload["scheduled_agent"] is False + assert payload["triggered_agent"] is False + assert payload["trigger_error"] == "NOT_WIRED" + + +@pytest.mark.asyncio +async def test_notifier_send_result_failure_records_sanitized_receipt_no_wake(kanban_home): + """If passive send returns success=false, persist SEND_FAILED without waking or leaking raw error.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + leaked = "super-secret-token-123456" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="send result failure", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + trigger_agent=True, + ) + kb.complete_task(conn, tid, result="done") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner._gateway_loop = asyncio.get_running_loop() + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + return {"success": False, "error": f"https://example.test/?access_token={leaked}"} + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + fake_adapter.handle_message = AsyncMock() + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + fake_adapter.handle_message.assert_not_called() + + conn = kb.connect() + try: + events = kb.list_events(conn, tid) + finally: + conn.close() + receipt_events = [e for e in events if e.kind == "notify_active_wake_receipt"] + assert len(receipt_events) == 1 + payload = receipt_events[0].payload + assert payload is not None + assert payload["success"] is False + assert payload["scheduled_agent"] is False + assert payload["triggered_agent"] is False + assert payload["trigger_error"] == "SEND_FAILED" + assert "receipt_correlation" in payload + assert leaked not in json.dumps(payload) + assert "access_token" not in json.dumps(payload) + + +@pytest.mark.asyncio +async def test_notifier_send_failure_does_not_active_wake_or_record_receipt(kanban_home): + """If passive adapter.send fails, active wake is not scheduled and no success receipt is persisted.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="send failure", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + trigger_agent=True, + ) + kb.complete_task(conn, tid, result="done") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + runner._gateway_loop = asyncio.get_running_loop() + + fake_adapter = MagicMock() + + async def _raise_send(chat_id, msg, metadata=None): + runner._running = False + raise RuntimeError("send failed: token=super-secret-token-123456") + + fake_adapter.send = AsyncMock(side_effect=_raise_send) + fake_adapter.handle_message = AsyncMock() + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + fake_adapter.handle_message.assert_not_called() + + conn = kb.connect() + try: + events = kb.list_events(conn, tid) + finally: + conn.close() + assert [e for e in events if e.kind == "notify_active_wake_receipt"] == [] + + +@pytest.mark.asyncio +async def test_notifier_trigger_agent_false_does_not_mirror_passive_send(kanban_home): + """Passive notifier sends must not be mirrored into target sessions (no duplicate).""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="no mirror task", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat1", + trigger_agent=False, + ) + kb.complete_task(conn, tid, result="done") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._running = True + runner._kanban_sub_fail_counts = {} + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \ + patch("gateway.kanban_watchers.mirror_to_session", create=True) as mock_mirror: + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_called_once() + mock_mirror.assert_not_called() diff --git a/tests/tools/test_send_message_tool.py b/tests/tools/test_send_message_tool.py index dcdb8f83266b..b40eb8603278 100644 --- a/tests/tools/test_send_message_tool.py +++ b/tests/tools/test_send_message_tool.py @@ -4,6 +4,7 @@ import json import os import sys +from concurrent.futures import Future from types import ModuleType, SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -3097,3 +3098,400 @@ async def run_test(): finally: if media_path and os.path.exists(media_path): os.unlink(media_path) + + +# --------------------------------------------------------------------------- +# Active-wake receipt primitive +# --------------------------------------------------------------------------- + + +class TestActiveWakeReceipt: + """Regression tests for the gateway-owned active-wake receipt envelope. + + The receipt exposes classification fields for AgentFlow delivery_attempt + semantics without persisting raw transcripts, paths, or secrets. AgentFlow + remains the policy owner: Hermes only reports whether a wake was attempted. + """ + + def _make_config(self): + telegram_cfg = SimpleNamespace(enabled=True, token="***", extra={}) + return SimpleNamespace( + platforms={Platform.TELEGRAM: telegram_cfg}, + get_home_channel=lambda _platform: None, + ), telegram_cfg + + def _completed_schedule(self, handled): + def _schedule(coro, loop, **_kwargs): + handled.append((coro, loop)) + asyncio.run(coro) + fut = Future() + fut.set_result(None) + return fut + return _schedule + + def _active_wake_event_text_for(self, message): + config, _telegram_cfg = self._make_config() + handled = [] + scheduled = [] + loop = object() + + class FakeAdapter: + async def handle_message(self, event): + handled.append(event) + + runner = SimpleNamespace(adapters={Platform.TELEGRAM: FakeAdapter()}, _gateway_loop=loop) + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True, "message_id": "msg-safe"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: runner), \ + patch("agent.async_utils.safe_schedule_threadsafe", side_effect=self._completed_schedule(scheduled)), \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": message, + "trigger_agent": True, + } + ) + ) + + assert result["success"] is True + assert result["scheduled_agent"] is True + assert result["triggered_agent"] is True # legacy scheduling alias + assert "accepted_by_session" not in result + assert "operator_receipt" not in result + assert "_acceptance" not in result + mirror_mock.assert_not_called() + assert len(handled) == 1 + return handled[0].text + + def test_sent_and_triggered(self): + """Visible send succeeds + live adapter present => scheduled wake only.""" + config, telegram_cfg = self._make_config() + handled = [] + scheduled = [] + loop = object() + + class FakeAdapter: + async def handle_message(self, event): + handled.append(event) + + runner = SimpleNamespace(adapters={Platform.TELEGRAM: FakeAdapter()}, _gateway_loop=loop) + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True, "message_id": "msg-123"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: runner), \ + patch("agent.async_utils.safe_schedule_threadsafe", side_effect=self._completed_schedule(scheduled)) as schedule_mock, \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": "wake up", + "trigger_agent": True, + "correlation_id": "corr-abc", + } + ) + ) + + assert result["success"] is True + assert result["message_id"] == "msg-123" + assert result["scheduled_agent"] is True + assert result["triggered_agent"] is True # legacy scheduling alias only + assert "accepted_by_session" not in result + assert "operator_receipt" not in result + assert "_acceptance" not in result + assert result["receipt_correlation"] == "corr-abc" + assert "trigger_error" not in result + assert "mirrored" not in result + mirror_mock.assert_not_called() + schedule_mock.assert_called_once() + assert scheduled[0][1] is loop + assert len(handled) == 1 + event = handled[0] + assert event.text == "wake up" + assert event.source.chat_id == "12345" + assert event.source.platform == Platform.TELEGRAM + assert event.internal is True + assert event.source.user_id is None + assert event.source.user_name == "Hermes Active Wake" + + def test_active_wake_targets_channel_session_key_not_synthetic_user(self): + """Internal active wake targets the operator/channel key, not a hermes-active-wake ghost.""" + from tools.send_message_tool import _trigger_adapter_active_wake + + handled = [] + scheduled = [] + loop = object() + + class FakeAdapter: + async def handle_message(self, event): + handled.append(event) + + def _session_key_for_source(source): + assert source.user_id is None + return f"agent:main:{source.platform.value}:group:{source.chat_id}" + + with patch("agent.async_utils.safe_schedule_threadsafe", side_effect=self._completed_schedule(scheduled)): + result = _trigger_adapter_active_wake( + platform=Platform.DISCORD, + adapter=FakeAdapter(), + loop=loop, + platform_name="discord", + chat_id="1497895797579190357", + thread_id=None, + message="wake up", + runner=SimpleNamespace(_session_key_for_source=_session_key_for_source), + ) + + assert result["scheduled_agent"] is True + assert result["triggered_agent"] is True + assert result["target_session_key"] == "agent:main:discord:group:1497895797579190357" + assert "hermes-active-wake" not in result["target_session_key"] + assert result["_acceptance"]["target_session_key"] == result["target_session_key"] + assert handled[0].internal is True + assert handled[0].source.user_id is None + + def test_synthetic_wake_event_uses_sanitized_text(self): + """Active wake must never inject raw MEDIA/path directives into event text.""" + config, _telegram_cfg = self._make_config() + handled = [] + scheduled = [] + loop = object() + + class FakeAdapter: + async def handle_message(self, event): + handled.append(event) + + runner = SimpleNamespace(adapters={Platform.TELEGRAM: FakeAdapter()}, _gateway_loop=loop) + raw_message = "hello\nMEDIA:/tmp/hermes-active-wake-secret-sk_live_123456.pdf" + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True, "message_id": "msg-safe"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: runner), \ + patch("agent.async_utils.safe_schedule_threadsafe", side_effect=self._completed_schedule(scheduled)), \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": raw_message, + "trigger_agent": True, + } + ) + ) + + assert result["success"] is True + assert result["scheduled_agent"] is True + assert result["triggered_agent"] is True + mirror_mock.assert_not_called() + assert len(handled) == 1 + event_text = handled[0].text + assert event_text == "hello" + assert "MEDIA:" not in event_text + assert "/tmp/hermes-active-wake-secret-sk_live_123456.pdf" not in event_text + assert "secret" not in event_text.lower() + assert "sk_live" not in event_text + + @pytest.mark.parametrize( + "raw_message, forbidden", + [ + ( + "inline `MEDIA:/tmp/hermes-inline-sk_live_inline123.pdf` keep", + ["MEDIA:", "/tmp/hermes-inline-sk_live_inline123.pdf", "sk_live_inline123"], + ), + ( + "```json\n{\"artifact\": \"MEDIA:/tmp/hermes-fenced-sk_live_fenced123.pdf\"}\n```", + ["MEDIA:", "/tmp/hermes-fenced-sk_live_fenced123.pdf", "sk_live_fenced123"], + ), + ( + "> quoted MEDIA:/tmp/hermes-quoted-sk_live_quoted123.pdf", + ["MEDIA:", "/tmp/hermes-quoted-sk_live_quoted123.pdf", "sk_live_quoted123"], + ), + ( + '{"file": "/tmp/hermes-json-secret.pdf", "token": "sk_live_json123456"}', + ["/tmp/hermes-json-secret.pdf", "sk_live_json123456"], + ), + ( + "bare path /home/duckran/.hermes/audio_cache/voice-secret.ogg should redact", + ["/home/duckran/.hermes/audio_cache/voice-secret.ogg"], + ), + ( + "token fragments sk_live_token123456 and ghp_abcdefghijklmnopqrstuvwxyz1234", + ["sk_live_token123456", "ghp_abcdefghijklmnopqrstuvwxyz1234"], + ), + ( + "redacted-looking token fragments sk_liv...3456 and ghp_ab...1234", + ["sk_liv...3456", "ghp_ab...1234"], + ), + ( + "redacted-looking Slack fragments xoxb-1...cdef and xoxb-1234...abcdef", + ["xoxb-1...cdef", "xoxb-1234...abcdef"], + ), + ], + ) + def test_synthetic_wake_event_strictly_sanitizes_protected_spans(self, raw_message, forbidden): + """Active wake redacts raw artifacts even in protected Markdown/JSON spans.""" + event_text = self._active_wake_event_text_for(raw_message) + + for raw in forbidden: + assert raw not in event_text + assert "MEDIA:" not in event_text + assert "sk_live" not in event_text + + def test_sent_and_not_wired(self): + """Visible send succeeds but no gateway runner => NOT_WIRED.""" + config, _telegram_cfg = self._make_config() + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True, "message_id": "msg-456"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: None), \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": "wake up", + "trigger_agent": True, + } + ) + ) + + assert result["success"] is True + assert result["message_id"] == "msg-456" + assert result["scheduled_agent"] is False + assert result["triggered_agent"] is False + assert result["trigger_error"] == "NOT_WIRED" + assert result["receipt_correlation"].startswith("wake_") + mirror_mock.assert_not_called() + + def test_send_failed(self): + """Visible send fails: no active wake is scheduled.""" + config, _telegram_cfg = self._make_config() + loop = object() + + class FakeAdapter: + async def handle_message(self, _event): + raise AssertionError("send_failed must not wake agent") + + runner = SimpleNamespace(adapters={Platform.TELEGRAM: FakeAdapter()}, _gateway_loop=loop) + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"error": "Telegram send failed: bad chat"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: runner), \ + patch("agent.async_utils.safe_schedule_threadsafe") as schedule_mock, \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": "wake up", + "trigger_agent": True, + } + ) + ) + + assert result["success"] is False + assert "error" in result + assert result["scheduled_agent"] is False + assert result["triggered_agent"] is False + assert result["trigger_error"] == "SEND_FAILED" + assert result["receipt_correlation"].startswith("wake_") + schedule_mock.assert_not_called() + mirror_mock.assert_not_called() + + def test_visible_send_alone_does_not_close_origin_return(self): + """Without trigger_agent, the result must not contain triggered_agent. + + This is the classification contract: a visible send alone does not mark + an active wake as scheduled, so AgentFlow should not close origin_return + unless it sees scheduled_agent=true (or legacy triggered_agent=true). + """ + config, _telegram_cfg = self._make_config() + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True, "message_id": "msg-789"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: None), \ + patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": "just a note", + } + ) + ) + + assert result["success"] is True + assert result["message_id"] == "msg-789" + assert result["mirrored"] is True + assert "triggered_agent" not in result + assert "trigger_error" not in result + assert "receipt_correlation" not in result + mirror_mock.assert_called_once() + + def test_trigger_error_redacted_on_secrets(self): + """Scheduling/setup errors in trigger_error must be redacted.""" + config, _telegram_cfg = self._make_config() + leaked = "super-secret-token-123456" + loop = object() + + class BadAdapter: + async def handle_message(self, _event): + return None + + runner = SimpleNamespace(adapters={Platform.TELEGRAM: BadAdapter()}, _gateway_loop=loop) + + def _bad_schedule(coro, _loop, **_kwargs): + coro.close() + raise RuntimeError(f"boom: https://api.example.com/send?access_token={leaked}") + + with patch("gateway.config.load_gateway_config", return_value=config), \ + patch("tools.interrupt.is_interrupted", return_value=False), \ + patch("model_tools._run_async", side_effect=_run_async_immediately), \ + patch("tools.send_message_tool._send_to_platform", + new=AsyncMock(return_value={"success": True, "message_id": "msg-000"})), \ + patch("gateway.run._gateway_runner_ref", new=lambda: runner), \ + patch("agent.async_utils.safe_schedule_threadsafe", side_effect=_bad_schedule), \ + patch("gateway.mirror.mirror_to_session", return_value=True): + result = json.loads( + send_message_tool( + { + "action": "send", + "target": "telegram:12345", + "message": "wake up", + "trigger_agent": True, + } + ) + ) + + assert result["success"] is True + assert result["scheduled_agent"] is False + assert result["triggered_agent"] is False + assert leaked not in result["trigger_error"] + assert "access_token=***" in result["trigger_error"] diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index b654d8ff2ecf..6e739ddf80b9 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -6,6 +6,7 @@ """ import asyncio +import hashlib import json import logging import os @@ -73,6 +74,20 @@ r"\b(access_token|api[_-]?key|auth[_-]?token|signature|sig)\s*=\s*([^\s,;]+)", re.IGNORECASE, ) +_ACTIVE_WAKE_MEDIA_DIRECTIVE_RE = re.compile(r"(?i)MEDIA:[^\s`'\"<>),\]}]+") +_ACTIVE_WAKE_LOCAL_PATH_RE = re.compile( + r"(?),\]}]+" +) +_ACTIVE_WAKE_TOKEN_FRAGMENT_RE = re.compile( + r"\b(?:sk|pk)_(?:live|test)_[A-Za-z0-9][A-Za-z0-9_-]{6,}\b" + r"|\b(?:sk|pk)_(?:liv|tes)\.\.\.[A-Za-z0-9_-]{2,}\b" + r"|\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9_]{20,}\b" + r"|\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9_]{2,}\.\.\.[A-Za-z0-9_]{2,}\b" + r"|\bxox[baprs]-[A-Za-z0-9-]{10,}\b" + r"|\bxox[baprs]-[A-Za-z0-9-]{1,}\.\.\.[A-Za-z0-9-]{2,}\b", + re.IGNORECASE, +) def _sanitize_error_text(text) -> str: @@ -83,6 +98,28 @@ def _sanitize_error_text(text) -> str: return redacted +def _sanitize_active_wake_text(text: str) -> str: + """Strictly sanitize text before injecting it as an internal wake event. + + ``BasePlatformAdapter.extract_media`` intentionally preserves protected + spans for visible sends and passive mirrors. Active-wake is different: the + sanitized text becomes synthetic inbound model context, so raw attachment + directives, local artifact paths, and token-looking fragments must not + survive even inside inline code, fenced blocks, blockquotes, or JSON. + """ + redacted = redact_sensitive_text(text or "") + redacted = _URL_SECRET_QUERY_RE.sub(lambda m: f"{m.group(1)}***", redacted) + redacted = _GENERIC_SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", redacted) + redacted = _ACTIVE_WAKE_MEDIA_DIRECTIVE_RE.sub("", redacted) + redacted = _ACTIVE_WAKE_LOCAL_PATH_RE.sub("[redacted-path]", redacted) + redacted = _ACTIVE_WAKE_TOKEN_FRAGMENT_RE.sub("[redacted-token]", redacted) + + # Drop lines that only contained a stripped MEDIA directive. This preserves + # user-visible prose ("hello") while avoiding synthetic wake clutter. + cleaned_lines = [line.rstrip() for line in redacted.splitlines() if line.strip()] + return "\n".join(cleaned_lines).strip() + + def _error(message: str) -> dict: """Build a standardized error payload with redacted content.""" return {"error": _sanitize_error_text(message)} @@ -95,6 +132,18 @@ def _display_chat_id(platform_name: str, chat_id: str) -> str: return chat_id +def _stable_correlation_id(platform: str, chat_id: str, thread_id: str | None, message: str) -> str: + """Return a deterministic, non-secret correlation ID for a send attempt. + + The ID is stable for the same (platform, chat, thread, message) tuple so + retries and idempotent callers observe the same receipt correlation without + needing to generate UUIDs or persist state. The message content is hashed + so raw text never leaks into the correlation value. + """ + key = f"{platform}:{chat_id}:{thread_id or ''}:{message}" + return f"wake_{hashlib.sha256(key.encode('utf-8')).hexdigest()[:24]}" + + def _telegram_retry_delay(exc: Exception, attempt: int) -> float | None: retry_after = getattr(exc, "retry_after", None) if retry_after is not None: @@ -171,6 +220,14 @@ async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs) "message_id": { "type": "string", "description": "For action='react'/'unreact': id of the message to react to. Omit to target the most recent message received in that chat (usually the one being replied to)." + }, + "trigger_agent": { + "type": "boolean", + "description": "When true and the gateway has a live adapter for the target platform, request an active-wake turn on the target session after the visible send. The caller (e.g. AgentFlow) remains the policy owner; Hermes only exposes the attempt result in the receipt fields." + }, + "correlation_id": { + "type": "string", + "description": "Optional caller-owned correlation ID for the active-wake receipt. When omitted, a stable deterministic correlation ID is generated from the target and message content." } }, "required": [] @@ -299,6 +356,8 @@ def _handle_send(args): """Send a message to a platform target.""" target = args.get("target", "") message = args.get("message", "") + trigger_agent = bool(args.get("trigger_agent", False)) + correlation_id = (args.get("correlation_id") or "").strip() or None if not target or not message: return tool_error("Both 'target' and 'message' are required when action='send'") @@ -446,8 +505,16 @@ async def _open_slack_dm(token, user_id): if used_home_channel and isinstance(result, dict) and result.get("success"): result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})" - # Mirror the sent message into the target's gateway session - if isinstance(result, dict) and result.get("success") and mirror_text: + # Mirror passive sends into the target's gateway session. Active-wake + # sends are represented by the synthetic inbound event below; mirroring + # them too would create a duplicate transcript entry and could let a + # visible send alone masquerade as an origin-return close signal. + if ( + isinstance(result, dict) + and result.get("success") + and mirror_text + and not trigger_agent + ): try: from gateway.mirror import mirror_to_session from gateway.session_context import get_session_env @@ -467,11 +534,234 @@ async def _open_slack_dm(token, user_id): if isinstance(result, dict) and "error" in result: result["error"] = _sanitize_error_text(result["error"]) + + # Active-wake receipt envelope: gateway-owned primitive, policy owner + # remains the caller (AgentFlow). Do not persist raw transcripts or + # secrets here; only expose deterministic classification fields. + if trigger_agent: + active_wake_text = _sanitize_active_wake_text(mirror_text) + result = _attach_active_wake_receipt( + result, + platform_name=platform_name, + chat_id=chat_id, + thread_id=thread_id, + message=active_wake_text, + correlation_id=correlation_id, + ) + elif correlation_id: + # correlation_id is meaningless without active-wake tracking; still + # echo it back when provided so callers don't lose their trace ID. + result["receipt_correlation"] = correlation_id + return json.dumps(result) except Exception as e: return json.dumps(_error(f"Send failed: {e}")) +def _attach_active_wake_receipt( + send_result: dict, + *, + platform_name: str, + chat_id: str, + thread_id: str | None, + message: str, + correlation_id: str | None, +) -> dict: + """Attach an active-wake receipt to a visible send result. + + This is a thin gateway-owned primitive. It does NOT decide when active wake + is appropriate; the caller owns that policy. Hermes only reports whether a + wake was accepted by the live gateway and what the visible send outcome was, + using fields that AgentFlow can use for delivery_attempt classification: + + - success/error visible platform send status + - scheduled_agent true iff a wake turn was scheduled on the gateway loop + - triggered_agent legacy alias for scheduled_agent; not proof of + operator-session acceptance (deprecated) + - trigger_error SEND_FAILED, NOT_WIRED, or sanitized failure + - message_id platform message id when the send returned one + - receipt_correlation caller correlation or a deterministic stable id + """ + receipt = dict(send_result) + effective_correlation = correlation_id or _stable_correlation_id( + platform_name, chat_id, thread_id, message + ) + receipt["receipt_correlation"] = effective_correlation + + visible_success = bool(receipt.get("success")) + visible_error = receipt.get("error") + receipt["success"] = visible_success + if visible_error: + receipt["error"] = _sanitize_error_text(visible_error) + + if not visible_success: + # A failed visible send must never active-wake/close an origin_return. + receipt["scheduled_agent"] = False + receipt["triggered_agent"] = False + receipt["trigger_error"] = "SEND_FAILED" + return receipt + + trigger_result = _trigger_gateway_agent( + platform_name=platform_name, + chat_id=chat_id, + thread_id=thread_id, + message=message, + ) + receipt.update({k: v for k, v in trigger_result.items() if not k.startswith("_")}) + return receipt + + +def _trigger_gateway_agent( + *, + platform_name: str, + chat_id: str, + thread_id: str | None, + message: str, +) -> dict: + """Request an internal gateway wake for a sent cross-channel message. + + The tool may run in a worker thread while the gateway adapter belongs to the + main gateway asyncio loop. Schedule onto ``runner._gateway_loop`` instead of + running the adapter coroutine on an unrelated helper loop. + """ + try: + from gateway.run import _gateway_runner_ref + runner = _gateway_runner_ref() + except Exception: + runner = None + + try: + from gateway.config import Platform + platform = Platform(platform_name) + except Exception: + platform = None + + adapter = None + if runner is not None and platform is not None: + try: + adapter = runner.adapters.get(platform) + except Exception: + adapter = None + + loop = getattr(runner, "_gateway_loop", None) if runner is not None else None + if platform is None: + return { + "scheduled_agent": False, + "triggered_agent": False, + "trigger_error": "NOT_WIRED", + } + return _trigger_adapter_active_wake( + platform=platform, + adapter=adapter, + loop=loop, + platform_name=platform_name, + chat_id=chat_id, + thread_id=thread_id, + message=message, + runner=runner, + ) + + +def _trigger_adapter_active_wake( + *, + platform, + adapter, + loop, + platform_name: str, + chat_id: str, + thread_id: str | None, + message: str, + user_id: str | None = None, + user_name: str = "Hermes Active Wake", + runner=None, +) -> dict: + """Schedule one sanitized internal active-wake MessageEvent on an adapter. + + P1 receipt contract: the synthetic event targets the channel/thread session + directly instead of appending the old ``hermes-active-wake`` participant id, + so group-per-user isolation cannot create a ghost session. Scheduling is + still separate from acceptance; the running gateway mutates the attached + acceptance dict once it has resolved/claimed the target session. + """ + handle_message = getattr(adapter, "handle_message", None) if adapter is not None else None + if adapter is None or loop is None or not callable(handle_message): + return { + "scheduled_agent": False, + "triggered_agent": False, + "trigger_error": "NOT_WIRED", + } + + try: + from gateway.session import SessionSource + from gateway.platforms.base import MessageEvent, MessageType + + source = SessionSource( + platform=platform, + chat_id=str(chat_id), + chat_type="group", + thread_id=str(thread_id) if thread_id else None, + # Intentionally do not impersonate a real user, and do not use the + # legacy hermes-active-wake synthetic user id. With no participant + # id, build_session_key routes to the real channel/thread operator + # session instead of ``::hermes-active-wake``. + user_id=user_id, + user_name=user_name, + is_bot=False, + message_id=None, + ) + target_session_key = None + if runner is not None: + try: + target_session_key = runner._session_key_for_source(source) + except Exception: + target_session_key = None + acceptance: dict[str, object] = { + "accepted_by_session": False, + "started_by_session": False, + "active_wake_status": "scheduled", + } + if target_session_key: + acceptance["target_session_key"] = target_session_key + wake_event = MessageEvent( + text=message, + message_type=MessageType.TEXT, + source=source, + internal=True, + ) + setattr(wake_event, "_hermes_active_wake_acceptance", acceptance) + if target_session_key: + setattr(wake_event, "_hermes_active_wake_target_session_key", target_session_key) + from agent.async_utils import safe_schedule_threadsafe + future = safe_schedule_threadsafe( + handle_message(wake_event), + loop, + logger=logger, + log_message="active_wake request scheduling failed", + ) + if future is None: + return { + "scheduled_agent": False, + "triggered_agent": False, + "trigger_error": "NOT_WIRED", + } + # ``triggered_agent`` is retained for older callers but is scheduling + # semantics only. Do not treat it as operator-session acceptance. + return { + "scheduled_agent": True, + "triggered_agent": True, + "active_wake_status": "scheduled", + "target_session_key": target_session_key, + "_acceptance": acceptance, + } + except Exception as exc: + logger.debug("active_wake request failed for %s:%s: %s", platform_name, chat_id, exc) + return { + "scheduled_agent": False, + "triggered_agent": False, + "trigger_error": _sanitize_error_text(str(exc)) or "ACTIVE_WAKE_FAILED", + } + + def _parse_target_ref(platform_name: str, target_ref: str): """Parse a tool target into chat_id/thread_id and whether it is explicit.""" if platform_name == "telegram":