"""WeChat completion pushes for long gateway turns (outbound bridge, not shell hooks).""" from __future__ import annotations import logging import os import time from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: from butler.gateway.outbound_bridge import GatewayOutboundBridge from butler.contracts.completion_registry import set_completion_hooks from butler.contracts.gateway_registry import get_bridge_access from butler.defaults.env_defaults import ( GATEWAY_COMPLETION_NOTIFY_MIN_SECONDS, GATEWAY_DELEGATE_COMPLETION_MAX_EACH, GATEWAY_DELEGATE_COMPLETION_MODE_DEFAULT, ) from butler.utilities.env_parse import env_truthy, float_env, int_env from butler.gateway import delegate_push_dedup as _delegate_push_dedup from butler.gateway.completion_notify_ops import ( deliver_completion_push_safe, outbox_diagnostic_lines_safe, ) from butler.gateway.completion_telemetry import ( completion_push_stats, push_queue_pending_count, record_completion_push_enqueued, record_completion_push_failed, record_completion_push_sent, ) from butler.resilience.durable_outbox import ( enqueue_outbox_message, mark_outbox_failed, mark_outbox_sent, ) from butler.gateway.completion_policy import suppress_completion_after_main_enabled from butler.gateway.pii_scrub import scrub_outbound_text from butler.gateway.wechat_text_export import build_delegate_completion_message from butler.report import AgentReport, cache_report, format_for_wechat, get_last_report from butler.runtime.notify import ( mark_wechat_push_sent, should_enqueue_wechat_push_failure, wait_wechat_push_cooldown, wechat_push_cooldown_seconds, ) from butler.runtime.push_queue import enqueue_failed_push logger = logging.getLogger(__name__) def _optional_bridge() -> GatewayOutboundBridge | None: access = get_bridge_access() if access is not None: bridge = access.get_optional_bridge() if bridge is not None: return cast("GatewayOutboundBridge", bridge) from butler.gateway.outbound_bridge import get_gateway_bridge_optional return get_gateway_bridge_optional() def _env_bool(name: str, default: bool) -> bool: return cast(bool, env_truthy(name, default=default)) def _env_float(name: str, default: float) -> float: return cast(float, float_env(name, default)) def completion_notify_enabled() -> bool: return _env_bool("BUTLER_GATEWAY_COMPLETION_NOTIFY", True) def min_elapsed_for_push() -> float: return max( 0.0, _env_float( "BUTLER_GATEWAY_COMPLETION_NOTIFY_MIN_SECONDS", GATEWAY_COMPLETION_NOTIFY_MIN_SECONDS, ), ) def delegate_completion_enabled() -> bool: return _env_bool("BUTLER_GATEWAY_DELEGATE_COMPLETION_NOTIFY", True) def delegate_progress_notify_enabled() -> bool: """Mid-flight WeChat progress for long delegates (also enables task milestones).""" return _env_bool("BUTLER_GATEWAY_DELEGATE_PROGRESS_NOTIFY", False) def turn_completion_enabled() -> bool: return _env_bool("BUTLER_GATEWAY_TURN_COMPLETION_NOTIFY", True) def workflow_completion_enabled() -> bool: return _env_bool("BUTLER_GATEWAY_WORKFLOW_COMPLETION_NOTIFY", True) def timeout_completion_enabled() -> bool: return _env_bool("BUTLER_GATEWAY_TIMEOUT_COMPLETION_NOTIFY", True) def delegate_completion_mode() -> str: """last: only the final delegate in a turn pushes; each: up to N; once: first only.""" return (os.getenv("BUTLER_GATEWAY_DELEGATE_COMPLETION_MODE", GATEWAY_DELEGATE_COMPLETION_MODE_DEFAULT) or GATEWAY_DELEGATE_COMPLETION_MODE_DEFAULT).strip().lower() def delegate_completion_max_each() -> int: return cast( int, int_env( "BUTLER_GATEWAY_DELEGATE_COMPLETION_MAX_EACH", GATEWAY_DELEGATE_COMPLETION_MAX_EACH, min=1, ), ) def format_outbound_diagnostic_lines( session_key: str = "", *, chat_id: str = "", ) -> list[str]: """Lines for ``/诊断`` — completion notify policy + session push stats.""" mode = delegate_completion_mode() flags: list[str] = [] if completion_notify_enabled(): flags.append("完成提醒:开") flags.append(f"委派={mode}") if mode == "each": flags.append(f"each≤{delegate_completion_max_each()}") flags.append(f"最短{int(min_elapsed_for_push())}s") flags.append(f"冷却{int(wechat_push_cooldown_seconds())}s") if not delegate_completion_enabled(): flags.append("委派推:关") if not turn_completion_enabled(): flags.append("整轮推:关") if not timeout_completion_enabled(): flags.append("超时推:关") else: flags.append("完成提醒:关") lines = [f"出站策略: {' · '.join(flags)}"] stats = completion_push_stats(session_key or chat_id) if any(stats.values()): lines.append( "出站推送本轮: " f"成功{stats['sent']} 失败{stats['failed']} 入队{stats['enqueued']}" ) else: lines.append("出站推送本轮: 无记录") pending = push_queue_pending_count(chat_id=chat_id) if pending: lines.append(f"推送队列待发: {pending} 条") lines.extend(outbox_diagnostic_lines_safe(chat_id)) return lines def format_elapsed(seconds: float) -> str: total = max(0, int(seconds)) if total < 60: return f"{total} 秒" minutes, secs = divmod(total, 60) if minutes < 60: return f"{minutes} 分 {secs} 秒" if secs else f"{minutes} 分钟" hours, rem = divmod(minutes, 60) return f"{hours} 小时 {rem} 分钟" def build_report_push_text(report: AgentReport, *, prefix: str = "") -> str: body = cast(str, format_for_wechat(report)) if prefix: return f"{prefix}\n\n{body}".strip() return body def build_turn_complete_text(*, elapsed_seconds: float) -> str: return ( f"✅ 本轮处理已完成（用时约 {format_elapsed(elapsed_seconds)}）\n" "完整回复见上一条；发 /详细 可看报告。" ) def should_push_delegate_completion( bridge: GatewayOutboundBridge, elapsed_turn_seconds: float, ) -> bool: if not completion_notify_enabled() or not delegate_completion_enabled(): return False mode = delegate_completion_mode() if mode == "once" and bridge.delegate_push_count >= 1: return False if mode == "each" and bridge.delegate_push_count >= delegate_completion_max_each(): return False if bridge.ack_sent: return True if elapsed_turn_seconds >= min_elapsed_for_push(): return True return _env_bool("BUTLER_GATEWAY_DELEGATE_PUSH_ALWAYS", False) def should_push_workflow_completion( bridge: GatewayOutboundBridge, elapsed_turn_seconds: float, ) -> bool: if not completion_notify_enabled() or not workflow_completion_enabled(): return False if bridge.completion_push_sent: return False if bridge.ack_sent: return True return elapsed_turn_seconds >= min_elapsed_for_push() def should_push_turn_completion( bridge: GatewayOutboundBridge, elapsed_turn_seconds: float, ) -> bool: if not completion_notify_enabled() or not turn_completion_enabled(): return False if bridge.completion_push_sent or bridge.timeout_notified: return False if not bridge.ack_sent: return False return elapsed_turn_seconds >= min_elapsed_for_push() def should_push_timeout_completion( bridge: GatewayOutboundBridge, elapsed_turn_seconds: float, ) -> bool: if not completion_notify_enabled() or not timeout_completion_enabled(): return False if bridge.timeout_notified or bridge.completion_push_sent: return False return bridge.ack_sent or elapsed_turn_seconds >= min_elapsed_for_push() def build_timeout_text(*, timeout_seconds: float, elapsed_seconds: float) -> str: limit = int(timeout_seconds) return ( f"⏱ 处理超时（已超过 {limit} 秒，本轮约 {format_elapsed(elapsed_seconds)}）\n" "请稍后重试，或发 /诊断 查看状态。" ) def try_push_turn_timeout( bridge: GatewayOutboundBridge | None, *, timeout_seconds: float, elapsed_seconds: float, ) -> bool: br = bridge if br is None: return False if not should_push_timeout_completion(br, elapsed_seconds): return False return cast( bool, br.schedule_completion_push( build_timeout_text( timeout_seconds=timeout_seconds, elapsed_seconds=elapsed_seconds, ), kind="timeout", ), ) def flush_pending_delegate_completion(bridge: GatewayOutboundBridge) -> bool: """Push deferred delegate report (``delegate_completion_mode=last``).""" report = bridge.take_pending_delegate_report() if report is None: return False tid = str(getattr(report, "task_id", "") or "").strip() ok, reason = _delegate_push_dedup.should_deliver_delegate_push(bridge.chat_id, tid) if not ok: logger.info( "pending delegate flush suppressed chat=%s task=%s reason=%s", str(bridge.chat_id or "")[:12], tid, reason, ) return False if ( getattr(bridge, "_final_sent", False) and getattr(bridge, "_main_reply_chars", 0) > 0 ): if suppress_completion_after_main_enabled(): return False elapsed = time.monotonic() - bridge.turn_started_at if bridge.turn_started_at else 0.0 if not should_push_delegate_completion(bridge, elapsed): return False text = build_delegate_completion_message( report, prefix="📋 委派阶段完成", platform="wechat", ) return cast(bool, bridge.schedule_completion_push(text, kind="delegate")) def _workflow_push_prefix(report: AgentReport) -> str: if report.success: return "📋 工作流阶段完成" return "⚠️ 工作流未完成" async def deliver_completion_push( adapter: Any, chat_id: str, body: str, *, kind: str, ) -> bool: """Send completion text with shared WeChat cooldown; enqueue on retryable failure.""" import asyncio telemetry_key = chat_id or "" body = scrub_outbound_text(body) if kind == "delegate": if _delegate_push_dedup.maybe_defer_delegate_push(chat_id, body, kind=kind): return True ok, reason = _delegate_push_dedup.should_deliver_delegate_push(chat_id, "", body=body) if not ok: logger.info( "Gateway delegate push suppressed chat=%s reason=%s", str(chat_id or "")[:12], reason, ) return False outbox_id = enqueue_outbox_message(chat_id, body, kind=kind) await asyncio.to_thread(wait_wechat_push_cooldown) title = f"[Butler] {kind}完成提醒" async def _on_success() -> None: await asyncio.to_thread(mark_wechat_push_sent) mark_outbox_sent(outbox_id) record_completion_push_sent(session_key=telemetry_key) if kind == "delegate": _delegate_push_dedup.mark_delegate_push_delivered(chat_id, "", body=body) def _on_failure(exc: Exception) -> None: mark_outbox_failed(outbox_id, error=str(exc)) if should_enqueue_wechat_push_failure(str(exc)): enqueue_failed_push(title, body, chat_id=chat_id) record_completion_push_enqueued(session_key=telemetry_key) else: record_completion_push_failed(session_key=telemetry_key) return cast( bool, await deliver_completion_push_safe( lambda: adapter.send(chat_id, body), on_success=_on_success, on_failure=_on_failure, kind=kind, ), ) def try_push_workflow_failure( bridge: GatewayOutboundBridge | None, workflow_name: str, error: Exception | str, *, session_key: str = "", ) -> bool: br = bridge if br is None: return False report = get_last_report(session_key) if report is None: msg = str(error)[:2000] report = AgentReport( headline=f"工作流 {workflow_name} 失败", summary=msg, success=False, task_preview=f"workflow:{workflow_name}"[:200], issues=[msg[:500]], ) cache_report(report, session_key=session_key or "default") elapsed = time.monotonic() - br.turn_started_at if br.turn_started_at else 0.0 if not should_push_workflow_completion(br, elapsed): return False text = build_report_push_text(report, prefix=_workflow_push_prefix(report)) return cast(bool, br.schedule_completion_push(text, kind="workflow")) def try_push_agent_report( report: AgentReport, *, kind: str, bridge: GatewayOutboundBridge | None = None, elapsed_turn_seconds: float | None = None, ) -> bool: """Schedule a WeChat completion message via the outbound bridge.""" br = bridge or _optional_bridge() if br is None: return False elapsed = ( float(elapsed_turn_seconds) if elapsed_turn_seconds is not None else (time.monotonic() - br.turn_started_at if br.turn_started_at else 0.0) ) if kind == "delegate": if delegate_completion_mode() == "last": br.set_pending_delegate_report(report) return True if not should_push_delegate_completion(br, elapsed): return False prefix = "📋 委派阶段完成" elif kind == "workflow": if not should_push_workflow_completion(br, elapsed): return False prefix = _workflow_push_prefix(report) else: return False text = build_report_push_text(report, prefix=prefix) return cast(bool, br.schedule_completion_push(text, kind=kind)) def try_push_turn_complete( bridge: GatewayOutboundBridge | None, *, elapsed_seconds: float, ) -> bool: br = bridge or _optional_bridge() if br is None: return False if not should_push_turn_completion(br, elapsed_seconds): return False return cast( bool, br.schedule_completion_push( build_turn_complete_text(elapsed_seconds=elapsed_seconds), kind="turn", ), ) class _LiveOutboundCompletionHooks: def flush_pending_delegate_completion(self, bridge: Any) -> bool: return flush_pending_delegate_completion(cast("GatewayOutboundBridge", bridge)) def try_push_agent_report( self, report: Any, *, kind: str, bridge: Any, elapsed_turn_seconds: float, ) -> bool: return try_push_agent_report( report, kind=kind, bridge=cast("GatewayOutboundBridge | None", bridge), elapsed_turn_seconds=elapsed_turn_seconds, ) def try_push_turn_timeout( self, bridge: Any, *, timeout_seconds: float, elapsed_seconds: float, ) -> bool: return try_push_turn_timeout( cast("GatewayOutboundBridge | None", bridge), timeout_seconds=timeout_seconds, elapsed_seconds=elapsed_seconds, ) def try_push_turn_complete( self, bridge: Any, *, elapsed_seconds: float, ) -> bool: return try_push_turn_complete( cast("GatewayOutboundBridge | None", bridge), elapsed_seconds=elapsed_seconds, ) async def deliver_completion_push( self, adapter: Any, chat_id: str, body: str, *, kind: str, ) -> bool: return await deliver_completion_push(adapter, chat_id, body, kind=kind) def delegate_completion_mode(self) -> str: return delegate_completion_mode() def delegate_completion_max_each(self) -> int: return delegate_completion_max_each() def register_default_completion_hooks() -> None: set_completion_hooks(_LiveOutboundCompletionHooks()) register_default_completion_hooks()