butler/gateway/completion_notify.py
Rama main · réplica de lectura del repositorio original.
"""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()