diff --git a/README.md b/README.md index da81552..b770906 100644 --- a/README.md +++ b/README.md @@ -172,13 +172,64 @@ ACP bridge 额外暴露了 ZCode 新版协议方法，供编辑器/脚本调用 - `--prompt` 创建的会话可被 `session/list` 看到、被 `--resume` 或 ACP resume 恢复 - 反之亦然 +## 凭证与限流（自动化集成必读） + +### Model ID 格式（canonical） + +**canonical model id = `~/.zcode/v2/config.json` 里 `models` 的 key 原样**（如 `GLM-5.2`），**不加 provider 前缀**。`shared/credentials.py`、MCP server、ACP bridge 三处统一用原始 id。实测 `zai/GLM-5.2` 也兼容，但非 canonical，本项目不使用。 + +### 凭证注入：显式环境变量优先 + +三个组件注入凭证的合并顺序为 `{**config_creds, **os.environ}`——**已显式设置的环境变量覆盖 config 读出的值**。便于不改 config 临时调试/覆盖： + +```bash +# 临时用另一个模型跑 ACP bridge（覆盖 config 的 GLM-5.2） +ZCODE_MODEL=GLM-5-Turbo ./packages/acp-bridge/zcode-acp-bridge + +# 临时覆盖 MCP server 的 baseURL +ZCODE_BASE_URL=https://api.z.ai/api/anthropic ./packages/mcp-server/zcode-mcp-server +``` + +诊断"实际会注入哪些凭证"（不改 config、apiKey 脱敏）： +```bash +./packages/agent-help/zcode-agent-help --print-injected-env +# 输出每个 key 的来源（config vs env）+ 最终值（apiKey 脱敏） +``` + +### ZCODE_BASE_URL 残留自动检测（切换过 plan 的用户） + +**背景**：ZCode App 切换过订阅 plan 的用户，旧 plan 的 baseURL 会残留在子进程环境（即使该 plan 已失效）。例如 `ZCODE_BASE_URL=https://zcode.z.ai`（start-plan 残留）与 config 当前 enabled 的 `zai-coding-plan`（`api.z.ai`）不一致，会导致 headless 调用打到错误端点（404）。 + +**自动自愈**：bridge 启动时检测——若 env 的 `ZCODE_BASE_URL` 指向 config 里**另一个 provider** 的官方 endpoint（host 匹配），判定为 App 注入的残留 → **自动用 config enabled provider 的值** + stderr 告警。用户自建/代理 endpoint（不在 config 任何 provider 里）则正常尊重 env（issue #3 调试场景不受影响）。 + +```bash +# 用 --print-injected-env 看是否检测到残留 (标 🚫) +./packages/agent-help/zcode-agent-help --print-injected-env +``` + +### 限流与重试（MCP server） + +`zcode_review` 调用 headless zcode 做 PR 审查时，多个 gate 并发可能触发 provider 限流。MCP server 做了三层防护（issue #3）： + +| 机制 | 行为 | 配置 | +|------|------|------| +| **进程级文件锁** | 多个 MCP client 并发调用时，串行化 headless review，防并发触发限流 | `ZCODE_BRIDGE_REVIEW_LOCK=0` 关闭 | +| **provider 错误解析** | 识别 429 / 1302 / `Too Many Requests` / `请求过于频繁` / `retry-after`，区分限流/配额/其他 | — | +| **有限重试 + 退避** | 仅对**限流**错误重试（配额/Unauthorized 不重试），退避用 retry-after 或指数退避（`2^n+1`） | `ZCODE_BRIDGE_MAX_RETRIES`（默认 3） | + +注意：zcode 内部已有自己的指数退避重试（`_retryWithExponentialBackoff`），MCP 层的重试是补充，默认保守（max 3）。 + +### 聚焦审查 prompt 建议 + +为降低限流风险，自动化审查时建议：把 diff/证据作为 `--attach` 附件，prompt 写明"不要 spawn 子代理、不要探索文件系统、只基于附件推理"。 + ## 版本兼容性 本项目兼容以下 ZCode 版本，**对旧版完全向后兼容**： | ZCode CLI 版本 | 支持情况 | ACP bridge 流式 | 扩展方法 | |:--------------:|:--------:|:---------------:|:--------:| -| **0.15.0+**（App 3.2.0） | ✅ 完整 | **真流式**（事件驱动） | ✅ 全部（含 workspace/*、setThoughtLevel 等） | +| **0.15.0+**（App 3.2.0+） | ✅ 完整 | **真流式**（事件驱动） | ✅ 全部（含 workspace/*、setThoughtLevel 等） | | **0.14.8**（App 3.1.4） | ✅ 完整 | **真流式**（事件驱动） | ✅ fork/rewind/goal/compact/steer | | **0.14.5 ~ 0.14.7** | ✅ 兼容 | 伪流式（自动降级轮询） | ❌（旧版协议未实现） | | **< 0.14.5** | ⚠️ 未测 | — | — | diff --git a/packages/acp-bridge/zcode-acp-bridge b/packages/acp-bridge/zcode-acp-bridge index 798f178..969d473 100755 --- a/packages/acp-bridge/zcode-acp-bridge +++ b/packages/acp-bridge/zcode-acp-bridge @@ -101,6 +101,52 @@ def load_zcode_credentials(): return {} +def _merge_env_with_creds(creds): + """合并 config 凭证与环境变量: 显式非空 env 优先, 空串视为未设置; + 且自动检测/自愈 App 注入的残留 baseURL (issue #3)。 + + 源自 shared/credentials.py.merge_env_with_creds (单文件可独立运行故内嵌)。 + 残留检测: env 的 ZCODE_BASE_URL 若是 config 里"另一个 provider"的官方 endpoint + (App 切换 plan 后的残留), 自愈用 config 值; 用户自建 endpoint 则尊重 env。 + """ + merged = {**os.environ, **creds} + for k in ("ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY"): + v = os.environ.get(k) + if v: # 仅非空才覆盖 + merged[k] = v + # 残留检测 + env_bu = os.environ.get("ZCODE_BASE_URL", "") + config_bu = creds.get("ZCODE_BASE_URL", "") + if env_bu and config_bu and env_bu != config_bu: + all_hosts = set() + try: + from urllib.parse import urlparse as _up + with open(ZCODE_CREDS_PATH) as _f: + _cfg = json.load(_f) + for _p in _cfg.get("provider", {}).values(): + _bu = _p.get("options", {}).get("baseURL", "") + if _bu: + try: + _h = _up(_bu) + if _h.hostname: + all_hosts.add(f"{_h.scheme}://{_h.netloc}" if _h.scheme else _h.netloc) + except Exception: + pass + _eh = _up(env_bu) + env_host = f"{_eh.scheme}://{_eh.netloc}" if (_eh.scheme and _eh.hostname) else None + _ch = _up(config_bu) + config_host = f"{_ch.scheme}://{_ch.netloc}" if (_ch.scheme and _ch.hostname) else None + except Exception: + env_host = config_host = None + # env host 等于某 config provider host, 但不等于 enabled 那个 → 残留 + if env_host and config_host and env_host != config_host and env_host in all_hosts: + merged["ZCODE_BASE_URL"] = config_bu + log(f"⚠ ZCODE_BASE_URL 残留检测: env='{env_bu}' 是 config 里另一个 provider " + f"的 endpoint, 与 enabled provider (config='{config_bu}') 不一致。已自动用 " + f"config 值 (旧 plan 残留)。") + return merged + + # ============================================================ # ZCode app-server 的 client 封装 # ============================================================ @@ -114,7 +160,9 @@ class ZCodeBackend: """ def __init__(self): - env = {**os.environ, **load_zcode_credentials()} + # 显式非空环境变量优先 (issue #3 子项2): ZCODE_MODEL=xxx 临时覆盖生效, + # 但空串 env 视为未设置 (保留 config 值, 避免误清空)。 + env = _merge_env_with_creds(load_zcode_credentials()) self.proc = subprocess.Popen( [ZCODE_BIN, "app-server", "--stdio"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, @@ -1631,7 +1679,8 @@ class ACPBridge: """ translator = EventTranslator() t0 = time.time() - emitted_text = False # 本轮是否已通过事件发过文本 (用于决定兜底) + emitted_text = False # 本轮是否已通过事件发过文本 (决定兜底回复发不发) + emitted_output = False # 本轮是否已发过任何实质输出 (text/tool/patch, 决定失败检测) last_event_time = time.time() last_stall_check = 0 # 上次停滞检查的时间戳 (可重复检查) @@ -1663,15 +1712,25 @@ class ACPBridge: # 停滞收尾: 不调 differ.diff() (differ 不知道事件路径已发过什么, # 会重复发 TextDelta/tool 事件)。只做兜底: 若事件路径未发文本, # 用 _fetch_last_reply 取最后回复。 + stall_reply = None if not emitted_text: - reply_text = self._fetch_last_reply(zcode_sid, differ) - if reply_text: + stall_reply = self._fetch_last_reply(zcode_sid, differ) + if stall_reply: self._send_acp_notification("session/update", { "sessionId": acp_sid, "update": {"sessionUpdate": "agent_message_chunk", - "content": {"type": "text", "text": reply_text}, + "content": {"type": "text", "text": stall_reply}, "messageId": chunk_msg_id}, }) + # 失败检测 (issue #3 子项3d / Codex P1#3): 停滞收尾时若事件路径既没收到 + # turn.failed (否则已正常返回), 也无任何实质输出 (text/tool/patch), + # 疑似 turn 失败 (如 turn.completed 事件丢失 + provider 错误)。显式告警而非静默成功。 + # 用 emitted_output (含 tool/patch) 而非 emitted_text, 避免纯工具 turn 误判。 + if not emitted_output and not stall_reply: + log(" ⚠ 停滞收尾: turn 完成但事件路径无输出, 疑似失败") + return self._error(msg_id, -32603, + "zcode turn 完成但无输出 (停滞收尾疑似失败; " + "可能 turn 事件丢失或 provider 错误)") return {"jsonrpc": "2.0", "id": msg_id, "result": {"stopReason": "end_turn"}} continue @@ -1681,9 +1740,13 @@ class ACPBridge: # 翻译事件 → ACP 事件 dict 列表 acp_events = translator.translate(event) for ev in acp_events: - # 标记: 是否已通过 model.streaming 发过文本 + # 标记: 是否已通过 model.streaming 发过文本 (决定兜底回复) if ev.get("kind") == "TextDelta": emitted_text = True + # 标记: 是否已发过任何实质输出 text/tool/patch (决定失败检测, Codex P1#3: + # 纯工具 turn 有 ToolCallNew 但无 TextDelta, 失败检测必须认 tool 才不误判) + if ev.get("kind") in ("TextDelta", "ToolCallNew", "FilesChanged"): + emitted_output = True self._dispatch_event(ev, acp_sid, chunk_msg_id) # turn 结束判定 @@ -1777,8 +1840,26 @@ class ACPBridge: # turn 完成: 取一次完整 snapshot, 发本轮新增的 tool/text/reasoning/plan 事件 cur_snapshot = self._build_snapshot(zcode_sid, projection) events = differ.diff(None, cur_snapshot) + has_output = False for ev in events: + # 标记是否本轮产出了实质内容 (text/tool/patch)。PlanUpdate/UsageDelta + # 是辅助信息 (todos/usage 状态), 不是 turn 的实质产出, 不计入。 + if ev.get("kind") in ("TextDelta", "ToolCallNew", "FilesChanged"): + has_output = True self._dispatch_event(ev, acp_sid, chunk_msg_id) + # 失败检测 (issue #3 子项3d, 修复 PR#2 遗留): + # 轮询模式无法收到 turn.failed 事件 payload (zcode 只在事件里给 resultType, + # projection/messages 无失败标志, 实测确认)。但若 turn "完成"却完全无有效输出 + # (只有 step-start 半成品), 强烈暗示 turn 失败 (如 provider 错误)。 + # 此时返回明确错误而非静默 end_turn, 把"吞掉失败"变成"显式告警"。 + if not has_output: + reply_text = self._fetch_last_reply(zcode_sid, differ) + if not reply_text: + del self.pending_turns[msg_id] + log(" ⚠ turn 完成但无任何输出 (无 text/tool/patch), 疑似 turn 失败") + return self._error(msg_id, -32603, + "zcode turn 完成但无输出 (轮询模式疑似失败; " + "可能 provider 错误, 见 stderr 日志)") turn_completed = True break diff --git a/packages/agent-help/zcode-agent-help b/packages/agent-help/zcode-agent-help index 8049bc2..7a6bc61 100755 --- a/packages/agent-help/zcode-agent-help +++ b/packages/agent-help/zcode-agent-help @@ -14,6 +14,7 @@ zcode-agent-help — ZCode 开放生态能力说明书 zcode-agent-help # 完整能力清单 (JSON) zcode-agent-help --pretty # 人类可读的格式化输出 zcode-agent-help --section # 只看某一节 + zcode-agent-help --print-injected-env # 诊断: 会注入哪些凭证 (脱敏, issue #3) zcode-agent-help --help # 本帮助 可用 section: @@ -28,13 +29,118 @@ from pathlib import Path ZCODE_BIN = os.environ.get("ZCODE_BIN", "zcode") +# 模型 ID 的 canonical 格式: config 里 models 的 key 原样 (如 GLM-5.2), 不加 provider 前缀。 +# 实测 "zai/GLM-5.2" 也兼容, 但与 shared/credentials.py 权威实现不一致, 故统一用原始 id。 +ZCODE_CREDS_PATH = Path.home() / ".zcode" / "v2" / "config.json" + + +def _load_creds_internal(config_path=None): + """内嵌的凭证读取 (源自 shared/credentials.py, 单文件可独立运行故内嵌)。 + + 返回从 config 读出的凭证 dict (不含 env 覆盖); 读取失败返回 {}。 + 与 shared/credentials.py.load_zcode_credentials 行为一致 (原始 model id)。 + """ + path = config_path or ZCODE_CREDS_PATH + try: + with open(path) as f: + cfg = json.load(f) + for _pid, p in cfg.get("provider", {}).items(): + if p.get("enabled"): + opts = p.get("options", {}) + models = p.get("models", {}) + model_id = next(iter(models)) if models else "GLM-5.2" + return { + "ZCODE_MODEL": model_id, + "ZCODE_BASE_URL": opts.get("baseURL", ""), + "ANTHROPIC_API_KEY": opts.get("apiKey", ""), + } + except (FileNotFoundError, json.JSONDecodeError, KeyError): + return {} + except Exception: + return {} + return {} + + +def _mask_key(key): + """脱敏 apiKey: 只留前 4 后 4。""" + if not key or len(key) <= 12: + return "(短或空)" if not key else f"{key[:2]}...{key[-2:]}" + return f"{key[:4]}...{key[-4:]}" + + +def print_injected_env(): + """打印"如果现在注入凭证, 会用哪些值"(脱敏), 用于不改 config 诊断 model/baseURL 来源。 + + 显示每个 key 的来源 (config vs 显式 env) + 最终值 (apiKey 脱敏), + 体现"显式 env 优先"的合并语义 (issue #3 子项2)。 + """ + creds = _load_creds_internal() + if not creds: + print(f"❌ 未从 {ZCODE_CREDS_PATH} 读取到凭证 (config 无 enabled provider 或文件缺失)") + return 0 + # 读 config 里所有 provider 的 baseURL (残留检测用, host 集合) + all_config_hosts = set() + try: + from urllib.parse import urlparse as _up + with open(ZCODE_CREDS_PATH) as _f: + _cfg = json.load(_f) + for _p in _cfg.get("provider", {}).values(): + _bu = _p.get("options", {}).get("baseURL", "") + if _bu: + try: + _h = _up(_bu) + if _h.hostname: + all_config_hosts.add(f"{_h.scheme}://{_h.netloc}" if _h.scheme else _h.netloc) + except Exception: + pass + except Exception: + pass + + def _host(u): + try: + h = _up(u) + return f"{h.scheme}://{h.netloc}" if (h.scheme and h.hostname) else None + except Exception: + return None + + keys = ["ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY"] + print("=== 凭证注入诊断 (显式非空 env 优先于 config; 空串视为未设置) ===") + print(f"config 源: {ZCODE_CREDS_PATH}") + for k in keys: + config_val = creds.get(k, "") + env_val = os.environ.get(k) + if env_val: # 非空显式 env 才覆盖 (空串视为未设置, 与 _merge_env_with_creds 一致) + shown = _mask_key(env_val) if k == "ANTHROPIC_API_KEY" else env_val + print(f" {k}:") + print(f" config → {('脱敏 ' + _mask_key(config_val)) if k == 'ANTHROPIC_API_KEY' and config_val else (config_val or '(空)')}") + # ZCODE_BASE_URL 特殊: 检测 App 注入的残留官方 endpoint (host 匹配) + if k == "ZCODE_BASE_URL" and config_val and env_val != config_val: + _eh = _host(env_val) + _ch = _host(config_val) + if _eh and _ch and _eh != _ch and _eh in all_config_hosts: + print(f" env → {shown} 🚫 残留! 是 config 里另一个 provider 的 endpoint") + print(f" → bridge 会自动用 config 值: {config_val}") + print(" (App 切换 plan 后的残留; 在 config 切回该 provider 或用自建 endpoint 可消除)") + else: + print(f" env → {shown} ⚠ 使用此值 (env 覆盖, 自定义 endpoint)") + else: + print(f" env → {shown} ⚠ 使用此值 (env 覆盖)") + elif env_val == "": + # 空串 env: 视为未设置, 仍用 config + shown = _mask_key(config_val) if k == "ANTHROPIC_API_KEY" else config_val + print(f" {k}: {shown} (来自 config; env 为空串, 视为未设置)") + else: + shown = _mask_key(config_val) if k == "ANTHROPIC_API_KEY" else config_val + print(f" {k}: {shown} (来自 config)") + return 0 + # ============================================================ # 静态能力描述 (基于实测) # ============================================================ OVERVIEW = { "tool": "zcode", - "version_described": "0.15.0 (App 3.2.0)", + "version_described": "0.15.0 (App 3.2.1)", "what_it_is": ( "ZCode 是智谱(Z.AI)的 Agentic Coding CLI, 由 GLM 系列模型驱动。" "它能自主完成代码编写/审查/调试/分析等任务, 可调用工具(读写文件、" @@ -488,6 +594,10 @@ def main(): print(__doc__) return 0 + # 诊断: 打印"会注入哪些凭证"(脱敏), 不改 config 即可排查 model/baseURL 来源 + if "--print-injected-env" in args: + return print_injected_env() + if "--section" in args: idx = args.index("--section") if idx + 1 < len(args): diff --git a/packages/mcp-server/zcode-mcp-server b/packages/mcp-server/zcode-mcp-server index ee40594..504e946 100755 --- a/packages/mcp-server/zcode-mcp-server +++ b/packages/mcp-server/zcode-mcp-server @@ -15,10 +15,13 @@ Claude Code、Cursor 等) 能标准化地发现并调用 headless zcode。 依赖: 仅 Python3 标准库 (零第三方依赖) """ +import fcntl import json import os +import re import subprocess import sys +import time from pathlib import Path # zcode-agent-help 的路径 (能力发现的数据源) @@ -68,6 +71,173 @@ def load_zcode_credentials(): return {} +def _merge_env_with_creds(creds): + """合并 config 凭证与环境变量: 显式非空 env 优先, 空串视为未设置; + 且自动检测/自愈 App 注入的残留 baseURL (issue #3)。 + + 源自 shared/credentials.py.merge_env_with_creds (单文件可独立运行故内嵌)。 + 残留检测: env 的 ZCODE_BASE_URL 若是 config 里"另一个 provider"的官方 endpoint + (App 切换 plan 后的残留), 自愈用 config 值; 用户自建 endpoint 则尊重 env。 + """ + merged = {**os.environ, **creds} + for k in ("ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY"): + v = os.environ.get(k) + if v: # 仅非空才覆盖 + merged[k] = v + # 残留检测 + env_bu = os.environ.get("ZCODE_BASE_URL", "") + config_bu = creds.get("ZCODE_BASE_URL", "") + if env_bu and config_bu and env_bu != config_bu: + all_hosts = set() + try: + from urllib.parse import urlparse as _up + _cfg_path = Path.home() / ".zcode" / "v2" / "config.json" + with open(_cfg_path) as _f: + _cfg = json.load(_f) + for _p in _cfg.get("provider", {}).values(): + _bu = _p.get("options", {}).get("baseURL", "") + if _bu: + try: + _h = _up(_bu) + if _h.hostname: + all_hosts.add(f"{_h.scheme}://{_h.netloc}" if _h.scheme else _h.netloc) + except Exception: + pass + _eh = _up(env_bu) + env_host = f"{_eh.scheme}://{_eh.netloc}" if (_eh.scheme and _eh.hostname) else None + _ch = _up(config_bu) + config_host = f"{_ch.scheme}://{_ch.netloc}" if (_ch.scheme and _ch.hostname) else None + except Exception: + env_host = config_host = None + if env_host and config_host and env_host != config_host and env_host in all_hosts: + merged["ZCODE_BASE_URL"] = config_bu + log(f"⚠ ZCODE_BASE_URL 残留检测: env='{env_bu}' 是 config 里另一个 provider " + f"的 endpoint, 与 enabled provider (config='{config_bu}') 不一致。已自动用 " + f"config 值 (旧 plan 残留)。") + return merged + + + +# 识别 zcode headless 调用返回的 provider 限流/配额错误, 用于有限重试。 +# 修改时请同步 shared/provider_error.py。 +# ============================================================ +def _parse_provider_error(text): + """分析 zcode 调用输出文本, 返回 {is_rate_limit, is_quota, retry_after_sec, error_kind}。""" + if not text: + return {"is_rate_limit": False, "is_quota": False, "retry_after_sec": None, "error_kind": "unknown"} + sample = text if len(text) <= 2000 else text[:2000] + lower = sample.lower() + retry_after = _extract_retry_after(sample) + + # 限流 (可安全重试) — 明确限流词 (不含模糊的 exceeded) + rl_pats = [r"\b429\b", r"too many requests", r"ratelimit", r"rate_limit", + r"rate limit", r"\brate.?limit", r"\b1302\b", + r"请求过于频繁", r"频繁访问", r"频繁请求", r"slow\s*down", r"throttl"] + if any(re.search(p, lower) for p in rl_pats): + return {"is_rate_limit": True, "is_quota": False, "retry_after_sec": retry_after, "error_kind": "rate_limit"} + + # 配额耗尽 (不可短期重试) — 用精确的 quota/credit/配额 词, 不用模糊的 exceeded.*limit + # (会误吃 "rate limit exceeded")。中文收紧为带限定词。Codex review P1#2。 + quota_pats = [r"\bquota\b", r"配额(不足|耗尽|用完|超限)", + r"额度(不足|耗尽|用完)", r"余额不足", + r"insufficient.*credit", r"insufficient.*balance", r"\bcredit\b.*exhaust"] + if any(re.search(p, lower) for p in quota_pats): + return {"is_rate_limit": False, "is_quota": True, "retry_after_sec": retry_after, "error_kind": "quota"} + + # 明确带 retry-after 但无 quota/限流标志 → 视为限流兜底 + if retry_after is not None: + return {"is_rate_limit": True, "is_quota": False, "retry_after_sec": retry_after, "error_kind": "rate_limit"} + + # 其他 provider 错误 (APICallError/Unauthorized/invalid key 等, 不重试) + provider_pats = [r"apicallerror", r"ai_apicallerror", r"providerbusinesserror", + r"unauthorized", r"forbidden", r"invalid.*api.*key", + r"internal\s*server", r"service\s*unavailable", r"bad\s*gateway"] + if any(re.search(p, lower) for p in provider_pats): + return {"is_rate_limit": False, "is_quota": False, "retry_after_sec": None, + "error_kind": "provider_error"} + + return {"is_rate_limit": False, "is_quota": False, "retry_after_sec": None, "error_kind": "unknown"} + + +def _extract_retry_after(sample): + """从错误文本提取 retry-after 秒数 (限 [1,300])。""" + for pat in [r'retry-?after\s*[:=]\s*"?(\d+)', r'retryAfter"\s*:\s*"?(\d+)', + r'(\d+)\s*秒[后以]', r'try\s+again\s+in\s+(\d+)']: + m = re.search(pat, sample, re.IGNORECASE) + if m: + try: + return max(1, min(300, int(m.group(1)))) + except (TypeError, ValueError): + pass + return None + + +# ============================================================ +# 进程级文件锁 (防并发 headless review 触发限流, issue #3 子项3) +# 多个 MCP client 进程会各自 spawn zcode; 文件锁串行化 review 调用。 +# 默认开, ZCODE_BRIDGE_REVIEW_LOCK=0 关闭。 +# ZCODE_BRIDGE_LOCK_DIR 可覆盖锁文件目录 (测试用)。 +# ============================================================ +_DEFAULT_LOCK_DIR = Path.home() / ".zcode" + + +def _review_lock_path(): + """锁文件路径; 目录可由 ZCODE_BRIDGE_LOCK_DIR 覆盖 (测试隔离用)。""" + d = Path(os.environ.get("ZCODE_BRIDGE_LOCK_DIR", _DEFAULT_LOCK_DIR)) + return d / "zcode-bridge-review.lock" + + +class ReviewFileLock: + """跨进程文件锁 (fcntl.flock)。with 块内串行执行; 拿不到锁超时返回。 + + 打开/加锁失败时 (目录不存在/不可写/权限受限) 降级为无锁并 log 警告, + 而非打崩 tool — 锁是"防并发"的优化, 不应让权限问题阻断 review 本身。 + """ + + def __init__(self, timeout=300): + self.timeout = timeout + self.enabled = os.environ.get("ZCODE_BRIDGE_REVIEW_LOCK", "1") != "0" + self.fd = None + self.acquired = False + + def __enter__(self): + if not self.enabled: + return self + lock_path = _review_lock_path() + try: + lock_path.parent.mkdir(parents=True, exist_ok=True) + self.fd = open(lock_path, "w") + except OSError as e: + # 目录不存在/不可写 → 降级无锁 (review 本身仍可跑) + log(f"⚠ review 锁文件无法创建 ({e}), 降级为无锁 (并发可能触发限流)") + self.fd = None + return self + t0 = time.time() + while True: + try: + fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + self.acquired = True + return self # 拿到锁 + except (OSError, BlockingIOError): + if time.time() - t0 > self.timeout: + self.fd.close() + self.fd = None + raise TimeoutError( + f"review 锁等待超时 ({self.timeout}s), 可能有并发 review 在跑" + ) from None + time.sleep(0.5) + + def __exit__(self, *exc): + if self.fd is not None: + if self.acquired: + try: + fcntl.flock(self.fd, fcntl.LOCK_UN) + except OSError: + pass + self.fd.close() + self.fd = None + + # ============================================================ # Tool 定义 # ============================================================ @@ -182,20 +352,58 @@ def tool_zcode_review(args): log(f"调用 zcode review: {len(attach_files)} 个输入") - # 注入凭证: MCP server 子进程不继承 shell 环境变量, 需自行读取 - env = {**os.environ, **load_zcode_credentials()} + # 注入凭证: MCP server 子进程不继承 shell 环境变量, 需自行读取。 + # 显式非空环境变量优先 (issue #3 子项2): 已设置非空的会覆盖 config; 空串视为未设置。 + env = _merge_env_with_creds(load_zcode_credentials()) + # 有限重试配置 (issue #3 子项3c): 仅对 provider 限流错误重试, 退避指数。 + # clamp 到 >=0, 防止 ZCODE_BRIDGE_MAX_RETRIES=-1 导致不执行 zcode (Codex P2) + max_retries = max(0, _env_int("ZCODE_BRIDGE_MAX_RETRIES", 3)) + + last_err = None try: - result = subprocess.run( - cmd, capture_output=True, text=True, timeout=120, env=env - ) - output = result.stdout - if result.returncode != 0 and not output: - return {"content": [{"type": "text", "text": f"zcode 调用失败: {result.stderr}"}], + # 文件锁 (issue #3 子项3b): 串行化 headless review, 防并发触发限流 + with ReviewFileLock(timeout=300): + for attempt in range(max_retries + 1): + try: + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=120, env=env + ) + output = result.stdout + # 错误判定: 只解析 stderr (Claude review P1-1: 若把 stdout 审查正文 + # 也喂进 parse_provider_error, 审查"限流/配额"相关代码时会被误判 + # 成限流而白重试, 甚至把成功结果截断成报错)。zcode headless 错误 + # 常在 stderr (APICallError 堆栈), stdout 是审查结论正文。 + err_info = _parse_provider_error(result.stderr or "") + has_error = result.returncode != 0 or (result.stderr and err_info["error_kind"] != "unknown") + + if has_error and err_info["is_rate_limit"] and attempt < max_retries: + # 限流 → 退避后重试 (优先用 retry-after, 否则指数 + jitter) + wait = err_info["retry_after_sec"] or min(60, (2 ** attempt) + 1) + last_err = f"限流 (attempt {attempt+1}/{max_retries}): {err_info['error_kind']}" + log(f" {last_err}, {wait}s 后重试") + time.sleep(wait) + continue + + if has_error: + # 非限流错误, 或重试耗尽 → 返回错误 (重试过的话附说明) + suffix = f" (已重试 {attempt} 次)" if attempt > 0 else "" + err_text = result.stderr.strip() or output or "未知错误" + # 提取简短错误 (脱敏: provider 错误一般不含 apiKey, 但截断防泄漏) + return {"content": [{"type": "text", + "text": f"zcode 调用失败{suffix}: {err_text[:500]}"}], + "isError": True} + # 成功 (也可能 stdout 空但无错误 → 正常返回空) + return {"content": [{"type": "text", "text": output}]} + + except subprocess.TimeoutExpired: + return {"content": [{"type": "text", "text": "zcode review 超时 (120s)"}], + "isError": True} + # 理论不可达 (循环内都 return), 兜底 + return {"content": [{"type": "text", "text": f"zcode 调用失败: 重试耗尽 {last_err}"}], "isError": True} - return {"content": [{"type": "text", "text": output}]} - except subprocess.TimeoutExpired: - return {"content": [{"type": "text", "text": "zcode review 超时 (120s)"}], + except TimeoutError as e: + return {"content": [{"type": "text", "text": f"zcode review 获取锁超时: {e}"}], "isError": True} finally: # 只清理本次创建的临时文件 (不用 glob, 避免误删并发请求的文件) @@ -206,6 +414,14 @@ def tool_zcode_review(args): pass +def _env_int(name, default): + """从环境变量读整数, 失败用默认值。""" + try: + return int(os.environ.get(name, str(default))) + except (TypeError, ValueError): + return default + + TOOL_HANDLERS = { "get_zcode_capabilities": tool_get_zcode_capabilities, "zcode_review": tool_zcode_review, diff --git a/shared/credentials.py b/shared/credentials.py index 7afd028..a9b60c0 100644 --- a/shared/credentials.py +++ b/shared/credentials.py @@ -15,12 +15,15 @@ 本文件是权威实现; 修改凭证逻辑时, 请同步更新两处副本。 用法: - from shared.credentials import load_zcode_credentials - env = {**os.environ, **load_zcode_credentials()} + from shared.credentials import load_zcode_credentials, merge_env_with_creds + # 显式非空环境变量优先 (调试/覆盖用), 空串视为未设置, 且自动检测 baseURL 残留。 + # 不要手写 {**creds, **os.environ} (会绕过空串保护和残留自愈), 用 merge_env_with_creds。 + env = merge_env_with_creds(load_zcode_credentials()) """ import json from pathlib import Path +from urllib.parse import urlparse # zcode 桌面 App 的配置文件 (含 provider 凭证) ZCODE_CREDS_PATH = Path.home() / ".zcode" / "v2" / "config.json" @@ -63,6 +66,126 @@ def load_zcode_credentials(config_path=None): return {} +def _all_provider_base_urls(config_path=None): + """读取 config 里所有 provider 的 baseURL 集合 (用于残留检测)。 + + 用于判断 env 里的 ZCODE_BASE_URL 是否指向"config 里某个其他 provider" + (即 App 注入的残留), 而非用户自建 endpoint (合法自定义)。 + """ + path = config_path or ZCODE_CREDS_PATH + urls = set() + try: + with open(path) as f: + cfg = json.load(f) + for p in cfg.get("provider", {}).values(): + bu = p.get("options", {}).get("baseURL", "") + if bu: + urls.add(bu) + except Exception: + pass + return urls + + +def is_stale_env_base_url(env_base_url, config_base_url, all_config_urls): + """判断 env 的 baseURL 是否是 App 注入的"残留官方 endpoint"。 + + ZCode App 切换过 plan 的用户: 旧 plan 的 baseURL 会残留在子进程 env 里 + (即使该 plan 已 not_entitled), 而 config 的 enabled provider 是正确的。 + 本函数精确区分两种"env 与 config 不同": + - 残留: env 值 ∈ all_config_urls (是 config 里另一个 provider 的官方 endpoint) + → 应自愈用 config 值 + - 自定义: env 值 ∉ all_config_urls (用户故意指向自建/代理) + → 尊重 env (issue #3 调试场景) + + Args: + env_base_url: env 里的 ZCODE_BASE_URL + config_base_url: config enabled provider 的 baseURL + all_config_urls: _all_provider_base_urls() 的返回 + + Returns: + True 若判定为残留 (建议用 config 值); False 若尊重 env。 + """ + if not env_base_url or not config_base_url: + return False + if env_base_url == config_base_url: + return False # 一致, 无问题 + # env 值是 config 里某个 provider 的官方 endpoint (但不是 enabled 那个) → 残留。 + # 注意: App 注入的可能是根域名 (如 https://zcode.z.ai), config 存的是完整路径 + # (如 https://zcode.z.ai/api/v1/zcode-plan/anthropic), 故用 host 匹配而非精确匹配。 + env_host = _safe_host(env_base_url) + if not env_host: + return False + config_hosts = {_safe_host(u) for u in all_config_urls} + config_host = _safe_host(config_base_url) + # env 的 host 等于某个 config provider 的 host, 但不等于 enabled 那个 → 残留 + return env_host in config_hosts and env_host != config_host + + +def _safe_host(url): + """提取 URL 的 host (scheme://host[:port]), 解析失败返回 None。""" + try: + p = urlparse(url) + if not p.hostname: + return None + return f"{p.scheme}://{p.netloc}" if p.scheme else f"https://{p.netloc}" + except Exception: + return None + + +# 受凭证注入影响的 env key (显式设置时优先于 config, 但空串视为未设置) +_CRED_ENV_KEYS = ("ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY") + + +def merge_env_with_creds(creds=None, environ=None, config_path=None, warn=None): + """合并 config 凭证与环境变量, 实现"显式 env 优先, 空串视为未设置, + 且自动检测/自愈 App 注入的残留 baseURL"。 + + issue #3 子项2: 让 ZCODE_MODEL=xxx 临时覆盖生效, 但避免空串 env + (如 ZCODE_MODEL="") 把 config 的有效值覆盖成空。 + + 残留检测 (产品级): ZCode App 切换过 plan 的用户, 旧 plan 的 baseURL 会 + 残留在子进程 env (即使已 not_entitled)。若 env 的 ZCODE_BASE_URL 是 config + 里"另一个 provider"的官方 endpoint (非 enabled 那个), 判定为残留 → 用 config + 值 + 告警。用户自建 endpoint (不在 config 任何 provider 里) 则正常尊重 env。 + + Args: + creds: load_zcode_credentials() 的返回 (config 凭证), 默认 {} + environ: 环境变量 dict, 默认 os.environ + config_path: 可选, 用于读 all_provider_urls 做残留判定 + warn: 可选回调 fn(msg), 用于告警 (通常传 log) + + Returns: + 合并后的 env dict: config 凭证作基底, 仅非空显式 env 覆盖之 (残留除外)。 + """ + import os + if creds is None: + creds = {} + if environ is None: + environ = os.environ + # 基底: 完整 os.environ + config 凭证 (config 覆盖 os.environ 里的同名空串/旧值) + merged = {**environ, **creds} + # 显式非空 env 再覆盖回去 (解决"显式 env 优先") + for k in _CRED_ENV_KEYS: + v = environ.get(k) + if v: # 仅非空才覆盖 (空串视为未设置, 保留 config 值) + merged[k] = v + + # 残留检测: env 的 baseURL 是否是 App 注入的"另一个 plan 的官方 endpoint" + env_bu = environ.get("ZCODE_BASE_URL", "") + config_bu = creds.get("ZCODE_BASE_URL", "") + if env_bu and config_bu and env_bu != config_bu: + all_urls = _all_provider_base_urls(config_path) + if is_stale_env_base_url(env_bu, config_bu, all_urls): + # 残留: env 值是 config 里另一个 provider 的官方 endpoint → 自愈用 config 值 + merged["ZCODE_BASE_URL"] = config_bu + if warn: + warn(f"⚠ ZCODE_BASE_URL 残留检测: env='{env_bu}' 是 config 里另一个 " + f"provider 的 endpoint, 与 enabled provider (config='{config_bu}') " + f"不一致。已自动用 config 值。若需用该 endpoint, 请在 config 切换 " + f"provider 或用自建 endpoint (非 config 内置)。") + return merged + + if __name__ == "__main__": # 自测: 打印读取到的凭证 (脱敏) creds = load_zcode_credentials() diff --git a/shared/provider_error.py b/shared/provider_error.py new file mode 100644 index 0000000..fdb1f95 --- /dev/null +++ b/shared/provider_error.py @@ -0,0 +1,156 @@ +""" +shared/provider_error.py — 解析 zcode headless 调用的 provider 限流/配额错误 + +zcode --prompt 失败时, provider 端错误 (限流/配额) 会出现在: + - stderr 文本里 (APICallError / AI_APICallError 堆栈, Vercel AI SDK) + - 或 --json 输出的 error 字段 (若有) + +这些错误的文本形态 (从 zcode.cjs 与实测确认): + - "429" / "Too Many Requests" / "RateLimit" / "rate_limit" (通用 HTTP 限流) + - "1302" (z.ai 特定业务码) + - "retry-after" / "retryAfter" header 值 + - 中文: "请求过于频繁" / "频繁" / "配额" + +本模块是纯函数, 无副作用, 便于单测。MCP/ACP bridge 各内嵌一份副本 +(保持单文件可独立运行特性), 修改时请同步。 + +用法: + from shared.provider_error import parse_provider_error + info = parse_provider_error(stderr_text) + if info["is_rate_limit"] and info["retry_after_sec"]: + time.sleep(info["retry_after_sec"]) +""" + +import re + + +def parse_provider_error(text): + """分析 zcode 调用输出文本, 判断是否 provider 限流/配额错误。 + + Args: + text: zcode 的 stderr / stdout / error 文本 (任意混合, 可含堆栈)。 + + Returns: + dict: + is_rate_limit: bool 是否限流类错误 (可安全重试) + is_quota: bool 是否配额/额度耗尽 (通常不可短期重试) + retry_after_sec: int|None 建议等待秒数 (从 retry-after / "请 X 秒后" 提取) + error_kind: str "rate_limit"|"quota"|"provider_error"|"unknown" + raw_hint: str 简短错误上下文 (脱敏, 仅用于日志/返回调用方) + """ + if not text: + return {"is_rate_limit": False, "is_quota": False, "retry_after_sec": None, + "error_kind": "unknown", "raw_hint": ""} + + sample = text if len(text) <= 2000 else text[:2000] # 只看头部, 避免大堆栈 + + # 简短上下文: 取第一个非堆栈的有意义行 + raw_hint = _extract_hint(sample) + + lower = sample.lower() + + # 先尝试提取 retry-after (多处会用到) + retry_after = _extract_retry_after(sample) + + # 限流 (可安全重试) — 明确限流词 (不含模糊的 exceeded) + rate_limit_patterns = [ + r"\b429\b", r"too many requests", r"ratelimit", r"rate_limit", + r"rate limit", r"\brate.?limit", r"\b1302\b", r"请求过于频繁", + r"频繁访问", r"频繁请求", r"slow\s*down", r"throttl", + ] + if any(re.search(p, lower) for p in rate_limit_patterns): + return {"is_rate_limit": True, "is_quota": False, "retry_after_sec": retry_after, + "error_kind": "rate_limit", "raw_hint": raw_hint} + + # 配额/额度耗尽 (通常不可短期重试) — 用精确的 quota/credit/配额 词, 不用模糊的 + # "exceeded.*limit" (会误吃 "rate limit exceeded")。Codex review P1#2。 + quota_patterns = [ + r"\bquota\b", r"配额(不足|耗尽|用完|超限)", r"额度(不足|耗尽|用完)", + r"余额不足", r"insufficient.*(quota|credit|balance)", r"\bcredit\b.*exhaust", + ] + if any(re.search(p, lower) for p in quota_patterns): + return {"is_rate_limit": False, "is_quota": True, "retry_after_sec": retry_after, + "error_kind": "quota", "raw_hint": raw_hint} + + # 明确带 retry-after 但无 quota/限流标志 → 视为限流兜底 (很多 API 只给 retry-after) + if retry_after is not None: + return {"is_rate_limit": True, "is_quota": False, "retry_after_sec": retry_after, + "error_kind": "rate_limit", "raw_hint": raw_hint} + + # 其他 provider 错误 (有 APICallError 标志, 但不是限流/配额) → 不重试 + provider_patterns = [ + r"apicallerror", r"ai_apicallerror", r"providerbusinesserror", + r"unauthorized", r"forbidden", r"invalid.*api.*key", + r"internal\s*server", r"service\s*unavailable", r"bad\s*gateway", + ] + if any(re.search(p, lower) for p in provider_patterns): + return {"is_rate_limit": False, "is_quota": False, "retry_after_sec": None, + "error_kind": "provider_error", "raw_hint": raw_hint} + + return {"is_rate_limit": False, "is_quota": False, "retry_after_sec": None, + "error_kind": "unknown", "raw_hint": raw_hint} + + +def _extract_hint(sample): + """从错误文本提取简短上下文 (跳过堆栈行, 脱敏)。""" + for line in sample.splitlines(): + line = line.strip() + if not line: + continue + # 跳过 JS 堆栈行 + if line.startswith("at ") or line.startswith(" at "): + continue + if line.startswith("{") or line.startswith("[Object]") or "[Object]" in line: + continue + # 截断到合理长度 + if len(line) > 120: + line = line[:120] + "..." + return line + return "" + + +def _extract_retry_after(sample): + """从错误文本提取 retry-after 秒数 (支持英文/中文/header 形式)。 + + 匹配: + "retry-after: 30" / "Retry-After: 30" + "retryAfter: 30" + "请在 30 秒后" / "请 30 秒后重试" + "try again in 30 seconds" + """ + # retry-after header (秒), 形如 retry-after: 30 或 retry-after: "30" + m = re.search(r'retry-?after\s*[:=]\s*"?(\d+)', sample, re.IGNORECASE) + if m: + return _bounded(int(m.group(1))) + # retryAfter JSON 字段 + m = re.search(r'retryAfter"\s*:\s*"?(\d+)', sample, re.IGNORECASE) + if m: + return _bounded(int(m.group(1))) + # 中文 "请在 N 秒后" / "N 秒后重试" + m = re.search(r"(\d+)\s*秒[后以]", sample) + if m: + return _bounded(int(m.group(1))) + # "try again in N seconds" + m = re.search(r"try\s+again\s+in\s+(\d+)", sample, re.IGNORECASE) + if m: + return _bounded(int(m.group(1))) + return None + + +def _bounded(seconds, lo=1, hi=300): + """限制 retry_after 到合理范围 [1, 300] 秒, 防止 provider 返回异常大值导致长眠。""" + try: + s = int(seconds) + except (TypeError, ValueError): + return None + return max(lo, min(hi, s)) + + +if __name__ == "__main__": + import sys + if len(sys.argv) > 1: + text = " ".join(sys.argv[1:]) + else: + text = sys.stdin.read() + import json + print(json.dumps(parse_provider_error(text), indent=2, ensure_ascii=False)) diff --git a/skills/zcode-bridge-guide/SKILL.md b/skills/zcode-bridge-guide/SKILL.md index 616d36a..5a44681 100644 --- a/skills/zcode-bridge-guide/SKILL.md +++ b/skills/zcode-bridge-guide/SKILL.md @@ -8,7 +8,7 @@ user-invocable: true # 驱动 ZCode（三模式通用说明书） > 本 skill 是 [zcode-open-bridge](https://github.com/tizerluo/zcode-open-bridge) 项目的配套说明书。 -> 兼容 ZCode CLI **0.14.5 ~ 0.15.0+**（App 3.2.0）。新版功能（事件驱动真流式、fork/rewind/goal/compact/steer、workspace/*、setThoughtLevel 思考强度控制）在旧版上自动降级或返回 `-32603`。 +> 兼容 ZCode CLI **0.14.5 ~ 0.15.0+**（App 3.2.0+，实测含 3.2.1）。新版功能（事件驱动真流式、fork/rewind/goal/compact/steer、workspace/*、setThoughtLevel 思考强度控制）在旧版上自动降级或返回 `-32603`。 ## 前置条件 @@ -116,9 +116,9 @@ 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"] + # canonical model id = config 里 models 的 key 原样 (如 GLM-5.2), 不加 provider 前缀。 + # 实测 "zai/GLM-5.2" 也兼容, 但与本项目的权威实现 (shared/credentials.py) 不一致, 故统一用原始 id。 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 @@ -356,7 +356,7 @@ npm test # 全量，看实际数字 | ZCode CLI 版本 | 支持情况 | 差异 | |:--------------:|:--------:|------| -| **0.15.0+** (App 3.2.0) | ✅ 完整 | ACP bridge 真流式；全部扩展方法可用（含 workspace/*、setThoughtLevel） | +| **0.15.0+** (App 3.2.0+) | ✅ 完整 | ACP bridge 真流式；全部扩展方法可用（含 workspace/*、setThoughtLevel） | | **0.14.8** (App 3.1.4) | ✅ 完整 | ACP bridge 真流式；fork/rewind/goal/compact/steer 可用；workspace/* 与 setThoughtLevel 返回 -32603 | | **0.14.5 ~ 0.14.7** | ✅ 兼容 | ACP bridge 自动降级伪流式；扩展方法不可用（协议未实现） | | **< 0.14.5** | ⚠️ 未测 | CLI `--prompt` 基本可用；ACP bridge 未验证 | diff --git a/tests/test_credentials.py b/tests/test_credentials.py new file mode 100644 index 0000000..3733ab4 --- /dev/null +++ b/tests/test_credentials.py @@ -0,0 +1,297 @@ +""" +test_credentials.py — 凭证读取与 env 优先级单测 + +验证 shared/credentials.py 的 model id 读取 (canonical 原始格式, 子项1) 和 +"显式 env 优先"的合并语义 (子项2), 补齐此前的零测试覆盖。 + + C0 正常读取: 返回第一个 enabled provider 的 model/baseURL/apiKey + C1 model id 是原始格式 (config 里 models 的 key 原样, 如 GLM-5.2, 不加 zai/ 前缀) + C2 models 为空 → 兜底 GLM-5.2 + C3 无 enabled provider → 返回 {} + C4 config 文件缺失 → 返回 {} (不崩) + C5 config JSON 损坏 → 返回 {} (不崩) + C6 env 优先级: 显式 os.environ 覆盖 config 读出的值 (子项2 核心) + C7 多 provider: 只取第一个 enabled + C8 apiKey 为空: 仍返回 (baseURL/apiKey 可空) + +运行: python3 tests/test_credentials.py +依赖: 仅 Python 标准库 + shared/credentials.py +""" + +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "shared")) +from credentials import ( # noqa: E402 + is_stale_env_base_url, + load_zcode_credentials, + merge_env_with_creds, +) + + +def _write_config(config_dict): + """写一个临时 config.json, 返回路径。""" + f = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) + json.dump(config_dict, f) + f.close() + return f.name + + +def _config_with_provider(provider_id="builtin:zai-coding-plan", enabled=True, + models=None, base_url="https://api.z.ai/api/anthropic", + api_key="sk-test-key-123456"): + if models is None: + models = {"GLM-5.2": {}} + return { + "provider": { + provider_id: { + "enabled": enabled, + "options": {"baseURL": base_url, "apiKey": api_key}, + "models": models, + } + } + } + + +class TestCredentials(unittest.TestCase): + """凭证读取 + model id canonical 格式 + env 优先级""" + + def setUp(self): + self._tmp_paths = [] + + def tearDown(self): + for p in self._tmp_paths: + try: + os.unlink(p) + except OSError: + pass + + def _creds(self, config_dict): + path = _write_config(config_dict) + self._tmp_paths.append(path) + return load_zcode_credentials(config_path=path) + + # ---------- C0: 正常读取 ---------- + def test_c0_normal_read(self): + """C0: 正常读取第一个 enabled provider""" + c = self._creds(_config_with_provider()) + self.assertEqual(c["ZCODE_MODEL"], "GLM-5.2") + self.assertEqual(c["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic") + self.assertEqual(c["ANTHROPIC_API_KEY"], "sk-test-key-123456") + + # ---------- C1: model id 原始格式 (无 zai/ 前缀) ---------- + def test_c1_model_id_raw_format(self): + """C1: model id = config 里 models 的 key 原样, 不加 provider 前缀 (子项1)""" + c = self._creds(_config_with_provider( + provider_id="builtin:zai-coding-plan", # builtin:zai 也不加 zai/ 前缀 + models={"GLM-5.2": {}, "GLM-5-Turbo": {}})) + self.assertEqual(c["ZCODE_MODEL"], "GLM-5.2", + "canonical model id 必须是原始 key, 不加 zai/ 前缀") + self.assertNotIn("zai/", c["ZCODE_MODEL"]) + + def test_c1b_custom_model_id(self): + """C1b: 自定义 model id (如第三方 provider) 原样返回""" + c = self._creds(_config_with_provider( + provider_id="custom:openai", models={"gpt-custom": {}})) + self.assertEqual(c["ZCODE_MODEL"], "gpt-custom") + + # ---------- C2: models 为空兜底 ---------- + def test_c2_empty_models_fallback(self): + """C2: models 为空 → 兜底 GLM-5.2""" + c = self._creds(_config_with_provider(models={})) + self.assertEqual(c["ZCODE_MODEL"], "GLM-5.2") + + # ---------- C3: 无 enabled provider ---------- + def test_c3_no_enabled_provider(self): + """C3: 无 enabled provider → {}""" + c = self._creds(_config_with_provider(enabled=False)) + self.assertEqual(c, {}) + + # ---------- C4: 文件缺失 ---------- + def test_c4_missing_file(self): + """C4: config 文件不存在 → {} (不崩)""" + c = load_zcode_credentials(config_path="/nonexistent/path/config.json") + self.assertEqual(c, {}) + + # ---------- C5: JSON 损坏 ---------- + def test_c5_corrupt_json(self): + """C5: config JSON 损坏 → {} (不崩)""" + f = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) + f.write("{ this is not valid json") + f.close() + self._tmp_paths.append(f.name) + c = load_zcode_credentials(config_path=f.name) + self.assertEqual(c, {}) + + # ---------- C6: env 优先级 (子项2 核心) ---------- + def test_c6_env_overrides_config(self): + """C6: 显式 os.environ 覆盖 config 读出的值 (子项2a 合并顺序: creds 在左)""" + c = self._creds(_config_with_provider(models={"GLM-5.2": {}})) + # 模拟 bridge 的合并: {**creds, **os.environ} → 显式 env 赢 + fake_env = {"ZCODE_MODEL": "GLM-5-Turbo"} # 显式覆盖 + merged = {**c, **fake_env} + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5-Turbo", + "显式 env 应覆盖 config 读出的 model") + + def test_c6b_env_partial_override(self): + """C6b: 只覆盖部分 (baseURL 来自 config, model 来自 env)""" + c = self._creds(_config_with_provider(models={"GLM-5.2": {}})) + fake_env = {"ZCODE_MODEL": "GLM-5-Turbo"} # 只设 model + merged = {**c, **fake_env} + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5-Turbo") + self.assertEqual(merged["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic", + "未覆盖的仍来自 config") + + def test_c6c_no_env_uses_config(self): + """C6c: 无显式 env → 用 config 值 (原有行为不变)""" + c = self._creds(_config_with_provider(models={"GLM-5.2": {}})) + merged = {**c, **{}} # 无显式 env + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5.2") + + # ---------- C9: merge_env_with_creds 空串不覆盖 (Codex P0) ---------- + def test_c9_empty_env_not_override(self): + """C9: 空串 env 视为未设置, 保留 config 值 (Codex P0 修复)""" + c = self._creds(_config_with_provider(models={"GLM-5.2": {}})) + merged = merge_env_with_creds(c, {"ZCODE_MODEL": "", "OTHER": "x"}) + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5.2", + "空串 env 不应覆盖 config 的有效值") + self.assertEqual(merged["OTHER"], "x", "非凭证 env 应保留") + + def test_c9b_nonempty_env_overrides(self): + """C9b: 非空 env 覆盖 config (merge_env_with_creds)""" + c = self._creds(_config_with_provider(models={"GLM-5.2": {}})) + merged = merge_env_with_creds(c, {"ZCODE_MODEL": "GLM-5-Turbo"}) + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5-Turbo") + + def test_c9c_empty_apikey_not_clear_config(self): + """C9c: ANTHROPIC_API_KEY="" 不应清空 config 的 key""" + c = self._creds(_config_with_provider(api_key="sk-real-key-123456")) + merged = merge_env_with_creds(c, {"ANTHROPIC_API_KEY": ""}) + self.assertEqual(merged["ANTHROPIC_API_KEY"], "sk-real-key-123456") + + # ---------- C7: 多 provider ---------- + def test_c7_multiple_providers_first_enabled(self): + """C7: 多 provider, 只取第一个 enabled""" + cfg = { + "provider": { + "disabled-one": {"enabled": False, "options": {}, + "models": {"model-A": {}}}, + "enabled-one": {"enabled": True, "options": {"baseURL": "https://x"}, + "models": {"model-B": {}}}, + } + } + c = self._creds(cfg) + self.assertEqual(c["ZCODE_MODEL"], "model-B") + + # ---------- C8: apiKey/baseURL 为空 ---------- + def test_c8_empty_apikey(self): + """C8: apiKey 为空仍正常返回 (key 字段为空串)""" + c = self._creds(_config_with_provider(api_key="")) + self.assertEqual(c["ANTHROPIC_API_KEY"], "") + self.assertEqual(c["ZCODE_MODEL"], "GLM-5.2") + + # ---------- C10: ZCODE_BASE_URL 残留检测 (App 切换 plan 后的残留自愈) ---------- + def _config_multi_provider(self, enabled_url, stale_url): + """构造两个 provider 的 config: enabled 用 enabled_url, 另一个用 stale_url。""" + return { + "provider": { + "builtin:zai-coding-plan": { + "enabled": True, + "options": {"baseURL": enabled_url, "apiKey": "key-good"}, + "models": {"GLM-5.2": {}}, + }, + "builtin:zai-start-plan": { + "enabled": False, + "options": {"baseURL": stale_url, "apiKey": "jwt-stale"}, + "models": {"GLM-5.2": {}}, + }, + } + } + + def _multi_path(self, enabled_url, stale_url): + """写多 provider config 到临时文件, 返回路径 (残留检测需 config_path)。""" + return _write_config(self._config_multi_provider(enabled_url, stale_url)) + + def test_c10_stale_env_self_heals(self): + """C10: env baseURL 是 config 另一 provider 的残留 → 自愈用 config 值""" + path = self._multi_path("https://api.z.ai/api/anthropic", + "https://zcode.z.ai/api/v1/zcode-plan/anthropic") + self._tmp_paths.append(path) + c = load_zcode_credentials(config_path=path) + # env 设为 stale provider 的根域名 (App 注入的真实形态); 传 config_path 让检测读到临时 config + merged = merge_env_with_creds(c, {"ZCODE_BASE_URL": "https://zcode.z.ai"}, + config_path=path) + self.assertEqual(merged["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic", + "残留 env 应被自愈为 config enabled provider 的值") + + def test_c10b_stale_warning_fired(self): + """C10b: 残留时 warn 回调被触发""" + path = self._multi_path("https://api.z.ai/api/anthropic", + "https://zcode.z.ai/api/v1/zcode-plan/anthropic") + self._tmp_paths.append(path) + c = load_zcode_credentials(config_path=path) + warnings = [] + merge_env_with_creds(c, {"ZCODE_BASE_URL": "https://zcode.z.ai"}, + config_path=path, warn=warnings.append) + self.assertTrue(any("残留" in w for w in warnings), "残留时应触发告警") + + def test_c10c_custom_endpoint_respected(self): + """C10c: env baseURL 是用户自建 (不在 config 任何 provider) → 尊重 env""" + path = self._multi_path("https://api.z.ai/api/anthropic", + "https://zcode.z.ai/api/v1/zcode-plan/anthropic") + self._tmp_paths.append(path) + c = load_zcode_credentials(config_path=path) + merged = merge_env_with_creds(c, {"ZCODE_BASE_URL": "https://my-proxy.example.com"}, + config_path=path) + self.assertEqual(merged["ZCODE_BASE_URL"], "https://my-proxy.example.com", + "自定义 endpoint 应被尊重 (issue #3 调试场景)") + + def test_c10d_consistent_no_warning(self): + """C10d: env 与 config 一致 → 不告警, 用一致值""" + path = self._multi_path("https://api.z.ai/api/anthropic", + "https://zcode.z.ai/api/v1/zcode-plan/anthropic") + self._tmp_paths.append(path) + c = load_zcode_credentials(config_path=path) + warnings = [] + merged = merge_env_with_creds(c, {"ZCODE_BASE_URL": "https://api.z.ai/api/anthropic"}, + config_path=path, warn=warnings.append) + self.assertEqual(merged["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic") + self.assertEqual(warnings, [], "一致时不应告警") + + def test_c10e_stale_exact_url_also_heals(self): + """C10e: env baseURL 是 stale provider 的完整 URL (非根域名) → 同样自愈""" + stale = "https://zcode.z.ai/api/v1/zcode-plan/anthropic" + path = self._multi_path("https://api.z.ai/api/anthropic", stale) + self._tmp_paths.append(path) + c = load_zcode_credentials(config_path=path) + merged = merge_env_with_creds(c, {"ZCODE_BASE_URL": stale}, config_path=path) + self.assertEqual(merged["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic") + + def test_c10f_is_stale_helper_host_matching(self): + """C10f: is_stale_env_base_url 用 host 匹配 (根域名 vs 完整路径)""" + all_urls = {"https://api.z.ai/api/anthropic", + "https://zcode.z.ai/api/v1/zcode-plan/anthropic"} + # env 根域名 zcode.z.ai, config 是 api.z.ai → 残留 (host 不同, env host 在 all_urls 里) + self.assertTrue(is_stale_env_base_url( + "https://zcode.z.ai", "https://api.z.ai/api/anthropic", all_urls)) + # env 是自建 host → 非残留 + self.assertFalse(is_stale_env_base_url( + "https://my-proxy.com", "https://api.z.ai/api/anthropic", all_urls)) + # env 与 config 一致 → 非残留 + self.assertFalse(is_stale_env_base_url( + "https://api.z.ai/api/anthropic", "https://api.z.ai/api/anthropic", all_urls)) + + def test_c10g_no_config_path_still_safe(self): + """C10g: config_path 默认 (None) 也能安全检测, 不崩 (CI 无 ~/.zcode 时)""" + # 不传 config_path; CI 环境 ~ 不存在 config, _all_provider_base_urls 返回空 set, + # 残留检测应安全跳过 (自建 endpoint 尊重 env, 不崩) + c = {"ZCODE_BASE_URL": "https://api.z.ai/api/anthropic"} + merged = merge_env_with_creds(c, {"ZCODE_BASE_URL": "https://self-hosted.test"}) + self.assertEqual(merged["ZCODE_BASE_URL"], "https://self-hosted.test") + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/test_mcp_retry_lock.py b/tests/test_mcp_retry_lock.py new file mode 100644 index 0000000..6f0deef --- /dev/null +++ b/tests/test_mcp_retry_lock.py @@ -0,0 +1,350 @@ +""" +test_mcp_retry_lock.py — MCP server 限流重试与文件锁单测 + +验证 zcode-mcp-server 的 tool_zcode_review (issue #3 子项3b+3c): + - provider 限流错误 → 有限重试 (指数退避) + - 非限流错误 → 不重试, 直接返回 + - 进程级文件锁 (ReviewFileLock) 可获取/释放/禁用/超时 + +用 monkeypatch 替换 subprocess.run (不真跑 zcode), 模拟限流/非限流错误。 +为避免重试测试真的 sleep, 也 patch time.sleep 为 no-op。 + + RL0 文件锁: 获取 + 自动释放 + RL1 文件锁: ZCODE_BRIDGE_REVIEW_LOCK=0 禁用 + RL2 文件锁: 同进程二次获取 (重入, fcntl LOCK_EX 在同 fd 上是递归的) + RT0 重试: 限流错误 (429) → 重试, 最终成功 + RT1 重试: 重试耗尽仍限流 → 返回带"已重试 N 次"的错误 + RT2 重试: 非限流错误 (Unauthorized) → 不重试, 直接返回错误 + RT3 重试: 首次成功 → 不重试 + RT4 重试: max_retries 可由 env 配置 (ZCODE_BRIDGE_MAX_RETRIES) + +运行: python3 tests/test_mcp_retry_lock.py +依赖: 仅 Python 标准库 + zcode-mcp-server 模块 +""" + +import os +import types +import unittest + +# 加载 zcode-mcp-server (单文件, exec 导入, 跳过 main) +MCP_PATH = os.path.join( + os.path.dirname(__file__), "..", "packages", "mcp-server", "zcode-mcp-server" +) + + +def _load_mcp_module(): + mod = types.ModuleType("zcode_mcp_server") + # MCP server 顶层用 Path(__file__) 解析 AGENT_HELP_BIN, exec 进新模块需提供 __file__ + mod.__file__ = MCP_PATH + with open(MCP_PATH) as f: + code = f.read() + code_no_main = code.split('if __name__ == "__main__":')[0] + exec(code_no_main, mod.__dict__) + return mod + + +class _FakeCompletedProcess: + """模拟 subprocess.run 的返回值。""" + + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class TestReviewFileLock(unittest.TestCase): + """进程级文件锁 (ReviewFileLock) 测试。 + + 用 ZCODE_BRIDGE_LOCK_DIR 隔离到临时目录, 避免写真实 ~/.zcode (Codex P1-2: + 沙箱/CI 里真实 HOME 可能不可写, 导致 PermissionError)。 + """ + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + cls.Lock = cls.mod.ReviewFileLock + + def setUp(self): + import tempfile + self._tmpdir = tempfile.mkdtemp(prefix="zcode-lock-test-") + self._saved_dir = os.environ.get("ZCODE_BRIDGE_LOCK_DIR") + os.environ["ZCODE_BRIDGE_LOCK_DIR"] = self._tmpdir + + def tearDown(self): + import shutil + if self._saved_dir is None: + os.environ.pop("ZCODE_BRIDGE_LOCK_DIR", None) + else: + os.environ["ZCODE_BRIDGE_LOCK_DIR"] = self._saved_dir + shutil.rmtree(self._tmpdir, ignore_errors=True) + + def test_rl0_acquire_release(self): + """RL0: 锁可获取, with 块结束自动释放""" + with self.Lock(timeout=5) as lock: + self.assertTrue(lock.acquired, "应成功获取锁") + self.assertIsNotNone(lock.fd) + + def test_rl1_disabled_by_env(self): + """RL1: ZCODE_BRIDGE_REVIEW_LOCK=0 → 禁用锁 (enabled=False)""" + old = os.environ.get("ZCODE_BRIDGE_REVIEW_LOCK") + try: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + lock = self.Lock(timeout=5) + self.assertFalse(lock.enabled, "env=0 应禁用锁") + with lock: + pass # 禁用时直接通过, 不获取文件锁 + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + def test_rl2_reentrant_same_process(self): + """RL2: 同进程顺序获取两次 (第一次释放后再取) 不死锁""" + with self.Lock(timeout=5): + pass + with self.Lock(timeout=5): + pass # 第二次也能拿到 (前次已释放) + + def test_rl3_lock_file_in_temp_dir(self): + """RL3: 锁文件创建在隔离的临时目录, 不写真实 ~/.zcode""" + with self.Lock(timeout=5): + lock_files = os.listdir(self._tmpdir) + self.assertIn("zcode-bridge-review.lock", lock_files, + "锁文件应在 ZCODE_BRIDGE_LOCK_DIR 指定的临时目录") + + def test_rl4_degrade_when_dir_unwritable(self): + """RL4: 锁目录不可写时降级为无锁 (不崩, Codex P1-2)""" + # 指向一个不存在的、且无法 mkdir 的路径 (父父目录无权限通常难造, + # 改用: 指向一个已存在的普通文件作为 dir, mkdir 会失败) + import tempfile + f = tempfile.NamedTemporaryFile(delete=False) + f.close() + self.addCleanup(os.unlink, f.name) + os.environ["ZCODE_BRIDGE_LOCK_DIR"] = f.name # 这是一个文件, 不是目录 + with self.Lock(timeout=5) as lock: + # 应降级为无锁 (fd=None), 而非抛异常 + self.assertIsNone(lock.fd, "目录无效时应降级无锁, fd=None") + + +class TestRetryLogic(unittest.TestCase): + """tool_zcode_review 的限流重试逻辑测试 (monkeypatch subprocess.run + time.sleep)""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def _patch(self, mod, completed_processes): + """patch subprocess.run 返回序列 + time.sleep no-op。返回 call 计数器。""" + calls = {"n": 0} + real_sleep = mod.time.sleep + saved_run = mod.subprocess.run + + def fake_run(*a, **kw): + i = calls["n"] + calls["n"] += 1 + if i < len(completed_processes): + return completed_processes[i] + return completed_processes[-1] # 超出则重复最后一个 + + def no_sleep(_secs): + pass + + mod.subprocess.run = fake_run + mod.time.sleep = no_sleep + return calls, saved_run, real_sleep + + def _restore(self, mod, saved_run, real_sleep): + mod.subprocess.run = saved_run + mod.time.sleep = real_sleep + + def _review(self, mod, **kwargs): + """调 tool_zcode_review, 默认给 code (避免真文件)。""" + args = {"code": "print('x')", "focus": "测试", **kwargs} + return mod.tool_zcode_review(args) + + # ---------- RT0: 限流重试后成功 ---------- + def test_rt0_rate_limit_then_success(self): + """RT0: 首次 429 限流 → 重试 → 第二次成功""" + mod = self.mod + procs = [ + _FakeCompletedProcess(returncode=0, stdout="", stderr="429 Too Many Requests"), + _FakeCompletedProcess(returncode=0, stdout="审查结论: OK", stderr=""), + ] + # 禁用文件锁, 避免测试间锁干扰 + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertEqual(calls["n"], 2, "应重试一次 (共调用 2 次)") + self.assertNotIn("isError", result, "最终成功不应 isError") + self.assertIn("审查结论", result["content"][0]["text"]) + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + # ---------- RT1: 重试耗尽 ---------- + def test_rt1_rate_limit_exhausted(self): + """RT1: 始终限流 → 重试耗尽, 返回带重试次数的错误""" + mod = self.mod + procs = [_FakeCompletedProcess(returncode=0, stdout="", + stderr="429 Too Many Requests")] # 每次都限流 + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + old_mr = os.environ.pop("ZCODE_BRIDGE_MAX_RETRIES", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + os.environ["ZCODE_BRIDGE_MAX_RETRIES"] = "2" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertTrue(result.get("isError"), "重试耗尽应返回错误") + self.assertIn("已重试", result["content"][0]["text"]) + self.assertEqual(calls["n"], 3, "max_retries=2 → 共调用 3 次 (1+2 重试)") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + if old_mr is None: + os.environ.pop("ZCODE_BRIDGE_MAX_RETRIES", None) + else: + os.environ["ZCODE_BRIDGE_MAX_RETRIES"] = old_mr + + # ---------- RT2: 非限流不重试 ---------- + def test_rt2_non_rate_limit_no_retry(self): + """RT2: Unauthorized (provider_error) → 不重试, 直接返回""" + mod = self.mod + procs = [_FakeCompletedProcess(returncode=1, stdout="", + stderr="APICallError: Unauthorized")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertTrue(result.get("isError")) + self.assertEqual(calls["n"], 1, "非限流错误不应重试") + self.assertNotIn("已重试", result["content"][0]["text"], + "未重试不应出现'已重试'字样") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + # ---------- RT3: 首次成功 ---------- + def test_rt3_success_first_try(self): + """RT3: 首次成功 → 不重试""" + mod = self.mod + procs = [_FakeCompletedProcess(returncode=0, stdout="OK", stderr="")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertNotIn("isError", result) + self.assertEqual(calls["n"], 1) + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + # ---------- RT4: 配额错误不重试 ---------- + def test_rt4_quota_no_retry(self): + """RT4: quota exceeded (配额耗尽) → 不重试""" + mod = self.mod + procs = [_FakeCompletedProcess(returncode=1, stdout="", + stderr="quota exceeded")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertTrue(result.get("isError")) + self.assertEqual(calls["n"], 1, "配额耗尽不应重试") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + # ---------- RT5: exit=0 + stderr=Unauthorized 不当成功 (Codex P1-1) ---------- + def test_rt5_exit0_with_stderr_error_not_success(self): + """RT5: zcode exit=0 但 stderr 含 APICallError: Unauthorized → 当错误返回 (Codex P1-1) + + zcode 错误常 exit=0 + stdout 空 + stderr 含堆栈。修复前 parser 副本缺 + provider_error 分支, 会误判为成功空输出。补分支后应识别为错误。 + """ + mod = self.mod + procs = [_FakeCompletedProcess( + returncode=0, stdout="", + stderr="APICallError [AI_APICallError]: Unauthorized\n at ...")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertTrue(result.get("isError"), + "exit=0+stderr=Unauthorized 应识别为错误, 不当成功空输出") + self.assertEqual(calls["n"], 1, "provider_error 不重试") + self.assertIn("Unauthorized", result["content"][0]["text"]) + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + # ---------- RT6: stdout 含限流词 + stderr 非空 → 不误判限流 (Claude P1-1) ---------- + def test_rt6_review_text_with_rate_keywords_not_misjudged(self): + """RT6: 审查"限流逻辑"代码的结论含 429/频繁, 但 stderr 非空诊断 → 不当限流重试 + + Claude review P1-1: 若把 stdout 审查正文喂进 parse_provider_error, 审查限流 + 相关代码会被误判限流而白重试。修复: 只解析 stderr。 + """ + mod = self.mod + # stdout 是审查结论 (含 429/频繁), stderr 有诊断 (非限流) + procs = [_FakeCompletedProcess( + returncode=0, + stdout="这段代码处理 429 限流和频繁请求的逻辑建议加退避。额度计算也有问题。", + stderr="some diagnostic noise\nfinished")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + # 关键: 不应重试 (calls=1), 不应 isError, 应返回审查结论 + self.assertEqual(calls["n"], 1, "审查正文含限流词不应触发重试") + self.assertNotIn("isError", result, "成功审查不应被误判为错误") + self.assertIn("429", result["content"][0]["text"], "审查结论应原样返回") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/test_polling_failure.py b/tests/test_polling_failure.py new file mode 100644 index 0000000..c4589dd --- /dev/null +++ b/tests/test_polling_failure.py @@ -0,0 +1,158 @@ +""" +test_polling_failure.py — 轮询/停滞收尾的 turn 失败检测单测 + +修复 PR#2 遗留 (issue #3 子项3d): 轮询路径 _run_polling_turn 和事件路径停滞收尾 +都不感知 turn.failed (zcode 只在事件 payload 给 resultType, projection/messages 无失败标志, +实测确认)。修复方式: turn "完成"(idle)但无任何有效输出 (text/tool/patch) 时, 疑似失败, +返回 -32603 而非静默 end_turn。 + +用增强 FakeBackend (按方法路由响应) + patch time.sleep 模拟 turn 流程, 不真跑 zcode。 + + PF0 轮询: turn 完成(idle)有正常输出 → end_turn (原有行为不破坏) + PF1 轮询: turn 完成(idle)但无输出 (失败) → -32603 (修复后) + PF2 轮询: turn 从未启动 → -32603 "未启动" + +运行: python3 tests/test_polling_failure.py +依赖: 仅 Python 标准库 + acp-bridge 模块 +""" + +import os +import types +import unittest + +BRIDGE_PATH = os.path.join( + os.path.dirname(__file__), "..", "packages", "acp-bridge", "zcode-acp-bridge" +) + + +def _load_bridge_module(): + 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 _RoutingBackend: + """按 method 路由响应的 FakeBackend; 可记录调用次数。 + + responses: {method: response_dict} 或 {method: [resp1, resp2, ...]} (序列) + """ + + def __init__(self, responses): + self.responses = responses + self.calls = [] + self._seq_idx = {} + + def request(self, msg_id, method, params=None, timeout=30): + self.calls.append(method) + resp = self.responses.get(method) + if isinstance(resp, list): + i = self._seq_idx.get(method, 0) + self._seq_idx[method] = i + 1 + return (resp[i] if i < len(resp) else (resp[-1] if resp else {"result": {}})), [] + return (resp if resp is not None else {"result": {}}), [] + + def send(self, msg): + pass + + +class TestPollingFailureDetection(unittest.TestCase): + """轮询/停滞收尾的 turn 失败检测 (子项3d)""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_bridge_module() + cls.Bridge = cls.mod.ACPBridge + # bridge 用 `import time; time.sleep(...)` 和 `time.time()`。 + # patch 全局 time.sleep 为 no-op + time.time 快速推进, 否则轮询循环 + # 的 120s 超时会真跑满 (测试会卡 120s/用例)。 + import time as _time_mod + cls._real_sleep = _time_mod.sleep + cls._real_time = _time_mod.time + _t = [0.0] + _time_mod.sleep = lambda _s: None + # time.time 每次调用推进 0.6s (≈轮询间隔), 让 120s 超时在 ~200 次瞬时迭代后触发, + # 而非真实等待 120s。PF0/PF1 只需 2~3 次迭代就能走完 running→idle。 + _time_mod.time = lambda: (_t.__setitem__(0, _t[0] + 0.6), _t[0])[1] + + @classmethod + def tearDownClass(cls): + import time as _time_mod + _time_mod.sleep = cls._real_sleep + _time_mod.time = cls._real_time + + def _new_bridge(self, responses): + b = self.Bridge() + b.backend = _RoutingBackend(responses) + return b + + def _run_polling(self, bridge, zcode_sid="sess_test", projection_seq=None, + messages=None): + """直接调 _run_polling_turn (跳过 _on_session_prompt 的前置)。""" + acp_sid = zcode_sid + bridge.session_map[acp_sid] = zcode_sid + msg_id = 1 + turn = {"zcode_sid": zcode_sid, "cancelled": False, "perms_responses": {}} + bridge.pending_turns[msg_id] = turn + differ = bridge._get_or_create_differ(zcode_sid) + return bridge._run_polling_turn(acp_sid, zcode_sid, msg_id, turn, + chunk_msg_id="chunk_1", differ=differ) + + def test_pf0_polling_success_with_output(self): + """PF0: 轮询 turn 完成且有文本输出 → 正常 end_turn (原有行为不破坏)""" + bridge = self._new_bridge({ + # session/read: 先 running 再 idle (poll_once 调 read) + "session/read": [ + {"result": {"projection": {"status": "running", "totalTokenCount": 100, + "contextWindow": 1000000}}}, + {"result": {"projection": {"status": "idle", "totalTokenCount": 100, + "contextWindow": 1000000}}}, + ], + # session/messages: 返回有文本的 assistant 消息 + "session/messages": {"result": {"messages": [ + {"info": {"role": "user"}, "parts": [{"type": "text", "text": "hi"}]}, + {"info": {"role": "assistant"}, "parts": [{"type": "text", "text": "你好"}]}, + ], "todos": []}}, + }) + resp = self._run_polling(bridge) + self.assertNotIn("error", resp, "有输出应正常 end_turn") + self.assertEqual(resp["result"]["stopReason"], "end_turn") + + def test_pf1_polling_failure_no_output(self): + """PF1: 轮询 turn 完成但无任何输出 (失败) → -32603 (修复后不再静默 end_turn)""" + bridge = self._new_bridge({ + "session/read": [ + {"result": {"projection": {"status": "running", "totalTokenCount": 100, + "contextWindow": 1000000}}}, + {"result": {"projection": {"status": "idle", "totalTokenCount": 100, + "contextWindow": 1000000}}}, + ], + # messages 只有 step-start 半成品 (失败 turn 的典型表现, 实测确认) + "session/messages": {"result": {"messages": [ + {"info": {"role": "user"}, "parts": [{"type": "text", "text": "hi"}]}, + {"info": {"role": "assistant"}, "parts": [{"type": "step-start"}]}, + ], "todos": []}}, + }) + resp = self._run_polling(bridge) + self.assertIn("error", resp, "无输出应判定疑似失败, 返回 error") + self.assertEqual(resp["error"]["code"], -32603) + self.assertIn("无输出", resp["error"]["message"]) + + def test_pf2_polling_turn_never_started(self): + """PF2: turn 从未启动 (status 一直非 running) → -32603 '未启动'""" + bridge = self._new_bridge({ + # status 始终 idle (turn 没启动过) + "session/read": {"result": {"projection": {"status": "idle", + "totalTokenCount": 0, + "contextWindow": 1000000}}}, + "session/messages": {"result": {"messages": [], "todos": []}}, + }) + resp = self._run_polling(bridge) + self.assertIn("error", resp) + self.assertIn("未启动", resp["error"]["message"]) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/test_provider_error.py b/tests/test_provider_error.py new file mode 100644 index 0000000..af97d10 --- /dev/null +++ b/tests/test_provider_error.py @@ -0,0 +1,189 @@ +""" +test_provider_error.py — parse_provider_error 错误解析单测 + +验证 shared/provider_error.py 能正确分类 zcode headless 调用返回的 provider 错误, +并提取 retry-after, 用于 MCP server 的有限重试决策 (issue #3 子项3a)。 + + PE0 限流 429 (HTTP 通用) + PE1 限流 1302 (z.ai 特定业务码) + PE2 限流中文文案 (请求过于频繁 / 频繁访问) + PE3 限流 + retry-after 秒数提取 (header / retryAfter / 中文 N秒后 / try again in) + PE4 纯 retry-after 也判定为限流 (很多 API 错误体只给 retry-after) + PE5 配额耗尽 (quota / 余额不足, 不可短期重试) + PE6 其他 provider 错误 (Unauthorized, 不重试) + PE7 空文本 / 无错误文本 → unknown + PE8 retry-after 边界 (超大值截断到 300, 含堆栈的长文本只看头部) + +运行: python3 tests/test_provider_error.py +依赖: 仅 Python 标准库 + shared/provider_error.py +""" + +import os +import sys +import unittest + +# shared/provider_error.py 是普通 .py, 直接 import +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "shared")) +from provider_error import parse_provider_error as pe # noqa: E402 + + +class TestParseProviderError(unittest.TestCase): + """parse_provider_error 的分类与 retry-after 提取单测""" + + def _rl(self, text): + """断言是限流, 返回完整结果供进一步检查。""" + r = pe(text) + self.assertTrue(r["is_rate_limit"], f"应判定为限流: {text!r} → {r}") + return r + + def _not_rl(self, text, kind=None): + r = pe(text) + self.assertFalse(r["is_rate_limit"], f"不应判定为限流: {text!r} → {r}") + if kind: + self.assertEqual(r["error_kind"], kind) + return r + + # ---------- PE0: 429 通用限流 ---------- + def test_pe0_429_rate_limit(self): + """PE0: 含 429 → rate_limit""" + r = self._rl("APICallError: Too Many Requests 429") + self.assertEqual(r["error_kind"], "rate_limit") + + def test_pe0b_too_many_requests(self): + """PE0b: "Too Many Requests" 文案 (无数字) → rate_limit""" + self._rl("Error: Too Many Requests. Please slow down.") + + # ---------- PE1: 1302 z.ai 特定码 ---------- + def test_pe1_1302_zai_code(self): + """PE1: z.ai 业务码 1302 → rate_limit""" + r = self._rl("Error 1302: 请求过于频繁") + self.assertEqual(r["error_kind"], "rate_limit") + + def test_pe1b_1302_alone(self): + """PE1b: 单独 1302 → rate_limit""" + self._rl("request failed with code 1302") + + # ---------- PE2: 中文限流文案 ---------- + def test_pe2_chinese_rate_limit(self): + """PE2: 中文"请求过于频繁"/"频繁访问" → rate_limit""" + self._rl("错误: 请求过于频繁,请稍后重试") + self._rl("频繁访问, 触发限流") + + def test_pe2b_rate_limit_snake(self): + """PE2b: rate_limit_error 文案 → rate_limit""" + self._rl('{"error":{"type":"rate_limit_error"}}') + + # ---------- PE3: retry-after 提取 ---------- + def test_pe3_retry_after_header(self): + """PE3: retry-after header 秒数提取""" + r = self._rl("429 Too Many Requests\nretry-after: 30") + self.assertEqual(r["retry_after_sec"], 30) + + def test_pe3b_retry_after_retryafter_json(self): + """PE3b: retryAfter JSON 字段""" + r = self._rl('429 {"error":{"retryAfter": "45"}}') + self.assertEqual(r["retry_after_sec"], 45) + + def test_pe3c_retry_after_chinese_seconds(self): + """PE3c: 中文 "请在 N 秒后" 提取""" + r = self._rl("429 请在 60 秒后重试") + self.assertEqual(r["retry_after_sec"], 60) + + def test_pe3d_retry_after_try_again(self): + """PE3d: "try again in N seconds" 提取""" + r = self._rl("429 Please try again in 20 seconds") + self.assertEqual(r["retry_after_sec"], 20) + + # ---------- PE4: 纯 retry-after 判定限流 ---------- + def test_pe4_bare_retry_after_is_rate_limit(self): + """PE4: 文本只有 retry-after (无 429) 也判为限流""" + r = self._rl("retry-after: 30") + self.assertTrue(r["is_rate_limit"]) + self.assertEqual(r["retry_after_sec"], 30) + + # ---------- PE5: 配额耗尽 ---------- + def test_pe5_quota_not_retryable(self): + """PE5: quota/配额 → is_quota, 非限流 (不可短期重试)""" + r = self._not_rl("quota exceeded for this month", kind="quota") + self.assertTrue(r["is_quota"]) + + def test_pe5b_balance_insufficient(self): + """PE5b: 余额不足 → quota""" + r = self._not_rl("余额不足, 请充值", kind="quota") + self.assertTrue(r["is_quota"]) + + def test_pe5c_quota_with_retry_after(self): + """PE5c: quota 仍可携带 retry_after (供调用方决策; 超大值被截断到 300)""" + r = pe("quota exceeded, retry-after: 3600") + self.assertTrue(r["is_quota"]) + self.assertFalse(r["is_rate_limit"]) + self.assertEqual(r["retry_after_sec"], 300, "retry_after 被截断到 300") + + # ---------- PE5d: rate limit exceeded 不被 quota 误吃 (Codex P1#2) ---------- + def test_pe5d_rate_limit_exceeded_not_quota(self): + """PE5d: 'rate limit exceeded' 应判为限流 (rate_limit), 而非被 quota 的 exceeded.*limit 吃掉""" + r = self._rl("Rate limit exceeded, retry-after: 30") + self.assertEqual(r["error_kind"], "rate_limit") + self.assertFalse(r["is_quota"]) + + def test_pe5e_quota_vs_rate_limit_disambig(self): + """PE5e: quota 需带限定词, 裸'额度/余额'不当配额 (避免审查正文误判)""" + # 真配额 + self.assertTrue(pe("额度不足")["is_quota"]) + # 裸词不当配额 (审查正文可能含) + self.assertNotIn(pe("代码里的额度计算逻辑")["error_kind"], ["quota"]) + self.assertNotIn(pe("频繁调用的优化建议")["error_kind"], ["quota", "rate_limit"]) + + # ---------- PE6: 其他 provider 错误 ---------- + def test_pe6_unauthorized_not_rate_limit(self): + """PE6: Unauthorized → provider_error, 非限流 (不重试)""" + self._not_rl("APICallError: Unauthorized", kind="provider_error") + + def test_pe6b_invalid_key(self): + """PE6b: invalid api key → provider_error""" + self._not_rl("invalid x-api-key", kind="provider_error") + + # ---------- PE7: 空文本 / 无错误 ---------- + def test_pe7_empty_text(self): + """PE7: 空文本 → unknown""" + r = pe("") + self.assertFalse(r["is_rate_limit"]) + self.assertEqual(r["error_kind"], "unknown") + self.assertIsNone(r["retry_after_sec"]) + + def test_pe7b_no_error_text(self): + """PE7b: 正常无错文本 → unknown""" + r = self._not_rl("这是一段正常的代码审查结论, 没有错误") + self.assertEqual(r["error_kind"], "unknown") + + def test_pe7c_none_text(self): + """PE7c: None 入参 → unknown (不崩)""" + r = pe(None) + self.assertEqual(r["error_kind"], "unknown") + + # ---------- PE8: 边界 ---------- + def test_pe8_retry_after_bounded(self): + """PE8: retry-after 超大值截断到 300 (防 provider 异常导致长眠)""" + r = self._rl("429 retry-after: 99999") + self.assertEqual(r["retry_after_sec"], 300) + + def test_pe8b_long_stacktrace_head_recognized(self): + """PE8b: 超长文本, 错误在头部 (<2000) → 能识别; 在尾部 (>2000) → 看不到""" + # 错误在头部, 能识别 + head_text = "APICallError: 429 Too Many Requests\n" + "x" * 3000 + self.assertTrue(pe(head_text)["is_rate_limit"], "头部含 429 应被识别") + # 错误在尾部 (>2000 字符后), 头部全是填充 → 看不到 (这是已知的截断行为) + tail_text = "x" * 3000 + "\nAPICallError: 429 Too Many Requests" + self.assertFalse(pe(tail_text)["is_rate_limit"], + "尾部 (>2000) 的错误不在取样头部, 识别不到 (已知截断行为)") + + def test_pe8c_raw_hint_skips_stacktrace(self): + """PE8c: raw_hint 跳过 JS 堆栈行, 取有意义的首行""" + text = " at /Applications/ZCode.app/...\n at process.x\nAPICallError: 429\n" + r = pe(text) + self.assertNotIn("at ", r["raw_hint"], "raw_hint 不应是堆栈行") + self.assertIn("APICallError", r["raw_hint"]) + + +if __name__ == "__main__": + unittest.main(verbosity=2)