butler/gateway/handler_helpers.py
Rama main · réplica de lectura del repositorio original.
"""Gateway message handler helper functions.
Extracted from message_handler.py. Contains text normalization,
command classification, project overview building, and session
management utilities.
"""
from __future__ import annotations
import logging
import os
import re
import threading
from pathlib import Path
from typing import Any, Callable, TYPE_CHECKING, cast
from butler.tools._file_cache import read_json_cached
from butler.core.loop_types import LoopCallbacks
from butler.core.auto_continue import resolve_auto_continue_user_message
from butler.core.best_effort import safe_best_effort
from butler.defaults.env_defaults import BUTLER_HOME_DEFAULT, ONBOARDING_WELCOME_DEFAULT
from butler.utilities.env_parse import env_truthy, float_env, int_env
from butler.gateway.command_registry import lookup
from butler.gateway.outbound_bridge import get_current_bridge
from butler.mcp.registry_hook import disconnect_mcp_session
from butler.runtime.service import list_jobs_status, runtime_enabled
from butler.tools.registry import get_tool_audit_events, reset_tool_audit_events
from butler.tools.reminder import _load_all
if TYPE_CHECKING:
from butler.core.agent_loop.loop import AgentLoop
logger = logging.getLogger(__name__)
# EXT-2 / 外部 SaaS：含这些词时不走 Butler 多项目 /总览、/项目 归一化
_EXTERNAL_INTEGRATION_MARKERS = (
"todoist",
"notion",
"linear",
"trello",
"asana",
)
def _mentions_external_integrations(text: str) -> bool:
lower = (text or "").strip().lower()
return any(marker in lower for marker in _EXTERNAL_INTEGRATION_MARKERS)
def _normalize_switch_request(text: str) -> str | None:
"""Map「切换到 <项目>」to ``/切换 <项目>`` (WeChat natural phrasing)."""
stripped = (text or "").strip().rstrip("。.!！?？")
_switch_prefixes = ("切换到", "切换至", "切换去", "切换回")
if stripped.startswith("切换") and not any(
stripped.startswith(p) for p in _switch_prefixes
):
name = stripped[len("切换") :].strip().rstrip("。.!！?？")
if name and not name.startswith("/"):
return f"/切换 {name}"
for prefix in (
"切换到",
"切换至",
"切换去",
"切换回",
"现在切到",
"现在切回",
"切到",
"切回",
"把项目切回去",
"把项目切回",
):
if stripped.startswith(prefix):
name = stripped[len(prefix) :].strip().rstrip("。.!！?？")
if name:
return f"/切换 {name}"
if stripped.startswith("切回去"):
name = stripped[len("切回去") :].strip().rstrip("。.!！?？")
if name:
return f"/切换 {name}"
return None
def _normalize_status_request(text: str) -> str | None:
"""Map project-status questions to ``/状态``."""
stripped = (text or "").strip().rstrip("？?").strip()
purpose_hints = (
"做什么",
"干什么",
"用途",
"目的",
"介绍",
"是干嘛",
"干啥",
"干什么用",
"用来做什么",
"用来干什么",
)
if any(hint in stripped for hint in purpose_hints):
return None
if stripped in (
"当前在哪个项目",
"当前是什么项目",
"当前什么项目",
"现在在哪个项目",
"现在在什么项目",
"当前项目是什么",
"确认一下当前项目",
"报一下当前项目",
):
return "/状态"
if any(
phrase in stripped
for phrase in ("哪个项目", "什么项目", "当前项目", "现在在哪个")
) and any(
hint in stripped
for hint in ("当前", "现在", "哪个", "什么", "确认", "报", "我是")
):
if any(
verb in stripped
for verb in ("读", "读取", "列出", "写", "删", "委派", "检查", "README", "文件")
):
return None
return "/状态"
if any(
phrase in stripped
for phrase in ("有哪些项目", "项目列表", "哪些workspace", "几个项目")
):
if _mentions_external_integrations(stripped):
return None
return "/项目"
if any(
phrase in stripped
for phrase in ("总览", "全部项目状态", "所有项目", "项目仪表盘", "dashboard")
):
if _mentions_external_integrations(stripped):
return None
return "/总览"
return None
def _normalize_new_session_request(text: str) -> str | None:
"""Allow ``/新对话`` / ``/新会话`` with trailing natural-language hints."""
stripped = (text or "").strip()
if stripped == "/新对话" or stripped.startswith("/新对话"):
return "/新对话"
if stripped == "/新会话" or stripped.startswith("/新会话"):
return "/新对话"
return None
def _normalize_memo_request(text: str) -> str | None:
"""Map natural phrases about memos to ``/备忘``."""
stripped = (text or "").strip().rstrip("。.!！?？")
if not stripped:
return None
if stripped in ("查看备忘", "我的备忘", "备忘录", "备忘列表", "看看备忘",
"有什么备忘", "备忘有哪些"):
return "/备忘"
return None
def _normalize_contacts_request(text: str) -> str | None:
"""Map natural phrases about contacts to ``/通讯录``."""
stripped = (text or "").strip().rstrip("。.!！?？")
if not stripped:
return None
if stripped in ("通讯录", "联系人", "我的通讯录", "查看通讯录", "联系人列表"):
return "/通讯录"
return None
def _normalize_expense_request(text: str) -> str | None:
"""Map natural phrases about expenses to ``/记账``."""
stripped = (text or "").strip().rstrip("。.!！?？")
if not stripped:
return None
if stripped in ("记账", "账单", "看看账单", "这个月花了多少", "本月支出",
"本月账单", "收支", "我的账单"):
return "/记账"
return None
def _normalize_habits_request(text: str) -> str | None:
"""Map natural phrases about habits to ``/打卡``."""
stripped = (text or "").strip().rstrip("。.!！?？")
if not stripped:
return None
if stripped in ("打卡", "习惯", "今日打卡", "我的习惯", "习惯打卡",
"看看打卡", "打卡情况"):
return "/打卡"
return None
def _normalize_detail_request(text: str) -> str | None:
"""Map WeChat-friendly「详细」to ``/详细`` without requiring a slash prefix."""
stripped = (text or "").strip()
if not stripped:
return None
lower = stripped.lower()
if "报错" in stripped or "错误信息" in stripped:
return None
for marker, cmd in (("/详细", "/详细"), ("/detail", "/detail")):
idx = lower.find(marker)
if idx >= 0:
rest = stripped[idx + len(marker) :].strip().rstrip("，,。.!！?？")
return f"{cmd} {rest}".strip() if rest else cmd
detail_aliases = {
"/详细",
"/detail",
"详细",
"detail",
"查看详细",
"看详细",
"完整报告",
"详细信息",
"查看详细信息",
"看一下详细",
"看一下详细信息",
"我要看一下详细信息",
"看详细信息",
}
if lower in detail_aliases or stripped in detail_aliases:
return "/详细"
for prefix in ("详细", "详细信息", "看一下详细"):
if stripped.startswith(prefix) and len(stripped) > len(prefix):
rest = stripped[len(prefix) :].strip()
if rest.startswith("信息"):
rest = rest[2:].strip()
if rest:
return "/详细 " + rest
if stripped.startswith("详细 "):
return "/详细 " + stripped[3:].strip()
if lower.startswith("detail "):
return "/detail " + stripped[7:].strip()
return None
def _gateway_run_callbacks() -> LoopCallbacks | None:
bridge = get_current_bridge()
if bridge is None:
return None
def _on_stream_delta(chunk: str) -> None:
bridge.append_stream_preview(chunk)
return LoopCallbacks(
on_tool_start=bridge.on_tool_start,
on_tool_complete=bridge.on_tool_complete,
on_stream_delta=_on_stream_delta,
)
def _is_prequeue_interrupt_command(text: str) -> bool:
stripped = (text or "").strip().lower()
if not stripped.startswith("/"):
return False
cmd = stripped.split(maxsplit=1)[0]
return cmd in {
"/stop",
"/停止",
"/中断",
"/cancel-loop",
"/停止循环",
"/stoploop",
}
def apply_auto_continue_rewrite(session_key: str, text: str) -> str | None:
"""Sprint 16 TST-10-5 第八批: 抽自 message_handler.py:426-433 pre-dispatch rewriter.
如果用户输入是 continue marker 且有 fresh pending, 返回改写后的 text (供后续 dispatch);
否则返 None (text 不变). 异常吞掉 + logger.debug, 与原 inline try/except 行为一致.
注意: 这是 text rewriter, **不**是 slash dispatch handler — 返回值是 "新 text" 而非 reply 字符串.
"""
def _run() -> str | None:
return cast(str | None, resolve_auto_continue_user_message(session_key, text))
return cast(
str | None,
safe_best_effort(
_run,
label="handler_helpers.auto_continue",
default=None,
),
)
def _env_int(name: str, default: int) -> int:
return cast(int, int_env(name, default))
def _env_float(name: str, default: float) -> float:
return cast(float, float_env(name, default))
def _is_sessionless_command(text: str) -> bool:
"""Decide whether *text* names a sessionless slash command.
Sprint 18-4 D CRITICAL-2: 真源 = command_registry. 已注册命令(含 aliases)
立即识别, 动态注册的新命令零漂移. 未注册命令返 False (降级到 session
队列后 LLM, 与原硬编码 set 中"未注册"成员的实际行为对齐 — 那些命令
在 dispatch 失败后同样 fallback 到 LLM).
行为变化: 之前 27 项已注册但不在 set 的命令 (新 dispatch 命令) 现在
sessionless, 立即响应; 之前 45 项在 set 但未注册的命令现在等 session
队列, 仅延迟差异 (最终 LLM 接收, 功能不变).
"""
stripped = (text or "").strip()
if not stripped.startswith("/"):
return False
parts = stripped.split(maxsplit=1)
cmd = parts[0].lower()
if not cmd:
return False
return lookup(cmd) is not None
def _tool_audit_summary(session_key: str) -> dict[str, Any]:
def _run() -> dict[str, Any]:
events = get_tool_audit_events(limit=50, session_key=session_key)
failed = [event for event in events if not event.get("ok", False)]
codes = sorted({str(event.get("code")) for event in failed if event.get("code")})
return {"total": len(events), "failed": len(failed), "codes": codes}
return safe_best_effort(
_run,
label="handler_helpers.tool_audit_summary",
default={"total": 0, "failed": 0, "codes": []},
) or {"total": 0, "failed": 0, "codes": []}
def _reset_tool_audit_events(session_key: str | None = None) -> None:
def _run() -> None:
reset_tool_audit_events(session_key)
safe_best_effort(_run, label="handler_helpers.reset_tool_audit", default=None)
_WELCOMED_SESSIONS: set[str] = set()
_WELCOMED_LOCK = threading.Lock()
_WELCOME_TEXT = """你好，我是 Butler 管家（莎丽）。
建议三步上手：
1️⃣ /切换 <项目名> — 选定当前项目（附摘要）
2️⃣ 只读一句 — 例：「读取 README 前 10 行并摘要」
3️⃣ 试一次委派 — 例：「交给开发代理：只读检查 docs/ 目录」
日常：/简报 · /帮助（五个说法：查改批记管）
首周 playbook：docs/guides/owner-first-week-2026-06.md
分工与七步剧本：/分工 · /帮助"""
# User already asked for identity/capabilities — LLM reply supersedes static welcome.
_WELCOME_SUPERSEDED_PATTERNS: tuple[re.Pattern[str], ...] = (
re.compile(r"介绍.{0,8}(自己|你)|自我介绍", re.I),
re.compile(r"你是谁|你叫什么|什么名字|who\s+are\s+you", re.I),
re.compile(r"(可以|能|会).{0,12}(干什么|做什么|帮我|什么)", re.I),
re.compile(r"(什么|哪些).{0,8}(能力|功能|命令)", re.I),
re.compile(r"^/?帮助$", re.I),
)
def _user_turn_supersedes_welcome(user_text: str) -> bool:
text = (user_text or "").strip()
if not text:
return False
return any(pat.search(text) for pat in _WELCOME_SUPERSEDED_PATTERNS)
def _mark_session_welcomed(session_key: str) -> None:
"""Persist first-contact marker without emitting welcome text."""
with _WELCOMED_LOCK:
if session_key in _WELCOMED_SESSIONS:
return
_WELCOMED_SESSIONS.add(session_key)
def _persist() -> None:
marker = Path(os.getenv("BUTLER_HOME", BUTLER_HOME_DEFAULT)).expanduser() / "welcomed_sessions.txt"
if marker.is_file():
known = set(marker.read_text(encoding="utf-8").strip().splitlines())
if session_key in known:
return
marker.parent.mkdir(parents=True, exist_ok=True)
with open(marker, "a", encoding="utf-8") as f:
f.write(session_key + "\n")
safe_best_effort(_persist, label="handler_helpers.mark_welcomed", default=None)
def _maybe_welcome_prefix(session_key: str, user_text: str = "") -> str:
"""Return welcome text for first-time sessions, empty string otherwise."""
if os.getenv("BUTLER_ONBOARDING_WELCOME", ONBOARDING_WELCOME_DEFAULT).strip() == "0":
return ""
if _user_turn_supersedes_welcome(user_text):
_mark_session_welcomed(session_key)
return ""
with _WELCOMED_LOCK:
if session_key in _WELCOMED_SESSIONS:
return ""
_WELCOMED_SESSIONS.add(session_key)
marker = Path(os.getenv("BUTLER_HOME", BUTLER_HOME_DEFAULT)).expanduser() / "welcomed_sessions.txt"
def _known_sessions() -> set[str]:
if not marker.is_file():
return set()
return set(marker.read_text(encoding="utf-8").strip().splitlines())
if session_key in (safe_best_effort(
_known_sessions,
label="handler_helpers.welcome_known",
default=set(),
) or set()):
return ""
def _persist() -> None:
marker.parent.mkdir(parents=True, exist_ok=True)
with open(marker, "a", encoding="utf-8") as f:
f.write(session_key + "\n")
safe_best_effort(_persist, label="handler_helpers.welcome_marker", default=None)
return _WELCOME_TEXT
def _safe_overview_sub(fn: Callable[[], str | None], label: str) -> str | None:
"""Run a `_build_project_overview` sub-info extractor; swallow + log on failure."""
return cast(
str | None,
safe_best_effort(fn, label=f"handler_helpers.overview.{label}", default=None),
)
def _todos_subinfo(ws: Path) -> str | None:
todos_path = ws / ".butler" / "todos.json"
if not todos_path.is_file():
return None
todos_data = read_json_cached(todos_path)
if not isinstance(todos_data, list):
todos_data = []
pending = [t for t in todos_data if t.get("status") == "pending"]
if pending:
return f"待办 {len(pending)} 项"
return None
def _jobs_subinfo(ws: Path, project_name: str) -> str | None:
jobs_path = ws / "runtime" / "jobs.yaml"
if not jobs_path.is_file():
return None
if not runtime_enabled():
return None
job_rows = list_jobs_status(project_name)
if job_rows:
return f"定时任务 {len(job_rows)} 个"
return None
def _summary_subinfo(ws: Path) -> str | None:
summary_path = ws / ".butler" / "session_summary.json"
if not summary_path.is_file():
return None
summary = read_json_cached(summary_path)
if not isinstance(summary, dict):
summary = {}
turns = summary.get("turns", 0)
if turns:
return f"上次会话 {turns} 轮"
return None
def _build_project_overview(orchestrator: Any, session_key: str) -> str:
"""Build a multi-project dashboard for /总览."""
pm = orchestrator.project_manager
projects = pm.list_projects()
if not projects:
return "暂无项目。"
current = pm.resolve_active_project_name(session_key=session_key)
lines = ["📋 项目总览", ""]
for p in sorted(projects, key=lambda x: x.name):
mark = "▸ " if p.name == current else " "
header = f"{mark}{p.name}"
if p.name == current:
header += "（当前）"
sub: list[str] = []
workspace = getattr(p, "workspace", None)
if workspace:
from pathlib import Path
ws = Path(workspace)
for label, msg in (
("todos", _safe_overview_sub(lambda: _todos_subinfo(ws), "todos")),
("jobs", _safe_overview_sub(lambda: _jobs_subinfo(ws, p.name), "jobs")),
("summary", _safe_overview_sub(lambda: _summary_subinfo(ws), "summary")),
):
if msg:
sub.append(msg)
pack = getattr(p, "pack", "") or ""
ptype = p.type or ""
desc_parts = [ptype]
if pack:
desc_parts.append(f"pack={pack}")
if sub:
desc_parts.append("｜".join(sub))
lines.append(f"{header} [{', '.join(desc_parts)}]")
if p.description:
lines.append(f" {p.description[:60]}")
lines.append("")
lines.append("提醒：")
def _reminders() -> str:
reminders = [r for r in _load_all() if r.get("status") == "pending"]
if reminders:
rows = [f" 待触发提醒 {len(reminders)} 个"]
for r in reminders[:3]:
rows.append(f" · {r.get('due_human', '')} — {r.get('message', '')[:40]}")
return "\n".join(rows)
return " 无待触发提醒"
reminder_block = safe_best_effort(
_reminders,
label="handler_helpers.reminders",
default=" 提醒系统不可用",
)
lines.append(reminder_block or " 提醒系统不可用")
return "\n".join(lines)
def _inject_previous_session_summary(loop: "AgentLoop", project: Any) -> None:
"""Inject previous session summary into a new AgentLoop for context continuity."""
def _run() -> None:
if not env_truthy("BUTLER_SESSION_SUMMARY", default=True):
return
if project is None:
return
workspace = getattr(project, "workspace", None)
if not workspace:
return
from pathlib import Path
summary_path = Path(workspace) / ".butler" / "session_summary.json"
if not summary_path.is_file():
return
raw = read_json_cached(summary_path)
if not isinstance(raw, dict):
return
parts: list[str] = []
proj_name = raw.get("project", "")
turns = raw.get("turns", 0)
if proj_name:
parts.append(f"上次会话项目：{proj_name}，{turns} 轮对话")
for field, label in [
("persona", "用户偏好"),
("preference", "设置变更"),
("experience", "经验记录"),
]:
items = raw.get(field) or []
if items:
parts.append(f"{label}：" + "；".join(str(i) for i in items[:5]))
if not parts:
return
context = "[上次会话摘要]\n" + "\n".join(parts) + "\n[/上次会话摘要]"
if hasattr(loop, "_messages"):
loop._messages.append({"role": "system", "content": context})
logger.debug("Injected previous session summary (%d chars)", len(context))
safe_best_effort(_run, label="handler_helpers.session_summary_inject", default=None)
def _on_gateway_session_removed(session_key: str) -> None:
_reset_tool_audit_events(session_key)
def _disconnect() -> None:
disconnect_mcp_session(session_key)
safe_best_effort(_disconnect, label="handler_helpers.mcp_disconnect", default=None)