butler/gateway/outbound_bridge.py
Rama main · réplica de lectura del repositorio original.
"""Thread-safe WeChat (and native gateway) outbound UX: typing, ack, milestones."""
from __future__ import annotations
import asyncio
import logging
import os
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Optional, Protocol, cast
from butler.contracts.completion_ports import OutboundCompletionHooks
from butler.contracts.completion_registry import get_completion_hooks
from butler.core.best_effort import async_safe_best_effort, safe_best_effort
from butler.defaults.env_defaults import (
GATEWAY_PROGRESS_MAX_ACK_MESSAGES,
GATEWAY_STREAM_PREVIEW_DEFAULT,
)
from butler.utilities.env_parse import env_truthy, float_env, int_env
from butler.gateway.completion_policy import (
max_supplementary_per_turn,
suppress_completion_after_main_enabled,
)
from butler.gateway.item_event_sink import record_thread_item
from butler.gateway.outbound_bridge_ops import (
run_milestone_timer_tick,
run_typing_refresh_loop,
schedule_coro_threadsafe,
send_adapter_message,
send_progress_ack,
)
from butler.gateway.outbound_events import delegate_event, stream_event, workflow_event
from butler.gateway.pii_scrub import scrub_outbound_text
from butler.gateway.progressive_stream import maybe_schedule_progressive_reply
from butler.gateway.task_milestone import task_milestone_enabled
logger = logging.getLogger(__name__)
_thread_bridge = threading.local()
MILESTONE_TOOLS = frozenset(
{"delegate_task", "run_workflow", "write_file", "patch", "terminal"}
)
class _TypingAdapter(Protocol):
async def send_typing(self, chat_id: str, metadata: Optional[dict[str, Any]] = None) -> None: ...
async def stop_typing(self, chat_id: str) -> None: ...
async def send(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
metadata: Optional[dict[str, Any]] = None,
) -> Any: ...
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 set_current_bridge(bridge: GatewayOutboundBridge | None) -> None:
_thread_bridge.bridge = bridge
def get_current_bridge() -> GatewayOutboundBridge | None:
return getattr(_thread_bridge, "bridge", None)
def get_gateway_bridge_optional() -> GatewayOutboundBridge | None:
return get_current_bridge()
def _env_int(name: str, default: int, *, min_value: int = 0) -> int:
return cast(int, int_env(name, default, min=min_value))
def _completion_hooks() -> OutboundCompletionHooks | None:
return get_completion_hooks()
@dataclass
class GatewayOutboundBridge:
"""Coordinates typing refresh and at-most-one ack per inbound turn."""
adapter: _TypingAdapter
chat_id: str
loop: asyncio.AbstractEventLoop
typing_enabled: bool = field(default_factory=lambda: _env_bool("BUTLER_GATEWAY_TYPING_ENABLED", True))
typing_refresh_seconds: float = field(
default_factory=lambda: _env_float("BUTLER_GATEWAY_TYPING_REFRESH_SECONDS", 4.0)
)
ack_enabled: bool = field(default_factory=lambda: _env_bool("BUTLER_GATEWAY_PROGRESS_ACK_ENABLED", True))
ack_seconds: float = field(
default_factory=lambda: _env_float("BUTLER_GATEWAY_PROGRESS_ACK_SECONDS", 30.0)
)
max_ack_messages: int = field(default_factory=lambda: int(
os.getenv("BUTLER_GATEWAY_PROGRESS_MAX_ACK_MESSAGES", str(GATEWAY_PROGRESS_MAX_ACK_MESSAGES)) or str(GATEWAY_PROGRESS_MAX_ACK_MESSAGES)
))
_started_at: float = field(default=0.0, init=False)
_final_sent: bool = field(default=False, init=False)
_ack_sent: bool = field(default=False, init=False)
_completion_push_sent: bool = field(default=False, init=False)
_delegate_push_count: int = field(default=0, init=False)
_pending_delegate_report: Any = field(default=None, init=False)
_timeout_notified: bool = field(default=False, init=False)
_typing_task: asyncio.Task[None] | None = field(default=None, init=False)
_ack_task: asyncio.Task[None] | None = field(default=None, init=False)
_closed: bool = field(default=False, init=False)
_task_milestone_sent: bool = field(default=False, init=False)
_task_milestone_count: int = field(default=0, init=False)
_milestone_task: asyncio.Task[None] | None = field(default=None, init=False)
delegate_role: str = field(default="", init=False)
workflow_name: str = field(default="", init=False)
workflow_step: str = field(default="", init=False)
workflow_step_index: int = field(default=0, init=False)
workflow_step_total: int = field(default=0, init=False)
last_tool_name: str = field(default="", init=False)
_last_turn_elapsed: float = field(default=0.0, init=False)
_main_reply_chars: int = field(default=0, init=False)
_supplementary_scheduled: int = field(default=0, init=False)
_delegate_push_inflight: int = field(default=0, init=False)
_outbound_events: list[Any] = field(default_factory=list, init=False)
_stream_preview: str = field(default="", init=False)
_stream_chars: int = field(default=0, init=False)
workflow_step_phase: str = field(default="", init=False)
slash_single_bubble: bool = field(default=False, init=False)
scrape_urls_seen: set[str] = field(default_factory=set, init=False)
inbound_user_text: str = field(default="", init=False)
network_search_web_used: bool = field(default=False, init=False)
firecrawl_search_count: int = field(default=0, init=False)
firecrawl_agent_count: int = field(default=0, init=False)
firecrawl_feedback_count: int = field(default=0, init=False)
web_search_empty_count: int = field(default=0, init=False)
@property
def turn_started_at(self) -> float:
return self._started_at
@property
def ack_sent(self) -> bool:
return self._ack_sent
@property
def completion_push_sent(self) -> bool:
return self._completion_push_sent
@property
def delegate_push_count(self) -> int:
return self._delegate_push_count
@property
def timeout_notified(self) -> bool:
return self._timeout_notified
def set_pending_delegate_report(self, report: Any) -> None:
self._pending_delegate_report = report
def take_pending_delegate_report(self) -> Any:
report = self._pending_delegate_report
self._pending_delegate_report = None
return report
@classmethod
def for_event(
cls,
adapter: Any,
*,
chat_id: str,
loop: asyncio.AbstractEventLoop,
ensure_typing: Callable[[], Any] | None = None,
) -> GatewayOutboundBridge:
bridge = cls(adapter=adapter, chat_id=chat_id, loop=loop)
bridge._ensure_typing = ensure_typing # type: ignore[attr-defined]
return bridge
async def start_turn(self) -> None:
self._started_at = time.monotonic()
self._final_sent = False
self._ack_sent = False
self._completion_push_sent = False
self._delegate_push_count = 0
self._pending_delegate_report = None
self._timeout_notified = False
self._closed = False
self._task_milestone_sent = False
self._task_milestone_count = 0
self.delegate_role = ""
self.workflow_name = ""
self._main_reply_chars = 0
self._supplementary_scheduled = 0
self._delegate_push_inflight = 0
self.scrape_urls_seen = set()
self.inbound_user_text = ""
self.network_search_web_used = False
self.firecrawl_search_count = 0
self.firecrawl_agent_count = 0
self.firecrawl_feedback_count = 0
self.web_search_empty_count = 0
ensure = getattr(self, "_ensure_typing", None)
if callable(ensure):
await async_safe_best_effort(
ensure,
label="outbound_bridge.ensure_typing",
default=None,
)
if self.typing_enabled:
await self._safe_send_typing()
self._typing_task = asyncio.create_task(
self._typing_refresh_loop(),
name=f"butler-typing-{self.chat_id[:8]}",
)
if self.ack_enabled and self.ack_seconds > 0 and self.max_ack_messages > 0:
self._ack_task = asyncio.create_task(
self._ack_timer(),
name=f"butler-ack-{self.chat_id[:8]}",
)
def _start_milestone_timer() -> None:
if task_milestone_enabled():
self._milestone_task = asyncio.create_task(
self._milestone_timer(),
name=f"butler-milestone-{self.chat_id[:8]}",
)
safe_best_effort(
_start_milestone_timer,
label="outbound_bridge.milestone_timer",
default=None,
)
async def end_turn(self) -> None:
if self._closed:
return
self._closed = True
for task in (self._typing_task, self._ack_task, self._milestone_task):
if task and not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
self._typing_task = None
self._ack_task = None
self._milestone_task = None
if self.typing_enabled:
await self._safe_stop_typing()
def mark_final_sent(self, *, main_reply_chars: int = 0) -> None:
self._final_sent = True
self._main_reply_chars = max(0, int(main_reply_chars or 0))
if self._ack_task and not self._ack_task.done():
self._ack_task.cancel()
self._ack_task = None
def _should_suppress_redundant_completion(self, kind: str) -> bool:
if not suppress_completion_after_main_enabled():
return False
if not self._final_sent or self._main_reply_chars <= 0:
return False
return kind in ("turn", "delegate", "workflow")
def _reserve_supplementary_slot(self) -> bool:
cap = max_supplementary_per_turn()
if cap <= 0:
return False
if self._supplementary_scheduled >= cap:
return False
self._supplementary_scheduled += 1
return True
def _release_supplementary_slot(self) -> None:
if self._supplementary_scheduled > 0:
self._supplementary_scheduled -= 1
def emit_threadsafe(self, fn: Callable[[], None]) -> None:
if self._closed:
return
try:
self.loop.call_soon_threadsafe(fn)
except RuntimeError:
logger.debug("outbound bridge loop closed; drop event")
def record_outbound_event(self, event: Any) -> None:
payload = safe_best_effort(
lambda: event.to_dict() if hasattr(event, "to_dict") else dict(event),
label="outbound_bridge.event_payload",
default=None,
)
if payload is None:
return
self._outbound_events.append(payload)
if len(self._outbound_events) > 24:
self._outbound_events.pop(0)
if payload.get("kind") == "thread_item":
def _record_thread_item() -> None:
record_thread_item(payload)
safe_best_effort(
_record_thread_item,
label="outbound_bridge.record_thread_item",
default=None,
)
def recent_outbound_events(self) -> list[dict[str, Any]]:
return list(self._outbound_events)
def append_stream_preview(self, delta: str) -> None:
if self._closed or not delta:
return
if not env_truthy("BUTLER_GATEWAY_STREAM_PREVIEW", default=GATEWAY_STREAM_PREVIEW_DEFAULT):
return
def _apply() -> None:
self._stream_chars += len(delta)
combined = (self._stream_preview + delta)[-400:]
self._stream_preview = combined
self.record_outbound_event(stream_event(phase="delta", chars=len(delta)))
maybe_schedule_progressive_reply(self, combined)
self.emit_threadsafe(_apply)
def notify_delegate_start(self, role: str, *, preview: str = "") -> None:
role = str(role or "").strip()
if not role:
return
self.flush_pending_delegate_completion()
self.delegate_role = role
self.record_outbound_event(delegate_event(role, phase="start", preview=preview[:80]))
logger.info("Gateway milestone delegate_start role=%s preview=%r", role, preview[:40])
def notify_workflow_step(
self,
workflow_name: str,
step_id: str,
*,
step_index: int = 0,
step_total: int = 0,
phase: str = "start",
error: str = "",
) -> None:
def _apply() -> None:
self.workflow_name = str(workflow_name or "").strip()
self.workflow_step = str(step_id or "").strip()
self.workflow_step_index = int(step_index)
self.workflow_step_total = int(step_total)
self.workflow_step_phase = str(phase or "start").strip()
self.record_outbound_event(
workflow_event(
self.workflow_name,
self.workflow_step,
phase=self.workflow_step_phase,
step_index=self.workflow_step_index,
step_total=self.workflow_step_total,
error=error,
)
)
logger.info(
"Gateway milestone workflow_step wf=%s step=%s phase=%s (%s/%s)",
self.workflow_name,
self.workflow_step,
self.workflow_step_phase,
self.workflow_step_index,
self.workflow_step_total,
)
self.emit_threadsafe(_apply)
def on_tool_start(self, name: str, args: dict[str, Any]) -> None:
name = str(name or "").strip()
if name not in MILESTONE_TOOLS:
return
def _apply() -> None:
self.last_tool_name = name
if name == "delegate_task":
role = str((args or {}).get("role") or "").strip()
if role:
self.delegate_role = role
elif name == "run_workflow":
wf = str((args or {}).get("workflow_name") or (args or {}).get("name") or "").strip()
if wf:
self.workflow_name = wf
self.emit_threadsafe(_apply)
def on_tool_complete(self, name: str, result: str) -> None:
del name, result
def flush_pending_delegate_completion(self) -> None:
hooks = _completion_hooks()
if hooks is not None:
hooks.flush_pending_delegate_completion(self)
def notify_delegate_finished(self, report: Any) -> None:
"""Push or defer delegate AgentReport (see delegate_completion_mode)."""
hooks = _completion_hooks()
if hooks is None:
return
elapsed = time.monotonic() - self._started_at if self._started_at else 0.0
hooks.try_push_agent_report(
report,
kind="delegate",
bridge=self,
elapsed_turn_seconds=elapsed,
)
def notify_turn_timeout(self, *, timeout_seconds: float) -> None:
hooks = _completion_hooks()
if hooks is None:
return
elapsed = time.monotonic() - self._started_at if self._started_at else 0.0
hooks.try_push_turn_timeout(
self,
timeout_seconds=timeout_seconds,
elapsed_seconds=elapsed,
)
def notify_workflow_finished(self, report: Any) -> None:
hooks = _completion_hooks()
if hooks is None:
return
elapsed = time.monotonic() - self._started_at if self._started_at else 0.0
hooks.try_push_agent_report(
report,
kind="workflow",
bridge=self,
elapsed_turn_seconds=elapsed,
)
def record_turn_elapsed(self, elapsed_seconds: float) -> None:
self._last_turn_elapsed = max(0.0, float(elapsed_seconds))
def maybe_notify_turn_complete_after_reply(self) -> None:
"""Call after the main WeChat reply was sent (see platforms/base.py)."""
self.flush_pending_delegate_completion()
hooks = _completion_hooks()
if hooks is None:
return
elapsed = self._last_turn_elapsed
if elapsed <= 0 and self._started_at:
elapsed = time.monotonic() - self._started_at
hooks.try_push_turn_complete(self, elapsed_seconds=elapsed)
def schedule_supplementary_reply(self, text: str, *, kind: str = "queued") -> bool:
"""Send an extra user-visible message (e.g. drained inbound queue)."""
if self._closed or not (text or "").strip():
return False
if self._final_sent and kind in ("queued", "progressive_stream"):
return False
if not self._reserve_supplementary_slot():
logger.info(
"Gateway supplementary reply capped kind=%s chat_id=%s",
kind,
self.chat_id[:12],
)
return False
body = (text or "").strip()
body = scrub_outbound_text(body)
if len(body) > 4000:
body = body[:3997] + "..."
async def _send() -> None:
ok = await send_adapter_message(
self.adapter,
self.chat_id,
body,
label="outbound_bridge.supplementary_send",
)
if ok:
logger.info(
"Gateway supplementary reply sent kind=%s chat_id=%s len=%d",
kind,
self.chat_id[:12],
len(body),
)
else:
self._release_supplementary_slot()
if schedule_coro_threadsafe(
self.loop,
_send,
label="outbound_bridge.supplementary_schedule",
):
return True
self._release_supplementary_slot()
return False
def schedule_completion_push(self, text: str, *, kind: str) -> bool:
"""Thread-safe: send an extra completion message."""
if self._closed or not (text or "").strip():
return False
if self._should_suppress_redundant_completion(kind):
if kind == "delegate":
self.take_pending_delegate_report()
logger.info(
"Gateway completion push suppressed after main reply kind=%s",
kind,
)
return False
if kind in ("turn", "workflow") and self._completion_push_sent:
return False
if kind == "timeout" and (self._timeout_notified or self._completion_push_sent):
return False
if kind == "delegate":
hooks = _completion_hooks()
mode = hooks.delegate_completion_mode() if hooks is not None else "last"
if mode == "each":
max_each = hooks.delegate_completion_max_each() if hooks is not None else 1
inflight = self._delegate_push_count + self._delegate_push_inflight
if inflight >= max_each:
return False
self._delegate_push_inflight += 1
if not self._reserve_supplementary_slot():
if kind == "delegate" and self._delegate_push_inflight > 0:
self._delegate_push_inflight -= 1
logger.info(
"Gateway completion push capped kind=%s chat_id=%s",
kind,
self.chat_id[:12],
)
return False
body = (text or "").strip()
body = scrub_outbound_text(body)
if len(body) > 4000:
body = body[:3997] + "..."
async def _send() -> None:
hooks = _completion_hooks()
ok = False
if hooks is not None:
ok = bool(
await hooks.deliver_completion_push(
self.adapter,
self.chat_id,
body,
kind=kind,
)
)
if ok:
if kind == "delegate":
self._delegate_push_count += 1
else:
self._completion_push_sent = True
if kind == "timeout":
self._timeout_notified = True
logger.info(
"Gateway completion push sent kind=%s chat_id=%s len=%d",
kind,
self.chat_id[:12],
len(body),
)
else:
self._release_supplementary_slot()
if kind == "delegate" and self._delegate_push_inflight > 0:
self._delegate_push_inflight -= 1
if schedule_coro_threadsafe(
self.loop,
_send,
label="outbound_bridge.completion_schedule",
):
return True
self._release_supplementary_slot()
if kind == "delegate" and self._delegate_push_inflight > 0:
self._delegate_push_inflight -= 1
return False
async def maybe_send_ack(self) -> None:
if self._closed or self._final_sent or self._ack_sent:
return
if self.max_ack_messages <= 0:
return
elapsed = int(time.monotonic() - self._started_at)
text = self._build_ack_text(elapsed)
if not text:
return
text = scrub_outbound_text(text)
await send_progress_ack(self, text=text, elapsed=elapsed)
async def _milestone_timer(self) -> None:
try:
await run_milestone_timer_tick(self)
except asyncio.CancelledError:
raise
def _build_ack_text(self, elapsed: int) -> str:
eta = self._estimate_eta(elapsed)
if self.delegate_role:
return (
f"正在处理：已委派 {self.delegate_role}，完成后回复摘要。"
f"（已用 {elapsed}s{eta}）可发 /health 查看状态"
)
if self.workflow_name:
step_part = ""
if self.workflow_step:
if self.workflow_step_total > 0:
step_part = f"（{self.workflow_step}，{self.workflow_step_index}/{self.workflow_step_total}）"
else:
step_part = f"（{self.workflow_step}）"
return (
f"正在处理：工作流 {self.workflow_name} 执行中{step_part}。"
f"（已用 {elapsed}s{eta}）可发 /health 查看状态"
)
base = (
f"正在处理，请稍候（已用 {elapsed}s{eta}）。"
"完成后回复摘要；可发 /health 查看状态。"
)
preview = (self._stream_preview or "").strip()
if preview:
snippet = preview[-120:].replace("\n", " ")
return f"{base}\n…{snippet}"
return base
@staticmethod
def _estimate_eta(elapsed: int) -> str:
if elapsed < 10:
return "，预计 15-30s"
if elapsed < 30:
return "，预计还需 10-30s"
if elapsed < 60:
return "，预计还需 30-60s"
return ""
async def _typing_refresh_loop(self) -> None:
try:
await run_typing_refresh_loop(self)
except asyncio.CancelledError:
raise
async def _ack_timer(self) -> None:
try:
await asyncio.sleep(self.ack_seconds)
if not self._closed and not self._final_sent:
await self.maybe_send_ack()
except asyncio.CancelledError:
raise
async def _safe_send_typing(self) -> None:
async def _run() -> None:
await self.adapter.send_typing(self.chat_id)
await async_safe_best_effort(
_run,
label="outbound_bridge.send_typing",
default=None,
)
async def _safe_stop_typing(self) -> None:
async def _run() -> None:
await self.adapter.stop_typing(self.chat_id)
await async_safe_best_effort(
_run,
label="outbound_bridge.stop_typing",
default=None,
)
# Backward-compatible re-export of ``merge_loop_callbacks``.
#
# The implementation moved to :mod:`butler.core.loop_callbacks_merge` to fix
# audit R1-2 (core → gateway layering violation). This re-export preserves the
# historical import path ``from butler.gateway.outbound_bridge import
# merge_loop_callbacks`` for existing callers.
from butler.core.loop_callbacks_merge import merge_loop_callbacks # noqa: E402,F401
__all__ = [
"GatewayOutboundBridge",
"get_current_bridge",
"get_gateway_bridge_optional",
"max_supplementary_per_turn",
"merge_loop_callbacks",
"set_current_bridge",
"suppress_completion_after_main_enabled",
]