diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4422413..6c9e6d0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -65,8 +65,9 @@ jobs: python-version: ${{ matrix.python-version }} # 测试用 stdlib unittest, 无需安装任何依赖 + # unittest discover 自动发现 tests/ 下所有 test_*.py - name: Run tests - run: python3 tests/test_projection_differ.py -v + run: python3 -m unittest discover -s tests -p 'test_*.py' -v perms: name: Executable bit check diff --git a/README.md b/README.md index 97af244..f2fb0ca 100644 --- a/README.md +++ b/README.md @@ -116,26 +116,73 @@ zcode --prompt "继续" --resume sess_xxxx |----------|:----:| | initialize / session/new / session/prompt / session/cancel | ✅ | | session/list / session/resume | ✅ | -| tool_call / tool_call_update（工具调用展示）| ✅ | +| tool_call / tool_call_update（工具调用展示）| ✅ 实时 | | usage_update（token 用量）| ✅ | -| agent_message_chunk（文本输出）| ✅ | -| agent_thought_chunk（思考过程）| ⚠️ 代码就位，GLM-5.2 不产数据 | +| agent_message_chunk（文本输出）| ✅ **真流式**（0.14.8+）| +| agent_thought_chunk（思考过程）| ✅ 流式（GLM-5-Turbo）| | plan（任务清单）| ⚠️ 代码就位，数据驱动 | | diff（文件变更）| ⚠️ 仅文件名，无 diff 内容 | +#### 双模式自动降级 + +ACP bridge 支持**事件驱动**（真流式）和**轮询**（伪流式）两种模式，自动选择： + +- **事件驱动模式**（ZCode CLI ≥ 0.14.8）：通过 `session/subscribe` 订阅事件流，`model.streaming` 事件携带 `text_delta` 逐段推送，实现**真正的流式文本输出**。工具调用状态也实时推送（scheduled → started → progress → result）。 +- **轮询模式**（旧版 CLI 或订阅失败）：自动降级为轮询 `session/read`，turn 完成后整段发文本（伪流式）。 + +#### 扩展方法（非标准 ACP） + +ACP bridge 额外暴露了 ZCode 新版（0.14.8+）的协议方法，供编辑器/脚本调用： + +| 扩展方法 | 作用 | params | +|----------|------|--------| +| `session/fork` | 从 checkpoint 分叉新会话 | `{sessionId, target?}` | +| `session/rewind` | 回退工作区文件到 checkpoint | `{sessionId, target?, expectedRevision?}` | +| `session/goal` | 读取/设置 session 目标 | `{sessionId, action: show\|set\|replace\|clear, objective?}` | +| `session/compact` | 压缩对话上下文 | `{sessionId}` | +| `session/steer` | turn 进行中追加指令 | `{sessionId, content}` | + ## 会话存储 `--prompt` 和 ACP bridge **共享同一套会话存储**（`~/.zcode/cli/db/db.sqlite`），互通互恢复： - `--prompt` 创建的会话可被 `session/list` 看到、被 `--resume` 或 ACP resume 恢复 - 反之亦然 +## 版本兼容性 + +本项目兼容以下 ZCode 版本，**对旧版完全向后兼容**： + +| ZCode CLI 版本 | 支持情况 | ACP bridge 流式 | 扩展方法 | +|:--------------:|:--------:|:---------------:|:--------:| +| **0.14.8+**（App 3.1.4） | ✅ 完整 | **真流式**（事件驱动） | ✅ fork/rewind/goal/compact/steer | +| **0.14.5 ~ 0.14.7** | ✅ 兼容 | 伪流式（自动降级轮询） | ❌（旧版协议未实现） | +| **< 0.14.5** | ⚠️ 未测 | — | — | + +**降级行为**（自动，无需手动配置）： +- ACP bridge 检测到 `session/subscribe` 不可用时，自动切换到轮询 `session/read`（伪流式） +- 扩展方法（fork/rewind/goal/compact/steer）在旧版 ZCode 上会透传后端错误（`-32603 zcode failed: ...`），不影响标准 ACP 方法（new/prompt/cancel/list/resume） + +## Skill（驱动说明书） + +项目附带一个通用 skill [zcode-bridge-guide](skills/zcode-bridge-guide/SKILL.md)，覆盖： +- 三种接入模式（CLI / ACP / MCP）的完整使用方法 +- 凭证配置、非交互 shell 坑 +- 真流式 vs 伪流式双模式说明 +- 扩展协议方法（fork/rewind/goal/compact/steer） +- 任务书模板、独立复核、版本兼容性 + +复制到 ZCode 的 skill 目录即可使用： +```bash +cp -r skills/zcode-bridge-guide ~/.zcode/skills/ +``` + ## ⚠️ 重要限制（请务必阅读） 本项目是建立在**闭源 ZCode** 之上的非官方桥接器，存在以下固有限制： 1. **依赖闭源软件**：必须先安装 ZCode。ZCode 升级可能随时破坏本项目（协议字段靠逆向确认，无官方保证）。 2. **工具调用 turn 不稳定**：ZCode app-server 的工具调用 turn 时长在 38s～100s+ 波动，有时不完成。 -3. **流式是伪流式**：ZCode 是快照式架构（turn 期间不推送中间事件），ACP bridge 的流式文本是 turn 完成后整段发。 +3. **流式输出**：ZCode CLI ≥ 0.14.8 支持事件推送（`session/subscribe`），ACP bridge 在此版本下实现**真流式**（逐段推送）；旧版自动降级为伪流式（turn 完成后整段发）。 4. **diff 无内容**：ZCode 协议层不暴露 oldText/newText，只能列文件名。 5. **GLM-5.2 无推理输出**：思考过程（agent_thought_chunk）在 GLM-5.2 下不触发，需 GLM-5-Turbo。 6. **TUI 不可用**：`zcode` 无参数直接运行（TUI 模式）在独立终端报错（缺 `@zcode/tui` 模块），仅 headless 模式可用。 @@ -151,8 +198,11 @@ zcode-open-bridge/ │ └── acp-bridge/ # ACP 桥接 (experimental) ├── shared/ │ └── credentials.py # 凭证读取 (单一真相源) +├── skills/ +│ └── zcode-bridge-guide/ # 驱动 ZCode 的通用 skill (说明书) ├── tests/ -│ └── test_projection_differ.py +│ ├── test_projection_differ.py +│ └── test_event_translator.py ├── LICENSE # MIT └── README.md ``` diff --git a/packages/acp-bridge/zcode-acp-bridge b/packages/acp-bridge/zcode-acp-bridge index 2539b5e..aa7ac79 100755 --- a/packages/acp-bridge/zcode-acp-bridge +++ b/packages/acp-bridge/zcode-acp-bridge @@ -4,7 +4,7 @@ zcode-acp-bridge — 把 headless ZCode 桥接为 ACP (Agent Client Protocol) Ag 让 Zed / JetBrains 等 ACP 兼容编辑器能通过标准 ACP 协议调用 zcode (GLM-5.2)。 -工作原理: +工作原理 (双模式自动选择): ACP Client (编辑器) zcode-acp-bridge zcode app-server │ │ │ │ ACP initialize │ │ @@ -15,35 +15,37 @@ zcode-acp-bridge — 把 headless ZCode 桥接为 ACP (Agent Client Protocol) Ag │ {sessionId} │◀─────────────────────────────│ │◀─────────────────────────────│ │ │ │ │ - │ ACP session/prompt │ zcode session/send (不等) │ + │ ACP session/prompt │ zcode session/subscribe │ │─────────────────────────────▶│─────────────────────────────▶│ - │ │ {accepted} │ - │ │◀─────────────────────────────│ - │ │ │ - │ │ 监听 state.updated │ - │ │◀─────────────────────────────│ - │ │ 直到 status:idle │ - │ │ │ - │ │ zcode session/messages │ + │ │ register listener │ + │ │ zcode session/send │ │ │─────────────────────────────▶│ - │ │ (拿完整回复文本) │ + │ │ │ + │ │ session/event 推送: │ + │ │ model.streaming → 流式文本 │ + │ │ tool.updated → 工具状态 │ + │ │ turn.completed → 收尾 │ │ │◀─────────────────────────────│ │ │ │ - │ session/update notification │ │ - │ (agent_message_chunk) │ │ + │ session/update (流式) │ │ + │ agent_message_chunk (逐段) │ │ + │ tool_call / tool_call_update│ │ │◀─────────────────────────────│ │ │ session/prompt result │ │ │ {stopReason: end_turn} │ │ │◀─────────────────────────────│ │ -MVP 范围 (阶段1): + 若 session/subscribe 不可用 (旧版 CLI <0.14.8): 自动降级为轮询 session/read, + turn 完成后整段发文本 (伪流式)。 + +能力范围: ✅ initialize 握手 ✅ session/new → session/create - ✅ session/prompt → session/send + 监听 + 返回 - ✅ 文本输出 (非流式, turn 结束一次性发) - ✅ session/cancel 处理 - ❌ 流式文本 (阶段2) - ❌ 工具调用展示 (阶段2) + ✅ session/prompt → 事件驱动真流式 (0.14.8+) / 轮询伪流式 (旧版降级) + ✅ 流式文本 (model.streaming → agent_message_chunk 逐段) + ✅ 工具调用展示 (tool.updated → tool_call/tool_call_update 实时) + ✅ session/cancel → session/stop + ✅ 扩展方法: fork/rewind/goal/compact/steer (0.14.8+) 协议: ACP (JSON-RPC 2.0 over stdio) 对外; ZCode Protocol (无 jsonrpc 字段) 对内 依赖: 仅 Python3 标准库 @@ -52,7 +54,6 @@ MVP 范围 (阶段1): import json import os import queue -import select import subprocess import sys import threading @@ -100,7 +101,13 @@ def load_zcode_credentials(): # ZCode app-server 的 client 封装 # ============================================================ class ZCodeBackend: - """管理 zcode app-server 子进程, 提供异步消息读取""" + """管理 zcode app-server 子进程, 后台 reader 线程分发消息。 + + 消息分三种, 由后台 _reader_loop 读取并路由: + - 响应 (有 id): → _response_queues[id] 的 Queue + - session/event 推送: → 当前 _event_listener 的 Queue + - 其他 notification: → _notification_queue + """ def __init__(self): env = {**os.environ, **load_zcode_credentials()} @@ -109,47 +116,139 @@ class ZCodeBackend: stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1, env=env, ) - # _lock 同时保护 send 和 readline, 防止多线程读写竞态 - self._lock = threading.Lock() + # send 与 close 用同一把锁保护子进程管道 + self._send_lock = threading.Lock() + # 响应路由: msg_id → Queue; request() 注册后读, 收到后清理 + self._response_queues = {} + self._resp_lock = threading.Lock() + # 通用 notification 队列 (非 session/event 的 notification) + self._notification_queue = queue.Queue() + # 当前事件监听器: {zcode_sid: EventStreamListener} + self._event_listeners = {} + self._listeners_lock = threading.Lock() + # reader 线程健康标志: 退出后设 True, request() 快速失败而非慢速假死 + self._reader_dead = False + self._reader_stop = False + self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True) + self._reader_thread.start() log(f"已启动 zcode app-server (pid={self.proc.pid})") + def _reader_loop(self): + """后台线程: 持续读 zcode stdout, 按消息类型分发到对应队列。""" + while not self._reader_stop: + try: + line = self.proc.stdout.readline() + if not line: + break # 进程关闭 + line = line.strip() + if not line: + continue + try: + msg = json.loads(line) + except json.JSONDecodeError: + continue + method = msg.get("method") + msg_id = msg.get("id") + if msg_id is not None and method is None: + # 响应: 有 id 无 method → 路由到对应 Queue + with self._resp_lock: + q = self._response_queues.pop(msg_id, None) + if q is not None: + q.put(msg) + else: + # 没人等: 放通用队列兜底 + self._notification_queue.put(msg) + elif method == "session/event": + # 事件推送: 路由到对应会话的 listener + ev = msg.get("params", {}) + sid = ev.get("sessionId") + with self._listeners_lock: + listener = self._event_listeners.get(sid) + if listener is not None: + listener.handle_event(ev) + else: + # 没人监听: 忽略 (订阅是按需的) + pass + elif method is not None: + # 其他 notification (如 state.updated) → 通用队列 + self._notification_queue.put(msg) + else: + # 带有 id 和 method 的消息 (不常见), 当响应处理 + with self._resp_lock: + q = self._response_queues.pop(msg_id, None) + if q is not None: + q.put(msg) + except Exception as e: + log(f"reader 线程异常: {e}") + break + # reader 退出: 设标志, 后续 request() 快速失败而非慢速假死 + self._reader_dead = True + # 清理所有等待中的响应队列, 让它们立即得到错误 + with self._resp_lock: + for q in self._response_queues.values(): + q.put({"error": {"message": "zcode backend reader exited"}}) + self._response_queues.clear() + log("reader 线程退出 (backend 已标记为 dead)") + + def register_event_listener(self, zcode_sid, listener): + """注册事件监听器, 之后该 sid 的 session/event 会路由到它。""" + with self._listeners_lock: + self._event_listeners[zcode_sid] = listener + + def unregister_event_listener(self, zcode_sid): + """注销事件监听器。""" + with self._listeners_lock: + self._event_listeners.pop(zcode_sid, None) + def send(self, msg): """发一条消息给 zcode (无 jsonrpc 字段)""" - with self._lock: + with self._send_lock: self.proc.stdin.write(json.dumps(msg, ensure_ascii=False) + "\n") self.proc.stdin.flush() def readline(self, timeout=2.0): - """非阻塞读一行, 超时返回 None。加锁避免与并发读竞态""" - with self._lock: - r, _, _ = select.select([self.proc.stdout], [], [], timeout) - if r: - line = self.proc.stdout.readline() - return line.strip() if line else None + """[兼容旧调用] 从通用 notification 队列读一条消息。 + + 注: reader 线程模式下, 所有响应走 request(), 所有 notification 走队列。 + 旧代码中 readline 直接读 stdout 的用法已被 request() 取代。 + """ + try: + return self._notification_queue.get(timeout=timeout) + except queue.Empty: return None def request(self, msg_id, method, params=None, timeout=30): - """同步请求: 发送并等待对应 id 的响应, 收集期间的所有 notification""" + """同步请求: 注册响应队列, 发送, 等待对应 id 的响应。 + + 期间到达的其他 notification 由 reader 线程异步分发到各自的队列, + 不会被本方法吞掉 (修复旧版 request 与事件推送的竞态)。 + """ if params is None: params = {} - notifications = [] - self.send({"id": msg_id, "method": method, "params": params}) - import time - t0 = time.time() - while time.time() - t0 < timeout: - line = self.readline(0.5) - if line: - try: - d = json.loads(line) - if d.get("id") == msg_id: - return d, notifications # 响应 + 期间的 notifications - elif "method" in d and "id" not in d: - notifications.append(d) # notification - except json.JSONDecodeError: - pass - return {"error": {"message": "timeout"}}, notifications + # reader 已死: 快速失败, 不让调用方卡满 timeout + if self._reader_dead: + return {"error": {"message": "zcode backend reader exited (backend dead)"}}, [] + # 注册响应队列 + resp_q = queue.Queue() + with self._resp_lock: + self._response_queues[msg_id] = resp_q + try: + self.send({"id": msg_id, "method": method, "params": params}) + except (BrokenPipeError, OSError) as e: + with self._resp_lock: + self._response_queues.pop(msg_id, None) + self._reader_dead = True + return {"error": {"message": f"zcode backend pipe broken: {e}"}}, [] + try: + resp = resp_q.get(timeout=timeout) + return resp, [] # notifications 已由 reader 分发, 不再在这里收集 + except queue.Empty: + with self._resp_lock: + self._response_queues.pop(msg_id, None) + return {"error": {"message": "timeout"}}, [] def close(self): + self._reader_stop = True try: self.proc.terminate() self.proc.wait(timeout=3) @@ -172,23 +271,245 @@ class TurnMonitor: self.bridge = bridge # 用于 _next_id / _drain_inbox def poll_once(self): - """发一次 session/read, 返回 projection dict 或 None。""" + """发一次 session/read, 返回 projection dict 或 None。 + + 注: reader 线程模式下用 request() 拿响应 (不再直接 readline)。 + """ poll_id = self.bridge._next_id() - self.backend.send({"id": poll_id, "method": "session/read", - "params": {"sessionId": self.zcode_sid}}) - poll_t0 = time.time() - while time.time() - poll_t0 < 3: - line = self.backend.readline(0.5) - if line: - try: - d = json.loads(line) - if d.get("id") == poll_id: - return d.get("result", {}).get("projection", {}) - except json.JSONDecodeError: - pass + resp, _ = self.backend.request(poll_id, "session/read", + {"sessionId": self.zcode_sid}, timeout=5) + if "error" not in resp: + return resp.get("result", {}).get("projection", {}) return None +# ============================================================ +# EventStreamListener: 订阅 zcode 事件流 (session/subscribe + session/event) +# ============================================================ +# 新版 ZCode (CLI 0.14.8+) 增加了事件推送机制: +# session/subscribe(deliveryKind=desktop-continuous) 开启推送 +# 之后服务器主动发 {method:"session/event", params:{type, payload, seq, ...}} +# 流式文本通过 model.streaming 事件 (kind=text_delta) 逐段推送, 不再等 turn 结束。 +class EventStreamListener: + """订阅 zcode 事件流, 把 session/event 推送入队供主循环消费。 + + 用法: + listener = EventStreamListener(backend, zcode_sid) + snapshot = listener.subscribe() # 订阅, 返回初始快照 (或 None 降级) + if snapshot is not None: + backend.register_event_listener(zcode_sid, listener) + # ... 主循环里 listener.poll_event() 取事件 ... + backend.unregister_event_listener(zcode_sid) + """ + + def __init__(self, backend, zcode_sid): + self.backend = backend + self.sid = zcode_sid + self.last_seq = 0 # seq 游标 + self.subscribed = False + self._event_queue = queue.Queue() + + def subscribe(self, msg_id): + """调 session/subscribe 拿初始快照 + eventSeq 水位线。 + + 返回: snapshot dict (含 projection/messages) 或 None (失败, 调用方应降级轮询)。 + """ + resp, _ = self.backend.request(msg_id, "session/subscribe", { + "sessionId": self.sid, + "deliveryKind": "desktop-continuous", + "includeSnapshot": True, + "afterSeq": 0, + }, timeout=10) + if "error" in resp: + log(f" session/subscribe 失败 (降级轮询): {resp['error'].get('message', '')}") + return None + result = resp.get("result", {}) + self.last_seq = result.get("eventSeq", 0) + self.subscribed = True + log(f" ✓ session/subscribe 成功 (eventSeq={self.last_seq})") + return result.get("snapshot") + + def handle_event(self, event): + """被 ZCodeBackend._reader_loop 调用: 把 session/event params 入队。""" + seq = event.get("seq", self.last_seq) + if seq > self.last_seq: + self.last_seq = seq + self._event_queue.put(event) + + def poll_event(self, timeout=0.5): + """取一条事件, 超时返回 None。""" + try: + return self._event_queue.get(timeout=timeout) + except queue.Empty: + return None + + +# ============================================================ +# EventTranslator: 把 zcode 事件 → ACP 事件 dict (复用 ProjectionDiffer 的 kind) +# ============================================================ +# 实测 zcode 事件类型 (session/event params.type): +# turn.started → 标记 turn 启动 +# model.streaming → 流式文本/推理增量 (payload.kind=text_delta/reasoning_delta) +# tool.updated → 工具调用 (payload.kind=scheduled/started/progress/result/batch) +# turn.completed → turn 完成 (payload.usage/tokenCount/resultType) +# turn.failed → turn 失败 +class EventTranslator: + """把 zcode session/event 翻译成 ACP 事件 dict 列表。 + + 输出格式与 ProjectionDiffer.diff() 一致 (kind 标识), 可直接喂给 _dispatch_event。 + """ + + # zcode tool 名 → ACP ToolKind (复用 ProjectionDiffer 的映射) + TOOL_KIND_MAP = { + "Bash": "execute", "bash": "execute", + "Edit": "edit", "Write": "edit", "edit": "edit", "write": "edit", + "Read": "read", "read": "read", + "Grep": "search", "grep": "search", "Glob": "search", + "WebFetch": "fetch", "WebSearch": "fetch", "web_search": "fetch", + } + + def __init__(self): + self.turn_started = False + self.turn_done = False + self.turn_result_type = None # success/cancelled/error_... + self._seen_tool_ids = set() # 已通知过的 toolCallId (scheduled) + self._tool_names = {} # call_id → tool_name (result 事件不带 toolName) + + def translate(self, event): + """翻译一条 zcode event → ACP 事件 dict 列表 (可能为空)。 + + Args: + event: session/event 的 params (含 type, payload, seq)。 + Returns: + list of event dict (kind 格式, 同 ProjectionDiffer)。 + """ + etype = event.get("type", "") + payload = event.get("payload") or {} + results = [] + + if etype == "turn.started": + self.turn_started = True + log(" [event] turn.started") + + elif etype == "model.streaming": + results.extend(self._translate_streaming(payload)) + + elif etype == "tool.updated": + results.extend(self._translate_tool(payload)) + + elif etype == "turn.completed": + self.turn_done = True + self.turn_result_type = payload.get("resultType", "success") + results.extend(self._translate_turn_done(payload)) + log(f" [event] turn.completed (resultType={self.turn_result_type})") + + elif etype == "turn.failed": + self.turn_done = True + self.turn_result_type = payload.get("resultType", "error") + log(f" [event] turn.failed (resultType={self.turn_result_type})") + + # 其他类型 (session.updated/titleUpdated 等) 不产出 ACP 事件 + return results + + def _translate_streaming(self, payload): + """model.streaming → TextDelta / ReasoningDelta (逐段增量)。""" + results = [] + kind = payload.get("kind", "") + delta = payload.get("delta", "") + if not delta: + return results + if kind == "text_delta": + results.append({"kind": "TextDelta", "text": delta}) + elif kind == "reasoning_delta": + results.append({"kind": "ReasoningDelta", "text": delta}) + return results + + def _translate_tool(self, payload): + """tool.updated → ToolCallNew / ToolCallUpdate。""" + results = [] + tkind = payload.get("kind", "") + call_id = payload.get("toolCallId", "") + tool_name = payload.get("toolName", "") + + if tkind == "scheduled": + if call_id and call_id not in self._seen_tool_ids: + self._seen_tool_ids.add(call_id) + self._tool_names[call_id] = tool_name + acp_kind = self.TOOL_KIND_MAP.get(tool_name, "execute") + results.append({ + "kind": "ToolCallNew", + "call_id": call_id, + "tool": tool_name or "unknown", + "acp_kind": acp_kind, + "status": "pending", + }) + + elif tkind == "started": + if call_id: + results.append({ + "kind": "ToolCallUpdate", + "call_id": call_id, + "status": "in_progress", + }) + + elif tkind == "progress": + if call_id: + # progress 事件含 stdoutTail/stderrTail (实时输出) + output = payload.get("stdoutTail", "") or payload.get("stderrTail", "") + results.append({ + "kind": "ToolCallUpdate", + "call_id": call_id, + "status": "in_progress", + "output": output, + }) + + elif tkind == "result": + if call_id: + output = payload.get("result", "") + results.append({ + "kind": "ToolCallUpdate", + "call_id": call_id, + "status": "completed", + "output": str(output)[:2000] if output else "", + }) + + elif tkind == "error": + if call_id: + results.append({ + "kind": "ToolCallUpdate", + "call_id": call_id, + "status": "failed", + "output": payload.get("error", ""), + }) + + elif tkind == "batch": + # batch: 多个工具一起完成。payload 含 toolCallIds[] + successCount/errorCount。 + # 实测 zcode 通常也会逐个发 result, 但作为兜底: 若 batch 里有未完成的 call_id, + # 补发 completed, 防止工具在编辑器里永远卡在 in_progress。 + batch_ids = payload.get("toolCallIds", []) + error_count = payload.get("errorCount", 0) + for bid in batch_ids: + # 只补发 scheduled 过的 (避免给未知 id 发幽灵事件) + if bid in self._seen_tool_ids: + results.append({ + "kind": "ToolCallUpdate", + "call_id": bid, + "status": "failed" if error_count > 0 else "completed", + }) + return results + + def _translate_turn_done(self, payload): + """turn.completed → UsageDelta (最终用量)。""" + results = [] + usage = payload.get("usage") or {} + token_count = payload.get("tokenCount", 0) + used = usage.get("totalTokens") or token_count or 0 + size = usage.get("contextWindow", 0) + if size > 0 or used > 0: + results.append({"kind": "UsageDelta", "used": used, "size": size}) + return results + + # ============================================================ # ProjectionDiffer: 对比两次快照, 输出事件列表 # ============================================================ @@ -433,6 +754,19 @@ class ACPBridge: if method == "session/resume": return self._on_session_resume(msg_id, req.get("params", {})) + # ===== 扩展方法 (非标准 ACP, 透传 zcode 新版协议) ===== + # 这些方法不在 ACP baseline 中, 是 zcode-specific 的能力暴露。 + if method == "session/fork": + return self._on_session_fork(msg_id, req.get("params", {})) + if method == "session/rewind": + return self._on_session_rewind(msg_id, req.get("params", {})) + if method == "session/goal": + return self._on_session_goal(msg_id, req.get("params", {})) + if method == "session/compact": + return self._on_session_compact(msg_id, req.get("params", {})) + if method == "session/steer": + return self._on_session_steer(msg_id, req.get("params", {})) + # 未实现的 method if method and msg_id is not None: return self._error(msg_id, -32601, f"Method not supported: {method}") @@ -536,6 +870,141 @@ class ACPBridge: log(f"session/resume → {target_sid}") return {"jsonrpc": "2.0", "id": msg_id, "result": {}} + # ---------- 扩展方法 (非标准 ACP, 透传 zcode 新版协议 0.14.8+) ---------- + # 以下方法不在 ACP baseline 中, 暴露 zcode-specific 能力。 + # 编辑器/脚本可通过这些方法调用 fork/rewind/goal/compact/steer。 + + def _on_session_fork(self, msg_id, params): + """扩展 session/fork → zcode session/fork: 从 checkpoint 分叉新会话。 + + params: {sessionId, target?} + target: 默认 {kind:"latestCheckpoint"}; 或 {kind:"checkpoint", checkpointId} + """ + self.ensure_backend() + acp_sid = params.get("sessionId") + zcode_sid = self.session_map.get(acp_sid) or acp_sid + zc_params = {"sessionId": zcode_sid} + # target 默认 latestCheckpoint; 显式 checkpointId 则构造 {kind, checkpointId} + if params.get("checkpointId"): + zc_params["target"] = {"kind": "checkpoint", "checkpointId": params["checkpointId"]} + else: + zc_params["target"] = params.get("target", {"kind": "latestCheckpoint"}) + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/fork", zc_params, timeout=15) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode fork failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + new_sid = result.get("sessionId") + if new_sid: + self.session_map[new_sid] = new_sid + log(f"session/fork → {new_sid or '?'}") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + def _on_session_rewind(self, msg_id, params): + """扩展 session/rewind → zcode session/rewind: 回退工作区文件到 checkpoint。 + + params: {sessionId, target?, expectedRevision?} + target: 默认 {kind:"latestCheckpoint"}; 或 {kind:"checkpoint", checkpointId} + """ + self.ensure_backend() + acp_sid = params.get("sessionId") + zcode_sid = self.session_map.get(acp_sid) or acp_sid + zc_params = {"sessionId": zcode_sid} + if params.get("checkpointId"): + zc_params["target"] = {"kind": "checkpoint", "checkpointId": params["checkpointId"]} + else: + zc_params["target"] = params.get("target", {"kind": "latestCheckpoint"}) + if params.get("expectedRevision") is not None: + zc_params["expectedRevision"] = params["expectedRevision"] + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/rewind", zc_params, timeout=15) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode rewind failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + log("session/rewind → ok") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + def _on_session_goal(self, msg_id, params): + """扩展 session/goal → zcode session/goal: 读取或设置 session 目标。 + + params: {sessionId, action?, objective?} + - action 默认 "show" (读取); "set" 设置; "replace" 替换; "clear" 清除; "pause"/"resume" 暂停/恢复 + - objective: 目标文本 (action=set/replace 时必填) + + 注意: action=set/replace 会启动内部 AI turn (异步), 必须等待其完成 + 否则后续调用会报 "Cannot manage goals while a prompt is running"。 + """ + self.ensure_backend() + acp_sid = params.get("sessionId") + zcode_sid = self.session_map.get(acp_sid) or acp_sid + action = params.get("action", "show") + zc_params = {"sessionId": zcode_sid, "action": action} + if action in ("set", "replace") and params.get("objective") is not None: + zc_params["objective"] = params["objective"] + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/goal", zc_params, timeout=15) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode goal failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + # set/replace 启动内部 turn, 等待其 prompt lock 释放 (用 goal show 探测) + # 实测 goal set 的内部 AI turn 耗时 10-45s+ 不等, 必须等锁释放 + # 否则后续 show/compact 会报 "prompt is running" + if action in ("set", "replace"): + released = self._wait_for_turn_idle(zcode_sid, timeout=60, probe_method="session/goal") + if released: + log(f"session/goal action={action} → ok (内部 turn lock 已释放)") + else: + log(f"session/goal action={action} → ok (⚠ lock 等待超时, 后续 show/compact 可能暂不可用)") + else: + log(f"session/goal action={action} → ok") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + def _on_session_compact(self, msg_id, params): + """扩展 session/compact → zcode session/compact: 压缩对话上下文。 + + params: {sessionId} + 注: compact 本身是同步的 (返回 snapshot), 不启动内部 turn。 + 但若前一个 goal set 的内部 turn 仍在跑, compact 会报 "prompt is running"。 + goal set handler 已等待 lock 释放, 所以正常流程不会撞此问题。 + """ + self.ensure_backend() + acp_sid = params.get("sessionId") + zcode_sid = self.session_map.get(acp_sid) or acp_sid + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/compact", + {"sessionId": zcode_sid}, timeout=30) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode compact failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + log("session/compact → ok") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + def _on_session_steer(self, msg_id, params): + """扩展 session/steer → zcode session/steer: turn 进行中追加指令。 + + params: {sessionId, content} + 注意: steer 只在 turn 运行中有效; 无活动 turn 时会返回错误。 + """ + self.ensure_backend() + acp_sid = params.get("sessionId") + zcode_sid = self.session_map.get(acp_sid) or acp_sid + content = params.get("content", "") + if not content.strip(): + return self._error(msg_id, -32602, "steer 需要 content") + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/steer", + {"sessionId": zcode_sid, "content": content}, timeout=15) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode steer failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + log(f"session/steer → kind={result.get('kind', '?')}") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + @staticmethod def _to_iso(ts): """zcode 的毫秒时间戳 → ISO 8601 字符串 (ACP SessionInfo.updatedAt 期望)""" @@ -547,6 +1016,54 @@ class ACPBridge: except Exception: return None + def _wait_for_turn_idle(self, zcode_sid, timeout=45, probe_method=None): + """等待 zcode 会话的内部 turn 完成后 lock 释放。 + + session/goal(set)、session/compact 等方法会启动内部 AI turn, + 它们在 request() 返回后仍在后台运行。projection.status=idle 并不 + 意味着内部 prompt lock 已释放 (实测 goal set 后 status 很快变 idle, + 但 "prompt is running" lock 持续 10-25s)。因此可靠的方式是: + 重试一个探测调用直到不再报 "while a prompt is running"。 + + Args: + zcode_sid: zcode 会话 ID + timeout: 最大等待秒数 + probe_method: 可选, 探测用的 zcode 方法 (如 "session/goal" show)。 + 若为 None 则用 session/read 轮询 status。 + Returns: + True 若 lock 已释放; False 超时。 + """ + t0 = time.time() + if probe_method: + # 用探测调用重试, 直到不再报 "prompt is running" + while time.time() - t0 < timeout: + probe_id = self._next_id() + resp, _ = self.backend.request(probe_id, probe_method, + {"sessionId": zcode_sid, "action": "show"}, + timeout=10) + if "error" not in resp: + log(f" [probe] lock 已释放 (elapsed {round(time.time()-t0,1)}s)") + return True # lock 已释放 + err_msg = resp.get("error", {}).get("message", "") + # timeout 或 "prompt is running" 都继续等; 其他错误视为释放 + if "prompt is running" in err_msg or "timeout" in err_msg.lower(): + time.sleep(2) + continue + log(f" [probe] 非锁错误, 视为释放: {err_msg[:50]}") + return True + log(f" [probe] 等待超时 ({timeout}s), lock 可能仍占用") + return False + else: + # fallback: 轮询 status + monitor = TurnMonitor(self.backend, zcode_sid, self) + while time.time() - t0 < timeout: + proj = monitor.poll_once() + if proj and proj.get("status") == "idle": + time.sleep(1) + return True + time.sleep(0.5) + return False + def _on_session_prompt(self, msg_id, params): """ACP session/prompt → zcode session/send + 监听 state.updated + 返回 @@ -593,48 +1110,173 @@ class ACPBridge: differ.mark_seen(baseline_msgs) chunk_msg_id = f"msg_{uuid.uuid4().hex[:12]}" - # 发送给 zcode (立即返回 accepted, 不等) - zc_id = self._next_id() - self.backend.send({"id": zc_id, "method": "session/send", - "params": {"sessionId": zcode_sid, "content": prompt_text}}) - # 消费 send 响应并校验 (修复 P1.2: 失败要返回错误, 不能默默继续) - send_accepted = False - send_t0 = time.time() - while time.time() - send_t0 < 5: - line = self.backend.readline(1.0) - if line: - try: - d = json.loads(line) - if d.get("id") == zc_id: - if "error" in d: - del self.pending_turns[msg_id] - return self._error(msg_id, -32603, - f"zcode send failed: {d['error'].get('message','')}") - if d.get("result", {}).get("accepted"): - send_accepted = True - break - except json.JSONDecodeError: - pass - if not send_accepted: - del self.pending_turns[msg_id] - return self._error(msg_id, -32603, "zcode send 未被接受 (无 accepted 响应)") - - # zcode 不主动推送 turn 完成事件, 改用轮询 session/read 查 status - # 注意: 必须先观察到 status=running 才算 turn 真正启动, 否则会误判上一个 idle 为完成 - # 阶段1: 轮询保持轻量 (只读 projection 发 usage_update); - # turn 完成(idle)后才取一次完整 snapshot 发 tool/text/reasoning/plan 事件。 + # ===== 双模式: 优先事件驱动, 失败降级轮询 ===== + # 新版 ZCode (0.14.8+) 支持 session/subscribe 推送流式事件 (model.streaming)。 + # 事件驱动模式下: model.streaming → 逐段 TextDelta (真流式); + # tool.updated → 实时工具调用状态。 + # 降级模式 (旧版 CLI / subscribe 失败): 轮询 session/read, turn 结束整段发 (伪流式)。 + + # ⚠️ 时序关键: 必须先 subscribe + register listener, 再 session/send。 + # 否则 send 和 register 之间的 session/event 会被 reader 线程丢弃, + # 短 turn 的 turn.completed 可能丢失导致 120s 超时。 + listener = EventStreamListener(self.backend, zcode_sid) + sub_id = self._next_id() + snapshot = listener.subscribe(sub_id) + + if snapshot is not None: + # 事件驱动模式: 先注册 listener (确保不丢事件), 再 send + self.backend.register_event_listener(zcode_sid, listener) + try: + # 现在 send prompt (listener 已注册, 不会丢事件) + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/send", + {"sessionId": zcode_sid, "content": prompt_text}, + timeout=10) + if "error" in resp: + del self.pending_turns[msg_id] + return self._error(msg_id, -32603, + f"zcode send failed: {resp['error'].get('message', '')}") + if not resp.get("result", {}).get("accepted", False): + del self.pending_turns[msg_id] + return self._error(msg_id, -32603, "zcode send 未被接受 (无 accepted 响应)") + + result = self._run_event_turn( + listener, acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ) + finally: + self.backend.unregister_event_listener(zcode_sid) + return result + else: + # 降级: 轮询模式 — send 在轮询循环前发 + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "session/send", + {"sessionId": zcode_sid, "content": prompt_text}, + timeout=10) + if "error" in resp: + del self.pending_turns[msg_id] + return self._error(msg_id, -32603, + f"zcode send failed: {resp['error'].get('message', '')}") + if not resp.get("result", {}).get("accepted", False): + del self.pending_turns[msg_id] + return self._error(msg_id, -32603, "zcode send 未被接受 (无 accepted 响应)") + log(" ⟳ 降级轮询模式 (事件流不可用)") + return self._run_polling_turn( + acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ) + + def _run_event_turn(self, listener, acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ): + """事件驱动 turn 循环: 消费 session/event 推送, 翻译成 ACP 事件实时发。 + + 真流式: model.streaming 的 text_delta 逐段发为 agent_message_chunk。 + """ + translator = EventTranslator() + t0 = time.time() + emitted_text = False # 本轮是否已通过事件发过文本 (用于决定兜底) + last_event_time = time.time() + last_stall_check = 0 # 上次停滞检查的时间戳 (可重复检查) + + while time.time() - t0 < 120: + # 每轮先处理 inbox: 让 session/cancel 真正生效 + self._drain_inbox() + if turn["cancelled"]: + self._cancel_backend_turn(zcode_sid) + del self.pending_turns[msg_id] + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"stopReason": "cancelled"}} + + # 从事件队列取一条 (超时 0.5s) + event = listener.poll_event(timeout=0.5) + if event is None: + # 无事件。如果 turn 已启动且超过 15s 无事件, + # 可能 turn.completed 事件丢失 → 用 session/read 确认并轮询收尾。 + # 每 15s 可重复检查 (长静默 turn 不会只查一次就放弃)。 + if translator.turn_started \ + and time.time() - last_event_time > 15 \ + and time.time() - last_stall_check > 15: + last_stall_check = time.time() + log(" ⚠ 事件流停滞, 用 session/read 确认 turn 状态") + monitor = TurnMonitor(self.backend, zcode_sid, self) + proj = monitor.poll_once() + if proj and proj.get("status") == "idle": + log(" ✓ 确认 turn 已完成 (事件可能丢失), 收尾") + del self.pending_turns[msg_id] + # 停滞收尾: 不调 differ.diff() (differ 不知道事件路径已发过什么, + # 会重复发 TextDelta/tool 事件)。只做兜底: 若事件路径未发文本, + # 用 _fetch_last_reply 取最后回复。 + if not emitted_text: + reply_text = self._fetch_last_reply(zcode_sid, differ) + if reply_text: + self._send_acp_notification("session/update", { + "sessionId": acp_sid, + "update": {"sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": reply_text}, + "messageId": chunk_msg_id}, + }) + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"stopReason": "end_turn"}} + continue + + last_event_time = time.time() + + # 翻译事件 → ACP 事件 dict 列表 + acp_events = translator.translate(event) + for ev in acp_events: + # 标记: 是否已通过 model.streaming 发过文本 + if ev.get("kind") == "TextDelta": + emitted_text = True + self._dispatch_event(ev, acp_sid, chunk_msg_id) + + # turn 结束判定 + if translator.turn_done: + del self.pending_turns[msg_id] + result_type = translator.turn_result_type + if result_type == "cancelled": + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"stopReason": "cancelled"}} + if result_type and result_type.startswith("error_"): + log(f"session/prompt: turn 失败 ({result_type})") + return self._error(msg_id, -32603, f"zcode turn failed: {result_type}") + + # turn 正常完成。若未收到任何 text_delta (如纯工具调用 turn), 兜底发最后回复 + if not emitted_text: + reply_text = self._fetch_last_reply(zcode_sid, differ) + if reply_text: + self._send_acp_notification("session/update", { + "sessionId": acp_sid, + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": reply_text}, + "messageId": chunk_msg_id, + }, + }) + log(f"session/prompt 完成 (事件): stopReason=end_turn, " + f"{'兜底' if not emitted_text else '流式'} 回复") + else: + log("session/prompt 完成 (事件): stopReason=end_turn, 流式文本已发") + + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"stopReason": "end_turn"}} + + # 超时 + del self.pending_turns[msg_id] + log("session/prompt 超时: 事件模式下 120s 未完成") + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"stopReason": "max_turn_requests"}} + + def _run_polling_turn(self, acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ): + """轮询降级 turn 循环: 轮询 session/read 检测 status, turn 完成整段发事件。 + + 这是旧版 ZCode (无事件推送) 的兼容模式。 + """ monitor = TurnMonitor(self.backend, zcode_sid, self) last_usage = None t0 = time.time() last_status = None turn_started = False - consecutive_none = 0 # 连续 poll 失败计数 (P1.3) + consecutive_none = 0 # 连续 poll 失败计数 turn_completed = False while time.time() - t0 < 120: # 每轮先处理 inbox: 让 session/cancel 真正生效 self._drain_inbox() if turn["cancelled"]: - # 修复 P2.1: 向后端转发取消, 而非仅本地标记 self._cancel_backend_turn(zcode_sid) del self.pending_turns[msg_id] return {"jsonrpc": "2.0", "id": msg_id, @@ -643,7 +1285,6 @@ class ACPBridge: projection = monitor.poll_once() status = projection.get("status") if projection else None - # 修复 P1.3: poll 返回 None 的处理 if projection is None: consecutive_none += 1 if consecutive_none >= 10: # 连续 ~5s 无响应 @@ -672,7 +1313,6 @@ class ACPBridge: elif status == "idle" and turn_started: log(" ✓ turn 完成 (status=idle)") # turn 完成: 取一次完整 snapshot, 发本轮新增的 tool/text/reasoning/plan 事件 - # differ 已标记历史消息为已见, 只会产出本轮新增 (P1.1) cur_snapshot = self._build_snapshot(zcode_sid, projection) events = differ.diff(None, cur_snapshot) for ev in events: @@ -682,7 +1322,7 @@ class ACPBridge: time.sleep(0.5) # 轮询间隔 - # 修复 P1.2: turn 未完成 (超时或从未 running) 时返回明确错误, 不假装 end_turn + # turn 未完成 (超时或从未 running) del self.pending_turns[msg_id] if not turn_completed: if not turn_started: @@ -693,7 +1333,6 @@ class ACPBridge: "result": {"stopReason": "max_turn_requests"}} # turn 正常完成。differ 已发本轮新增 text; 若本轮无新增 text, 兜底发最后一条 assistant 回复 - # 修复 P1: 兜底要排除 differ 已见的消息, 避免把上一轮旧回复发出 if not differ.emitted_text_this_turn: differ.emitted_text_this_turn = False # reset reply_text = self._fetch_last_reply(zcode_sid, differ) diff --git a/packages/agent-help/zcode-agent-help b/packages/agent-help/zcode-agent-help index b870cea..b1009c6 100755 --- a/packages/agent-help/zcode-agent-help +++ b/packages/agent-help/zcode-agent-help @@ -271,22 +271,26 @@ ECOSYSTEM = { "role": "把 zcode 桥接为 ACP (Agent Client Protocol) Agent", "maturity": "experimental", "transport": "ACP over stdio (JSON-RPC 2.0) 对外; ZCode Protocol 对内", + "streaming_mode": "双模式: 事件驱动 (真流式, CLI 0.14.8+) / 轮询 (伪流式, 自动降级)", "acp_methods_implemented": [ "initialize", "session/new", "session/prompt", "session/cancel", "session/list", "session/resume", + # 扩展方法 (非标准 ACP, 透传 zcode 0.14.8+ 协议): + "session/fork", "session/rewind", "session/goal", + "session/compact", "session/steer", ], "acp_events_emitted": [ - {"event": "tool_call/tool_call_update", "status": "已验证"}, - {"event": "usage_update", "status": "已验证"}, - {"event": "agent_message_chunk", "status": "已验证"}, - {"event": "agent_thought_chunk", "status": "代码就位, GLM-5.2 无数据"}, - {"event": "plan", "status": "代码就位, 数据驱动"}, - {"event": "diff (文件列表)", "status": "代码就位, 仅文件名无内容"}, + {"event": "tool_call/tool_call_update", "status": "✅ 实时 (事件驱动)"}, + {"event": "usage_update", "status": "✅ 已验证"}, + {"event": "agent_message_chunk", "status": "✅ 真流式 (model.streaming text_delta)"}, + {"event": "agent_thought_chunk", "status": "✅ 流式 (GLM-5-Turbo reasoning_delta)"}, + {"event": "plan", "status": "⚠️ 代码就位, 数据驱动"}, + {"event": "diff (文件列表)", "status": "⚠️ 仅文件名无内容"}, ], "use_case": "让 Zed/JetBrains 等 ACP 兼容编辑器调用 zcode", "limitations": [ "工具调用 turn 不稳定时 turn 完成检测受影响", - "流式文本是伪流式 (zcode 快照式架构)", + "旧版 CLI (<0.14.8) 降级为伪流式 (turn 完成后整段发)", "diff 无内容 (zcode 协议层不暴露 oldText/newText)", "P4b 权限请求未实现 (yolo 模式不触发)", ], diff --git a/skills/zcode-bridge-guide/SKILL.md b/skills/zcode-bridge-guide/SKILL.md new file mode 100644 index 0000000..6a2ffc6 --- /dev/null +++ b/skills/zcode-bridge-guide/SKILL.md @@ -0,0 +1,392 @@ +--- +name: zcode-bridge-guide +version: 1.0.0 +description: 驱动 ZCode（智谱 GLM 系列 coding agent）的通用说明书。覆盖三种接入模式（CLI --prompt / ACP bridge / MCP tools）、凭证配置、真流式/伪流式双模式、扩展协议方法、任务书模板、已知坑。需要把 ZCode 当子代理编排、跑编码/审查任务、或集成进编辑器时用。兼容 ZCode CLI 0.14.5 ~ 0.14.8+。 +user-invocable: true +--- + +# 驱动 ZCode（三模式通用说明书） + +> 本 skill 是 [zcode-open-bridge](https://github.com/tizerluo/zcode-open-bridge) 项目的配套说明书。 +> 兼容 ZCode CLI **0.14.5 ~ 0.14.8+**（App 3.1.4）。新版功能（事件驱动真流式、fork/rewind/goal/compact/steer）在旧版上自动降级。 + +## 前置条件 + +1. 已安装 [ZCode](https://zcode.z.ai) 桌面 App（含 CLI） +2. 已通过 ZCode 登录（凭证存在 `~/.zcode/v2/config.json`） +3. （可选）已 clone zcode-open-bridge 仓库 + +## 入口与认证 + +### 让 `zcode` 命令可用 + +ZCode 的 CLI 藏在 App 内部，默认不在 PATH： + +```bash +# macOS +ln -s /Applications/ZCode.app/Contents/Resources/glm/zcode.cjs ~/.local/bin/zcode +``` + +### 凭证配置（⚠️ `--prompt` 模式必读） + +**关键坑**：`--prompt` / `--target` 模式**不读** `~/.zcode/cli/config.json` 的 model 字段，必须用环境变量注入。ZCode 桌面 App 的凭证存在 `~/.zcode/v2/config.json` 的 `provider` 段。 + +**推荐方案**：在 `~/.zshrc` 配置一个 shell 函数，动态读取凭证（不明文存 key）： + +```bash +zcode() { + local cfg="$HOME/.zcode/v2/config.json" + eval "$(python3 -c " +import json +c=json.load(open('$cfg')) +for k,v in c['provider'].items(): + if v.get('enabled'): + o=v['options'] + print(f'export ZCODE_MODEL=\"{next(iter(v.get(\"models\",{}))) or \"GLM-5.2\"}\"') + print(f'export ZCODE_BASE_URL=\"{o.get(\"baseURL\",\"\")}\"') + print(f'export ANTHROPIC_API_KEY=\"{o.get(\"apiKey\",\"\")}\"') + break +")" + command zcode "$@" +} +``` + +> ACP bridge 和 MCP server 内部已实现同样的凭证读取逻辑（`shared/credentials.py`），无需额外配置。 + +### ⚠️ 非交互 shell 坑（Codex/agent 编排注意） + +非交互 shell（`zsh -lc`）不会 source `~/.zshrc`，`zcode` 可能解析成 npm binary 而非 shell function。检查： +- `zsh -ic 'type -a zcode'`（交互路径，才会看到 function） +- 若非交互不可用，改用显式 env 包装（见下方 CLI 模式） + +--- + +## 三种模式，按场景选 + +| 模式 | 适合场景 | 会话能力 | 流式 | 接入方式 | +|------|---------|---------|:----:|---------| +| **CLI `--prompt`** | 单轮任务（审查、解释、生成） | 无（一次跑完） | ❌ | `zcode --prompt` | +| **ACP bridge** | 多轮编排、子代理、编辑器集成 | 有（session/resume/fork） | ✅ 真流式 (0.14.8+) | `zcode-acp-bridge` | +| **MCP tools** | MCP client 内直接调用（轻量） | 有（MCP server 管理） | ❌ | `zcode-mcp-server` | + +**选择指南**： +- 简单任务（一轮跑完）→ **CLI** +- 需要多轮交互 / 流式输出 / 结构化事件 / resume → **ACP bridge** +- 在 MCP client（Claude Code/Cursor）内直接调用 → **MCP tools** + +--- + +## 模式一：CLI `--prompt`（最简单） + +### 执行形态（改代码） +```bash +zcode --prompt "" --attach --mode build --json +# --mode build 允许写文件 + 执行命令 +# --mode plan 只读分析，不改文件（最安全，适合审查） +# --mode edit 允许写文件，不执行命令 +# --mode yolo 全自动（--prompt 的默认模式） +# --json 输出 JSON（含 sessionId, response, usage） +# --attach 附带文件（可重复） +# --resume 续接指定 session（--json 返回 sessionId） +# --cwd 指定工作目录 +``` + +### 长程任务（设定 session 目标） +```bash +# 方式1: --target (与 --prompt 互斥) +zcode --target "重构认证模块" --mode yolo + +# 方式2: prompt 内用 /goal (等价于 --target) +zcode --prompt "/goal 重构认证模块" --mode yolo + +# 替换已有目标 +zcode --target "新目标" --target-replace +``` + +### 评审形态（只读） +```bash +zcode --prompt "" --attach --mode plan --json +``` + +### 显式 env 包装（非交互 shell 用） +```bash +read -r model base_url api_key < <(python3 - <<'PY' +import json, pathlib +cfg = json.loads((pathlib.Path.home() / ".zcode/v2/config.json").read_text()) +for pid, p in cfg.get("provider", {}).items(): + if p.get("enabled"): + o = p["options"] + m = "GLM-5.2" if "GLM-5.2" in p.get("models", {}) else next(iter(p.get("models", {})), "") + if pid.startswith("builtin:zai"): + m = f"zai/{m}" + print(m, o.get("baseURL", ""), o.get("apiKey", "")) + break +PY +) +ANTHROPIC_API_KEY="$api_key" ZCODE_BASE_URL="$base_url" ZCODE_MODEL="$model" \ + zcode --prompt "" --mode plan --json +``` + +### ⚠️ CLI 模式已知限制 +1. **不支持 stdin 管道**：不能 `cat file | zcode --prompt`，必须用 `--attach` +2. **无流式输出**（`--stream-json` 不支持）：turn 结束一次性返回 +3. **tool call 轮次不稳定**：duration 38s~100s+，有时不完成 +4. **自由 prompt 可能触发限流**：开放式 prompt 会 spawn 多个 explore 子代理，撞 z.ai 限流。修法：把证据塞进 `--attach`，prompt 写明「不要调用工具/不要 spawn 子代理，只基于附件推理」 + +--- + +## 模式二：ACP bridge（多轮编排，真流式） + +ACP bridge 是把 ZCode 当子代理编排的**最佳方式**——支持多轮会话、session resume、真流式文本、实时工具调用追踪。 + +### 启动 +```bash +# 方式1: 直接运行（需要 zcode 在 PATH 或用 ZCODE_BIN 指定） +./packages/acp-bridge/zcode-acp-bridge + +# 方式2: 指定 zcode 路径 +ZCODE_BIN=/Applications/ZCode.app/Contents/Resources/glm/zcode.cjs \ + ./packages/acp-bridge/zcode-acp-bridge + +# 启动后等待 ACP JSON-RPC 消息（stdio） +``` + +### ACP 协议流程 + +``` +1. initialize → 握手，声明 capabilities +2. session/new → 创建会话（返回 sessionId） +3. session/prompt → 发送 prompt（非阻塞，返回 accepted） +4. 收 session/update → 流式事件（文本逐段、工具实时状态、usage） +5. 等 stopReason → end_turn / cancelled / max_turn_requests +6. session/resume → 续接已有会话（跨多轮） +7. session/cancel → 取消正在执行的 turn +``` + +### 从脚本驱动 ACP bridge（Python 示例） + +```python +import json, subprocess, select, time + +proc = subprocess.Popen( + ["./packages/acp-bridge/zcode-acp-bridge"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + text=True, bufsize=1, + env={**__import__("os").environ, "ZCODE_BIN": "zcode"} +) + +def send(msg): + proc.stdin.write(json.dumps(msg) + "\n") + proc.stdin.flush() + +# 1. initialize +send({"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": 1}}) +# ... 读取 id=1 响应 + +# 2. session/new +send({"jsonrpc": "2.0", "id": 2, "method": "session/new", + "params": {"cwd": "/path/to/project"}}) +# ... 读取 sessionId + +# 3. session/prompt (会收到流式 session/update notification) +sid = "sess_xxx" +send({"jsonrpc": "2.0", "id": 3, "method": "session/prompt", + "params": {"sessionId": sid, + "prompt": [{"type": "text", "text": "审查这段代码"}]}}) + +# 4. 收集流式事件 + 等最终响应 +chunks = [] +while True: + line = proc.stdout.readline() + if not line: + break + msg = json.loads(line) + if msg.get("id") == 3: + print("stopReason:", msg["result"]["stopReason"]) + break + elif msg.get("method") == "session/update": + update = msg["params"]["update"] + su = update.get("sessionUpdate", "") + if su == "agent_message_chunk": + chunks.append(update["content"]["text"]) + elif su in ("tool_call", "tool_call_update"): + print("工具调用:", update.get("toolCallId"), update.get("status")) + +print("回复:", "".join(chunks)) +proc.terminate() +``` + +### 真流式 vs 伪流式（双模式自动降级） + +ACP bridge 根据运行时检测自动选择模式： + +| 模式 | 条件 | 文本到达 | 工具状态 | +|------|------|----------|----------| +| **事件驱动（真流式）** | CLI ≥ 0.14.8 且 subscribe 成功 | 逐段推送（2~8+ 段） | 实时（scheduled→started→result） | +| **轮询（伪流式）** | 旧版 CLI 或 subscribe 失败 | turn 完成后整段发 | turn 完成后一次性 | + +**无需手动选择**——bridge 先尝试 `session/subscribe`，失败则自动降级轮询。 + +### 扩展方法（非标准 ACP，0.14.8+ 新增） + +| 方法 | 作用 | params | +|------|------|--------| +| `session/fork` | 从 checkpoint 分叉新会话 | `{sessionId, target?}` | +| `session/rewind` | 回退工作区文件到 checkpoint | `{sessionId, target?, expectedRevision?}` | +| `session/goal` | 读取/设置 session 目标 | `{sessionId, action: show\|set\|replace\|clear, objective?}` | +| `session/compact` | 压缩对话上下文 | `{sessionId}` | +| `session/steer` | turn 进行中追加指令 | `{sessionId, content}` | + +> ⚠️ `session/goal action=set` 会启动内部 AI turn（异步），耗时 10~45s。bridge 会自动等待 prompt lock 释放后返回，但调用方需预期较长延迟。 + +### ⚠️ ACP 模式已知限制 +1. **默认 `mode=yolo`**：所有 prompt 可能触发无确认的文件修改和命令执行。信任环境要隔离（worktree / 受信目录） +2. **diff 无内容**：协议层不暴露 oldText/newText，只能列文件名 +3. **轮次时间不稳定**：同 CLI 模式，38s~100s+ +4. **会话存储共享**：ACP bridge 和 CLI `--prompt` 共享 `~/.zcode/cli/db/db.sqlite`，可跨方式 resume + +--- + +## 模式三：MCP tools（MCP client 内直接调用） + +MCP server 暴露两个标准 MCP tool，供 Claude Code / Cursor 等 MCP client 调用。 + +### 注册到 MCP client + +在 MCP client 的配置（如 `~/.zcode/cli/config.json` 的 `mcp.servers`）中添加： + +```json +{ + "mcpServers": { + "zcode-mcp": { + "command": "/path/to/zcode-open-bridge/packages/mcp-server/zcode-mcp-server", + "args": [] + } + } +} +``` + +### 可用 tools + +| Tool | 用途 | 安全性 | +|------|------|--------| +| `get_zcode_capabilities` | 返回 ZCode 完整能力清单 | 只读 | +| `zcode_review` | 调用 ZCode 审查代码 | 只读（`--mode plan`，不改文件） | + +### `zcode_review` 参数 +- `files`：要审查的文件路径列表 +- `code`：直接传入代码文本（与 files 二选一或组合） +- `focus`：审查重点（如 "安全性"、"性能"、"找 bug"） +- `cwd`：工作目录（影响 zcode 的项目上下文） + +--- + +## 任务书模板（派 ZCode 跑任务时用） + +派 ZCode 跑编码任务时，**不要只给一个 prompt**——写结构化任务书。 + +### 必备要素 +```markdown +## 必读清单 +- 现状代码锚点：<要扩展的函数/要复用的测试基建的具体路径> + +## 执行要点 +- <安全纪律、约束、依赖白名单> + +## 完成标准（硬门） +- build 命令： +- test 命令： +- 期望结果：<具体数字/状态> + +## 职责边界 +- 只改文件 + 跑 build/test，不碰 git +- 遇到改变边界的问题写 BLOCKED.md 后停止 + +## 结束总结要求 +- 做了什么、build/test 结果、关键决策 +``` + +### 为什么这样写 +- **必读清单**：省 agent 自己摸索，减少幻觉（无锚点时 agent 可能自创接口） +- **完成标准硬门**：agent 自报 build/test 结果不可信，必须本地重跑验证 +- **职责边界**：agent 不碰 git（编排层拥有 git），遇到问题写 BLOCKED.md 停止 + +--- + +## 独立复核（不信 agent 自报） + +**核心原则**：agent 报的 build/test 结果不可信，必须本地重跑验证。 + +```bash +cd +npm install # agent 可能加了依赖 +npm run build +npm test # 全量，看实际数字 +# 对比 agent 报的数字 vs 实际数字 +``` + +--- + +## 版本兼容性 + +本 skill 及 zcode-open-bridge 项目兼容以下 ZCode 版本： + +| ZCode CLI 版本 | 支持情况 | 差异 | +|:--------------:|:--------:|------| +| **0.14.8+** (App 3.1.4) | ✅ 完整 | ACP bridge 真流式；fork/rewind/goal/compact/steer 可用 | +| **0.14.5 ~ 0.14.7** | ✅ 兼容 | ACP bridge 自动降级伪流式；扩展方法不可用（协议未实现） | +| **< 0.14.5** | ⚠️ 未测 | CLI `--prompt` 基本可用；ACP bridge 未验证 | + +**降级行为**： +- ACP bridge 检测到 `session/subscribe` 不可用时，自动切换到轮询 `session/read`（伪流式） +- 扩展方法（fork/rewind/goal/compact/steer）在旧版 ZCode 上会透传后端错误（`-32603 zcode failed: ...`），不影响标准 ACP 方法 + +--- + +## 成功判定 + +### CLI 模式 +- `returncode == 0` + stdout 含有效内容 → 成功 +- `--json` 模式：检查 `response` 字段是否存在 +- `sessionId` 可用于后续 `--resume` + +### ACP 模式 +- `session/prompt` 最终响应 `stopReason: end_turn` → 正常完成 +- `stopReason: cancelled` → 被取消 +- `stopReason: max_turn_requests` → 超时（120s） +- 流式事件中的 `agent_message_chunk` → 文本输出（真流式有多段） + +### MCP 模式 +- `zcode_review` 返回 `{content: [{type: "text", text: "..."}]}` +- `isError: true` → 失败 + +--- + +## 自检 + +```bash +# 1. CLI 模式自检 +zcode --prompt "Reply with exactly: ZCODE_OK" --json 2>&1 | tail -3 +# 输出含 ZCODE_OK + rc=0 → 通 + +# 2. ACP bridge 握手自检 +echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":1}}' | \ + ./packages/acp-bridge/zcode-acp-bridge 2>/dev/null | head -1 | python3 -c " +import sys, json +d = json.loads(sys.stdin.readline()) +print('ACP OK:', d.get('result', {}).get('agentInfo', {}).get('name', 'unknown')) +" + +# 3. 能力发现自检 +./packages/agent-help/zcode-agent-help --pretty 2>/dev/null | head -5 +``` + +--- + +## 参考 + +- **项目仓库**：[zcode-open-bridge](https://github.com/tizerluo/zcode-open-bridge) +- **能力发现说明书**：`zcode-agent-help --pretty`（输出 ZCode 全部能力清单） +- **协议文档**：README 的「ACP bridge」和「扩展方法」章节 +- **同系列驱动 skill**：`drive-zcode`（个人实战版，含 spec-pr-pipeline 经验） diff --git a/tests/test_event_translator.py b/tests/test_event_translator.py new file mode 100644 index 0000000..5b2d576 --- /dev/null +++ b/tests/test_event_translator.py @@ -0,0 +1,306 @@ +""" +test_event_translator.py — EventTranslator 翻译逻辑单测 + +验证事件驱动模式下 EventTranslator 能正确把 zcode session/event 事件 +(model.streaming, tool.updated, turn.started/completed) 翻译成 ACP 事件 dict: + E0 流式文本 (model.streaming text_delta → TextDelta) + E1 流式推理 (model.streaming reasoning_delta → ReasoningDelta) + E2 工具新增 (tool.updated scheduled → ToolCallNew) + E3 工具开始 (tool.updated started → ToolCallUpdate in_progress) + E4 工具完成 (tool.updated result → ToolCallUpdate completed) + E5 工具进度 (tool.updated progress → ToolCallUpdate with output) + E6 turn 生命周期 (turn.started + turn.completed → 状态标记 + UsageDelta) + E7 turn 失败 (turn.failed → turn_done 标记) + E8 空 delta 过滤 (model.streaming 无 delta → 无事件) + E9 batch/忽略类型 (session.updated → 无事件) + E10 多段流式合并 (多条 text_delta → 多个 TextDelta) + +运行: python3 tests/test_event_translator.py +依赖: 仅 Python 标准库 + 本项目的 acp-bridge 模块 +""" + +import os +import types +import unittest + +# 把 acp-bridge 模块加载进来 (单文件, 用 exec 导入类定义, 跳过 main) +BRIDGE_PATH = os.path.join( + os.path.dirname(__file__), "..", "packages", "acp-bridge", "zcode-acp-bridge" +) + + +def _load_bridge_module(): + """加载 acp-bridge 的类定义 (跳过 if __name__ 块)""" + mod = types.ModuleType("acp_bridge") + with open(BRIDGE_PATH) as f: + code = f.read() + code_no_main = code.split('if __name__ == "__main__":')[0] + # 提供 exec 所需的 log 占位 (模块级函数) + exec(code_no_main, mod.__dict__) + return mod + + +def _event(etype, payload=None): + """构造一个 session/event params dict""" + return {"type": etype, "payload": payload or {}, "seq": 0, "sessionId": "sess_test"} + + +class TestEventTranslator(unittest.TestCase): + """EventTranslator 的事件翻译单测""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_bridge_module() + cls.Translator = cls.mod.EventTranslator + + def _new_translator(self): + return self.Translator() + + # ---------- E0: 流式文本 ---------- + def test_e0_text_delta(self): + """E0: model.streaming text_delta → TextDelta""" + t = self._new_translator() + events = t.translate(_event("model.streaming", {"kind": "text_delta", "delta": "你好"})) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["kind"], "TextDelta") + self.assertEqual(events[0]["text"], "你好") + + # ---------- E1: 流式推理 ---------- + def test_e1_reasoning_delta(self): + """E1: model.streaming reasoning_delta → ReasoningDelta""" + t = self._new_translator() + events = t.translate(_event("model.streaming", {"kind": "reasoning_delta", "delta": "思考中"})) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["kind"], "ReasoningDelta") + self.assertEqual(events[0]["text"], "思考中") + + # ---------- E2: 工具新增 ---------- + def test_e2_tool_new(self): + """E2: tool.updated scheduled → ToolCallNew""" + t = self._new_translator() + events = t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": "call_1", "toolName": "Bash", + })) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["kind"], "ToolCallNew") + self.assertEqual(events[0]["call_id"], "call_1") + self.assertEqual(events[0]["tool"], "Bash") + self.assertEqual(events[0]["acp_kind"], "execute") + self.assertEqual(events[0]["status"], "pending") + + def test_e2b_tool_kind_mapping(self): + """E2b: tool 名 → ACP ToolKind 映射 (Read→read, Edit→edit, Grep→search)""" + t = self._new_translator() + for zname, expected in [("Read", "read"), ("Edit", "edit"), ("Grep", "search"), + ("WebSearch", "fetch"), ("Glob", "search")]: + events = t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": f"call_{zname}", "toolName": zname, + })) + self.assertEqual(events[0]["acp_kind"], expected, f"映射错误: {zname}") + + def test_e2c_tool_dedup(self): + """E2c: 同一 toolCallId 的 scheduled 不重复发 ToolCallNew""" + t = self._new_translator() + payload = {"kind": "scheduled", "toolCallId": "call_dup", "toolName": "Bash"} + t.translate(_event("tool.updated", payload)) + events2 = t.translate(_event("tool.updated", payload)) + self.assertEqual(len(events2), 0, "重复 scheduled 不应再发 ToolCallNew") + + # ---------- E3: 工具开始 ---------- + def test_e3_tool_started(self): + """E3: tool.updated started → ToolCallUpdate in_progress""" + t = self._new_translator() + t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": "call_2", "toolName": "Bash", + })) + events = t.translate(_event("tool.updated", { + "kind": "started", "toolCallId": "call_2", "toolName": "Bash", + })) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["kind"], "ToolCallUpdate") + self.assertEqual(events[0]["status"], "in_progress") + + # ---------- E4: 工具完成 ---------- + def test_e4_tool_result(self): + """E4: tool.updated result → ToolCallUpdate completed""" + t = self._new_translator() + t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": "call_3", "toolName": "Read", + })) + events = t.translate(_event("tool.updated", { + "kind": "result", "toolCallId": "call_3", "result": "文件内容...", + })) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["status"], "completed") + self.assertEqual(events[0]["output"], "文件内容...") + + # ---------- E5: 工具进度 ---------- + def test_e5_tool_progress(self): + """E5: tool.updated progress → ToolCallUpdate (含实时输出)""" + t = self._new_translator() + t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": "call_4", "toolName": "Bash", + })) + events = t.translate(_event("tool.updated", { + "kind": "progress", "toolCallId": "call_4", "stdoutTail": "正在运行...", + })) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["status"], "in_progress") + self.assertEqual(events[0]["output"], "正在运行...") + + # ---------- E6: turn 生命周期 ---------- + def test_e6_turn_lifecycle(self): + """E6: turn.started + turn.completed → 状态标记 + UsageDelta""" + t = self._new_translator() + self.assertFalse(t.turn_started) + + t.translate(_event("turn.started")) + self.assertTrue(t.turn_started) + self.assertFalse(t.turn_done) + + events = t.translate(_event("turn.completed", { + "resultType": "success", "tokenCount": 100, "usage": {"contextWindow": 200000}, + })) + self.assertTrue(t.turn_done) + self.assertEqual(t.turn_result_type, "success") + # turn.completed 应产出 UsageDelta + usage_events = [e for e in events if e["kind"] == "UsageDelta"] + self.assertEqual(len(usage_events), 1) + self.assertEqual(usage_events[0]["used"], 100) + + # ---------- E7: turn 失败 ---------- + def test_e7_turn_failed(self): + """E7: turn.failed → turn_done 标记 + result_type""" + t = self._new_translator() + t.translate(_event("turn.failed", {"resultType": "error_max_turns"})) + self.assertTrue(t.turn_done) + self.assertEqual(t.turn_result_type, "error_max_turns") + + # ---------- E8: 空 delta 过滤 ---------- + def test_e8_empty_delta(self): + """E8: model.streaming 无 delta → 不产出事件""" + t = self._new_translator() + events = t.translate(_event("model.streaming", {"kind": "text_delta", "delta": ""})) + self.assertEqual(len(events), 0) + + # ---------- E9: 忽略类型 ---------- + def test_e9_ignored_types(self): + """E9: session.updated/titleUpdated → 不产出事件""" + t = self._new_translator() + for etype in ["session.updated", "session.titleUpdated", "streamRecovery.updated"]: + events = t.translate(_event(etype, {})) + self.assertEqual(len(events), 0, f"{etype} 不应产出事件") + + # ---------- E10: 多段流式 ---------- + def test_e10_multi_segment_streaming(self): + """E10: 多条 text_delta → 多个 TextDelta (逐段, 不合并)""" + t = self._new_translator() + segments = ["第一段", "第二段", "第三段"] + all_events = [] + for seg in segments: + all_events.extend(t.translate(_event("model.streaming", { + "kind": "text_delta", "delta": seg, + }))) + self.assertEqual(len(all_events), 3) + # 验证顺序和内容 + combined = "".join(e["text"] for e in all_events) + self.assertEqual(combined, "第一段第二段第三段") + + # ---------- E11: 完整 turn 模拟 ---------- + def test_e11_full_turn_sequence(self): + """E11: 模拟一个完整 turn 的事件序列 (started → text → tool → completed)""" + t = self._new_translator() + all_events = [] + + # turn 开始 + all_events.extend(t.translate(_event("turn.started"))) + + # 流式文本 2 段 + all_events.extend(t.translate(_event("model.streaming", {"kind": "text_delta", "delta": "我来"}))) + all_events.extend(t.translate(_event("model.streaming", {"kind": "text_delta", "delta": "读取文件"}))) + + # 工具调用完整序列 + all_events.extend(t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": "call_full", "toolName": "Read"}))) + all_events.extend(t.translate(_event("tool.updated", { + "kind": "started", "toolCallId": "call_full", "toolName": "Read"}))) + all_events.extend(t.translate(_event("tool.updated", { + "kind": "result", "toolCallId": "call_full", "result": "内容"}))) + + # turn 完成 + all_events.extend(t.translate(_event("turn.completed", { + "resultType": "success", "tokenCount": 50, "usage": {"contextWindow": 100000}}))) + + # 验证事件序列 + kinds = [e["kind"] for e in all_events] + self.assertEqual(kinds, [ + "TextDelta", "TextDelta", # 2 段流式文本 + "ToolCallNew", # scheduled + "ToolCallUpdate", # started + "ToolCallUpdate", # result + "UsageDelta", # turn completed usage + ]) + self.assertTrue(t.turn_done) + self.assertEqual(t.turn_result_type, "success") + + # ---------- E12: tool error ---------- + def test_e12_tool_error(self): + """E12: tool.updated error → ToolCallUpdate failed""" + t = self._new_translator() + t.translate(_event("tool.updated", { + "kind": "scheduled", "toolCallId": "call_err", "toolName": "Bash"})) + events = t.translate(_event("tool.updated", { + "kind": "error", "toolCallId": "call_err", "error": "命令失败"})) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["status"], "failed") + self.assertEqual(events[0]["output"], "命令失败") + + # ---------- E13: batch tool completion ---------- + def test_e13_tool_batch(self): + """E13: tool.updated batch → 为 scheduled 过的 call_id 补发 completed""" + t = self._new_translator() + # 两个工具 scheduled + t.translate(_event("tool.updated", {"kind": "scheduled", "toolCallId": "b1", "toolName": "Read"})) + t.translate(_event("tool.updated", {"kind": "scheduled", "toolCallId": "b2", "toolName": "Bash"})) + # batch 事件 (含 toolCallIds) + events = t.translate(_event("tool.updated", { + "kind": "batch", "toolCallIds": ["b1", "b2"], "successCount": 2, "errorCount": 0})) + # 应为两个 scheduled 过的 id 各补发 completed + self.assertEqual(len(events), 2) + statuses = {e["call_id"]: e["status"] for e in events} + self.assertEqual(statuses["b1"], "completed") + self.assertEqual(statuses["b2"], "completed") + + def test_e13b_batch_with_errors(self): + """E13b: batch 有 error → 对应 call_id 标记 failed""" + t = self._new_translator() + t.translate(_event("tool.updated", {"kind": "scheduled", "toolCallId": "be1", "toolName": "Bash"})) + events = t.translate(_event("tool.updated", { + "kind": "batch", "toolCallIds": ["be1"], "successCount": 0, "errorCount": 1})) + self.assertEqual(events[0]["status"], "failed") + + def test_e13c_batch_ignores_unknown_ids(self): + """E13c: batch 里的未知 toolCallId (未 scheduled) 不补发""" + t = self._new_translator() + t.translate(_event("tool.updated", {"kind": "scheduled", "toolCallId": "known", "toolName": "Read"})) + events = t.translate(_event("tool.updated", { + "kind": "batch", "toolCallIds": ["known", "unknown"], "successCount": 2, "errorCount": 0})) + # 只有 "known" 补发, "unknown" 忽略 + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["call_id"], "known") + + # ---------- E14: reader dead fail-fast ---------- + def test_e14_turn_usage_total_tokens(self): + """E14: turn.completed 的 usage.totalTokens 优先于 tokenCount""" + t = self._new_translator() + events = t.translate(_event("turn.completed", { + "resultType": "success", + "tokenCount": 50, + "usage": {"totalTokens": 200, "contextWindow": 100000}, + })) + usage_events = [e for e in events if e["kind"] == "UsageDelta"] + self.assertEqual(usage_events[0]["used"], 200) # totalTokens 优先 + + +if __name__ == "__main__": + unittest.main(verbosity=2)