"""Gateway runner best-effort helpers (P0-A).""" from __future__ import annotations import asyncio import logging from typing import Any from butler.core.best_effort import async_safe_best_effort, safe_best_effort from butler.skills.similarity import _ensure_jieba from butler.ops.degradation_registry import sync_all_startup_degradations from butler.resilience.message_queue import restore_persisted_queue from butler.mcp.async_runner import graceful_shutdown_mcp_stack from butler.memory.semantic_index import close_all_semantic_indices from butler.tools.reminder import poll_due_reminders from butler.resilience.durable_outbox import ( durable_outbox_enabled, enqueue_outbox_message, list_pending_outbox, mark_outbox_sent, ) logger = logging.getLogger(__name__) def warmup_gateway_runtime_safe(butler: Any) -> None: def _skills() -> None: _ensure_jieba() mgr = getattr(butler._orchestrator, "_skill_manager", None) if mgr is not None: mgr.list_skills() logger.info("Gateway runtime warmup complete (skills/jieba)") safe_best_effort(_skills, label="runner.warmup_skills", default=None) def _degradation() -> None: sync_all_startup_degradations() safe_best_effort(_degradation, label="runner.warmup_degradation", default=None) def restore_persisted_queue_safe() -> int: def _run() -> int: return int(restore_persisted_queue() or 0) result = safe_best_effort(_run, label="runner.queue_restore", default=0) return int(result or 0) async def disconnect_adapter_safe(adapter: Any) -> None: async def _run() -> None: await adapter.disconnect() await async_safe_best_effort( _run, label=f"runner.disconnect.{getattr(adapter, 'name', '?')}", default=None, ) def shutdown_mcp_stack_safe() -> None: def _run() -> None: graceful_shutdown_mcp_stack(timeout=5.0) safe_best_effort(_run, label="runner.mcp_shutdown", default=None) def close_semantic_indices_safe() -> None: def _run() -> None: close_all_semantic_indices() safe_best_effort(_run, label="runner.semantic_index_shutdown", default=None) def sync_send_via_adapter_loud( adapters: list[Any], chat_id: str, text: str, ) -> bool: """Send via first adapter; log warning and try next on failure.""" if not chat_id or not text: return False for adapter in adapters: if not hasattr(adapter, "send"): continue try: loop = None try: loop = asyncio.get_running_loop() except RuntimeError: pass if loop and loop.is_running(): asyncio.run_coroutine_threadsafe( adapter.send(chat_id, text), loop ).result(timeout=30) else: asyncio.run(adapter.send(chat_id, text)) return True except Exception as exc: logger.warning("_sync_send_via_adapter failed: %s", exc) continue logger.warning("No adapter with send() available") return False def push_reminder_safe(adapters: list[Any], reminder: dict[str, Any]) -> None: def _run() -> None: import os from butler.defaults.env_defaults import OWNER_WECHAT_ID_DEFAULT text = f"⏰ 提醒:{reminder.get('message', '')}\n(设定时间:{reminder.get('due_human', '')})" chat_id = os.getenv("BUTLER_OWNER_WECHAT_ID", OWNER_WECHAT_ID_DEFAULT) if not chat_id: logger.warning( "BUTLER_OWNER_WECHAT_ID not set, reminder not pushed: %s", reminder.get("id"), ) return if durable_outbox_enabled(): entry_id = enqueue_outbox_message(chat_id, text, kind="reminder") if sync_send_via_adapter_loud(adapters, chat_id, text): mark_outbox_sent(entry_id) logger.info("Reminder sent+marked via outbox: %s", reminder.get("id")) else: logger.warning( "Reminder enqueued but send failed (will retry on replay): %s", reminder.get("id"), ) elif sync_send_via_adapter_loud(adapters, chat_id, text): logger.info("Reminder fired (direct): %s", reminder.get("id")) else: logger.warning("Reminder direct send failed: %s", reminder.get("id")) safe_best_effort(_run, label="runner.reminder_push", default=None) def poll_due_reminders_safe() -> list[dict[str, Any]]: def _run() -> list[dict[str, Any]]: fired = poll_due_reminders() return fired if isinstance(fired, list) else [] result = safe_best_effort(_run, label="runner.reminder_poll", default=[]) return result if isinstance(result, list) else [] def replay_outbox_entry_safe( adapters: list[Any], entry: dict[str, Any], ) -> bool: """Replay one outbox entry; return True if sent and marked.""" chat_id = entry.get("chat_id", "") body = entry.get("body", "") entry_id = entry.get("entry_id", "") if not chat_id or not body: return False if not sync_send_via_adapter_loud(adapters, chat_id, body): logger.warning("Outbox replay send failed for entry %s", entry_id) return False mark_outbox_sent(entry_id) return True def replay_pending_outbox_safe(adapters: list[Any]) -> tuple[int, int]: """Return (sent_count, total_pending).""" def _run() -> tuple[int, int]: if not durable_outbox_enabled(): return 0, 0 pending = list_pending_outbox() if not pending: return 0, 0 sent = 0 for entry in pending: if replay_outbox_entry_safe(adapters, entry): sent += 1 return sent, len(pending) result = safe_best_effort(_run, label="runner.outbox_replay", default=(0, 0)) if isinstance(result, tuple) and len(result) == 2: return int(result[0]), int(result[1]) return 0, 0