From 1d3428a2888041163563dd7710676b538f520226 Mon Sep 17 00:00:00 2001 From: chengma Date: Tue, 14 Jul 2026 13:05:19 +0800 Subject: [PATCH] feat(governance): automate harness consistency checks (phase 3) --- .gitea/ISSUE_TEMPLATE/task.md | 2 +- .gitea/workflows/harness-governance.yml | 21 + .gitignore | 3 + AGENTS.md | 9 + README.md | 10 +- docs/00-ai-start-here.md | 4 +- docs/README.md | 7 +- docs/adoption-checklist.md | 2 +- docs/clean-state-checklist.md | 1 + docs/current-state.md | 2 +- docs/gitea-collaboration.md | 39 +- docs/gitea-mcp.md | 22 +- docs/method-map.md | 4 +- docs/tasks/README.md | 7 +- gitea.env.example | 4 +- scripts/audit_gitea_coordination.py | 688 ++++++++++++++++++++++++ scripts/setup_gitea_labels.py | 11 +- scripts/test_gitea_claim_race.py | 166 ++++++ scripts/validate_agent_context.py | 101 ++-- scripts/validate_harness_governance.py | 685 +++++++++++++++++++++++ tasks.md | 2 +- tests/test_governance.py | 419 +++++++++++++++ 22 files changed, 2149 insertions(+), 60 deletions(-) create mode 100644 .gitea/workflows/harness-governance.yml create mode 100644 scripts/audit_gitea_coordination.py create mode 100644 scripts/test_gitea_claim_race.py create mode 100644 scripts/validate_harness_governance.py create mode 100644 tests/test_governance.py diff --git a/.gitea/ISSUE_TEMPLATE/task.md b/.gitea/ISSUE_TEMPLATE/task.md index b1835f5..500a78a 100644 --- a/.gitea/ISSUE_TEMPLATE/task.md +++ b/.gitea/ISSUE_TEMPLATE/task.md @@ -32,5 +32,5 @@ labels: - claimed_by: `【领取后填写非敏感 agent-id】` - lease_until: `【领取后填写 RFC 3339 时间】` -领取必须遵循 `docs/gitea-collaboration.md` 的原子 claim 流程。不要在本 Issue 粘贴 Token、Authorization header 或私有配置。 +领取必须遵循 `docs/gitea-collaboration.md` 的 dispatcher 串行分配与 claim 标记流程。不要在本 Issue 粘贴 Token、Authorization header 或私有配置。 创建后先把 Issue 编号回填任务文件并合入默认分支，再按主要变更选择唯一 `type/docs` 或 `type/code`、一个 `priority/*` 和 `status/todo`。映射提交完成前不可领取。 diff --git a/.gitea/workflows/harness-governance.yml b/.gitea/workflows/harness-governance.yml new file mode 100644 index 0000000..c3f53ce --- /dev/null +++ b/.gitea/workflows/harness-governance.yml @@ -0,0 +1,21 @@ +name: Harness governance + +on: + push: + pull_request: + +permissions: read-all + +jobs: + validate: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + persist-credentials: false + - name: Validate templates and governance + run: | + python scripts/validate_agent_context.py + python -m unittest discover -s tests -p "test_*.py" + python scripts/validate_harness_governance.py diff --git a/.gitignore b/.gitignore index c35cda9..a4a31de 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,6 @@ gitea.env.* !gitea.env.example *.stderr.log +# Python 本地校验缓存 +__pycache__/ +*.py[cod] diff --git a/AGENTS.md b/AGENTS.md index fecf4fb..dbab8b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -65,3 +65,12 @@ Get-ChildItem -Recurse -File ``` 如修改链接或文件名，使用 `rg` 搜索旧名称和新名称，确认引用一致。 + +涉及上下文清单、任务协议或 Gitea 模板时，再运行： + +```powershell +python -m unittest discover -s tests -p "test_*.py" +python scripts/validate_harness_governance.py +``` + +远端协调审计是可选只读检查，需要本机私有 Gitea 配置：`python scripts/audit_gitea_coordination.py --repo 【owner/repo】 --dispatcher 【Gitea登录名】`。 diff --git a/README.md b/README.md index f158ba0..9f6a126 100644 --- a/README.md +++ b/README.md @@ -33,10 +33,15 @@ | [`docs/agent-context.schema.json`](docs/agent-context.schema.json) | 上下文清单结构契约 | | [`docs/agent-context.md`](docs/agent-context.md) | 上下文清单的读取、缓存、权威来源和断连降级说明 | | [`docs/gitea-mcp.md`](docs/gitea-mcp.md) | 可选：Gitea MCP 共享文档与任务协调接入、安全和降级规则 | -| [`docs/gitea-collaboration.md`](docs/gitea-collaboration.md) | 可选：Issue / 任务文件 / PR 映射、原子领取和写路径防撞协议 | +| [`docs/gitea-collaboration.md`](docs/gitea-collaboration.md) | 可选：Issue / 任务文件 / PR 映射、串行分配和写路径防撞协议 | | [`scripts/validate_agent_context.py`](scripts/validate_agent_context.py) | 零第三方依赖校验上下文清单、Schema 和仓库相对路径 | | [`scripts/setup_gitea_labels.py`](scripts/setup_gitea_labels.py) | 默认只读预览远端差异、显式 `--apply` 的 Gitea 协作标签初始化脚本 | +| [`scripts/validate_harness_governance.py`](scripts/validate_harness_governance.py) | 离线检查导航、链接、任务、模板、工作流和敏感信息 | +| [`scripts/audit_gitea_coordination.py`](scripts/audit_gitea_coordination.py) | 只读审计远端任务映射、状态、分支、PR、写路径和过期 claim | +| [`scripts/test_gitea_claim_race.py`](scripts/test_gitea_claim_race.py) | 显式 `--apply` 的目标实例 claim 分支并发兼容性 smoke | +| [`tests/test_governance.py`](tests/test_governance.py) | 标准库治理回归测试 | | [`.gitea/ISSUE_TEMPLATE/task.md`](.gitea/ISSUE_TEMPLATE/task.md) / [`.gitea/PULL_REQUEST_TEMPLATE.md`](.gitea/PULL_REQUEST_TEMPLATE.md) | Gitea 任务 Issue 与 PR 模板 | +| [`.gitea/workflows/harness-governance.yml`](.gitea/workflows/harness-governance.yml) | push / PR 离线治理检查模板；需仓库 Actions 和 runner 已启用 | | [`docs/method-map.md`](docs/method-map.md) | 失败模式 → 首要修复 → 工件的诊断 / 导航对照表 | | [`docs/evaluator-rubric.md`](docs/evaluator-rubric.md) | 单次会话输出的结构化评审评分表 | | [`docs/quality-document.md`](docs/quality-document.md) | 代码库长期健康度追踪（产品域 × 架构层评级） | @@ -79,7 +84,8 @@ 项目进入多轮长期开发后，再按需启用： - `docs/gitea-mcp.md`：需要跨 agent 读取 Gitea 文档、Issue 和 PR 时启用。 -- `docs/gitea-collaboration.md`：需要多 agent 原子领取、独立分支 / worktree 和写路径防撞时启用。 +- `docs/gitea-collaboration.md`：需要 dispatcher 串行分配、独立分支 / worktree 和写路径防撞时启用。 +- `scripts/validate_harness_governance.py`：在本地和 CI 使用同一套离线一致性检查。 - `docs/clean-state-checklist.md`：每轮结束前检查仓库是否可恢复。 - `docs/method-map.md`：遇到失败模式时定位该补哪个工件。 - `docs/evaluator-rubric.md`：评审单次 agent 输出质量。 diff --git a/docs/00-ai-start-here.md b/docs/00-ai-start-here.md index 67d647d..9d63257 100644 --- a/docs/00-ai-start-here.md +++ b/docs/00-ai-start-here.md @@ -45,7 +45,7 @@ 4. 运行 `./init.sh`（Windows 原生 PowerShell 用 `./init.ps1`）：统一安装、验证、打印启动命令；如果脚本提示命令未替换，先配置脚本顶部三个命令。 5. 跑一条基础 smoke / 端到端路径，确认基线没坏。 6. **如果基线已坏，先修基线**，不要在坏的起点上叠新功能。 -7. 基线绿了，再从 `docs/tasks/` 为当前 agent 领取唯一任务（路线图见 [`06-tasks.md`](06-tasks.md)）；启用 Gitea 时由 dispatcher 串行分配并取得原子 claim。 +7. 基线绿了，再从 `docs/tasks/` 为当前 agent 领取唯一任务（路线图见 [`06-tasks.md`](06-tasks.md)）；启用 Gitea 时由 dispatcher 串行分配并创建 claim 标记。 ## 当前阶段 @@ -66,7 +66,7 @@ - 每个 agent 只领取一个 frontmatter `status: TODO` 且依赖均 `DONE` 的任务文件，取编号最靠前的；项目可并行多个 `write_paths` 互不重叠的任务。 - 若 `docs/tasks/` 暂无可领任务，先按 [`06-tasks.md`](06-tasks.md) 路线图把下一个建议任务落成任务文件，再领取。 - 未启用 Gitea 时，开始前在独立分支 / worktree 把该文件 frontmatter 的 `status` 改为 `DOING`。 -- 启用 Gitea 时，由 dispatcher 按 [`gitea-collaboration.md`](gitea-collaboration.md) 串行检查写路径并原子创建 `claims/T-<编号>`；worker 只接受已读回确认的分配。标签、assignee 和更新后读回不能代替同任务领取锁。 +- 启用 Gitea 时，由 dispatcher 按 [`gitea-collaboration.md`](gitea-collaboration.md) 串行检查写路径并创建 `claims/T-<编号>` 防御性标记；worker 只接受已读回确认的分配。不要把标签、assignee、读回或普通 create-branch API 单独当作并发锁。 - 本轮只完成这一个任务；验收通过后改为 `DONE`。 - **执行记录写进该任务文件的 `## 执行记录`**（改了什么、跑了什么验证、结果、决策）。 - 若项目现实发生变化（启动/验证路径、目录结构、blocker），覆盖更新 [`current-state.md`](current-state.md)；未启用 Gitea 时以任务文件 frontmatter 为状态权威，启用后以 Issue 为实时状态，不逐任务改写快照。 diff --git a/docs/README.md b/docs/README.md index 1ec9054..e53d45f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -30,7 +30,7 @@ - [当前实现状态](current-state.md)：可覆盖的当前快照，记录仓库现实状态、可运行命令和下一步可做任务。 - [Agent 上下文清单](agent-context.md) / [`agent-context.json`](agent-context.json) / [`Schema`](agent-context.schema.json)：按任务类型选择文档、用提交 / 文件 SHA 避免重复读取。 - [Gitea MCP 接入](gitea-mcp.md)：可选的共享文档、Issue / PR 协调、安全配置和断连降级规则。 -- [Gitea 多 Agent 协作](gitea-collaboration.md)：可选的任务映射、原子 claim、写路径防撞和 PR 状态协议。 +- [Gitea 多 Agent 协作](gitea-collaboration.md)：可选的任务映射、串行分配、防重复 claim 和 PR 状态协议。 - [收尾检查清单](clean-state-checklist.md)：会话结束前逐项检查，保证下一轮无需人工修复即可继续。 - [方法对照表](method-map.md)：失败模式 → 首要修复 → 工件；出问题先查这里对症补工件。 - [评审评分表](evaluator-rubric.md)：单次会话输出的结构化评审（6 维 0-2 分 + 校准说明）。 @@ -38,7 +38,12 @@ - [`../init.sh`](../init.sh) / [`../init.ps1`](../init.ps1)：标准启动与验证入口脚本（根目录），统一安装、验证和启动命令。按操作系统二选一：WSL / Git Bash / macOS / Linux 用 `init.sh`，Windows 原生 PowerShell 用 `init.ps1`；换技术栈只改脚本顶部三个命令变量；未替换前脚本会主动失败，避免把示例命令误当真实项目命令。 - [`../scripts/validate_agent_context.py`](../scripts/validate_agent_context.py)：零第三方依赖校验上下文清单、Schema 和仓库相对路径。 - [`../scripts/setup_gitea_labels.py`](../scripts/setup_gitea_labels.py)：默认只读预览远端差异、显式写入的 Gitea 协作标签初始化脚本。 +- [`../scripts/validate_harness_governance.py`](../scripts/validate_harness_governance.py)：离线检查导航、链接、任务、模板、工作流和敏感信息。 +- [`../scripts/audit_gitea_coordination.py`](../scripts/audit_gitea_coordination.py)：只读审计远端任务映射、状态、分支、PR、写路径和过期 claim。 +- [`../scripts/test_gitea_claim_race.py`](../scripts/test_gitea_claim_race.py)：显式写入并安全清理的 claim 分支并发兼容性 smoke。 +- [`../tests/test_governance.py`](../tests/test_governance.py)：标准库治理回归测试。 - [Gitea Issue 模板](../.gitea/ISSUE_TEMPLATE/task.md) / [PR 模板](../.gitea/PULL_REQUEST_TEMPLATE.md)：任务映射、写路径和验证证据字段。 +- [Gitea Actions 工作流](../.gitea/workflows/harness-governance.yml)：push / PR 离线治理模板；实际执行依赖仓库 Actions 和 runner。 ## 任务 / 进度 / 当前状态 diff --git a/docs/adoption-checklist.md b/docs/adoption-checklist.md index 58ad1e2..2426152 100644 --- a/docs/adoption-checklist.md +++ b/docs/adoption-checklist.md @@ -31,7 +31,7 @@ 推荐随后补齐：`docs/01-vision.md`、`docs/02-requirements.md`、`docs/03-tech-stack.md`、`docs/04-architecture.md`、`docs/api.md`、`docs/routes.md`、`docs/clean-state-checklist.md`；`progress.md` 可选（历史归档 / 项目级大事记）。 -需要 Gitea 多 Agent 协作时，再复制 `docs/gitea-mcp.md`、`docs/gitea-collaboration.md`、`.gitea/` 模板和 `scripts/setup_gitea_labels.py`；先只读预览远端标签差异，再由维护者显式 `--apply`。 +需要 Gitea 多 Agent 协作时，再复制 `docs/gitea-mcp.md`、`docs/gitea-collaboration.md`、`.gitea/` 模板、`scripts/setup_gitea_labels.py`、`scripts/audit_gitea_coordination.py`、离线治理脚本和测试；先只读预览远端标签差异，再由维护者显式 `--apply`。Actions 工作流仅在仓库已启用 Actions 且 runner 可用时生效。 ## 接入步骤 diff --git a/docs/clean-state-checklist.md b/docs/clean-state-checklist.md index 4e1129b..3ff29f0 100644 --- a/docs/clean-state-checklist.md +++ b/docs/clean-state-checklist.md @@ -14,5 +14,6 @@ - [ ] 代码处于可安全恢复的状态（必要时已提交，提交信息清晰）。 - [ ] 本轮实际修改没有超出任务 `write_paths`，也没有与其他活跃任务发生路径重叠。 - [ ] 启用 Gitea 时，Issue 的唯一 `status/*`、claim / 工作分支、PR 和任务状态彼此一致；未完成任务没有误删 claim。 +- [ ] 已运行 `python scripts/validate_harness_governance.py`；启用且可连接 Gitea 时，还运行了只读 `python scripts/audit_gitea_coordination.py --repo 【owner/repo】 --dispatcher 【Gitea登录名】`。 任意一项不满足，就先补到满足，再结束会话。 diff --git a/docs/current-state.md b/docs/current-state.md index 8e29579..33b68c5 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -61,7 +61,7 @@ 2. 读 `docs/00-ai-start-here.md`。 3. 读 `docs/05-coding-rules.md`。 4. 在 `docs/tasks/` 找到 frontmatter `status: TODO` 且依赖均 `DONE` 的任务文件；暂无任务时，先按 `docs/06-tasks.md` 路线图落成任务文件。 -5. 未启用 Gitea 时，在独立分支 / worktree 把任务改为 `DOING`；启用 Gitea 时，先按 `gitea-collaboration.md` 由 dispatcher 完成原子 claim 和路径分配，worker 读回成功后再开始。 +5. 未启用 Gitea 时，在独立分支 / worktree 把任务改为 `DOING`；启用 Gitea 时，先按 `gitea-collaboration.md` 由 dispatcher 串行完成路径分配和 claim 标记，worker 读回成功后再开始。 ## 维护规则 diff --git a/docs/gitea-collaboration.md b/docs/gitea-collaboration.md index 70e56f7..2e8f441 100644 --- a/docs/gitea-collaboration.md +++ b/docs/gitea-collaboration.md @@ -8,7 +8,7 @@ | --- | --- | --- | | `docs/tasks/T-<编号>.md` | 任务规格、依赖、允许写路径、验收标准、可审计执行证据 | Token、实例地址、临时聊天 | | Gitea Issue | 实时状态、领取者、阻塞、结构化 claim / 续租记录 | 需求正文的唯一副本 | -| `claims/T-<编号>` 分支 | 唯一领取锁；分支存在表示任务已被领取 | 工作提交 | +| `claims/T-<编号>` 分支 | 防御性领取标记；分支存在表示 dispatcher 已分配任务 | 工作提交、跨 dispatcher 的线性化锁 | | `agent//T-<编号>` 分支 | 单个 agent 的任务提交 | 其他任务的顺手修改 | | Pull Request | 评审、验证证据、合并决策 | 未进入 Git 的隐含上下文 | @@ -44,13 +44,13 @@ Issue 是实时状态权威；默认分支尚未合入工作提交时，其任 因此 dispatcher 能从默认分支可靠定位任务 ↔ Issue；没有双向映射或没有 `status/todo` 的任务都不可领取。 -## 串行分配与原子领取 +## 串行分配与防重复领取 -仅修改 assignee / `status/doing` 再读回不是原子操作：两个 agent 可能先后覆盖并各自读到成功。MVP 指定一个 dispatcher（主 agent 或维护者）串行执行分配；worker 不并发自选任务。dispatcher 再使用 Gitea “同名分支只能创建一次”的约束，防止同一任务因重试或旁路操作被重复领取： +仅修改 assignee / `status/doing` 再读回不是原子操作：两个 agent 可能先后覆盖并各自读到成功。MVP 的互斥保证来自单一 dispatcher（主 agent 或维护者）串行执行分配；worker 不并发自选任务。claim 分支用于识别已分配任务并拦截顺序重试 / 常见旁路，不把 Gitea 的普通 create-branch API 当作线性化锁： 1. 读取默认分支任务文件和对应 Issue，确认双向映射、依赖均为 `DONE`、Issue 为 `status/todo`，且目标 worker 没有其他活跃任务。 2. 读取默认分支头提交 SHA，记为 `context_ref`。串行检查所有活跃预留的 `write_paths`，不得与本任务重叠。 -3. 从精确的 `context_ref` 原子创建 `claims/T-<编号>`。创建成功者获得该任务锁；收到 `409` 或“分支已存在”即停止，不得仅因分支 SHA 相同就判定成功。 +3. 从精确的 `context_ref` 创建 `claims/T-<编号>` 防御性标记。若已存在、返回非成功或状态不确定就停止并人工核查；并发冲突在不同版本中可能表现为 `409` 或 `5xx`，不得自动无限重试，也不得仅因分支 SHA 相同就判定本次分配成功。 4. 创建 `agent//T-<编号>` 工作分支，更新 Issue 为 `status/doing`，按项目规则设置 assignee，并追加结构化 claim 评论。 5. dispatcher 读回 Issue 和两个分支；不一致时先修复协调状态，不把任务交给 worker。 6. worker 在独立 worktree 读回分配结果，再把工作分支任务文件更新为 `DOING`，写入 `context_ref`、claim / 工作分支和已接受的 `write_paths`；提交只触碰允许路径。 @@ -63,11 +63,13 @@ Issue 是实时状态权威；默认分支尚未合入工作提交时，其任 CLAIM task: T-123 claimed_by: 【agent-id】 -allocated_by: 【dispatcher-id】 +allocated_by: 【dispatcher 的 Gitea 登录名】 context_ref: 【40 位提交 SHA】 claim_branch: claims/T-123 work_branch: agent/【agent-id】/T-123 -write_paths: 【仓库相对路径列表】 +write_paths: + - docs/tasks/T-123.md + - 【其他仓库相对路径】 claimed_at: 【RFC 3339 时间】 lease_until: 【RFC 3339 时间】 ``` @@ -76,10 +78,10 @@ lease_until: 【RFC 3339 时间】 ## 写路径防撞 -- 默认分支任务文件定义初始 `write_paths`；领取后，最新被 dispatcher 接受的结构化 claim / scope 评论与工作分支任务文件共同定义活跃预留。二者不一致时暂停工作。 +- 默认分支任务文件定义初始 `write_paths`；领取后，由配置的 dispatcher Gitea 身份发布、且 `allocated_by` 与评论作者一致的最新完整 CLAIM / CLAIM RENEWAL，与工作分支任务文件共同定义活跃预留。二者不一致时暂停工作。 - `write_paths` 必须列出任务文件本身及预期修改的文件或目录；共享配置、锁文件、导航文件也要列入。 - 两条路径相同，或一条是另一条的目录前缀，视为重叠；活跃任务不得存在重叠路径。 -- 发现必须修改范围外文件时，先停止并在 Issue 提议扩展范围；dispatcher 串行复查其他活跃预留，接受后追加结构化 scope 评论，worker 同步更新工作分支任务文件，二者都完成后才能继续。 +- 发现必须修改范围外文件时，先停止并在 Issue 提议扩展范围；dispatcher 串行复查其他活跃预留，接受后追加包含全部字段和新路径的 `CLAIM RENEWAL`，worker 同步更新工作分支任务文件，二者都完成后才能继续。审计始终以最后一个完整 CLAIM 块为准。 - 活跃任务由 Issue 的 `status/doing`、`status/blocked`、`status/review` 判定。每个 agent 同时最多一个活跃任务；项目可以并行多个写路径互不重叠的任务。 建议 worktree 命令： @@ -101,10 +103,18 @@ git worktree add ../【项目】-T-123 -b agent/【agent-id】/T-123 origin/agen ## 过期 claim 与断连 -- agent 应在 `lease_until` 前用新评论续租；续租不更换 claim 分支。 +- worker 应在 `lease_until` 前请求续租；dispatcher 串行复查后，由自己的 Gitea 身份发布包含全部字段的 `CLAIM RENEWAL`。单次租期最长 24 小时，续租不得更换 `task`、`claimed_by`、`allocated_by`、`context_ref`、claim / 工作分支；审计以最后一个由配置 dispatcher 发布的有效块为准。 - claim 过期不等于可以自动抢占。维护者先检查 Issue 最后活动、工作分支新提交和 PR，再评论回收原因并人工删除 claim 分支。 - Gitea / MCP 断连时，只能继续已经确认归属自己的任务；不能领取新任务、释放锁或猜测远端状态。 +只读检测命令： + +```powershell +python scripts/audit_gitea_coordination.py --repo 【owner/repo】 --dispatcher 【Gitea登录名】 +``` + +从默认分支的 clean checkout 运行审计。`--dispatcher`（或非敏感环境变量 `GITEA_DISPATCHER_LOGIN`）指定唯一可信的 dispatcher Gitea 登录名；审计只接受该账号发布且 `allocated_by` 一致的 CLAIM。它会核对标签、任务依赖、任务 ↔ Issue、claim / 工作分支、PR、活跃写路径和 `lease_until`，不会写远端。发现过期 claim 后不自动删除：维护者先查 Issue 最后活动、分支新提交和 PR，再评论回收原因，确认无人继续工作后才通过 UI 或受控 REST 删除。 + ## 初始化标签 先预览，再显式写入： @@ -118,4 +128,13 @@ python scripts/setup_gitea_labels.py --repo 【owner/repo】 --apply ## 并发验收 -在专用测试任务上让两个进程同时创建同一个临时 claim 分支，验收结果必须是恰好一个 `201`、另一个 `409`。测试前后均需人工确认分支名，清理由受控 REST / UI 完成；不得在业务任务上试验。 +可用 `python scripts/test_gitea_claim_race.py --repo 【owner/repo】 --apply` 在唯一 `claims/__probe__/race-*` 临时分支做兼容性 smoke，期望恰好一个 `201`、一个 `409`。脚本只在名称前缀和 SHA 都符合预期时清理并复查 404。一次 smoke 结果不能证明 create-branch 线性化；无论结果如何，MVP 仍依赖 dispatcher 串行分配。不得在业务任务分支上试验。 + +## 自动化与升级阈值 + +- `python scripts/validate_harness_governance.py` 完全离线检查上下文清单、导航、本地链接、任务 frontmatter / 依赖 / 写路径、Gitea 模板和已跟踪文本中的敏感值。 +- `.gitea/workflows/harness-governance.yml` 在 push / PR 运行标准库测试和离线检查，不注入本机长期 PAT，也不运行远端审计。平台仍会提供 job token，工作流用 `permissions: read-all` 和 `persist-credentials: false` 收窄权限与留存。 +- Actions 模板只有合入默认分支、仓库启用 Actions 且带 Python 3.10+ 的 `ubuntu-latest` runner 可用时才会真正执行；内网 runner 还要能取得 `actions/checkout@v4`。没有 runner 时，以相同本地命令作为验收证据，不宣称 CI 已跑绿。 +- 退出码统一：`0` 通过，`1` 发现一致性问题，`2` 配置、网络或运行前提缺失。敏感信息检查只输出规则、文件和行号，不回显命中正文。 + +MVP 不实现 webhook、协调服务或独立 dashboard。只有出现以下任一信号才重新评估：单项目约 20 个以上并发任务、dispatcher 成为持续瓶颈、跨仓库聚合成为刚需、重复出现路径分配竞态，或审计 / 合规要求集中查询。届时优先增加原子 allocation 服务和 webhook 索引，再评估只读 dashboard；不把前端看板当作并发控制器。 diff --git a/docs/gitea-mcp.md b/docs/gitea-mcp.md index 6ab536e..38a2dba 100644 --- a/docs/gitea-mcp.md +++ b/docs/gitea-mcp.md @@ -17,6 +17,9 @@ GITEA_URL=【Gitea 实例根地址，不含 /api/v1】 GITEA_TOKEN=【最小权限 Personal Access Token】 +# 启用远端协调审计时设置；不是秘密： +GITEA_DISPATCHER_LOGIN=【唯一 dispatcher 的 Gitea 登录名】 + # 仅当团队明确接受 HTTP 下 Token 明文传输风险时设置： GITEA_ALLOW_INSECURE_HTTP=1 @@ -94,6 +97,23 @@ Gitea 中的文件与本地 `docs/` 是同一 Git 工件的远端与 checkout， 多 agent 协作时遵循 [`gitea-collaboration.md`](gitea-collaboration.md)： - 任务文件保存规格和长期证据，Issue 保存实时状态，PR 保存评审与合并决策。 -- 领取互斥依赖唯一 `claims/T-<编号>` 分支；assignee、`status/doing` 和读回仅作状态确认。 +- 领取互斥依赖 dispatcher 串行分配；`claims/T-<编号>` 是防御性标记，assignee、`status/doing` 和读回仅作状态确认。 - MVP 由单一 dispatcher 串行分配任务并检查 `write_paths`，每个 worker 使用 `agent//T-<编号>` 和独立 worktree；每任务 claim 只解决同任务重复领取，不单独保证跨任务路径互斥。 - MCP 可创建 claim / 工作分支，但当前没有创建标签或删除分支工具。标签用 `python scripts/setup_gitea_labels.py --repo 【owner/repo】 --apply` 幂等初始化；过期 claim 由维护者通过 UI 或受控 REST 人工回收。 + +## 治理检查 + +离线检查不需要 Token，可放进 Gitea Actions： + +```powershell +python -m unittest discover -s tests -p "test_*.py" +python scripts/validate_harness_governance.py +``` + +远端一致性检查单独运行，严格只读： + +```powershell +python scripts/audit_gitea_coordination.py --repo 【owner/repo】 --dispatcher 【Gitea登录名】 +``` + +远端审计区分“不一致”（退出码 1）和配置 / 网络 / 权限失败（退出码 2），并验证 dispatcher 评论身份、任务依赖和过期 claim；`--dispatcher` 可由非敏感环境变量 `GITEA_DISPATCHER_LOGIN` 代替。它不会更新标签、关闭 Issue、合并 PR 或删除分支。 diff --git a/docs/method-map.md b/docs/method-map.md index 1fb0aa2..14a7616 100644 --- a/docs/method-map.md +++ b/docs/method-map.md @@ -17,8 +17,10 @@ | 文档堆叠失控 | 入口文件越来越长，"每次失败加一句" | 渐进披露，入口保持薄 | [`00-ai-start-here.md`](00-ai-start-here.md) + 拆分到具体文档 | | 每轮全量重读 | 多 agent 反复拉取全部文档，慢且容易混入无关上下文 | 用任务路由和提交 / 文件 SHA 增量读取 | [`agent-context.md`](agent-context.md) + [`agent-context.json`](agent-context.json) | | 多 agent 抢改任务文件 | 多个 agent 并发时抢改同一个看板/进度文件，出现"读到旧版本"、ID 撞号、合并冲突 | 一任务一文件（默认模式已内建），执行记录进任务文件，不逐任务改共享收尾文件 | [`tasks/README.md`](tasks/README.md) | -| 多 agent 重复领取 | 两个 agent 同时把同一 Issue 改为 doing，读回后都以为成功 | 用唯一 claim 分支做原子锁，标签只展示状态 | [`gitea-collaboration.md`](gitea-collaboration.md) | +| 多 agent 重复领取 | 两个 agent 同时把同一 Issue 改为 doing，读回后都以为成功 | 由 dispatcher 串行分配，claim 分支只作防御性标记，标签只展示状态 | [`gitea-collaboration.md`](gitea-collaboration.md) | | 多 agent 写路径碰撞 | 不同任务同时修改同一目录或共享配置，合并时才发现冲突 | dispatcher 串行声明 / 比较 `write_paths`，worker 使用独立 worktree | [`gitea-collaboration.md`](gitea-collaboration.md) + [`tasks/README.md`](tasks/README.md) | +| claim 长期占用 | Issue 仍 doing，但 agent 已退出或分支无活动，后续任务无法分配 | 只读审计 `lease_until`，人工核实后回收，不自动抢占 | [`gitea-collaboration.md`](gitea-collaboration.md) + [`../scripts/audit_gitea_coordination.py`](../scripts/audit_gitea_coordination.py) | +| 规则悄悄漂移 | 导航、任务元数据、模板或敏感配置在多轮提交后不一致 | 用同一离线治理命令在本地和 Actions 检查 | [`../scripts/validate_harness_governance.py`](../scripts/validate_harness_governance.py) | ## 使用原则 diff --git a/docs/tasks/README.md b/docs/tasks/README.md index 94c1276..309fde3 100644 --- a/docs/tasks/README.md +++ b/docs/tasks/README.md @@ -40,6 +40,7 @@ write_paths: # 允许修改的仓库相对路径 ## 方案 ## 验收要点 ## 边界（不改什么） +## 协作约束 ## 执行记录 ``` @@ -52,7 +53,7 @@ write_paths: # 允许修改的仓库相对路径 - **执行记录写进本任务文件的 `## 执行记录` 一节**（改了什么、跑了什么验证、结果、决策）——不逐任务追加共享的 `progress.md`（可选历史归档）、也不逐任务覆盖 `current-state.md`（项目级快照，只在启动/验证路径、目录结构或 blocker 变化时更新）。 - **只改自己那个任务文件**；不要编辑别人正在做的任务文件。 -未启用 Gitea 时，在独立分支 / worktree 中把任务文件从 `TODO` 改为 `DOING` 即可。启用 Gitea 时，必须先按 [`../gitea-collaboration.md`](../gitea-collaboration.md) 原子创建唯一 `claims/T-<编号>` 分支；assignee、标签和“更新后读回”都不能代替领取锁。 +未启用 Gitea 时，在独立分支 / worktree 中把任务文件从 `TODO` 改为 `DOING` 即可。启用 Gitea 时，必须先按 [`../gitea-collaboration.md`](../gitea-collaboration.md) 由 dispatcher 串行分配并创建 `claims/T-<编号>` 防御性标记；assignee、标签、读回和普通 create-branch API 都不能单独提供并发互斥。 ## 与 Gitea Issue / PR 的映射（可选） @@ -60,7 +61,7 @@ write_paths: # 允许修改的仓库相对路径 - 先把任务文件合入默认分支，再创建 Issue；随后把 Issue 编号回填任务文件并合入默认分支，最后才添加 `status/todo`。映射未完成的 Issue 不可领取。 - Issue、claim 分支、工作分支和 PR 都携带同一个 `T-<编号>`；不得用一个 PR 顺带完成多个任务。 - 工作分支命名为 `agent//T-<编号>`，每个 agent 使用独立 worktree。 -- MVP 由一个 dispatcher / 主 agent 串行分配任务。唯一 claim 分支保证同一任务不被重复领取，dispatcher 的串行检查保证不同任务的 `write_paths` 不冲突。 +- MVP 由一个 dispatcher / 主 agent 串行分配任务，以此保证同一任务不被重复领取，并保证不同任务的 `write_paths` 不冲突。唯一 claim 分支只拦截顺序重试和常见旁路，不能替代 dispatcher 的互斥保证。 - 进入评审后 Issue 使用唯一 `status/review`；PR 合并且默认分支任务文件为 `DONE` 后，Issue 才能关闭并标记 `status/done`。 - 领取、结构化 claim 评论、过期锁回收和分支清理的完整规则见 [`../gitea-collaboration.md`](../gitea-collaboration.md)。 @@ -95,4 +96,4 @@ write_paths: # 允许修改的仓库相对路径 - [`../06-tasks.md`](../06-tasks.md)：只读路线图（Phase 划分、里程碑、Backlog、建议拆分清单）；未启用 Gitea 时以任务文件 frontmatter 为准，启用后以 Issue 为实时状态、合并后的任务文件为长期事实。 - `../../progress.md`：可选工件，用作历史归档或项目级大事记；执行记录写各任务文件，不逐任务追加。 - [`../current-state.md`](../current-state.md)：项目级快照（启动/验证路径、目录要点、blocker）；任务状态不在此维护，可由脚本汇总 frontmatter 生成。 -- [`../gitea-collaboration.md`](../gitea-collaboration.md)：启用 Gitea 时的任务映射、原子领取、写路径防撞和 PR 状态协议。 +- [`../gitea-collaboration.md`](../gitea-collaboration.md)：启用 Gitea 时的任务映射、串行分配、防重复 claim、写路径防撞和 PR 状态协议。 diff --git a/gitea.env.example b/gitea.env.example index 1cfb3fc..1e927e1 100644 --- a/gitea.env.example +++ b/gitea.env.example @@ -2,9 +2,11 @@ GITEA_URL=http://gitea.example.invalid:3000 GITEA_TOKEN=【replace-with-minimum-scope-token】 +# Optional non-secret identity used to verify dispatcher-authored CLAIM comments. +GITEA_DISPATCHER_LOGIN=【dispatcher-gitea-login】 + # Required only when the team explicitly accepts HTTP credential exposure risk. GITEA_ALLOW_INSECURE_HTTP=1 # Optional: bypass local HTTP/SOCKS proxy for this Gitea instance. GITEA_DIRECT=1 - diff --git a/scripts/audit_gitea_coordination.py b/scripts/audit_gitea_coordination.py new file mode 100644 index 0000000..ae8ed93 --- /dev/null +++ b/scripts/audit_gitea_coordination.py @@ -0,0 +1,688 @@ +#!/usr/bin/env python3 +"""Read-only audit of Harness Coding task coordination in Gitea.""" + +from __future__ import annotations + +import argparse +import base64 +import os +import re +import sys +import urllib.parse +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from setup_gitea_labels import ApiError, GiteaClient, LABELS, validate_config +from validate_harness_governance import ( + TASK_ID, + is_safe_repo_path, + parse_frontmatter, + parse_frontmatter_text, + scopes_overlap, +) + + +STATUS_LABELS = { + "status/todo", + "status/doing", + "status/blocked", + "status/review", + "status/done", +} +ACTIVE_LABELS = {"status/doing", "status/blocked", "status/review"} +READY_OR_ACTIVE_LABELS = ACTIVE_LABELS | {"status/todo"} +TASK_IN_TITLE = re.compile(r"^\[(T-\d{3}[a-z]?)\]") +BODY_TASK_ID = re.compile(r"(?m)^\s*-\s*task_id:\s*`?(T-\d{3}[a-z]?)`?\s*$") +BODY_TASK_FILE = re.compile( + r"(?m)^\s*-\s*task_file:\s*`?(docs/tasks/T-\d{3}[a-z]?\.md)`?\s*$" +) +FIELD = re.compile(r"(?m)^\s*(?:-\s*)?([a-z_]+):\s*`?([^`\r\n]+?)`?\s*$") +MAX_LEASE = timedelta(hours=24) +CLOCK_SKEW = timedelta(minutes=5) +CLAIM_IDENTITY_FIELDS = ( + "task", + "claimed_by", + "allocated_by", + "context_ref", + "claim_branch", + "work_branch", +) + + +@dataclass(frozen=True, order=True) +class AuditFinding: + issue: int + rule: str + message: str + + def render(self) -> str: + subject = "repository" if self.issue <= 0 else f"issue #{self.issue}" + return f"ERROR [{self.rule}] {subject}: {self.message}" + + +@dataclass +class RemoteTask: + number: int + task_id: str + state: str + status: str | None + labels: set[str] + body: str + work_branch: str | None = None + context_ref: str | None = None + claimed_by: str | None = None + claimed_at: datetime | None = None + lease_until: datetime | None = None + write_paths: list[str] | None = None + + +def label_names(item: dict[str, Any]) -> set[str]: + labels = item.get("labels") + if not isinstance(labels, list): + return set() + return { + label["name"] + for label in labels + if isinstance(label, dict) and isinstance(label.get("name"), str) + } + + +def paged(client: GiteaClient, path: str) -> list[dict[str, Any]]: + result: list[dict[str, Any]] = [] + seen_pages: set[tuple[str, ...]] = set() + page = 1 + separator = "&" if "?" in path else "?" + while True: + values = client.request("GET", f"{path}{separator}limit=50&page={page}") + if not isinstance(values, list): + raise RuntimeError("Gitea 分页响应格式异常。") + if not values: + return result + if not all(isinstance(value, dict) for value in values): + raise RuntimeError("Gitea 分页响应包含非对象条目。") + signature = tuple( + str(value.get("id") or value.get("number") or value.get("name")) + for value in values + ) + if signature in seen_pages or page > 1000: + raise RuntimeError("Gitea 分页重复，已停止以避免无限读取。") + seen_pages.add(signature) + result.extend(value for value in values if isinstance(value, dict)) + page += 1 + + +def parse_fields(text: str) -> dict[str, str]: + return {match.group(1): match.group(2).strip() for match in FIELD.finditer(text)} + + +def parse_write_paths(text: str) -> list[str]: + lines = text.splitlines() + values: list[str] = [] + collecting = False + for line in lines: + if re.match(r"^\s*(?:-\s*)?write_paths:\s*$", line): + collecting = True + continue + if collecting: + item = re.match(r"^\s+-\s+`?([^`\r\n]+?)`?\s*$", line) + if item: + value = item.group(1).strip() + if "【" not in value: + values.append(value) + continue + if line.strip(): + break + return values + + +def parse_datetime(value: str | None) -> datetime | None: + if not value or "【" in value: + return None + if not re.fullmatch( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})", + value, + ): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def select_latest_claim( + comments: list[dict[str, Any]], task_id: str, dispatcher_login: str | None +) -> tuple[str | None, list[str]]: + required = { + "task", + "claimed_by", + "allocated_by", + "context_ref", + "claim_branch", + "work_branch", + "claimed_at", + "lease_until", + } + if not dispatcher_login: + return None, ["未配置可信 dispatcher Gitea 登录名，无法验证 CLAIM 作者。"] + + claims: list[tuple[int, str, dict[str, str], str, str]] = [] + for index, comment in enumerate(comments): + body = comment.get("body") + if not isinstance(body, str): + continue + lines = body.strip().splitlines() + if not lines or lines[0].strip() not in {"CLAIM", "CLAIM RENEWAL"}: + continue + marker = lines[0].strip() + fields = parse_fields(body) + if fields.get("task") != task_id or not required.issubset(fields): + continue + if not parse_write_paths(body): + continue + user = comment.get("user") + author = str(user.get("login") or "") if isinstance(user, dict) else "" + comment_id = comment.get("id") + order = comment_id if isinstance(comment_id, int) else index + claims.append((order, marker, fields, author, body)) + + selected: str | None = None + identity: tuple[str, ...] | None = None + errors: list[str] = [] + for _, marker, fields, author, body in sorted(claims, key=lambda item: item[0]): + if author != dispatcher_login or fields.get("allocated_by") != dispatcher_login: + errors.append("CLAIM 必须由配置的 dispatcher 账号发布，且 allocated_by 与作者一致。") + continue + candidate_identity = tuple(fields.get(key, "") for key in CLAIM_IDENTITY_FIELDS) + if marker == "CLAIM": + if identity is not None: + errors.append("同一任务存在重复的初始 CLAIM。") + continue + identity = candidate_identity + selected = body + continue + if identity is None: + errors.append("CLAIM RENEWAL 之前缺少有效初始 CLAIM。") + continue + if candidate_identity != identity: + errors.append("CLAIM RENEWAL 改变了任务、领取者、dispatcher 或分支身份字段。") + continue + selected = body + return selected, sorted(set(errors)) + + +def latest_claim( + comments: list[dict[str, Any]], task_id: str, dispatcher_login: str | None +) -> str | None: + return select_latest_claim(comments, task_id, dispatcher_login)[0] + + +def pull_request_head(pr: dict[str, Any]) -> str: + head = pr.get("head") + return str(head.get("ref") or "") if isinstance(head, dict) else "" + + +def branch_commits(branches: list[dict[str, Any]]) -> dict[str, str]: + result: dict[str, str] = {} + for branch in branches: + name = branch.get("name") + commit = branch.get("commit") + if isinstance(name, str) and isinstance(commit, dict) and isinstance( + commit.get("id"), str + ): + result[name] = commit["id"] + return result + + +def validate_pull_request( + pr: dict[str, Any], task: RemoteTask +) -> list[AuditFinding]: + findings: list[AuditFinding] = [] + if not task.work_branch or pull_request_head(pr) != task.work_branch: + findings.append( + AuditFinding(task.number, "pull-request", "PR head 与 claim 工作分支不一致。") + ) + body = pr.get("body") + body = body if isinstance(body, str) else "" + required_values = [f"docs/tasks/{task.task_id}.md"] + if task.context_ref: + required_values.append(task.context_ref) + required_values.extend(task.write_paths or []) + if not re.search(rf"(?i)\bCloses\s+#{task.number}\b", body): + findings.append(AuditFinding(task.number, "pull-request", "PR body 未链接对应 Issue。")) + if any(value not in body for value in required_values): + findings.append( + AuditFinding(task.number, "pull-request", "PR body 缺少任务、context_ref 或写路径。") + ) + return findings + + +def remote_task_metadata( + client: GiteaClient, task_id: str, branch: str +) -> tuple[dict[str, Any] | None, list[str]]: + file_path = urllib.parse.quote(f"docs/tasks/{task_id}.md", safe="/") + ref = urllib.parse.quote(branch, safe="") + try: + response = client.request("GET", f"/contents/{file_path}?ref={ref}") + except ApiError as exc: + if exc.status == 404: + return None, ["工作分支缺少任务文件。"] + raise + if not isinstance(response, dict) or not isinstance(response.get("content"), str): + return None, ["工作分支任务文件响应格式异常。"] + try: + text = base64.b64decode(response["content"]).decode("utf-8") + except (ValueError, UnicodeDecodeError): + return None, ["工作分支任务文件不是有效 UTF-8 / base64。"] + metadata, _, errors = parse_frontmatter_text(text) + return metadata, errors + + +def local_tasks(root: Path) -> dict[str, dict[str, Any]]: + result: dict[str, dict[str, Any]] = {} + task_dir = root / "docs" / "tasks" + if not task_dir.is_dir(): + return result + for path in sorted(task_dir.glob("T-*.md")): + metadata, _, _ = parse_frontmatter(path) + task_id = metadata.get("id") + if isinstance(task_id, str) and TASK_ID.fullmatch(task_id): + result[task_id] = metadata + return result + + +def audit_labels(client: GiteaClient) -> list[AuditFinding]: + findings: list[AuditFinding] = [] + existing = client.list_labels() + for desired in LABELS: + current = existing.get(desired["name"]) + if current is None: + findings.append(AuditFinding(0, "labels", f"缺少 {desired['name']}。")) + elif bool(current.get("exclusive")) != desired["exclusive"]: + findings.append(AuditFinding(0, "labels", f"{desired['name']} exclusive 属性不一致。")) + return findings + + +def audit_repository( + root: Path, + client: GiteaClient, + now: datetime, + dispatcher_login: str | None = None, +) -> tuple[list[AuditFinding], int]: + findings = audit_labels(client) + issues = [ + issue + for issue in paged(client, "/issues?state=all&type=issues") + if "kind/task" in label_names(issue) and not issue.get("pull_request") + ] + branches = branch_commits(paged(client, "/branches")) + pull_requests = paged(client, "/pulls?state=all") + local = local_tasks(root) + remote: dict[str, RemoteTask] = {} + + for issue in issues: + number = issue.get("number") + title = issue.get("title") + body = issue.get("body") + state = issue.get("state") + if not isinstance(number, int) or not isinstance(title, str): + continue + body = body if isinstance(body, str) else "" + title_match = TASK_IN_TITLE.search(title) + body_match = BODY_TASK_ID.search(body) + task_id = title_match.group(1) if title_match else "" + if not task_id: + findings.append(AuditFinding(number, "mapping", "标题缺少 [T-编号]。")) + if body_match is None or body_match.group(1) != task_id: + findings.append(AuditFinding(number, "mapping", "正文 task_id 与标题不一致。")) + task_file_match = BODY_TASK_FILE.search(body) + if task_id and ( + task_file_match is None + or task_file_match.group(1) != f"docs/tasks/{task_id}.md" + ): + findings.append(AuditFinding(number, "mapping", "task_file 与任务 ID 不一致。")) + if task_id in remote: + findings.append(AuditFinding(number, "mapping", "任务 ID 映射到多个 Issue。")) + + labels = label_names(issue) + statuses = sorted(labels & STATUS_LABELS) + status = statuses[0] if len(statuses) == 1 else None + local_issue = local.get(task_id, {}).get("issue") if task_id else None + if len(statuses) > 1: + findings.append(AuditFinding(number, "status", "存在多个 status/* 标签。")) + elif not statuses and local_issue == number: + findings.append(AuditFinding(number, "status", "已映射任务缺少 status/* 标签。")) + if status: + if task_id not in local or local_issue != number: + findings.append(AuditFinding(number, "mapping", "可领取 Issue 未映射默认分支任务文件。")) + else: + local_status = local[task_id].get("status") + expected_local = "DONE" if status == "status/done" else "TODO" + if local_status != expected_local: + findings.append( + AuditFinding( + number, + "status", + f"远端 {status} 要求默认分支任务为 {expected_local}。", + ) + ) + if status in READY_OR_ACTIVE_LABELS: + deps = local[task_id].get("deps") + if not isinstance(deps, list): + findings.append( + AuditFinding(number, "dependency", "默认分支任务 deps 不是列表。") + ) + else: + unready = sorted( + str(dep) + for dep in deps + if not isinstance(dep, str) + or dep not in local + or local[dep].get("status") != "DONE" + ) + if unready: + findings.append( + AuditFinding( + number, + "dependency", + "任务依赖尚未全部 DONE：" + ", ".join(unready) + "。", + ) + ) + for prefix in ("type/", "priority/"): + scoped = [name for name in labels if name.startswith(prefix)] + if len(scoped) != 1: + findings.append( + AuditFinding(number, "labels", f"可领取任务必须恰有一个 {prefix} 标签。") + ) + if status == "status/done" and state != "closed": + findings.append(AuditFinding(number, "status", "status/done 的 Issue 必须关闭。")) + if status != "status/done" and state == "closed": + findings.append(AuditFinding(number, "status", "未完成 Issue 不应关闭。")) + + task = RemoteTask( + number=number, + task_id=task_id, + state=state if isinstance(state, str) else "", + status=status, + labels=labels, + body=body, + write_paths=parse_write_paths(body), + ) + if task_id: + remote[task_id] = task + + local_metadata = local.get(task_id, {}) + if status == "status/done": + work_branch = local_metadata.get("work_branch") + context_ref = local_metadata.get("context_ref") + write_paths = local_metadata.get("write_paths") + task.work_branch = work_branch if isinstance(work_branch, str) else None + task.context_ref = context_ref if isinstance(context_ref, str) else None + task.write_paths = ( + [value for value in write_paths if isinstance(value, str)] + if isinstance(write_paths, list) + else [] + ) + if not task.work_branch or not task.context_ref or not task.write_paths: + findings.append( + AuditFinding(number, "task-file", "DONE 任务缺少长期分支、context_ref 或写路径。") + ) + + claim_branch = f"claims/{task_id}" if task_id else "" + if status in ACTIVE_LABELS: + if claim_branch not in branches: + findings.append(AuditFinding(number, "claim", "活跃任务缺少 claim 分支。")) + comments = paged(client, f"/issues/{number}/comments") + claim, claim_errors = select_latest_claim( + comments, task_id, dispatcher_login + ) + for message in claim_errors: + findings.append(AuditFinding(number, "claim-author", message)) + if claim is None: + findings.append( + AuditFinding(number, "claim", "活跃任务缺少由可信 dispatcher 发布的结构化 CLAIM。") + ) + else: + fields = parse_fields(claim) + task.work_branch = fields.get("work_branch") + task.context_ref = fields.get("context_ref") + task.claimed_by = fields.get("claimed_by") + task.claimed_at = parse_datetime(fields.get("claimed_at")) + task.lease_until = parse_datetime(fields.get("lease_until")) + claim_paths = parse_write_paths(claim) + if claim_paths: + task.write_paths = claim_paths + required_claim = { + "task", + "claimed_by", + "allocated_by", + "context_ref", + "claim_branch", + "work_branch", + "claimed_at", + "lease_until", + } + missing_claim = sorted(required_claim - set(fields)) + if missing_claim: + findings.append( + AuditFinding( + number, + "claim", + "CLAIM 缺少字段：" + ", ".join(missing_claim) + "。", + ) + ) + if fields.get("task") != task_id: + findings.append(AuditFinding(number, "claim", "CLAIM task 不一致。")) + if fields.get("claim_branch") != claim_branch: + findings.append(AuditFinding(number, "claim", "CLAIM claim_branch 不一致。")) + context_ref = task.context_ref + if not context_ref or not re.fullmatch(r"[0-9a-fA-F]{40}", context_ref): + findings.append(AuditFinding(number, "claim", "CLAIM context_ref 无效。")) + elif branches.get(claim_branch) != context_ref: + findings.append(AuditFinding(number, "claim", "claim 分支 SHA 与 context_ref 不一致。")) + if ( + not task.claimed_by + or not re.fullmatch(r"[A-Za-z0-9._-]+", task.claimed_by) + or task.work_branch != f"agent/{task.claimed_by}/{task_id}" + ): + findings.append(AuditFinding(number, "claim", "claimed_by 与工作分支命名不一致。")) + if ( + not claim_paths + or len(claim_paths) != len(set(claim_paths)) + or any(not is_safe_repo_path(path) for path in claim_paths) + or f"docs/tasks/{task_id}.md" not in claim_paths + ): + findings.append( + AuditFinding( + number, + "claim", + "CLAIM write_paths 必须安全、唯一并包含任务文件。", + ) + ) + if not task.work_branch or task.work_branch not in branches: + findings.append(AuditFinding(number, "claim", "工作分支不存在。")) + elif task.work_branch: + metadata, metadata_errors = remote_task_metadata( + client, task_id, task.work_branch + ) + for _ in metadata_errors: + findings.append(AuditFinding(number, "task-file", "工作分支任务文件无效。")) + if metadata is not None: + expected_status = { + "status/doing": "DOING", + "status/blocked": "BLOCKED", + "status/review": "DONE", + }.get(status) + comparisons = { + "id": task_id, + "issue": number, + "context_ref": context_ref, + "claim_branch": claim_branch, + "work_branch": task.work_branch, + "status": expected_status, + } + for key, expected in comparisons.items(): + if metadata.get(key) != expected: + findings.append( + AuditFinding( + number, + "task-file", + f"工作分支任务字段 {key} 与协调状态不一致。", + ) + ) + metadata_paths = metadata.get("write_paths") + if not isinstance(metadata_paths, list) or set(metadata_paths) != set( + claim_paths + ): + findings.append( + AuditFinding( + number, + "task-file", + "工作分支 write_paths 与 CLAIM 不一致。", + ) + ) + if task.claimed_at is None: + findings.append(AuditFinding(number, "stale", "CLAIM 缺少有效 claimed_at。")) + elif task.claimed_at > now + CLOCK_SKEW: + findings.append(AuditFinding(number, "stale", "claimed_at 超出允许时钟偏差。")) + if task.lease_until is None: + findings.append(AuditFinding(number, "stale", "CLAIM 缺少有效 lease_until。")) + elif task.claimed_at is not None: + if task.lease_until <= task.claimed_at: + findings.append(AuditFinding(number, "stale", "lease_until 必须晚于 claimed_at。")) + elif task.lease_until - task.claimed_at > MAX_LEASE: + findings.append(AuditFinding(number, "stale", "claim 租期不得超过 24 小时。")) + if task.lease_until <= now: + findings.append(AuditFinding(number, "stale", "claim 已过期，需人工审查回收。")) + elif status == "status/todo" and claim_branch in branches: + findings.append(AuditFinding(number, "claim", "TODO 仍存在 claim 分支。")) + + matching_prs = [ + pr + for pr in pull_requests + if task_id and TASK_IN_TITLE.search(str(pr.get("title") or "")) + and TASK_IN_TITLE.search(str(pr.get("title") or "")).group(1) == task_id + ] + if status == "status/review": + open_prs = [pr for pr in matching_prs if pr.get("state") == "open"] + if len(open_prs) != 1: + findings.append( + AuditFinding(number, "pull-request", "status/review 必须恰有一个 open PR。") + ) + else: + findings.extend(validate_pull_request(open_prs[0], task)) + if status == "status/done": + merged_prs = [pr for pr in matching_prs if bool(pr.get("merged"))] + if len(merged_prs) != 1: + findings.append( + AuditFinding(number, "pull-request", "status/done 必须恰有一个 merged PR。") + ) + else: + findings.extend(validate_pull_request(merged_prs[0], task)) + + for task_id, metadata in sorted(local.items()): + issue_number = metadata.get("issue") + if isinstance(issue_number, int): + mapped = remote.get(task_id) + if mapped is None or mapped.number != issue_number: + findings.append(AuditFinding(issue_number, "mapping", f"本地 {task_id} 没有唯一远端映射。")) + elif metadata.get("status") == "DONE" and mapped.status != "status/done": + findings.append(AuditFinding(issue_number, "status", "本地 DONE 与远端状态不一致。")) + + active = sorted( + (task for task in remote.values() if task.status in ACTIVE_LABELS), + key=lambda task: task.task_id, + ) + workers: dict[str, str] = {} + for task in active: + if not task.claimed_by: + continue + previous = workers.get(task.claimed_by) + if previous: + findings.append( + AuditFinding( + task.number, + "worker-overlap", + f"claimed_by 同时活跃于 {previous} 和 {task.task_id}。", + ) + ) + else: + workers[task.claimed_by] = task.task_id + for index, left in enumerate(active): + for right in active[index + 1 :]: + if any( + scopes_overlap(a, b) + for a in left.write_paths or [] + for b in right.write_paths or [] + ): + findings.append( + AuditFinding( + right.number, + "scope-overlap", + f"活跃 write_paths 与 {left.task_id} 重叠。", + ) + ) + return sorted(set(findings)), len(issues) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="只读审计 Gitea 任务协调状态。") + parser.add_argument("--repo", help="目标 owner/repo；也可设置 GITEA_REPOSITORY。") + parser.add_argument( + "--dispatcher", + default=os.environ.get("GITEA_DISPATCHER_LOGIN"), + help="可信 dispatcher 的 Gitea 登录名；也可设置 GITEA_DISPATCHER_LOGIN。", + ) + parser.add_argument( + "--root", + type=Path, + default=Path(__file__).resolve().parents[1], + help="本地仓库根目录。", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + root = args.root.resolve() + if not root.is_dir() or not (root / "docs" / "tasks").is_dir(): + print("ERROR: --root 必须是包含 docs/tasks 的仓库目录。", file=sys.stderr) + return 2 + repo = args.repo or os.environ.get("GITEA_REPOSITORY") + if not repo: + print("ERROR: 需要 --repo owner/repo 或 GITEA_REPOSITORY。", file=sys.stderr) + return 2 + try: + root_url, owner, name = validate_config( + os.environ.get("GITEA_URL", ""), + os.environ.get("GITEA_TOKEN", ""), + repo, + ) + client = GiteaClient(root_url, owner, name, os.environ["GITEA_TOKEN"]) + findings, count = audit_repository( + root, client, datetime.now(timezone.utc), args.dispatcher + ) + except ValueError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + except ApiError as exc: + print(f"ERROR: Gitea 只读审计失败（HTTP {exc.status}）。", file=sys.stderr) + return 2 + except RuntimeError: + print("ERROR: Gitea 只读审计失败（网络、代理或响应格式异常）。", file=sys.stderr) + return 2 + if findings: + for finding in findings: + print(finding.render(), file=sys.stderr) + print(f"Gitea 协调审计失败：{len(findings)} 项不一致。", file=sys.stderr) + return 1 + print(f"Gitea 协调审计通过：检查 {count} 个 kind/task Issue，未执行远端写入。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/setup_gitea_labels.py b/scripts/setup_gitea_labels.py index 6c0b9b9..8d54c3b 100644 --- a/scripts/setup_gitea_labels.py +++ b/scripts/setup_gitea_labels.py @@ -171,16 +171,23 @@ class GiteaClient: def list_labels(self) -> dict[str, dict[str, Any]]: result: dict[str, dict[str, Any]] = {} + seen_pages: set[tuple[str, ...]] = set() page = 1 while True: labels = self.request("GET", f"/labels?limit=50&page={page}") if not isinstance(labels, list): raise RuntimeError("Gitea labels 响应格式异常。") + if not labels: + return result + if not all(isinstance(label, dict) for label in labels): + raise RuntimeError("Gitea labels 响应包含非对象条目。") + signature = tuple(str(label.get("id") or label.get("name")) for label in labels) + if signature in seen_pages or page > 1000: + raise RuntimeError("Gitea labels 分页重复，已停止以避免无限读取。") + seen_pages.add(signature) for label in labels: if isinstance(label, dict) and isinstance(label.get("name"), str): result[label["name"]] = label - if len(labels) < 50: - return result page += 1 diff --git a/scripts/test_gitea_claim_race.py b/scripts/test_gitea_claim_race.py new file mode 100644 index 0000000..ea82465 --- /dev/null +++ b/scripts/test_gitea_claim_race.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +"""Compatibility smoke for Gitea's same-name claim branch race behavior.""" + +from __future__ import annotations + +import argparse +import os +import sys +import threading +import urllib.parse +import uuid +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from typing import Any + +from setup_gitea_labels import ApiError, GiteaClient, validate_config + + +PROBE_PREFIX = "claims/__probe__/race-" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="并发创建唯一临时 claim 分支，smoke 期望一个 201、一个 409。" + ) + parser.add_argument("--repo", help="目标 owner/repo；也可设置 GITEA_REPOSITORY。") + parser.add_argument( + "--apply", + action="store_true", + help="执行两次写入并清理临时分支；省略时只读并打印计划。", + ) + return parser.parse_args() + + +def new_probe_branch() -> str: + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return f"{PROBE_PREFIX}{stamp}-{uuid.uuid4().hex[:12]}" + + +def branch_commit(client: GiteaClient, branch: str) -> str | None: + encoded = urllib.parse.quote(branch, safe="") + try: + response = client.request("GET", f"/branches/{encoded}") + except ApiError as exc: + if exc.status == 404: + return None + raise + if not isinstance(response, dict): + raise RuntimeError("Gitea branch 响应格式异常。") + commit = response.get("commit") + if not isinstance(commit, dict) or not isinstance(commit.get("id"), str): + raise RuntimeError("Gitea branch 响应缺少 commit.id。") + return commit["id"] + + +def repository_base(client: GiteaClient) -> tuple[str, str]: + repository = client.request("GET", "") + if not isinstance(repository, dict) or not isinstance( + repository.get("default_branch"), str + ): + raise RuntimeError("Gitea repository 响应缺少 default_branch。") + default_branch = repository["default_branch"] + commit = branch_commit(client, default_branch) + if commit is None: + raise RuntimeError("默认分支不存在。") + return default_branch, commit + + +def create_once(client: GiteaClient, barrier: threading.Barrier, branch: str, ref: str) -> int: + barrier.wait(timeout=10) + try: + client.request( + "POST", + "/branches", + {"new_branch_name": branch, "old_ref_name": ref}, + ) + return 201 + except ApiError as exc: + return exc.status + + +def cleanup_probe(client: GiteaClient, branch: str, expected_sha: str) -> None: + if not branch.startswith(PROBE_PREFIX): + raise RuntimeError("拒绝清理非探针分支。") + actual_sha = branch_commit(client, branch) + if actual_sha is None: + return + if actual_sha != expected_sha: + raise RuntimeError("探针分支 SHA 与预期不一致，已保留供人工检查。") + encoded = urllib.parse.quote(branch, safe="") + client.request("DELETE", f"/branches/{encoded}") + if branch_commit(client, branch) is not None: + raise RuntimeError("探针分支清理后仍然存在。") + + +def client_for(root: str, owner: str, repo: str, token: str) -> GiteaClient: + return GiteaClient(root, owner, repo, token) + + +def run_probe(root: str, owner: str, repo: str, token: str, branch: str, sha: str) -> list[int]: + barrier = threading.Barrier(2) + clients = [client_for(root, owner, repo, token) for _ in range(2)] + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit(create_once, client, barrier, branch, sha) for client in clients + ] + return sorted(future.result(timeout=40) for future in futures) + + +def main() -> int: + args = parse_args() + repo_value = args.repo or os.environ.get("GITEA_REPOSITORY") + if not repo_value: + print("ERROR: 需要 --repo owner/repo 或 GITEA_REPOSITORY。", file=sys.stderr) + return 2 + token = os.environ.get("GITEA_TOKEN", "") + try: + root, owner, repo = validate_config( + os.environ.get("GITEA_URL", ""), token, repo_value + ) + control = client_for(root, owner, repo, token) + default_branch, sha = repository_base(control) + branch = new_probe_branch() + print(f"目标仓库：{repo_value}") + print(f"基准分支：{default_branch} @ {sha}") + print(f"临时分支：{branch}") + if not args.apply: + print("dry-run：未写入；追加 --apply 才会执行竞态探针和受控清理。") + return 0 + + results: list[int] = [] + probe_error: Exception | None = None + try: + results = run_probe(root, owner, repo, token, branch, sha) + except Exception as exc: # cleanup still has to run after partial writes + probe_error = exc + try: + cleanup_probe(control, branch, sha) + except (ApiError, RuntimeError) as cleanup_error: + print(f"ERROR: 清理失败：{cleanup_error}", file=sys.stderr) + return 2 + if probe_error is not None: + print("ERROR: 竞态请求未完整返回；临时分支已安全清理。", file=sys.stderr) + return 2 + print("竞态结果：" + ", ".join(str(status) for status in results)) + if results != [201, 409]: + print( + "ERROR: 未得到恰好一个 201 和一个 409；目标实例不符合预期 smoke，临时分支已安全清理。", + file=sys.stderr, + ) + return 1 + print("claim 并发兼容性 smoke 通过；这不证明线性化，临时分支已删除并确认 404。") + return 0 + except ValueError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + except ApiError as exc: + print(f"ERROR: Gitea 探针失败（HTTP {exc.status}）。", file=sys.stderr) + return 2 + except RuntimeError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validate_agent_context.py b/scripts/validate_agent_context.py index b1e6965..615d693 100644 --- a/scripts/validate_agent_context.py +++ b/scripts/validate_agent_context.py @@ -3,6 +3,7 @@ from __future__ import annotations +import argparse import json import re import sys @@ -10,8 +11,6 @@ from pathlib import Path, PurePosixPath from typing import Any -ROOT = Path(__file__).resolve().parents[1] -MANIFEST = ROOT / "docs" / "agent-context.json" EXPECTED_SCHEMA = "docs/agent-context.schema.json" REQUIRED_TOP_LEVEL = { "schema", @@ -34,18 +33,25 @@ SENSITIVE_KEY = re.compile(r"(?:token|password|secret|credential)", re.IGNORECAS URI_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:") -def load_json(path: Path, errors: list[str]) -> Any: +def load_json(path: Path, root: Path, errors: list[str]) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: - errors.append(f"文件不存在：{path.relative_to(ROOT).as_posix()}") + errors.append(f"文件不存在：{display_path(path, root)}") except json.JSONDecodeError as exc: errors.append( - f"JSON 语法错误：{path.relative_to(ROOT).as_posix()}:{exc.lineno}:{exc.colno}" + f"JSON 语法错误：{display_path(path, root)}:{exc.lineno}:{exc.colno}" ) return None +def display_path(path: Path, root: Path) -> str: + try: + return path.relative_to(root).as_posix() + except ValueError: + return path.as_posix() + + def require_mapping(value: Any, name: str, errors: list[str]) -> dict[str, Any]: if not isinstance(value, dict): errors.append(f"{name} 必须是对象。") @@ -64,7 +70,7 @@ def require_string_list(value: Any, name: str, errors: list[str]) -> list[str]: return value -def validate_repo_path(value: str, name: str, errors: list[str]) -> None: +def validate_repo_path(root: Path, value: str, name: str, errors: list[str]) -> None: path = PurePosixPath(value) if ( path.is_absolute() @@ -75,7 +81,7 @@ def validate_repo_path(value: str, name: str, errors: list[str]) -> None: errors.append(f"{name} 必须是安全的仓库相对路径：{value}") return - target = ROOT.joinpath(*path.parts) + target = root.joinpath(*path.parts) if not target.exists(): errors.append(f"{name} 引用路径不存在：{value}") @@ -92,34 +98,36 @@ def find_sensitive_keys(value: Any, location: str, errors: list[str]) -> None: find_sensitive_keys(child, f"{location}[{index}]", errors) -def main() -> int: +def validate_manifest(root: Path) -> list[str]: + root = root.resolve() + manifest_path = root / "docs" / "agent-context.json" errors: list[str] = [] - manifest = load_json(MANIFEST, errors) - schema = load_json(ROOT / EXPECTED_SCHEMA, errors) + manifest = load_json(manifest_path, root, errors) + schema = load_json(root / EXPECTED_SCHEMA, root, errors) if manifest is None or schema is None: - return report(errors) + return errors if not isinstance(schema, dict) or schema.get("type") != "object": errors.append("agent-context.schema.json 不是有效的对象 Schema。") - root = require_mapping(manifest, "manifest", errors) - actual_keys = set(root) + root_object = require_mapping(manifest, "manifest", errors) + actual_keys = set(root_object) missing = sorted(REQUIRED_TOP_LEVEL - actual_keys) unexpected = sorted(actual_keys - REQUIRED_TOP_LEVEL) if missing: errors.append("缺少顶层字段：" + ", ".join(missing)) if unexpected: errors.append("存在未知顶层字段：" + ", ".join(unexpected)) - if root.get("schema") != EXPECTED_SCHEMA: + if root_object.get("schema") != EXPECTED_SCHEMA: errors.append(f"schema 必须是 {EXPECTED_SCHEMA}。") - if root.get("schema_version") != 1: + if root_object.get("schema_version") != 1: errors.append("schema_version 必须为 1。") - authority = require_mapping(root.get("authority"), "authority", errors) + authority = require_mapping(root_object.get("authority"), "authority", errors) for key in ("bootstrap", "framework_templates", "project_facts", "coordination"): if not isinstance(authority.get(key), str) or not authority[key]: errors.append(f"authority.{key} 必须是非空字符串。") - bootstrap = require_mapping(root.get("bootstrap"), "bootstrap", errors) + bootstrap = require_mapping(root_object.get("bootstrap"), "bootstrap", errors) always_read = require_string_list( bootstrap.get("always_read"), "bootstrap.always_read", errors ) @@ -127,7 +135,7 @@ def main() -> int: if missing_bootstrap: errors.append("bootstrap.always_read 缺少：" + ", ".join(missing_bootstrap)) - routes = require_mapping(root.get("routes"), "routes", errors) + routes = require_mapping(root_object.get("routes"), "routes", errors) if not routes: errors.append("routes 至少需要一个任务类型。") @@ -137,7 +145,7 @@ def main() -> int: paths = require_string_list(value, f"routes.{route}", errors) path_values.extend((path, f"routes.{route}") for path in paths) - tasks = require_mapping(root.get("tasks"), "tasks", errors) + tasks = require_mapping(root_object.get("tasks"), "tasks", errors) if set(tasks) != TASK_PATH_KEYS: errors.append("tasks 必须且只能包含 roadmap、directory、template。") for key in sorted(TASK_PATH_KEYS): @@ -147,7 +155,7 @@ def main() -> int: else: errors.append(f"tasks.{key} 必须是非空字符串。") - refresh = require_mapping(root.get("refresh"), "refresh", errors) + refresh = require_mapping(root_object.get("refresh"), "refresh", errors) expected_refresh = { "context_ref": "default_branch_head_sha", "cache_key": "file_sha", @@ -157,7 +165,7 @@ def main() -> int: if refresh != expected_refresh: errors.append("refresh 必须使用约定的提交 SHA 与文件 SHA 刷新策略。") - degraded = require_mapping(root.get("degraded_mode"), "degraded_mode", errors) + degraded = require_mapping(root_object.get("degraded_mode"), "degraded_mode", errors) expected_degraded = { "continue_claimed_task": True, "claim_new_task": False, @@ -167,25 +175,52 @@ def main() -> int: errors.append("degraded_mode 必须禁止领取新任务和写入远端状态。") for value, name in path_values: - validate_repo_path(value, name, errors) - find_sensitive_keys(root, "manifest", errors) + validate_repo_path(root, value, name, errors) + find_sensitive_keys(root_object, "manifest", errors) + return errors + +def manifest_summary(root: Path) -> tuple[int, int]: + manifest = json.loads( + (root / "docs" / "agent-context.json").read_text(encoding="utf-8") + ) + paths = {manifest["schema"]} + paths.update(manifest["bootstrap"]["always_read"]) + for values in manifest["routes"].values(): + paths.update(values) + paths.update(manifest["tasks"].values()) + return len(manifest["routes"]), len(paths) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="校验 Agent 上下文清单。") + parser.add_argument( + "--root", + type=Path, + default=Path(__file__).resolve().parents[1], + help="仓库根目录；默认取脚本上一级。", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + root = args.root.resolve() + if not root.is_dir(): + print("ERROR: 仓库根目录不存在。", file=sys.stderr) + return 2 + errors = validate_manifest(root) if errors: - return report(errors) - - unique_paths = {value for value, _ in path_values} + for error in errors: + print(f"ERROR: {error}", file=sys.stderr) + return 1 + route_count, path_count = manifest_summary(root) print( "agent-context 校验通过：" - f"{len(routes)} 个任务路由，{len(unique_paths)} 个有效仓库路径。" + f"{route_count} 个任务路由，{path_count} 个有效仓库路径。" ) return 0 -def report(errors: list[str]) -> int: - for error in errors: - print(f"ERROR: {error}", file=sys.stderr) - return 1 - - if __name__ == "__main__": raise SystemExit(main()) diff --git a/scripts/validate_harness_governance.py b/scripts/validate_harness_governance.py new file mode 100644 index 0000000..2ae6d96 --- /dev/null +++ b/scripts/validate_harness_governance.py @@ -0,0 +1,685 @@ +#!/usr/bin/env python3 +"""Offline governance checks for a Harness Coding repository.""" + +from __future__ import annotations + +import argparse +import re +import subprocess +import sys +import unicodedata +import urllib.parse +from dataclasses import dataclass +from datetime import date +from pathlib import Path, PurePosixPath +from typing import Any, Iterable + +from validate_agent_context import validate_manifest + + +TASK_ID = re.compile(r"^T-\d{3}[a-z]?$") +SHA40 = re.compile(r"^[0-9a-fA-F]{40}$") +MARKDOWN_LINK = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)") +URI_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:") +AUTH_VALUE = re.compile( + r"(?i)authorization\s*[:=]\s*['\"]?(?:basic|bearer|token)\s+[A-Za-z0-9._~+/=-]{8,}" +) +TOKEN_ASSIGNMENT = re.compile( + r"(?i)^\s*\{?\s*(?:(?:export\s+)?(?:\$env:)?GITEA_TOKEN|['\"]GITEA_TOKEN['\"])" + r"\s*[:=]\s*(.*?)\s*[,}]?\s*$" +) +URL_CREDENTIAL = re.compile(r"(?i)https?://[^/\s:@]+:[^/\s@]+@") +GITEA_TOKEN_LITERAL = re.compile(r"\bgta_[A-Za-z0-9_-]{16,}\b") +CMD_TOKEN_ASSIGNMENT = re.compile( + r"(?ix)^\s*(?:" + r"setx\s+(?:\"GITEA_TOKEN\"|GITEA_TOKEN)\s+(?:\"([^\"]*)\"|(.*?))" + r"|set\s+(?:\"GITEA_TOKEN\s*=\s*([^\"]*)\"|GITEA_TOKEN\s*=\s*(.*?))" + r")\s*$" +) +DOTNET_TOKEN_SETTER = re.compile( + r"(?is)\[Environment\]::SetEnvironmentVariable\s*\(\s*['\"]GITEA_TOKEN['\"]" + r"\s*,\s*(['\"])(.*?)\1" +) +SAFE_VARIABLE_REFERENCE = re.compile( + r"(?i)(?:\$\{[A-Za-z_][A-Za-z0-9_]*\}|\$env:[A-Za-z_][A-Za-z0-9_]*|" + r"\$[A-Za-z_][A-Za-z0-9_]*|%[A-Za-z_][A-Za-z0-9_]*%)" +) +TASK_REQUIRED_FIELDS = { + "id", + "title", + "phase", + "deps", + "status", + "created", + "issue", + "context_ref", + "claim_branch", + "work_branch", + "write_paths", +} +TASK_REQUIRED_SECTIONS = { + "问题 / 背景", + "方案", + "验收要点", + "边界（不改什么）", + "协作约束", + "执行记录", +} +VALID_STATUS = {"TODO", "DOING", "DONE", "BLOCKED"} +ACTIVE_STATUS = {"DOING", "BLOCKED"} +KNOWN_TEXT_SUFFIXES = { + ".md", + ".py", + ".ps1", + ".sh", + ".json", + ".yaml", + ".yml", + ".toml", + ".txt", + ".env", + ".example", +} + + +@dataclass(frozen=True, order=True) +class Finding: + rule: str + path: str + line: int + message: str + + def render(self) -> str: + location = self.path if self.line <= 0 else f"{self.path}:{self.line}" + return f"ERROR [{self.rule}] {location}: {self.message}" + + +@dataclass +class Task: + path: Path + metadata: dict[str, Any] + body: str + + @property + def task_id(self) -> str: + value = self.metadata.get("id") + return value if isinstance(value, str) else "" + + @property + def status(self) -> str: + value = self.metadata.get("status") + return value if isinstance(value, str) else "" + + +def relative(path: Path, root: Path) -> str: + return path.relative_to(root).as_posix() + + +def read_text(path: Path) -> str | None: + try: + data = path.read_bytes() + if data.startswith((b"\xff\xfe", b"\xfe\xff")): + return data.decode("utf-16") + return data.decode("utf-8-sig") + except (OSError, UnicodeDecodeError): + return None + + +def candidate_files(root: Path) -> list[Path]: + command = [ + "git", + "-C", + str(root), + "ls-files", + "--cached", + "--others", + "--exclude-standard", + "-z", + ] + try: + result = subprocess.run( + command, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) + names = [name for name in result.stdout.decode("utf-8").split("\0") if name] + return sorted(root / PurePosixPath(name) for name in names if (root / name).is_file()) + except (OSError, subprocess.CalledProcessError, UnicodeDecodeError): + return sorted( + path for path in root.rglob("*") if path.is_file() and ".git" not in path.parts + ) + + +def parse_scalar(value: str) -> Any: + value = value.split(" #", 1)[0].strip() + if not value or value.lower() in {"null", "~"}: + return None + if value == "[]": + return [] + if value.startswith("[") and value.endswith("]"): + inner = value[1:-1].strip() + return [] if not inner else [parse_scalar(item) for item in inner.split(",")] + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + value = value[1:-1] + if value.isdigit(): + return int(value) + return value + + +def parse_frontmatter(path: Path) -> tuple[dict[str, Any], str, list[str]]: + text = read_text(path) + if text is None: + return {}, "", ["文件不是 UTF-8 文本。"] + return parse_frontmatter_text(text) + + +def parse_frontmatter_text(text: str) -> tuple[dict[str, Any], str, list[str]]: + lines = text.splitlines() + if not lines or lines[0].strip() != "---": + return {}, text, ["缺少起始 frontmatter 分隔符。"] + try: + end = next(index for index in range(1, len(lines)) if lines[index].strip() == "---") + except StopIteration: + return {}, text, ["缺少结束 frontmatter 分隔符。"] + + metadata: dict[str, Any] = {} + current_list: str | None = None + errors: list[str] = [] + for number, raw in enumerate(lines[1:end], start=2): + if not raw.strip() or raw.lstrip().startswith("#"): + continue + item = re.match(r"^\s+-\s+(.+)$", raw) + if item and current_list: + metadata[current_list].append(parse_scalar(item.group(1))) + continue + field = re.match(r"^([A-Za-z_][A-Za-z0-9_-]*):(?:\s*(.*))?$", raw) + if not field: + errors.append(f"frontmatter 第 {number} 行语法不受支持。") + current_list = None + continue + key, raw_value = field.groups() + if key in metadata: + errors.append(f"frontmatter 字段重复：{key}。") + value = parse_scalar(raw_value or "") + if value is None and not (raw_value or "").strip(): + value = [] + current_list = key + else: + current_list = None + metadata[key] = value + return metadata, "\n".join(lines[end + 1 :]), errors + + +def is_safe_repo_path(value: str) -> bool: + path = PurePosixPath(value) + return bool(value) and value == value.strip() and not ( + path.is_absolute() + or ".." in path.parts + or "\\" in value + or URI_SCHEME.match(value) + or "【" in value + or any(character in value for character in "*?[]{}") + or any(ord(character) < 32 for character in value) + ) + + +def normalize_scope(value: str) -> tuple[str, ...]: + return tuple( + unicodedata.normalize("NFC", part).casefold() + for part in PurePosixPath(value.rstrip("/")).parts + if part not in {"."} + ) + + +def scopes_overlap(left: str, right: str) -> bool: + left_parts = normalize_scope(left) + right_parts = normalize_scope(right) + if not left_parts or not right_parts: + return True + width = min(len(left_parts), len(right_parts)) + return left_parts[:width] == right_parts[:width] + + +def section_content(body: str, heading: str) -> str: + pattern = re.compile( + rf"(?ms)^##\s+{re.escape(heading)}\s*$\n(.*?)(?=^##\s+|\Z)" + ) + match = pattern.search(body) + return "" if match is None else match.group(1).strip() + + +def validate_tasks(root: Path) -> list[Finding]: + findings: list[Finding] = [] + task_dir = root / "docs" / "tasks" + template = task_dir / "_template.md" + if template.is_file(): + metadata, body, errors = parse_frontmatter(template) + for message in errors: + findings.append(Finding("task-template", relative(template, root), 0, message)) + missing = sorted(TASK_REQUIRED_FIELDS - set(metadata)) + if missing: + findings.append( + Finding( + "task-template", + relative(template, root), + 0, + "缺少字段：" + ", ".join(missing), + ) + ) + headings = set(re.findall(r"(?m)^##\s+(.+?)\s*$", body)) + missing_sections = sorted(TASK_REQUIRED_SECTIONS - headings) + if missing_sections: + findings.append( + Finding( + "task-template", + relative(template, root), + 0, + "缺少章节：" + ", ".join(missing_sections), + ) + ) + else: + findings.append(Finding("task-template", "docs/tasks/_template.md", 0, "文件不存在。")) + + tasks: dict[str, Task] = {} + issue_numbers: dict[int, str] = {} + for path in sorted(task_dir.glob("T-*.md")) if task_dir.is_dir() else []: + rel = relative(path, root) + metadata, body, errors = parse_frontmatter(path) + for message in errors: + findings.append(Finding("task-frontmatter", rel, 0, message)) + filename_id = path.stem + if not TASK_ID.fullmatch(filename_id): + findings.append(Finding("task-id", rel, 0, "文件名必须是 T-<三位编号>[可选小写后缀]。")) + missing = sorted(TASK_REQUIRED_FIELDS - set(metadata)) + if missing: + findings.append( + Finding("task-frontmatter", rel, 0, "缺少字段：" + ", ".join(missing)) + ) + task_id = metadata.get("id") + if task_id != filename_id: + findings.append(Finding("task-id", rel, 0, "frontmatter id 必须与文件名一致。")) + if isinstance(task_id, str) and task_id in tasks: + findings.append(Finding("task-id", rel, 0, "任务 ID 重复。")) + status = metadata.get("status") + if status not in VALID_STATUS: + findings.append(Finding("task-status", rel, 0, "status 不在允许枚举中。")) + title = metadata.get("title") + if not isinstance(title, str) or not title.strip() or "【" in title: + findings.append(Finding("task-metadata", rel, 0, "title 必须是已填写的非空字符串。")) + phase = metadata.get("phase") + if type(phase) is not int or phase < 0: + findings.append(Finding("task-metadata", rel, 0, "phase 必须是非负整数。")) + created = metadata.get("created") + try: + if not isinstance(created, str): + raise ValueError + date.fromisoformat(created) + except ValueError: + findings.append(Finding("task-metadata", rel, 0, "created 必须是 YYYY-MM-DD。")) + deps = metadata.get("deps") + if not isinstance(deps, list) or not all(isinstance(dep, str) for dep in deps): + findings.append(Finding("task-deps", rel, 0, "deps 必须是任务 ID 数组。")) + elif task_id in deps: + findings.append(Finding("task-deps", rel, 0, "任务不得依赖自身。")) + elif any(not TASK_ID.fullmatch(dep) for dep in deps): + findings.append(Finding("task-deps", rel, 0, "deps 含无效任务 ID。")) + write_paths = metadata.get("write_paths") + if not isinstance(write_paths, list) or not write_paths: + findings.append(Finding("task-scope", rel, 0, "write_paths 必须是非空数组。")) + else: + values = [value for value in write_paths if isinstance(value, str)] + if len(values) != len(write_paths) or any(not is_safe_repo_path(value) for value in values): + findings.append(Finding("task-scope", rel, 0, "write_paths 含不安全或非字符串路径。")) + if len(values) != len(set(values)): + findings.append(Finding("task-scope", rel, 0, "write_paths 含重复路径。")) + if rel not in values: + findings.append(Finding("task-scope", rel, 0, "write_paths 必须包含任务文件自身。")) + issue = metadata.get("issue") + if issue is not None and (type(issue) is not int or issue <= 0): + findings.append(Finding("task-issue", rel, 0, "issue 必须是正整数或 null。")) + elif type(issue) is int: + if issue in issue_numbers: + findings.append(Finding("task-issue", rel, 0, "Issue 编号与其他任务重复。")) + issue_numbers[issue] = filename_id + context_ref = metadata.get("context_ref") + if context_ref is not None and ( + not isinstance(context_ref, str) or not SHA40.fullmatch(context_ref) + ): + findings.append(Finding("task-claim", rel, 0, "context_ref 必须是 40 位 SHA 或 null。")) + claim_branch = metadata.get("claim_branch") + if claim_branch is not None and claim_branch != f"claims/{filename_id}": + findings.append(Finding("task-claim", rel, 0, "claim_branch 与任务 ID 不一致。")) + work_branch = metadata.get("work_branch") + if work_branch is not None and ( + not isinstance(work_branch, str) + or not re.fullmatch(rf"agent/[^/]+/{re.escape(filename_id)}", work_branch) + ): + findings.append(Finding("task-claim", rel, 0, "work_branch 格式或任务 ID 不一致。")) + if status == "TODO": + for key in ("context_ref", "claim_branch", "work_branch"): + if metadata.get(key) is not None: + findings.append(Finding("task-claim", rel, 0, f"TODO 的 {key} 必须为 null。")) + if issue is not None and status in ACTIVE_STATUS: + if not isinstance(context_ref, str) or not SHA40.fullmatch(context_ref): + findings.append(Finding("task-claim", rel, 0, "Gitea 活跃任务缺少 40 位 context_ref。")) + if metadata.get("claim_branch") != f"claims/{filename_id}": + findings.append(Finding("task-claim", rel, 0, "claim_branch 与任务 ID 不一致。")) + if not isinstance(work_branch, str) or not work_branch.endswith(f"/{filename_id}"): + findings.append(Finding("task-claim", rel, 0, "work_branch 与任务 ID 不一致。")) + headings = set(re.findall(r"(?m)^##\s+(.+?)\s*$", body)) + missing_sections = sorted(TASK_REQUIRED_SECTIONS - headings) + if missing_sections: + findings.append( + Finding("task-sections", rel, 0, "缺少章节：" + ", ".join(missing_sections)) + ) + if status == "DONE": + evidence = section_content(body, "执行记录") + if not evidence or "（做完在此记录" in evidence or "【" in evidence: + findings.append(Finding("task-evidence", rel, 0, "DONE 缺少真实执行证据。")) + if isinstance(task_id, str): + tasks[task_id] = Task(path, metadata, body) + + for task_id, task in sorted(tasks.items()): + rel = relative(task.path, root) + deps = task.metadata.get("deps") + if not isinstance(deps, list): + continue + for dep in deps: + if dep not in tasks: + findings.append(Finding("task-deps", rel, 0, f"依赖任务不存在：{dep}。")) + elif task.status != "TODO" and tasks[dep].status != "DONE": + findings.append(Finding("task-deps", rel, 0, f"非 TODO 任务依赖尚未 DONE：{dep}。")) + + visiting: set[str] = set() + visited: set[str] = set() + + def visit(task_id: str) -> None: + if task_id in visiting: + findings.append( + Finding("task-deps", relative(tasks[task_id].path, root), 0, "依赖图存在环。") + ) + return + if task_id in visited: + return + visiting.add(task_id) + deps = tasks[task_id].metadata.get("deps") + if isinstance(deps, list): + for dep in deps: + if dep in tasks: + visit(dep) + visiting.remove(task_id) + visited.add(task_id) + + for task_id in sorted(tasks): + visit(task_id) + + active = [task for task in tasks.values() if task.status in ACTIVE_STATUS] + for index, left in enumerate(sorted(active, key=lambda task: task.task_id)): + left_paths = left.metadata.get("write_paths", []) + if not isinstance(left_paths, list): + continue + for right in sorted(active, key=lambda task: task.task_id)[index + 1 :]: + right_paths = right.metadata.get("write_paths", []) + if not isinstance(right_paths, list): + continue + if any( + isinstance(a, str) and isinstance(b, str) and scopes_overlap(a, b) + for a in left_paths + for b in right_paths + ): + findings.append( + Finding( + "task-scope-overlap", + relative(right.path, root), + 0, + f"活跃任务与 {left.task_id} 的 write_paths 重叠。", + ) + ) + return findings + + +def markdown_targets(text: str) -> Iterable[tuple[int, str]]: + in_fence = False + for line_number, line in enumerate(text.splitlines(), start=1): + stripped = line.lstrip() + if stripped.startswith("```") or stripped.startswith("~~~"): + in_fence = not in_fence + continue + if in_fence: + continue + for match in MARKDOWN_LINK.finditer(line): + yield line_number, match.group(1).strip() + + +def clean_link_target(raw: str) -> str | None: + if raw.startswith("<") and ">" in raw: + target = raw[1 : raw.index(">")] + else: + target = raw.split(maxsplit=1)[0] + target = urllib.parse.unquote(target).split("#", 1)[0].split("?", 1)[0] + if ( + not target + or target.startswith("#") + or target.startswith("//") + or URI_SCHEME.match(target) + or "【" in target + ): + return None + return target + + +def validate_markdown_links(root: Path, files: list[Path]) -> list[Finding]: + findings: list[Finding] = [] + for path in files: + if path.suffix.lower() != ".md": + continue + text = read_text(path) + if text is None: + continue + for line, raw in markdown_targets(text): + target = clean_link_target(raw) + if target is None: + continue + resolved = root / target.lstrip("/") if target.startswith("/") else path.parent / target + try: + resolved.resolve().relative_to(root.resolve()) + except ValueError: + findings.append( + Finding("markdown-link", relative(path, root), line, "链接逃出仓库根目录。") + ) + continue + if not resolved.exists(): + findings.append( + Finding("markdown-link", relative(path, root), line, "本地链接目标不存在。") + ) + return findings + + +def validate_navigation(root: Path) -> list[Finding]: + findings: list[Finding] = [] + root_readme = read_text(root / "README.md") or "" + docs_readme = read_text(root / "docs" / "README.md") or "" + for doc in sorted((root / "docs").glob("*.md")): + root_target = f"docs/{doc.name}" + if root_target not in root_readme: + findings.append(Finding("navigation", "README.md", 0, f"未登记 {root_target}。")) + if doc.name != "README.md" and f"({doc.name})" not in docs_readme: + findings.append( + Finding("navigation", "docs/README.md", 0, f"未登记 {doc.name}。") + ) + + required_root_entries = ( + "scripts/validate_agent_context.py", + "scripts/setup_gitea_labels.py", + "scripts/validate_harness_governance.py", + "scripts/audit_gitea_coordination.py", + "scripts/test_gitea_claim_race.py", + "tests/test_governance.py", + ".gitea/ISSUE_TEMPLATE/task.md", + ".gitea/PULL_REQUEST_TEMPLATE.md", + ".gitea/workflows/harness-governance.yml", + ) + for entry in required_root_entries: + if entry not in root_readme: + findings.append(Finding("navigation", "README.md", 0, f"未登记 {entry}。")) + return findings + + +def safe_token_assignment(value: str) -> bool: + value = value.strip().rstrip(",}").strip().strip("'\"") + upper = value.upper() + return ( + not value + or value.startswith(("【", "<")) + or SAFE_VARIABLE_REFERENCE.fullmatch(value) is not None + or upper in {"REPLACE", "CHANGEME", "EXAMPLE"} + or upper.startswith(("REPLACE_", "CHANGEME_", "EXAMPLE_")) + ) + + +def validate_secrets(root: Path, files: list[Path]) -> list[Finding]: + findings: list[Finding] = [] + for path in files: + rel = relative(path, root) + lower_name = path.name.lower() + if lower_name == "gitea.env" or ( + lower_name.startswith("gitea.env.") and lower_name != "gitea.env.example" + ): + findings.append(Finding("secret-file", rel, 0, "私有 Gitea 环境文件不得被跟踪。")) + text = read_text(path) + if text is None: + if path.suffix.lower() in KNOWN_TEXT_SUFFIXES or lower_name in { + ".env", + "dockerfile", + "makefile", + }: + findings.append( + Finding("secret-scan", rel, 0, "已跟踪文本无法安全解码并扫描。") + ) + continue + for line_number, line in enumerate(text.splitlines(), start=1): + token_assignment = TOKEN_ASSIGNMENT.match(line) + cmd_assignment = CMD_TOKEN_ASSIGNMENT.match(line) + rules = [] + if token_assignment and not safe_token_assignment(token_assignment.group(1)): + rules.append("GITEA_TOKEN 实值") + if cmd_assignment: + cmd_value = next( + (value for value in cmd_assignment.groups() if value is not None), + "", + ) + if not safe_token_assignment(cmd_value): + rules.append("Windows 命令 Token 实值") + if AUTH_VALUE.search(line): + rules.append("Authorization 实值") + if URL_CREDENTIAL.search(line): + rules.append("URL 内嵌凭据") + if GITEA_TOKEN_LITERAL.search(line): + rules.append("Gitea Token 字面值") + for rule in rules: + findings.append(Finding("secret-value", rel, line_number, f"检测到{rule}。")) + for match in DOTNET_TOKEN_SETTER.finditer(text): + if not safe_token_assignment(match.group(2)): + line_number = text.count("\n", 0, match.start()) + 1 + findings.append( + Finding( + "secret-value", + rel, + line_number, + "检测到 .NET 环境变量 Token 实值。", + ) + ) + return findings + + +def require_markers(root: Path, path_string: str, markers: Iterable[str]) -> list[Finding]: + path = root / PurePosixPath(path_string) + if not path.is_file(): + return [Finding("required-artifact", path_string, 0, "文件不存在。")] + text = read_text(path) or "" + return [ + Finding("required-artifact", path_string, 0, f"缺少标记：{marker}。") + for marker in markers + if marker not in text + ] + + +def validate_gitea_artifacts(root: Path) -> list[Finding]: + findings = [] + findings.extend( + require_markers( + root, + ".gitea/ISSUE_TEMPLATE/task.md", + ("task_id:", "task_file:", "context_ref:", "write_paths:", "lease_until:"), + ) + ) + findings.extend( + require_markers( + root, + ".gitea/PULL_REQUEST_TEMPLATE.md", + ("Closes #", "task_file:", "context_ref:", "write_paths:", "验证证据"), + ) + ) + workflow = ".gitea/workflows/harness-governance.yml" + findings.extend( + require_markers( + root, + workflow, + ( + "push:", + "pull_request:", + "actions/checkout@v4", + "permissions: read-all", + "persist-credentials: false", + "python -m unittest discover", + "python scripts/validate_harness_governance.py", + ), + ) + ) + return findings + + +def validate_repository(root: Path) -> list[Finding]: + files = candidate_files(root) + findings = [ + Finding("agent-context", "docs/agent-context.json", 0, message) + for message in validate_manifest(root) + ] + findings.extend(validate_navigation(root)) + findings.extend(validate_markdown_links(root, files)) + findings.extend(validate_tasks(root)) + findings.extend(validate_secrets(root, files)) + findings.extend(validate_gitea_artifacts(root)) + return sorted(set(findings)) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="离线校验 Harness Coding 仓库治理工件。") + parser.add_argument( + "--root", + type=Path, + default=Path(__file__).resolve().parents[1], + help="仓库根目录；默认取脚本上一级。", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + root = args.root.resolve() + if not root.is_dir(): + print("ERROR: 仓库根目录不存在。", file=sys.stderr) + return 2 + findings = validate_repository(root) + if findings: + for finding in findings: + print(finding.render(), file=sys.stderr) + print(f"治理校验失败：{len(findings)} 项不一致。", file=sys.stderr) + return 1 + print("治理校验通过：上下文、导航、链接、任务、模板、工作流与敏感信息均一致。") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tasks.md b/tasks.md index 0b98f50..becb2bb 100644 --- a/tasks.md +++ b/tasks.md @@ -97,7 +97,7 @@ Get-ChildItem -Recurse -File | H-601 | Phase 0：建立 Gitea MCP 安全与连接基线 | H-409 | 固定 MCP 版本；私有配置不入库；HTTP 需显式确认风险；读写审批和断连降级规则清楚 | DONE | | H-602 | Phase 1：增加上下文清单与按需读取流程 | H-601 | 有机器可读清单和无第三方依赖验证；agent 按任务类型读取；同一 SHA 不重复加载 | DONE | | H-603 | Phase 2：建立 Issue / 任务文件 / PR 多 Agent 协调协议 | H-602 | 任务映射、领取读回校验、分支 / worktree 和写路径防撞规则完整 | DONE | -| H-604 | Phase 3：增加自动化治理与一致性检查 | H-603 | 导航、清单、任务元数据、敏感信息和 Gitea Actions 检查可运行 | TODO | +| H-604 | Phase 3：增加自动化治理与一致性检查 | H-603 | 离线导航、清单、任务元数据和敏感信息检查可运行；Actions 模板就绪，实际运行以 runner 启用为前提 | DONE | ## Backlog diff --git a/tests/test_governance.py b/tests/test_governance.py new file mode 100644 index 0000000..f41ec73 --- /dev/null +++ b/tests/test_governance.py @@ -0,0 +1,419 @@ +from __future__ import annotations + +import base64 +import sys +import tempfile +import unittest +from datetime import datetime, timezone +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + +from audit_gitea_coordination import ( + audit_repository, + latest_claim, + paged, + parse_datetime, + parse_write_paths, + select_latest_claim, +) +from setup_gitea_labels import LABELS, NoRedirect, build_plan +from test_gitea_claim_race import PROBE_PREFIX, new_probe_branch +from validate_agent_context import validate_manifest +from validate_harness_governance import ( + validate_markdown_links, + validate_navigation, + validate_repository, + validate_secrets, + validate_tasks, + is_safe_repo_path, + scopes_overlap, +) + + +class RepositoryIntegrationTests(unittest.TestCase): + def test_repository_governance_passes(self) -> None: + self.assertEqual([], validate_repository(ROOT)) + + def test_context_manifest_passes(self) -> None: + self.assertEqual([], validate_manifest(ROOT)) + + +class OfflineRuleTests(unittest.TestCase): + def test_broken_markdown_link_is_reported(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + page = root / "page.md" + page.write_text("[missing](missing.md)\n", encoding="utf-8") + findings = validate_markdown_links(root, [page]) + self.assertEqual("markdown-link", findings[0].rule) + + def test_navigation_omission_is_reported(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "docs").mkdir() + (root / "README.md").write_text("# root\n", encoding="utf-8") + (root / "docs" / "README.md").write_text("# docs\n", encoding="utf-8") + (root / "docs" / "new.md").write_text("# new\n", encoding="utf-8") + findings = validate_navigation(root) + self.assertTrue(any(item.path == "README.md" for item in findings)) + self.assertTrue(any(item.path == "docs/README.md" for item in findings)) + + def test_task_status_and_scope_errors_are_reported(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + task_dir = root / "docs" / "tasks" + task_dir.mkdir(parents=True) + template = (ROOT / "docs" / "tasks" / "_template.md").read_text( + encoding="utf-8" + ) + (task_dir / "_template.md").write_text(template, encoding="utf-8") + task = template.replace("T-XXX", "T-001").replace( + "status: TODO", "status: INVALID" + ).replace("title: 一句话任务名", "title: []").replace( + "phase: 1", "phase: banana" + ).replace("created: 【日期】", "created: nonsense").replace( + "issue: null", "issue: 0" + ) + (task_dir / "T-001.md").write_text(task, encoding="utf-8") + rules = {finding.rule for finding in validate_tasks(root)} + self.assertIn("task-status", rules) + self.assertIn("task-scope", rules) + self.assertIn("task-metadata", rules) + self.assertIn("task-issue", rules) + + def test_done_task_requires_done_dependencies(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + task_dir = root / "docs" / "tasks" + task_dir.mkdir(parents=True) + template = (ROOT / "docs" / "tasks" / "_template.md").read_text( + encoding="utf-8" + ) + (task_dir / "_template.md").write_text(template, encoding="utf-8") + base = ( + template.replace("created: 【日期】", "created: 2026-07-14") + .replace(" - 【允许修改的仓库相对路径】\n", "") + .replace( + "（做完在此记录：改了哪些文件、跑的验证命令与结果、阻塞、关键决策。\n执行记录只写进本任务文件，不逐任务追加共享的 `progress.md`，避免多 agent 抢改共享文件。）", + "验证：python -m unittest，结果通过。", + ) + ) + first = base.replace("T-XXX", "T-001") + second = ( + base.replace("T-XXX", "T-002") + .replace("deps: []", "deps: [T-001]") + .replace("status: TODO", "status: DONE") + ) + (task_dir / "T-001.md").write_text(first, encoding="utf-8") + (task_dir / "T-002.md").write_text(second, encoding="utf-8") + findings = validate_tasks(root) + self.assertTrue( + any(item.rule == "task-deps" and item.path.endswith("T-002.md") for item in findings) + ) + + def test_scope_prefix_overlap(self) -> None: + self.assertTrue(scopes_overlap("src/api/", "src/api/users.py")) + self.assertTrue(scopes_overlap("README.md", "README.md")) + self.assertTrue(scopes_overlap(".", "src/api/users.py")) + self.assertTrue(scopes_overlap("Src/API", "src/api/users.py")) + self.assertFalse(scopes_overlap("src/api/", "src/ui/")) + self.assertFalse(is_safe_repo_path("src/**")) + self.assertFalse(is_safe_repo_path("src/\tapi")) + + def test_secret_finding_does_not_echo_value(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + path = root / "tracked.txt" + secret = "private-" + "credential-value" + path.write_text("GITEA_" + "TOKEN=" + secret + "\n", encoding="utf-8") + findings = validate_secrets(root, [path]) + rendered = "\n".join(finding.render() for finding in findings) + self.assertTrue(findings) + self.assertNotIn(secret, rendered) + + def test_secret_formats_are_detected(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + key = "GITEA_" + "TOKEN" + secret = "another-" + "private-value" + authorization = "Author" + "ization" + paths = [] + for name, content in ( + (".env", f"{key}={secret}\n"), + ("config.ps1", f"$env:{key} = '{secret}'\n"), + ("config.json", f'{{"{key}": "{secret}"}}\n'), + ("config.yml", f"{key}: {secret}\n"), + ("defaults.env", f"{key}=${{TOKEN:-{secret}}}\n"), + ("configure.cmd", f'set "{key}={secret}"\n'), + ("headers.txt", f"{authorization}: Basic dXNl" + "cjpwYXNz\n"), + ): + path = root / name + path.write_text(content, encoding="utf-8") + paths.append(path) + findings = validate_secrets(root, paths) + self.assertGreaterEqual(len(findings), 7) + self.assertNotIn(secret, "\n".join(item.render() for item in findings)) + + def test_windows_token_setters_and_utf16_are_detected(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + key = "GITEA_" + "TOKEN" + secret = "windows-" + "private-value" + setter = "[Environment]::SetEnvironmentVariable" + path = root / "configure.ps1" + path.write_text( + f'{setter}("{key}", "{secret}", "User")\n' + f'setx {key} {secret}\n', + encoding="utf-16", + ) + findings = validate_secrets(root, [path]) + self.assertGreaterEqual(len(findings), 2) + self.assertNotIn(secret, "\n".join(item.render() for item in findings)) + + +class GiteaHelperTests(unittest.TestCase): + def test_label_plan_detects_exclusive_change(self) -> None: + desired = next(label for label in LABELS if label["name"] == "status/todo") + existing = { + desired["name"]: { + "id": 1, + "name": desired["name"], + "color": desired["color"], + "description": desired["description"], + "exclusive": False, + } + } + actions = {item[1]["name"]: item[0] for item in build_plan(existing)} + self.assertEqual("update", actions["status/todo"]) + + def test_redirect_handler_refuses_redirect(self) -> None: + handler = NoRedirect() + self.assertIsNone( + handler.redirect_request(None, None, 302, "Found", {}, "http://example.invalid") + ) + + def test_claim_parsers(self) -> None: + comment = """CLAIM +task: T-123 +claimed_by: worker-1 +allocated_by: dispatcher-1 +context_ref: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa +claim_branch: claims/T-123 +work_branch: agent/worker-1/T-123 +write_paths: + - docs/tasks/T-123.md + - src/api/ +claimed_at: 2029-05-31T12:00:00Z +lease_until: 2029-06-01T12:00:00Z +""" + comments = [ + {"body": "note"}, + {"id": 1, "body": comment, "user": {"login": "dispatcher-1"}}, + ] + self.assertEqual(comment, latest_claim(comments, "T-123", "dispatcher-1")) + self.assertEqual(["docs/tasks/T-123.md", "src/api/"], parse_write_paths(comment)) + self.assertGreater( + parse_datetime("2030-01-01T00:00:00Z"), + datetime(2029, 1, 1, tzinfo=timezone.utc), + ) + self.assertIsNone(parse_datetime("2030-01-01")) + quoted = {"id": 99, "body": "Discussion quoted CLAIM and task: T-123"} + self.assertEqual( + comment, + latest_claim( + [ + { + "id": 1, + "body": comment, + "user": {"login": "dispatcher-1"}, + }, + quoted, + ], + "T-123", + "dispatcher-1", + ), + ) + + renewal = comment.replace("CLAIM\n", "CLAIM RENEWAL\n", 1).replace( + "claimed_at: 2029-05-31T12:00:00Z", + "claimed_at: 2029-06-01T00:00:00Z", + ) + selected, errors = select_latest_claim( + comments + + [ + { + "id": 2, + "body": renewal, + "user": {"login": "dispatcher-1"}, + } + ], + "T-123", + "dispatcher-1", + ) + self.assertEqual(renewal, selected) + self.assertEqual([], errors) + + changed_identity = renewal.replace("claimed_by: worker-1", "claimed_by: worker-2") + selected, errors = select_latest_claim( + comments + + [ + {"id": 2, "body": renewal, "user": {"login": "worker-1"}}, + { + "id": 3, + "body": changed_identity, + "user": {"login": "dispatcher-1"}, + }, + ], + "T-123", + "dispatcher-1", + ) + self.assertEqual(comment, selected) + self.assertGreaterEqual(len(errors), 2) + + def test_pagination_reads_until_empty_page(self) -> None: + class FakePagedClient: + def __init__(self) -> None: + self.pages: list[int] = [] + + def request(self, method: str, path: str) -> object: + page = int(path.rsplit("page=", 1)[1]) + self.pages.append(page) + if page == 1: + return [{"id": number} for number in range(1, 21)] + if page == 2: + return [{"id": 21}] + return [] + + client = FakePagedClient() + values = paged(client, "/items") # type: ignore[arg-type] + self.assertEqual(21, len(values)) + self.assertEqual([1, 2, 3], client.pages) + + def test_probe_branch_is_unique_and_scoped(self) -> None: + first = new_probe_branch() + second = new_probe_branch() + self.assertTrue(first.startswith(PROBE_PREFIX)) + self.assertNotEqual(first, second) + + def test_valid_active_remote_task_audits_cleanly(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + task_dir = root / "docs" / "tasks" + task_dir.mkdir(parents=True) + template = (ROOT / "docs" / "tasks" / "_template.md").read_text( + encoding="utf-8" + ) + default_task = template.replace("T-XXX", "T-123").replace( + "issue: null", "issue: 1" + ) + (task_dir / "T-123.md").write_text(default_task, encoding="utf-8") + sha = "a" * 40 + work_task = ( + default_task.replace("status: TODO", "status: DOING") + .replace("context_ref: null", f"context_ref: {sha}") + .replace("claim_branch: null", "claim_branch: claims/T-123") + .replace("work_branch: null", "work_branch: agent/worker-1/T-123") + .replace(" - 【允许修改的仓库相对路径】\n", "") + ) + claim = f"""CLAIM +task: T-123 +claimed_by: worker-1 +allocated_by: dispatcher-1 +context_ref: {sha} +claim_branch: claims/T-123 +work_branch: agent/worker-1/T-123 +write_paths: + - docs/tasks/T-123.md +claimed_at: 2029-05-31T12:00:00Z +lease_until: 2029-06-01T12:00:00Z +""" + labels = [ + {"name": "kind/task"}, + {"name": "type/code"}, + {"name": "priority/p1"}, + {"name": "status/doing"}, + ] + + class FakeClient: + def list_labels(self) -> dict[str, dict[str, object]]: + return { + label["name"]: { + "name": label["name"], + "exclusive": label["exclusive"], + } + for label in LABELS + } + + def request(self, method: str, path: str, payload: object = None) -> object: + self.assert_get(method) + page = int(path.rsplit("page=", 1)[1]) if "page=" in path else 1 + if page > 1: + return [] + if path.startswith("/issues?state=all"): + return [ + { + "number": 1, + "title": "[T-123] valid", + "body": "- task_id: `T-123`\n- task_file: `docs/tasks/T-123.md`\n- write_paths:\n - `docs/tasks/T-123.md`\n", + "state": "open", + "labels": labels, + } + ] + if path.startswith("/branches?"): + return [ + {"name": "claims/T-123", "commit": {"id": sha}}, + {"name": "agent/worker-1/T-123", "commit": {"id": "b" * 40}}, + ] + if path.startswith("/pulls?"): + return [] + if path.startswith("/issues/1/comments?"): + return [ + { + "body": claim, + "user": {"login": "dispatcher-1"}, + } + ] + if path.startswith("/contents/docs/tasks/T-123.md?"): + return { + "content": base64.b64encode(work_task.encode("utf-8")).decode( + "ascii" + ) + } + self.fail(f"unexpected path: {path}") + + def assert_get(self, method: str) -> None: + if method != "GET": + self.fail("audit attempted a write") + + def fail(self, message: str) -> None: + raise AssertionError(message) + + findings, count = audit_repository( + root, + FakeClient(), # type: ignore[arg-type] + datetime(2029, 6, 1, tzinfo=timezone.utc), + "dispatcher-1", + ) + self.assertEqual(1, count) + self.assertEqual([], findings) + + (task_dir / "T-123.md").write_text( + default_task.replace("deps: []", "deps: [T-122]"), + encoding="utf-8", + ) + findings, _ = audit_repository( + root, + FakeClient(), # type: ignore[arg-type] + datetime(2029, 6, 1, tzinfo=timezone.utc), + "dispatcher-1", + ) + self.assertTrue(any(item.rule == "dependency" for item in findings)) + + +if __name__ == "__main__": + unittest.main()