From 94c34f3d500083938a6e7fd7c98cee341220c125 Mon Sep 17 00:00:00 2001 From: olegbrok Date: Tue, 4 Aug 2026 15:12:40 -0700 Subject: [PATCH] Recover approval-gate backlogs --- src/pinky_daemon/agent_registry.py | 204 ++++++++++++++++-- src/pinky_daemon/api.py | 33 ++- src/pinky_daemon/broker.py | 112 ++++++++-- tests/test_approval_gate_recovery.py | 300 +++++++++++++++++++++++++++ 4 files changed, 614 insertions(+), 35 deletions(-) create mode 100644 tests/test_approval_gate_recovery.py diff --git a/src/pinky_daemon/agent_registry.py b/src/pinky_daemon/agent_registry.py index 93b75e15..360bfe22 100644 --- a/src/pinky_daemon/agent_registry.py +++ b/src/pinky_daemon/agent_registry.py @@ -1353,6 +1353,8 @@ def _init_tables(self) -> None: last_error TEXT NOT NULL DEFAULT '', notification_destination TEXT NOT NULL DEFAULT '{}', fallback_path TEXT NOT NULL DEFAULT '[]', + aging_reprompt_count INTEGER NOT NULL DEFAULT 0, + high_signal_alerted_at REAL NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL, FOREIGN KEY (agent_name) REFERENCES agents(name) ON DELETE CASCADE, @@ -1707,6 +1709,26 @@ def _migrate(self) -> None: ) _log("agent_registry: migrated — added sender_id to pending_messages") + # Approval maintenance metadata (#998). Aging re-prompts are bounded + # independently from transport retries/new-message notifications, and + # high-signal alerts are durable so one approved principal does not + # page the owner on every message while its channel awaits approval. + ar_existing = { + row[1] for row in self._db.execute("PRAGMA table_info(approval_requests)").fetchall() + } + if "aging_reprompt_count" not in ar_existing: + self._db.execute( + "ALTER TABLE approval_requests " + "ADD COLUMN aging_reprompt_count INTEGER NOT NULL DEFAULT 0" + ) + _log("agent_registry: migrated — added aging_reprompt_count to approval_requests") + if "high_signal_alerted_at" not in ar_existing: + self._db.execute( + "ALTER TABLE approval_requests " + "ADD COLUMN high_signal_alerted_at REAL NOT NULL DEFAULT 0" + ) + _log("agent_registry: migrated — added high_signal_alerted_at to approval_requests") + # Seed main_agent default: if unset, adopt the oldest enabled agent. # New installs get their main agent auto-assigned at create time (see # ``register``); this migration covers pre-existing installs whose @@ -3918,6 +3940,14 @@ def approve_user( approved_by=excluded.approved_by, updated_at=excluded.updated_at""", (agent_name, chat_id, display_name, approved_by, now, now), ) + # Approval state and its durable notification aggregate must transition + # together regardless of caller (API, owner command, or migration). + self._db.execute( + """UPDATE approval_requests + SET gate_state='approved', next_retry_at=0, updated_at=? + WHERE agent_name=? AND chat_id=?""", + (now, agent_name, chat_id), + ) self._db.commit() _log(f"agents: approved user {chat_id} for {agent_name}") row = self._db.execute( @@ -3941,6 +3971,12 @@ def deny_user(self, agent_name: str, chat_id: str) -> bool: DO UPDATE SET status='denied', updated_at=excluded.updated_at""", (agent_name, chat_id, now, now), ) + self._db.execute( + """UPDATE approval_requests + SET gate_state='denied', next_retry_at=0, updated_at=? + WHERE agent_name=? AND chat_id=?""", + (now, agent_name, chat_id), + ) self._db.commit() return cursor.rowcount > 0 @@ -4191,6 +4227,94 @@ def delete_pending_messages(self, agent_name: str, chat_id: str = "") -> int: self._db.commit() return cursor.rowcount + def list_approval_backlogs( + self, + agent_name: str = "", + chat_id: str = "", + *, + now: float | None = None, + ) -> list[dict]: + """Return every undelivered approval backlog, grouped by agent/chat. + + This deliberately mirrors the fleet-wide incident diagnostic: group + undelivered ``pending_messages`` by ``(agent_name, chat_id)`` and join + both the gate status and the original sender's approval status. The + latter identifies the high-signal case where an approved principal is + being silently held behind a still-pending group/channel gate. + """ + where = ["pm.delivered=0"] + params: list[object] = [] + if agent_name: + where.append("pm.agent_name=?") + params.append(agent_name) + if chat_id: + where.append("pm.chat_id=?") + params.append(chat_id) + rows = self._db.execute( + f"""SELECT pm.agent_name, pm.chat_id, + COALESCE(gate.status, 'missing'), + COALESCE(gate.display_name, ''), + COUNT(*), MIN(pm.created_at), MAX(pm.created_at), + COALESCE(ar.id, 0), + COALESCE(ar.notification_state, 'missing'), + COALESCE(ar.last_notified_at, 0), + COUNT(DISTINCT CASE WHEN principal.status='approved' + THEN pm.sender_id END), + GROUP_CONCAT(DISTINCT CASE WHEN principal.status='approved' + THEN pm.sender_id END) + FROM pending_messages AS pm + LEFT JOIN approved_users AS gate + ON gate.agent_name=pm.agent_name AND gate.chat_id=pm.chat_id + LEFT JOIN approved_users AS principal + ON principal.agent_name=pm.agent_name + AND principal.chat_id=pm.sender_id + LEFT JOIN approval_requests AS ar + ON ar.agent_name=pm.agent_name AND ar.chat_id=pm.chat_id + WHERE {' AND '.join(where)} + GROUP BY pm.agent_name, pm.chat_id, gate.status, + gate.display_name, ar.id, ar.notification_state, + ar.last_notified_at + ORDER BY MIN(pm.created_at), pm.agent_name, pm.chat_id""", + params, + ).fetchall() + observed_at = time.time() if now is None else now + return [ + { + "agent_name": row[0], + "chat_id": row[1], + "gate_status": row[2], + "display_name": row[3], + "undelivered_count": row[4], + "oldest_held_at": row[5], + "newest_held_at": row[6], + "oldest_age_seconds": max(0, int(observed_at - row[5])), + "request_id": row[7], + "notification_state": row[8], + "last_notified_at": row[9], + "approved_principal_count": row[10], + "approved_principal_ids": row[11].split(",") if row[11] else [], + "high_signal": row[2] != "approved" and row[10] > 0, + } + for row in rows + ] + + def get_approval_backlog_health(self, agent_name: str = "") -> dict: + """Owner/operator-facing summary of every undelivered approval row.""" + backlogs = self.list_approval_backlogs(agent_name) + pending = [row for row in backlogs if row["gate_status"] in ("pending", "missing")] + approved = [row for row in backlogs if row["gate_status"] == "approved"] + denied = [row for row in backlogs if row["gate_status"] == "denied"] + high_signal = [row for row in backlogs if row["high_signal"]] + return { + "healthy": not backlogs, + "pending_chats": len(pending), + "approved_stranded_chats": len(approved), + "denied_stranded_chats": len(denied), + "high_signal_chats": len(high_signal), + "undelivered_messages": sum(row["undelivered_count"] for row in backlogs), + "backlogs": backlogs, + } + # ── Approval Requests (#863 emergency lane) ───────────── @staticmethod @@ -4206,8 +4330,30 @@ def _approval_request_dict(row) -> dict: "notification_destination": json.loads(row[14] or "{}"), "fallback_path": json.loads(row[15] or "[]"), "created_at": row[16], "updated_at": row[17], + "aging_reprompt_count": row[18], + "high_signal_alerted_at": row[19], } + def _enrich_approval_request(self, request: dict) -> dict: + backlogs = self.list_approval_backlogs( + request["agent_name"], request["chat_id"], + ) + if backlogs: + request.update({ + "undelivered_count": backlogs[0]["undelivered_count"], + "approved_principal_count": backlogs[0]["approved_principal_count"], + "approved_principal_ids": backlogs[0]["approved_principal_ids"], + "high_signal": backlogs[0]["high_signal"], + }) + else: + request.update({ + "undelivered_count": 0, + "approved_principal_count": 0, + "approved_principal_ids": [], + "high_signal": False, + }) + return request + def get_approval_request(self, agent_name: str, chat_id: str) -> dict | None: """Return the stable request for a legacy ``(agent, chat_id)`` gate. @@ -4219,11 +4365,12 @@ def get_approval_request(self, agent_name: str, chat_id: str) -> dict | None: gate_state, held_count, oldest_held_at, notification_state, notification_attempts, notified_held_count, last_notified_at, next_retry_at, last_error, notification_destination, - fallback_path, created_at, updated_at + fallback_path, created_at, updated_at, + aging_reprompt_count, high_signal_alerted_at FROM approval_requests WHERE agent_name=? AND chat_id=?""", (agent_name, chat_id), ).fetchone() - return self._approval_request_dict(row) if row else None + return self._enrich_approval_request(self._approval_request_dict(row)) if row else None def record_approval_hold( self, agent_name: str, chat_id: str, *, target_name: str = "", @@ -4302,15 +4449,21 @@ def queue_pending_message_with_approval_request( raise RuntimeError("approval request missing after atomic hold commit") return message_id, request - def begin_approval_notification(self, request_id: int, *, reset_attempts: bool) -> None: + def begin_approval_notification( + self, + request_id: int, + *, + reset_attempts: bool, + aging_reprompt: bool = False, + ) -> None: """Mark a notification cycle active, optionally resetting re-notify attempts.""" - reset_sql = ", notification_attempts=0" if reset_attempts else "" self._db.execute( - f"""UPDATE approval_requests - SET notification_state='retrying', last_error='', updated_at=? - {reset_sql} - WHERE id=? AND gate_state='pending'""", - (time.time(), request_id), + """UPDATE approval_requests + SET notification_state='retrying', last_error='', updated_at=?, + notification_attempts=CASE WHEN ? THEN 0 ELSE notification_attempts END, + aging_reprompt_count=aging_reprompt_count + ? + WHERE id=? AND gate_state='pending'""", + (time.time(), int(reset_attempts), int(aging_reprompt), request_id), ) self._db.commit() @@ -4332,18 +4485,25 @@ def record_approval_notification_failure( self._db.commit() def record_approval_notification_delivered( - self, request_id: int, *, destination: dict, fallback_path: list[dict], + self, + request_id: int, + *, + destination: dict, + fallback_path: list[dict], + high_signal: bool = False, ) -> None: now = time.time() self._db.execute( """UPDATE approval_requests SET notification_state='delivered', notification_attempts=0, notified_held_count=held_count, last_notified_at=?, next_retry_at=0, - last_error='', notification_destination=?, fallback_path=?, updated_at=? + last_error='', notification_destination=?, fallback_path=?, updated_at=?, + high_signal_alerted_at=CASE WHEN ? THEN ? ELSE high_signal_alerted_at END WHERE id=? AND gate_state='pending'""", ( now, json.dumps(destination, separators=(",", ":")), - json.dumps(fallback_path, separators=(",", ":")), now, request_id, + json.dumps(fallback_path, separators=(",", ":")), now, + int(high_signal), now, request_id, ), ) self._db.commit() @@ -4360,19 +4520,31 @@ def settle_approval_request(self, agent_name: str, chat_id: str, state: str) -> self._db.commit() def list_due_approval_notifications(self, now: float | None = None) -> list[dict]: + """Return retry-due rows plus delivered rows eligible for maintenance. + + The broker applies the aging/new-hold/high-signal policy to delivered + candidates. Including them here lets the daemon re-prompt without a new + inbound message waking the approval path. + """ due_at = time.time() if now is None else now rows = self._db.execute( """SELECT id, agent_name, chat_id, target_name, is_channel, gate_state, held_count, oldest_held_at, notification_state, notification_attempts, notified_held_count, last_notified_at, next_retry_at, last_error, notification_destination, - fallback_path, created_at, updated_at + fallback_path, created_at, updated_at, + aging_reprompt_count, high_signal_alerted_at FROM approval_requests - WHERE gate_state='pending' AND notification_state='retrying' - AND next_retry_at <= ? ORDER BY next_retry_at, id""", + WHERE gate_state='pending' + AND ((notification_state='retrying' AND next_retry_at <= ?) + OR notification_state IN ('delivered', 'failed')) + ORDER BY next_retry_at, id""", (due_at,), ).fetchall() - return [self._approval_request_dict(row) for row in rows] + return [ + self._enrich_approval_request(self._approval_request_dict(row)) + for row in rows + ] def get_approval_notification_health(self, agent_name: str) -> dict: rows = self._db.execute( diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 4ae0f110..b67fd8fb 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -6184,6 +6184,11 @@ async def list_all_approved_users(): """List all approved users across all agents.""" return {"users": agents.list_all_approved_users()} + @app.get("/system/approval-backlog") + async def approval_backlog(): + """Fleet-wide owner/operator view of undelivered approval-gate rows.""" + return agents.get_approval_backlog_health() + # ── Outreach Platform Configuration Routes ────────── from pinky_daemon.routes.outreach import router as _outreach_router from pinky_daemon.routes.outreach import set_dependencies as _outreach_set_deps @@ -8190,6 +8195,7 @@ async def list_pending_messages(name: str): "pending_users": len(by_chat), "total_messages": len(messages), "by_sender": by_chat, + "approval_backlog": agents.get_approval_backlog_health(name), } @app.delete("/agents/{name}/pending-messages/{chat_id}") @@ -11181,6 +11187,18 @@ def _resolve_signing_key(agent_name: str) -> str | None: # carries the one-time owner digest until confirmed delivery. await _resume_grandfather_migration(agents, broker) + # #998: heal every approved-but-undelivered backlog, not only rows + # created by the grandfather migration. This catches approval changes + # made by older migrations/direct registry paths before any poller can + # accept new traffic; the broker's maintenance loop keeps reconciling + # later runtime transitions as well. + reconciled = await broker.reconcile_approved_pending_messages() + if reconciled: + _log( + "startup: reconciled " + f"{reconciled} approved pending message(s)" + ) + # Boot policy: the main agent always starts. Enabled siblings start only # when the shutdown manifest proves they had a live streaming transport; # all other siblings stay dormant until an inbound/scheduled wake. @@ -12563,7 +12581,11 @@ async def agent_health(agent_name: str): approval_notification_health = agents.get_approval_notification_health( agent_name, ) - checks = {"approval_notifications": approval_notification_health} + approval_backlog_health = agents.get_approval_backlog_health(agent_name) + checks = { + "approval_notifications": approval_notification_health, + "approval_backlog": approval_backlog_health, + } return { "agent": agent_name, @@ -12599,10 +12621,17 @@ def _health_recommendation(session, heartbeat, tasks, errors, checks=None) -> st if len(errors) >= 3: issues.append("high_error_rate") approval_check = (checks or {}).get("approval_notifications", {}) + backlog_check = (checks or {}).get("approval_backlog", {}) if approval_check.get("failed"): issues.append("owner_notification_failed") elif approval_check.get("retrying"): issues.append("owner_notification_retrying") + if backlog_check.get("approved_stranded_chats"): + issues.append("approved_messages_stranded") + if backlog_check.get("high_signal_chats"): + issues.append("approved_principal_blocked") + elif backlog_check.get("pending_chats"): + issues.append("approval_pending") if not issues: return "healthy" @@ -12614,6 +12643,8 @@ def _health_recommendation(session, heartbeat, tasks, errors, checks=None) -> st return "unstable" if "owner_notification_failed" in issues: return "needs_attention" + if "approved_messages_stranded" in issues or "approved_principal_blocked" in issues: + return "needs_attention" return "degraded" # time is imported at module level diff --git a/src/pinky_daemon/broker.py b/src/pinky_daemon/broker.py index 65d0df00..74fca449 100644 --- a/src/pinky_daemon/broker.py +++ b/src/pinky_daemon/broker.py @@ -43,6 +43,7 @@ # five cycles; every cycle walks the canonical primary + ordered fallbacks. _APPROVAL_RENOTIFY_INTERVAL_SEC = 4 * 60 * 60 _APPROVAL_RENOTIFY_HELD_COUNT = 10 +_APPROVAL_AGING_REPROMPT_AFTER_SEC = (24 * 60 * 60, 72 * 60 * 60) _APPROVAL_NOTIFY_RETRY_BASE_SEC = 30 _APPROVAL_NOTIFY_RETRY_MAX_SEC = 30 * 60 _APPROVAL_NOTIFY_MAX_ATTEMPTS = 5 @@ -309,6 +310,7 @@ def __init__( self._background_tasks: set[asyncio.Task] = set() self._approval_notification_task: asyncio.Task | None = None self._approval_notification_locks: dict[int, asyncio.Lock] = {} + self._approval_flush_locks: dict[tuple[str, str], asyncio.Lock] = {} @property def send_callback(self): @@ -1017,8 +1019,17 @@ async def _handle_auth_code_reply(self, message: BrokerMessage) -> bool: def _format_approval_notification(request: dict) -> str: approval_key = request["chat_id"] agent_name = request["agent_name"] - held_count = request["held_count"] + held_count = request.get("undelivered_count", request["held_count"]) oldest_age = max(0, int(time.time() - request["oldest_held_at"])) + alert = "" + if request.get("high_signal"): + principals = ", ".join(request.get("approved_principal_ids") or []) + alert = ( + "🚨 APPROVAL GATE BLACK-HOLE RISK\n" + f"This pending chat is holding messages from an already-approved " + f"{agent_name} principal" + f"{f' ({principals})' if principals else ''}.\n\n" + ) if request["is_channel"]: subject = ( f"{agent_name} has messages held from a new channel " @@ -1033,7 +1044,7 @@ def _format_approval_notification(request: dict) -> str: ) action = "Review the request:" return ( - f"🆕 {subject}\n\n" + f"{alert}🆕 {subject}\n\n" f"Held messages: {held_count}; oldest: {oldest_age}s\n" f"{action}\n" f"/approve_{approval_key}\n" @@ -1041,19 +1052,33 @@ def _format_approval_notification(request: dict) -> str: ) @staticmethod - def _approval_request_needs_notification(request: dict, now: float) -> bool: + def _approval_notification_reason(request: dict, now: float) -> str: if request["gate_state"] != "pending": - return False + return "" + if request.get("undelivered_count", request["held_count"]) < 1: + return "" if request["notification_state"] == "retrying": - return request["next_retry_at"] <= now + return "retry" if request["next_retry_at"] <= now else "" + aging_count = request.get("aging_reprompt_count", 0) + if aging_count < len(_APPROVAL_AGING_REPROMPT_AFTER_SEC): + threshold = _APPROVAL_AGING_REPROMPT_AFTER_SEC[aging_count] + if now - request["oldest_held_at"] >= threshold: + return "aging" + if request.get("high_signal") and not request.get("high_signal_alerted_at"): + return "high_signal" if request["notification_state"] == "failed": - return False + return "" new_holds = request["held_count"] - request["notified_held_count"] elapsed = now - request["last_notified_at"] - return ( - new_holds >= _APPROVAL_RENOTIFY_HELD_COUNT - or (new_holds > 0 and elapsed >= _APPROVAL_RENOTIFY_INTERVAL_SEC) - ) + if new_holds >= _APPROVAL_RENOTIFY_HELD_COUNT: + return "new_holds" + if new_holds > 0 and elapsed >= _APPROVAL_RENOTIFY_INTERVAL_SEC: + return "new_holds" + return "" + + @staticmethod + def _approval_request_needs_notification(request: dict, now: float) -> bool: + return bool(MessageBroker._approval_notification_reason(request, now)) async def _notify_approval_request(self, request: dict) -> None: """Deliver one durable approval notification over ordered fallbacks.""" @@ -1064,12 +1089,18 @@ async def _notify_approval_request(self, request: dict) -> None: request["agent_name"], request["chat_id"], ) now = time.time() - if not current or not self._approval_request_needs_notification(current, now): + reason = ( + self._approval_notification_reason(current, now) + if current else "" + ) + if not current or not reason: return - reset_attempts = current["notification_state"] == "delivered" + reset_attempts = current["notification_state"] in ("delivered", "failed") self._registry.begin_approval_notification( - request_id, reset_attempts=reset_attempts, + request_id, + reset_attempts=reset_attempts, + aging_reprompt=reason == "aging", ) current = self._registry.get_approval_request( current["agent_name"], current["chat_id"], @@ -1103,6 +1134,7 @@ async def _notify_approval_request(self, request: dict) -> None: request_id, destination=destination, fallback_path=fallback_path, + high_signal=bool(current.get("high_signal")), ) _log( "broker: owner approval notification delivered " @@ -1135,15 +1167,46 @@ async def _notify_approval_request(self, request: dict) -> None: async def retry_due_approval_notifications(self) -> int: """Attempt every due durable notification; return requests examined.""" - due = self._registry.list_due_approval_notifications() + now = time.time() + due = [ + request + for request in self._registry.list_due_approval_notifications(now) + if self._approval_request_needs_notification(request, now) + ] for request in due: await self._notify_approval_request(request) return len(due) + async def reconcile_approved_pending_messages(self) -> int: + """Flush held rows whose approval gate is already approved. + + This is the systemic catch-all for transitions outside the two normal + approval endpoints (migrations, direct registry callers, and old rows + surviving an upgrade). It is safe to run at startup and continuously; + ``handle_approval`` serializes each agent/chat flush and checkpoints + successful messages individually. + """ + delivered = 0 + for backlog in self._registry.list_approval_backlogs(): + if backlog["gate_status"] != "approved": + continue + agent_name = backlog["agent_name"] + chat_id = backlog["chat_id"] + try: + self._registry.settle_approval_request(agent_name, chat_id, "approved") + delivered += await self.handle_approval(agent_name, chat_id) + except Exception as exc: + _log( + "ERROR broker: approved pending-message reconcile failed " + f"for {agent_name}/{chat_id}: {exc}" + ) + return delivered + async def run_approval_notification_retries(self) -> None: """Daemon loop that resumes retrying receipts after process restarts.""" while True: try: + await self.reconcile_approved_pending_messages() await self.retry_due_approval_notifications() except asyncio.CancelledError: raise @@ -1394,6 +1457,14 @@ def _format_prompt(self, message: BrokerMessage) -> str: return f"{header}\n{body}" async def handle_approval(self, agent_name: str, chat_id: str) -> int: + """Serialize and flush one approved chat's held messages.""" + lock = self._approval_flush_locks.setdefault( + (agent_name, chat_id), asyncio.Lock(), + ) + async with lock: + return await self._handle_approval_unlocked(agent_name, chat_id) + + async def _handle_approval_unlocked(self, agent_name: str, chat_id: str) -> int: """When a pending user is approved, deliver their held messages. Successful rows are checkpointed individually so a later-row failure @@ -1425,7 +1496,11 @@ async def handle_approval(self, agent_name: str, chat_id: str) -> int: timestamp=msg["created_at"], is_group=bool(msg.get("is_group")), ) - await self._route_streaming(agent_name, broker_msg) + handoff = await self._route_streaming(agent_name, broker_msg) + if handoff is False: + raise RuntimeError( + f"streaming handoff unavailable for {agent_name}/{chat_id}" + ) # Checkpoint each successful handoff before attempting the next # row. A later-row failure can then retry without deterministically @@ -1458,7 +1533,7 @@ def _get_streaming_session(self, agent_name: str, chat_id: str = ""): # Fall back to main return sessions.get("main") - async def _route_streaming(self, agent_name: str, message: BrokerMessage) -> None: + async def _route_streaming(self, agent_name: str, message: BrokerMessage) -> bool: """Route a message via streaming session — non-blocking. Resolution order: @@ -1586,7 +1661,7 @@ async def _route_streaming(self, agent_name: str, message: BrokerMessage) -> Non agent_name, message.platform, message.chat_id, f"⚠️ {agent_name} is not running right now. Try again later.", ) - return + return False # Show typing indicator if self._typing_callback: @@ -1619,7 +1694,7 @@ async def _route_streaming(self, agent_name: str, message: BrokerMessage) -> Non agent_name, message.platform, message.chat_id, "I received your voice message but couldn't transcribe it — please try again or send text.", ) - return + return False else: self._voice_pending.pop((agent_name, message.chat_id), None) @@ -1668,6 +1743,7 @@ async def _route_streaming(self, agent_name: str, message: BrokerMessage) -> Non await self._start_typing(agent_name, message.platform, message.chat_id, streaming) self._stats["routed"] += 1 _log(f"broker: streamed message to {agent_name} (non-blocking)") + return True async def inject_agent_message( self, diff --git a/tests/test_approval_gate_recovery.py b/tests/test_approval_gate_recovery.py new file mode 100644 index 00000000..433113d2 --- /dev/null +++ b/tests/test_approval_gate_recovery.py @@ -0,0 +1,300 @@ +from __future__ import annotations + +import time + +import pytest +from fastapi.testclient import TestClient + +from pinky_daemon.agent_registry import AgentRegistry +from pinky_daemon.api import create_api +from pinky_daemon.broker import BrokerMessage, MessageBroker +from pinky_daemon.sessions import SessionManager + + +def _registry(tmp_path) -> AgentRegistry: + registry = AgentRegistry(db_path=str(tmp_path / "agents.db")) + registry.register("onesie", model="sonnet", working_dir=str(tmp_path)) + return registry + + +def _broker(registry: AgentRegistry, sent: list[str]) -> MessageBroker: + async def send_callback( + agent_name: str, + platform: str, + chat_id: str, + content: str, + *, + account_id: str = "", + ) -> None: + sent.append(content) + + return MessageBroker( + registry, + SessionManager(), + send_callback=send_callback, + ) + + +def _owner_notifications(registry: AgentRegistry) -> None: + registry.set_primary_user("owner", display_name="Brad") + registry.set_owner_notification_destinations([ + { + "platform": "telegram", + "account_id": "owner-bot", + "conversation_id": "owner", + "principal_id": "owner", + } + ]) + + +@pytest.mark.asyncio +async def test_aging_reprompts_at_24h_and_72h_then_stops(tmp_path) -> None: + registry = _registry(tmp_path) + sent: list[str] = [] + broker = _broker(registry, sent) + _owner_notifications(registry) + + registry.add_pending_user("onesie", "C_TRAVEL", "Travel group") + _, request = registry.queue_pending_message_with_approval_request( + agent_name="onesie", + platform="telegram", + chat_id="C_TRAVEL", + reply_chat_id="C_TRAVEL", + sender_name="Guest", + sender_id="guest", + content="Are you here?", + is_group=True, + target_name="Travel group", + ) + await broker._notify_approval_request(request) + assert len(sent) == 1 + + now = time.time() + registry._db.execute( + "UPDATE approval_requests SET oldest_held_at=? WHERE agent_name=? AND chat_id=?", + (now - (25 * 60 * 60), "onesie", "C_TRAVEL"), + ) + registry._db.commit() + assert await broker.retry_due_approval_notifications() == 1 + assert len(sent) == 2 + assert registry.get_approval_request("onesie", "C_TRAVEL")[ + "aging_reprompt_count" + ] == 1 + + registry._db.execute( + "UPDATE approval_requests SET oldest_held_at=? WHERE agent_name=? AND chat_id=?", + (now - (73 * 60 * 60), "onesie", "C_TRAVEL"), + ) + registry._db.commit() + assert await broker.retry_due_approval_notifications() == 1 + assert len(sent) == 3 + assert registry.get_approval_request("onesie", "C_TRAVEL")[ + "aging_reprompt_count" + ] == 2 + + assert await broker.retry_due_approval_notifications() == 0 + assert len(sent) == 3 + + +@pytest.mark.asyncio +async def test_aging_does_not_reprompt_without_undelivered_messages(tmp_path) -> None: + registry = _registry(tmp_path) + sent: list[str] = [] + broker = _broker(registry, sent) + _owner_notifications(registry) + registry.add_pending_user("onesie", "C_EMPTY", "Empty group") + _, request = registry.queue_pending_message_with_approval_request( + agent_name="onesie", + platform="telegram", + chat_id="C_EMPTY", + sender_name="Guest", + sender_id="guest", + content="already handled", + is_group=True, + target_name="Empty group", + ) + await broker._notify_approval_request(request) + registry.mark_pending_delivered("onesie", "C_EMPTY") + registry._db.execute( + "UPDATE approval_requests SET oldest_held_at=? WHERE agent_name=? AND chat_id=?", + (time.time() - (96 * 60 * 60), "onesie", "C_EMPTY"), + ) + registry._db.commit() + + assert await broker.retry_due_approval_notifications() == 0 + assert len(sent) == 1 + + +@pytest.mark.asyncio +async def test_pending_channel_pages_loudly_for_approved_principal(tmp_path) -> None: + registry = _registry(tmp_path) + sent: list[str] = [] + broker = _broker(registry, sent) + _owner_notifications(registry) + registry.approve_user("onesie", "dmitri", "Dmitri", "owner") + + await broker.handle_inbound(BrokerMessage( + platform="telegram", + chat_id="C_TRAVEL", + sender_name="Guest", + sender_id="guest", + content="first held message", + agent_name="onesie", + is_group=True, + )) + assert len(sent) == 1 + assert not sent[0].startswith("🚨") + + await broker.handle_inbound(BrokerMessage( + platform="telegram", + chat_id="C_TRAVEL", + sender_name="Dmitri", + sender_id="dmitri", + content="Are you here?", + agent_name="onesie", + is_group=True, + )) + + assert len(sent) == 2 + assert sent[1].startswith("🚨 APPROVAL GATE BLACK-HOLE RISK") + assert "already-approved onesie principal (dmitri)" in sent[1] + request = registry.get_approval_request("onesie", "C_TRAVEL") + assert request["high_signal"] is True + assert request["high_signal_alerted_at"] > 0 + + +@pytest.mark.asyncio +async def test_approved_transition_settles_and_reconcile_flushes(tmp_path, monkeypatch) -> None: + registry = _registry(tmp_path) + broker = _broker(registry, []) + routed: list[BrokerMessage] = [] + + async def route(agent_name: str, message: BrokerMessage) -> None: + routed.append(message) + + monkeypatch.setattr(broker, "_route_streaming", route) + registry.add_pending_user("onesie", "C_TRAVEL", "Travel group") + registry.queue_pending_message_with_approval_request( + agent_name="onesie", + platform="telegram", + chat_id="C_TRAVEL", + reply_chat_id="C_TRAVEL", + sender_name="Dmitri", + sender_id="dmitri", + content="standing instruction", + is_group=True, + target_name="Travel group", + ) + + registry.approve_user("onesie", "C_TRAVEL", "Travel group", "migration") + + assert registry.get_approval_request("onesie", "C_TRAVEL")["gate_state"] == "approved" + assert await broker.reconcile_approved_pending_messages() == 1 + assert [message.content for message in routed] == ["standing instruction"] + assert registry.get_pending_messages("onesie", "C_TRAVEL") == [] + + +@pytest.mark.asyncio +async def test_startup_reconcile_heals_direct_migration_approval(tmp_path, monkeypatch) -> None: + registry = _registry(tmp_path) + broker = _broker(registry, []) + routed: list[BrokerMessage] = [] + + async def route(agent_name: str, message: BrokerMessage) -> None: + routed.append(message) + + monkeypatch.setattr(broker, "_route_streaming", route) + registry.add_pending_user("onesie", "C_LEGACY", "Legacy group") + registry.queue_pending_message_with_approval_request( + agent_name="onesie", + platform="telegram", + chat_id="C_LEGACY", + reply_chat_id="C_LEGACY", + sender_name="Dmitri", + sender_id="dmitri", + content="legacy held row", + is_group=True, + target_name="Legacy group", + ) + registry._db.execute( + "UPDATE approved_users SET status='approved' WHERE agent_name=? AND chat_id=?", + ("onesie", "C_LEGACY"), + ) + registry._db.commit() + + assert registry.get_approval_request("onesie", "C_LEGACY")["gate_state"] == "pending" + assert await broker.reconcile_approved_pending_messages() == 1 + assert registry.get_approval_request("onesie", "C_LEGACY")["gate_state"] == "approved" + assert [message.content for message in routed] == ["legacy held row"] + + +@pytest.mark.asyncio +async def test_reconcile_keeps_row_undelivered_when_session_handoff_fails( + tmp_path, monkeypatch, +) -> None: + registry = _registry(tmp_path) + broker = _broker(registry, []) + + async def unavailable(agent_name: str, message: BrokerMessage) -> bool: + return False + + monkeypatch.setattr(broker, "_route_streaming", unavailable) + registry.add_pending_user("onesie", "C_OFFLINE", "Offline group") + registry.queue_pending_message_with_approval_request( + agent_name="onesie", + platform="telegram", + chat_id="C_OFFLINE", + reply_chat_id="C_OFFLINE", + sender_name="Dmitri", + sender_id="dmitri", + content="do not drop me", + is_group=True, + target_name="Offline group", + ) + registry.approve_user("onesie", "C_OFFLINE", "Offline group", "migration") + + assert await broker.reconcile_approved_pending_messages() == 0 + assert [ + message["content"] + for message in registry.get_pending_messages("onesie", "C_OFFLINE") + ] == ["do not drop me"] + + +def test_backlog_diagnostic_is_fleet_and_agent_visible(tmp_path) -> None: + app = create_api( + max_sessions=10, + default_working_dir=str(tmp_path), + db_path=str(tmp_path / "pinky.db"), + ) + registry = app.state.agents + registry.register("onesie", model="sonnet", working_dir=str(tmp_path)) + registry.set_primary_user("owner", display_name="Brad") + registry.approve_user("onesie", "dmitri", "Dmitri", "owner") + registry.add_pending_user("onesie", "C_TRAVEL", "Travel group") + registry.queue_pending_message_with_approval_request( + agent_name="onesie", + platform="telegram", + chat_id="C_TRAVEL", + reply_chat_id="C_TRAVEL", + sender_name="Dmitri", + sender_id="dmitri", + content="Are you here?", + is_group=True, + target_name="Travel group", + ) + client = TestClient(app) + + fleet = client.get("/system/approval-backlog") + health = client.get("/agents/onesie/health") + + assert fleet.status_code == 200 + assert fleet.json()["pending_chats"] == 1 + assert fleet.json()["high_signal_chats"] == 1 + assert fleet.json()["undelivered_messages"] == 1 + row = fleet.json()["backlogs"][0] + assert (row["agent_name"], row["chat_id"], row["undelivered_count"]) == ( + "onesie", "C_TRAVEL", 1, + ) + check = health.json()["checks"]["approval_backlog"] + assert check["high_signal_chats"] == 1 + assert health.json()["recommendation"] == "needs_attention"