feat: 无状态原子化改造 v0.2.0 — pipelines/ 分层 + 双工录音 + pipx 自动更新修复

## 架构改造

将 recorder/tunnel/managed_mcp 三个有状态单例模块拆成
mindcli/pipelines/ 下的无状态管线 + health.py 调用方持有状态。
参照 hermes-overlay/infra/pipelines/anyfile2md.py 的分层模式。

### pipelines/ 层（无状态）
- audio_capture.py: capture() 工厂函数返回独立 CaptureHandle，
  无单例无互斥，双工模式（system+mic 并行）在代码层面可用
- tunnel_session.py: connect() 工厂函数 + on_status 回调，
  消除 health ⇄ tunnel 循环耦合（单向数据流）
- tool_proxy.py: ToolProxy 替代 ManagedMCP，非单例

### health.py 改造
- _active_captures dict 按 chatId 索引，可多实例并存
- _tunnel_handle 由调用方持有，on_status 回调更新状态
- /record/stop 支持 ?chatId= 停单路或全停
- /record/status 返回所有活跃录音列表

### cli.py 改造
- chat/ask 走 run_agent headless + Cloud Gateway JWT（铁律 A）
- 保留 --offline 走 vendor TUI（铁律 C：断开即自治）
- mind update 修复 pipx 场景：
  - 检测 pipx venv → pipx reinstall
  - 非 pipx → sys.executable -m pip（修复 venv 里 pip 找不到）
  - 防降级保护（远端版本低于本地时不升级）
  - 远端 upgradeCmd 字段下发

### 顺手修复
- health.py / capability.py 的 HERMES_COMMIT → VENDOR_COMMIT
- 版本号 0.1.0 → 0.2.0（__init__.py + pyproject.toml）
- 新增 versions.json 仓库模板（installCmd 改为 pipx，新增 upgradeCmd）

### 删除
- recorder.py → 逻辑迁入 pipelines/audio_capture.py
- tunnel.py → 逻辑迁入 pipelines/tunnel_session.py
- managed_mcp.py → 逻辑迁入 pipelines/tool_proxy.py

SPEC: docs/SPEC_mindcli_atomization.md

mindcli/__init__.py

@@ -11,7 +11,7 @@ POC 验证结论：sys.path.insert(0, _vendor_dir) 一行即可，
 import os
 import sys
 
-__version__ = "0.1.0"
+__version__ = "0.2.0"
 
 # ── Vendor 路径注入 ──────────────────────────────────────────
 # 将 _vendor/ 目录加入 sys.path 头部，

mindcli/capability.py

@@ -71,7 +71,7 @@ def _scan_mcp_servers() -> list[dict[str, str]]:
 
 
 def _get_vendor_commit() -> str:
-    commit_file = os.path.join(mindcli._VENDOR_DIR, "HERMES_COMMIT")
+    commit_file = os.path.join(mindcli._VENDOR_DIR, "VENDOR_COMMIT")
     try:
         with open(commit_file) as f:
             return f.read().strip()

mindcli/cli.py

@@ -1,8 +1,10 @@
 """
 Mind CLI — 命令行入口。
 
-通过 Click 定义 `mind` 命令族，内部委托给 _vendor/ 中的 Hermes CLI。
-Phase 0 只实现 chat / ask / health 三个核心命令。
+通过 Click 定义 `mind` 命令族。
+- chat/ask 走 _vendor/run_agent.py 的 headless AIAgent（铁律 A：决策权归云端）
+- LLM 调用 100% 走 Cloud Gateway（JWT 计费），不在本地拉起 vendor TUI
+- --offline 备选：走 _vendor/cli.py 的完整 TUI（铁律 C：断开即自治）
 """
 
 import click
@@ -14,6 +16,44 @@ import sys
 import mindcli  # noqa: F401 — 触发 __init__.py 的 sys.path 注入
 
 
+# ── Cloud Gateway 配置 ────────────────────────────────────
+
+def _get_cloud_gateway_url() -> str:
+    """Cloud Gateway 的 LLM API base URL。"""
+    return os.environ.get(
+        "MINDOS_GATEWAY_URL",
+        "https://agent.brainwork.club/mindos-next/llm",
+    )
+
+
+def _get_jwt() -> str | None:
+    """获取 JWT（从 Tunnel 句柄或环境变量）。
+
+    优先从 health server 的 tunnel handle 获取；
+    离线/无 tunnel 时回退到 MINDOS_JWT 环境变量。
+    """
+    jwt = os.environ.get("MINDOS_JWT")
+    if jwt:
+        return jwt
+
+    # 尝试从 health server 查询 tunnel 状态获取 JWT
+    #（JWT 存在 tunnel handle 内存中，不落盘）
+    import urllib.request
+    try:
+        req = urllib.request.Request("http://127.0.0.1:8660/tunnel/status")
+        with urllib.request.urlopen(req, timeout=2) as resp:
+            data = json.loads(resp.read())
+            if data.get("status") == "connected":
+                # tunnel 已连接，JWT 在 handle 内存中
+                # 通过环境变量 MINDOS_JWT 传递（由 /tunnel/activate 时设置）
+                return os.environ.get("MINDOS_JWT")
+    except Exception:
+        pass
+    return None
+
+
+# ── Click 命令组 ──────────────────────────────────────────
+
 @click.group(invoke_without_command=True)
 @click.version_option(version=mindcli.__version__, prog_name="mind")
 @click.pass_context
@@ -27,11 +67,79 @@ def main(ctx):
 @click.option("--model", "-m", default="", help="模型名称（默认使用配置文件）")
 @click.option("--skills", "-s", multiple=True, help="加载指定 skill")
 @click.option("--resume", "-r", default="", help="恢复指定会话 ID")
-def chat(model, skills, resume):
-    """进入交互式 Chat（复用 Hermes CLI 的完整 TUI）。"""
+@click.option("--offline", is_flag=True, default=False,
+              help="离线模式：使用 vendor TUI + 本地 LLM 配置（铁律 C：断开即自治）")
+def chat(model, skills, resume, offline):
+    """进入交互式 Chat。
+
+    默认走 Cloud Gateway（JWT 计费）。
+    --offline 走 _vendor/cli.py 的完整 TUI（本地 LLM 配置）。
+    """
+    if offline:
+        _chat_offline(model, skills, resume)
+        return
+
+    _chat_cloud(model, skills, resume)
+
+
+def _chat_cloud(model, skills, resume):
+    """走 Cloud Gateway 的交互式 chat（run_agent headless）。"""
+    jwt = _get_jwt()
+    if not jwt:
+        click.echo("❌ 未找到 JWT，请先连接 Tunnel：mind tunnel connect --token <JWT>")
+        click.echo("   或设置环境变量: export MINDOS_JWT=<your-jwt>")
+        sys.exit(1)
+
+    from run_agent import AIAgent
+
+    agent = AIAgent(
+        base_url=_get_cloud_gateway_url(),
+        api_key=jwt,
+        model=model or None,
+        enabled_toolsets=list(skills) if skills else None,
+        session_id=resume or None,
+    )
+
+    click.echo("🤖 Mind CLI (Cloud Gateway 模式)")
+    click.echo("=" * 50)
+
+    # 交互循环
+    conversation_history = []
+    system_message = None
+
+    while True:
+        try:
+            user_input = input("\n你: ").strip()
+        except (EOFError, KeyboardInterrupt):
+            click.echo("\n👋 再见！")
+            break
+
+        if not user_input:
+            continue
+        if user_input.lower() in ("exit", "quit", "q"):
+            click.echo("👋 再见！")
+            break
+
+        try:
+            result = agent.run_conversation(
+                user_message=user_input,
+                system_message=system_message,
+                conversation_history=conversation_history,
+            )
+            # 更新会话历史
+            if result.get("messages"):
+                conversation_history = result["messages"]
+            response = result.get("response", "")
+            if response:
+                click.echo(f"\n🤖 {response}")
+        except Exception as e:
+            click.echo(f"\n❌ 错误: {e}")
+
+
+def _chat_offline(model, skills, resume):
+    """离线模式：走 vendor cli.py 的完整 TUI（本地 LLM 配置）。"""
     from cli import main as hermes_main
 
-    # 构建 Hermes CLI 参数
     kwargs = {}
     if model:
         kwargs["model"] = model
@@ -48,21 +156,42 @@ def chat(model, skills, resume):
 @click.option("--model", "-m", default="", help="模型名称")
 @click.option("--format", "fmt", default="text", help="输出格式: text/json/markdown")
 def ask(question, model, fmt):
-    """单次查询（= hermes -q）。"""
-    from cli import main as hermes_main
+    """单次查询（= hermes -q）。
 
-    kwargs = {"query": question}
-    if model:
-        kwargs["model"] = model
+    走 Cloud Gateway（JWT 计费）。
+    """
+    jwt = _get_jwt()
+    if not jwt:
+        click.echo("❌ 未找到 JWT，请先连接 Tunnel：mind tunnel connect --token <JWT>")
+        sys.exit(1)
 
-    hermes_main(**kwargs)
+    from run_agent import AIAgent
+
+    agent = AIAgent(
+        base_url=_get_cloud_gateway_url(),
+        api_key=jwt,
+        model=model or None,
+    )
+
+    try:
+        result = agent.run_conversation(user_message=question)
+        response = result.get("response", "")
+
+        if fmt == "json":
+            click.echo(json.dumps({"question": question, "answer": response},
+                                  ensure_ascii=False, indent=2))
+        elif fmt == "markdown":
+            click.echo(f"## Q: {question}\n\n{response}")
+        else:
+            click.echo(response)
+    except Exception as e:
+        click.echo(f"❌ 错误: {e}", err=True)
+        sys.exit(1)
 
 
 @main.command()
 def health():
     """显示本地 CLI 健康状态。"""
-    import json
-
     status = {
         "ok": True,
         "version": mindcli.__version__,
@@ -160,6 +289,7 @@ def record_start(token, chat_id, source):
     """开始录音（独立音频源 → Cloud ASR）。
 
     双工模式下，CLI 和浏览器各自独立推送，不做混音。
+    可多次调用不同 --source 启动多路录音（双工模式）。
     """
     import urllib.request
     body = json.dumps({"token": token, "chatId": chat_id, "source": source}).encode()
@@ -182,19 +312,26 @@ def record_start(token, chat_id, source):
 
 
 @record.command(name="stop")
-def record_stop():
+@click.option("--chat-id", default="", help="停止指定 chatId 的录音（默认停止全部）")
+def record_stop(chat_id):
     """停止录音。"""
     import urllib.request
+    url = "http://127.0.0.1:8660/record/stop"
+    if chat_id:
+        url += f"?chatId={chat_id}"
     try:
         req = urllib.request.Request(
-            "http://127.0.0.1:8660/record/stop",
+            url,
             data=b"{}",
             headers={"Content-Type": "application/json"},
         )
         with urllib.request.urlopen(req, timeout=10) as resp:
             result = json.loads(resp.read())
             if result.get("ok"):
-                click.echo(f"⏹️ 录音已停止 duration={result.get('duration')}s")
+                if "stopped" in result:
+                    click.echo(f"⏹️ 已停止 {result['stopped']} 路录音")
+                else:
+                    click.echo(f"⏹️ 录音已停止 duration={result.get('duration')}s")
             else:
                 click.echo(f"❌ {result.get('error')}")
     except Exception as e:
@@ -210,7 +347,9 @@ def record_status():
         with urllib.request.urlopen(req, timeout=2) as resp:
             data = json.loads(resp.read())
             if data.get("running"):
-                click.echo(f"🎙️ 录音中 duration={data.get('duration')}s chatId={data.get('chatId')}")
+                for cap in data.get("captures", []):
+                    click.echo(f"🎙️ 录音中 duration={cap.get('duration')}s "
+                              f"chatId={cap.get('chatId')} source={cap.get('source')}")
             else:
                 click.echo("⏹️ 未在录音")
     except Exception:
@@ -225,36 +364,117 @@ def update():
     click.echo(f"当前版本: v{mindcli.__version__}")
     click.echo("正在检查最新版本...")
 
-    remoteVersion = _check_remote_version()
-    if remoteVersion and remoteVersion != mindcli.__version__:
-        click.echo(f"⬆️  发现新版本 v{remoteVersion}")
-        click.echo("正在升级...")
-        installCmd = "pip install --upgrade git+https://git.brainwork.club/lidf/MindOS_CLI.git"
-        result = subprocess.run(installCmd.split(), capture_output=True, text=True)
-        if result.returncode == 0:
-            click.echo(f"✅ 已升级到 v{remoteVersion}")
-            click.echo("   请重启 Mind CLI（mind start）以生效")
-        else:
-            click.echo(f"❌ 升级失败: {result.stderr[:200]}")
-    elif remoteVersion:
-        click.echo(f"✅ 已是最新版本 v{mindcli.__version__}")
-    else:
+    remote = _fetch_remote_info()
+    if not remote:
         click.echo("⚠️  无法检查远端版本（网络问题？）")
+        return
+
+    remoteVersion = remote.get("version")
+    if not remoteVersion:
+        click.echo("⚠️  远端版本信息异常")
+        return
+
+    if remoteVersion == mindcli.__version__:
+        click.echo(f"✅ 已是最新版本 v{mindcli.__version__}")
+        return
+
+    # 防降级：远端版本低于本地时，不升级（服务器 versions.json 可能未同步）
+    if _compare_versions(remoteVersion, mindcli.__version__) < 0:
+        click.echo(f"⚠️  本地 v{mindcli.__version__} 已高于远端 v{remoteVersion}（服务器版本未同步？）")
+        return
+
+    click.echo(f"⬆️  发现新版本 v{remoteVersion}")
+    if remote.get("releaseNotes"):
+        click.echo(f"   {remote['releaseNotes']}")
+    click.echo("正在升级...")
+
+    # 优先用远端下发的 upgradeCmd；无则本地检测安装方式
+    installCmd = remote.get("upgradeCmd") or _detect_upgrade_command()
+    click.echo(f"   执行: {installCmd}")
+
+    result = subprocess.run(
+        installCmd,
+        shell=True,  # 含引号/管道，必须 shell=True
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode == 0:
+        click.echo(f"✅ 已升级到 v{remoteVersion}")
+        click.echo("   请重启 Mind CLI（mind start）以生效")
+    else:
+        click.echo(f"❌ 升级失败: {result.stderr[:300]}")
+        click.echo("   可手动执行: pipx reinstall mindos-cli")
 
 
-def _check_remote_version():
-    """从 versions.json 获取远端 CLI 最新版本号。"""
+def _compare_versions(v1: str, v2: str) -> int:
+    """比较两个语义化版本号。返回 -1 (v1<v2), 0 (相等), 1 (v1>v2)。"""
+    def _parse(v: str):
+        parts = []
+        for p in v.strip().split("."):
+            try:
+                parts.append(int(p))
+            except ValueError:
+                parts.append(0)
+        return parts
+    a, b = _parse(v1), _parse(v2)
+    # 补齐长度
+    while len(a) < len(b):
+        a.append(0)
+    while len(b) < len(a):
+        b.append(0)
+    for x, y in zip(a, b):
+        if x < y:
+            return -1
+        if x > y:
+            return 1
+    return 0
+
+
+def _detect_upgrade_command() -> str:
+    """检测当前安装方式，返回正确的升级命令。
+
+    pipx 安装 → pipx reinstall（从 git URL 重新安装到隔离 venv）
+    pip 安装  → python -m pip install --upgrade（用当前解释器的 pip）
+    """
+    import sys
+
+    # 检测是否运行在 pipx 的 venv 中
+    # pipx venv 路径特征：包含 .local/share/pipx/venvs/<包名>
+    python_path = sys.executable
+    if "pipx/venvs" in python_path or "pipx\\venvs" in python_path:
+        return "pipx reinstall mindos-cli"
+
+    # 默认 pip 升级（用 sys.executable -m pip，避免 PATH 里找不到 pip）
+    return f"{sys.executable} -m pip install --upgrade git+https://git.brainwork.club/lidf/MindOS_CLI.git"
+
+
+def _fetch_remote_info() -> dict | None:
+    """从 versions.json 获取远端 CLI 版本信息。
+
+    返回 {"version": str, "upgradeCmd": str|None, "releaseNotes": str|None} 或 None。
+    """
     import urllib.request
     versionsUrl = "https://dl.brainwork.club/mindos-next/versions.json"
     try:
         req = urllib.request.Request(versionsUrl)
         with urllib.request.urlopen(req, timeout=5) as resp:
             data = json.loads(resp.read())
-            return data.get("cli", {}).get("version")
+            cli = data.get("cli", {})
+            return {
+                "version": cli.get("version"),
+                "upgradeCmd": cli.get("upgradeCmd"),
+                "releaseNotes": cli.get("releaseNotes"),
+            }
     except Exception:
         return None
 
 
+def _check_remote_version():
+    """从 versions.json 获取远端 CLI 最新版本号（兼容旧接口）。"""
+    info = _fetch_remote_info()
+    return info.get("version") if info else None
+
+
 def _get_vendor_commit() -> str:
     """读取 _vendor/VENDOR_COMMIT 文件获取 vendor 版本。"""
     commitFile = os.path.join(mindcli._VENDOR_DIR, "VENDOR_COMMIT")

mindcli/health.py

@@ -21,16 +21,15 @@ import mindcli
 
 logger = logging.getLogger("mindcli.health")
 
-# ── 全局状态（Phase 2 tunnel.py 会写入） ──────────────────
+# ── 运行时状态（由调用方管线通过回调更新） ──────────────
 _tunnel_status = "disconnected"
 _tool_count = 0
 
+# ── 活跃录音句柄（按 chatId 索引，可多实例并存 = 双工模式）──
+_active_captures: dict[str, "object"] = {}  # chatId → CaptureHandle
 
-def set_tunnel_status(status: str, tools: int = 0) -> None:
-    """由 tunnel.py 调用，更新隧道状态。"""
-    global _tunnel_status, _tool_count
-    _tunnel_status = status
-    _tool_count = tools
+# ── Tunnel 句柄（由 /tunnel/activate 创建，单连接）────────
+_tunnel_handle = None  # TunnelHandle | None
 
 
 # asyncio 事件循环（tunnel 需要）
@@ -47,6 +46,17 @@ def _detect_capabilities() -> list[str]:
     return caps
 
 
+def _on_tunnel_status(status: str, tools: int = 0) -> None:
+    """Tunnel 状态变更回调（由 TunnelHandle 通过 on_status 调用）。
+
+    替代旧版 tunnel.py 反向 import health.set_tunnel_status 的全局变量写入，
+    实现单向数据流：Tunnel → 回调 → health 局部状态。
+    """
+    global _tunnel_status, _tool_count
+    _tunnel_status = status
+    _tool_count = tools
+
+
 class _HealthHandler(BaseHTTPRequestHandler):
     """处理 /health 和 /tunnel/activate 请求。"""
 
@@ -64,18 +74,21 @@ class _HealthHandler(BaseHTTPRequestHandler):
             }, ensure_ascii=False)
             self._respond(200, body)
         elif self.path == "/tunnel/status":
-            from mindcli.tunnel import get_tunnel_client
-            client = get_tunnel_client()
             body = json.dumps({
-                "status": client.status,
-                "userId": client.user_id,
+                "status": _tunnel_handle.status if _tunnel_handle else "disconnected",
+                "userId": _tunnel_handle.user_id if _tunnel_handle else None,
                 "tools": _tool_count,
             })
             self._respond(200, body)
         elif self.path == "/record/status":
             try:
-                from mindcli.recorder import get_recorder
-                body = json.dumps(get_recorder().status(), ensure_ascii=False)
+                # 遍历所有活跃录音句柄（双工模式下可有多路）
+                if _active_captures:
+                    captures = [h.status() for h in _active_captures.values()]
+                else:
+                    captures = []
+                body = json.dumps({"running": len(captures) > 0, "captures": captures},
+                                  ensure_ascii=False)
                 self._respond(200, body)
             except Exception as e:
                 self._respond(500, json.dumps({"error": str(e)}))
@@ -84,11 +97,13 @@ class _HealthHandler(BaseHTTPRequestHandler):
 
     def do_POST(self):
         """处理 POST 请求。"""
-        if self.path == "/tunnel/activate":
+        # 去掉 query string 后再匹配路由（/record/stop?chatId=xxx → /record/stop）
+        path = self.path.split("?")[0]
+        if path == "/tunnel/activate":
             self._handle_tunnel_activate()
-        elif self.path == "/record/start":
+        elif path == "/record/start":
             self._handle_record_start()
-        elif self.path == "/record/stop":
+        elif path == "/record/stop":
             self._handle_record_stop()
         else:
             self._respond(404, json.dumps({"error": "Not Found"}))
@@ -107,14 +122,24 @@ class _HealthHandler(BaseHTTPRequestHandler):
                 self._respond(400, json.dumps({"error": "Missing token or tunnelUrl"}))
                 return
 
-            # 在 asyncio 事件循环中启动 tunnel
-            from mindcli.tunnel import get_tunnel_client
-            client = get_tunnel_client()
+            # 在 asyncio 事件循环中启动 tunnel（无状态管线 + 回调）
+            from mindcli.pipelines.tunnel_session import connect as tunnel_connect
 
             if _loop and _loop.is_running():
-                future = asyncio.run_coroutine_threadsafe(
-                    client.activate(token, tunnel_url), _loop
-                )
+                # 若已有旧 handle，先断开
+                global _tunnel_handle
+                async def _activate():
+                    global _tunnel_handle
+                    if _tunnel_handle:
+                        await _tunnel_handle.disconnect()
+                    _tunnel_handle = await tunnel_connect(
+                        url=tunnel_url,
+                        jwt=token,
+                        on_status=_on_tunnel_status,
+                    )
+                    return {"ok": True, "status": "connecting"}
+
+                future = asyncio.run_coroutine_threadsafe(_activate(), _loop)
                 result = future.result(timeout=5)
             else:
                 result = {"ok": True, "status": "no_event_loop"}
@@ -146,19 +171,21 @@ class _HealthHandler(BaseHTTPRequestHandler):
                     f"?token={token}&chatId={chat_id}&meetingId={meeting_id}&source=system"
                 )
 
-            from mindcli.recorder import get_recorder
-            recorder = get_recorder()
             source = data.get("source", "system")  # "system" 或 "mic"
 
             if _loop and _loop.is_running():
+                from mindcli.pipelines.audio_capture import capture as capture_audio
                 future = asyncio.run_coroutine_threadsafe(
-                    recorder.start(
+                    capture_audio(
                         ws_url=ws_base, chat_id=chat_id,
                         meeting_id=meeting_id, source=source,
                     ),
                     _loop,
                 )
-                result = future.result(timeout=10)
+                handle = future.result(timeout=10)
+                # 存入 _active_captures，按 chatId 索引（双工模式可多路并存）
+                _active_captures[chat_id or meeting_id] = handle
+                result = {"ok": True, "meetingId": handle.meeting_id, "source": source}
             else:
                 result = {"error": "事件循环未运行，请先 mind start"}
 
@@ -170,12 +197,32 @@ class _HealthHandler(BaseHTTPRequestHandler):
     def _handle_record_stop(self):
         """停止本地录音。"""
         try:
-            from mindcli.recorder import get_recorder
-            recorder = get_recorder()
+            # 支持 /record/stop?chatId=xxx 停止单路；无参则停止所有
+            from urllib.parse import urlparse, parse_qs
+            parsed = urlparse(self.path)
+            qs = parse_qs(parsed.query)
+            target_chat = qs.get("chatId", [None])[0]
 
             if _loop and _loop.is_running():
-                future = asyncio.run_coroutine_threadsafe(recorder.stop(), _loop)
-                result = future.result(timeout=10)
+                if target_chat:
+                    # 停止指定 chatId 的单路录音
+                    handle = _active_captures.pop(target_chat, None)
+                    if handle:
+                        future = asyncio.run_coroutine_threadsafe(handle.stop(), _loop)
+                        result = future.result(timeout=10)
+                    else:
+                        result = {"error": f"无活跃录音 chatId={target_chat}"}
+                elif _active_captures:
+                    # 停止所有活跃录音（全局 [停止录音]）
+                    results = []
+                    for cid in list(_active_captures.keys()):
+                        handle = _active_captures.pop(cid, None)
+                        if handle:
+                            future = asyncio.run_coroutine_threadsafe(handle.stop(), _loop)
+                            results.append(future.result(timeout=10))
+                    result = {"ok": True, "stopped": len(results), "results": results}
+                else:
+                    result = {"error": "未在录音"}
             else:
                 result = {"error": "事件循环未运行"}
 
@@ -214,7 +261,7 @@ class _ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
 
 
 def _get_vendor_commit() -> str:
-    commit_file = os.path.join(mindcli._VENDOR_DIR, "HERMES_COMMIT")
+    commit_file = os.path.join(mindcli._VENDOR_DIR, "VENDOR_COMMIT")
     try:
         with open(commit_file) as f:
             return f.read().strip()

mindcli/pipelines/__init__.py

@@ -0,0 +1,10 @@
+"""
+Mind CLI 无状态管线层。
+
+参照 hermes-overlay/infra/pipelines/anyfile2md.py 的分层模式：
+  - 管线层只做编排，不做逻辑。无状态：不写 DB、不推 SSE、不管 session。
+  - 状态化职责（生命周期管理、端口绑定、状态聚合）下沉到调用方（health.py）。
+
+每个模块的入口是工厂函数（capture / connect / execute），返回轻量句柄，
+调用方持有句柄并管理其生命周期。
+"""

mindcli/pipelines/audio_capture.py

@@ -1,5 +1,5 @@
 """
-Mind CLI — 系统拾音引擎 (Phase 4)
+Mind CLI — 系统拾音管线（无状态）。
 
 双层架构·音频服务层：
   CLI 作为音频服务层的一个独立源，只负责采集系统音频。
@@ -11,6 +11,10 @@ Mind CLI — 系统拾音引擎 (Phase 4)
 
 不做混音。混音 = 伪需求。"关联两份转写"是 Agent 层的智力工作。
 
+无状态：不持有进程级单例、不反向 import 调用方模块。
+每次 capture() 返回独立 CaptureHandle，可多实例并存（双工模式）。
+状态化职责（哪些 handle 在跑、何时 stop）由调用方（health.py）管理。
+
 Cloud 端复用 dashscope_realtime.py 管线，零新增代码。
 协议与前端 RecordingService 完全一致：
   客户端→服务端: binary PCM16 帧 / {"type":"stop"}
@@ -24,58 +28,104 @@ import threading
 import time
 from typing import Callable, Literal
 
-logger = logging.getLogger("mindcli.recorder")
+logger = logging.getLogger("mindcli.pipelines.audio_capture")
 
 # ── 常量 ──────────────────────────────────────────────────
 TARGET_SAMPLE_RATE = 16000   # Cloud ASR 要求 16kHz
 CHUNK_DURATION_MS = 100      # 每帧 100ms
 CHUNK_SAMPLES = TARGET_SAMPLE_RATE * CHUNK_DURATION_MS // 1000  # 1600
 
-# ── 全局单例 ──────────────────────────────────────────────
-_recorder: "SystemRecorder | None" = None
 
-
-def get_recorder() -> "SystemRecorder":
-    """获取全局 SystemRecorder 单例。"""
-    global _recorder
-    if _recorder is None:
-        _recorder = SystemRecorder()
-    return _recorder
-
-
-class SystemRecorder:
+async def capture(
+    ws_url: str,
+    chat_id: str = "",
+    meeting_id: str = "",
+    source: Literal["system", "mic"] = "system",
+    on_text: Callable[[str, str], None] | None = None,
+) -> "CaptureHandle":
     """
-    系统拾音引擎 — 音频服务层的一个独立源。
+    启动一路音频采集，返回独立句柄。
 
-    只负责单一音频源的采集 + WS 推送。
-    不做混音（双工模式下浏览器和 CLI 各自独立推送到 Cloud ASR）。
+    无单例、无互斥。可多次调用，每次返回独立 handle，
+    双工模式 = 两个 handle 并存（system + mic 各一路）。
+
+    Args:
+        ws_url: Cloud ASR WebSocket URL（含 token/chatId/meetingId query）
+        chat_id: 对话 ID
+        meeting_id: 录音批次 ID
+        source: "system"（ScreenCaptureKit）或 "mic"（sounddevice）
+        on_text: 收到转写文本的回调 (type, text)
+
+    Returns:
+        CaptureHandle（已启动采集）
+
+    Raises:
+        RuntimeError: WebSocket 连接失败或音频源启动失败
+    """
+    handle = CaptureHandle(
+        ws_url=ws_url,
+        chat_id=chat_id,
+        meeting_id=meeting_id or f"cli_rec_{int(time.time() * 1000)}",
+        source=source,
+        on_text=on_text,
+    )
+    await handle._start()
+    return handle
+
+
+class CaptureHandle:
+    """
+    一路音频采集的句柄。生命周期由调用方管理。
+
+    非单例——双工模式下可同时存在多个 CaptureHandle 实例，
+    各自持有独立的 WS 连接、音频缓冲区、采集资源。
     """
 
-    def __init__(self):
+    def __init__(
+        self,
+        ws_url: str,
+        chat_id: str,
+        meeting_id: str,
+        source: str,
+        on_text: Callable[[str, str], None] | None,
+    ):
+        self._ws_url = ws_url
+        self._chat_id = chat_id
+        self._meeting_id = meeting_id
+        self._source = source
+        self._on_text = on_text
+
         self._running = False
         self._ws = None
-        self._source: str = "system"
         self._start_time = 0.0
-        self._chat_id = ""
-        self._meeting_id = ""
+
         # 音频缓冲区（线程安全）
         self._audio_buf: bytearray = bytearray()
         self._buf_lock = threading.Lock()
+
         # 采集资源
         self._mic_stream = None       # sounddevice.InputStream
         self._sc_stream = None        # SCStream
         self._sc_delegate = None
+
         # 推送线程
         self._push_thread: threading.Thread | None = None
+
         # asyncio 事件循环引用（start 时保存）
         self._loop: asyncio.AbstractEventLoop | None = None
-        # 事件回调
-        self._on_text: Callable[[str, str], None] | None = None
 
     @property
     def is_running(self) -> bool:
         return self._running
 
+    @property
+    def chat_id(self) -> str:
+        return self._chat_id
+
+    @property
+    def meeting_id(self) -> str:
+        return self._meeting_id
+
     def status(self) -> dict:
         """返回当前录音状态。"""
         return {
@@ -86,57 +136,31 @@ class SystemRecorder:
             "meetingId": self._meeting_id,
         }
 
-    async def start(
-        self,
-        ws_url: str,
-        chat_id: str = "",
-        meeting_id: str = "",
-        source: Literal["system", "mic"] = "system",
-        on_text: Callable[[str, str], None] | None = None,
-    ) -> dict:
-        """
-        开始录音（单一源）。
-
-        Args:
-            ws_url: Cloud ASR WebSocket URL（含 token/chatId/meetingId query）
-            chat_id: 对话 ID
-            meeting_id: 录音批次 ID
-            source: "system"（ScreenCaptureKit）或 "mic"（sounddevice）
-            on_text: 收到转写文本的回调 (type, text)
-
-        Returns:
-            {"ok": True, "meetingId": "..."} 或 {"error": "..."}
-        """
-        if self._running:
-            return {"error": "录音已在进行中"}
-
-        self._chat_id = chat_id
-        self._meeting_id = meeting_id or f"cli_rec_{int(time.time() * 1000)}"
-        self._source = source
-        self._on_text = on_text
+    async def _start(self) -> dict:
+        """启动采集（由 capture() 调用）。"""
         self._audio_buf.clear()
         self._loop = asyncio.get_running_loop()
 
         # 1. 连接 Cloud ASR WebSocket
         try:
             import websockets
-            self._ws = await websockets.connect(ws_url)
-            logger.info("[Recorder] WS 已连接: %s", ws_url[:80])
+            self._ws = await websockets.connect(self._ws_url)
+            logger.info("[AudioCapture] WS 已连接: %s", self._ws_url[:80])
         except Exception as e:
-            logger.error("[Recorder] WS 连接失败: %s", e)
-            return {"error": f"WebSocket 连接失败: {e}"}
+            logger.error("[AudioCapture] WS 连接失败: %s", e)
+            raise RuntimeError(f"WebSocket 连接失败: {e}")
 
         # 2. 启动音频采集
         try:
-            if source == "system":
+            if self._source == "system":
                 self._start_system_audio()
             else:
                 self._start_mic()
         except Exception as e:
-            logger.error("[Recorder] 音频源 '%s' 启动失败: %s", source, e)
+            logger.error("[AudioCapture] 音频源 '%s' 启动失败: %s", self._source, e)
             await self._ws.close()
             self._ws = None
-            return {"error": f"音频源启动失败: {e}"}
+            raise RuntimeError(f"音频源启动失败: {e}")
 
         # 3. 启动推送线程
         self._running = True
@@ -147,9 +171,9 @@ class SystemRecorder:
         # 4. 启动 WS 接收协程（转写结果）
         asyncio.create_task(self._ws_recv_loop())
 
-        logger.info("[Recorder] 录音已开始 source=%s chatId=%s meetingId=%s",
-                    source, chat_id, self._meeting_id)
-        return {"ok": True, "meetingId": self._meeting_id, "source": source}
+        logger.info("[AudioCapture] 录音已开始 source=%s chatId=%s meetingId=%s",
+                    self._source, self._chat_id, self._meeting_id)
+        return {"ok": True, "meetingId": self._meeting_id, "source": self._source}
 
     async def stop(self) -> dict:
         """停止录音。"""
@@ -176,7 +200,7 @@ class SystemRecorder:
                 pass
             self._ws = None
 
-        logger.info("[Recorder] 录音已停止 duration=%.1fs", duration)
+        logger.info("[AudioCapture] 录音已停止 duration=%.1fs", duration)
         return {"ok": True, "duration": duration, "meetingId": self._meeting_id}
 
     # ── 麦克风采集（sounddevice）────────────────────────────
@@ -187,7 +211,7 @@ class SystemRecorder:
 
         def _callback(indata, frames, time_info, status):
             if status:
-                logger.debug("[Recorder] mic status: %s", status)
+                logger.debug("[AudioCapture] mic status: %s", status)
             with self._buf_lock:
                 self._audio_buf.extend(indata.tobytes())
 
@@ -199,7 +223,7 @@ class SystemRecorder:
             callback=_callback,
         )
         self._mic_stream.start()
-        logger.info("[Recorder] 麦克风已启动 @%dHz", TARGET_SAMPLE_RATE)
+        logger.info("[AudioCapture] 麦克风已启动 @%dHz", TARGET_SAMPLE_RATE)
 
     # ── 系统音频采集（ScreenCaptureKit）─────────────────────
 
@@ -244,7 +268,7 @@ class SystemRecorder:
         # 创建 delegate
         _ensure_delegate_class()
         self._sc_delegate = _SCStreamDelegate.alloc().init()
-        self._sc_delegate._recorder = self
+        self._sc_delegate._handle = self  # ← delegate 持有 handle 引用
 
         # 创建 stream
         self._sc_stream = SCStream.alloc().initWithFilter_configuration_delegate_(
@@ -272,7 +296,7 @@ class SystemRecorder:
         if error_holder[0]:
             raise RuntimeError(f"SCStream 启动失败: {error_holder[0]}")
 
-        logger.info("[Recorder] 系统音频已启动 (ScreenCaptureKit @%dHz)", TARGET_SAMPLE_RATE)
+        logger.info("[AudioCapture] 系统音频已启动 (ScreenCaptureKit @%dHz)", TARGET_SAMPLE_RATE)
 
     def _on_system_audio(self, raw_bytes: bytes):
         """系统音频回调。SCStream 输出 float32 PCM，需要转为 int16。"""
@@ -340,13 +364,13 @@ class SystemRecorder:
                     asyncio.run_coroutine_threadsafe(ws.send(chunk), self._loop)
                     send_count += 1
                 except Exception as e:
-                    logger.debug("[Recorder] WS send err: %s", e)
+                    logger.debug("[AudioCapture] WS send err: %s", e)
                     break
                 offset += frame_bytes
 
             # 每 ~5s 打一次发送统计
             if send_count > 0 and send_count % 50 == 0:
-                logger.info("[Recorder] WS sent %d frames (%.1fs)",
+                logger.info("[AudioCapture] WS sent %d frames (%.1fs)",
                             send_count, send_count * CHUNK_DURATION_MS / 1000)
 
     async def _ws_recv_loop(self):
@@ -358,16 +382,16 @@ class SystemRecorder:
                     msg_type = data.get("type", "")
                     text = data.get("text", "")
                     if msg_type in ("partial", "final") and text:
-                        logger.info("[Recorder] %s: %s", msg_type, text[:50])
+                        logger.info("[AudioCapture] %s: %s", msg_type, text[:50])
                         if self._on_text:
                             self._on_text(msg_type, text)
                     elif msg_type == "error":
-                        logger.error("[Recorder] ASR error: %s", data.get("message"))
+                        logger.error("[AudioCapture] ASR error: %s", data.get("message"))
                 except (json.JSONDecodeError, TypeError):
                     pass
         except Exception as e:
             if self._running:
-                logger.warning("[Recorder] WS recv 断开: %s", e)
+                logger.warning("[AudioCapture] WS recv 断开: %s", e)
 
 
 # ── 工具函数 ──────────────────────────────────────────────
@@ -381,7 +405,7 @@ def _sync_get_sharable_content():
 
     def _handler(content, error):
         if error:
-            logger.error("[Recorder] SCShareableContent error: %s", error)
+            logger.error("[AudioCapture] SCShareableContent error: %s", error)
         result[0] = content
         event.set()
 
@@ -406,12 +430,12 @@ def _define_sc_delegate():
     class _Delegate(NSObject):
         """接收 SCStream 音频样本的 delegate。"""
 
-        _recorder = None
+        _handle = None  # ← CaptureHandle 引用（替代旧 _recorder）
 
         def stream_didOutputSampleBuffer_ofType_(self, stream, sample_buffer, output_type):
             if output_type != SCStreamOutputTypeAudio:
                 return
-            if not self._recorder or not self._recorder.is_running:
+            if not self._handle or not self._handle.is_running:
                 return
 
             try:
@@ -422,9 +446,9 @@ def _define_sc_delegate():
                 # CMBlockBufferCopyDataBytes 返回 (OSStatus, bytes_data)
                 status, raw_data = CoreMedia.CMBlockBufferCopyDataBytes(block_buf, 0, length, None)
                 if status == 0 and raw_data:
-                    self._recorder._on_system_audio(raw_data)
+                    self._handle._on_system_audio(raw_data)
             except Exception as e:
-                logger.warning("[Recorder] SCStream sample error: %s", e, exc_info=True)
+                logger.warning("[AudioCapture] SCStream sample error: %s", e, exc_info=True)
 
     return _Delegate
 

mindcli/pipelines/tool_proxy.py

@@ -1,8 +1,11 @@
 """
-Mind CLI — Managed MCP 治理层。
+Mind CLI — 工具代理管线（无状态）。
 
 在 _vendor/tools/ 之上加白名单过滤。Cloud 审批通过的工具才能执行。
 Managed 模式下：只允许 approved_tools 列表中的工具。
+
+无状态：不持有进程级单例、不反向 import 调用方模块。
+白名单 set 由调用方传入，生命周期由调用方管理（通常是 TunnelHandle 持有）。
 """
 
 import asyncio
@@ -10,23 +13,26 @@ import logging
 import os
 import subprocess
 import sys
-from typing import Any
+import shlex
+from typing import Any, Callable
 
-logger = logging.getLogger("mindcli.managed_mcp")
+logger = logging.getLogger("mindcli.pipelines.tool_proxy")
 
 
-class ManagedMCP:
+class ToolProxy:
     """
     治理层：只暴露 Cloud 审批通过的工具。
 
     Cloud 通过 Tunnel 握手下发 approved_tools 白名单，
-    后续 tool_call 请求先过白名单检查，再委托到 _vendor/tools/ 执行。
+    后续 tool_call 请求先过白名单检查，再委托到内置执行器执行。
+
+    非单例——可被任意调用方构造和持有。
     """
 
     def __init__(self, approved_tools: list[str] | None = None):
         self._approved: set[str] = set(approved_tools or [])
         # 工具名 → 执行函数的映射
-        self._executors: dict[str, Any] = {}
+        self._executors: dict[str, Callable] = {}
         self._register_executors()
 
     def update_approved(self, tools: list[str]) -> None:
@@ -36,9 +42,9 @@ class ManagedMCP:
         added = self._approved - old
         removed = old - self._approved
         if added:
-            logger.info("[ManagedMCP] 新增审批工具: %s", added)
+            logger.info("[ToolProxy] 新增审批工具: %s", added)
         if removed:
-            logger.info("[ManagedMCP] 移除审批工具: %s", removed)
+            logger.info("[ToolProxy] 移除审批工具: %s", removed)
 
     def is_approved(self, tool_name: str) -> bool:
         """检查工具是否在白名单中。"""
@@ -56,7 +62,7 @@ class ManagedMCP:
             {"output": "...", "exit_code": 0} 或 {"error": "..."}
         """
         if not self.is_approved(tool_name):
-            logger.warning("[ManagedMCP] 工具 '%s' 未审批，拒绝执行", tool_name)
+            logger.warning("[ToolProxy] 工具 '%s' 未审批，拒绝执行", tool_name)
             return {"error": f"Tool '{tool_name}' not approved by Cloud"}
 
         executor = self._executors.get(tool_name)
@@ -67,7 +73,7 @@ class ManagedMCP:
             result = await executor(params)
             return result
         except Exception as e:
-            logger.error("[ManagedMCP] 工具 '%s' 执行异常: %s", tool_name, e)
+            logger.error("[ToolProxy] 工具 '%s' 执行异常: %s", tool_name, e)
             return {"error": f"Execution failed: {str(e)}"}
 
     def _register_executors(self) -> None:
@@ -141,7 +147,7 @@ class ManagedMCP:
         if not pattern:
             return {"error": "No pattern specified"}
 
-        cmd = f"grep -rn {_shell_quote(pattern)} {_shell_quote(path)}"
+        cmd = f"grep -rn {shlex.quote(pattern)} {shlex.quote(path)}"
         return await self._exec_terminal({"command": cmd, "timeout": 15})
 
     async def _exec_file_ops(self, params: dict) -> dict:
@@ -151,13 +157,13 @@ class ManagedMCP:
         dst = params.get("destination", "")
 
         if op == "copy":
-            cmd = f"cp -r {_shell_quote(src)} {_shell_quote(dst)}"
+            cmd = f"cp -r {shlex.quote(src)} {shlex.quote(dst)}"
         elif op == "move":
-            cmd = f"mv {_shell_quote(src)} {_shell_quote(dst)}"
+            cmd = f"mv {shlex.quote(src)} {shlex.quote(dst)}"
         elif op == "delete":
-            cmd = f"rm -rf {_shell_quote(src)}"
+            cmd = f"rm -rf {shlex.quote(src)}"
         elif op == "list":
-            cmd = f"ls -la {_shell_quote(src)}"
+            cmd = f"ls -la {shlex.quote(src)}"
         else:
             return {"error": f"Unknown operation: {op}"}
 
@@ -172,12 +178,6 @@ class ManagedMCP:
             return {"error": f"Unsupported language: {language}"}
 
         return await self._exec_terminal({
-            "command": f"{sys.executable} -c {_shell_quote(code)}",
+            "command": f"{sys.executable} -c {shlex.quote(code)}",
             "timeout": 30,
         })
-
-
-def _shell_quote(s: str) -> str:
-    """简单的 shell 参数转义。"""
-    import shlex
-    return shlex.quote(s)

mindcli/pipelines/tunnel_session.py

@@ -1,18 +1,20 @@
 """
-Mind CLI — WebSocket Tunnel 客户端。
+Mind CLI — WebSocket Tunnel 会话管线（无状态）。
 
 连接到 Cloud 端 mindcli_bridge，接收工具调用指令并在本地执行。
 采用 Browser-Donated JWT 认证：浏览器授权 CLI，CLI 不需独立认证。
+
+无状态：不持有进程级单例、不反向 import 调用方模块（health.py）。
+状态变更通过 on_status 回调通知调用方，工具调用通过 on_dispatch 回调派发。
+生命周期由调用方（health.py）管理。
 """
 
 import asyncio
 import json
 import logging
-import os
-import time
-from typing import Any
+from typing import Callable
 
-logger = logging.getLogger("mindcli.tunnel")
+logger = logging.getLogger("mindcli.pipelines.tunnel_session")
 
 # 连接状态
 DISCONNECTED = "disconnected"
@@ -20,24 +22,60 @@ CONNECTING = "connecting"
 CONNECTED = "connected"
 
 
-class TunnelClient:
+async def connect(
+    url: str,
+    jwt: str,
+    on_dispatch: Callable[[dict], None] | None = None,
+    on_status: Callable[[str, int], None] | None = None,
+) -> "TunnelHandle":
     """
-    CLI → Cloud WebSocket 隧道。
+    建立 Tunnel 连接，返回句柄。
+
+    Args:
+        url: Cloud Tunnel WebSocket URL
+        jwt: MindPass JWT（浏览器提供）
+        on_dispatch: 工具调用派发回调（可选，默认走内部 ToolProxy）
+        on_status: 状态变更回调 (status_str, tool_count)
+
+    Returns:
+        TunnelHandle（已在后台启动连接循环）
+    """
+    handle = TunnelHandle(url, jwt, on_dispatch, on_status)
+    handle._start_connect_loop()
+    return handle
+
+
+class TunnelHandle:
+    """
+    CLI → Cloud WebSocket 隧道句柄。
+
+    非单例——生命周期由调用方管理。
 
     生命周期：
-    1. 浏览器 POST /tunnel/activate → 提供 JWT + tunnelUrl
-    2. TunnelClient.connect() → WebSocket 握手 + 能力协商
-    3. 消息循环：接收 tool_call → ManagedMCP 执行 → 返回结果
+    1. 调用方调 connect() → 后台启动 _connect_loop
+    2. 握手 + 能力协商 → Cloud 下发 approved_tools → 创建 ToolProxy
+    3. 消息循环：接收 tool_call → ToolProxy 执行 → 返回结果
     4. 心跳维持 30s / 断线指数退避重连
     """
 
-    def __init__(self):
+    def __init__(
+        self,
+        url: str,
+        jwt: str,
+        on_dispatch: Callable[[dict], None] | None = None,
+        on_status: Callable[[str, int], None] | None = None,
+    ):
+        self._tunnel_url = url
+        self._jwt = jwt
+        self._on_dispatch = on_dispatch
+        self._on_status = on_status
+
         self._status = DISCONNECTED
         self._ws = None
-        self._jwt: str | None = None
-        self._tunnel_url: str | None = None
         self._user_id: str | None = None
-        self._managed_mcp = None
+        self._tool_proxy = None  # ToolProxy 实例，握手成功后创建
+
+        # 后台任务
         self._reconnect_task: asyncio.Task | None = None
         self._heartbeat_task: asyncio.Task | None = None
         self._message_task: asyncio.Task | None = None
@@ -47,9 +85,6 @@ class TunnelClient:
         self._max_reconnect_delay = 30.0
         self._reconnect_attempts = 0
 
-        # 回调：通知 health.py 更新状态
-        self._on_status_change = None
-
     @property
     def status(self) -> str:
         return self._status
@@ -58,21 +93,21 @@ class TunnelClient:
     def user_id(self) -> str | None:
         return self._user_id
 
-    def set_status_callback(self, callback) -> None:
-        """设置状态变更回调（由 health.py 注册）。"""
-        self._on_status_change = callback
-
-    def _set_status(self, status: str) -> None:
+    def _set_status(self, status: str, tools: int = 0) -> None:
         self._status = status
-        if self._on_status_change:
-            self._on_status_change(status)
+        if self._on_status:
+            self._on_status(status, tools)
+
+    def _start_connect_loop(self) -> None:
+        """在当前 event loop 中后台启动连接循环。"""
+        self._reconnect_task = asyncio.create_task(self._connect_loop())
 
     async def activate(self, jwt: str, tunnel_url: str) -> dict:
         """
-        浏览器授权激活 Tunnel。
+        更新 JWT + URL 并重新连接（由调用方在浏览器重新授权时调用）。
 
         Args:
-            jwt: MindPass JWT（浏览器提供）
+            jwt: 新的 MindPass JWT
             tunnel_url: Cloud Tunnel WebSocket URL
 
         Returns:
@@ -84,9 +119,8 @@ class TunnelClient:
         # 取消旧连接
         await self.disconnect()
 
-        # 后台启动连接
+        # 后台启动新连接
         self._reconnect_task = asyncio.create_task(self._connect_loop())
-
         return {"ok": True, "status": "connecting"}
 
     async def disconnect(self) -> None:
@@ -112,6 +146,11 @@ class TunnelClient:
         self._reconnect_delay = 1.0
         logger.info("[Tunnel] 已断开")
 
+    def update_approved(self, tools: list[str]) -> None:
+        """热更新工具白名单（Cloud 下发时调用）。"""
+        if self._tool_proxy:
+            self._tool_proxy.update_approved(tools)
+
     async def _connect_loop(self) -> None:
         """连接循环：握手 → 消息循环 → 断线重连。"""
         while True:
@@ -173,17 +212,16 @@ class TunnelClient:
         # 接收审批结果
         raw = await asyncio.wait_for(self._ws.recv(), timeout=10)
         approval = json.loads(raw)
+        approved: list[str] = []
         if approval.get("type") == "approved_tools":
             approved = approval.get("tools", [])
-            # 初始化 ManagedMCP
-            from mindcli.managed_mcp import ManagedMCP
-            self._managed_mcp = ManagedMCP(approved_tools=approved)
+            # 初始化 ToolProxy（从 pipelines.tool_proxy 导入，无状态）
+            from mindcli.pipelines.tool_proxy import ToolProxy
+            self._tool_proxy = ToolProxy(approved_tools=approved)
             logger.info("[Tunnel] 审批通过工具: %s", approved)
 
-        # 更新 health 状态
-        from mindcli.health import set_tunnel_status
-        set_tunnel_status("connected", len(approved) if approval.get("type") == "approved_tools" else 0)
-        self._set_status(CONNECTED)
+        # ★ 通过回调通知调用方状态（不反向 import health.py）
+        self._set_status(CONNECTED, len(approved))
 
     async def _run(self) -> None:
         """主消息循环：接收 Cloud 指令 → 本地执行 → 返回结果。"""
@@ -195,8 +233,8 @@ class TunnelClient:
                     await self._handle_tool_call(msg)
                 elif msg.get("type") == "approved_tools":
                     # 热更新白名单
-                    if self._managed_mcp:
-                        self._managed_mcp.update_approved(msg.get("tools", []))
+                    if self._tool_proxy:
+                        self._tool_proxy.update_approved(msg.get("tools", []))
                 elif msg.get("type") == "ping":
                     await self._ws.send(json.dumps({"type": "pong"}))
                 else:
@@ -214,10 +252,13 @@ class TunnelClient:
 
         logger.info("[Tunnel] 工具调用: %s (id=%s)", tool_name, call_id)
 
-        if not self._managed_mcp:
-            result = {"error": "ManagedMCP not initialized"}
+        if self._on_dispatch:
+            # 调用方自定义派发
+            result = self._on_dispatch(msg)
+        elif not self._tool_proxy:
+            result = {"error": "ToolProxy not initialized"}
         else:
-            result = await self._managed_mcp.execute(tool_name, tool_args)
+            result = await self._tool_proxy.execute(tool_name, tool_args)
 
         # 截断过大的输出（防止 WS 阻塞）
         output = result.get("output", "")
@@ -231,12 +272,3 @@ class TunnelClient:
             "result": result,
         }
         await self._ws.send(json.dumps(response))
-
-
-# ── 全局单例 ──────────────────────────────────────────
-_tunnel_client = TunnelClient()
-
-
-def get_tunnel_client() -> TunnelClient:
-    """获取全局 TunnelClient 实例。"""
-    return _tunnel_client