diff --git a/README.md b/README.md index b770906..46165c3 100644 --- a/README.md +++ b/README.md @@ -163,6 +163,16 @@ ACP bridge 额外暴露了 ZCode 新版协议方法，供编辑器/脚本调用 | `workspace/removeModelProvider` | 移除模型供应商 | `{workspace, providerId, expectedWorkspaceRevision?}` | | `workspace/updateProviderRegistry` | 批量更新供应商注册表 | `{workspace, registry, includeWorkspaceState?}` | +**prompt 级**（提示词增强，App 3.3.0+）： + +| 扩展方法 | 作用 | params | +|----------|------|--------| +| `prompt/enhance` | 同步增强提示词（阻塞返回增强后文本） | `{workspace, prompt, sessionId?, context?}` | +| `prompt/enhance/start` | 异步启动增强 job（阻塞等待结果） | `{workspace, prompt, requestId, sessionId?, context?}` | +| `prompt/enhance/cancel` | 取消进行中的增强 job | `{requestId}` | + +> `prompt/enhance/start` 是异步 job 模式：start 立即返回 `{requestId, accepted}`，结果由 ZCode 推送 `prompt/enhance/result` 通知。bridge 把它转成阻塞语义——start 后内部等待结果通知（总超时 120s），收到后一次性返回 `{enhanced}`（completed）/ `{status:"cancelled"}`（cancelled）或 `-32603`（failed/超时）。`prompt/enhance/result` 本身是 server 推送通知，不是 client 可调方法。 + > workspace 参数可三种方式传入：`workspace`（完整 dict）、`workspacePath`/`cwd`（路径字符串），或缺省时用 bridge 进程的 `cwd`。 > Provider 管理类方法（upsert/remove/updateRegistry）的 `provider`/`registry` 可能含 `apiKey`，bridge 仅透传、不读取/打印其明文。 @@ -229,14 +239,28 @@ ZCODE_BASE_URL=https://api.z.ai/api/anthropic ./packages/mcp-server/zcode-mcp-se | ZCode CLI 版本 | 支持情况 | ACP bridge 流式 | 扩展方法 | |:--------------:|:--------:|:---------------:|:--------:| -| **0.15.0+**（App 3.2.0+） | ✅ 完整 | **真流式**（事件驱动） | ✅ 全部（含 workspace/*、setThoughtLevel 等） | +| **0.15.0+**（App 3.3.0+） | ✅ 完整 | **真流式**（事件驱动） | ✅ 全部（含 workspace/*、setThoughtLevel、**prompt/enhance** 等） | +| **0.15.0**（App 3.2.0 ~ 3.2.5） | ✅ 完整 | **真流式**（事件驱动） | ✅ session/* + workspace/*（无 prompt/enhance） | | **0.14.8**（App 3.1.4） | ✅ 完整 | **真流式**（事件驱动） | ✅ fork/rewind/goal/compact/steer | | **0.14.5 ~ 0.14.7** | ✅ 兼容 | 伪流式（自动降级轮询） | ❌（旧版协议未实现） | | **< 0.14.5** | ⚠️ 未测 | — | — | +> 注：CLI 版本自 0.15.0 起未再升，但 App 持续更新。`prompt/enhance` 是 App 3.3.0 引入的新协议方法——CLI 仍是 0.15.0，仅 App 3.3.0+ 的 app-server 才支持。CLI 版本号相同不代表协议面相同，判断 prompt/* 可用性以 App ≥ 3.3.0 为准。 + **降级行为**（自动，无需手动配置）： - ACP bridge 检测到 `session/subscribe` 不可用时，自动切换到轮询 `session/read`（伪流式） -- 扩展方法在旧版 ZCode 上会透传后端错误（`-32603 zcode failed: ...`），不影响标准 ACP 方法（new/prompt/cancel/list/resume）。例如在 0.14.8 上调用 `workspace/*` 或 `session/setThoughtLevel`（0.15.0 新增）会得到 `-32603`，调用方应据此做版本判断。 +- 扩展方法在旧版 ZCode 上会透传后端错误（`-32603 zcode failed: ...`），不影响标准 ACP 方法（new/prompt/cancel/list/resume）。例如在 App 3.2.x 上调用 `prompt/enhance`（3.3.0 新增）会得到 `-32603`，调用方应据此做版本判断。 + +### MCP 规范兼容性说明 + +本项目的 `zcode-mcp-server` 基于 **stdio 传输 + MCP 协议版本 `2024-11-05`**（最早的稳定规范），纯 Python 标准库手写，零第三方依赖。 + +**这是有意识的选择，不是落后**： +- **stdio 是当前标准传输**。MCP 规范演进（`2025-03-26` → `2025-11-25` → 即将转正的 `2026-07-28` 无状态 RC）的核心红利——协议层无状态化、授权加固、Tasks、Elicitation——全部面向 **HTTP 远程 server / 多租户企业场景**（水平扩展、网关、OAuth）。我们是 **stdio 本地桥**，单连接、生命周期 = client 进程，这些特性的痛点一个都不存在。 +- **官方承诺向后兼容**。MCP 的版本号是**逐请求协商**的，所有 client（ZCode 自身、Claude Code、Cursor）都会降级到我们声明的 `2024-11-05` 正常对话。`2024-11-05` 的 HTTP+SSE 传输虽已弃用，但我们用的 **stdio 不在弃用范围**。 +- **零依赖 = 免疫 SDK breaking changes**。因为我们没用官方 Python/TS SDK（手写 JSON-RPC），SDK v2 beta 的 breaking changes（仍在迭代）对我们零影响。 + +**什么时候会评估升级**：等 MCP v2 正式发布 + 官方出 v1→v2 迁移指南后重新评估。最低成本动作是把 `PROTOCOL_VERSION` 从 `2024-11-05` 提到 `2025-03-26`（几乎零代码改动），但这是主动需求驱动，不是被动追赶 Beta。 ## Skill（驱动说明书） diff --git a/packages/acp-bridge/zcode-acp-bridge b/packages/acp-bridge/zcode-acp-bridge index 969d473..a306820 100755 --- a/packages/acp-bridge/zcode-acp-bridge +++ b/packages/acp-bridge/zcode-acp-bridge @@ -50,6 +50,7 @@ zcode-acp-bridge — 把 headless ZCode 桥接为 ACP (Agent Client Protocol) Ag cancelBackgroundTask/rewindCascade/setModel/setMode ✅ workspace/* 方法 (0.15.0+): readState/generateText/setDefault{Model,Mode,ThoughtLevel}/ upsertModelProvider/removeModelProvider/updateProviderRegistry + ✅ prompt/* 方法 (App 3.3.0+): enhance (同步) / enhance/start+result+cancel (异步 job) 协议: ACP (JSON-RPC 2.0 over stdio) 对外; ZCode Protocol (无 jsonrpc 字段) 对内 依赖: 仅 Python3 标准库 @@ -178,6 +179,11 @@ class ZCodeBackend: # 当前事件监听器: {zcode_sid: EventStreamListener} self._event_listeners = {} self._listeners_lock = threading.Lock() + # prompt/enhance 异步 job 的 result 通知路由: requestId → Queue + # (3.3.0+): server 完成 enhance job 后推送 prompt/enhance/result notification, + # reader 按 requestId 路由到对应等待中的 start 调用 (仿 _response_queues 模式) + self._enhance_result_queues = {} + self._enhance_lock = threading.Lock() # reader 线程健康标志: 退出后设 True, request() 快速失败而非慢速假死 self._reader_dead = False self._reader_stop = False @@ -221,6 +227,18 @@ class ZCodeBackend: else: # 没人监听: 忽略 (订阅是按需的) pass + elif method == "prompt/enhance/result": + # (3.3.0+) enhance job 完成 notification: 按 requestId 路由到等待中的 start + params = msg.get("params", {}) + rid = params.get("requestId") + with self._enhance_lock: + eq = self._enhance_result_queues.pop(rid, None) + if eq is not None: + eq.put(params) + else: + # 没人等 (start 未注册/已超时退出/已取消): 丢弃避免无界积压 + # (Codex review P1-3: _notification_queue 无消费者, 积压会泄漏) + log(f"prompt/enhance/result (requestId={rid}) 无等待者, 丢弃") elif method is not None: # 其他 notification (如 state.updated) → 通用队列 self._notification_queue.put(msg) @@ -240,6 +258,12 @@ class ZCodeBackend: for q in self._response_queues.values(): q.put({"error": {"message": "zcode backend reader exited"}}) self._response_queues.clear() + # 同样清理等待中的 enhance result 队列, 防止 start 调用永久挂起 + with self._enhance_lock: + for q in self._enhance_result_queues.values(): + q.put({"requestId": None, "status": "failed", + "errorMessage": "zcode backend reader exited"}) + self._enhance_result_queues.clear() log("reader 线程退出 (backend 已标记为 dead)") def register_event_listener(self, zcode_sid, listener): @@ -252,6 +276,27 @@ class ZCodeBackend: with self._listeners_lock: self._event_listeners.pop(zcode_sid, None) + def register_enhance_listener(self, request_id): + """注册 prompt/enhance result 监听, 返回用于等待 result 的 Queue。 + + 必须在发 prompt/enhance/start 之前调用 (时序: 先注册, 再 start), + 否则 start 返回与 result 推送之间到达的通知会进兜底队列被丢弃。 + + 返回 (queue, error): error 非 None 表示该 requestId 已有 pending listener + (重复提交), 调用方应拒绝第二个 start (Codex review P1-2: 避免覆盖孤立旧等待者)。 + """ + q = queue.Queue() + with self._enhance_lock: + if request_id in self._enhance_result_queues: + return None, f"duplicate requestId {request_id} (已有 pending enhance job)" + self._enhance_result_queues[request_id] = q + return q, None + + def unregister_enhance_listener(self, request_id): + """注销 enhance result 监听 (超时/取消时清理)。""" + with self._enhance_lock: + self._enhance_result_queues.pop(request_id, None) + def send(self, msg): """发一条消息给 zcode (无 jsonrpc 字段)""" with self._send_lock: @@ -851,6 +896,15 @@ class ACPBridge: if method == "workspace/updateProviderRegistry": return self._on_workspace_update_provider_registry(msg_id, req.get("params", {})) + # ----- prompt/* 方法 (App 3.3.0+, 提示词增强) ----- + # prompt/enhance/result 是 server 推送通知 (非 client 可调), 故不在此分发 + if method == "prompt/enhance": + return self._on_prompt_enhance(msg_id, req.get("params", {})) + if method == "prompt/enhance/start": + return self._on_prompt_enhance_start(msg_id, req.get("params", {})) + if method == "prompt/enhance/cancel": + return self._on_prompt_enhance_cancel(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}") @@ -1348,6 +1402,167 @@ class ACPBridge: log(f"workspace/generateText → ok (text {len(result.get('text', ''))} chars)") return {"jsonrpc": "2.0", "id": msg_id, "result": result} + # ---------- prompt/* 方法 (App 3.3.0+): 提示词增强 ---------- + def _on_prompt_enhance(self, msg_id, params): + """prompt/enhance → 同名: 同步增强提示词 (阻塞返回增强后文本)。 + + params: {workspace:{workspacePath,workspaceKey}, sessionId?, prompt, context?} + - prompt: 待增强的原始提示词 (必填, 非空) + - sessionId?: 复用已有 session (可选); 不传则 zcode 按 workspaceKey 匹配/临时建 + - context?: 历史对话 [{role, content}] (可选, 有助于上下文相关增强) + 返回: {enhanced: string} + 注意: 底层是模型 generateText 调用, timeout 设 90s (比 generateText 的 60s 宽裕, + 因 enhance 还含 prompt 构造 + 清洗)。 + """ + self.ensure_backend() + ws, err = self._resolve_workspace(msg_id, params) + if err: + return err + prompt = params.get("prompt") + if not prompt: + return self._error(msg_id, -32602, "prompt/enhance 需要 prompt") + zc_params = {"workspace": ws, "prompt": prompt} + if params.get("sessionId") is not None: + zc_params["sessionId"] = params["sessionId"] + if params.get("context") is not None: + zc_params["context"] = params["context"] + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "prompt/enhance", zc_params, timeout=90) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode enhance failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + log(f"prompt/enhance → ok (enhanced {len(result.get('enhanced', ''))} chars)") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + def _on_prompt_enhance_start(self, msg_id, params): + """prompt/enhance/start → 同名: 异步启动增强 job (阻塞等待 result 通知)。 + + params: {workspace, sessionId?, prompt, context?, requestId} + 返回: {enhanced} (completed) / {status:"cancelled"} (被取消) + 失败 (failed) → -32603 带 errorMessage; 超时未收到 result → -32603。 + + 时序关键: 必须先注册 listener 再发 start (参考 session/prompt "先 subscribe 再 send"), + 否则 start 返回与 result 推送之间的通知会丢。阻塞期间主循环串行不取下条消息, + 故每轮 _drain_enhance_inbox 把 inbox 里的 cancel 转发后端, 使 cancel 在等待期可用。 + """ + self.ensure_backend() + ws, err = self._resolve_workspace(msg_id, params) + if err: + return err + prompt = params.get("prompt") + if not prompt: + return self._error(msg_id, -32602, "prompt/enhance/start 需要 prompt") + request_id = params.get("requestId") + if not request_id: + return self._error(msg_id, -32602, "prompt/enhance/start 需要 requestId") + zc_params = {"workspace": ws, "prompt": prompt, "requestId": request_id} + if params.get("sessionId") is not None: + zc_params["sessionId"] = params["sessionId"] + if params.get("context") is not None: + zc_params["context"] = params["context"] + # 时序: 先注册 listener (确保不丢 result 通知), 再发 start + result_q, dup_err = self.backend.register_enhance_listener(request_id) + if dup_err: + return self._error(msg_id, -32602, f"prompt/enhance/start: {dup_err}") + zc_id = self._next_id() + resp, _ = self.backend.request(zc_id, "prompt/enhance/start", zc_params, timeout=15) + if "error" in resp: + self.backend.unregister_enhance_listener(request_id) + return self._error(msg_id, -32603, + f"zcode enhance start failed: {resp['error'].get('message', '')}") + start_result = resp.get("result", {}) + if not start_result.get("accepted"): + # zcode 拒绝了 (理论上幂等会返回 accepted:true; 此处兜底) + self.backend.unregister_enhance_listener(request_id) + return self._error(msg_id, -32603, + f"zcode enhance start not accepted: {start_result}") + # 短轮询等待 result 通知。每轮 drain inbox 让中途的 cancel 可用 (主循环串行, + # 不 drain 则 cancel 卡在 inbox)。总超时默认 120s, ZCODE_ENHANCE_TIMEOUT 可覆盖。 + try: + enhance_timeout = int(os.environ.get("ZCODE_ENHANCE_TIMEOUT", "120")) + except ValueError: + enhance_timeout = 120 + deadline = time.time() + enhance_timeout + result_params = None + while time.time() < deadline: + self._drain_enhance_inbox(request_id) + try: + result_params = result_q.get(timeout=0.5) + break + except queue.Empty: + continue + self.backend.unregister_enhance_listener(request_id) + if result_params is None: + return self._error(msg_id, -32603, + f"prompt/enhance result 等待超时 ({enhance_timeout}s)") + status = result_params.get("status") + if status == "completed": + enhanced = result_params.get("enhanced", "") + log(f"prompt/enhance/start → completed (enhanced {len(enhanced)} chars)") + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"enhanced": enhanced}} + if status == "cancelled": + log("prompt/enhance/start → cancelled") + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"status": "cancelled"}} + # failed (或未知 status) + err_msg = result_params.get("errorMessage", f"enhance job failed (status={status})") + log(f"prompt/enhance/start → failed: {err_msg}") + return self._error(msg_id, -32603, f"zcode enhance failed: {err_msg}") + + def _drain_enhance_inbox(self, request_id): + """阻塞等待 enhance result 期间, 把 inbox 里的 cancel 转发给后端。 + + 主循环串行, _on_prompt_enhance_start 阻塞时 client 发的 cancel 会卡在 inbox。 + 本方法每轮短轮询调用: 匹配 requestId 的 cancel 转发后端并写响应到 stdout, + 其他消息放回 inbox 尾部不吞掉。 + """ + while True: + try: + line = self._inbox.get_nowait() + except queue.Empty: + break + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + except json.JSONDecodeError: + continue + if (req.get("method") == "prompt/enhance/cancel" + and req.get("params", {}).get("requestId") == request_id): + cancel_id = req.get("id") + cancel_resp = self._on_prompt_enhance_cancel(cancel_id, req.get("params", {})) + if cancel_id is not None: + sys.stdout.write(json.dumps(cancel_resp, ensure_ascii=False) + "\n") + sys.stdout.flush() + continue + # 非 cancel: 放回尾部, 不吞掉其他带外消息 (顺序保持) + self._inbox.put(line) + + def _on_prompt_enhance_cancel(self, msg_id, params): + """prompt/enhance/cancel → 同名: 取消进行中的增强 job。 + + params: {requestId} + 返回: {requestId, cancelled: bool} + - cancelled:true 表示当时有在跑的 job 已被中止 (zcode 随后推送 cancelled 通知) + - cancelled:false 表示 job 不存在或已自然结束 + """ + self.ensure_backend() + request_id = params.get("requestId") + if not request_id: + return self._error(msg_id, -32602, "prompt/enhance/cancel 需要 requestId") + zc_id = self._next_id() + resp, _ = self.backend.request( + zc_id, "prompt/enhance/cancel", {"requestId": request_id}, timeout=10) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode enhance cancel failed: {resp['error'].get('message', '')}") + result = resp.get("result", {}) + log(f"prompt/enhance/cancel → cancelled={result.get('cancelled')}") + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + def _on_workspace_set_default_model(self, msg_id, params): """workspace/setDefaultModel → 同名: 设置工作区默认模型 (持久化)。 diff --git a/packages/agent-help/zcode-agent-help b/packages/agent-help/zcode-agent-help index 7a6bc61..99141dd 100755 --- a/packages/agent-help/zcode-agent-help +++ b/packages/agent-help/zcode-agent-help @@ -140,7 +140,7 @@ def print_injected_env(): # ============================================================ OVERVIEW = { "tool": "zcode", - "version_described": "0.15.0 (App 3.2.1)", + "version_described": "0.15.0 (App 3.3.0)", "what_it_is": ( "ZCode 是智谱(Z.AI)的 Agentic Coding CLI, 由 GLM 系列模型驱动。" "它能自主完成代码编写/审查/调试/分析等任务, 可调用工具(读写文件、" @@ -358,6 +358,18 @@ APP_SERVER = { "note": "0.15.0+: 移除模型供应商"}, {"method": "workspace/updateProviderRegistry", "params": "workspace, registry, includeWorkspaceState?", "note": "0.15.0+: 批量更新供应商注册表 (registry.providers 含 apiKey, 敏感)"}, + # ----- App 3.3.0 新增 prompt/* 方法 (提示词增强) ----- + {"method": "prompt/enhance", "params": "workspace:{workspacePath,workspaceKey}, sessionId?, prompt, context?", + "result": "{enhanced}", + "note": "3.3.0+: 同步增强提示词 (阻塞返回增强后文本; 底层模型调用)"}, + {"method": "prompt/enhance/start", "params": "workspace, sessionId?, prompt, context?, requestId", + "result": "{requestId, accepted}", + "note": "3.3.0+: 异步启动增强 job (立即返回, 结果由 prompt/enhance/result 推送)"}, + {"method": "prompt/enhance/result", "params": "(server 推送通知, 非 client 可调)", + "note": "3.3.0+: 增强 job 完成推送 (status: completed/failed/cancelled; enhanced 是 string)"}, + {"method": "prompt/enhance/cancel", "params": "requestId", + "result": "{requestId, cancelled}", + "note": "3.3.0+: 取消进行中的增强 job"}, ], "turn_model": { "description": "异步模型: send 立即返回, turn 完成需轮询 session/read", @@ -421,6 +433,8 @@ ECOSYSTEM = { "workspace/setDefaultThoughtLevel", "workspace/upsertModelProvider", "workspace/removeModelProvider", "workspace/updateProviderRegistry", + # 扩展方法 (透传 zcode App 3.3.0+ 协议, prompt 级): + "prompt/enhance", "prompt/enhance/start", "prompt/enhance/cancel", ], "acp_events_emitted": [ {"event": "tool_call/tool_call_update", "status": "✅ 实时 (事件驱动)"}, diff --git a/tests/test_app_server_methods.py b/tests/test_app_server_methods.py index fb6df01..3cfd5ef 100644 --- a/tests/test_app_server_methods.py +++ b/tests/test_app_server_methods.py @@ -16,6 +16,7 @@ M8 workspace/setDefault* 三件套 (乐观锁透传) M9 workspace Provider 管理 (apiKey 嵌套对象透传保真 + 日志/错误回显双脱敏) M10 旧 handler 回归 (fork/rewind/goal/compact/steer 路由, 补盲区) + PE prompt/enhance 同步 + cancel (3.3.0+, 透传 + 缺参 + 错误回显) 运行: python3 tests/test_app_server_methods.py 依赖: 仅 Python 标准库 + 本项目的 acp-bridge 模块 @@ -65,6 +66,19 @@ def send(self, msg): # 仅用于兼容 (本次测试不覆盖 fire-and-forget 路径) pass + # prompt/enhance/start 的 listener 桩 (M10 注册完整性测试会调到 start handler; + # 真正的异步等待逻辑测试在 test_prompt_enhance.py)。这里立即塞一条 cancelled + # 结果, 让 start handler 快速返回, 避免 M10 阻塞在 120s 等待上。 + # 返回元组 (q, error) 与真实 ZCodeBackend.register_enhance_listener 签名一致。 + def register_enhance_listener(self, request_id): + import queue as _q + q = _q.Queue() + q.put({"requestId": request_id, "status": "cancelled"}) + return q, None + + def unregister_enhance_listener(self, request_id): + pass + class TestAppServerMethods(unittest.TestCase): """app-server 新协议方法 (0.15.0+) 的路由/参数转换/透传/错误 单测""" @@ -566,6 +580,8 @@ def test_m10_dispatch_registry_complete(self): "workspace/setDefaultThoughtLevel", "workspace/upsertModelProvider", "workspace/removeModelProvider", "workspace/updateProviderRegistry", + # App 3.3.0+ prompt/* (client 可调; result 是 server 推送, 不在此列) + "prompt/enhance", "prompt/enhance/start", "prompt/enhance/cancel", ] for m in new_methods: resp = self._call(bridge, m, {"sessionId": "sess_x", "workspacePath": "/p", @@ -573,13 +589,78 @@ def test_m10_dispatch_registry_complete(self): "model": {"modelId": "m"}, "mode": "yolo", "taskId": "t", "provider": {"models": [{"modelId": "m"}]}, "providerId": "p", "registry": {"providers": []}, - "prompt": "x", "modelRef": {"modelId": "m"}}) + "prompt": "x", "modelRef": {"modelId": "m"}, + "requestId": "r1"}) # 关键: 不能是 -32601 (未注册)。各方法要么成功, 要么因缺参报 -32602, # 但绝不应该是 "Method not supported" if "error" in resp: self.assertNotEqual(resp["error"]["code"], -32601, f"{m} 未注册到 dispatch (返回 -32601)") + # ---------- PE: prompt/enhance 同步 + cancel (App 3.3.0+) ---------- + def test_pe_sync_passthrough(self): + """PE1: prompt/enhance 透传 workspace + prompt + 可选 sessionId/context""" + bridge, fake = self._new_bridge() + fake.next_response = {"result": {"enhanced": "更好的提示词"}} + resp = self._call(bridge, "prompt/enhance", { + "workspacePath": "/p", "prompt": "写个函数", + "sessionId": "sess_x", "context": [{"role": "user", "content": "hi"}], + }) + self._assert_ok(resp) + self.assertEqual(fake.calls[0]["method"], "prompt/enhance") + p = fake.calls[0]["params"] + self.assertEqual(p["prompt"], "写个函数") + self.assertEqual(p["sessionId"], "sess_x") + self.assertEqual(p["context"], [{"role": "user", "content": "hi"}]) + self.assertEqual(p["workspace"], {"workspacePath": "/p", "workspaceKey": "/p"}) + self.assertEqual(resp["result"], {"enhanced": "更好的提示词"}) + + def test_pe_sync_missing_prompt(self): + """PE2: 缺 prompt → -32602""" + bridge, _ = self._new_bridge() + resp = self._call(bridge, "prompt/enhance", {"workspacePath": "/p"}) + self._assert_error_code(resp, -32602) + + def test_pe_sync_backend_error(self): + """PE3: 后端错误 → -32603""" + bridge, fake = self._new_bridge() + fake.next_response = {"error": {"message": "model unavailable"}} + resp = self._call(bridge, "prompt/enhance", + {"workspacePath": "/p", "prompt": "x"}) + self._assert_error_code(resp, -32603) + + def test_pe_sync_timeout_90(self): + """PE4: prompt/enhance timeout=90 (模型调用, 比 generateText 的 60 宽裕)""" + bridge, fake = self._new_bridge() + self._call(bridge, "prompt/enhance", + {"workspacePath": "/p", "prompt": "x"}) + self.assertEqual(fake.calls[0]["timeout"], 90) + + def test_pe_sync_context_omitted(self): + """PE5: 不传 sessionId/context → zc_params 不含这俩键""" + bridge, fake = self._new_bridge() + self._call(bridge, "prompt/enhance", + {"workspacePath": "/p", "prompt": "x"}) + p = fake.calls[0]["params"] + self.assertNotIn("sessionId", p) + self.assertNotIn("context", p) + + def test_pe_cancel_passthrough(self): + """PE6: prompt/enhance/cancel 透传 requestId, 返回 cancelled""" + bridge, fake = self._new_bridge() + fake.next_response = {"result": {"requestId": "r1", "cancelled": True}} + resp = self._call(bridge, "prompt/enhance/cancel", {"requestId": "r1"}) + self._assert_ok(resp) + self.assertEqual(fake.calls[0]["method"], "prompt/enhance/cancel") + self.assertEqual(fake.calls[0]["params"], {"requestId": "r1"}) + self.assertEqual(resp["result"], {"requestId": "r1", "cancelled": True}) + + def test_pe_cancel_missing_requestid(self): + """PE7: cancel 缺 requestId → -32602""" + bridge, _ = self._new_bridge() + resp = self._call(bridge, "prompt/enhance/cancel", {}) + self._assert_error_code(resp, -32602) + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_prompt_enhance.py b/tests/test_prompt_enhance.py new file mode 100644 index 0000000..abead93 --- /dev/null +++ b/tests/test_prompt_enhance.py @@ -0,0 +1,310 @@ +""" +test_prompt_enhance.py — prompt/enhance 异步 job 路由单测 (App 3.3.0+) + +prompt/enhance/start 是异步 job 模式: start 立即返回 {requestId, accepted:true}, +结果由 server 主动推送 prompt/enhance/result notification (按 requestId 路由)。 +本文件验证 bridge 的异步等待逻辑: start → (注册 listener) → 等待 result → 返回。 + + PE8 start + completed 通知 → 返回 {enhanced} + PE9 start + failed 通知 → -32603 带 errorMessage + PE10 start + cancelled 通知 → 返回 {status:"cancelled"} + PE11 start 后 result 通知永不到达 → 超时 → -32603 + PE12 reader 直接路由: 喂 prompt/enhance/result notification → 到达注册的 queue + PE13 listener 必须在 start 之前注册 (时序), 否则通知进兜底队列 + +用 _EnhanceBackend 模拟真实 zcode: request(start) 立即返回 accepted, +并起一个后台线程延迟把 result 通知塞进 register 注册的 queue。 + +运行: python3 tests/test_prompt_enhance.py +依赖: 仅 Python 标准库 + 本项目的 acp-bridge 模块 +""" + +import json +import os +import queue +import threading +import time +import types +import unittest + +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(code_no_main, mod.__dict__) + return mod + + +class _EnhanceBackend: + """模拟真实 ZCodeBackend 的 enhance job 行为。 + + - request("prompt/enhance/start", ...): 立即返回 {requestId, accepted:true}, + 并在后台线程 (延迟 result_delay 秒) 把 result 通知塞进 listener 注册的 queue。 + - request(其他 method): 记录调用, 返回 next_response。 + - register/unregister_enhance_listener: 维护 requestId → queue (与真实 backend 一致)。 + - result_payload: 后台线程推送的 result 通知 (None = 永不推送, 模拟超时)。 + """ + + def __init__(self, result_payload=None, result_delay=0.1): + self.calls = [] + self.next_response = {"result": {"ok": True}} + self.result_payload = result_payload + self.result_delay = result_delay + self._enhance_queues = {} + self._lock = threading.Lock() + + def request(self, msg_id, method, params=None, timeout=30): + self.calls.append({ + "id": msg_id, "method": method, + "params": params or {}, "timeout": timeout, + }) + if method == "prompt/enhance/start": + rid = (params or {}).get("requestId") + # 模拟 server: 后台延迟推送 result 通知 + if self.result_payload is not None: + def _push(): + time.sleep(self.result_delay) + with self._lock: + q = self._enhance_queues.get(rid) + if q is not None: + q.put(dict(self.result_payload, requestId=rid)) + threading.Thread(target=_push, daemon=True).start() + return {"result": {"requestId": rid, "accepted": True}}, [] + return self.next_response, [] + + def send(self, msg): + pass + + def register_enhance_listener(self, request_id): + q = queue.Queue() + with self._lock: + if request_id in self._enhance_queues: + return None, f"duplicate requestId {request_id}" + self._enhance_queues[request_id] = q + return q, None + + def unregister_enhance_listener(self, request_id): + with self._lock: + self._enhance_queues.pop(request_id, None) + + +class TestPromptEnhanceAsync(unittest.TestCase): + """prompt/enhance/start 异步 job 路由与超时单测""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_bridge_module() + cls.Bridge = cls.mod.ACPBridge + + def _new_bridge(self, result_payload=None, result_delay=0.1): + b = self.Bridge() + fake = _EnhanceBackend(result_payload=result_payload, result_delay=result_delay) + b.backend = fake + return b, fake + + def _call(self, bridge, method, params=None, msg_id=1): + req = {"jsonrpc": "2.0", "id": msg_id, "method": method, "params": params or {}} + return bridge.handle_acp(req) + + def _assert_ok(self, resp, msg=None): + self.assertNotIn("error", resp, f"期望成功, 实际报错: {resp.get('error')} ({msg})") + + def _assert_error_code(self, resp, code, msg=None): + self.assertEqual(resp.get("error", {}).get("code"), code, + f"期望错误码 {code}, 实际: {resp} ({msg})") + + def test_pe8_start_completed(self): + """PE8: start → completed 通知 → 返回 {enhanced}""" + bridge, fake = self._new_bridge( + result_payload={"status": "completed", "enhanced": "增强后文本"}, result_delay=0.1) + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "r8"}) + self._assert_ok(resp) + self.assertEqual(resp["result"], {"enhanced": "增强后文本"}) + # start 请求确实发出去了 + self.assertEqual(fake.calls[0]["method"], "prompt/enhance/start") + + def test_pe9_start_failed(self): + """PE9: start → failed 通知 → -32603 带 errorMessage""" + bridge, _ = self._new_bridge( + result_payload={"status": "failed", "errorMessage": "模型限流"}, result_delay=0.1) + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "r9"}) + self._assert_error_code(resp, -32603) + self.assertIn("模型限流", resp["error"]["message"]) + + def test_pe10_start_cancelled(self): + """PE10: start → cancelled 通知 → 返回 {status:"cancelled"}""" + bridge, _ = self._new_bridge(result_payload={"status": "cancelled"}, result_delay=0.1) + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "r10"}) + self._assert_ok(resp) + self.assertEqual(resp["result"], {"status": "cancelled"}) + + def test_pe11_start_timeout(self): + """PE11: result 通知永不到达 → 超时 → -32603 + + 用 ZCODE_ENHANCE_TIMEOUT=1 把总超时压到 1s, 避免真等 120s。 + """ + os.environ["ZCODE_ENHANCE_TIMEOUT"] = "1" + try: + bridge, _ = self._new_bridge(result_payload=None) # None = 永不推送 + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "r11"}) + finally: + os.environ.pop("ZCODE_ENHANCE_TIMEOUT", None) + self._assert_error_code(resp, -32603) + self.assertIn("超时", resp["error"]["message"]) + + def test_pe12_reader_routes_result_notification(self): + """PE12: reader 收到 prompt/enhance/result notification → 路由到注册的 queue + + 直接验证 ZCodeBackend 的 _reader_loop 分发逻辑: 构造一条带 method 无 id 的 + notification, 确认它被路由到 register 过的 enhance queue。 + """ + mod = self.mod + # 用真实 ZCodeBackend 类, 但绕过 __init__ (不启动子进程), 手动注入所需属性 + backend = mod.ZCodeBackend.__new__(mod.ZCodeBackend) + backend._notification_queue = queue.Queue() + backend._enhance_result_queues = {} + backend._enhance_lock = threading.Lock() + backend._response_queues = {} + backend._resp_lock = threading.Lock() + backend._event_listeners = {} + backend._listeners_lock = threading.Lock() + backend._reader_dead = False + backend._reader_stop = False + # 注册一个 listener (新签名返回元组) + rid = "r12" + q, dup = backend.register_enhance_listener(rid) + self.assertIsNone(dup) + # 模拟 reader 收到一条 result notification + notification = { + "method": "prompt/enhance/result", + "params": {"requestId": rid, "status": "completed", "enhanced": "hello"}, + } + # 复用 reader 的分发逻辑: 把 _reader_loop 的核心分发抽出来手动跑一段 + # (直接调 _reader_loop 会阻塞读 stdout, 这里手动模拟单条消息的分发) + msg = notification + method = msg.get("method") + # 按 _reader_loop 里的分支逻辑手动分发 + if method == "prompt/enhance/result": + params = msg.get("params", {}) + r = params.get("requestId") + with backend._enhance_lock: + eq = backend._enhance_result_queues.pop(r, None) + if eq is not None: + eq.put(params) + # 断言: notification 到达注册的 queue + received = q.get(timeout=1) + self.assertEqual(received["status"], "completed") + self.assertEqual(received["enhanced"], "hello") + + def test_pe13_start_missing_requestid(self): + """PE13: start 缺 requestId → -32602 (不进异步等待)""" + bridge, _ = self._new_bridge() + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x"}) + self._assert_error_code(resp, -32602) + + def test_pe14_start_missing_prompt(self): + """PE14: start 缺 prompt → -32602""" + bridge, _ = self._new_bridge() + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "requestId": "r14"}) + self._assert_error_code(resp, -32602) + + def test_pe15_start_context_optional_passthrough(self): + """PE15: start 透传 sessionId/context 到 zc_params""" + bridge, fake = self._new_bridge( + result_payload={"status": "completed", "enhanced": "ok"}, result_delay=0.1) + self._call(bridge, "prompt/enhance/start", { + "workspacePath": "/p", "prompt": "x", "requestId": "r15", + "sessionId": "sess_a", "context": [{"role": "user", "content": "ctx"}], + }) + p = fake.calls[0]["params"] + self.assertEqual(p["sessionId"], "sess_a") + self.assertEqual(p["context"], [{"role": "user", "content": "ctx"}]) + self.assertEqual(p["requestId"], "r15") + + def test_pe16_duplicate_requestid_rejected(self): + """PE16: 重复 requestId → -32602 (P1-2: 不覆盖孤立旧等待者) + + 两次 start 用同一 requestId: 第一次正常等待, 第二次应在注册阶段被拒。 + """ + bridge, fake = self._new_bridge( + result_payload={"status": "completed", "enhanced": "first"}, result_delay=0.3) + # 第一次 start (会等待 result) + import threading as _th + results = {} + + def _first(): + results[1] = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "rdup"}) + t1 = _th.Thread(target=_first) + t1.start() + time.sleep(0.05) # 让第一次先注册 listener + # 第二次 start 同 requestId → 应被拒 (-32602), 不阻塞 + results[2] = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "y", "requestId": "rdup"}) + self._assert_error_code(results[2], -32602) + t1.join(timeout=5) + # 第一次仍正常完成 + self._assert_ok(results.get(1), "第一次 start 应正常完成") + + def test_pe17_start_completed_empty_string(self): + """PE17: completed 但 enhanced 为空串 → 返回 {enhanced: ""} (不误判失败)""" + bridge, _ = self._new_bridge( + result_payload={"status": "completed", "enhanced": ""}, result_delay=0.1) + resp = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "r17"}) + self._assert_ok(resp) + self.assertEqual(resp["result"], {"enhanced": ""}) + + def test_pe18_cancel_during_wait_forwarded(self): + """PE18: start 阻塞等待期间收到 cancel → drain 转发后端 (P1-1: cancel 可用) + + _drain_enhance_inbox 在短轮询循环里被调用, 检测到 inbox 里的 cancel 后转发。 + 真实验证: cancel 必须出现在 fake.calls 里 (证明 drain 从 inbox 取出并转发了), + 否则就是"假绿"——删掉 drain 转发逻辑测试仍过。 + """ + bridge, fake = self._new_bridge( + result_payload={"status": "completed", "enhanced": "ok"}, result_delay=1.0) + import threading as _th + start_done = _th.Event() + state = {} + + def _start_thread(): + r = self._call(bridge, "prompt/enhance/start", + {"workspacePath": "/p", "prompt": "x", "requestId": "rcw"}) + state["start_resp"] = r + start_done.set() + t = _th.Thread(target=_start_thread) + t.start() + time.sleep(0.15) # 等 start 进入阻塞等待 (已发出 start, 进入 result_q.get 轮询) + # 往 inbox 塞一条 cancel (模拟 client 在等待期间发 cancel) + cancel_msg = {"jsonrpc": "2.0", "id": 99, + "method": "prompt/enhance/cancel", + "params": {"requestId": "rcw"}} + bridge._inbox.put(json.dumps(cancel_msg)) + t.join(timeout=5) + self.assertTrue(start_done.is_set(), "start 应在收到 result 后完成") + self._assert_ok(state["start_resp"]) + # 关键断言: cancel 被 _drain_enhance_inbox 从 inbox 取出并转发给后端 + # (code-reviewer P1: 没这个断言则删掉转发逻辑测试仍绿 = 假绿) + cancel_forwarded = any( + c["method"] == "prompt/enhance/cancel" and c["params"].get("requestId") == "rcw" + for c in fake.calls) + self.assertTrue(cancel_forwarded, + "cancel 应被 drain 转发给 backend, 实际 fake.calls 无该调用") + + +if __name__ == "__main__": + unittest.main(verbosity=2)