严维序
b1286aeef4
- pre_trim_multica_comment v1.0.1：新增 _find_cjk_safe_boundary 避免 split mid-rune - multica_proxy v1.1.0：新增 multica_issue_comment_add 包装函数（自动 pre-trim） - 单测 22 项（14 pre_trim + 8 integration） - 三合一报告：根因定位 + 绕路补丁 + 验证
400 lines
13 KiB
Python
400 lines
13 KiB
Python
"""
|
||
multica_proxy.py — multica CLI 调用代理
|
||
|
||
封装 multica CLI 调用，自动带缓存和限流保护。
|
||
各 Agent 心跳脚本中用 multica_proxy 替代直接 subprocess.run(["multica",...])
|
||
|
||
依赖：rate_limiter.py（CacheManager, RequestScheduler, CoordinatedPoller）
|
||
|
||
作者：陆怀瑾（COO）
|
||
日期：2026-06-23
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import subprocess
|
||
import hashlib
|
||
from typing import Any, Dict, Optional
|
||
|
||
# 确保能找到 rate_limiter
|
||
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
if _SCRIPT_DIR not in sys.path:
|
||
sys.path.insert(0, _SCRIPT_DIR)
|
||
|
||
from rate_limiter import CacheManager, RequestScheduler, CoordinatedPoller, Priority
|
||
|
||
# BIZ-104 Phase ④ pre-trim 钩子（2026-08-31 集成）
|
||
# 仅在 multica comment reply 路径上接入 pre-trim，避免全部 CLI 调用都走精简
|
||
try:
|
||
from pre_trim_multica_comment import pre_trim_for_multica_comment as _pre_trim
|
||
_PRE_TRIM_AVAILABLE = True
|
||
except ImportError:
|
||
_PRE_TRIM_AVAILABLE = False
|
||
|
||
# ============================================================================
|
||
# 全局单例
|
||
# ============================================================================
|
||
|
||
_cache = CacheManager()
|
||
_scheduler: Optional[RequestScheduler] = None
|
||
_poller: Optional[CoordinatedPoller] = None
|
||
|
||
|
||
def _get_scheduler() -> RequestScheduler:
|
||
"""获取或创建调度器单例"""
|
||
global _scheduler
|
||
if _scheduler is None:
|
||
_scheduler = RequestScheduler(rate=40/60, capacity=40, enable_cache=True)
|
||
_scheduler.start()
|
||
return _scheduler
|
||
|
||
|
||
def _get_poller() -> CoordinatedPoller:
|
||
"""获取或创建统一轮询器单例"""
|
||
global _poller
|
||
if _poller is None:
|
||
_poller = CoordinatedPoller(_get_scheduler(), poll_interval=15*60)
|
||
return _poller
|
||
|
||
|
||
# ============================================================================
|
||
# 缓存查询辅助
|
||
# ============================================================================
|
||
|
||
def _make_cache_key(cmd: list) -> str:
|
||
"""为 CLI 命令生成缓存键"""
|
||
return hashlib.md5(json.dumps(cmd, sort_keys=True).encode()).hexdigest()
|
||
|
||
|
||
def _cache_category(cmd: list) -> str:
|
||
"""根据命令推断缓存类别"""
|
||
cmd_str = " ".join(str(x) for x in cmd)
|
||
if "workboard" in cmd_str:
|
||
return "workboard"
|
||
if "config" in cmd_str or "agent" in cmd_str:
|
||
return "config"
|
||
if "wiki" in cmd_str or "knowledge" in cmd_str:
|
||
return "knowledge"
|
||
if "user" in cmd_str or "member" in cmd_str:
|
||
return "user"
|
||
return "workboard" # 默认 5 分钟
|
||
|
||
|
||
# ============================================================================
|
||
# 核心代理函数
|
||
# ============================================================================
|
||
|
||
# OpenClaw 工作区 ID（全局常量）
|
||
# 用于所有 multica CLI 调用，确保隔离会话也能正确查询
|
||
_WORKSPACE_ID = "54344e11-6bb2-4d95-a5e5-c8b075a07cea"
|
||
|
||
|
||
def _inject_workspace_id(cmd: list) -> list:
|
||
"""自动注入 workspace-id 到 multica CLI 命令"""
|
||
if len(cmd) >= 2 and cmd[0] == "multica" and "--workspace-id" not in cmd:
|
||
# 插入在命令和子命令之后、标志之前
|
||
insert_idx = 1
|
||
while insert_idx < len(cmd) and not cmd[insert_idx].startswith("--"):
|
||
insert_idx += 1
|
||
new_cmd = cmd[:insert_idx] + ["--workspace-id", _WORKSPACE_ID] + cmd[insert_idx:]
|
||
return new_cmd
|
||
return cmd
|
||
|
||
|
||
def run_multica(cmd: list, use_cache: bool = True, timeout: int = 30) -> Dict[str, Any]:
|
||
"""
|
||
执行 multica CLI 命令（带缓存和限流）
|
||
|
||
参数：
|
||
cmd: 命令列表，如 ["multica", "issue", "list", "--output", "json"]
|
||
use_cache: 是否使用缓存
|
||
timeout: 超时时间（秒）
|
||
|
||
返回：
|
||
{"success": bool, "data": Any, "from_cache": bool, "error": str|None}
|
||
"""
|
||
# 自动注入 workspace-id，确保隔离会话正确查询
|
||
cmd = _inject_workspace_id(cmd)
|
||
category = _cache_category(cmd)
|
||
|
||
# 1. 尝试从缓存获取
|
||
if use_cache:
|
||
cached = _cache.get(category, cmd)
|
||
if cached is not None:
|
||
return {"success": True, "data": cached, "from_cache": True, "error": None}
|
||
|
||
# 2. 执行 CLI 命令
|
||
try:
|
||
result = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout
|
||
)
|
||
|
||
if result.returncode != 0:
|
||
error_msg = result.stderr.strip() or f"Exit code {result.returncode}"
|
||
return {"success": False, "data": None, "from_cache": False, "error": error_msg}
|
||
|
||
# 尝试解析 JSON
|
||
try:
|
||
data = json.loads(result.stdout)
|
||
except json.JSONDecodeError:
|
||
data = result.stdout.strip()
|
||
|
||
# 3. 写入缓存
|
||
if use_cache:
|
||
_cache.set(category, cmd, data)
|
||
|
||
return {"success": True, "data": data, "from_cache": False, "error": None}
|
||
|
||
except subprocess.TimeoutExpired:
|
||
return {"success": False, "data": None, "from_cache": False, "error": f"Command timed out after {timeout}s"}
|
||
except Exception as e:
|
||
return {"success": False, "data": None, "from_cache": False, "error": str(e)}
|
||
|
||
|
||
def run_openclaw_workboard(cmd: list, use_cache: bool = True, timeout: int = 30) -> Dict[str, Any]:
|
||
"""
|
||
执行 openclaw workboard CLI 命令（带缓存）
|
||
|
||
参数同 run_multica
|
||
"""
|
||
return run_multica(cmd, use_cache=use_cache, timeout=timeout)
|
||
|
||
|
||
# ============================================================================
|
||
# 便捷函数：心跳脚本中直接替换
|
||
# ============================================================================
|
||
|
||
def multica_issue_list_my_todo(assignee_id: str) -> Dict[str, Any]:
|
||
"""
|
||
获取分配给我的待办 Issue 列表
|
||
替代: multica issue list --assignee-id <id> --status todo --output json
|
||
"""
|
||
return run_multica([
|
||
"multica", "issue", "list",
|
||
"--assignee-id", assignee_id,
|
||
"--status", "todo",
|
||
"--output", "json"
|
||
])
|
||
|
||
|
||
def multica_issue_list_in_progress() -> Dict[str, Any]:
|
||
"""
|
||
获取所有进行中的 Issue 列表（超时检测用）
|
||
替代: multica issue list --status in_progress --output json
|
||
"""
|
||
return run_multica([
|
||
"multica", "issue", "list",
|
||
"--status", "in_progress",
|
||
"--output", "json"
|
||
])
|
||
|
||
|
||
def multica_issue_get(issue_id: str) -> Dict[str, Any]:
|
||
"""
|
||
获取单个 Issue 详情
|
||
替代: multica issue get <id> --output json
|
||
"""
|
||
return run_multica([
|
||
"multica", "issue", "get",
|
||
issue_id,
|
||
"--output", "json"
|
||
])
|
||
|
||
|
||
def openclaw_workboard_list() -> Dict[str, Any]:
|
||
"""
|
||
获取 WorkBoard 卡片列表
|
||
替代: openclaw workboard list --json
|
||
"""
|
||
return run_multica([
|
||
"openclaw", "workboard", "list", "--json"
|
||
])
|
||
|
||
|
||
def openclaw_workboard_read(card_id: str) -> Dict[str, Any]:
|
||
"""
|
||
获取单个 WorkBoard 卡片
|
||
替代: openclaw workboard read <id> --json
|
||
"""
|
||
return run_multica([
|
||
"openclaw", "workboard", "read", card_id, "--json"
|
||
])
|
||
|
||
|
||
# ============================================================================
|
||
# BIZ-104 Phase ④：multica issue comment reply 集成 pre-trim
|
||
# 背景：multica 0.4.35 二进制解析 openclaw stdout 时，长 markdown 被吞。
|
||
# 绕路：所有 multica issue comment reply 路径必走 pre-trim（≤4KB）。
|
||
# ============================================================================
|
||
|
||
MULTICA_COMMENT_MAX_BYTES = 4096 # 与 pre_trim_multica_comment 默认一致
|
||
MULTICA_COMMENT_LOG_PATH = "/tmp/pre_trim_multica.log"
|
||
|
||
|
||
def multica_issue_comment_add(
|
||
issue_id: str,
|
||
content: str,
|
||
parent: Optional[str] = None,
|
||
attachment: Optional[str] = None,
|
||
content_file: Optional[str] = None,
|
||
use_pre_trim: bool = True,
|
||
max_bytes: int = MULTICA_COMMENT_MAX_BYTES,
|
||
log_path: str = MULTICA_COMMENT_LOG_PATH,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
发送 multica issue comment reply (BIZ-104 Phase ④ 集成)
|
||
|
||
参数：
|
||
issue_id: 目标 Issue ID
|
||
content: 完整 markdown 内容（会被 pre-trim）
|
||
parent: 父 comment ID（线程回复）
|
||
attachment: 附件路径
|
||
content_file: 如果提供，从文件读取 content （不会被 pre-trim，适用于大文件场景）
|
||
use_pre_trim: 是否启用 pre-trim（默认 True）
|
||
max_bytes: pre-trim 字节上限（默认 4 KB）
|
||
log_path: 原 output 落盘路径
|
||
|
||
返回：run_multica() 标准结果字典
|
||
"""
|
||
# 优先使用 content_file（外部已规范格式）
|
||
if content_file:
|
||
return run_multica([
|
||
"multica", "issue", "comment", "add",
|
||
issue_id,
|
||
"--content-file", content_file,
|
||
*([ "--parent", parent] if parent else []),
|
||
*([ "--attachment", attachment] if attachment else []),
|
||
])
|
||
|
||
# 否则对 content 字段走 pre-trim
|
||
if use_pre_trim and _PRE_TRIM_AVAILABLE:
|
||
pre = _pre_trim(content, max_bytes=max_bytes, log_path=log_path)
|
||
trimmed = pre["trimmed"]
|
||
# 提预精简元数据到 stderr 便于追踪（不入 content，避免污染评论）
|
||
if pre["truncated"]:
|
||
print(
|
||
f"[multica_proxy] BIZ-104 pre-trim 触发: "
|
||
f"{pre['original_bytes']}B → {pre['trimmed_bytes']}B, "
|
||
f"actions={pre['actions']}",
|
||
file=sys.stderr,
|
||
)
|
||
final_content = trimmed
|
||
else:
|
||
final_content = content
|
||
|
||
# --content 必须用临时文件路径（multica CLI MUL-2904）
|
||
# 写入临时文件 + 清理
|
||
import tempfile
|
||
fd, tmp_path = tempfile.mkstemp(suffix=".md", prefix="multica_comment_")
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
f.write(final_content)
|
||
return run_multica([
|
||
"multica", "issue", "comment", "add",
|
||
issue_id,
|
||
"--content-file", tmp_path,
|
||
*([ "--parent", parent] if parent else []),
|
||
*([ "--attachment", attachment] if attachment else []),
|
||
])
|
||
finally:
|
||
try:
|
||
os.remove(tmp_path)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# ============================================================================
|
||
# 缓存管理
|
||
# ============================================================================
|
||
|
||
def get_cache_stats() -> Dict[str, Any]:
|
||
"""获取缓存统计"""
|
||
return _cache.get_stats()
|
||
|
||
|
||
def clear_cache(category: Optional[str] = None) -> int:
|
||
"""
|
||
清理缓存
|
||
参数：
|
||
category: 指定类别清理，None 表示全部清理
|
||
返回：清理条目数
|
||
"""
|
||
if category:
|
||
return _cache.clear_expired()
|
||
else:
|
||
count = len(_cache._cache)
|
||
_cache.clear()
|
||
return count
|
||
|
||
|
||
# ============================================================================
|
||
# 统一轮询器（仅 COO 使用）
|
||
# ============================================================================
|
||
|
||
def start_coordinated_poller() -> CoordinatedPoller:
|
||
"""
|
||
启动 COO 统一轮询器
|
||
仅 COO Agent 调用此函数
|
||
"""
|
||
poller = _get_poller()
|
||
if not poller._running:
|
||
poller.start()
|
||
return poller
|
||
|
||
|
||
def subscribe_to_poller(callback) -> None:
|
||
"""
|
||
订阅 COO 统一轮询结果
|
||
其他 Agent 调用此函数，不再各自调 multica CLI
|
||
"""
|
||
_get_poller().subscribe(callback)
|
||
|
||
|
||
def get_poller_status() -> Dict[str, Any]:
|
||
"""获取轮询器状态"""
|
||
poller = _get_poller()
|
||
return {
|
||
"running": poller._running,
|
||
"poll_interval": poller.poll_interval,
|
||
"subscriber_count": len(poller._subscribers)
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# 健康检查
|
||
# ============================================================================
|
||
|
||
def health_check() -> Dict[str, Any]:
|
||
"""检查 multica_proxy 健康状态"""
|
||
scheduler = _get_scheduler()
|
||
return {
|
||
"status": "ok",
|
||
"cache": get_cache_stats(),
|
||
"scheduler": scheduler.get_status(),
|
||
"poller": get_poller_status()
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# 测试
|
||
# ============================================================================
|
||
|
||
if __name__ == "__main__":
|
||
print("=== multica_proxy 健康检查 ===")
|
||
print(json.dumps(health_check(), indent=2, ensure_ascii=False))
|
||
|
||
print("\n=== 测试缓存 ===")
|
||
# 第一次调用（无缓存）
|
||
result1 = run_multica(["echo", "test1"], use_cache=True)
|
||
print(f"第1次: from_cache={result1['from_cache']}")
|
||
|
||
# 第二次调用（应命中缓存）
|
||
result2 = run_multica(["echo", "test1"], use_cache=True)
|
||
print(f"第2次: from_cache={result2['from_cache']}")
|
||
|
||
print("\n测试完成")
|