diff --git a/examples/workflows/pr-quality-gatekeeper/.gitignore b/examples/workflows/pr-quality-gatekeeper/.gitignore new file mode 100644 index 0000000..4156ef4 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/.gitignore @@ -0,0 +1,3 @@ +outputs/ +__pycache__/ +*.pyc diff --git a/examples/workflows/pr-quality-gatekeeper/README.md b/examples/workflows/pr-quality-gatekeeper/README.md new file mode 100644 index 0000000..997aa9b --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/README.md @@ -0,0 +1,69 @@ +# PR 质量门禁工作流（pr-quality-gatekeeper） + +把已收录的 [`gitlink-gatekeeper` Skill](../../../skills/gitlink-gatekeeper/SKILL.md)（Policy-as-Code 合并门禁）包成**可直接运行的端到端工作流**： + +> **采集 → 路由 → 裁决 → 回写/善后**：读取一个真实 PR 的元信息/变更文件/commits/CI，按变更路径建议 reviewer，依 `gatekeeper.yaml` 策略算出**确定性 0–100 评分卡**与**三态裁决**（PASS / REQUEST_CHANGES / COMMENT），并（仅在 `--apply` 时）把评分卡评论、裁决标签、tracking issue 真实回写到 GitLink。 + +与仓库内已有能力的关系：`label` 命令（裁决标签）→ `gitlink-gatekeeper` Skill（裁决知识）→ **本工作流（可复现闭环）**，三层共用同一套策略文件，互为支撑而非重复。 + +## 交付物 + +- `scripts/gatekeeper_workflow.py`：单 PR 门禁闭环（纯标准库，Python ≥3.9，零第三方依赖） +- `scripts/gatekeeper_sweep.py`：**仓库级批量体检**——对全部 open PR 逐个 dry-run，产出治理报告 +- `owner-rules.example.yaml`：变更路径 → reviewer 的路由表样例 +- `config.example.yaml`：工作流配置样例（命令行参数可覆盖） +- `findings.example.json`：AI/人工审查发现注入样例（**来自对真实 PR diff 的真实审查**，行号可复核） +- `docs/architecture.md` · `docs/quickstart.md` · `docs/runbook.md` · `docs/verification.md` +- `ci-example/`：Gitea Actions 接入示例（PR 触发自动门禁，退出码 2 = REQUEST_CHANGES） +- `examples/demo-outputs/`：真实平台运行产物（PASS 90 评分卡 / 注入发现后的 55 分评分卡 / 113 个 open PR 的全仓体检报告） +- `tests/test_scoring.py`：确定性回归护栏（同输入 → 同分 → 同裁决） + +## 快速运行（默认 dry-run，不写远端） + +```bash +npm install -g @gitlink-ai/cli # ≥0.2.0，自带 label 命令与 gitlink-gatekeeper Skill +gitlink-cli auth login + +python3 scripts/gatekeeper_workflow.py \ + --owner --repo --pr \ + --policy ../../../skills/gitlink-gatekeeper/examples/gatekeeper.yaml \ + --owner-rules owner-rules.example.yaml \ + --output-dir outputs +``` + +- 注入审查发现得到含扣分的评分卡：加 `--findings findings.example.json` +- 真实回写（评论 + 标签 + tracking issue）：加 `--apply`（请先在自有仓库演练） +- 全仓批量体检（只读，零写入）： + +```bash +python3 scripts/gatekeeper_sweep.py \ + --owner --repo \ + --policy ../../../skills/gitlink-gatekeeper/examples/gatekeeper.yaml \ + --owner-rules owner-rules.example.yaml \ + --output-dir sweep-out --date-label $(date +%F) +``` + +更多见 [`docs/quickstart.md`](docs/quickstart.md) 与 [`docs/runbook.md`](docs/runbook.md)。 + +## 已在真实平台验证 + +全部证据见 [`docs/verification.md`](docs/verification.md)，要点： + +| 验证 | 对象 | 结果 | +|------|------|------| +| dry-run | 本仓库真实 PR（pull_request_id 15222） | ✅ PASS 90/100，8 个变更文件路由正确 | +| 注入真实审查发现 | 同一 PR + `findings.example.json` | ❌ REQUEST_CHANGES 55/100（裁决翻转，确定性可复算） | +| `--apply` 真实回写 | 自有 fork 的演练 PR | 评分卡评论 + tracking issue + 裁决标签全部由 API 回执确认 | +| **全仓批量体检** | 本仓库**全部 113 个 open PR** | 113/113 成功：PASS 105 / COMMENT 6 / REQUEST_CHANGES 2，均分 88.5；96% 未关联 issue | +| 单测 | `tests/test_scoring.py` | 全绿（锁定四个权威裁决案例的分值与裁决） | + +## 设计要点 + +- **确定性评分**：AI 只负责产出「发现列表」（可选注入），扣分与裁决由纯函数完成——同策略 + 同 PR → 同裁决，可逐位手算复现、可审计。 +- **安全默认**：默认 dry-run 什么都不写；即便策略开了 `auto_merge`，也必须 `verdict == PASS` 且显式 `--apply` 才会合并；强语义的 approve/reject 始终留给人，自动裁决只以建议性 `common` 评论 + 标签呈现。 +- **原生适配 GitLink**：PR 标题/描述取自 `pr +view` 的 `issue.subject/description`；标签挂载走「`label +list` 查 id → Raw API `POST /:owner/:repo/issues/`」；尊重 `common/approved/rejected` 三态 review。 +- **零依赖、零常驻**：纯标准库脚本 + `gitlink-cli`，无需部署 webhook 服务或数据库，CI 一条 step 即可接入（见 `ci-example/`）；确定性意味着**大规模治理零 AI 成本**。 + +## 许可证 + +随仓库 [MulanPSL-2.0](../../../LICENSE)。 diff --git a/examples/workflows/pr-quality-gatekeeper/ci-example/README.md b/examples/workflows/pr-quality-gatekeeper/ci-example/README.md new file mode 100644 index 0000000..8455c4e --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/ci-example/README.md @@ -0,0 +1,27 @@ +# CI 集成示例 —— 门禁接 CI + +本目录演示如何把 **gitlink-gatekeeper** 的 PR 看门人门禁接到 CI 上，让裁决直接挡住不达标的 PR。 + +> 这是**示例**，不是开箱即用的生产配置；`gitlink-cli` 的安装方式、PR 编号字段名需按你的 runner 实际情况调整。 + +## 文件 + +- [`gatekeeper.gitea.yml`](gatekeeper.gitea.yml)：Gitea Actions 工作流（GitLink 基于 Gitea，语法与 GitHub Actions 兼容）。 + +## 用法 + +1. 把 `gatekeeper.gitea.yml` 复制到目标仓库的 `.gitea/workflows/` 目录。 +2. 在仓库 **Settings → Actions → Secrets** 新增 `GITLINK_TOKEN`，值为有权读取该仓库 PR 的访问令牌（供 `gitlink-cli` 认证）。**Token 切勿写进仓库或日志。** +3. 提一个 PR 触发工作流即可。 + +## 工作原理 + +- 触发：PR 的 `opened` / `synchronize` / `reopened` 事件。 +- 步骤：检出 → 准备 Python 3.9（脚本纯标准库，无需装依赖）→ 装 `gitlink-cli` → 跑 `scripts/gatekeeper_workflow.py` 采集本次 PR 上下文并评分裁决。 +- **退出码即门禁**： + - `0` = PASS / COMMENT → job 通过，放行。 + - `2` = REQUEST_CHANGES → 工作流把它转成 job 失败，挡住该 PR。 + - `1` = 可预期错误（缺参数 / 未装 `gitlink-cli` 等）→ 同样失败。 +- 产物：评分卡与 `summary.json` 落在 `outputs/`，工作流用 `upload-artifact` 上传，便于在 CI 页面查看裁决依据。 + +调门禁松紧只需改 `--policy` 指向的 `gatekeeper.yaml`（策略字段说明见 [`gitlink-gatekeeper` Skill REFERENCE](../../../../skills/gitlink-gatekeeper/REFERENCE.md)）。 diff --git a/examples/workflows/pr-quality-gatekeeper/ci-example/gatekeeper.gitea.yml b/examples/workflows/pr-quality-gatekeeper/ci-example/gatekeeper.gitea.yml new file mode 100644 index 0000000..aa22cc6 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/ci-example/gatekeeper.gitea.yml @@ -0,0 +1,75 @@ +# gitlink-gatekeeper —— Gitea Actions CI 示例（GitLink 平台用） +# +# ⚠️ 这是一个「门禁接 CI」的演示示例，不是开箱即用的生产配置。 +# - GitLink 基于 Gitea，其 Actions 语法与 GitHub Actions 兼容，工作流放在 +# 仓库的 .gitea/workflows/ 目录下。把本文件复制过去并按需调整即可启用。 +# - 需要在仓库 Settings → Actions → Secrets 配置一个 GITLINK_TOKEN secret +# （供 gitlink-cli 认证、采集目标 PR 的上下文）。Token 切勿写进仓库。 +# - runner 需能访问 GitLink API；Python 3.9+ 与 gitlink-cli 的安装方式按实际 +# runner 镜像调整（下面 install 步骤仅为占位示意）。 +# +# 触发：对 PR 的 open / 同步事件运行门禁，脚本返回码 2（REQUEST_CHANGES） +# 会让本 job 失败，从而在 CI 上挡住该 PR（PASS/COMMENT 返回 0 即通过）。 + +name: gatekeeper + +on: + pull_request: + types: [opened, synchronize, reopened] + +jobs: + gatekeeper: + runs-on: ubuntu-latest + steps: + - name: 检出代码 + uses: actions/checkout@v4 + + - name: 准备 Python（纯标准库，无需装依赖） + uses: actions/setup-python@v5 + with: + python-version: "3.9" + + # 安装 gitlink-cli（示意：按 runner 实际情况替换为正确的安装/分发方式） + - name: 安装 gitlink-cli + run: | + # 例如从发布物下载或用包管理器安装，确保 PATH 里有 gitlink-cli + gitlink-cli --version + + # 跑门禁：采集本次 PR 上下文 → 评分 → 裁决。 + # REQUEST_CHANGES 时脚本退出码为 2；下面用 if/exit 把它转成 job 失败。 + - name: 运行 PR 看门人门禁 + env: + # gitlink-cli 通过该环境变量认证（对应仓库配置的 secret） + GITLINK_TOKEN: ${{ secrets.GITLINK_TOKEN }} + # Gitea 注入的 PR 编号；不同 runner 字段名可能不同，按实际调整 + PR_NUMBER: ${{ github.event.pull_request.number }} + run: | + set -o pipefail + python3 examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_workflow.py \ + --owner "${{ github.repository_owner }}" \ + --repo "${{ github.event.repository.name }}" \ + --pr "${PR_NUMBER}" \ + --policy skills/gitlink-gatekeeper/examples/gatekeeper.yaml \ + --owner-rules workflow/owner-rules.yaml \ + --cli-bin gitlink-cli + code=$? + # 退出码：0 = PASS/COMMENT（放行）；2 = REQUEST_CHANGES（挡住，让 job 失败）； + # 1 = 可预期错误（缺参数 / 未装 gitlink-cli 等），同样视为失败。 + if [ "$code" -eq 0 ]; then + echo "门禁通过（PASS/COMMENT）" + exit 0 + elif [ "$code" -eq 2 ]; then + echo "::error::门禁裁决 REQUEST_CHANGES，阻止合并" + exit 1 + else + echo "::error::门禁执行出错（退出码 $code）" + exit 1 + fi + + # 上传评分卡 / summary 产物，便于在 CI 页面查看裁决依据 + - name: 上传门禁产物 + if: always() + uses: actions/upload-artifact@v4 + with: + name: gatekeeper-scorecard + path: outputs/ diff --git a/examples/workflows/pr-quality-gatekeeper/config.example.yaml b/examples/workflows/pr-quality-gatekeeper/config.example.yaml new file mode 100644 index 0000000..2137027 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/config.example.yaml @@ -0,0 +1,19 @@ +# config.example.yaml — PR 看门人闭环的工作流配置（gatekeeper_workflow.py --config 用） +# +# 命令行参数（--owner/--repo/--pr/--policy/--owner-rules/--findings）会覆盖这里的同名字段。 +# 相对路径以本配置文件所在目录为基准解析。 + +owner: Gitlink # 目标仓库 owner（GitLink 主分支为 master） +repo: gitlink-cli # 目标仓库名 +pr: 1 # 目标 PR 编号（用 --pr 覆盖以复用同一配置跑不同 PR） + +# 策略文件（Policy-as-Code）。缺省 / 文件不存在时回退脚本内置默认策略（SSOT 第 2 节）。 +policy: ../skills/gitlink-gatekeeper/examples/gatekeeper.yaml + +# 文件路径 → reviewer 路由表（工作流步骤 1） +owner_rules: owner-rules.yaml + +# 可选：AI 审查发现 JSON（注入 review_findings 维度）。 +# 缺省为空 → 评分仍确定性可复现（同策略 + 同 PR 上下文 → 同裁决）。 +# 这里默认指向随附的示例（1 major + 2 minor），开箱即可跑通；用 --findings 覆盖为你自己的产物。 +findings: findings.sample.json diff --git a/examples/workflows/pr-quality-gatekeeper/docs/architecture.md b/examples/workflows/pr-quality-gatekeeper/docs/architecture.md new file mode 100644 index 0000000..ba74969 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/docs/architecture.md @@ -0,0 +1,66 @@ +# 架构与数据流 — PR 看门人闭环 + +本工作流采用「**采集 → 路由 → 裁决 → 回写/善后**」四段式流水线，把 `gitlink-gatekeeper` 的 Policy-as-Code 门禁包成一条可复现闭环。所有数值/字段/算法以已收录的 [`gitlink-gatekeeper` Skill REFERENCE](../../../../skills/gitlink-gatekeeper/REFERENCE.md) 为准。 + +## 设计目标 + +- **可复现**：同策略 + 同 PR 上下文 → 同评分卡 + 同裁决（确定性算法，SSOT 第 3–5 节）。 +- **可审计**：评分卡逐维列分 + 备注，硬门禁逐条列出命中原因，裁决可追溯到具体规则与 `gatekeeper.yaml` 版本。 +- **安全默认**：默认 dry-run，写操作需显式 `--apply`；绝不默认自动合并（SSOT 第 8 节）。 +- **低门槛**：仅依赖 `gitlink-cli` 与 Python 标准库（含内置 YAML 子集解析器，无第三方包）。 +- **边界清晰**：采集、路由、裁决、回写四段各自独立，便于单测与替换（如换一套评分维度只动裁决段）。 + +## ASCII 流程图 + +> 下面是数据流占位图：左侧为 `gitlink-cli` 采集，中间为本脚本的确定性处理，右侧为回写/善后的写操作（仅 `--apply` 时执行）。 + +``` + ┌──────────────────────────── gatekeeper_workflow.py ────────────────────────────┐ + │ │ + gitlink-cli (读) │ step 1: 路由 step 2: 裁决 step 3: 回写 + 善后 │ gitlink-cli (写, 仅 --apply) + ───────────────────── │ ─────────────────── ────────────────── ───────────────────────── │ ───────────────────────────── + │ │ + pr +view ──┐ │ ┌─ review_findings(40) ─┐ │ + pr +files ──┼──▶ 采集 │ changed_files │ test_coverage (20) │ verdict │ pr +comment ─▶ 评分卡评论 + ci +builds │ 归一化 │ │ │ pr_hygiene (15) │──▶ ── PASS ───────┐ │ + api .../ │ │ ▼ │ commit_quality(15) │ ── COMMENT ───┐ │ │ label +create ─▶ 裁决标签 + commits ──┘ │ owner-rules.yaml │ ci_status (10) │ ── REQ_CHG ─┐ │ │ │ (+ 挂 issue_tag_ids + ▲ │ (glob → reviewer) └───────────┬───────────┘ │ │ │ │ via Raw API POST + │ │ │ ▼ │ │ │ │ /:owner/:repo/issues/:id) + gatekeeper.yaml ─────┼─────┼──────────────▶ hard_gates 判定 ─┴─▶ total 0..100 ─▶ 阈值 │ │ │ │ + (Policy-as-Code) │ ▼ (SSOT 第4节) (SSOT 第3节) (第5节) │ │ │ │ issue +create ─▶ tracking issue + │ │ suggested_reviewers ──────────────────────────────────────────┘ │ │ │ (仅 REQUEST_CHANGES) + findings.json ───────┼──▶ review_findings 注入 │ │ │ + (可选, AI 审查) │ │ │ │ pr +merge ─▶ 合并 (受限: + │ ┌── outputs/*_scorecard.md ◀────────────┘ │ │ PASS + auto_merge + --apply) + │ 本地产物落盘 (总是) ───────┤ │ │ + │ └── outputs/*_summary.json ◀──────────────┘ │ + └─────────────────────────────────────────────────────────────────────────────────┘ + + dry-run（默认）：右侧写操作仅打印「将要执行的命令」，不实际调用 → 安全。 + --apply ：右侧写操作真正执行；其中合并需同时满足 PASS + 策略 auto_merge=true + --apply。 +``` + +## 四段职责 + +### ① 采集（collect_pr_context） +调只读 `gitlink-cli` 命令拿到 PR 元信息、变更文件、CI 状态、commits（端点未开放时降级，不阻断）。输出统一归一化为内部结构，兼容 GitLink Envelope 的多种字段名。 + +### ② 路由（route_reviewers） +读 `owner-rules.yaml`，对每个变更文件按 glob 顺序匹配（首个命中生效，顺序即优先级），产出 `reviewer → 文件清单`；未命中文件归 `default_reviewers`。结果写进评分卡的「Suggested reviewers」分区。**只产出建议，不调用任何写操作**——是否真正分配由维护者决定。 + +### ③ 裁决（score_dimensions / evaluate_hard_gates / decide_verdict） +- 五维加权评分（权重和=100，SSOT 第 3 节），可选注入 AI findings 影响 `review_findings`。 +- 硬门禁逐项判定（SSOT 第 4 节），任一命中即 `hard_gate_failed`。 +- 裁决判定树（SSOT 第 5 节）：硬门禁失败 → REQUEST_CHANGES；否则按总分与 `pass`/`request_changes` 阈值落三态。 +- 渲染评分卡（SSOT 第 6 节模板）。 + +### ④ 回写 + 善后（build_*_command + execute_write） +按裁决构造写操作计划：评分卡评论、裁决标签、（REQUEST_CHANGES 时）tracking issue、（受限）合并。dry-run 只打印计划；`--apply` 才逐条执行并记录结果到 `summary.json`。 + +## 为什么选这条链路 + +子赛题三要求用现有命令 / Skill 组合形成完整解决方案，且串联不少于 3 步。本链路： +1. 串联了 **4 个只读采集命令** + **最多 4 个写命令**，远超 3 步下限。 +2. 形成从「数据获取」到「治理动作落地」的端到端闭环，并能接入 CI（REQUEST_CHANGES 返回码 2）。 +3. 复用本作品自研的 `label` 命令组（子赛题一）与 gatekeeper 策略（子赛题二），三个子赛题在同一作品内闭环，相互增强。 diff --git a/examples/workflows/pr-quality-gatekeeper/docs/quickstart.md b/examples/workflows/pr-quality-gatekeeper/docs/quickstart.md new file mode 100644 index 0000000..36b0aab --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/docs/quickstart.md @@ -0,0 +1,48 @@ +# 最短复现路径（3 步） + +## 1. 安装与认证 + +```bash +npm install -g @gitlink-ai/cli # ≥0.2.0（label 命令与 gitlink-gatekeeper Skill 已内置） +gitlink-cli auth login # 或 export GITLINK_TOKEN=<私人令牌> +gitlink-cli auth status # 确认已登录 +``` + +## 2. 对任意真实 PR 出评分卡（dry-run，零写入） + +在本目录（`examples/workflows/pr-quality-gatekeeper/`）下： + +```bash +python3 scripts/gatekeeper_workflow.py \ + --owner Gitlink --repo gitlink-cli --pr \ + --policy ../../../skills/gitlink-gatekeeper/examples/gatekeeper.yaml \ + --owner-rules owner-rules.example.yaml \ + --output-dir outputs +``` + +产物：`outputs/__pr_scorecard.md`（评分卡）+ `_summary.json`（结构化摘要）。 +退出码：`0` = PASS/COMMENT，`2` = REQUEST_CHANGES（可直接当 CI 门禁用），`1` = 运行错误。 + +不带 `--policy` 也能跑（脚本内置同值默认策略）；想看含扣分的评分卡，加 `--findings findings.example.json`。 + +## 3. 可选进阶 + +- **真实回写**（评论 + 裁决标签 + tracking issue）：加 `--apply`。请先在自有 fork 演练；自动裁决只用建议性 `common` 评论，绝不替人 approve/reject，绝不自动合并。 +- **全仓体检**（只读批扫全部 open PR，出治理报告）： + +```bash +python3 scripts/gatekeeper_sweep.py \ + --owner Gitlink --repo gitlink-cli \ + --policy ../../../skills/gitlink-gatekeeper/examples/gatekeeper.yaml \ + --owner-rules owner-rules.example.yaml \ + --output-dir sweep-out --date-label $(date +%F) +``` + +- **CI 接入**：见 [`../ci-example/`](../ci-example/)（Gitea Actions，PR 触发自动门禁）。 +- **改门禁松紧**：复制一份 `gatekeeper.yaml` 改 `weights/hard_gates/thresholds`，字段说明见 [Skill REFERENCE](../../../../skills/gitlink-gatekeeper/REFERENCE.md)。 + +## 验证自己改动没破坏确定性 + +```bash +python3 tests/test_scoring.py # 同输入 → 同分 → 同裁决 的回归护栏 +``` diff --git a/examples/workflows/pr-quality-gatekeeper/docs/runbook.md b/examples/workflows/pr-quality-gatekeeper/docs/runbook.md new file mode 100644 index 0000000..1e965a1 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/docs/runbook.md @@ -0,0 +1,106 @@ +# 运行手册 — PR 看门人闭环 + +本手册覆盖 `scripts/gatekeeper_workflow.py` 的前置条件、运行步骤、参数、预期输出与回滚。数值/字段以已收录的 [`gitlink-gatekeeper` Skill REFERENCE](../../../../skills/gitlink-gatekeeper/REFERENCE.md) 为准。 + +## 1. 前置条件 + +- 已安装 `gitlink-cli` 且在 `PATH` 中（或用 `--cli-bin` 指定路径）。 +- 已完成登录：`gitlink-cli auth login`（Token 有效期 7 天，过期重新登录；详见 [gitlink-shared](../../../../skills/gitlink-shared/SKILL.md)）。 +- 对目标仓库有读权限；要 `--apply` 回写评论/标签/建 issue 时需写权限。 +- Python 3.8+（脚本纯标准库，无需 `pip install`）。 + +验证登录态： + +```bash +gitlink-cli auth status +gitlink-cli pr +view -i --owner --repo --format json # 确认目标 PR 可读 +``` + +## 2. 配置 + +编辑 [`../config.example.yaml`](../config.example.yaml)（或复制一份），填好 `owner`/`repo`/`pr` 与策略、路由表路径。命令行参数会覆盖配置同名字段，相对路径以配置文件所在目录为基准。 + +按需调整 [`../owner-rules.example.yaml`](../owner-rules.example.yaml)：把占位 reviewer 用户名替换成本仓库维护者，按「具体规则在前」排列 glob。 + +按需选择策略预设（均在 [`../../../../skills/gitlink-gatekeeper/examples/`](../../../../skills/gitlink-gatekeeper/examples/)）： +- `gatekeeper.yaml`：均衡基线（= SSOT 内置默认）。 +- `gatekeeper.strict.yaml`：严格预设。 +- `gatekeeper.lenient.yaml`：宽松预设。 +- 不指定 `--policy` 且配置无 `policy` 字段时，回退脚本内置默认策略（与 `gatekeeper.yaml` 等价）。 + +## 3. 运行步骤 + +### 步骤 A：dry-run 预览（安全默认，必做） + +```bash +python3 scripts/gatekeeper_workflow.py --config config.example.yaml --pr +``` + +此模式**不写任何东西**，只采集 + 评分 + 打印将要执行的写命令 + 落盘本地产物。先看评分卡与计划是否符合预期。 + +### 步骤 B：注入 AI 审查发现（可选） + +`review_findings` 维度默认按 0 发现计分。若已有 AI 代码审查（如 `gitlink-code-review` Skill）产出，整理成 JSON 注入： + +```json +{ "findings": [ + { "severity": "blocker", "message": "硬编码密钥", "file": "internal/auth/refresh.go", "line": 12 }, + { "severity": "minor", "message": "缺超时上下文", "file": "internal/auth/handler.go", "line": 40 } +] } +``` + +```bash +python3 scripts/gatekeeper_workflow.py --config config.example.yaml --pr --findings findings.json +``` + +`severity` 取值：`blocker` / `major` / `minor` / `nit`（其余忽略）。 + +### 步骤 C：apply 执行写操作 + +确认 dry-run 计划无误后，加 `--apply`： + +```bash +python3 scripts/gatekeeper_workflow.py --config config.example.yaml --pr --apply +``` + +将依次执行（按裁决）：回写评分卡评论 → 确保裁决标签存在 →（仅 REQUEST_CHANGES）创建 tracking issue。 +**合并不会自动发生**：仅当策略 `behavior.auto_merge: true` 且裁决为 `PASS` 且本次带 `--apply` 时，才追加 `pr +merge`。默认 `auto_merge: false`。 + +## 4. 参数速查 + +| 参数 | 说明 | 默认 | +|------|------|------| +| `--config` | 工作流配置 YAML（owner/repo/pr/policy/owner_rules/findings） | 无 | +| `--owner` / `--repo` / `--pr` | 覆盖配置中的目标 | 取自 config | +| `--policy` | `gatekeeper.yaml` 路径 | 内置默认策略 | +| `--owner-rules` | `owner-rules.yaml` 路径 | 取自 config | +| `--findings` | AI 审查发现 JSON | 空（0 发现） | +| `--cli-bin` | `gitlink-cli` 可执行路径 | `gitlink-cli` | +| `--skip-ci` | 跳过 CI 采集（`ci_status` 记 `unknown`） | 否 | +| `--output-dir` | 本地产物目录 | `outputs` | +| `--apply` | **执行写操作**；不传则仅预览 | 否（dry-run） | + +## 5. 预期输出 + +- 终端：三段进度（路由 / 裁决 / 回写）+ 评分概览 + 计划或执行结果 + 最终裁决。 +- 文件： + - `outputs/__pr_scorecard.md` — 评分卡（SSOT 第 6 节模板）。 + - `outputs/__pr_summary.json` — 结构化摘要（路由、各维得分、硬门禁、裁决、`planned_writes`、`executed`、产物路径）。 +- 退出码：`PASS`/`COMMENT` → `0`；`REQUEST_CHANGES` → `2`（可作 CI 门禁）；可预期错误（缺配置 / 未登录 / CLI 缺失）→ `1`。 + +样例评分卡见 [`../../../../skills/gitlink-gatekeeper/examples/scorecard-sample.md`](../../../../skills/gitlink-gatekeeper/examples/scorecard-sample.md)。 + +## 6. 回滚 + +dry-run 不产生任何远端副作用，无需回滚（本地产物可直接删 `outputs/`）。 + +`--apply` 后如需撤销： + +| 已做的写操作 | 回滚方式 | +|--------------|----------| +| 回写的评分卡评论 | 评论走 issue journals，在 PR 页面手动删除该评论即可；脚本不提供删除命令（避免误删他人评论） | +| 创建的裁决标签定义 | `gitlink-cli label +delete -i --owner --repo `（先 `label +list` 查 id） | +| 创建的 tracking issue | `gitlink-cli issue +close -n --owner --repo `（关闭而非删除，保留审计痕迹） | +| 已合并的 PR | **不可自动回滚**。这也是默认 `auto_merge: false` 的原因；合并前务必人工确认。如确需撤销，按仓库常规流程 revert commit | + +> 安全提示：任何 `--apply` 写操作前，脚本会在 dry-run 计划里完整复述将执行的命令。生产仓库建议先 dry-run，再 `--apply`。 diff --git a/examples/workflows/pr-quality-gatekeeper/docs/verification.md b/examples/workflows/pr-quality-gatekeeper/docs/verification.md new file mode 100644 index 0000000..c2d461e --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/docs/verification.md @@ -0,0 +1,55 @@ +# 真实平台验证记录 + +> 全部针对 **GitLink 线上真实平台** 运行（`gitlink-cli` + Token 认证），非 mock。 +> 他人仓库一律 dry-run（只读零写入）；写操作只在自有 fork 演练。 +> 运行环境：macOS（Apple Silicon）· Python 3.9 · `@gitlink-ai/cli` 0.2.0（npm 官方发布版，零本地构建）。 + +## A. dry-run：真实 PR → PASS 90/100 + +对本仓库真实 PR（`pull_request_id 15222`，feat/org-team-projects，8 个变更文件）： + +- 路由正确分流：README/docs/skill → doc-maintainer；`org.go` → go-reviewer；`org_test.go` → qa-reviewer +- 评分（确定性，可手算复现）：review 40/40 · test 20/20（1 src/1 test）· hygiene 10/15（desc✓/issue✗/size✓）· commit 15/15 · ci 5/10（unknown）= **90 → PASS** +- CI 取不到构建记录 → `unknown`：按策略**不触发**硬门禁（仅显式 `failing` 触发），只在 CI 维记半分 +- 产物：[`../examples/demo-outputs/scorecard-pass-90.md`](../examples/demo-outputs/scorecard-pass-90.md) + +## B. 注入真实审查发现 → REQUEST_CHANGES 55/100 + +同一 PR，注入 [`../findings.example.json`](../findings.example.json) 重跑：review_findings 40/40 → 5/40（1 major + 2 minor），总分 90 → 55，**裁决翻转为 REQUEST_CHANGES**。 + +**发现是真的，不是编的**——三条均来自对该 PR 真实 diff（head `bcc27bf`）的代码审查，标注 `shortcuts/org/org.go` 真实行号，任何人拉取该分支可逐条复核。其中 major：新增的 `parseBool` 只认字面 `"true"`，`--dry-run=1` 会被静默当 false，而该 flag 守护的是「批量移除团队全部项目」这一破坏性操作。 + +产物：[`../examples/demo-outputs/scorecard-findings-55.md`](../examples/demo-outputs/scorecard-findings-55.md) + +## C. `--apply` 真实回写（自有 fork 演练） + +在自有 fork 的演练 PR（故意「改源码不带测试」）上执行 `--apply`： + +- 触发硬门禁 `require_tests_for_src_changes` → REQUEST_CHANGES 40/100 +- GitLink API 回执确认三件写操作全部落地： + 1. 评分卡评论回写到 PR（comment id `472741`） + 2. 自动创建 tracking issue（id `143217`），汇总硬门禁 + 必修项 + 建议 reviewer，与 PR 双向回链 + 3. 裁决标签挂载到 PR 背后 issue（`label +list` 查 id → Raw API `POST /:owner/:repo/issues/`）——依赖本仓库的 `label` 命令（0.2.0 起官方发布版自带） + +## D. 全仓批量体检：113 个 open PR + +`gatekeeper_sweep.py` 对本仓库**全部 113 个 open PR** 逐个 dry-run（只读、零写入、零 AI 成本），113/113 成功： + +- 裁决分布：**PASS 105 · COMMENT 6 · REQUEST_CHANGES 2**；分数 min 70 / 中位 90 / 均值 88.5 / max 95 +- 治理洞察：**96% 的 open PR 未关联 issue**；2 个 PR 触发 `require_tests_for_src_changes`（改源码不带测试） +- 完整报告（含全量明细表）：[`../examples/demo-outputs/sweep-report-2026-06-10.md`](../examples/demo-outputs/sweep-report-2026-06-10.md) +- 诚实口径：批扫不注入审查发现（review_findings 维未评、按满分计），CI 统一 `--skip-ci`（unknown 半分）——总分代表「除人工/AI 审查外的工程卫生分」，偏乐观 + +## E. 单元测试（确定性回归护栏） + +```bash +$ python3 tests/test_scoring.py +OK +``` + +锁定四个权威裁决案例（PASS / REQUEST_CHANGES / COMMENT / 硬门禁直拒）的**总分与裁决**与 Skill 文档逐位一致；任何改动若破坏「同输入 → 同分 → 同裁决」，测试立即变红。 + +## 真实运行当场暴露过的问题（透明记录） + +- GitLink 的 PR 标题/描述在 `pr +view` 返回的 `issue.subject/description`，而非 `pull_request` 子对象——离线 mock 测不到，真实平台运行才暴露并修复。 +- npm 0.1.18 时代 `--apply` 的打标签步骤会报 `unknown command "label"`（彼时 `label` 命令尚未发布）；0.2.0 起官方发布版自带，整条闭环零本地构建跑通。 diff --git a/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/scorecard-findings-55.md b/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/scorecard-findings-55.md new file mode 100644 index 0000000..490dd3b --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/scorecard-findings-55.md @@ -0,0 +1,28 @@ +## 🛡️ Gatekeeper Report — PR #15222 feat(org): add team project binding shortcuts + +**Verdict: ❌ REQUEST_CHANGES** · Score: 55/100 · policy: gatekeeper.yaml@v1 + +| Dimension | Weight | Score | Notes | +|-----------|:------:|:-----:|-------| +| Review findings | 40 | 5/40 | 0 blocker / 1 major / 2 minor / 0 nit | +| Test coverage | 20 | 20/20 | 1 src / 1 test files | +| PR hygiene | 15 | 10/15 | desc ✓ / linked issue ✗ / size ✓ | +| Commit quality | 15 | 15/15 | 0/0 conventional | +| CI status | 10 | 5/10 | unknown | + +### 👥 Suggested reviewers (3) +- @doc-maintainer — 6 file(s): README.md, README.zh-CN.md, doc/changes/org-team-projects.md … +- @go-reviewer — 2 file(s): shortcuts/org/org.go, shortcuts/org/org_test.go +- @qa-reviewer — 1 file(s): shortcuts/org/org_test.go + +### 🔴 Must fix (1) +- [major] parseBool 只把字面 "true" 当真：用户传 --dry-run=1 / t / TRUE 以外写法会被静默解析为 false。该 flag 守护的是 team-projects-remove-all（批量移除团队全部项目）这类破坏性操作——预览意图被静默降级为真实执行。建议改用 strconv.ParseBool（与 shortcuts/common/runner.go:52 解析 flag 默认值的行为一致），无法识别的值应报错而非吞掉 — shortcuts/org/org.go:156 + +### 🟡 Should fix (2) +- [minor] team-projects-remove-all 一条命令清空团队全部项目绑定，除 --dry-run 外无确认机制；建议在 Description/help 标注危险性，或要求显式 --yes 二次确认 — shortcuts/org/org.go:103 +- [minor] dry-run 预览 payload 中 action（add_all_team_projects/remove_all_team_projects）与实际请求 path 段（create_all/destroy_all，见 :134）是两套词汇，排查问题时易误导；建议统一或在 payload 同时给出两者映射 — shortcuts/org/org.go:142 + +### Next steps +1. 评分低于阈值，按上方 Must/Should fix 修复后重新触发 gatekeeper +--- +*Generated by gitlink-gatekeeper · policy-as-code PR gate · re-run after changes* diff --git a/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/scorecard-pass-90.md b/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/scorecard-pass-90.md new file mode 100644 index 0000000..76754a1 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/scorecard-pass-90.md @@ -0,0 +1,21 @@ +## 🛡️ Gatekeeper Report — PR #15222 feat(org): add team project binding shortcuts + +**Verdict: ✅ PASS** · Score: 90/100 · policy: gatekeeper.yaml@v1 + +| Dimension | Weight | Score | Notes | +|-----------|:------:|:-----:|-------| +| Review findings | 40 | 40/40 | 0 blocker / 0 major / 0 minor / 0 nit | +| Test coverage | 20 | 20/20 | 1 src / 1 test files | +| PR hygiene | 15 | 10/15 | desc ✓ / linked issue ✗ / size ✓ | +| Commit quality | 15 | 15/15 | 0/0 conventional | +| CI status | 10 | 5/10 | unknown | + +### 👥 Suggested reviewers (3) +- @doc-maintainer — 6 file(s): README.md, README.zh-CN.md, doc/changes/org-team-projects.md … +- @go-reviewer — 2 file(s): shortcuts/org/org.go, shortcuts/org/org_test.go +- @qa-reviewer — 1 file(s): shortcuts/org/org_test.go + +### Next steps +1. 满足合并门禁；如策略开启 auto_merge 且操作者带 --apply，可执行合并 +--- +*Generated by gitlink-gatekeeper · policy-as-code PR gate · re-run after changes* diff --git a/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/sweep-report-2026-06-10.md b/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/sweep-report-2026-06-10.md new file mode 100644 index 0000000..5f21585 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/examples/demo-outputs/sweep-report-2026-06-10.md @@ -0,0 +1,138 @@ +# gatekeeper 仓库体检报告 —— Gitlink/gitlink-cli（2026-06-10） + +> 对 **113 个 open PR** 全量 dry-run（**只读，零写入**）· 策略 `gatekeeper.yaml` · 成功 113 / 失败 0 +> +> **诚实口径**：批扫未注入 AI 审查发现，review_findings 维按 0 发现计满分（**该维度未评**）；CI 维按 `--skip-ci` 统一记 unknown（半分）。其余维度为真实采集。因此**总分代表「除人工/AI 审查外的工程卫生分」，偏乐观**；裁决分布同理。 + +## 总览 + +- 裁决分布：COMMENT **6** · PASS **105** · REQUEST_CHANGES **2** +- 分数：min 70 / 中位 90 / 均值 88.5 / max 95 +- **0%** 的 PR 测试覆盖维 0 分（改动不带任何测试） +- **96%** 的 PR 未关联 issue +- **2%** 的 PR 触发 REQUEST_CHANGES（硬门禁或低分） + +硬门禁命中：`require_tests_for_src_changes` × 2 + +## 全量明细（按分数降序） + +| PR | 标题 | 作者 | 总分 | 裁决 | 硬门禁失败 | 卫生(描述/关联/体量) | +|----|------|------|-----:|------|-----------|---------------------| +| [#145](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/145) | fix(issue): preserve metadata during batch close | dtwdtw | 95 | PASS | — | ✓/✓/✓ | +| [#218](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/218) | feat(skills): 新增 科研Fork影响力分析 的skill : gitlink-re | yangsai | 90 | PASS | — | ✓/✗/✓ | +| [#177](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/177) | feat(wiki): add wiki management shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#217](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/217) | feat(commands): add command catalog export | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#216](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/216) | feat(api): support saved variables in batch plan | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#214](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/214) | feat(pr): add conversation comment shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#213](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/213) | feat(repo): add mirror sync shortcut | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#212](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/212) | feat(feedback): add feedback shortcut | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#211](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/211) | feat(repo): add profile view shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#210](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/210) | feat(skills): 新增维护者交接与分支治理 Skills | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#208](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/208) | feat(user): add pinned project shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#207](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/207) | feat(user): add statistics shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#206](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/206) | feat(commit): add commit inspection shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#204](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/204) | feat(org): 增强组织团队与成员管理快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#203](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/203) | feat(user): 增加用户画像分析快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#202](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/202) | feat(ignore): add ignore template shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#201](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/201) | feat(account): add account auth shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#200](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/200) | feat(pr): add review journal shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#199](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/199) | feat(code): add read-only code browsing shortcut | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#198](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/198) | feat(message): 增加消息中心快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#197](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/197) | feat(message-settings): 增加消息通知设置快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#194](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/194) | fix(pr): 补齐 pr +view 的合并与关闭时间字段 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#193](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/193) | feat(shortcut): add shortcuts/wiki | co63oc | 90 | PASS | — | ✓/✗/✓ | +| [#192](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/192) | feat(repo): add navigation unit shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#191](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/191) | feat(user): add profile shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#187](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/187) | feat(org): add team project bulk shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#186](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/186) | feat(ref): add branch and tag shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#185](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/185) | Add workflow pull request review queue | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#184](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/184) | Add workflow release notes generator | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#183](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/183) | feat(project): add lifecycle flow shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#182](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/182) | feat(issue): add journal maintenance shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#181](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/181) | feat(topic): add project topic shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#180](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/180) | feat(template): add project template shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#179](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/179) | feat(dataset): add research dataset shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#178](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/178) | feat(contents): add repository content shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#176](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/176) | feat(user): add dashboard shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#175](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/175) | feat(notification): add message and setting shor | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#174](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/174) | feat(public-key): add SSH key shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#173](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/173) | feat(account): add cancellation shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#172](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/172) | feat(account): add security shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#171](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/171) | feat(oauth): add token shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#170](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/170) | Add repository file search and batch commit shor | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#167](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/167) | feat(account): add email verification shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#164](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/164) | Add pull request review comment management short | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#163](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/163) | Add complete issue comment management shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#160](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/160) | Add GitLink feedback submission shortcut | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#158](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/158) | Add code trace analysis shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#153](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/153) | feat(shortcut): add shortcuts/ignore | co63oc | 90 | PASS | — | ✓/✗/✓ | +| [#151](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/151) | feat(transfer): add transfer request shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#135](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/135) | feat(dev): add developer resource shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#118](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/118) | feat(access): add project access shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#114](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/114) | feat(mirror): add mirror repository shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#113](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/113) | feat(todo): add request approval shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#107](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/107) | feat(star): add starred project shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#83](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/83) | feat(org): add team project binding shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#82](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/82) | feat(meta): add attachment and metadata shortcut | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#78](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/78) | feat(branch): complete OpenAPI shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#76](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/76) | feat(notification): add OpenAPI shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#72](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/72) | feat(template): add project template shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#70](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/70) | feat(user): add account and stats shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#65](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/65) | feat(wiki): add OpenAPI shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#64](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/64) | feat(dataset): add OpenAPI shortcuts | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#63](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/63) | feat(code): add repository code OpenAPI shortcut | wangyue111 | 90 | PASS | — | ✓/✗/✓ | +| [#152](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/152) | chore(doc): fix README.md | co63oc | 90 | PASS | — | ✓/✗/✓ | +| [#137](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/137) | feat(skills): 增强 7 个 Agent Skill + 新增 2 个 Skill（ | whale | 90 | PASS | — | ✓/✗/✓ | +| [#149](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/149) | feat(skills): 新增 学者/团队科研画像生成 的skill : gitlink-sc | yangsai | 90 | PASS | — | ✓/✗/✓ | +| [#148](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/148) | feat(skills): 新增 科研热点追踪与知识图谱构建 的skill : gitlink- | yangsai | 90 | PASS | — | ✓/✗/✓ | +| [#144](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/144) | feat(skills): 新增 3 个 Agent Skill — wiki-builder, | whale | 90 | PASS | — | ✓/✗/✓ | +| [#134](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/134) | 新增 shell 自动补全命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#99](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/99) | 新增 5 个仓库检查快捷命令 (languages/contributors/files/tag | jiangtx | 90 | PASS | — | ✓/✗/✓ | +| [#86](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/86) | fix: preserve issue metadata on update | dtwdtw | 90 | PASS | — | ✓/✗/✓ | +| [#73](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/73) | feat(user): add SSH key shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#67](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/67) | feat(repo): add repository units shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#60](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/60) | feat: add notification shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#58](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/58) | feat: add repository reaction shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#126](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/126) | feat(skills): 新增 gitlink-scaffold 社区健康文件体检 Skill | Ct201314 | 90 | PASS | — | ✓/✗/✓ | +| [#56](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/56) | feat: add git tag shortcut group | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#125](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/125) | feat(skills): 新增 gitlink-newcomer 新人引导 Skill | Ct201314 | 90 | PASS | — | ✓/✗/✓ | +| [#127](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/127) | feat(skills): 新增 gitlink-deps 依赖追踪 Skill | Ct201314 | 90 | PASS | — | ✓/✗/✓ | +| [#128](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/128) | feat(skills): 新增 gitlink-contributor 贡献者致谢与成长 Sk | Ct201314 | 90 | PASS | — | ✓/✗/✓ | +| [#129](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/129) | feat(skills): 新增 gitlink-kb 知识库问答 Skill | Ct201314 | 90 | PASS | — | ✓/✗/✓ | +| [#115](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/115) | feat: add catalog template shortcuts | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#116](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/116) | 新增仓库洞察快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#119](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/119) | 新增仓库转移快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#122](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/122) | 完善仓库 README 快捷命令 | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#50](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/50) | feat: add wiki shortcut group | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#54](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/54) | gitlink-growth 开源贡献者成长系统 Skill 贡献 | yingjie | 90 | PASS | — | ✓/✗/✓ | +| [#23](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/23) | feat: support fork metadata in pr create | Mengz | 90 | PASS | — | ✓/✗/✓ | +| [#196](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/196) | feat(release): 增加发布资产管理快捷命令 | Mengz | 88 | PASS | — | ✓/✗/✓ | +| [#215](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/215) | fix(client): improve API robustness | wangyue111 | 87 | PASS | — | ✓/✗/✓ | +| [#147](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/147) | feat(shortcuts): 新增 wiki/commit/file/star/watch | chroe | 86 | PASS | — | ✓/✗/✓ | +| [#209](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/209) | feat(milestone): 增加里程碑进度分析快捷命令 | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#205](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/205) | fix(issue): 修复详情缺失并保护更新元数据 | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#195](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/195) | feat(compare): 新增 compare 汇总与提交筛选能力 | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#190](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/190) | Add workflow release readiness gate | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#189](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/189) | Add workflow duplicate issue detection | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#188](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/188) | Add workflow dependency risk audit | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#165](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/165) | feat(issue): add batch maintenance shortcuts | wangyue111 | 85 | PASS | — | ✓/✗/✓ | +| [#159](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/159) | Add member application workflow shortcuts | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#77](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/77) | feat(journal): add issue and PR comment shortcut | wangyue111 | 85 | PASS | — | ✓/✗/✓ | +| [#150](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/150) | 新增 Issue 批量导出命令 | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#142](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/142) | 新增 PR 本地检出命令 | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#100](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/100) | 查看指定时间范围的开发统计 | jiangtx | 85 | PASS | — | ✗/✗/✓ | +| [#101](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/101) | 查看用户项目动态 | jiangtx | 85 | PASS | — | ✗/✗/✓ | +| [#21](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/21) | feat: add attachment shortcut group | Mengz | 85 | PASS | — | ✓/✗/✓ | +| [#139](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/139) | feat(wiki): 新增 Wiki 页面与目录管理 Shortcuts | whale | 82 | COMMENT | — | ✓/✗/✓ | +| [#130](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/130) | feat(workflows): 新增 gitlink-flow 社区运营自动化端到端工作流 | Ct201314 | 82 | COMMENT | — | ✓/✗/✓ | +| [#97](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/97) | 基础设施修复 | jiangtx | 81 | COMMENT | — | ✗/✗/✓ | +| [#123](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/123) | 新增 Release 资产下载命令 | Mengz | 80 | COMMENT | — | ✗/✗/✓ | +| [#103](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/103) | feat: 新建 pm 模块，添加 6 条项目管理命令 | wyxttn | 78 | COMMENT | — | — | +| [#131](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/131) | 子赛题三 - Java-Gatekeeper 端到端自动化质量门禁工作流 | xxxx12 | 75 | REQUEST_CHANGES | require_tests_for_src_changes | ✓/✓/✓ | +| [#30](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/30) | 增加wiki管理的shortcut | camelliamc | 74 | COMMENT | — | — | +| [#146](https://www.gitlink.org.cn/Gitlink/gitlink-cli/pulls/146) | feat: 新增 Showcase Dashboard 交互式展示页 | chroe | 70 | REQUEST_CHANGES | require_tests_for_src_changes | ✓/✗/✓ | + +## 这份报告说明了什么 + +- 同一份 `gatekeeper.yaml` 策略可以**无人值守地体检一个真实活跃仓库的全部积压**——确定性评分意味着大规模治理零 AI 成本，AI 只在需要语义判断（review_findings）时按需介入。 +- 任何人重跑本报告（`python3 scripts/gatekeeper_sweep.py`）会对同一组 PR 得到同样的分数与裁决。 diff --git a/examples/workflows/pr-quality-gatekeeper/findings.example.json b/examples/workflows/pr-quality-gatekeeper/findings.example.json new file mode 100644 index 0000000..611c12e --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/findings.example.json @@ -0,0 +1,23 @@ +{ + "_comment": "对 Gitlink/gitlink-cli PR #15222（feat/org-team-projects，head bcc27bf）真实 diff 的人工+AI 代码审查发现。每条均可在该 PR 的 shortcuts/org/org.go 对应行号复核——非样例数据。", + "findings": [ + { + "severity": "major", + "file": "shortcuts/org/org.go", + "line": 156, + "message": "parseBool 只把字面 \"true\" 当真：用户传 --dry-run=1 / t / TRUE 以外写法会被静默解析为 false。该 flag 守护的是 team-projects-remove-all（批量移除团队全部项目）这类破坏性操作——预览意图被静默降级为真实执行。建议改用 strconv.ParseBool（与 shortcuts/common/runner.go:52 解析 flag 默认值的行为一致），无法识别的值应报错而非吞掉" + }, + { + "severity": "minor", + "file": "shortcuts/org/org.go", + "line": 103, + "message": "team-projects-remove-all 一条命令清空团队全部项目绑定，除 --dry-run 外无确认机制；建议在 Description/help 标注危险性，或要求显式 --yes 二次确认" + }, + { + "severity": "minor", + "file": "shortcuts/org/org.go", + "line": 142, + "message": "dry-run 预览 payload 中 action（add_all_team_projects/remove_all_team_projects）与实际请求 path 段（create_all/destroy_all，见 :134）是两套词汇，排查问题时易误导；建议统一或在 payload 同时给出两者映射" + } + ] +} diff --git a/examples/workflows/pr-quality-gatekeeper/owner-rules.example.yaml b/examples/workflows/pr-quality-gatekeeper/owner-rules.example.yaml new file mode 100644 index 0000000..4615cad --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/owner-rules.example.yaml @@ -0,0 +1,62 @@ +# owner-rules.yaml — 文件路径 → reviewer 路由表（gitlink-gatekeeper 工作流步骤 1） +# +# 作用：PR 看门人闭环的第一步「路由」。脚本拉取 PR 变更文件后，按下面的 +# glob 规则把每个文件映射到建议 reviewer，写进评分卡的「Suggested +# reviewers」分区（REQUEST_CHANGES 时也写进 tracking issue）。 +# +# 语义（见 scripts/gatekeeper_workflow.py route_reviewers）： +# - rules 按顺序匹配，**首个命中的规则生效**（顺序即优先级，把更具体的放前面）。 +# - glob 用 Python fnmatch 语法（* 不跨目录段时也会匹配 /，与 fnmatch 行为一致）。 +# - 一个文件命中后不再继续匹配后续规则；多个 reviewer 写在同一规则的 reviewers 列表里。 +# - 未命中任何规则的文件归到 default_reviewers（兜底）。 +# +# 排序示例（首个命中即生效、顺序即优先级，请按需排序）： +# 下面把目录 glob `skills/**` 放在语言 glob `**/*.go` 之前，于是 `skills/foo.go` +# 会先命中 `skills/**` → 路由给 skill-owner，而不会落到 go-reviewer。若你希望 +# skills 下的 Go 文件仍由 go-reviewer 审，就把语言规则提到目录规则之前 +# （或在目录规则里收窄 glob，如 `skills/**/*.md`）。 +# +# 注意：本表只产出「建议」，不调用任何写操作；真正分配 reviewer 由维护者在 +# PR 页面决定。gatekeeper 不替人点提交（SSOT 第 8 节安全规则）。 +# +# reviewers 填 GitLink 用户名（login）。下方为占位示例，真实使用时替换为本仓库的维护者。 + +rules: + # —— 文档：只改文档走文档维护者，避免占用代码 reviewer —— + - glob: "docs/**" + reviewers: ["doc-maintainer"] + - glob: "**/*.md" + reviewers: ["doc-maintainer"] + + # —— 工作流 / 脚本本体 —— + - glob: "workflow/**" + reviewers: ["workflow-owner"] + - glob: "skills/**" + reviewers: ["skill-owner"] + + # —— 按语言路由到对应方向的 reviewer —— + - glob: "**/*_test.go" + reviewers: ["go-reviewer", "qa-reviewer"] + - glob: "**/*.go" + reviewers: ["go-reviewer"] + - glob: "test_*.py" + reviewers: ["py-reviewer", "qa-reviewer"] + - glob: "**/*.py" + reviewers: ["py-reviewer"] + - glob: "**/*.ts" + reviewers: ["fe-reviewer"] + - glob: "**/*.js" + reviewers: ["fe-reviewer"] + + # —— 高敏感区：CI / 依赖 / 安全配置，强制资深 reviewer —— + - glob: ".gitea/**" + reviewers: ["ci-owner", "security-reviewer"] + - glob: "**/Dockerfile" + reviewers: ["ci-owner"] + - glob: "go.mod" + reviewers: ["security-reviewer"] + - glob: "go.sum" + reviewers: ["security-reviewer"] + +# 未命中上面任何规则的文件，兜底分配给这些人 +default_reviewers: ["maintainer"] diff --git a/examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_sweep.py b/examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_sweep.py new file mode 100644 index 0000000..e0a04a6 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_sweep.py @@ -0,0 +1,227 @@ +# SPDX-License-Identifier: MulanPSL-2.0 +"""gatekeeper_sweep —— 对一个仓库的全部 open PR 批量跑门禁（只读 dry-run），出治理报告。 + +把单 PR 的「策略 → 评分卡 → 裁决」升级为仓库级体检： + 1. 翻页拉取 PR 列表，筛出 open； + 2. 逐个调用 gatekeeper_workflow.py（强制 dry-run，绝不 --apply，对远端零写入）； + 3. 汇总每个 PR 的 summary.json → 聚合统计 + 全量明细表 → sweep-report.md / sweep-summary.json。 + +诚实口径：批扫不注入 AI 审查发现（--findings），review_findings 维按 0 发现计满分， +报告中明确标注「该维度未评」；其余 4 维（测试/卫生/commit/CI）为真实采集结果。 +纯标准库，无第三方依赖。 +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +import time +import urllib.request +from pathlib import Path +from typing import Any + +API_BASE = "https://www.gitlink.org.cn/api" +HYGIENE_RE = re.compile(r"desc (✓|✗) / linked issue (✓|✗) / size (✓|✗)") + + +def fetch_open_prs(owner: str, repo: str, limit_pages: int = 20) -> list[dict[str, Any]]: + """翻页拉取 PR 列表并筛出 open（列表接口的 status 参数不可靠，按字段过滤）。""" + items: list[dict[str, Any]] = [] + page = 1 + while page <= limit_pages: + url = f"{API_BASE}/{owner}/{repo}/pulls.json?page={page}&limit=50" + with urllib.request.urlopen(url, timeout=30) as resp: + data = json.loads(resp.read().decode("utf-8")) + batch = data.get("issues") or [] + if not batch: + break + items.extend(batch) + if len(items) >= int(data.get("search_count") or 0): + break + page += 1 + return [it for it in items if it.get("pull_request_staus") == "open"] + + +def run_one( + workflow_script: Path, + owner: str, + repo: str, + number: int, + policy: Path, + owner_rules: Path, + cli_bin: str, + out_dir: Path, +) -> dict[str, Any]: + """对单个 PR 跑一次 dry-run 门禁，返回解析后的行记录（失败不抛，记 error）。""" + cmd = [ + sys.executable, + str(workflow_script), + "--owner", owner, + "--repo", repo, + "--pr", str(number), + "--policy", str(policy), + "--owner-rules", str(owner_rules), + "--cli-bin", cli_bin, + "--skip-ci", + "--output-dir", str(out_dir), + ] + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=180) + slug = f"{owner}_{repo}_pr{number}".replace("/", "_") + summary_path = out_dir / f"{slug}_summary.json" + if proc.returncode == 1 or not summary_path.exists(): + return {"number": number, "error": (proc.stderr or proc.stdout)[-200:].strip()} + summary = json.loads(summary_path.read_text(encoding="utf-8")) + hygiene = "" + scorecard_path = out_dir / f"{slug}_scorecard.md" + if scorecard_path.exists(): + m = HYGIENE_RE.search(scorecard_path.read_text(encoding="utf-8")) + if m: + hygiene = "/".join(m.groups()) # 例如 "✓/✗/✓"：描述/关联issue/体量 + return { + "number": number, + "verdict": summary.get("verdict"), + "total": summary.get("total"), + "scores": summary.get("scores", {}), + "hard_gate_failures": [f.get("gate") if isinstance(f, dict) else f + for f in summary.get("hard_gate_failures", [])], + "hygiene": hygiene, + "suggested_reviewers": summary.get("routing", {}).get("suggested_reviewers", []), + } + + +def aggregate(rows: list[dict[str, Any]]) -> dict[str, Any]: + ok = [r for r in rows if "error" not in r] + totals = sorted(r["total"] for r in ok) + verdicts: dict[str, int] = {} + gate_hits: dict[str, int] = {} + for r in ok: + verdicts[r["verdict"]] = verdicts.get(r["verdict"], 0) + 1 + for g in r["hard_gate_failures"]: + gate_hits[str(g)] = gate_hits.get(str(g), 0) + 1 + def pct(n: int) -> str: + return f"{100 * n / len(ok):.0f}%" if ok else "0%" + no_linked = sum(1 for r in ok if r["hygiene"] and r["hygiene"].split("/")[1] == "✗") + zero_cov = sum(1 for r in ok if r["scores"].get("test_coverage") == 0) + return { + "scanned": len(rows), + "succeeded": len(ok), + "failed": len(rows) - len(ok), + "verdicts": verdicts, + "score_min": totals[0] if totals else None, + "score_median": totals[len(totals) // 2] if totals else None, + "score_avg": round(sum(totals) / len(totals), 1) if totals else None, + "score_max": totals[-1] if totals else None, + "hard_gate_hits": gate_hits, + "pct_zero_test_coverage": pct(zero_cov), + "pct_no_linked_issue": pct(no_linked), + "pct_request_changes": pct(verdicts.get("REQUEST_CHANGES", 0)), + } + + +def render_report( + owner: str, repo: str, policy_label: str, date_label: str, + rows: list[dict[str, Any]], agg: dict[str, Any], + pr_meta: dict[int, dict[str, Any]], +) -> str: + ok = [r for r in rows if "error" not in r] + lines = [ + f"# gatekeeper 仓库体检报告 —— {owner}/{repo}（{date_label}）", + "", + f"> 对 **{agg['scanned']} 个 open PR** 全量 dry-run（**只读，零写入**）· 策略 `{policy_label}` · " + f"成功 {agg['succeeded']} / 失败 {agg['failed']}", + ">", + "> **诚实口径**：批扫未注入 AI 审查发现，review_findings 维按 0 发现计满分（**该维度未评**）；" + "CI 维按 `--skip-ci` 统一记 unknown（半分）。其余维度为真实采集。" + "因此**总分代表「除人工/AI 审查外的工程卫生分」，偏乐观**；裁决分布同理。", + "", + "## 总览", + "", + f"- 裁决分布：{' · '.join(f'{k} **{v}**' for k, v in sorted(agg['verdicts'].items()))}", + f"- 分数：min {agg['score_min']} / 中位 {agg['score_median']} / 均值 {agg['score_avg']} / max {agg['score_max']}", + f"- **{agg['pct_zero_test_coverage']}** 的 PR 测试覆盖维 0 分（改动不带任何测试）", + f"- **{agg['pct_no_linked_issue']}** 的 PR 未关联 issue", + f"- **{agg['pct_request_changes']}** 的 PR 触发 REQUEST_CHANGES（硬门禁或低分）", + "", + "硬门禁命中：" + ("；".join(f"`{g}` × {n}" for g, n in sorted(agg["hard_gate_hits"].items(), key=lambda x: -x[1])) or "无"), + "", + "## 全量明细（按分数降序）", + "", + "| PR | 标题 | 作者 | 总分 | 裁决 | 硬门禁失败 | 卫生(描述/关联/体量) |", + "|----|------|------|-----:|------|-----------|---------------------|", + ] + for r in sorted(ok, key=lambda x: -x["total"]): + meta = pr_meta.get(r["number"], {}) + title = str(meta.get("name", ""))[:48].replace("|", "\\|") + gates = ", ".join(str(g) for g in r["hard_gate_failures"]) or "—" + lines.append( + f"| [#{r['number']}](https://www.gitlink.org.cn/{owner}/{repo}/pulls/{r['number']}) " + f"| {title} | {meta.get('author_name', '?')} | {r['total']} | {r['verdict']} | {gates} | {r['hygiene'] or '—'} |" + ) + errs = [r for r in rows if "error" in r] + if errs: + lines += ["", "## 跑失败的 PR", ""] + lines += [f"- #{r['number']}：`{r['error']}`" for r in errs] + lines += [ + "", + "## 这份报告说明了什么", + "", + "- 同一份 `gatekeeper.yaml` 策略可以**无人值守地体检一个真实活跃仓库的全部积压**——" + "确定性评分意味着大规模治理零 AI 成本，AI 只在需要语义判断（review_findings）时按需介入。", + "- 任何人重跑本报告（`python3 scripts/gatekeeper_sweep.py`）会对同一组 PR 得到同样的分数与裁决。", + "", + ] + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="对全部 open PR 批量 dry-run 出治理报告") + parser.add_argument("--owner", default="Gitlink") + parser.add_argument("--repo", default="gitlink-cli") + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--owner-rules", dest="owner_rules", type=Path, required=True) + parser.add_argument("--cli-bin", default="gitlink-cli") + parser.add_argument("--output-dir", type=Path, default=Path("sweep-outputs")) + parser.add_argument("--date-label", default="sweep", help="报告日期标签（可复现：不取系统时间）") + parser.add_argument("--max", type=int, default=0, help="只跑前 N 个（0=全量），用于试跑") + parser.add_argument("--sleep", type=float, default=0.2, help="相邻 PR 间隔秒数（对平台礼貌）") + args = parser.parse_args(argv) + + workflow_script = Path(__file__).with_name("gatekeeper_workflow.py") + runs_dir = args.output_dir / "runs" + runs_dir.mkdir(parents=True, exist_ok=True) + + prs = fetch_open_prs(args.owner, args.repo) + if args.max: + prs = prs[: args.max] + pr_meta = {int(p["pull_request_number"]): p for p in prs} + print(f"open PR 共 {len(prs)} 个，开始批扫（dry-run，零写入）…", flush=True) + + rows: list[dict[str, Any]] = [] + for i, p in enumerate(prs, 1): + number = int(p["pull_request_number"]) + row = run_one(workflow_script, args.owner, args.repo, number, + args.policy, args.owner_rules, args.cli_bin, runs_dir) + rows.append(row) + tag = row.get("verdict", "ERROR") + print(f"[{i}/{len(prs)}] PR #{number} → {tag} {row.get('total', '')}", flush=True) + time.sleep(args.sleep) + + agg = aggregate(rows) + policy_label = args.policy.name + report = render_report(args.owner, args.repo, policy_label, args.date_label, rows, agg, pr_meta) + (args.output_dir / "sweep-report.md").write_text(report, encoding="utf-8") + (args.output_dir / "sweep-summary.json").write_text( + json.dumps({"aggregate": agg, "rows": rows}, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + print(f"\n报告：{args.output_dir / 'sweep-report.md'}") + print(f"汇总：{args.output_dir / 'sweep-summary.json'}") + print(f"裁决分布：{agg['verdicts']} · 均分 {agg['score_avg']}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_workflow.py b/examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_workflow.py new file mode 100644 index 0000000..601fc86 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/scripts/gatekeeper_workflow.py @@ -0,0 +1,1320 @@ +#!/usr/bin/env python3 +"""gitlink-gatekeeper 端到端「PR 看门人闭环」编排脚本。 + +把 gitlink-gatekeeper（Policy-as-Code PR 合并门禁）包成一条可复现的闭环，串联 +SSOT（docs/design.md）第 9 节定义的三步： + + 1. 路由：按变更文件路径匹配 owner-rules.yaml 的 glob，推导建议 reviewer。 + 2. 裁决：采集 PR 上下文 → 按 SSOT 第 3-5 节确定性算法算评分卡 + 三态裁决。 + 3. 回写 + 善后：评分卡作为评论回写 PR + 打裁决标签；若 REQUEST_CHANGES， + 自动创建一条 tracking issue 汇总必修项并回链 PR。 + +设计要点（与 SSOT 第 8 节安全规则一致）： + - **默认 dry-run**：不传 --apply 时只打印将要做什么，绝不写任何东西。 + - 写操作（回写评论 / 打标签 / 建 issue / 合并）一律需要显式 --apply。 + - **绝不默认自动合并**：仅当策略 behavior.auto_merge=true 且裁决=PASS 且 + 命令显式带 --apply 才会合并。 + - 纯 Python 标准库，无第三方依赖（含内置的 YAML 子集解析器）。 + +CLI 命令映射见 SSOT 第 7 节；所引用的 gitlink-cli 命令均已在 +upstream-gitlink-cli/shortcuts/ 下核验存在： + pr +view / pr +files / pr +diff / pr +comment / pr +review / pr +merge + issue +create / issue +comment + label +create / label +list + ci +builds +向 PR / issue 挂标签没有独立 shortcut，需走 Raw API + POST /:owner/:repo/issues/:id --body '{"issue_tag_ids":[...], ...}' +（PR 底层关联一个 issue，标签即挂在该 issue 上——与 gitlink-code-review 工作流 3 一致）。 +挂载行为受 --apply 显式门控：dry-run 只预览「将把标签挂到 PR」，仅在 --apply 时 +才真正发起 POST；回写 body 会带上原 subject/description，避免清空 PR 标题/描述。 +""" + +# SPDX-License-Identifier: MulanPSL-2.0 + +from __future__ import annotations + +import argparse +import fnmatch +import json +import re +import subprocess +import sys +from dataclasses import dataclass, field +from pathlib import Path +from shutil import which +from typing import Any, Iterable + + +# --------------------------------------------------------------------------- # +# 常量 / 异常 +# --------------------------------------------------------------------------- # + +SEVERITY_ORDER = ("blocker", "major", "minor", "nit") + +# SSOT 第 2 节「内置默认策略」——找不到 gatekeeper.yaml 时回退到这里。 +DEFAULT_POLICY: dict[str, Any] = { + "version": 1, + "weights": { + "review_findings": 40, + "test_coverage": 20, + "pr_hygiene": 15, + "commit_quality": 15, + "ci_status": 10, + }, + "hard_gates": { + "forbid_blocker_findings": True, + "require_ci_pass": True, + "require_tests_for_src_changes": True, + "require_linked_issue": False, + "max_changed_files": 80, + }, + "severity_penalty": {"blocker": 100, "major": 25, "minor": 5, "nit": 1}, + "thresholds": {"pass": 85, "request_changes": 60}, + "labels": { + "pass": "gatekeeper:pass", + "request_changes": "gatekeeper:needs-changes", + "comment": "gatekeeper:review", + }, + "source_globs": ["**/*.go", "**/*.py", "**/*.js", "**/*.ts", "**/*.rs", "**/*.java"], + "test_globs": ["**/*_test.go", "**/test_*.py", "**/*.test.*", "**/*.spec.*", "tests/**"], + "behavior": { + "dry_run_default": True, + "post_comment": True, + "apply_label": True, + "auto_merge": False, + "merge_method": "squash", + }, +} + +VERDICT_EMOJI = {"PASS": "✅", "REQUEST_CHANGES": "❌", "COMMENT": "💬"} + + +class WorkflowError(RuntimeError): + """工作流可预期的失败（缺配置、CLI 不存在、采集失败等）。""" + + +# --------------------------------------------------------------------------- # +# 极简 YAML 子集解析器（纯标准库，覆盖本工作流配置所需的语法） +# +# 支持：标量、嵌套映射（缩进）、`key: [a, b]` 行内列表、`- item` 块列表、 +# `#` 注释、true/false/整数/带引号字符串。 +# 不支持：锚点、多文档、多行字符串——本项目配置不需要。 +# --------------------------------------------------------------------------- # + +def _parse_scalar(token: str) -> Any: + token = token.strip() + if token == "" or token == "~" or token.lower() == "null": + return None + if (token.startswith('"') and token.endswith('"')) or ( + token.startswith("'") and token.endswith("'") + ): + return token[1:-1] + low = token.lower() + if low == "true": + return True + if low == "false": + return False + try: + return int(token) + except ValueError: + pass + try: + return float(token) + except ValueError: + return token + + +def _parse_inline_list(token: str) -> list[Any]: + inner = token.strip()[1:-1].strip() + if not inner: + return [] + # 朴素逗号切分；配置里的列表元素不含逗号，足够。 + return [_parse_scalar(part) for part in inner.split(",")] + + +def _strip_comment(line: str) -> str: + in_single = in_double = False + for idx, ch in enumerate(line): + if ch == "'" and not in_double: + in_single = not in_single + elif ch == '"' and not in_single: + in_double = not in_double + elif ch == "#" and not in_single and not in_double: + return line[:idx] + return line + + +def _parse_value_token(rest: str) -> Any: + """解析 `key:` 右侧的标量 / 行内列表。""" + if rest.startswith("["): + return _parse_inline_list(rest) + return _parse_scalar(rest) + + +def _clean_lines(text: str) -> list[tuple[int, str]]: + """返回 [(indent, stripped_content)]，已去注释 / 空行。""" + out: list[tuple[int, str]] = [] + for raw in text.splitlines(): + line = _strip_comment(raw).rstrip() + if not line.strip(): + continue + indent = len(line) - len(line.lstrip(" ")) + out.append((indent, line.strip())) + return out + + +def _parse_block(lines: list[tuple[int, str]], pos: int, indent: int) -> tuple[Any, int]: + """递归下降解析一个缩进块，返回 (value, next_pos)。 + + 根据块内第一行判断是 list（`- ...`）还是 dict（`key: ...`）。 + 支持「列表项是映射」（`- glob: x` 后跟同级缩进的 `reviewers: [...]`）。 + """ + if pos >= len(lines): + return {}, pos + + first_indent = lines[pos][0] + is_list = lines[pos][1].startswith("- ") + container: Any = [] if is_list else {} + + while pos < len(lines): + cur_indent, content = lines[pos] + if cur_indent < first_indent: + break + + if is_list: + if not content.startswith("- "): + break + item_body = content[2:].strip() + # 列表项内可能直接带 key: value（即列表元素是映射） + if ":" in item_body and not item_body.startswith(("[", '"', "'")): + key, _, rest = item_body.partition(":") + key, rest = key.strip(), rest.strip() + # 该列表元素起始的虚拟缩进 = 列表项内容的列位置 + inner_indent = cur_indent + 2 + item_map: dict[str, Any] = {} + if rest == "": + pos += 1 + sub, pos = _parse_block(lines, pos, inner_indent + 1) + item_map[key] = sub + else: + item_map[key] = _parse_value_token(rest) + pos += 1 + # 收编后续属于同一列表元素的 key（缩进 >= inner_indent 且不是新 `- `） + while pos < len(lines): + nxt_indent, nxt_content = lines[pos] + if nxt_indent < inner_indent or nxt_content.startswith("- "): + break + if ":" not in nxt_content: + break + k2, _, r2 = nxt_content.partition(":") + k2, r2 = k2.strip(), r2.strip() + if r2 == "": + pos += 1 + sub2, pos = _parse_block(lines, pos, nxt_indent + 1) + item_map[k2] = sub2 + else: + item_map[k2] = _parse_value_token(r2) + pos += 1 + container.append(item_map) + else: + container.append(_parse_scalar(item_body)) + pos += 1 + else: + if content.startswith("- ") or ":" not in content: + break + key, _, rest = content.partition(":") + key, rest = key.strip(), rest.strip() + if rest == "": + pos += 1 + sub, pos = _parse_block(lines, pos, cur_indent + 1) + container[key] = sub + else: + container[key] = _parse_value_token(rest) + pos += 1 + + return container, pos + + +def parse_simple_yaml(text: str) -> dict[str, Any]: + """解析本工作流所需的 YAML 子集，返回嵌套 dict。 + + 支持：嵌套映射、块列表、「列表项是映射」、行内列表、标量、注释。 + 不支持：锚点 / 多文档 / 多行字符串 / 复杂流式语法（本项目配置不需要）。 + """ + lines = _clean_lines(text) + if not lines: + return {} + value, _ = _parse_block(lines, 0, lines[0][0]) + if not isinstance(value, dict): + # 顶层是列表的情况：包一层（本项目顶层均为映射，保险处理） + return {"_root": value} + return value + + +# --------------------------------------------------------------------------- # +# 配置加载 +# --------------------------------------------------------------------------- # + +def load_yaml_file(path: Path) -> dict[str, Any]: + if not path.exists(): + raise WorkflowError(f"找不到配置文件：{path}") + return parse_simple_yaml(path.read_text(encoding="utf-8")) + + +def deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + """把 override 合并到 base 的副本上（用于策略覆盖默认值）。""" + result = json.loads(json.dumps(base)) # 深拷贝 + for key, value in (override or {}).items(): + if isinstance(value, dict) and isinstance(result.get(key), dict): + result[key] = deep_merge(result[key], value) + else: + result[key] = value + return result + + +def load_policy(policy_path: Path | None) -> dict[str, Any]: + """加载 gatekeeper.yaml；缺失时回退内置默认策略（SSOT 第 2 节）。""" + if policy_path is None or not policy_path.exists(): + return json.loads(json.dumps(DEFAULT_POLICY)) + user_policy = load_yaml_file(policy_path) + merged = deep_merge(DEFAULT_POLICY, user_policy) + _validate_policy(merged) + return merged + + +def _validate_policy(policy: dict[str, Any]) -> None: + weights = policy.get("weights", {}) + total = sum(int(v) for v in weights.values()) + if total != 100: + raise WorkflowError( + f"策略非法：weights 之和必须为 100，当前为 {total}（{weights}）" + ) + + +# --------------------------------------------------------------------------- # +# gitlink-cli 调用层 +# --------------------------------------------------------------------------- # + +def cli_available(cli_bin: str) -> bool: + return which(cli_bin) is not None + + +def run_cli_json( + cli_bin: str, + args: list[str], + owner: str, + repo: str, +) -> Any: + """运行只读 gitlink-cli 命令并解析 JSON 输出（仅采集步骤用）。""" + cmd = [cli_bin, *args, "--owner", owner, "--repo", repo, "--format", "json"] + proc = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8") + if proc.returncode != 0: + stderr = (proc.stderr or proc.stdout or "未知错误").strip() + raise WorkflowError(f"命令失败：{' '.join(cmd)}\n{stderr}") + return _parse_cli_json(proc.stdout) + + +def _parse_cli_json(text: str) -> Any: + stripped = (text or "").strip() + if not stripped: + return {} + try: + return json.loads(stripped) + except json.JSONDecodeError: + idxs = [i for i in (stripped.find("{"), stripped.find("[")) if i != -1] + if idxs: + return json.loads(stripped[min(idxs):]) + raise WorkflowError(f"无法解析 CLI JSON 输出：{stripped[:120]}") + + +@dataclass +class PlannedWrite: + """一个待执行的写操作（dry-run 时只打印，apply 时执行）。""" + + label: str # 人类可读说明 + command: list[str] # gitlink-cli 子命令（不含 --owner/--repo/--format） + note: str = "" # 备注（如 body 摘要） + + +def render_planned(write: PlannedWrite, owner: str, repo: str, cli_bin: str) -> str: + safe_cmd = " ".join([cli_bin, *write.command, "--owner", owner, "--repo", repo]) + # 折叠超长 body，避免刷屏 + return f" • {write.label}\n $ {safe_cmd[:400]}" + + +def execute_write( + write: PlannedWrite, + owner: str, + repo: str, + cli_bin: str, +) -> dict[str, Any]: + cmd = [cli_bin, *write.command, "--owner", owner, "--repo", repo, "--format", "json"] + proc = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8") + ok = proc.returncode == 0 + return { + "label": write.label, + "ok": ok, + "stderr": (proc.stderr or "").strip() if not ok else "", + "data": _parse_cli_json(proc.stdout) if ok and proc.stdout.strip() else None, + } + + +# --------------------------------------------------------------------------- # +# 数据归一化（兼容 GitLink Envelope 的多种字段名） +# --------------------------------------------------------------------------- # + +def unwrap(payload: Any) -> Any: + """剥掉 Envelope 的 data 外层。""" + if isinstance(payload, dict) and "data" in payload and "ok" in payload: + return payload["data"] + return payload + + +def first_value(item: dict[str, Any], keys: Iterable[str], default: Any = None) -> Any: + for key in keys: + if isinstance(item, dict) and key in item and item[key] not in (None, "", []): + return item[key] + return default + + +def extract_first_list(payload: Any, keys: Iterable[str]) -> list[Any]: + if isinstance(payload, list): + return payload + if isinstance(payload, dict): + for key in keys: + if isinstance(payload.get(key), list): + return payload[key] + for value in payload.values(): + found = extract_first_list(value, keys) + if found: + return found + return [] + + +def extract_file_paths(files_payload: Any) -> list[str]: + data = unwrap(files_payload) + items = extract_first_list(data, ("files", "diff", "entries", "items", "list")) + paths: list[str] = [] + for item in items: + if isinstance(item, dict): + p = first_value(item, ("path", "filename", "new_path", "name", "filepath")) + if p: + paths.append(str(p)) + elif isinstance(item, str): + paths.append(item) + return paths + + +# --------------------------------------------------------------------------- # +# 步骤 1：路由（owner-rules.yaml glob → reviewer） +# --------------------------------------------------------------------------- # + +def load_owner_rules(path: Path) -> tuple[list[dict[str, Any]], list[str]]: + """读取 owner-rules.yaml，返回 (rules, default_reviewers)，rules 为 [{glob, reviewers:[...]}, ...]（保序）。""" + if not path.exists(): + raise WorkflowError(f"找不到 owner-rules：{path}") + parsed = parse_simple_yaml(path.read_text(encoding="utf-8")) + rules_raw = parsed.get("rules", []) + rules: list[dict[str, Any]] = [] + if isinstance(rules_raw, list): + for entry in rules_raw: + if not isinstance(entry, dict): + continue + glob = entry.get("glob") or entry.get("path") + reviewers = entry.get("reviewers") + if isinstance(reviewers, str): + reviewers = [reviewers] + if glob and reviewers: + rules.append({"glob": str(glob), "reviewers": list(reviewers)}) + fallback = parsed.get("default_reviewers") or parsed.get("fallback") or [] + if isinstance(fallback, str): + fallback = [fallback] + return rules, list(fallback) + + +def route_reviewers( + changed_files: list[str], + rules: list[dict[str, Any]], + fallback: list[str], +) -> dict[str, Any]: + """把变更文件映射到 reviewer。返回 reviewer→匹配文件，及未命中文件。""" + assignments: dict[str, list[str]] = {} + matched: set[str] = set() + for path in changed_files: + for rule in rules: + if glob_match(path, rule["glob"]): + matched.add(path) + for reviewer in rule["reviewers"]: + assignments.setdefault(reviewer, []).append(path) + break # 首个命中规则生效（规则顺序即优先级） + unmatched = [p for p in changed_files if p not in matched] + if unmatched and fallback: + for reviewer in fallback: + assignments.setdefault(reviewer, []).extend(unmatched) + return { + "assignments": {k: sorted(set(v)) for k, v in assignments.items()}, + "unmatched": unmatched, + "suggested_reviewers": sorted(assignments.keys()), + } + + +# --------------------------------------------------------------------------- # +# 步骤 2：裁决（采集 + 评分，复用 SSOT 第 3-5 节确定性算法） +# --------------------------------------------------------------------------- # + +def glob_match(path: str, pattern: str) -> bool: + """fnmatch 包装：让 `**/X` 也能匹配顶层文件 `X`（fnmatch 默认要求至少一段目录）。 + + 这与 SSOT source_globs/test_globs（如 `**/*.go`）的直觉一致：`main.go` + 位于仓库根目录也应被视为源码。 + """ + if fnmatch.fnmatch(path, pattern): + return True + if pattern.startswith("**/"): + return fnmatch.fnmatch(path, pattern[3:]) + return False + + +def matches_any(path: str, globs: list[str]) -> bool: + return any(glob_match(path, g) for g in globs) + + +def count_src_test(changed_files: list[str], policy: dict[str, Any]) -> tuple[int, int]: + # test_globs 优先：同时匹配 source 与 test 的文件只计为 test，不计为 src（SSOT §3.2） + tests = sum(1 for p in changed_files if matches_any(p, policy["test_globs"])) + src = sum( + 1 + for p in changed_files + if matches_any(p, policy["source_globs"]) and not matches_any(p, policy["test_globs"]) + ) + return src, tests + + +CONVENTIONAL_RE = re.compile( + r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)(\([^)]+\))?!?:\s+\S+" +) + + +def count_conventional_commits(commits: list[str]) -> tuple[int, int]: + if not commits: + return 0, 0 + conforming = sum(1 for msg in commits if CONVENTIONAL_RE.match((msg or "").strip())) + return conforming, len(commits) + + +@dataclass +class Finding: + severity: str + message: str + file: str = "" + line: Any = "" + + +@dataclass +class ScoreInput: + pr_id: str + title: str + description: str + changed_files: list[str] + changed_src: int + changed_tests: int + commits: list[str] + ci_status: str # passing / failing / unknown + linked_issue: bool + findings: list[Finding] = field(default_factory=list) + + +def score_dimensions(inp: ScoreInput, policy: dict[str, Any]) -> dict[str, Any]: + """按 SSOT 第 3 节逐维评分。返回每维得分 + 备注。""" + w = policy["weights"] + sev = policy["severity_penalty"] + max_changed = int(policy["hard_gates"].get("max_changed_files", 0)) + + # 3.1 review_findings + penalty = sum(int(sev.get(f.severity, 0)) for f in inp.findings) + rf_w = w["review_findings"] + review_score = round(rf_w * max(0.0, 1 - penalty / rf_w)) if rf_w else 0 + + # 3.2 test_coverage + tc_w = w["test_coverage"] + if inp.changed_src == 0: + test_score = tc_w + elif inp.changed_tests == 0: + test_score = 0 + else: + ratio = min(1.0, inp.changed_tests / inp.changed_src) + test_score = round(tc_w * (0.5 + 0.5 * ratio)) + + # 3.3 pr_hygiene（三项各 1/3） + hy_w = w["pr_hygiene"] + hits = 0.0 + desc_ok = bool(inp.description) and len(inp.description.strip()) >= 30 + if desc_ok: + hits += 1 / 3 + if inp.linked_issue: + hits += 1 / 3 + n_files = len(inp.changed_files) + if max_changed == 0 or n_files <= max_changed / 2: + size_credit, size_mark = 1 / 3, "✓" # 不限体量或体量适中 → 满分 + elif n_files <= max_changed: + size_credit, size_mark = 1 / 6, "~" # 偏大但未超上限 → 半分 + else: + size_credit, size_mark = 0.0, "✗" # 超上限（通常已被硬门禁拦截） + hits += size_credit + hygiene_score = round(hy_w * hits) + + # 3.4 commit_quality + cq_w = w["commit_quality"] + conforming, total = count_conventional_commits(inp.commits) + commit_score = cq_w if total == 0 else round(cq_w * conforming / total) + + # 3.5 ci_status + ci_w = w["ci_status"] + if inp.ci_status == "passing": + ci_score = ci_w + elif inp.ci_status == "failing": + ci_score = 0 + else: + ci_score = round(ci_w * 0.5) + + sev_counts = {s: sum(1 for f in inp.findings if f.severity == s) for s in SEVERITY_ORDER} + + return { + "review_findings": { + "score": review_score, + "weight": rf_w, + "note": " / ".join(f"{sev_counts[s]} {s}" for s in SEVERITY_ORDER), + }, + "test_coverage": { + "score": test_score, + "weight": tc_w, + "note": f"{inp.changed_src} src / {inp.changed_tests} test files", + }, + "pr_hygiene": { + "score": hygiene_score, + "weight": hy_w, + "note": f"desc {'✓' if desc_ok else '✗'} / " + f"linked issue {'✓' if inp.linked_issue else '✗'} / " + f"size {size_mark}", + }, + "commit_quality": { + "score": commit_score, + "weight": cq_w, + "note": f"{conforming}/{total} conventional", + }, + "ci_status": {"score": ci_score, "weight": ci_w, "note": inp.ci_status}, + "total": review_score + test_score + hygiene_score + commit_score + ci_score, + } + + +def evaluate_hard_gates(inp: ScoreInput, policy: dict[str, Any]) -> list[dict[str, str]]: + """按 SSOT 第 4 节逐项判定硬门禁，返回命中的失败列表。""" + gates = policy["hard_gates"] + failures: list[dict[str, str]] = [] + has_blocker = any(f.severity == "blocker" for f in inp.findings) + if gates.get("forbid_blocker_findings") and has_blocker: + failures.append({"gate": "forbid_blocker_findings", "detail": "存在 blocker 级审查发现"}) + # 仅在 CI 明确 failing 时触发；unknown/无 build 记录不触发（SSOT §4） + if gates.get("require_ci_pass") and inp.ci_status == "failing": + failures.append({"gate": "require_ci_pass", "detail": "CI 明确失败（failing）"}) + if ( + gates.get("require_tests_for_src_changes") + and inp.changed_src > 0 + and inp.changed_tests == 0 + ): + failures.append( + { + "gate": "require_tests_for_src_changes", + "detail": f"改动了 {inp.changed_src} 个源码文件，但本 PR 未包含任何测试文件", + } + ) + if gates.get("require_linked_issue") and not inp.linked_issue: + failures.append({"gate": "require_linked_issue", "detail": "PR 未关联任何 Issue"}) + max_changed = int(gates.get("max_changed_files", 0)) + if max_changed > 0 and len(inp.changed_files) > max_changed: + failures.append( + { + "gate": "max_changed_files", + "detail": f"变更文件 {len(inp.changed_files)} 超过上限 {max_changed}", + } + ) + return failures + + +def decide_verdict( + total: int, hard_gate_failed: bool, policy: dict[str, Any] +) -> str: + """SSOT 第 5 节裁决判定树。""" + th = policy["thresholds"] + if hard_gate_failed: + return "REQUEST_CHANGES" + if total >= int(th["pass"]): + return "PASS" + if total < int(th["request_changes"]): + return "REQUEST_CHANGES" + return "COMMENT" + + +# --------------------------------------------------------------------------- # +# 评分卡渲染（SSOT 第 6 节模板） +# --------------------------------------------------------------------------- # + +def render_scorecard( + inp: ScoreInput, + dims: dict[str, Any], + failures: list[dict[str, str]], + verdict: str, + policy_label: str, + routing: dict[str, Any] | None, + tracking_issue: Any = None, +) -> str: + emoji = VERDICT_EMOJI[verdict] + total = dims["total"] + lines: list[str] = [] + lines.append(f"## 🛡️ Gatekeeper Report — PR #{inp.pr_id} {inp.title}") + lines.append("") + lines.append( + f"**Verdict: {emoji} {verdict}** · Score: {total}/100 · policy: {policy_label}" + ) + lines.append("") + lines.append("| Dimension | Weight | Score | Notes |") + lines.append("|-----------|:------:|:-----:|-------|") + lines.append( + f"| Review findings | {dims['review_findings']['weight']} | " + f"{dims['review_findings']['score']}/{dims['review_findings']['weight']} | " + f"{dims['review_findings']['note']} |" + ) + lines.append( + f"| Test coverage | {dims['test_coverage']['weight']} | " + f"{dims['test_coverage']['score']}/{dims['test_coverage']['weight']} | " + f"{dims['test_coverage']['note']} |" + ) + lines.append( + f"| PR hygiene | {dims['pr_hygiene']['weight']} | " + f"{dims['pr_hygiene']['score']}/{dims['pr_hygiene']['weight']} | " + f"{dims['pr_hygiene']['note']} |" + ) + lines.append( + f"| Commit quality | {dims['commit_quality']['weight']} | " + f"{dims['commit_quality']['score']}/{dims['commit_quality']['weight']} | " + f"{dims['commit_quality']['note']} |" + ) + lines.append( + f"| CI status | {dims['ci_status']['weight']} | " + f"{dims['ci_status']['score']}/{dims['ci_status']['weight']} | " + f"{dims['ci_status']['note']} |" + ) + lines.append("") + + if routing and routing.get("suggested_reviewers"): + lines.append(f"### 👥 Suggested reviewers ({len(routing['suggested_reviewers'])})") + for reviewer, paths in routing["assignments"].items(): + sample = ", ".join(paths[:3]) + (" …" if len(paths) > 3 else "") + lines.append(f"- @{reviewer} — {len(paths)} file(s): {sample}") + lines.append("") + + if failures: + lines.append(f"### ⛔ Hard gate failures ({len(failures)})") + for f in failures: + lines.append(f"- `{f['gate']}`: {f['detail']}") + lines.append("") + + must = [f for f in inp.findings if f.severity in ("blocker", "major")] + should = [f for f in inp.findings if f.severity == "minor"] + nits = [f for f in inp.findings if f.severity == "nit"] + + if must: + lines.append(f"### 🔴 Must fix ({len(must)})") + for f in must: + loc = f" — {f.file}:{f.line}" if f.file else "" + lines.append(f"- [{f.severity}] {f.message}{loc}") + lines.append("") + if should: + lines.append(f"### 🟡 Should fix ({len(should)})") + for f in should: + loc = f" — {f.file}:{f.line}" if f.file else "" + lines.append(f"- [{f.severity}] {f.message}{loc}") + lines.append("") + if nits: + lines.append(f"### 🔵 Nits ({len(nits)})") + for f in nits: + loc = f" — {f.file}:{f.line}" if f.file else "" + lines.append(f"- [{f.severity}] {f.message}{loc}") + lines.append("") + + lines.append("### Next steps") + if verdict == "PASS": + lines.append("1. 满足合并门禁；如策略开启 auto_merge 且操作者带 --apply，可执行合并") + elif verdict == "REQUEST_CHANGES": + if failures: + lines.append(f"1. 优先解除硬门禁：{failures[0]['gate']} — {failures[0]['detail']}") + else: + lines.append("1. 评分低于阈值，按上方 Must/Should fix 修复后重新触发 gatekeeper") + if tracking_issue not in (None, ""): + lines.append(f"2. 关联 tracking issue #{tracking_issue}（已自动汇总必修项，修复后逐项勾选）") + else: + lines.append("1. 处于观察区间，建议处理 Should fix 项后复跑以争取 PASS") + lines.append("---") + lines.append("*Generated by gitlink-gatekeeper · policy-as-code PR gate · re-run after changes*") + return "\n".join(lines) + + +def render_tracking_issue_body( + inp: ScoreInput, + dims: dict[str, Any], + failures: list[dict[str, str]], + routing: dict[str, Any] | None, + owner: str, + repo: str, +) -> str: + """REQUEST_CHANGES 时创建的 tracking issue 正文。""" + must = [f for f in inp.findings if f.severity in ("blocker", "major")] + lines = [ + f"## Tracking — gatekeeper 拦截 PR #{inp.pr_id}", + "", + f"PR：`{owner}/{repo}` #{inp.pr_id} {inp.title}", + f"裁决：**REQUEST_CHANGES** · Score {dims['total']}/100", + "", + ] + if failures: + lines.append("### 必须解除的硬门禁") + for f in failures: + lines.append(f"- [ ] `{f['gate']}`: {f['detail']}") + lines.append("") + if must: + lines.append("### 必修项（blocker / major）") + for f in must: + loc = f" — {f.file}:{f.line}" if f.file else "" + lines.append(f"- [ ] [{f.severity}] {f.message}{loc}") + lines.append("") + if routing and routing.get("suggested_reviewers"): + lines.append("### 建议 reviewer") + lines.append("- " + ", ".join(f"@{r}" for r in routing["suggested_reviewers"])) + lines.append("") + lines.append("> 修复后请在 PR 上复跑 gatekeeper；全部勾选完成后关闭本 issue。") + lines.append("> Generated by gitlink-gatekeeper workflow.") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- # +# 采集编排 +# --------------------------------------------------------------------------- # + +def detect_linked_issue(description: str) -> bool: + return bool(re.search(r"#\d+", description or "")) + + +def normalize_ci_status(builds_payload: Any) -> str: + data = unwrap(builds_payload) + items = extract_first_list(data, ("builds", "items", "list")) + if not items: + return "unknown" + statuses = [] + for item in items: + if isinstance(item, dict): + s = str(first_value(item, ("status", "state", "result"), "")).lower() + statuses.append(s) + if not statuses: + return "unknown" + latest = statuses[0] + if latest in ("success", "passing", "passed", "ok", "1"): + return "passing" + if latest in ("failure", "failing", "failed", "error", "2"): + return "failing" + return "unknown" + + +def collect_pr_context( + cli_bin: str, + owner: str, + repo: str, + pr_id: str, + skip_ci: bool, +) -> dict[str, Any]: + """采集 PR 上下文（只读 CLI 命令，见 SSOT 第 7 节）。""" + view = unwrap(run_cli_json(cli_bin, ["pr", "+view", "-i", pr_id], owner, repo)) + view = view if isinstance(view, dict) else {} + # GitLink 的 PR 由 issue 承载：title/description 在 view.issue（subject/description）， + # view.pull_request 只有合并相关字段。兼容直接返回 PR 对象的情况。 + issue = view.get("issue") if isinstance(view.get("issue"), dict) else {} + pr = view.get("pull_request") if isinstance(view.get("pull_request"), dict) else {} + # PR 背后承载的 issue id —— 给 PR 挂标签时的 Raw API 路径占位需要它。 + issue_id = first_value(issue, ("id", "issue", "number"), "") + issue_id = str(issue_id) if issue_id not in (None, "") else "" + title = str( + first_value(issue, ("subject", "title", "name")) + or first_value(view, ("title", "subject", "name")) + or first_value(pr, ("title", "subject", "name"), f"PR #{pr_id}") + ) + description = str( + first_value(issue, ("description", "body", "notes")) + or first_value(view, ("body", "description", "notes")) + or first_value(pr, ("body", "description", "notes"), "") + ) + + files_payload = run_cli_json(cli_bin, ["pr", "+files", "-i", pr_id], owner, repo) + changed_files = extract_file_paths(files_payload) + + # commits：SSOT 第 7 节用 Raw API GET /:owner/:repo/pulls/:id/commits。 + # 路径里的 :id 需替换为实际 PR 号；该端点在部分实例可能未开放，失败则降级 + # （commit_quality 按 total=0 给满分，见 SSOT 3.4），不阻断整条闭环。 + commits_payload: Any = {} + try: + commits_payload = run_cli_json( + cli_bin, + ["api", "GET", f"/:owner/:repo/pulls/{pr_id}/commits"], + owner, + repo, + ) + except WorkflowError: + commits_payload = {} + commit_msgs: list[str] = [] + citems = extract_first_list(unwrap(commits_payload), ("commits", "items", "list")) + for c in citems: + if isinstance(c, dict): + msg = first_value(c, ("message", "title", "commit_message"), "") + commit = c.get("commit") if isinstance(c.get("commit"), dict) else None + if not msg and commit: + msg = first_value(commit, ("message", "title"), "") + if msg: + commit_msgs.append(str(msg)) + + ci_status = "unknown" + if not skip_ci: + try: + ci_status = normalize_ci_status( + run_cli_json(cli_bin, ["ci", "+builds"], owner, repo) + ) + except WorkflowError: + ci_status = "unknown" + + return { + "title": title, + "description": description, + "issue_id": issue_id, + "changed_files": changed_files, + "commits": commit_msgs, + "ci_status": ci_status, + "linked_issue": detect_linked_issue(description), + } + + +def load_findings(path: Path | None) -> list[Finding]: + """从 --findings JSON 注入 AI 审查发现；缺省为空（结果仍确定性可复现）。""" + if path is None: + return [] + if not path.exists(): + raise WorkflowError(f"找不到 findings 文件：{path}") + raw = json.loads(path.read_text(encoding="utf-8")) + items = raw.get("findings", raw) if isinstance(raw, dict) else raw + findings: list[Finding] = [] + for item in items or []: + if not isinstance(item, dict): + continue + sev = str(item.get("severity", "")).lower() + if sev not in SEVERITY_ORDER: + continue + findings.append( + Finding( + severity=sev, + message=str(item.get("message", "")), + file=str(item.get("file", "")), + line=item.get("line", ""), + ) + ) + return findings + + +# --------------------------------------------------------------------------- # +# 善后：构造写操作计划（标签 / 评论 / tracking issue / 合并） +# --------------------------------------------------------------------------- # + +def build_label_command(label_name: str, color: str = "#1E90FF") -> PlannedWrite: + """确保裁决标签「定义」存在（label +create）。 + + 这是「打标签」两步中的第一步——仅创建标签定义本身。第二步「把标签挂到 + PR 背后的 issue」没有独立 shortcut，需走 Raw API + POST /:owner/:repo/issues/:id 带 issue_tag_ids（见模块 docstring 与 + attach_label_to_pr）。挂载是受 --apply 显式门控的安全写操作： + dry-run 只预览将要挂载的标签，apply 时才真正发起 POST。 + """ + return PlannedWrite( + label=f"确保标签存在：{label_name}", + command=["label", "+create", "-n", label_name, "-c", color], + note="如标签已存在会返回冲突，可忽略", + ) + + +def lookup_label_id( + cli_bin: str, owner: str, repo: str, label_name: str +) -> Any: + """用 `label +list` 按 name 找出标签 id；找不到返回 None。 + + GitLink 返回 data.issue_tags = [{id, name, ...}]。只读命令，dry-run 不调用。 + """ + payload = run_cli_json(cli_bin, ["label", "+list"], owner, repo) + tags = extract_first_list(unwrap(payload), ("issue_tags", "tags", "items", "list")) + for tag in tags: + if isinstance(tag, dict) and str(tag.get("name", "")) == label_name: + return tag.get("id") + return None + + +def build_attach_label_command( + issue_id: str, + tag_id: Any, + title: str, + description: str, + label_name: str, +) -> PlannedWrite: + """把标签挂到 PR 背后 issue 的 Raw API 写操作。 + + 必须带回原 subject/description，否则 GitLink 会把标题/描述清空 + （见 gitlink-shared 实测）。 + """ + body = json.dumps( + { + "issue_tag_ids": [tag_id], + "done_ratio": 0, + "subject": title, + "description": description, + }, + ensure_ascii=False, + ) + return PlannedWrite( + label=f"挂载标签到 PR(issue {issue_id})：{label_name}", + command=["api", "POST", f"/:owner/:repo/issues/{issue_id}", "--body", body], + note=f"issue_tag_ids=[{tag_id}]", + ) + + +def attach_label_to_pr( + cli_bin: str, + owner: str, + repo: str, + issue_id: str, + label_name: str, + title: str, + description: str, +) -> dict[str, Any]: + """apply 时：查 tag id → Raw API 挂到 PR 背后 issue。封装多步逻辑保持 main 可读。 + + 任何一步失败都返回 {ok: False, ...} 而非抛出，保证整条闭环优雅降级。 + """ + if not issue_id: + return { + "label": f"挂载标签到 PR：{label_name}", + "ok": False, + "stderr": "未取到 PR 背后 issue id，跳过挂载", + "data": None, + } + try: + tag_id = lookup_label_id(cli_bin, owner, repo, label_name) + except WorkflowError as exc: + return { + "label": f"挂载标签到 PR：{label_name}", + "ok": False, + "stderr": f"label +list 失败：{exc}", + "data": None, + } + if tag_id in (None, ""): + return { + "label": f"挂载标签到 PR：{label_name}", + "ok": False, + "stderr": f"label +list 未找到标签 {label_name} 的 id", + "data": None, + } + write = build_attach_label_command(issue_id, tag_id, title, description, label_name) + return execute_write(write, owner, repo, cli_bin) + + +def build_comment_command(pr_id: str, body: str) -> PlannedWrite: + return PlannedWrite( + label=f"回写评分卡评论到 PR #{pr_id}", + command=["pr", "+comment", "-i", pr_id, "-b", body], + note=f"评分卡 {len(body)} 字符", + ) + + +def build_tracking_issue_command(title: str, body: str) -> PlannedWrite: + return PlannedWrite( + label="创建 tracking issue 汇总必修项", + command=["issue", "+create", "-t", title, "-b", body], + note=title, + ) + + +def parse_created_issue_number(res: dict[str, Any]) -> Any: + """从 issue +create 的 execute_write 结果里回捕新建 issue 的编号/ID。 + + GitLink 字段命名不统一（issue / id / number / pull_request），用 first_value + 兼容多键；取不到返回 None。 + """ + if not res or not res.get("ok"): + return None + data = unwrap(res.get("data")) + if isinstance(data, dict): + return first_value(data, ("issue", "id", "number", "pull_request")) + return None + + +def build_merge_command(pr_id: str, method: str) -> PlannedWrite: + return PlannedWrite( + label=f"合并 PR #{pr_id}（{method}）", + command=["pr", "+merge", "-i", pr_id, "-m", method], + note="仅 PASS + auto_merge + --apply 时出现", + ) + + +# --------------------------------------------------------------------------- # +# 主流程 +# --------------------------------------------------------------------------- # + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="gatekeeper_workflow.py", + description="gitlink-gatekeeper 端到端 PR 看门人闭环：路由 reviewer → 裁决 → 回写评论/标签/tracking issue。" + " 默认 dry-run，写操作需 --apply。", + ) + parser.add_argument("--config", type=Path, help="工作流配置文件（YAML），含 owner/repo/policy/owner_rules 路径") + parser.add_argument("--owner", help="仓库 owner（覆盖配置）") + parser.add_argument("--repo", help="仓库名（覆盖配置）") + parser.add_argument("--pr", dest="pr_id", help="目标 PR 编号（覆盖配置）") + parser.add_argument("--policy", type=Path, help="gatekeeper.yaml 策略路径（缺省回退内置默认策略）") + parser.add_argument("--owner-rules", dest="owner_rules", type=Path, help="owner-rules.yaml 路径") + parser.add_argument("--findings", type=Path, help="AI 审查发现 JSON（注入 review_findings；缺省为空）") + parser.add_argument("--cli-bin", default="gitlink-cli", help="gitlink-cli 可执行文件路径") + parser.add_argument("--skip-ci", action="store_true", help="跳过 CI 采集（ci_status 记为 unknown）") + parser.add_argument("--output-dir", type=Path, default=Path("outputs"), help="本地产物输出目录") + parser.add_argument( + "--apply", + action="store_true", + help="执行写操作（回写评论 / 打标签 / 建 issue / 合并）。不传则仅预览（安全默认）。", + ) + parser.add_argument("--no-color", action="store_true", help="不输出彩色（保留位，当前纯文本）") + return parser.parse_args(argv) + + +def resolve_config(args: argparse.Namespace) -> dict[str, Any]: + cfg: dict[str, Any] = {} + if args.config: + cfg = load_yaml_file(args.config) + base = args.config.parent if args.config else Path.cwd() + + def resolve_path(value: Any) -> Path | None: + if not value: + return None + p = Path(value) + return p if p.is_absolute() else (base / p) + + owner = args.owner or cfg.get("owner") + repo = args.repo or cfg.get("repo") + pr_id = args.pr_id or (str(cfg["pr"]) if cfg.get("pr") is not None else None) + policy_path = args.policy or resolve_path(cfg.get("policy")) + owner_rules_path = args.owner_rules or resolve_path(cfg.get("owner_rules")) + findings_path = args.findings or resolve_path(cfg.get("findings")) + + if not owner or not repo: + raise WorkflowError("必须提供 owner 和 repo（通过 --config 或 --owner/--repo）") + if not pr_id: + raise WorkflowError("必须提供 PR 编号（通过 --config 的 pr 字段或 --pr）") + if not owner_rules_path: + raise WorkflowError("必须提供 owner-rules.yaml（通过 --config 的 owner_rules 或 --owner-rules）") + + return { + "owner": str(owner), + "repo": str(repo), + "pr_id": str(pr_id), + "policy_path": policy_path, + "owner_rules_path": owner_rules_path, + "findings_path": findings_path, + } + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + try: + conf = resolve_config(args) + owner, repo, pr_id = conf["owner"], conf["repo"], conf["pr_id"] + cli_bin = args.cli_bin + + policy = load_policy(conf["policy_path"]) + policy_label = ( + f"{conf['policy_path'].name}@v{policy['version']}" + if conf["policy_path"] and conf["policy_path"].exists() + else f"builtin-default@v{policy['version']}" + ) + rules, fallback = load_owner_rules(conf["owner_rules_path"]) + findings = load_findings(conf["findings_path"]) + + print(f"=== gitlink-gatekeeper PR 看门人闭环 ===") + print(f"目标：{owner}/{repo} PR #{pr_id} · policy: {policy_label}") + print(f"模式：{'APPLY（将执行写操作）' if args.apply else 'DRY-RUN（仅预览，不写任何东西）'}") + print() + + if not cli_available(cli_bin): + raise WorkflowError( + f"未找到 {cli_bin}，请先安装 gitlink-cli 并完成 `gitlink-cli auth login`" + ) + + # --- 步骤 1+2 采集 --- + print("[1/3] 采集 PR 上下文并路由 reviewer …") + ctx = collect_pr_context(cli_bin, owner, repo, pr_id, args.skip_ci) + changed_src, changed_tests = count_src_test(ctx["changed_files"], policy) + routing = route_reviewers(ctx["changed_files"], rules, fallback) + print(f" 变更文件 {len(ctx['changed_files'])} 个（src {changed_src} / test {changed_tests}）") + print(f" 建议 reviewer：{', '.join(routing['suggested_reviewers']) or '（无规则命中）'}") + if routing["unmatched"]: + print(f" 未命中规则文件 {len(routing['unmatched'])} 个 → fallback") + print() + + # --- 步骤 2 裁决 --- + print("[2/3] 按策略评分并裁决 …") + inp = ScoreInput( + pr_id=pr_id, + title=ctx["title"], + description=ctx["description"], + changed_files=ctx["changed_files"], + changed_src=changed_src, + changed_tests=changed_tests, + commits=ctx["commits"], + ci_status=ctx["ci_status"], + linked_issue=ctx["linked_issue"], + findings=findings, + ) + dims = score_dimensions(inp, policy) + failures = evaluate_hard_gates(inp, policy) + verdict = decide_verdict(dims["total"], bool(failures), policy) + scorecard = render_scorecard(inp, dims, failures, verdict, policy_label, routing) + print(f" Score: {dims['total']}/100 · 硬门禁失败 {len(failures)} 项 · 裁决: {verdict}") + print() + + # 落盘本地产物（无论 dry-run 与否都生成，便于复核 / 验证记录） + out_dir: Path = args.output_dir + out_dir.mkdir(parents=True, exist_ok=True) + slug = f"{owner}_{repo}_pr{pr_id}".replace("/", "_") + scorecard_path = out_dir / f"{slug}_scorecard.md" + summary_path = out_dir / f"{slug}_summary.json" + scorecard_path.write_text(scorecard + "\n", encoding="utf-8") + + # --- 步骤 3 善后 --- + print("[3/3] 回写 + 善后 …") + behavior = policy["behavior"] + labels = policy["labels"] + verdict_label = { + "PASS": labels["pass"], + "REQUEST_CHANGES": labels["request_changes"], + "COMMENT": labels["comment"], + }[verdict] + + results: list[dict[str, Any]] = [] + tracking_issue_no: Any = None + + # REQUEST_CHANGES：apply 时「先建 tracking issue → 回捕编号 → 把编号补进评分卡 + # Next steps → 再回写评论」，让评论与 issue 双向可追溯。 + issue_title = issue_body = "" + if verdict == "REQUEST_CHANGES": + issue_title = f"[gatekeeper] PR #{pr_id} 未通过门禁：{ctx['title'][:60]}" + issue_body = render_tracking_issue_body(inp, dims, failures, routing, owner, repo) + if args.apply: + print(" APPLY：执行写操作 …") + tracking_write = build_tracking_issue_command(issue_title, issue_body) + res = execute_write(tracking_write, owner, repo, cli_bin) + results.append(res) + tracking_issue_no = parse_created_issue_number(res) + status = "OK" if res["ok"] else f"FAIL（{res['stderr'][:120]}）" + print(f" • {tracking_write.label} → {status}") + if tracking_issue_no not in (None, ""): + print(f" • tracking issue 已建：#{tracking_issue_no}") + # 把编号补进评分卡 Next steps 后重新落盘（评论将带上回链） + scorecard = render_scorecard( + inp, dims, failures, verdict, policy_label, routing, + tracking_issue=tracking_issue_no, + ) + scorecard_path.write_text(scorecard + "\n", encoding="utf-8") + + # 构造（其余）写操作计划。评论用上面可能已带回链的 scorecard。 + plan: list[PlannedWrite] = [] + if behavior.get("post_comment", True): + plan.append(build_comment_command(pr_id, scorecard)) + if behavior.get("apply_label", True): + plan.append(build_label_command(verdict_label)) + # dry-run 下把 tracking issue 也列进计划预览（apply 路径已在上面执行掉） + if verdict == "REQUEST_CHANGES" and not args.apply: + plan.append(build_tracking_issue_command(issue_title, issue_body)) + if ( + verdict == "PASS" + and behavior.get("auto_merge", False) + and args.apply + ): + plan.append(build_merge_command(pr_id, behavior.get("merge_method", "squash"))) + + if not args.apply: + print(" DRY-RUN：以下写操作不会执行（加 --apply 才执行）：") + for w in plan: + print(render_planned(w, owner, repo, cli_bin)) + if behavior.get("apply_label", True): + print( + f" • （计划）apply 时将把标签 {verdict_label} 挂到 PR 背后 issue" + f"（label +list 查 id → Raw API POST /:owner/:repo/issues/）" + ) + else: + if verdict != "REQUEST_CHANGES": + print(" APPLY：执行写操作 …") + for w in plan: + res = execute_write(w, owner, repo, cli_bin) + status = "OK" if res["ok"] else f"FAIL（{res['stderr'][:120]}）" + print(f" • {w.label} → {status}") + results.append(res) + # 标签真正挂到 PR：create 之后查 id 并 Raw API 挂载（受 --apply 门控） + if behavior.get("apply_label", True): + attach_res = attach_label_to_pr( + cli_bin, owner, repo, ctx.get("issue_id", ""), + verdict_label, ctx["title"], ctx["description"], + ) + status = "OK" if attach_res["ok"] else f"FAIL（{attach_res['stderr'][:120]}）" + iid = ctx.get("issue_id", "") or "?" + print(f" • 挂载标签到 PR(issue {iid}) → {status}") + results.append(attach_res) + print() + + # 结构化摘要 + summary = { + "owner": owner, + "repo": repo, + "pr_id": pr_id, + "policy": policy_label, + "mode": "apply" if args.apply else "dry-run", + "routing": routing, + "scores": {k: v for k, v in dims.items() if k != "total"}, + "total": dims["total"], + "hard_gate_failures": failures, + "verdict": verdict, + "verdict_label": verdict_label, + "tracking_issue": tracking_issue_no, + "planned_writes": [ + {"label": w.label, "command": w.command, "note": w.note} for w in plan + ], + "executed": results, + "artifacts": { + "scorecard": scorecard_path.as_posix(), + "summary": summary_path.as_posix(), + }, + } + summary_path.write_text( + json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + print(f"评分卡已落盘：{scorecard_path}") + print(f"结构化摘要：{summary_path}") + print(f"最终裁决：{VERDICT_EMOJI[verdict]} {verdict} · {dims['total']}/100") + # REQUEST_CHANGES 时返回码 2，便于 CI 接入做门禁；PASS/COMMENT 返回 0。 + return 2 if verdict == "REQUEST_CHANGES" else 0 + + except WorkflowError as exc: + print(f"错误：{exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/workflows/pr-quality-gatekeeper/tests/test_scoring.py b/examples/workflows/pr-quality-gatekeeper/tests/test_scoring.py new file mode 100644 index 0000000..b5eb763 --- /dev/null +++ b/examples/workflows/pr-quality-gatekeeper/tests/test_scoring.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python3 +"""可复现评分单测 —— 把「同输入 → 同分 → 同裁决」从口号变成可验证事实。 + +纯标准库 unittest（Python 3.9 兼容）。直接 import `scripts/gatekeeper_workflow.py` +的确定性算法（score_dimensions / evaluate_hard_gates / decide_verdict），对四个 +权威裁决案例（skills/gitlink-gatekeeper/examples/decision-*.md 与 +scorecard-sample.md）构造等价的 ScoreInput，断言**总分**与**三态裁决**与文档逐位一致。 + +任意一处算法改动若改变了这四个案例的分值，本测试立即变红——即为「确定性」的回归护栏。 + +运行： + python3 workflow/tests/test_scoring.py +或： + python3 -m unittest workflow.tests.test_scoring # 在仓库根目录 + +数值来源（默认策略 gatekeeper.yaml，与脚本内置 DEFAULT_POLICY 一致）： + 权重 40/20/15/15/10；severity_penalty blocker=100/major=25/minor=5/nit=1； + thresholds pass=85 / request_changes=60；max_changed_files=80。 +""" + +from __future__ import annotations + +import importlib.util +import sys +import unittest +from pathlib import Path + +# --------------------------------------------------------------------------- # +# 以绝对路径加载被测脚本（它在 scripts/ 下、非包，按文件直接载入最稳） +# +# 注意：必须先把模块塞进 sys.modules 再 exec —— 被测脚本用了 +# `from __future__ import annotations`，Python 3.9 的 @dataclass 在解析字符串 +# 注解时会回查 sys.modules[cls.__module__]，未注册会取到 None 而报 +# AttributeError（'NoneType' object has no attribute '__dict__'）。 +# --------------------------------------------------------------------------- # +_SCRIPT = ( + Path(__file__).resolve().parent.parent / "scripts" / "gatekeeper_workflow.py" +) +_spec = importlib.util.spec_from_file_location("gatekeeper_workflow", _SCRIPT) +assert _spec and _spec.loader, f"无法定位被测脚本：{_SCRIPT}" +gw = importlib.util.module_from_spec(_spec) +sys.modules["gatekeeper_workflow"] = gw +_spec.loader.exec_module(gw) # type: ignore[union-attr] + +ScoreInput = gw.ScoreInput +Finding = gw.Finding +score_dimensions = gw.score_dimensions +evaluate_hard_gates = gw.evaluate_hard_gates +decide_verdict = gw.decide_verdict +# 默认策略（深拷贝一份，避免任何用例意外改到共享 dict） +import json as _json # noqa: E402 + +DEFAULT_POLICY = _json.loads(_json.dumps(gw.DEFAULT_POLICY)) + + +# --------------------------------------------------------------------------- # +# 构造辅助：把「严重度计数 / 文件数 / commit 计数」翻译成 ScoreInput 字段 +# --------------------------------------------------------------------------- # + +def _findings(blocker: int = 0, major: int = 0, minor: int = 0, nit: int = 0): + """按严重度计数生成 Finding 列表（message/file/line 对评分无影响，仅 severity 计 penalty）。""" + out = [] + for sev, n in (("blocker", blocker), ("major", major), ("minor", minor), ("nit", nit)): + for i in range(n): + out.append(Finding(severity=sev, message=f"{sev} #{i}", file="f.go", line=i + 1)) + return out + + +def _commits(conforming: int, total: int): + """生成 total 条 commit message，其中 conforming 条符合 Conventional Commits。""" + assert conforming <= total + msgs = [f"feat(mod{i}): conforming change {i}" for i in range(conforming)] + msgs += [f"wip update {i}" for i in range(total - conforming)] # 'wip ...' 不匹配规约 + return msgs + + +def _files(n: int): + """生成 n 个占位变更文件路径（仅用于 size 维度计 len，src/test 计数由字段直接给定）。""" + return [f"path/file_{i}.go" for i in range(n)] + + +def _build( + *, + pr_id: str, + title: str, + desc_len: int, + linked_issue: bool, + n_files: int, + src: int, + tests: int, + commits: tuple, # (conforming, total) + ci: str, + findings_counts: dict, +) -> ScoreInput: + description = "x" * desc_len if desc_len else "" + return ScoreInput( + pr_id=pr_id, + title=title, + description=description, + changed_files=_files(n_files), + changed_src=src, + changed_tests=tests, + commits=_commits(*commits), + ci_status=ci, + linked_issue=linked_issue, + findings=_findings(**findings_counts), + ) + + +def _run(inp: ScoreInput): + """跑完整确定性链路，返回 (total, verdict)。""" + dims = score_dimensions(inp, DEFAULT_POLICY) + failures = evaluate_hard_gates(inp, DEFAULT_POLICY) + verdict = decide_verdict(dims["total"], bool(failures), DEFAULT_POLICY) + return dims, failures, verdict + + +# --------------------------------------------------------------------------- # +# 四个权威案例 +# --------------------------------------------------------------------------- # + +class TestAuthoritativeCases(unittest.TestCase): + """对照 examples/ 下四个裁决记录，断言总分与裁决。""" + + def test_decision_pass(self): + # decision-pass.md：3 src / 2 test、desc 142(含#198)、4/4 commit、CI passing、 + # 0/0/1/2 findings → 33+17+15+15+10 = 90 → PASS + inp = _build( + pr_id="214", + title="feat(search): validate pagination params", + desc_len=142, + linked_issue=True, + n_files=5, + src=3, + tests=2, + commits=(4, 4), + ci="passing", + findings_counts={"minor": 1, "nit": 2}, + ) + dims, failures, verdict = _run(inp) + self.assertEqual(dims["review_findings"]["score"], 33) + self.assertEqual(dims["test_coverage"]["score"], 17) + self.assertEqual(dims["pr_hygiene"]["score"], 15) + self.assertEqual(dims["commit_quality"]["score"], 15) + self.assertEqual(dims["ci_status"]["score"], 10) + self.assertEqual(failures, []) + self.assertEqual(dims["total"], 90) + self.assertEqual(verdict, "PASS") + + def test_decision_request_changes(self): + # decision-request-changes.md：4 src / 0 test（触发硬门禁 + # require_tests_for_src_changes）、desc 88 无关联、2/3 commit、CI passing、 + # 0/1/1/2 findings → 8+0+10+10+10 = 38 → REQUEST_CHANGES + inp = _build( + pr_id="305", + title="refactor(billing): rework settlement pipeline", + desc_len=88, + linked_issue=False, + n_files=4, + src=4, + tests=0, + commits=(2, 3), + ci="passing", + findings_counts={"major": 1, "minor": 1, "nit": 2}, + ) + dims, failures, verdict = _run(inp) + self.assertEqual(dims["review_findings"]["score"], 8) + self.assertEqual(dims["test_coverage"]["score"], 0) + self.assertEqual(dims["pr_hygiene"]["score"], 10) + self.assertEqual(dims["commit_quality"]["score"], 10) + self.assertEqual(dims["ci_status"]["score"], 10) + gate_names = {f["gate"] for f in failures} + self.assertIn("require_tests_for_src_changes", gate_names) + self.assertEqual(dims["total"], 38) + self.assertEqual(verdict, "REQUEST_CHANGES") + + def test_decision_comment(self): + # decision-comment.md：2 src / 1 test、desc 52 无关联、2/3 commit、CI passing、 + # 0/0/3/2 findings → 23+15+10+10+10 = 68 ∈ [60,85) 且无硬门禁 → COMMENT + inp = _build( + pr_id="277", + title="feat(config): merge defaults on load", + desc_len=52, + linked_issue=False, + n_files=2, + src=2, + tests=1, + commits=(2, 3), + ci="passing", + findings_counts={"minor": 3, "nit": 2}, + ) + dims, failures, verdict = _run(inp) + self.assertEqual(dims["review_findings"]["score"], 23) + self.assertEqual(dims["test_coverage"]["score"], 15) + self.assertEqual(dims["pr_hygiene"]["score"], 10) + self.assertEqual(dims["commit_quality"]["score"], 10) + self.assertEqual(dims["ci_status"]["score"], 10) + self.assertEqual(failures, []) + self.assertEqual(dims["total"], 68) + self.assertEqual(verdict, "COMMENT") + + def test_scorecard_sample(self): + # scorecard-sample.md：4 src / 0 test（触发硬门禁）、desc 64 无关联、3/4 commit、 + # CI passing、0/1/2/1 findings → 4+0+10+11+10 = 35 → REQUEST_CHANGES + inp = _build( + pr_id="128", + title="feat(auth): add refresh-token rotation", + desc_len=64, + linked_issue=False, + n_files=6, + src=4, + tests=0, + commits=(3, 4), + ci="passing", + findings_counts={"major": 1, "minor": 2, "nit": 1}, + ) + dims, failures, verdict = _run(inp) + self.assertEqual(dims["review_findings"]["score"], 4) + self.assertEqual(dims["test_coverage"]["score"], 0) + self.assertEqual(dims["pr_hygiene"]["score"], 10) + self.assertEqual(dims["commit_quality"]["score"], 11) + self.assertEqual(dims["ci_status"]["score"], 10) + gate_names = {f["gate"] for f in failures} + self.assertIn("require_tests_for_src_changes", gate_names) + self.assertEqual(dims["total"], 35) + self.assertEqual(verdict, "REQUEST_CHANGES") + + +class TestVerdictBoundaries(unittest.TestCase): + """裁决判定树（decide_verdict）边界：与 thresholds pass=85 / request_changes=60 一致。""" + + def test_pass_threshold_inclusive(self): + self.assertEqual(decide_verdict(85, False, DEFAULT_POLICY), "PASS") + + def test_comment_band(self): + self.assertEqual(decide_verdict(60, False, DEFAULT_POLICY), "COMMENT") + self.assertEqual(decide_verdict(84, False, DEFAULT_POLICY), "COMMENT") + + def test_request_changes_below_band(self): + self.assertEqual(decide_verdict(59, False, DEFAULT_POLICY), "REQUEST_CHANGES") + + def test_hard_gate_short_circuits_high_score(self): + # 即便满分，硬门禁失败也直接 REQUEST_CHANGES + self.assertEqual(decide_verdict(100, True, DEFAULT_POLICY), "REQUEST_CHANGES") + + +if __name__ == "__main__": + unittest.main(verbosity=2)