butler/gateway/message_handler.py
Rama main · réplica de lectura del repositorio original.
"""Butler Gateway Message Handler.
Processes incoming messages from any platform (WeChat, Telegram, etc.)
through Butler's orchestration layer. Provides the bridge between
external gateways and Butler's AgentLoop.
Information flow:
User -> Platform Adapter -> Butler Handler -> AgentLoop -> Report Pipeline -> User
R1-6 (audit ``docs/reviews/project-deep-audit-2026-06-r1to8.md`` §R1-6):
The two god methods ``handle_message`` and ``_handle_message_locked``
have been decomposed into composable phase functions living in
:mod:`butler.gateway.message_pipelines` (pre-session phases) and
:mod:`butler.gateway.locked_phases` (in-session phases). This module
keeps the class definition, the per-session AgentLoop bookkeeping,
and a thin orchestrator that wires the phases together.
"""
from __future__ import annotations
import logging
import time as _time
from typing import Any, Optional, cast
from butler.core.agent_loop.loop import AgentLoop
from butler.core.loop_types import LoopResult
from butler.gateway.message_handler_ops import (
interrupt_session_loop_safe,
run_prequeue_interrupt_safe,
)
from butler.gateway.message_pipelines import (
_phase_resolve_session_key,
_phase_transform_inbound_text,
)
from butler.orchestrator import ButlerOrchestrator
from butler.gateway.delegate_push_dedup import gateway_inbound_guard
from butler.gateway.handler_commands import (
format_health_summary,
format_loop_response,
handle_slash_command,
)
from butler.gateway.inbound_drain import drain_queued_inbound
from butler.gateway.inbound_pipeline import build_default_inbound_pipeline
from butler.gateway.locked_turn_orchestrator import run_locked_message_turn
from butler.resilience.message_queue import message_queue_enabled
from butler.gateway.session_loop_factory import create_gateway_loop
from butler.gateway.turn_post_pipeline import run_turn_post_inbound_pipeline
from butler.utilities.env_parse import env_truthy
from butler.session.keys import chat_id_from_session_key, normalize_session_key
from butler.session.lifecycle import trigger_session_end
from butler.gateway.session_registry import GatewaySessionRegistry
logger = logging.getLogger(__name__)
class ButlerMessageHandler:
"""Handles messages from any platform through Butler's pipeline.
The handler maintains per-session AgentLoop instances and routes
messages through Butler's orchestration layer. After the R1-6
split, the two big methods (``handle_message`` and
``_handle_message_locked``) are thin orchestrators that wire
the phase functions in :mod:`message_pipelines` and
:mod:`locked_phases` together.
"""
def __init__(
self,
channel: str = "gateway",
orchestrator: Optional[ButlerOrchestrator] = None,
):
self.channel = channel
self._orchestrator = orchestrator or ButlerOrchestrator(user_id="owner", channel=channel)
self._session_registry = GatewaySessionRegistry(
self._create_loop_for_session,
finalize=self._finalize_session,
on_session_removed=_on_gateway_session_removed,
max_sessions=_env_int("BUTLER_GATEWAY_MAX_SESSIONS", 128),
idle_ttl_seconds=_env_float("BUTLER_GATEWAY_SESSION_IDLE_TTL_SECONDS", 7200),
)
self._sessions: dict[str, AgentLoop] = self._session_registry.sessions
self._health_by_session: dict[str, dict[str, Any]] = self._session_registry.health_by_session
self._inbound_pipeline = build_default_inbound_pipeline()
def _create_loop_for_session(self, session_key: str) -> AgentLoop:
return create_gateway_loop(self, session_key)
def _finalize_session(self, loop: AgentLoop) -> None:
trigger_session_end(self._orchestrator, loop, reason="finalize")
def _get_or_create_loop(self, session_key: str) -> AgentLoop:
self._session_registry.evict_idle()
return self._session_registry.get_or_create(session_key)
def resolve_session_key(
self,
*,
platform: str = "unknown",
external_id: str | None = None,
session_key: str | None = None,
) -> str:
"""Resolve ``platform:chat_id:project`` from external id and per-chat project."""
pm = self._orchestrator.project_manager
cid = str(external_id or "").strip()
if not cid and session_key:
cid = chat_id_from_session_key(session_key)
project = pm.get_project_name_for_chat(platform=platform, chat_id=cid or "default")
return cast(
str,
normalize_session_key(
platform=platform,
external_id=external_id,
session_key=session_key,
project=project,
),
)
def _should_queue_inbound(self, session_key: str, text: str) -> bool:
if not message_queue_enabled():
return False
stripped = (text or "").strip()
if not stripped or stripped.startswith("/"):
return False
return cast(bool, self._session_registry.is_session_active(session_key))
def _queue_push_via_bridge(self) -> bool:
from butler.defaults.env_defaults import GATEWAY_QUEUE_PUSH_VIA_BRIDGE_DEFAULT
return cast(bool, env_truthy("BUTLER_GATEWAY_QUEUE_PUSH_VIA_BRIDGE", default=GATEWAY_QUEUE_PUSH_VIA_BRIDGE_DEFAULT))
def _interrupt_session_loop(self, session_key: str) -> None:
interrupt_session_loop_safe(self._sessions, session_key)
def _format_prequeue_interrupt_reply(self, session_key: str) -> str:
"""Sprint 16 TST-10-5 第八批: 抽自 handle_message pre-dispatch hook (原 line 422-424).
中断 in-flight session loop, 返 "已请求停止" 提示. 防御性 try/except 包裹 interrupt 调用,
即使 interrupt 抛异常也返消息 (logger.debug), 不应阻断 caller.
"""
run_prequeue_interrupt_safe(
lambda: self._interrupt_session_loop(session_key),
log_debug=logger.debug,
)
return "已请求停止当前会话任务（含进行中的委派）。"
def _drain_queued_inbound(
self,
session_key: str,
*,
platform: str,
external_id: str | None,
primary_reply: str = "",
) -> str:
return cast(
str,
drain_queued_inbound(
self,
session_key,
platform=platform,
external_id=external_id,
primary_reply=primary_reply,
),
)
def _recover_registry_if_stale(self) -> None:
"""Clear a stuck ``reset_all`` flag that would block ``enter_session`` forever."""
reg = self._session_registry
with reg._lock:
if (
reg._resetting_all
and not reg._active_sessions
and reg._pending_session_entries == 0
):
logger.warning("Recovering stale gateway reset_all flag")
reg._resetting_all = False
reg._reset_condition.notify_all()
def handle_message(
self,
text: str,
*,
session_key: str | None = None,
platform: str = "unknown",
external_id: str | None = None,
) -> str:
"""Process an incoming message and return the response.
R1-6: this method is now a thin orchestrator that wires the
pre-session phase functions from
:mod:`butler.gateway.message_pipelines` together. The body is
reduced from ~334 non-blank lines to a linear flow of phase
calls; each phase encapsulates one inbound guard, transform,
or admission step.
"""
_t0 = _time.monotonic()
self._recover_registry_if_stale()
logger.info(
"Gateway handle_message start platform=%s external_id=%s preview=%r",
platform,
external_id,
(text or "")[:80],
)
session_key = _phase_resolve_session_key(
self, platform=platform, external_id=external_id, session_key=session_key,
)
text = _phase_transform_inbound_text(
text, platform=platform, external_id=external_id, session_key=session_key,
)
if not text.strip():
return ""
chat_id = str(external_id or "").strip()
with gateway_inbound_guard(chat_id):
return cast(
str,
run_turn_post_inbound_pipeline(
self,
text,
session_key=session_key,
platform=platform,
external_id=external_id,
t0=_t0,
),
)
def _handle_message_after_pipeline(
self,
text: str,
*,
session_key: str | None,
platform: str,
external_id: str | None,
_t0: float,
) -> str:
"""Backward-compatible alias; prefer :func:`turn_post_pipeline.run_turn_post_inbound_pipeline`."""
return cast(
str,
run_turn_post_inbound_pipeline(
self,
text,
session_key=session_key,
platform=platform,
external_id=external_id,
t0=_t0,
),
)
def _handle_message_locked(
self,
text: str,
*,
session_key: str = "default",
platform: str = "unknown",
external_id: str | None = None,
) -> str:
"""In-session pipeline orchestrator (ENG-3 → locked_turn_orchestrator)."""
return cast(
str,
run_locked_message_turn(
self,
text,
session_key=session_key,
platform=platform,
external_id=external_id,
),
)
def last_health_summary(self, session_key: str = "default") -> dict[str, Any]:
"""Return the latest best-effort runtime diagnostics for a session."""
return cast(dict[str, Any], self._session_registry.get_health(session_key))
def _handle_command(
self,
text: str,
*,
session_key: str = "default",
platform: str = "unknown",
external_id: str | None = None,
) -> Optional[str]:
return cast(
str | None,
handle_slash_command(
self,
text,
session_key=session_key,
platform=platform,
external_id=external_id,
),
)
def _format_health_summary(self, session_key: str = "default") -> str:
return cast(str, format_health_summary(self, session_key))
def _format_response(self, result: LoopResult, platform: str) -> str:
return cast(str, format_loop_response(result, platform))
from butler.gateway.handler_helpers import ( # noqa: F401, E402
_normalize_switch_request,
_normalize_status_request,
_normalize_new_session_request,
_normalize_detail_request,
_normalize_memo_request,
_normalize_contacts_request,
_normalize_expense_request,
_normalize_habits_request,
_gateway_run_callbacks,
_is_prequeue_interrupt_command,
_env_int,
_env_float,
_is_sessionless_command,
apply_auto_continue_rewrite,
_tool_audit_summary,
_reset_tool_audit_events,
_maybe_welcome_prefix,
_build_project_overview,
_inject_previous_session_summary,
_on_gateway_session_removed,
_WELCOMED_SESSIONS,
_WELCOME_TEXT,
)