diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6c9e6d0..2d19d13 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,7 +4,7 @@ # 三个 job: # 1. lint — ruff 检查 (代码风格 + bug 类规则) # 2. test — unittest 矩阵 (Python 3.11 / 3.13) -# 3. perms — 三个组件的可执行位校验 (本项目历史反复踩的坑: +# 3. perms — 四个组件的可执行位校验 (本项目历史反复踩的坑: # 编辑器/工具常把 100755 改回 100644, 导致用户下载后 # 无法直接执行。CI 在此兜底拦截) # @@ -32,11 +32,11 @@ jobs: with: python-version: "3.13" - # 只装 ruff; 运行时三个组件零第三方依赖 + # 只装 ruff; 运行时四个组件零第三方依赖 - name: Install ruff run: pip install ruff==0.15.17 - # 注意: 三个组件是「无后缀单文件脚本」, ruff 默认只扫 .py, + # 注意: 四个组件是「无后缀单文件脚本」, ruff 默认只扫 .py, # 必须显式把组件路径传进去, 否则会漏检核心代码。 - name: ruff check run: | @@ -44,6 +44,7 @@ jobs: packages/acp-bridge/zcode-acp-bridge \ packages/agent-help/zcode-agent-help \ packages/mcp-server/zcode-mcp-server \ + packages/review-gate/zcode-review-gate \ shared/ \ tests/ @@ -75,7 +76,7 @@ jobs: steps: - uses: actions/checkout@v4 - # 校验三个组件在 git 里保持可执行 (100755)。 + # 校验四个组件在 git 里保持可执行 (100755)。 # 历史上编辑器/Edit 类工具多次把权限改回 100644, 用户下载后 # 无法直接 ./zcode-xxx 运行。这一步专门兜底。 - name: Verify components are executable @@ -85,7 +86,8 @@ jobs: for f in \ packages/acp-bridge/zcode-acp-bridge \ packages/agent-help/zcode-agent-help \ - packages/mcp-server/zcode-mcp-server; do + packages/mcp-server/zcode-mcp-server \ + packages/review-gate/zcode-review-gate; do mode=$(git ls-files --stage -- "$f" | awk '{print $1}') if [ "$mode" != "100755" ]; then echo "❌ $f 模式为 $mode, 应为 100755 (可执行)" diff --git a/README.md b/README.md index 06386ea..44ab32e 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,7 @@ ZCode 是智谱 Z.AI 出品的 AI 编程 Agent，由 GLM 系列模型驱动。 | **zcode-agent-help** | 能力发现说明书：一次调用了解 ZCode 全部能力 | ✅ stable | [packages/agent-help](packages/agent-help) | | **zcode-mcp-server** | 把 ZCode 暴露为标准 MCP server，供 MCP client（Claude Code/Cursor/自身）调用 | ✅ stable | [packages/mcp-server](packages/mcp-server) | | **zcode-acp-bridge** | 把 ZCode 桥接为 ACP Agent，供 Zed/JetBrains 等编辑器调用 | ⚠️ experimental | [packages/acp-bridge](packages/acp-bridge) | +| **zcode-review-gate** | PR 自动审查闸门：轮询 open PR，新 head 调 `zcode_pr_review` 审查并回贴 verdict 评论 | ⚠️ experimental | [packages/review-gate](packages/review-gate) | ## 快速开始 @@ -199,6 +200,14 @@ ACP bridge 额外暴露了 ZCode 新版协议方法，供编辑器/脚本调用 > workspace 参数可三种方式传入：`workspace`（完整 dict）、`workspacePath`/`cwd`（路径字符串），或缺省时用 bridge 进程的 `cwd`。 > Provider 管理类方法（upsert/remove/updateRegistry）的 `provider`/`registry` 可能含 `apiKey`，bridge 仅透传、不读取/打印其明文。 +### Review gate（`zcode-review-gate`） + +PR 自动审查闸门守护进程（第 4 组件，experimental）：常驻轮询配置仓库的 open PR，对每个新 head sha 经 `zcode-mcp-server --call zcode_pr_review` 完成审查（锁/重试/只读护栏全部复用 bridge 同源路径），把带 verdict（✅ pass / ⚠️ concerns）的结果回贴为 PR 评论。同一 head sha 不重复审（state 文件去重），失败按指数退避重试，head 更新自动复活重审。token 不落盘（经 git≥2.31 的 `GIT_CONFIG_*` 环境变量进程内注入，不进 argv）。公开、通用，任何 GitHub 仓库可用。 + +安装、配置参考、systemd 部署与运维详见 [packages/review-gate/README.md](packages/review-gate/README.md)。 + +> 配套能力：mcp-server 新增 `--call TOOL ''` 一次性调用模式（脚本化入口，exit 0/1/2 分别对应 成功 / 用法错误或 handler 异常 / tool 执行失败），stdio 模式行为不变。 + ## 会话存储 `--prompt` 和 ACP bridge **共享同一套会话存储**（`~/.zcode/cli/db/db.sqlite`），互通互恢复： @@ -359,7 +368,8 @@ zcode-open-bridge/ ├── packages/ │ ├── agent-help/ # 能力发现说明书 (stable) │ ├── mcp-server/ # MCP 桥接 (stable) -│ └── acp-bridge/ # ACP 桥接 (experimental) +│ ├── acp-bridge/ # ACP 桥接 (experimental) +│ └── review-gate/ # PR 自动审查闸门 (experimental) ├── shared/ │ └── credentials.py # 凭证读取 (单一真相源) ├── skills/ diff --git a/packages/mcp-server/zcode-mcp-server b/packages/mcp-server/zcode-mcp-server index 0015851..5b315f7 100755 --- a/packages/mcp-server/zcode-mcp-server +++ b/packages/mcp-server/zcode-mcp-server @@ -1581,5 +1581,48 @@ def main(): log("stdin 关闭, 退出") +def call_once(argv): + """一次性调用模式: --call TOOL_NAME '' → 结果 JSON 打印到 stdout。 + + 用途: 脚本化/守护进程 (如 packages/review-gate) 的单次调用入口 — 不用起 + stdio MCP 会话, 直接拿一个 tool 的结果。锁/重试/yolo 只读护栏全走 + TOOL_HANDLERS 路径, 与 stdio 模式完全同源 (不重造任何安全机制)。 + + 退出码约定: + 0 — 调用成功 ({"ok": true, "result": ...}) + 1 — 用法错误 (参数个数不对 / 未知 tool / JSON 解析失败 / handler 抛异常) + 2 — tool 执行失败 (result 带 isError, 如目录不存在、diff 为空等) + """ + if len(argv) != 2: + log("用法: zcode-mcp-server --call TOOL_NAME ''") + return 1 + tool_name, raw_args = argv + if tool_name not in TOOL_HANDLERS: + log(f"未知 tool: {tool_name} (可选: {sorted(TOOL_HANDLERS)})") + return 1 + try: + arguments = json.loads(raw_args) + except json.JSONDecodeError as e: + log(f"args JSON 解析失败: {e}") + return 1 + if not isinstance(arguments, dict): + log("args 必须是 JSON 对象") + return 1 + try: + result = TOOL_HANDLERS[tool_name](arguments) + except Exception as e: + # 与 stdio 路径一致: 内部细节不外泄到 stdout, 只走 log (P1-5) + log(f"tool {tool_name} 异常: {e}") + print(json.dumps({"ok": False, "error": "内部错误, 详见 stderr 日志"}, + ensure_ascii=False)) + return 1 + out = {"ok": not result.get("isError"), "result": result} + print(json.dumps(out, ensure_ascii=False)) + return 0 if out["ok"] else 2 + + if __name__ == "__main__": + # argv 带 --call → 一次性调用模式; 无参数 → 原 stdio JSON-RPC 循环 + if len(sys.argv) > 1 and sys.argv[1] == "--call": + sys.exit(call_once(sys.argv[2:])) main() diff --git a/packages/review-gate/README.md b/packages/review-gate/README.md new file mode 100644 index 0000000..37f9799 --- /dev/null +++ b/packages/review-gate/README.md @@ -0,0 +1,184 @@ +# zcode-review-gate — PR 自动审查闸门 + +常驻轮询守护进程：监控配置仓库的 open PR，对每个新 head sha 调 bridge 的 +`zcode_pr_review` 完成审查（git diff + mimosa 深扫 + ZCode 只读复核）， +把带 verdict（pass / concerns）的结果回贴为 PR 评论。同一 head sha 不重复审 +（state 文件去重），失败按指数退避重试。公开、通用，任何 GitHub 仓库可用。 + +``` +┌─────────────┐ 轮询 open PR ┌──────────────┐ +│ GitHub API │ ◄────────────── │ │ +└──────┬──────┘ │ │ + │ 新 head sha? │ review-gate │ (state 文件去重/退避) + ▼ │ │ + git clone/fetch ──────────────► │ │ + │ └──────┬───────┘ + ▼ │ --call zcode_pr_review +┌─────────────────┐ ▼ +│ zcode-mcp-server│ (git diff + mimosa 深扫 + ZCode 只读复核, +│ (子进程) │ 锁/重试/只读护栏全在 bridge 侧同源复用) +└──────┬──────────┘ + │ 报告 → 解析 P0/P1/P2 → verdict + ▼ +┌─────────────┐ +│ PR 评论 │ ✅ pass / ⚠️ concerns + 完整报告 (details 折叠) +└─────────────┘ +``` + +## 前置条件 + +- 已安装 [ZCode](https://zcode.z.ai) CLI 并完成登录（凭证在 `~/.zcode/v2/config.json`） +- 已安装 mimosa 安全扫描插件（`zcode_pr_review` 依赖；见顶层 README） +- `zcode-mcp-server` 可用（本仓库 `packages/mcp-server`，带 `--call` 模式）： + cp/ln 到 `~/.local/bin/` 或在配置里用 `mcp_server` 指绝对路径 +- git 可用 +- **GitHub token**（repo scope）：拉取公开仓可匿名，但**发评论必须 token**。 + 解析级联：`GITHUB_TOKEN` env → `GH_TOKEN` env → `gh auth token` + +## 安装 + +```bash +# 1. 脚本就位 (二选一) +cp packages/review-gate/zcode-review-gate ~/.local/bin/ +# 或: ln -s "$(pwd)/packages/review-gate/zcode-review-gate" ~/.local/bin/ + +# 2. 写配置 +mkdir -p ~/.config/zcode-review-gate +cat > ~/.config/zcode-review-gate/config.json <<'EOF' +{ + "repos": ["owner/your-repo"], + "poll_interval_seconds": 300 +} +EOF + +# 3. 先手动跑一轮验证 (看 stderr 日志, 不装 daemon 也能用) +zcode-review-gate --once + +# 4. 装 systemd --user 单元 +mkdir -p ~/.config/systemd/user +cp packages/review-gate/zcode-review-gate.service ~/.config/systemd/user/ +systemctl --user daemon-reload +systemctl --user enable --now zcode-review-gate + +# 5. 服务器上希望注销后仍运行 (无 linger 时 user manager 随最后一个会话退出) +loginctl enable-linger "$USER" +``` + +注意事项： + +- **PATH**：systemd --user 默认 `PATH=/usr/local/bin:/usr/bin:/bin`，不含 + `~/.local/bin`（GC-8G 实测踩坑）。unit 模板已默认带 + `Environment=PATH=%h/.local/bin:...`；此外 gate 解析 `mcp_server` 纯 + 命令名时若 PATH 找不到会自动回退试 `~/.local/bin/`，双保险。 + 但 `git`、`zcode` 仍依赖 PATH——自己改 unit 时别把默认 PATH 行删掉。 +- **sudo -u 场景**：用 `sudo -u systemctl --user ...` 操作别人的 + user manager 时，需要 `XDG_RUNTIME_DIR=/run/user/$(id -u )`， + 否则连不上 user bus。 +- **token**：不要把 token 写进 config.json（脚本也不读）。用 + `systemctl --user edit zcode-review-gate` 加 + `Environment=GITHUB_TOKEN=...` override，或 `gh auth login` 后靠级联解析。 + +## 配置参考 + +`~/.config/zcode-review-gate/config.json`（全部键可选，有默认值）： + +| 键 | 默认 | 说明 | +|---|---|---| +| `repos` | `[]` | 监控的仓库列表，`owner/name` 格式。**缺省且 CLI 无 `--repo` 时拒绝启动** | +| `poll_interval_seconds` | `300` | 轮询间隔（秒） | +| `state_file` | `~/.local/state/zcode-review-gate/state.json` | 状态文件（去重/退避） | +| `clone_root` | `~/.local/state/zcode-review-gate/clones` | 仓库 clone 存放目录 | +| `mcp_server` | `zcode-mcp-server` | bridge mcp-server 可执行名/路径（须支持 `--call`）。纯命令名且 PATH 找不到时自动回退试 `~/.local/bin/`（存在且可执行才用，log DEBUG 记录解析结果） | +| `github_api` | `https://api.github.com` | GitHub API base（企业版可改）。clone 的 web 宿主按惯例推导：`api.github.com`→`github.com`，`/api/v3`→``（GHE），其他形态回退 `github.com` | +| `review.depth` | `deep` | 审查深度：`normal`（快扫）/ `deep`（含业务逻辑投研） | +| `review.focus` | `""` | 额外审查重点（透传给 zcode prompt） | +| `retry.base_seconds` | `300` | 退避基数：失败后 `base * 2^(attempts-1)` | +| `retry.max_seconds` | `3600` | 退避封顶 | +| `retry.max_attempts` | `5` | 最大尝试次数，达到后 `gave_up`（head 更新才复活） | +| `comment.enabled` | `true` | 是否回贴 PR 评论（false 时只审只落状态，dry-run 用） | +| `comment.max_body` | `60000` | 评论体上限（GitHub 硬上限 65536，留余量；超出截断报告正文） | + +环境变量覆盖（优先级高于配置文件）： + +| env | 覆盖的键 | +|---|---| +| `GATE_CONFIG` | 配置文件路径本身（等价 `--config`） | +| `GATE_STATE_FILE` | `state_file` | +| `GATE_CLONE_ROOT` | `clone_root` | +| `GATE_MCP_SERVER` | `mcp_server` | +| `GATE_POLL_INTERVAL` | `poll_interval_seconds` | + +路径值支持 `~` 展开。 + +## 卸载 + +```bash +systemctl --user disable --now zcode-review-gate +rm ~/.config/systemd/user/zcode-review-gate.service +systemctl --user daemon-reload +# 按需清理: ~/.local/bin/zcode-review-gate +# ~/.config/zcode-review-gate/ +# ~/.local/state/zcode-review-gate/ (state + clones, 可能很大) +``` + +## 运维 + +```bash +# 看日志 +journalctl --user -u zcode-review-gate -f + +# 手动跑一轮 (调试/cron) +zcode-review-gate --once + +# 只审指定 PR (实测用, 不影响其他 PR) +zcode-review-gate --once --repo owner/repo --pr 5 + +# 更细的日志 +zcode-review-gate --once --log-level DEBUG +``` + +state 文件（默认 `~/.local/state/zcode-review-gate/state.json`）记录每个 PR 的 +审查状态：`head_sha` / `status` / `verdict` / `counts` / `report` / +`attempts` / `next_retry_at` / `comment_url` / `error`。 +`status ∈ pending|reviewed|failed|comment_failed|gave_up`：`pending` 是 +一轮处理中的瞬态；`comment_failed` 表示审查已成功但评论没发出去——此时 +`report`/`verdict`/`counts` 已缓存，重试时同 head **只补评论不重跑审查**； +`reviewed` 后 `report` 缓存清空。 +**想强制重审某个 PR：删掉对应条目**（或把 PR 推一个新 commit，head 变化会 +自动复活重审）。 +注意：`gave_up` 时缓存的 `report`（每条最多 `comment.max_body` 字符）会 +留在 state 文件里供人工排查——长期积攒关注 state 文件体量，可定期清理 +已完结 PR 的条目。 + +deep 档审查耗时长：gate 起审查子进程时已自动透传 +`ZCODE_BRIDGE_REVIEW_TIMEOUT=3600`，自身等子进程的总超时再加 120s 留给 +mimosa 扫描与进程收尾（=3720），mcp-server 侧不会以默认 300s 提前掐断 +zcode。 + +## 限制 + +- **评论以 token 身份发出**：用什么 token 评论就显示什么账号，建议专用 bot + 账号或 GitHub App token。 +- **评论 at-least-once**：评论请求发出后响应丢失（超时/连接断开）会按失败 + 重试，而 GitHub issue comments 没有幂等键——极端情况下同一 head 可能 + 出现重复评论，属已知限制（方向仍是宁多勿漏）。 +- **verdict 依赖报告文本解析**：从报告开头解析 `P0/P1/P2` 条数得出 + pass/concerns；解析失败时 fail-safe 为 **concerns**（宁错拦不错放）， + 评论里会标注"严重度分布解析失败，请人工核对"。 +- **单线程串行**：逐仓逐 PR 串行审查；并发安全靠 bridge mcp-server 侧的 + 跨进程文件锁兜底（多实例同时跑也不会并发打爆 zcode 限流）。 +- **fork PR**：走 `refs/pull/{n}/head` 拉取，无需加 fork 远端； + 审查的是 PR head 快照本身。 +- token 不落盘：经 git≥2.31 的 `GIT_CONFIG_COUNT/KEY/VALUE` 环境变量逐 + 命令注入 `http.extraHeader`（env 只对本用户可见，优于 argv）， + clone URL / git config / state 文件里都不会有 token。 + 认证形态分两路：git smart-HTTP 走 **Basic** header + （`x-access-token:` 的 base64——实测 GitHub 的 git 端点拒绝 + OAuth token 的 Bearer 形式），REST API 走 **Bearer** header。 + +## 测试 + +```bash +python3 tests/test_review_gate.py # 组件单测 (全 fake, 不碰网络/git/真实路径) +python3 tests/test_mcp_call.py # mcp-server --call 模式单测 +``` diff --git a/packages/review-gate/zcode-review-gate b/packages/review-gate/zcode-review-gate new file mode 100755 index 0000000..21d0f58 --- /dev/null +++ b/packages/review-gate/zcode-review-gate @@ -0,0 +1,928 @@ +#!/usr/bin/env python3 +""" +zcode-review-gate — PR 自动审查闸门守护进程 (zcode-open-bridge 第 4 组件) + +常驻轮询配置的 GitHub 仓库: 对每个 open PR 的新 head sha, 通过 bridge 的 +`zcode-mcp-server --call zcode_pr_review` 完成审查 (git diff + mimosa 深扫 ++ ZCode 只读复核), 把带 verdict (pass/concerns) 的结果回贴为 PR 评论。 + +设计要点: + - 审查一律经 mcp-server 子进程 (--call 一次性模式) 完成: 跨进程文件锁、 + 限流退避重试、yolo 只读护栏全部复用 TOOL_HANDLERS 同源路径, 本组件 + 不 import/exec mcp-server 源码, 更不重造这些机制。 + - 同一 head sha 不重复审 (state 文件去重); 失败按指数退避重试, 超限 + gave_up; head 更新 (force-push / 新 commit) 自动复活重审。 + - token 不落盘: 只经 git≥2.31 的 GIT_CONFIG_COUNT/KEY/VALUE 环境变量 + 逐命令进程内注入 http.extraHeader (env 只对本用户可见, 优于 argv); + clone URL / .git/config / state / config 文件里都没有 token。 + - verdict 解析是确定性 fail-safe: 严重度分布解析不出来一律按 concerns。 + +依赖: 仅 Python3 标准库 (零第三方依赖) +日志: 全部走 stderr (带时间戳前缀) +""" + +import argparse +import base64 +import json +import os +import re +import shutil +import signal +import subprocess +import sys +import time +import urllib.error +import urllib.request +from datetime import datetime + +VERSION = "0.1.0" + +DEFAULT_CONFIG_PATH = "~/.config/zcode-review-gate/config.json" +DEFAULT_STATE_FILE = "~/.local/state/zcode-review-gate/state.json" +DEFAULT_CLONE_ROOT = "~/.local/state/zcode-review-gate/clones" +DEFAULT_MCP_SERVER = "zcode-mcp-server" +DEFAULT_GITHUB_API = "https://api.github.com" + +HTTP_TIMEOUT = 30 # GitHub API 单请求超时 (秒) +GIT_TIMEOUT = 120 # 单条 git 命令超时 (秒; fetch/rev-parse 等) +CLONE_TIMEOUT = 600 # clone 单独放宽 (大仓首次克隆慢) +ZCODE_REVIEW_TIMEOUT = 3600 # 传给 mcp-server 的 zcode 审查预算 (透传 env; + # mcp-server 侧 _review_timeout 上限正好 3600) +REVIEW_TIMEOUT = ZCODE_REVIEW_TIMEOUT + 120 # gate 等子进程的总超时: + # zcode 预算 + mimosa 扫描与进程收尾余量 120s, + # 防 zcode 报告已出却被 gate 提前掐掉 +PULLS_PER_PAGE = 100 # pulls 列表每页条数 (GitHub 上限) +PULLS_MAX_PAGES = 10 # 分页兜底: 单仓 open PR 超过 1000 个极罕见 +VERDICT_SCAN_HEAD = 3000 # verdict 只扫报告前 N 字符 (汇总段在开头, 防正文干扰) +MIN_COMMENT_BODY = 2000 # comment.max_body 下限 (小于模板开销的极端配置钳到这) + +# 严重度分布解析: 在报告开头找 "P0 ... <数字>" 三段。确定性 fail-safe: +# 宁可解析失败按 concerns, 也不猜。P 必须大写 (不用 IGNORECASE), +# 防正文里 "p0" 之类的偶然命中。 +# [^\dP] 排除 P 是关键: "P0/P1/P2 各 0/0/2" 枚举格式里 P0 后面紧跟 +# "/P1", 若不排除 P 会把 P1 的数字算到 P0 头上 — 枚举格式统一解析失败 +# 回 None (评论标注"解析失败请人工核对"), 也不拿错数字。 +_RE_P0 = re.compile(r"P0[^\dP]{0,12}(\d+)(?!\d)") +_RE_P1 = re.compile(r"P1[^\dP]{0,12}(\d+)(?!\d)") +_RE_P2 = re.compile(r"P2[^\dP]{0,12}(\d+)(?!\d)") + +_TRUNC_MARK = "\n\n[... 报告超长已截断 ...]" + +_LOG_LEVELS = {"DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40} +_LOG_LEVEL = "INFO" + +_SHUTDOWN = False + + +def log(msg, level="INFO"): + """日志走 stderr, 带时间戳前缀; 低于 --log-level 阈值的不打。""" + if _LOG_LEVELS.get(level, 20) < _LOG_LEVELS.get(_LOG_LEVEL, 20): + return + ts = time.strftime("%Y-%m-%d %H:%M:%S") + print(f"[{ts}] {msg}", file=sys.stderr, flush=True) + + +# ============================================================ +# 错误分类 (决定走退避还是跳仓) +# ============================================================ +class RateLimited(Exception): + """GitHub API 限流 (403 + X-RateLimit-Remaining: 0)。 + + reset_at 为 unix 秒; 本轮跳过该仓, 下个 poll 周期自然重试 + (不烧 PR 重试次数 — 限流是仓级事件, 不是单个 PR 的错)。 + """ + + def __init__(self, reset_at): + super().__init__(f"GitHub API rate limited (reset at {reset_at})") + self.reset_at = reset_at + + +class RetryableError(Exception): + """可重试错误 (5xx / 网络异常等): 进该 PR 的指数退避。""" + + +class RepoHardError(Exception): + """仓库级硬错误 (404 不存在/无权限): log + 跳过该仓, 不进 PR 退避。""" + + +class GitError(Exception): + """git 命令失败 (clone/fetch 等)。 + + transient=True 表示瞬时故障 (git 不可执行 / 命令超时等): 调用方不应 + 据此做破坏性恢复 (如删 clone 重建), 上抛当本轮仓级失败即可; + transient=False (rc≠0) 才是真损坏。 + """ + + def __init__(self, msg, transient=False): + super().__init__(msg) + self.transient = transient + + +# ============================================================ +# 配置 +# ============================================================ +def _as_int(value, default): + try: + return int(value) + except (TypeError, ValueError): + return default + + +class GateConfig: + """运行配置: 全部键可选有默认值; env 覆盖见 apply_env_overrides。 + + 对应配置文件 (JSON): + repos / poll_interval_seconds / state_file / clone_root / mcp_server / + github_api / review.{depth,focus} / retry.{base_seconds,max_seconds, + max_attempts} / comment.{enabled,max_body} + """ + + def __init__(self, data): + data = data or {} + self.repos = list(data.get("repos") or []) + # 钳下限: 过小值会打爆 GitHub 限流/把自己逼疯, 宁可启动即纠偏 + self.poll_interval = max(_as_int(data.get("poll_interval_seconds"), + 300), 30) + self.state_file = os.path.expanduser( + data.get("state_file") or DEFAULT_STATE_FILE) + self.clone_root = os.path.expanduser( + data.get("clone_root") or DEFAULT_CLONE_ROOT) + self.mcp_server = data.get("mcp_server") or DEFAULT_MCP_SERVER + self.github_api = (data.get("github_api") + or DEFAULT_GITHUB_API).rstrip("/") + + review = data.get("review") or {} + self.review_depth = review.get("depth") or "deep" + if self.review_depth not in ("normal", "deep"): + # mcp-server 侧只认 normal|deep, 非法值会在审查时报错烧重试次数, + # 不如启动时就回退默认值并告警 + log(f"非法 review.depth {self.review_depth!r}, 回退 deep", "WARNING") + self.review_depth = "deep" + self.review_focus = review.get("focus") or "" + + retry = data.get("retry") or {} + self.retry_base = max(_as_int(retry.get("base_seconds"), 300), 1) + self.retry_max = max(_as_int(retry.get("max_seconds"), 3600), + self.retry_base) + self.retry_max_attempts = max(_as_int(retry.get("max_attempts"), 5), 1) + + comment = data.get("comment") or {} + self.comment_enabled = bool(comment.get("enabled", True)) + self.comment_max_body = _as_int(comment.get("max_body"), 60000) + + def apply_env_overrides(self): + """env 覆盖 (优先级高于配置文件, 方便 systemd unit 里调 Environment=)。""" + env = os.environ + if env.get("GATE_STATE_FILE"): + self.state_file = os.path.expanduser(env["GATE_STATE_FILE"]) + if env.get("GATE_CLONE_ROOT"): + self.clone_root = os.path.expanduser(env["GATE_CLONE_ROOT"]) + if env.get("GATE_MCP_SERVER"): + self.mcp_server = env["GATE_MCP_SERVER"] + if env.get("GATE_POLL_INTERVAL"): + self.poll_interval = _as_int(env["GATE_POLL_INTERVAL"], + self.poll_interval) + + +def load_config(path): + """读 JSON 配置: 文件不存在 → 全默认; 存在但解析失败 → 致命错误退出。 + + 解析失败不退回默认是有意的: 用户明明写了配置却被静默忽略最难排查 + (接着报"未配置仓库"会让人找错方向), 直接报真正的错。 + """ + expanded = os.path.expanduser(path) + if not os.path.isfile(expanded): + log(f"配置文件 {expanded} 不存在, 使用默认配置") + return GateConfig({}) + try: + with open(expanded) as f: + data = json.load(f) + except (json.JSONDecodeError, OSError) as e: + raise SystemExit(f"配置文件 {expanded} 解析失败: {e}") from e + if not isinstance(data, dict): + raise SystemExit(f"配置文件 {expanded} 根节点必须是 JSON 对象") + log(f"已加载配置 {expanded}") + cfg = GateConfig(data) + cfg.apply_env_overrides() + return cfg + + +# ============================================================ +# GitHub token (级联解析, 不落盘) +# ============================================================ +def resolve_token(): + """GitHub token 级联: GITHUB_TOKEN → GH_TOKEN → `gh auth token`。 + + 无 token 不是致命错误: 公开仓的 pulls 列表与 git clone/fetch 可匿名; + 但发评论必须要 token — 评论时无 token 算失败, 进 PR 退避。 + """ + for var in ("GITHUB_TOKEN", "GH_TOKEN"): + token = (os.environ.get(var) or "").strip() + if token: + return token + try: + p = subprocess.run(["gh", "auth", "token"], + capture_output=True, text=True, timeout=10) + if p.returncode == 0 and p.stdout.strip(): + return p.stdout.strip() + except (OSError, subprocess.SubprocessError) as e: + log(f"gh auth token 不可用 ({e}); 未装 gh 或不在 PATH 属正常", "DEBUG") + return None + + +# ============================================================ +# GitHub REST API (urllib) +# ============================================================ +def github_request(token, api_base, method, path, body=None): + """一次 GitHub REST 调用, 返回解析后的 JSON; 错误按语义分类抛异常。""" + headers = { + "User-Agent": "zcode-review-gate", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + if token: + headers["Authorization"] = f"Bearer {token}" + data = None + if body is not None: + data = json.dumps(body, ensure_ascii=False).encode("utf-8") + headers["Content-Type"] = "application/json" + req = urllib.request.Request(api_base + path, data=data, + headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp: + raw = resp.read().decode("utf-8") + return json.loads(raw) if raw.strip() else None + except urllib.error.HTTPError as e: + hdrs = e.headers or {} + # 限流识别 (403 或 429): remaining=0 / Retry-After 头 / 裸 429 都算; + # 等待时间优先 Retry-After (相对秒), 否则 X-RateLimit-Reset (unix 秒) + if e.code in (403, 429): + retry_after = (hdrs.get("Retry-After") or "").strip() + if (e.code == 429 or retry_after + or hdrs.get("X-RateLimit-Remaining") == "0"): + if retry_after.isdigit(): + reset = int(time.time()) + int(retry_after) + else: + raw = (hdrs.get("X-RateLimit-Reset") or "0").strip() + reset = int(raw) if raw.isdigit() else 0 + raise RateLimited(reset) from e + if e.code == 404: + raise RepoHardError(f"404 Not Found: {api_base}{path}") from e + if e.code >= 500: + raise RetryableError(f"HTTP {e.code}: {api_base}{path}") from e + # 其他 4xx (401/422/非限流 403): 重试多半也失败, 但会在 + # max_attempts 后 gave_up, 方向 fail-safe + raise RetryableError(f"HTTP {e.code} (未分类 4xx): {api_base}{path}") from e + except (urllib.error.URLError, OSError, json.JSONDecodeError) as e: + raise RetryableError(f"网络/解析错误: {e}") from e + + +def list_open_prs(token, api_base, owner, repo): + """分页拉取 open PR (per_page=100, 最多 PULLS_MAX_PAGES 页兜底)。""" + prs = [] + for page in range(1, PULLS_MAX_PAGES + 1): + batch = github_request( + token, api_base, "GET", + f"/repos/{owner}/{repo}/pulls?state=open" + f"&per_page={PULLS_PER_PAGE}&page={page}") + if not batch: + break + prs.extend(batch) + if len(batch) < PULLS_PER_PAGE: + break + return prs + + +def post_comment(token, api_base, owner, repo, pr_number, body): + """在 PR 下发评论 (issues comments 端点; PR 本质上是带代码的 issue)。""" + return github_request( + token, api_base, "POST", + f"/repos/{owner}/{repo}/issues/{pr_number}/comments", + body={"body": body}) + + +# ============================================================ +# git 操作 (subprocess; token 只经进程内 header 注入) +# ============================================================ +def _basic_auth_header(token): + """拼 git smart-HTTP 用的 Authorization Basic 头值。 + + GC-8G 实测: GitHub 的 git smart-HTTP 拒绝 OAuth token (gho_ 前缀) 的 + Bearer 形式 (remote: invalid credentials), 必须 Basic — 用户名 + x-access-token (GitHub 官方约定的占位用户名), 密码为 token。 + REST API 侧 Bearer 没问题, 不受影响 (见 github_request)。 + """ + raw = base64.b64encode(f"x-access-token:{token}".encode()) + return f"Authorization: Basic {raw.decode('ascii')}" + + +def git(args, token=None, timeout=GIT_TIMEOUT): + """执行一条 git 命令, 失败抛 GitError, 成功返回 stdout。 + + token 经 git≥2.31 的 GIT_CONFIG_COUNT/KEY/VALUE 环境变量注入 + http.extraHeader (逐命令, 不写 .git/config): env 只对本用户可见, + 严格优于 argv (-c 会把 token 摆进 ps)。错误消息只回显业务 args 与 + git 的 stderr, 绝不拼接 env, 保证 token 不进日志/异常。 + """ + cmd = ["git"] + list(args) + env = None + if token: + env = dict(os.environ) + env["GIT_CONFIG_COUNT"] = "1" + env["GIT_CONFIG_KEY_0"] = "http.extraHeader" + env["GIT_CONFIG_VALUE_0"] = _basic_auth_header(token) + try: + p = subprocess.run(cmd, capture_output=True, text=True, + timeout=timeout, env=env) + except subprocess.TimeoutExpired as e: + raise GitError(f"git {args[0]} 超时 ({timeout}s)", transient=True) from e + except OSError as e: + raise GitError(f"git 不可执行: {e}", transient=True) from e + if p.returncode != 0: + detail = (p.stderr or "").strip()[:500] + raise GitError(f"git {args[0]} 失败 (exit={p.returncode}): {detail}") + return p.stdout + + +def _git_base_url(github_api): + """从 API base 推导 git web base (clone URL 的宿主)。 + + 规则: https://api.github.com → https://github.com (公有云惯例); + https://ghe.example.com/api/v3 → https://ghe.example.com (GHE 惯例, + web 与 API 同 host); 其他形态回退 https://github.com。 + """ + api = (github_api or DEFAULT_GITHUB_API).rstrip("/") + if api == "https://api.github.com": + return "https://github.com" + if api.endswith("/api/v3"): + return api[: -len("/api/v3")] + return "https://github.com" + + +def ensure_clone(cfg, token, owner, repo): + """确保本地 clone 存在, 返回 clone 路径。 + + clone URL 不含 token (匿名 https + env 注入认证, token 不落盘); + 目录名 owner__repo ("/" 不能进目录名)。 + 自愈: 上次 clone 中断/失败留下的半成品 (有 .git 目录但 rev-parse + rc≠0) 删除重建; clone 失败也清理残留, 让下轮从干净状态重试。 + rev-parse 的瞬时故障 (git 缺失/磁盘卡顿导致的 OSError/超时) 直接 + 上抛当本轮仓级失败 — 健康 clone 不能因一次卡顿被误删。 + """ + dest = os.path.join(cfg.clone_root, f"{owner}__{repo}") + if os.path.isdir(os.path.join(dest, ".git")): + try: + git(["-C", dest, "rev-parse", "--git-dir"], token=token) + return dest + except GitError as e: + if e.transient: + raise # 上抛给 process_repo 当仓级失败, 下轮自然重试 + log(f"{dest} 是不完整 clone (rev-parse 不过), 删除重建", "WARNING") + shutil.rmtree(dest, ignore_errors=True) + os.makedirs(cfg.clone_root, exist_ok=True) + url = f"{_git_base_url(cfg.github_api)}/{owner}/{repo}.git" + log(f"clone {owner}/{repo} → {dest}") + try: + git(["clone", url, dest], token=token, timeout=CLONE_TIMEOUT) + except GitError: + shutil.rmtree(dest, ignore_errors=True) + raise + return dest + + +def fetch_repo(clone, token): + """每轮一次的整仓 fetch (--prune 清掉远端已删的分支)。""" + git(["-C", clone, "fetch", "origin", "--prune"], token=token) + + +def fetch_pr_refs(clone, token, pr_number, base_ref): + """拉单个 PR 的 head ref 与 base 分支, 保证 head sha 与 merge-base 本地可见。 + + PR head 走 refs/pull/{n}/head (fork 来的 PR 也能拿到), 落到 + refs/remotes/pr/{n}; base 落到 refs/remotes/origin/{base_ref}, + 供审查参数里的 `origin/` 引用。 + """ + git(["-C", clone, "fetch", "origin", + f"+refs/pull/{pr_number}/head:refs/remotes/pr/{pr_number}"], + token=token) + git(["-C", clone, "fetch", "origin", + f"+refs/heads/{base_ref}:refs/remotes/origin/{base_ref}"], + token=token) + + +# ============================================================ +# 审查调用 (经 mcp-server --call 子进程) +# ============================================================ +def _result_text(data): + """从 --call 输出取 result.content[0].text (MCP result 的标准形状)。""" + result = (data or {}).get("result") or {} + content = result.get("content") or [] + if content and isinstance(content[0], dict): + return content[0].get("text") or "" + return "" + + +def _resolve_mcp_server(cfg): + """解析 mcp_server 配置值为可执行路径。 + + GC-8G 实测踩坑: systemd --user 默认 PATH 不含 ~/.local/bin, 而标准 + 安装法把组件装在 ~/.local/bin — 纯命令名 (无路径分隔符) 且 PATH 里 + 找不到时, 回退试 ~/.local/bin/ (存在且可执行才用); 都没有按 + 原样交给 subprocess (OSError 走既有"无法执行 mcp-server"错误路径)。 + 含路径分隔符的值原样使用。 + """ + name = cfg.mcp_server + if os.sep in name or (os.altsep and os.altsep in name): + return name + if shutil.which(name): + log(f"mcp_server 解析: {name} (PATH 命中)", "DEBUG") + return name # PATH 里能找到, 按原样 (子进程同样走 PATH) + fallback = os.path.expanduser(f"~/.local/bin/{name}") + if os.path.isfile(fallback) and os.access(fallback, os.X_OK): + log(f"mcp_server 解析: {name} → {fallback} " + f"(PATH 未命中, 回退 ~/.local/bin)", "DEBUG") + return fallback + log(f"mcp_server 解析: {name} (PATH 与 ~/.local/bin 均未见, 按原样)", + "DEBUG") + return name + + +def run_review(cfg, clone, base_ref, head_sha): + """经 mcp-server --call 调 zcode_pr_review, 返回 (ok, 报告正文 | 错误描述)。 + + 锁/限流重试/只读护栏全在 mcp-server 侧 (TOOL_HANDLERS 同源路径), 这里 + 只做进程调用与结果解包。base 必须带 origin/ 前缀 — mcp-server 的 diff + 是 base...head 三点语法 (merge-base 语义), 引用要指向远端跟踪分支。 + ZCODE_BRIDGE_REVIEW_TIMEOUT 显式透传 ZCODE_REVIEW_TIMEOUT (3600): + mcp-server 侧 zcode 审查默认 300s 超时, deep 档远远不够; gate 自身 + 等子进程的总超时 REVIEW_TIMEOUT 在此基础上再加 120s, 留给 mimosa + 扫描与进程收尾, 防 zcode 报告已出却被 gate 提前掐掉。 + """ + args = { + "path": clone, + "base": f"origin/{base_ref}", + "head": head_sha, + "depth": cfg.review_depth, + "focus": cfg.review_focus, + } + cmd = [_resolve_mcp_server(cfg), "--call", "zcode_pr_review", + json.dumps(args, ensure_ascii=False)] + env = dict(os.environ) + env["ZCODE_BRIDGE_REVIEW_TIMEOUT"] = str(ZCODE_REVIEW_TIMEOUT) + try: + p = subprocess.run(cmd, capture_output=True, text=True, + timeout=REVIEW_TIMEOUT, env=env) + except subprocess.TimeoutExpired: + return False, f"审查子进程超时 ({REVIEW_TIMEOUT}s)" + except OSError as e: + return False, f"无法执行 mcp-server ({cfg.mcp_server}): {e}" + out = (p.stdout or "").strip() + try: + data = json.loads(out) + except json.JSONDecodeError: + return False, f"mcp-server 输出非 JSON (exit={p.returncode}): {out[:300]}" + if p.returncode != 0 or not data.get("ok"): + detail = _result_text(data)[:500] + # 失败时把子进程 stderr 尾部带进错误 (超时/限流等细节都在 stderr) + stderr_tail = (p.stderr or "").strip()[-500:] + if stderr_tail: + detail = f"{detail} | stderr: {stderr_tail}" + return False, f"审查失败 (exit={p.returncode}): {detail}" + text = _result_text(data) + if not text: + return False, "mcp-server 返回空报告" + return True, text + + +# ============================================================ +# verdict 解析 (确定性, fail-safe) +# ============================================================ +def parse_severity_counts(report): + """从报告前 VERDICT_SCAN_HEAD 字符解析 (P0, P1, P2) 条数; 任一缺失 → None。""" + head = (report or "")[:VERDICT_SCAN_HEAD] + counts = [] + for pattern in (_RE_P0, _RE_P1, _RE_P2): + m = pattern.search(head) + if not m: + return None + counts.append(int(m.group(1))) + return tuple(counts) + + +def verdict_from_counts(counts): + """counts=None (解析失败) → concerns (fail-safe); P0/P1>0 → concerns; 否则 pass。""" + if counts is None: + return "concerns" + return "concerns" if (counts[0] > 0 or counts[1] > 0) else "pass" + + +def build_comment_body(verdict, counts, head_sha, report, max_body): + """拼 GitHub 评论 markdown; 超 max_body 时截断报告正文并标注。 + + max_body 默认 60000, 低于 GitHub 评论 65536 硬上限留余量; 只截报告 + 正文, 头部的 verdict 表格与尾部署名永远完整。max_body 钳下限 + MIN_COMMENT_BODY: 小于模板开销的极端配置不至于截出残破 markdown。 + """ + max_body = max(_as_int(max_body, 60000), MIN_COMMENT_BODY) + + def assemble(body_text): + icon = "✅ pass" if verdict == "pass" else "⚠️ concerns" + if counts is not None: + sev = f"P0 × {counts[0]} · P1 × {counts[1]} · P2 × {counts[2]}" + else: + sev = "解析失败（严重度分布解析失败，请人工核对）" + conclusion = ("未发现阻断问题，可以合并" if verdict == "pass" + else "存在需关注的问题，合并前请处理") + return ( + f"## ZCode Review Gate：{icon}\n\n" + "| 项 | 值 |\n" + "|---|---|\n" + f"| head | `{head_sha[:12]}` |\n" + f"| 严重度分布 | {sev} |\n" + f"| 结论 | {conclusion} |\n" + "\n" + "
完整审查报告（zcode_pr_review：git diff + " + "mimosa 深扫 + ZCode 只读复核）\n" + "\n" + f"{body_text}\n" + "\n" + "
\n" + "\n" + "---\n" + "由 [zcode-review-gate]" + "(https://github.com/tizerluo/zcode-open-bridge) 自动审查 · " + "写工具物理禁用 · 同一 head 不重复审" + ) + + body = assemble(report) + if len(body) <= max_body: + return body + budget = max(max_body - len(assemble("")) - len(_TRUNC_MARK), 0) + return assemble(report[:budget] + _TRUNC_MARK) + + +# ============================================================ +# 状态文件 (JSON, 原子写) +# ============================================================ +class StateStore: + """PR 审查状态 (JSON 文件, 原子写: 同目录 tmp + os.replace)。 + + 形状: {"prs": {"owner/repo#5": {head_sha, status, verdict, counts, + report, attempts, next_retry_at, reviewed_at, comment_url, error}}} + status ∈ pending|reviewed|failed|comment_failed|gave_up: + pending — 一轮处理中的瞬态; 崩溃残留时下轮自然恢复 + failed — 审查链路失败 (git/审查子进程), 进退避 + comment_failed — 审查已成功但评论失败; report/verdict/counts 已缓存, + 重试时同 head 只补评论, 不重跑昂贵的 zcode + reviewed — 评论已发出 (或评论被禁用); report 缓存随即清空 + gave_up — attempts 达上限; 同 head 不再动, 新 head 才复活 + report 只存评论失败期间的缓存 (截断到 comment.max_body), reviewed 后 + 置 None 防 state 膨胀。 + """ + + def __init__(self, path): + self.path = path + self.data = {"prs": {}} + self._load() + + def _load(self): + try: + with open(self.path) as f: + data = json.load(f) + except FileNotFoundError: + return + except (json.JSONDecodeError, OSError) as e: + log(f"state 文件 {self.path} 损坏, 从零开始: {e}", "WARNING") + return + if isinstance(data, dict) and isinstance(data.get("prs"), dict): + self.data = data + else: + log(f"state 文件 {self.path} 形状非法, 从零开始", "WARNING") + + def save(self): + """原子落盘 (os.replace 同目录换名, 防中途断电留半个 JSON)。""" + directory = os.path.dirname(self.path) + if directory: + os.makedirs(directory, exist_ok=True) + tmp = self.path + ".tmp" + with open(tmp, "w") as f: + json.dump(self.data, f, ensure_ascii=False, indent=1) + os.replace(tmp, self.path) + + def get(self, key): + return self.data["prs"].get(key) + + def put(self, key, entry): + self.data["prs"][key] = entry + + +# ============================================================ +# 去重 / 退避状态机 +# ============================================================ +def needs_review(entry, head_sha, now): + """去重/退避判定: 这个 PR 的当前 head 本轮要不要处理。""" + if entry is None: + return True # 首见 + if entry.get("head_sha") != head_sha: + return True # head 更新 → 重审 (复活) + status = entry.get("status") + if status in ("reviewed", "gave_up"): + return False # 同 sha 已审 / 已放弃 + if status in ("failed", "comment_failed") \ + and float(entry.get("next_retry_at") or 0) > now: + return False # 退避未到点 + return True # failed 到点重试 / comment_failed 补评论 + # / pending 断点恢复 + + +def mark_failure(entry, error, now, cfg, status="failed"): + """记录一次失败: attempts+1; 超限 gave_up, 否则指数退避。 + + 退避: base * 2**(attempts-1) 封顶 max_seconds + (默认 300 → 600 → 1200 → 2400 → gave_up@5次)。 + status="comment_failed" 用于评论失败 (审查结果已缓存在 entry, + 重试只补评论); 缓存不清, gave_up 时也保留供人工排查。 + """ + entry["attempts"] = int(entry.get("attempts") or 0) + 1 + entry["error"] = str(error)[:500] + if status == "failed": + entry["verdict"] = None # 审查链路失败: 旧 verdict 不再可信 + if entry["attempts"] >= cfg.retry_max_attempts: + entry["status"] = "gave_up" + entry["next_retry_at"] = 0.0 + log(f" 已达最大重试 {cfg.retry_max_attempts} 次, gave_up " + f"(head 更新才复活): {entry['error'][:200]}") + else: + entry["status"] = status + delay = min(cfg.retry_base * 2 ** (entry["attempts"] - 1), + cfg.retry_max) + entry["next_retry_at"] = now + delay + log(f" 失败 (第 {entry['attempts']} 次, {status}), " + f"{int(delay)}s 后重试: {entry['error'][:200]}") + + +def mark_reviewed(entry, verdict, comment_url): + """审查+评论完成: 落 verdict 与时间戳, attempts 归零。""" + entry["status"] = "reviewed" + entry["verdict"] = verdict + entry["error"] = "" + entry["attempts"] = 0 + entry["next_retry_at"] = 0.0 + entry["comment_url"] = comment_url + entry["reviewed_at"] = datetime.now().isoformat(timespec="seconds") + + +# ============================================================ +# 主流程 +# ============================================================ +def _pr_info(pr): + """从 pulls REST 响应提取处理所需字段。 + + REST 的 head sha 在 pr["head"]["sha"] (GraphQL 里叫 headRefOid), + base 分支名在 pr["base"]["ref"]。 + """ + return { + "number": pr.get("number"), + "head_sha": (pr.get("head") or {}).get("sha") or "", + "base_ref": (pr.get("base") or {}).get("ref") or "main", + "title": pr.get("title") or "", + } + + +def process_pr(cfg, state, token, owner, repo, clone, info): + """单个 PR 全链路: 取 ref → 审查 → verdict → 评论 → 落状态。 + + RateLimited (评论时遇限流) 不在此捕获, 上抛 process_repo 停本轮; + 限流不烧 attempts, 且审查结果已缓存, 下轮只补评论。 + now 在进入时现取 (一轮可能跑很久, 退避基准不许用轮初时间)。 + """ + key = f"{owner}/{repo}#{info['number']}" + head_sha = info["head_sha"] + now = time.time() + entry = state.get(key) + if entry is not None and entry.get("head_sha") != head_sha: + log(f"{key}: 新 head {head_sha[:12]} " + f"(旧 {str(entry.get('head_sha') or '')[:12]}), 重置重审") + entry = None # head 变化 → 换新 entry, attempts/缓存随之清零 + if entry is None: + entry = {"head_sha": head_sha, "status": "pending", "verdict": None, + "counts": None, "report": None, + "attempts": 0, "next_retry_at": 0.0, "reviewed_at": "", + "comment_url": "", "error": ""} + state.put(key, entry) + + if entry.get("status") == "comment_failed" and entry.get("report"): + # 审查已成功过、只是评论没发出去: 沿用缓存结果, 只补评论, + # 不为一次评论失败重跑昂贵的 zcode 审查 + report = entry["report"] + verdict = entry.get("verdict") or "concerns" + counts = tuple(entry["counts"]) if entry.get("counts") else None + log(f"{key}: 沿用缓存审查结果 (verdict={verdict}), 只补评论") + else: + try: + fetch_pr_refs(clone, token, info["number"], info["base_ref"]) + except GitError as e: + mark_failure(entry, f"git fetch 失败: {e}", now, cfg) + return + + log(f"{key}: 开始审查 head={head_sha[:12]} " + f"base=origin/{info['base_ref']} depth={cfg.review_depth}") + ok, payload = run_review(cfg, clone, info["base_ref"], head_sha) + if not ok: + mark_failure(entry, payload, now, cfg) + return + + report = payload + counts = parse_severity_counts(report) + verdict = verdict_from_counts(counts) + if counts is None: + log(f"{key}: verdict=concerns (严重度分布解析失败, fail-safe 请人工核对)") + else: + log(f"{key}: verdict={verdict} " + f"(P0={counts[0]} P1={counts[1]} P2={counts[2]})") + # 缓存审查结果: 万一评论失败, 下轮同 head 只补评论不重审 + entry["report"] = report[: cfg.comment_max_body] + entry["verdict"] = verdict + entry["counts"] = list(counts) if counts is not None else None + + comment_url = "" + if cfg.comment_enabled: + if not token: + mark_failure(entry, "发评论需要 GitHub token " + "(GITHUB_TOKEN/GH_TOKEN/gh auth token), 当前均无", + now, cfg, status="comment_failed") + return + body = build_comment_body(verdict, counts, head_sha, report, + cfg.comment_max_body) + try: + resp = post_comment(token, cfg.github_api, owner, repo, + info["number"], body) + comment_url = (resp or {}).get("html_url") or "" + log(f"{key}: 评论已发布 {comment_url or '(无返回 url)'}") + except RetryableError as e: + mark_failure(entry, f"评论发布失败: {e}", now, cfg, + status="comment_failed") + return + except RepoHardError as e: + # 评论 404: PR 在拉取列表后被关闭/删除属常态, 不烧重试次数, + # 留下次轮询的列表自然收敛 + log(f"{key}: 评论端点 404 (PR 已关闭?), 跳过: {e}", "WARNING") + return + except RateLimited: + # 限流不烧 attempts (仓级事件), 但置 comment_failed 让下轮 + # 走缓存只补评论; 上抛停本轮 + entry["status"] = "comment_failed" + entry["next_retry_at"] = 0.0 + raise + else: + log(f"{key}: 评论已禁用 (comment.enabled=false), 只落状态") + + mark_reviewed(entry, verdict, comment_url) + entry["report"] = None # 已 reviewed, 清缓存防 state 膨胀 (verdict/counts 保留) + + +def process_repo(cfg, state, token, repo_full, pr_filter): + """一个仓库的一轮: 拉 open PR → 筛出待审 → 准备 clone → 逐 PR 处理。""" + owner, sep, repo = repo_full.partition("/") + if not sep or not owner or not repo or "/" in repo: + log(f"非法仓库格式 {repo_full!r} (应为 owner/name), 跳过", "ERROR") + return + log(f"--- {owner}/{repo} ---") + try: + raw_prs = list_open_prs(token, cfg.github_api, owner, repo) + except RateLimited as e: + reset = (time.strftime("%H:%M:%S", time.localtime(e.reset_at)) + if e.reset_at else "未知") + log(f"{repo_full}: API 限流, 本轮跳过 (reset: {reset})", "WARNING") + return + except RepoHardError as e: + log(f"{repo_full}: {e} — 仓库不存在或无权限, 本轮跳过", "ERROR") + return + except RetryableError as e: + log(f"{repo_full}: 拉取 PR 列表失败, 下轮重试: {e}", "WARNING") + return + + infos = [i for i in (_pr_info(p) for p in raw_prs) + if i["number"] and i["head_sha"]] + if pr_filter: + infos = [i for i in infos if i["number"] in pr_filter] + now = time.time() + todo = [i for i in infos + if needs_review(state.get(f"{owner}/{repo}#{i['number']}"), + i["head_sha"], now)] + if not todo: + log(f"{repo_full}: {len(infos)} 个 open PR, 无待审新 head") + return + log(f"{repo_full}: {len(infos)} 个 open PR, {len(todo)} 个待审") + + try: + clone = ensure_clone(cfg, token, owner, repo) + fetch_repo(clone, token) + except GitError as e: + # 仓库级 git 失败 (网络抖动/磁盘满等) 不该烧各 PR 的重试次数: + # 本轮整体跳过, 下个 poll 周期自然重试 + log(f"{repo_full}: git 准备失败, 本轮跳过 (不消耗 PR 重试): {e}", + "ERROR") + return + + for info in todo: + key = f"{owner}/{repo}#{info['number']}" + try: + process_pr(cfg, state, token, owner, repo, clone, info) + except RateLimited as e: + log(f"{repo_full}: 发评论时遇限流, 本轮停止 " + f"(reset: {e.reset_at})", "WARNING") + return + except Exception as e: # 单 PR 意外不拖垮守护进程 (fail-safe 进退避) + log(f"{key}: 未预期异常: {e}", "ERROR") + entry = state.get(key) + if entry is not None: + mark_failure(entry, f"内部异常: {e}", time.time(), cfg) + finally: + state.save() # 每个 PR 处理完立即落盘, 崩溃不丢进度 + + +def run_once(cfg, pr_filter=None): + """跑一轮全部仓库。""" + log("===== 轮询开始 =====") + token = resolve_token() + if not token: + log("无 GitHub token: 公开仓可匿名只读, 但评论会失败进退避 " + "(设 GITHUB_TOKEN 或 gh auth login)", "WARNING") + state = StateStore(cfg.state_file) + for repo_full in cfg.repos: + process_repo(cfg, state, token, repo_full, pr_filter) + log("===== 轮询结束 =====") + + +# ============================================================ +# CLI / 守护循环 +# ============================================================ +def _on_signal(signum, _frame): + """SIGTERM/SIGINT 只置标志, 主循环在本轮边界优雅退出。""" + global _SHUTDOWN + log(f"收到信号 {signum}, 本轮结束后退出") + _SHUTDOWN = True + + +def _interruptible_sleep(seconds): + """分段 sleep, 每秒检查一次退出标志 (响应 SIGTERM 不拖延)。""" + deadline = time.time() + seconds + while not _SHUTDOWN and time.time() < deadline: + time.sleep(min(1.0, max(0.0, deadline - time.time()))) + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="zcode-review-gate", + description="PR 自动审查闸门: 轮询 GitHub open PR, 对新 head 调 " + "zcode_pr_review 审查并把 verdict 回贴为 PR 评论") + parser.add_argument( + "--config", metavar="PATH", + default=os.environ.get("GATE_CONFIG", DEFAULT_CONFIG_PATH), + help=f"配置文件路径 (默认 {DEFAULT_CONFIG_PATH}; env GATE_CONFIG 覆盖)") + parser.add_argument("--once", action="store_true", + help="只跑一轮立即退出 (cron / 调试实测用)") + parser.add_argument("--repo", action="append", dest="repos", default=[], + metavar="owner/name", + help="只监控指定仓库 (可多次; 覆盖配置文件 repos)") + parser.add_argument("--pr", action="append", dest="prs", type=int, + default=[], metavar="N", + help="只处理指定 PR 号 (可多次; 调试/实测用)") + parser.add_argument("--log-level", default="INFO", + choices=sorted(_LOG_LEVELS), + help="日志级别 (默认 INFO)") + parser.add_argument("--version", action="version", + version=f"%(prog)s {VERSION}") + args = parser.parse_args(argv) + + global _LOG_LEVEL + _LOG_LEVEL = args.log_level + + cfg = load_config(args.config) + if args.repos: + cfg.repos = args.repos # CLI 显式指定优先于配置文件 repos + if not cfg.repos: + print("错误: 未配置任何仓库 — 在配置文件 repos 或用 " + "--repo owner/name 指定", file=sys.stderr) + return 2 + pr_filter = set(args.prs) or None + + signal.signal(signal.SIGTERM, _on_signal) + signal.signal(signal.SIGINT, _on_signal) + + if args.once: + run_once(cfg, pr_filter) + return 0 + + log(f"进入守护循环: repos={cfg.repos} poll_interval={cfg.poll_interval}s " + f"state={cfg.state_file} clone_root={cfg.clone_root}") + while not _SHUTDOWN: + run_once(cfg, pr_filter) + _interruptible_sleep(cfg.poll_interval) + log("优雅退出") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/review-gate/zcode-review-gate.service b/packages/review-gate/zcode-review-gate.service new file mode 100644 index 0000000..d58bc90 --- /dev/null +++ b/packages/review-gate/zcode-review-gate.service @@ -0,0 +1,38 @@ +# zcode-review-gate.service — systemd --user 单元模板 +# +# 安装: +# mkdir -p ~/.config/systemd/user +# cp zcode-review-gate.service ~/.config/systemd/user/ +# systemctl --user daemon-reload +# systemctl --user enable --now zcode-review-gate +# +# 按需修改 (%h = 当前用户 home): +# - ExecStart 指向 zcode-review-gate 脚本的实际位置 +# (下面假设 cp/ln 到 ~/.local/bin/zcode-review-gate) +# - --config 指向你的 config.json +# - GitHub token 经 Environment 注入 (二选一, 也可靠 `gh auth token` 级联): +# Environment=GITHUB_TOKEN=ghp_xxx +# 更安全的方式是 `systemctl --user edit zcode-review-gate` 写 override, +# 避免 token 进 unit 文件本体。 + +[Unit] +Description=ZCode Review Gate — PR 自动审查闸门 (zcode-open-bridge) +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=%h/.local/bin/zcode-review-gate --config %h/.config/zcode-review-gate/config.json +Restart=on-failure +RestartSec=30 +# 长审查中的 stop: SIGTERM 后主循环在本轮边界退出, 给 mcp-server/zcode +# 子进程留收敛时间; 超时仍 SIGKILL — 万一撞 ReviewFileLock 残锁, +# mcp-server 侧 300s 锁超时兜底, 不会死锁。 +TimeoutStopSec=120 +# systemd --user 默认 PATH=/usr/local/bin:/usr/bin:/bin, 不含 ~/.local/bin +# (GC-8G 实测踩坑), 而组件/git/zcode 通常就装在 ~/.local/bin — 默认带上 +Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +Environment=PYTHONUNBUFFERED=1 + +[Install] +WantedBy=default.target diff --git a/tests/test_mcp_call.py b/tests/test_mcp_call.py new file mode 100644 index 0000000..6f9a49b --- /dev/null +++ b/tests/test_mcp_call.py @@ -0,0 +1,111 @@ +""" +test_mcp_call.py — mcp-server --call 一次性调用模式单测 + +对真实脚本文件起子进程验证 (跨进程, 无法 monkeypatch handler): + - 未知 tool → exit 1 + - args 非 JSON → exit 1 + - args 非 JSON 对象 → exit 1 + - handler 返回 isError (目录不存在的 zcode_pr_review) → exit 2, + stdout 是合法 JSON 且 ok=false + - get_zcode_capabilities → exit 0, ok=true + (该 handler 只调 packages/agent-help/zcode-agent-help, 零外部依赖; + 显式设 ZCODE_AGENT_HELP_BIN 指向仓库内副本, 防环境里已有同名变量 + 指向安装态旧版/缺失路径) + +运行: python3 tests/test_mcp_call.py +依赖: 仅 Python 标准库 + 仓库内两个组件 +""" + +import json +import os +import subprocess +import sys +import unittest + +REPO_ROOT = os.path.join(os.path.dirname(__file__), "..") +MCP_PATH = os.path.join(REPO_ROOT, "packages", "mcp-server", "zcode-mcp-server") +AGENT_HELP_PATH = os.path.abspath( + os.path.join(REPO_ROOT, "packages", "agent-help", "zcode-agent-help") +) + + +def _run_call(*args): + env = dict(os.environ) + env["ZCODE_AGENT_HELP_BIN"] = AGENT_HELP_PATH + return subprocess.run( + [sys.executable, MCP_PATH, "--call", *args], + capture_output=True, text=True, timeout=60, env=env, + ) + + +class TestMcpCall(unittest.TestCase): + def test_unknown_tool_exit_1(self): + p = _run_call("no_such_tool", "{}") + self.assertEqual(p.returncode, 1) + self.assertIn("未知 tool", p.stderr) + + def test_missing_args_exit_1(self): + """--call 只给 tool 名不给 json (或全缺) → 用法错误 exit 1""" + p = _run_call("get_zcode_capabilities") + self.assertEqual(p.returncode, 1) + + def test_handler_exception_exit_1(self): + """handler 抛异常 → exit 1 且 stdout 是 ok:false 的合法 JSON。 + + 真实异常路径 (非 mock): agent-help 二进制不存在时 + tool_get_zcode_capabilities 里的 subprocess.run 抛 FileNotFoundError。 + """ + env = dict(os.environ) + env["ZCODE_AGENT_HELP_BIN"] = "/nonexistent-agent-help-xyz" + p = subprocess.run( + [sys.executable, MCP_PATH, "--call", "get_zcode_capabilities", "{}"], + capture_output=True, text=True, timeout=60, env=env, + ) + self.assertEqual(p.returncode, 1) + data = json.loads(p.stdout) + self.assertFalse(data["ok"]) + self.assertIn("error", data) + + def test_bad_json_exit_1(self): + p = _run_call("zcode_pr_review", "not-json{") + self.assertEqual(p.returncode, 1) + + def test_non_dict_args_exit_1(self): + p = _run_call("zcode_pr_review", "[1, 2]") + self.assertEqual(p.returncode, 1) + + def test_tool_error_exit_2(self): + """handler 返回 isError → exit 2, stdout 合法 JSON 且 ok=false""" + p = _run_call("zcode_pr_review", + json.dumps({"path": "/nonexistent-path-zcode-gate-test"})) + self.assertEqual(p.returncode, 2, msg=f"stderr: {p.stderr[:300]}") + data = json.loads(p.stdout) # stdout 必须整体是合法 JSON + self.assertFalse(data["ok"]) + self.assertTrue(data["result"]["isError"]) + + def test_ok_exit_0(self): + """get_zcode_capabilities 只调仓库内 agent-help, 无网络/zcode 依赖""" + p = _run_call("get_zcode_capabilities", "{}") + self.assertEqual(p.returncode, 0, msg=f"stderr: {p.stderr[:500]}") + data = json.loads(p.stdout) + self.assertTrue(data["ok"]) + text = data["result"]["content"][0]["text"] + self.assertIn("zcode", text) + + def test_stdio_mode_untouched(self): + """无 --call 参数时仍是 stdio server: initialize 走 JSON-RPC 握手""" + req = {"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-11-25"}} + p = subprocess.run( + [sys.executable, MCP_PATH], + input=json.dumps(req) + "\n", + capture_output=True, text=True, timeout=30, + ) + self.assertEqual(p.returncode, 0) + resp = json.loads(p.stdout.strip().splitlines()[0]) + self.assertEqual(resp["id"], 1) + self.assertIn("protocolVersion", resp["result"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_review_gate.py b/tests/test_review_gate.py new file mode 100644 index 0000000..766f05b --- /dev/null +++ b/tests/test_review_gate.py @@ -0,0 +1,1026 @@ +""" +test_review_gate.py — zcode-review-gate 组件单测 (全 fake) + +subprocess.run 与 urllib.request.urlopen 一律 mock 掉: 不碰真实网络、 +真实 git、真实 ~/.config / ~/.local。state/clone_root 用临时目录。 + +覆盖: + - token 级联 (env 优先 / gh fallback / 全灭 None) + - pulls 分页拼装 + per_page/page 参数 + Authorization 头 + - 错误分类 (403+remaining=0 / 429 / 403+Retry-After → RateLimited; + 404 → RepoHardError; 5xx/网络/401/422 → RetryableError) + - 去重状态机 (同 sha 跳过 / 新 sha 重审 / failed 退避 / comment_failed 退避 + / gave_up 跳过 / 新 sha 复活) + - 退避数学 min(base*2**(attempts-1), max) + 超限 gave_up + comment_failed 保 verdict + - verdict 解析 (三段齐 / 缺一段 None / 枚举格式 None / 无关数字 fail-safe 钉住 + / p0>0 concerns / 全 0 pass / None concerns) + - 评论 body 超 max_body 截断 + max_body 钳下限 + - state 原子写 (os.replace 被调) + 损坏恢复 + 往返 + - 配置默认值 + env 覆盖 + 下限钳制 + - git: token 经 GIT_CONFIG_* env 注入且 argv 无 token / GitError 不带 token + - ensure_clone: 半成品重建 / clone 失败清理 / web 宿主推导 + - run_review: 坏 JSON / 空报告 / OSError / TimeoutExpired / 超时透传 / stderr 尾部 + - mcp_server 解析: PATH 命中 / ~/.local/bin 回退 / 不可执行不用 / 原样兜底 + - --once 全链路 (新 sha → 审+评+落盘; 同 sha 不重复; 新 sha 重审 attempts 清零) + - 评论失败两路径: RetryableError → comment_failed+退避+缓存, 下轮只补评论; + RateLimited → 不烧 attempts 有缓存, 下轮只补评论 + - CLI: --repo 覆盖配置 repos; --pr 过滤生效; 有 PR 失败 --once 仍 exit 0 + +运行: python3 tests/test_review_gate.py +依赖: 仅 Python 标准库 +""" + +import base64 +import email.message +import json +import os +import re +import shutil +import subprocess +import tempfile +import types +import unittest +import urllib.error +from unittest import mock + +GATE_PATH = os.path.join( + os.path.dirname(__file__), "..", "packages", "review-gate", "zcode-review-gate" +) + + +def _load_gate_module(): + """exec 加载无后缀组件 (跳过 __main__ 尾巴), 与 test_security_review 同惯例。""" + mod = types.ModuleType("zcode_review_gate") + mod.__file__ = GATE_PATH + with open(GATE_PATH) as f: + code = f.read() + code_no_main = code.split('if __name__ == "__main__":')[0] + exec(code_no_main, mod.__dict__) + return mod + + +class _CP: + """假 subprocess.CompletedProcess""" + + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class _FakeResp: + """假 urlopen 响应: 支持 with 语境 + read()""" + + def __init__(self, payload): + self._raw = json.dumps(payload).encode("utf-8") + + def read(self): + return self._raw + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + +def _make_http_error(code, headers=None): + hdrs = email.message.Message() + for k, v in (headers or {}).items(): + hdrs[k] = v + return urllib.error.HTTPError("https://api.github.com/x", code, + "err", hdrs, None) + + +class _EnvGuard(unittest.TestCase): + """保存/恢复本文件用到的环境变量 (与 test_security_review 同惯例)。""" + + ENV_KEYS = ("GITHUB_TOKEN", "GH_TOKEN", "GATE_CONFIG", "GATE_STATE_FILE", + "GATE_CLONE_ROOT", "GATE_MCP_SERVER", "GATE_POLL_INTERVAL") + + def setUp(self): + self._saved = {k: os.environ.get(k) for k in self.ENV_KEYS} + + def tearDown(self): + for k, v in self._saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + +class _GateCase(_EnvGuard): + """公共基座: 加载模块 + 临时目录。""" + + def setUp(self): + super().setUp() + self.mod = _load_gate_module() + self.tmp = tempfile.mkdtemp(prefix="zrg-test-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + super().tearDown() + + +# ============================================================ +# token 级联 +# ============================================================ +class TestTokenCascade(_GateCase): + def setUp(self): + super().setUp() + os.environ.pop("GITHUB_TOKEN", None) + os.environ.pop("GH_TOKEN", None) + + def test_github_token_priority(self): + os.environ["GITHUB_TOKEN"] = "t1" + os.environ["GH_TOKEN"] = "t2" + self.assertEqual(self.mod.resolve_token(), "t1") + + def test_gh_token_fallback(self): + os.environ["GH_TOKEN"] = "t2" + self.assertEqual(self.mod.resolve_token(), "t2") + + def test_gh_cli_fallback(self): + def fake_run(cmd, *a, **kw): + self.assertEqual(cmd[:2], ["gh", "auth"]) + return _CP(returncode=0, stdout="ghp_xyz\n", stderr="") + with mock.patch.object(self.mod.subprocess, "run", fake_run): + self.assertEqual(self.mod.resolve_token(), "ghp_xyz") + + def test_all_missing_returns_none(self): + def fake_run(cmd, *a, **kw): + raise FileNotFoundError("gh: command not found") + with mock.patch.object(self.mod.subprocess, "run", fake_run): + self.assertIsNone(self.mod.resolve_token()) + + +# ============================================================ +# GitHub API: 分页 / 错误分类 / 头 +# ============================================================ +class TestGithubApi(_GateCase): + def test_pagination_and_params(self): + urls, seen_headers = [], [] + + def fake_urlopen(req, timeout=None, **kw): + urls.append(req.full_url) + seen_headers.append(req.get_header("Authorization")) + page = int(re.search(r"[?&]page=(\d+)", req.full_url).group(1)) + payload = {1: [{"number": i} for i in range(100)], + 2: [{"number": 100}]}.get(page, []) + return _FakeResp(payload) + + with mock.patch.object(self.mod.urllib.request, "urlopen", fake_urlopen): + prs = self.mod.list_open_prs("tok", "https://api.github.com", "o", "r") + self.assertEqual(len(prs), 101) + self.assertEqual(len(urls), 2) + self.assertIn("state=open", urls[0]) + self.assertIn("per_page=100", urls[0]) + self.assertIn("page=1", urls[0]) + self.assertIn("page=2", urls[1]) + self.assertEqual(seen_headers[0], "Bearer tok") + + def _assert_raises(self, err, exc_type): + def fake_urlopen(req, timeout=None, **kw): + raise err + with mock.patch.object(self.mod.urllib.request, "urlopen", fake_urlopen): + with self.assertRaises(exc_type): + self.mod.list_open_prs("tok", "https://api.github.com", "o", "r") + + def test_rate_limited_403_remaining_0(self): + err = _make_http_error(403, {"X-RateLimit-Remaining": "0", + "X-RateLimit-Reset": "1893456000"}) + def fake_urlopen(req, timeout=None, **kw): + raise err + with mock.patch.object(self.mod.urllib.request, "urlopen", fake_urlopen): + with self.assertRaises(self.mod.RateLimited) as cm: + self.mod.list_open_prs("tok", "https://api.github.com", "o", "r") + self.assertEqual(cm.exception.reset_at, 1893456000) + + def test_rate_limited_429_bare(self): + self._assert_raises(_make_http_error(429), self.mod.RateLimited) + + def test_rate_limited_403_retry_after(self): + err = _make_http_error(403, {"Retry-After": "120"}) + def fake_urlopen(req, timeout=None, **kw): + raise err + import time as _time + before = int(_time.time()) + with mock.patch.object(self.mod.urllib.request, "urlopen", fake_urlopen): + with self.assertRaises(self.mod.RateLimited) as cm: + self.mod.list_open_prs("tok", "https://api.github.com", "o", "r") + # Retry-After 是相对秒, reset 应落在 now+120 附近 + self.assertGreaterEqual(cm.exception.reset_at, before + 119) + self.assertLessEqual(cm.exception.reset_at, before + 125) + + def test_404_is_repo_hard_error(self): + self._assert_raises(_make_http_error(404), self.mod.RepoHardError) + + def test_5xx_is_retryable(self): + self._assert_raises(_make_http_error(502), self.mod.RetryableError) + + def test_401_422_are_retryable(self): + # 非限流 4xx: 重试多半也失败, 但 gave_up 兜底, 方向 fail-safe + self._assert_raises(_make_http_error(401), self.mod.RetryableError) + self._assert_raises(_make_http_error(422), self.mod.RetryableError) + + def test_network_error_is_retryable(self): + def fake_urlopen(req, timeout=None, **kw): + raise urllib.error.URLError("connection refused") + with mock.patch.object(self.mod.urllib.request, "urlopen", fake_urlopen): + with self.assertRaises(self.mod.RetryableError): + self.mod.list_open_prs("tok", "https://api.github.com", "o", "r") + + +# ============================================================ +# 去重状态机 +# ============================================================ +class TestNeedsReview(_GateCase): + NOW = 1_000_000.0 + + def test_first_seen(self): + self.assertTrue(self.mod.needs_review(None, "sha1", self.NOW)) + + def test_same_sha_reviewed_skips(self): + e = {"head_sha": "sha1", "status": "reviewed"} + self.assertFalse(self.mod.needs_review(e, "sha1", self.NOW)) + + def test_new_sha_rereviews(self): + e = {"head_sha": "sha1", "status": "reviewed"} + self.assertTrue(self.mod.needs_review(e, "sha2", self.NOW)) + + def test_failed_backoff_not_due_skips(self): + e = {"head_sha": "sha1", "status": "failed", + "next_retry_at": self.NOW + 100} + self.assertFalse(self.mod.needs_review(e, "sha1", self.NOW)) + + def test_failed_backoff_due_reviews(self): + e = {"head_sha": "sha1", "status": "failed", + "next_retry_at": self.NOW - 1} + self.assertTrue(self.mod.needs_review(e, "sha1", self.NOW)) + + def test_comment_failed_backoff_not_due_skips(self): + e = {"head_sha": "sha1", "status": "comment_failed", + "next_retry_at": self.NOW + 100} + self.assertFalse(self.mod.needs_review(e, "sha1", self.NOW)) + + def test_comment_failed_due_retries(self): + e = {"head_sha": "sha1", "status": "comment_failed", + "next_retry_at": 0.0} + self.assertTrue(self.mod.needs_review(e, "sha1", self.NOW)) + + def test_gave_up_skips_same_sha(self): + e = {"head_sha": "sha1", "status": "gave_up"} + self.assertFalse(self.mod.needs_review(e, "sha1", self.NOW)) + + def test_gave_up_revives_on_new_sha(self): + e = {"head_sha": "sha1", "status": "gave_up", "attempts": 5} + self.assertTrue(self.mod.needs_review(e, "sha2", self.NOW)) + + +# ============================================================ +# 退避数学 +# ============================================================ +class TestBackoff(_GateCase): + def _cfg(self, **retry): + base = {"base_seconds": 300, "max_seconds": 3600, "max_attempts": 5} + base.update(retry) + return self.mod.GateConfig({"retry": base}) + + def test_backoff_sequence(self): + cfg = self._cfg() + now = 1_000_000.0 + entry = {"attempts": 0} + for i, delay in enumerate([300, 600, 1200, 2400], start=1): + self.mod.mark_failure(entry, "err", now, cfg) + self.assertEqual(entry["attempts"], i) + self.assertEqual(entry["status"], "failed") + self.assertEqual(entry["next_retry_at"] - now, delay) + + def test_gave_up_at_max_attempts(self): + cfg = self._cfg() + entry = {"attempts": 4} + self.mod.mark_failure(entry, "err", 0.0, cfg) + self.assertEqual(entry["status"], "gave_up") + self.assertEqual(entry["attempts"], 5) + self.assertEqual(entry["next_retry_at"], 0.0) + + def test_backoff_capped_at_max(self): + cfg = self._cfg(max_seconds=500, max_attempts=9) + entry = {"attempts": 2} + self.mod.mark_failure(entry, "err", 0.0, cfg) + # 300 * 2**2 = 1200 → 封顶 500 + self.assertEqual(entry["next_retry_at"], 500.0) + + def test_comment_failed_status_and_verdict_kept(self): + # 评论失败: status=comment_failed, 缓存的 verdict 不被清 (重试要用) + cfg = self._cfg() + entry = {"attempts": 0, "verdict": "pass", "report": "r"} + self.mod.mark_failure(entry, "err", 0.0, cfg, status="comment_failed") + self.assertEqual(entry["status"], "comment_failed") + self.assertEqual(entry["verdict"], "pass") + # 审查链路失败则清 verdict (旧结论不再可信) + entry2 = {"attempts": 0, "verdict": "pass"} + self.mod.mark_failure(entry2, "err", 0.0, cfg) + self.assertIsNone(entry2["verdict"]) + + +# ============================================================ +# verdict 解析 +# ============================================================ +class TestVerdict(_GateCase): + def test_parse_all_three(self): + text = "汇总: P0: 1 条, P1: 2 条, P2: 3 条\n详情..." + self.assertEqual(self.mod.parse_severity_counts(text), (1, 2, 3)) + + def test_parse_cn_format(self): + text = "P0 × 0 · P1 × 0 · P2 × 12" + self.assertEqual(self.mod.parse_severity_counts(text), (0, 0, 12)) + + def test_parse_missing_one_returns_none(self): + text = "P0: 0 条, P1: 1 条" # 缺 P2 + self.assertIsNone(self.mod.parse_severity_counts(text)) + + def test_parse_enumeration_format_returns_none(self): + # "P0/P1/P2 各 0/0/2" 枚举格式: P0 后紧跟 "/P1", 不排除 P 会把 + # P1 的数字算到 P0 头上 — 统一解析失败回 None (走人工核对标注), + # 也不拿错数字 + text = "汇总: P0/P1/P2 各 0/0/2 条, findings 确认 1 条" + self.assertIsNone(self.mod.parse_severity_counts(text)) + # 对应的 verdict 走向: None → concerns (fail-safe) + self.assertEqual(self.mod.verdict_from_counts(None), "concerns") + + def test_parse_unrelated_number_fail_safe(self): + # "P0 级问题参见 2024 年报": 无关数字仍会被当作计数 (2024), + # 钉住该行为 — 这是有意取舍: 解析到异常大数 → concerns, + # 方向 fail-safe (宁误拦, 不漏放), 人工看评论即可分辨 + text = "P0 级问题参见 2024 年报; P1: 0 条; P2: 1 条" + self.assertEqual(self.mod.parse_severity_counts(text), (2024, 0, 1)) + self.assertEqual( + self.mod.verdict_from_counts( + self.mod.parse_severity_counts(text)), "concerns") + + def test_parse_only_scans_head(self): + # 分布信息在 3000 字符以外 → 视为缺失 (防正文偶然命中) + text = "没有分布" + "x" * 4000 + "P0: 9 P1: 9 P2: 9" + self.assertIsNone(self.mod.parse_severity_counts(text)) + + def test_verdict_p0_concerns(self): + self.assertEqual(self.mod.verdict_from_counts((1, 0, 0)), "concerns") + + def test_verdict_p1_concerns(self): + self.assertEqual(self.mod.verdict_from_counts((0, 2, 0)), "concerns") + + def test_verdict_all_zero_pass(self): + self.assertEqual(self.mod.verdict_from_counts((0, 0, 5)), "pass") + + def test_verdict_none_concerns(self): + self.assertEqual(self.mod.verdict_from_counts(None), "concerns") + + +# ============================================================ +# 评论 body +# ============================================================ +class TestCommentBody(_GateCase): + SHA = "abcdef1234567890" + "0" * 24 + + def test_pass_body(self): + body = self.mod.build_comment_body( + "pass", (0, 0, 1), self.SHA, "报告正文", 60000) + self.assertIn("✅ pass", body) + self.assertIn("`abcdef123456`", body) + self.assertIn("P0 × 0 · P1 × 0 · P2 × 1", body) + self.assertIn("未发现阻断问题，可以合并", body) + self.assertIn("报告正文", body) + self.assertIn("zcode_pr_review", body) + self.assertNotIn("截断", body) + + def test_concerns_parse_failure_body(self): + body = self.mod.build_comment_body( + "concerns", None, self.SHA, "r", 60000) + self.assertIn("⚠️ concerns", body) + self.assertIn("解析失败", body) + self.assertIn("合并前请处理", body) + + def test_truncation_over_max_body(self): + report = "报" * 100000 + body = self.mod.build_comment_body( + "concerns", (1, 0, 0), self.SHA, report, 60000) + self.assertLessEqual(len(body), 60000) + self.assertIn("报告超长已截断", body) + # 头尾结构完整 (只截报告正文) + self.assertIn("## ZCode Review Gate", body) + self.assertIn("同一 head 不重复审", body) + + def test_max_body_clamped_to_floor(self): + # 极端配置 max_body=10 (< 模板开销) → 钳到 2000, 不至于截出残破 markdown + body = self.mod.build_comment_body( + "pass", (0, 0, 0), self.SHA, "短报告", 10) + self.assertIn("## ZCode Review Gate", body) + self.assertIn("短报告", body) + self.assertNotIn("截断", body) + long_body = self.mod.build_comment_body( + "pass", (0, 0, 0), self.SHA, "报" * 5000, 10) + self.assertLessEqual(len(long_body), 2000) + self.assertIn("报告超长已截断", long_body) + + +# ============================================================ +# state 文件 +# ============================================================ +class TestStateStore(_GateCase): + def test_atomic_save_uses_os_replace(self): + path = os.path.join(self.tmp, "sub", "state.json") + store = self.mod.StateStore(path) + store.put("o/r#1", {"head_sha": "x", "status": "reviewed"}) + with mock.patch.object(self.mod.os, "replace", + wraps=os.replace) as m_replace: + store.save() + m_replace.assert_called_once() + with open(path) as f: + data = json.load(f) + self.assertEqual(data["prs"]["o/r#1"]["head_sha"], "x") + self.assertFalse(os.path.exists(path + ".tmp")) # tmp 不残留 + + def test_corrupt_state_starts_fresh(self): + path = os.path.join(self.tmp, "state.json") + with open(path, "w") as f: + f.write("{not json") + store = self.mod.StateStore(path) + self.assertEqual(store.data, {"prs": {}}) + + def test_roundtrip(self): + path = os.path.join(self.tmp, "state.json") + s1 = self.mod.StateStore(path) + s1.put("o/r#2", {"head_sha": "y"}) + s1.save() + s2 = self.mod.StateStore(path) + self.assertEqual(s2.get("o/r#2")["head_sha"], "y") + + +# ============================================================ +# 配置 +# ============================================================ +class TestConfig(_GateCase): + def test_defaults(self): + cfg = self.mod.GateConfig({}) + self.assertEqual(cfg.poll_interval, 300) + self.assertEqual(cfg.review_depth, "deep") + self.assertEqual(cfg.retry_base, 300) + self.assertEqual(cfg.retry_max_attempts, 5) + self.assertTrue(cfg.comment_enabled) + self.assertEqual(cfg.comment_max_body, 60000) + self.assertTrue( + cfg.state_file.endswith(".local/state/zcode-review-gate/state.json")) + self.assertEqual(cfg.mcp_server, "zcode-mcp-server") + + def test_env_overrides(self): + os.environ["GATE_STATE_FILE"] = "/tmp/x/state.json" + os.environ["GATE_POLL_INTERVAL"] = "42" + os.environ["GATE_MCP_SERVER"] = "/opt/mcp" + cfg = self.mod.GateConfig({}) + cfg.apply_env_overrides() + self.assertEqual(cfg.state_file, "/tmp/x/state.json") + self.assertEqual(cfg.poll_interval, 42) + self.assertEqual(cfg.mcp_server, "/opt/mcp") + + def test_bad_depth_falls_back(self): + cfg = self.mod.GateConfig({"review": {"depth": "ultra"}}) + self.assertEqual(cfg.review_depth, "deep") + + def test_lower_bound_clamps(self): + cfg = self.mod.GateConfig({ + "poll_interval_seconds": 5, # → 30 + "retry": {"base_seconds": 0, # → 1 + "max_seconds": 0, # → 钳到 ≥ base + "max_attempts": 0}}) # → 1 + self.assertEqual(cfg.poll_interval, 30) + self.assertEqual(cfg.retry_base, 1) + self.assertGreaterEqual(cfg.retry_max, cfg.retry_base) + self.assertEqual(cfg.retry_max_attempts, 1) + + +# ============================================================ +# git token 注入 (env 方式) / GitError 不含 token +# ============================================================ +class TestGitTokenHeader(_GateCase): + def _capture(self, token): + seen = {} + + def fake_run(cmd, *a, **kw): + seen["cmd"] = list(cmd) + seen["env"] = kw.get("env") + return _CP(returncode=0, stdout="ok", stderr="") + + with mock.patch.object(self.mod.subprocess, "run", fake_run): + self.mod.git(["fetch", "origin"], token=token) + return seen + + def test_token_injected_via_env_not_argv(self): + seen = self._capture("secret-tok") + # argv 里绝不允许出现 token (ps 对本机所有用户可见) + self.assertNotIn("secret-tok", " ".join(seen["cmd"])) + # env 注入 GIT_CONFIG_* 三件套 (git≥2.31) + env = seen["env"] + self.assertEqual(env["GIT_CONFIG_COUNT"], "1") + self.assertEqual(env["GIT_CONFIG_KEY_0"], "http.extraHeader") + self.assertTrue( + env["GIT_CONFIG_VALUE_0"].startswith("Authorization: Basic ")) + + def test_basic_header_decodes_to_x_access_token(self): + # GC-8G 实测: GitHub git smart-HTTP 拒绝 OAuth token 的 Bearer 形式, + # 必须 Basic (x-access-token: 的 base64); REST 侧 Bearer 不受影响 + seen = self._capture("secret-tok") + value = seen["env"]["GIT_CONFIG_VALUE_0"] + scheme, b64 = value.split(" ", 2)[1:] + self.assertEqual(scheme, "Basic") + decoded = base64.b64decode(b64).decode("utf-8") + self.assertEqual(decoded, "x-access-token:secret-tok") + + def test_no_token_no_env_override(self): + seen = self._capture(None) + self.assertIsNone(seen["env"]) # 不传 env → 子进程原样继承 + + def test_git_error_message_excludes_token(self): + # 红线: GitError 消息/异常 repr 不能带 env 内容 + def fake_run(cmd, *a, **kw): + return _CP(returncode=128, stdout="", stderr="fatal: auth failed") + with mock.patch.object(self.mod.subprocess, "run", fake_run): + with self.assertRaises(self.mod.GitError) as cm: + self.mod.git(["fetch"], token="secret-tok") + self.assertNotIn("secret-tok", str(cm.exception)) + self.assertNotIn("secret-tok", repr(cm.exception)) + + +# ============================================================ +# ensure_clone: 自愈 / 清理 / web 宿主推导 +# ============================================================ +class TestEnsureClone(_GateCase): + def _cfg(self, api="https://api.github.com"): + return self.mod.GateConfig({ + "clone_root": os.path.join(self.tmp, "clones"), + "github_api": api}) + + def test_incomplete_clone_rebuilt(self): + """半成品 clone (有 .git 目录但 rev-parse 不过) → 删除重建""" + cfg = self._cfg() + dest = os.path.join(self.tmp, "clones", "o__r") + os.makedirs(os.path.join(dest, ".git")) + calls = [] + + def fake_run(cmd, *a, **kw): + calls.append(list(cmd)) + if "rev-parse" in cmd: + return _CP(returncode=128, stdout="", stderr="not a git repo") + if "clone" in cmd: + os.makedirs(os.path.join(cmd[-1], ".git"), exist_ok=True) + return _CP(returncode=0, stdout="", stderr="") + return _CP(returncode=0, stdout="", stderr="") + + with mock.patch.object(self.mod.subprocess, "run", fake_run): + out = self.mod.ensure_clone(cfg, "tok", "o", "r") + self.assertEqual(out, dest) + self.assertTrue(any("clone" in c for c in calls)) # 触发了重建 + + def test_clone_failure_cleans_up(self): + """clone 失败 (留了半个目录) → GitError 且残留被清理""" + cfg = self._cfg() + dest = os.path.join(self.tmp, "clones", "o__r") + + def fake_run(cmd, *a, **kw): + if "clone" in cmd: + os.makedirs(os.path.join(cmd[-1], ".git"), exist_ok=True) + return _CP(returncode=128, stdout="", stderr="boom") + return _CP(returncode=0, stdout="", stderr="") + + with mock.patch.object(self.mod.subprocess, "run", fake_run): + with self.assertRaises(self.mod.GitError): + self.mod.ensure_clone(cfg, "tok", "o", "r") + self.assertFalse(os.path.exists(dest)) + + def test_revparse_transient_keeps_clone(self): + """rev-parse 超时/OSError (瞬时故障) → GitError 上抛, 健康 clone 不被误删""" + cfg = self._cfg() + dest = os.path.join(self.tmp, "clones", "o__r") + os.makedirs(os.path.join(dest, ".git")) + + def fake_run(cmd, *a, **kw): + if "rev-parse" in cmd: + raise subprocess.TimeoutExpired(cmd=list(cmd), timeout=120) + if "clone" in cmd: # 不应走到: 瞬时故障不触发重建 + raise AssertionError("瞬时故障不应删除重建") + return _CP(returncode=0, stdout="", stderr="") + + with mock.patch.object(self.mod.subprocess, "run", fake_run): + with self.assertRaises(self.mod.GitError) as cm: + self.mod.ensure_clone(cfg, "tok", "o", "r") + self.assertTrue(cm.exception.transient) + self.assertTrue(os.path.isdir(os.path.join(dest, ".git"))) # 未被删 + + def test_git_base_url_derivation(self): + self.assertEqual(self.mod._git_base_url("https://api.github.com"), + "https://github.com") + self.assertEqual(self.mod._git_base_url("https://ghe.example.com/api/v3"), + "https://ghe.example.com") + self.assertEqual(self.mod._git_base_url("https://weird.example.com"), + "https://github.com") + + def test_clone_uses_derived_url(self): + cfg = self._cfg("https://ghe.example.com/api/v3") + cmds = [] + + def fake_run(cmd, *a, **kw): + cmds.append(list(cmd)) + if "clone" in cmd: + os.makedirs(os.path.join(cmd[-1], ".git"), exist_ok=True) + return _CP(returncode=0, stdout="", stderr="") + + with mock.patch.object(self.mod.subprocess, "run", fake_run): + self.mod.ensure_clone(cfg, "tok", "o", "r") + clone_cmd = next(c for c in cmds if "clone" in c) + self.assertIn("https://ghe.example.com/o/r.git", clone_cmd) + + +# ============================================================ +# mcp_server 解析 (PATH → ~/.local/bin 回退 → 原样) +# ============================================================ +class TestResolveMcpServer(_GateCase): + """GC-8G 实测: systemd --user 默认 PATH 不含 ~/.local/bin, + 纯命令名需回退 ~/.local/bin 找组件。""" + + ENV_KEYS = _EnvGuard.ENV_KEYS + ("HOME",) + + def _cfg(self, name="zcode-mcp-server"): + return self.mod.GateConfig({"mcp_server": name}) + + def _make_local_exe(self, mode=0o755): + os.environ["HOME"] = self.tmp + local_bin = os.path.join(self.tmp, ".local", "bin") + os.makedirs(local_bin) + exe = os.path.join(local_bin, "zcode-mcp-server") + with open(exe, "w") as f: + f.write("#!/bin/sh\n") + os.chmod(exe, mode) + return exe + + def test_which_hit_returns_name(self): + with mock.patch.object(self.mod.shutil, "which", + return_value="/usr/bin/zcode-mcp-server"): + self.assertEqual(self.mod._resolve_mcp_server(self._cfg()), + "zcode-mcp-server") + + def test_fallback_to_local_bin(self): + """which 返回 None + ~/.local/bin 存在可执行文件 → 选用回退路径""" + exe = self._make_local_exe() + with mock.patch.object(self.mod.shutil, "which", return_value=None): + self.assertEqual(self.mod._resolve_mcp_server(self._cfg()), exe) + + def test_fallback_requires_executable(self): + """~/.local/bin 里文件存在但不可执行 → 不用, 按原样交给 subprocess""" + self._make_local_exe(mode=0o644) + with mock.patch.object(self.mod.shutil, "which", return_value=None): + self.assertEqual(self.mod._resolve_mcp_server(self._cfg()), + "zcode-mcp-server") + + def test_missing_everywhere_returns_as_is(self): + """PATH 与 ~/.local/bin 都没有 → 原样 (OSError 走既有错误路径)""" + os.environ["HOME"] = self.tmp + with mock.patch.object(self.mod.shutil, "which", return_value=None): + self.assertEqual(self.mod._resolve_mcp_server(self._cfg()), + "zcode-mcp-server") + + def test_path_value_used_as_is(self): + """含路径分隔符的值原样使用, 不做任何探测""" + with mock.patch.object(self.mod.shutil, "which") as m_which: + self.assertEqual( + self.mod._resolve_mcp_server(self._cfg("/opt/mcp/server")), + "/opt/mcp/server") + m_which.assert_not_called() + + +# ============================================================ +# run_review 异常分支 / 超时透传 / stderr 尾部 +# ============================================================ +class TestRunReview(_GateCase): + def _cfg(self): + return self.mod.GateConfig({}) + + def _run_with(self, cp=None, exc=None): + def fake_run(cmd, *a, **kw): + if exc is not None: + raise exc + return cp + with mock.patch.object(self.mod.subprocess, "run", fake_run): + return self.mod.run_review(self._cfg(), "/clone", "main", "sha") + + def test_bad_json_output(self): + ok, err = self._run_with(_CP(returncode=0, stdout="not json", stderr="")) + self.assertFalse(ok) + self.assertIn("非 JSON", err) + + def test_empty_report(self): + payload = {"ok": True, "result": {"content": []}} + ok, err = self._run_with( + _CP(returncode=0, stdout=json.dumps(payload), stderr="")) + self.assertFalse(ok) + self.assertIn("空报告", err) + + def test_mcp_server_missing_oserror(self): + ok, err = self._run_with(exc=OSError("No such file or directory")) + self.assertFalse(ok) + self.assertIn("无法执行", err) + + def test_subprocess_timeout(self): + ok, err = self._run_with( + exc=subprocess.TimeoutExpired(cmd="x", timeout=1)) + self.assertFalse(ok) + self.assertIn("超时", err) + + def test_timeout_env_passthrough(self): + """zcode 审查预算 (3600) 透传给 mcp-server; gate 总超时再多留 + 120s 给 mimosa 扫描与收尾 (REVIEW_TIMEOUT = ZCODE_REVIEW_TIMEOUT + 120)""" + seen = {} + + def fake_run(cmd, *a, **kw): + seen.update(kw) + payload = {"ok": True, "result": { + "content": [{"type": "text", "text": "P0: 0 P1: 0 P2: 0"}]}} + return _CP(returncode=0, stdout=json.dumps(payload), stderr="") + + with mock.patch.object(self.mod.subprocess, "run", fake_run): + ok, _report = self.mod.run_review(self._cfg(), "/c", "main", "s") + self.assertTrue(ok) + self.assertEqual(self.mod.REVIEW_TIMEOUT, + self.mod.ZCODE_REVIEW_TIMEOUT + 120) + self.assertEqual(seen["timeout"], self.mod.REVIEW_TIMEOUT) + self.assertEqual(seen["env"]["ZCODE_BRIDGE_REVIEW_TIMEOUT"], + str(self.mod.ZCODE_REVIEW_TIMEOUT)) + + def test_stderr_tail_in_error(self): + payload = {"ok": False, "result": { + "content": [{"type": "text", "text": "炸了"}], "isError": True}} + ok, err = self._run_with( + _CP(returncode=2, stdout=json.dumps(payload), stderr="x" * 600)) + self.assertFalse(ok) + self.assertIn("stderr:", err) + self.assertIn("x" * 100, err) # stderr 尾部在错误里 + + +# ============================================================ +# --once 全链路 (全 fake) +# ============================================================ +class TestOnceEndToEnd(_GateCase): + def setUp(self): + super().setUp() + os.environ["GITHUB_TOKEN"] = "fake-token-123" + os.environ.pop("GH_TOKEN", None) + self.pr_sha = "a" * 40 + self.pr_list = [self._mkpr(5, self.pr_sha)] + self.comment_behavior = "ok" # ok | retryable | ratelimited + self.api_calls = [] # (method, url, body) + self.mcp_calls = [] # 审查 args dict + self.git_calls = [] # (cmd list, env) + + def _mkpr(self, number, sha, base="main"): + return {"number": number, "title": f"pr {number}", + "head": {"sha": sha}, "base": {"ref": base}} + + def _fake_urlopen(self, req, timeout=None, **kw): + url = req.full_url + method = req.get_method() + body = json.loads(req.data.decode("utf-8")) if req.data else None + self.api_calls.append((method, url, body)) + if "/pulls?" in url: + return _FakeResp(self.pr_list) + if url.endswith("/comments"): + if self.comment_behavior == "retryable": + raise _make_http_error(500) + if self.comment_behavior == "ratelimited": + raise _make_http_error(403, {"X-RateLimit-Remaining": "0", + "X-RateLimit-Reset": "1893456000"}) + return _FakeResp( + {"html_url": "https://github.com/octo/hello#issuecomment-1"}) + raise AssertionError(f"未预期 URL: {url}") + + def _fake_run(self, cmd, *a, **kw): + if cmd[0] == "git": + self.git_calls.append((list(cmd), kw.get("env"))) + if "clone" in cmd: + # 假 clone 也要建出 .git, 否则每轮都重复 clone + os.makedirs(os.path.join(cmd[-1], ".git"), exist_ok=True) + return _CP(returncode=0, stdout="", stderr="") + if "--call" in cmd: + idx = cmd.index("--call") + self.assertEqual(cmd[idx + 1], "zcode_pr_review") + self.mcp_calls.append(json.loads(cmd[idx + 2])) + report = "汇总: P0: 0 条, P1: 0 条, P2: 2 条\n一切正常。" + payload = {"ok": True, "result": { + "content": [{"type": "text", "text": report}]}} + return _CP(returncode=0, stdout=json.dumps(payload), stderr="") + raise AssertionError(f"未预期命令: {cmd}") + + def _write_config(self, repos=("octo/hello",)): + cfg = {"repos": list(repos), + "state_file": os.path.join(self.tmp, "state.json"), + "clone_root": os.path.join(self.tmp, "clones"), + "review": {"depth": "deep", "focus": ""}, + "retry": {"base_seconds": 300, "max_seconds": 3600, + "max_attempts": 5}, + "comment": {"enabled": True, "max_body": 60000}} + path = os.path.join(self.tmp, "config.json") + with open(path, "w") as f: + json.dump(cfg, f) + return path + + def _run_once(self, cfg_path, extra_args=()): + with mock.patch.object(self.mod.subprocess, "run", self._fake_run), \ + mock.patch.object(self.mod.urllib.request, "urlopen", + self._fake_urlopen): + return self.mod.main(["--once", "--config", cfg_path, + *extra_args]) + + def _state(self): + with open(os.path.join(self.tmp, "state.json")) as f: + return json.load(f)["prs"] + + def _state_entry(self, key="octo/hello#5"): + return self._state()[key] + + def _force_retry_due(self, key="octo/hello#5"): + """模拟退避到点: 把 next_retry_at 拨回过去 (不等真实 300s)""" + path = os.path.join(self.tmp, "state.json") + with open(path) as f: + st = json.load(f) + st["prs"][key]["next_retry_at"] = 0 + with open(path, "w") as f: + json.dump(st, f) + + def test_full_cycle_dedup_and_new_sha(self): + cfg_path = self._write_config() + + # 第一轮: 新 sha → clone/fetch/审查/评论/state 落盘 + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1) + posts = [c for c in self.api_calls if c[0] == "POST"] + self.assertEqual(len(posts), 1) + self.assertIn("/repos/octo/hello/issues/5/comments", posts[0][1]) + + # 审查参数: base 带 origin/ 前缀, head 为 sha, depth 透传 + call_args = self.mcp_calls[0] + self.assertEqual(call_args["base"], "origin/main") + self.assertEqual(call_args["head"], "a" * 40) + self.assertEqual(call_args["depth"], "deep") + self.assertTrue(call_args["path"].endswith("octo__hello")) + + # 评论体: pass + 严重度分布 + 署名 + body = posts[0][2]["body"] + self.assertIn("✅ pass", body) + self.assertIn("P0 × 0 · P1 × 0 · P2 × 2", body) + self.assertIn("同一 head 不重复审", body) + + # git: clone URL 与 argv 均无 token; token 只走 GIT_CONFIG_* env + # (Basic header, 明文 token 本身也不在 env 值里 — 只有 base64 形态) + clone_cmd, clone_env = next( + (c, e) for c, e in self.git_calls if "clone" in c) + url_arg = next(a for a in clone_cmd if a.startswith("https://")) + self.assertNotIn("fake-token-123", url_arg) + self.assertNotIn("fake-token-123", " ".join(clone_cmd)) + self.assertEqual(clone_env["GIT_CONFIG_VALUE_0"], + self.mod._basic_auth_header("fake-token-123")) + + # state 落盘: reviewed/pass, report 缓存已清 + entry = self._state_entry() + self.assertEqual(entry["status"], "reviewed") + self.assertEqual(entry["verdict"], "pass") + self.assertEqual(entry["head_sha"], "a" * 40) + self.assertEqual(entry["attempts"], 0) + self.assertTrue(entry["comment_url"]) + self.assertIsNone(entry["report"]) + + # 第二轮同 sha: 不重复审查、不重复评论 + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1) + self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 1) + + # 第三轮新 sha: 重审 + 重评论, attempts 从 0 重新计 + self.pr_sha = "b" * 40 + self.pr_list = [self._mkpr(5, self.pr_sha)] + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 2) + self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 2) + entry = self._state_entry() + self.assertEqual(entry["head_sha"], "b" * 40) + self.assertEqual(entry["attempts"], 0) + self.assertEqual(entry["status"], "reviewed") + + def test_review_failure_goes_backoff_then_skip(self): + """审查失败 → failed + 退避; 第二轮退避未到 → 跳过不再调审查。 + 同时钉住语义: 有 PR 失败时 --once 仍 exit 0 (失败记在 state, 不传染退出码)""" + cfg_path = self._write_config() + fail_payload = {"ok": False, "result": { + "content": [{"type": "text", "text": "炸了"}], "isError": True}} + + def failing_run(cmd, *a, **kw): + if "--call" in cmd: + return _CP(returncode=2, stdout=json.dumps(fail_payload), + stderr="") + return self._fake_run(cmd, *a, **kw) + + with mock.patch.object(self.mod.subprocess, "run", failing_run), \ + mock.patch.object(self.mod.urllib.request, "urlopen", + self._fake_urlopen): + self.assertEqual(self.mod.main(["--once", "--config", cfg_path]), 0) + entry = self._state_entry() + self.assertEqual(entry["status"], "failed") + self.assertEqual(entry["attempts"], 1) + self.assertGreater(entry["next_retry_at"], 0) + # 评论不应发出 (审查就失败了) + self.assertEqual( + len([c for c in self.api_calls if c[0] == "POST"]), 0) + + # 第二轮: 退避未到, 不再调审查, 退出码仍 0 + self.assertEqual(self.mod.main(["--once", "--config", cfg_path]), 0) + entry = self._state_entry() + self.assertEqual(entry["attempts"], 1) # 没增加 + + def test_comment_retryable_failure_caches_and_comment_only_retry(self): + """审查成功+评论 RetryableError → comment_failed+退避+缓存; + 退避到点后的下一轮: 只补评论, 不重跑 mcp 审查""" + cfg_path = self._write_config() + self.comment_behavior = "retryable" # 评论 500 + + # 第一轮: 审查成功, 评论失败 → comment_failed + 缓存 + attempts=1 + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1) + self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 1) + entry = self._state_entry() + self.assertEqual(entry["status"], "comment_failed") + self.assertEqual(entry["attempts"], 1) + self.assertGreater(entry["next_retry_at"], 0) + self.assertTrue(entry["report"]) # 审查结果已缓存 + self.assertEqual(entry["verdict"], "pass") + self.assertEqual(entry["counts"], [0, 0, 2]) + + # 退避到点 (拨回 next_retry_at), 评论恢复 → 只补评论 + self.comment_behavior = "ok" + self._force_retry_due() + git_marker = len(self.git_calls) + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1) # 没有重跑审查 + self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 2) + # 缓存路径不拉 PR ref (fetch_pr_refs 只在真实审查前跑) + round2_git = self.git_calls[git_marker:] + self.assertFalse(any(any("refs/pull/" in a for a in c) + for c, _e in round2_git)) + entry = self._state_entry() + self.assertEqual(entry["status"], "reviewed") + self.assertIsNone(entry["report"]) # reviewed 后缓存清掉 + self.assertEqual(entry["attempts"], 0) + + def test_comment_ratelimited_no_attempts_burn_comment_only_retry(self): + """评论遇限流 → 不烧 attempts 但有缓存; 下轮只补评论""" + cfg_path = self._write_config() + self.comment_behavior = "ratelimited" + + # 第一轮: 审查成功, 评论限流 → comment_failed, attempts 不增 + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1) + entry = self._state_entry() + self.assertEqual(entry["status"], "comment_failed") + self.assertEqual(entry["attempts"], 0) # 限流不烧重试次数 + self.assertTrue(entry["report"]) + + # 第二轮 (限流 reset 语义上下轮自然重试): 只补评论 + self.comment_behavior = "ok" + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1) # 仍未重跑审查 + self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 2) + entry = self._state_entry() + self.assertEqual(entry["status"], "reviewed") + self.assertIsNone(entry["report"]) + + def test_repo_flag_overrides_config_repos(self): + """CLI --repo 覆盖配置文件 repos""" + cfg_path = self._write_config(repos=("octo/hello",)) + rc = self._run_once(cfg_path, extra_args=("--repo", "octo/other")) + self.assertEqual(rc, 0) + gets = [u for m, u, _b in self.api_calls if m == "GET"] + self.assertTrue(all("/repos/octo/other/" in u for u in gets)) + self.assertIn("octo/other#5", self._state()) + self.assertNotIn("octo/hello#5", self._state()) + + def test_pr_filter(self): + """--pr 只处理指定 PR""" + self.pr_list = [self._mkpr(5, self.pr_sha), self._mkpr(6, "c" * 40)] + cfg_path = self._write_config() + rc = self._run_once(cfg_path, extra_args=("--pr", "6")) + self.assertEqual(rc, 0) + self.assertEqual(len(self.mcp_calls), 1) + posts = [c for c in self.api_calls if c[0] == "POST"] + self.assertEqual(len(posts), 1) + self.assertIn("issues/6/comments", posts[0][1]) + self.assertIn("octo/hello#6", self._state()) + self.assertNotIn("octo/hello#5", self._state()) + + +if __name__ == "__main__": + unittest.main()