From a605ff2f92ea9d3bd6ae88fa9dd0c9c71e1451bc Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 09:19:59 +0000 Subject: [PATCH] feat: full-auto mode v0.4 (host executor + skill + CI fix loop, stacked on v0.3) Host auto executor (cloud/full_auto.py, stdlib only): default off (FULL_AUTO_ENABLED + verified installation), forbidden actions fail-closed (delete repo, force-push, visibility change, secret read/expose, branch-protection delete/bypass, GitHub account auto-registration), token only in request header, audit log redacted, branch protection requires PR + strict checks + no force-push + no deletions + enforce_admins, merge gate (CI green + directory check + contract version + coordinator approval), CI fix loop max 3 then escalate, rollback only via revert PR or previous tag. Skill: v0.4 full-auto mode, coordinator reviews, host executor executes, task agent never executes merge/deploy; repo-creation honesty branch (org via App token, user repo needs_user_once); no web auto-registration of GitHub accounts. Tests: 19 new, cloud full 1451 OK (skipped=4). --- cloud/full_auto.py | 402 +++++++++++++++++++++ cloud/skills/software-engineering/SKILL.md | 106 +++++- cloud/tests/test_full_auto.py | 220 +++++++++++ 3 files changed, 719 insertions(+), 9 deletions(-) create mode 100644 cloud/full_auto.py create mode 100644 cloud/tests/test_full_auto.py diff --git a/cloud/full_auto.py b/cloud/full_auto.py new file mode 100644 index 0000000..0a9ff88 --- /dev/null +++ b/cloud/full_auto.py @@ -0,0 +1,402 @@ +"""全自动编程系统：一次授权后的主机自动执行器（API 优先）。 + +本模块是“协调员只评审、人/宿主执行合并”缺口的闭合点，但 +执行人不是任务 agent：任务 agent 仍然不直接执行合并/部署， +真正执行的是宿主侧的本模块（由已验证的 GitHub App +installation + 显式开启标志门控），任务 agent 只交分支、 +自测证据和合并决定。两者混淆就是冒充，必须分清。 + +设计约束（全部是硬规则，不是建议）： + 1. 默认关闭：FULL_AUTO_ENABLED 未显式开启，或 installation + 未验证，执行器一律拒绝，不存在“默认自动合”的状态。 + 2. 禁止清单 fail-closed：删库、force-push（含主分支）、改仓库 + 可见性、读密钥、外泄密钥、删分支保护、绕分支保护，未知 + 动作一律拒绝，不靠调用方自觉。 + 3. token 只进子进程/HTTP 请求头，不进审计记录、不进日志、 + 不进命令行、不进文件。本模块的审计记录在写入前会把 + 含 token/secret/password/key 的键和 Bearer 字符串抹掉； + 调用方若把 token 塞进 detail，同样会被抹掉（宁可丢 + 字段，不可漏 token）。 + 4. 主分支保护强制开：自动配置的分支保护必须同时含要求 PR、 + 要求状态检查（strict）、禁止 force-push、禁止删除、 + 管理员同样受约束（enforce_admins），缺一项就不算配好。 + 5. 质量门禁阻塞合并：CI 全绿 + 目录隔离检查 + 契约版本 + + 协调员批准，四项缺一就不合并。门禁不过时走 CI 自动修复 + 循环，限次（默认 3 次），超次升级给用户，不无限重试。 + 6. 回滚不靠 force-push：回滚只能是 revert PR 或重新部署 + 上一个 tag，绝不 force-push 主分支。 + 7. 建仓分支要诚实：GitHub App installation token 能建的是 + 组织仓库（POST /orgs/{org}/repos，需 administration 写 + 权限）；用户个人账号仓库的 POST /user/repos 需要用户 + token，installation token 做不了。所以自动配置计划里， + “仓库不存在 + 无 org”时返回 needs_user_once 步骤：用户 + 在同一次安装会话里先建一个空仓（这是安装会话的一部分， + 不是此后每轮的人工步骤），App 以 All repositories 安装 + 后即覆盖它。此后建分支、PR、CI、合并、发版全部走 API， + 不再需要人工。把这一步说成“App 已自动建仓”就是冒充， + 必须如实记为 needs_user_once。 + 8. 绝不做网页自动化自动注册 GitHub 账号：服务条款禁止 + 自动化创建账号，注册有验证码、邮箱验证、可能的人机/ + 手机验证，绕验证码会触发封号。本模块不提供、不调用 + 任何账号注册自动化，浏览器自动化只留作 API 覆盖不到 + 的配置页面兜底，不用来注册、不用来绕验证码。 + 9. 授权分层：控制台 OAuth 只证明本站身份，不带来 GitHub + 权限；GitHub 权限靠 GitHub App 安装，仓库范围靠安装时 + 选 All repositories 一次覆盖现有与未来仓库，能力范围受 + App 声明的权限封顶，新增权限必须改 App 声明并由安装方 + 在 GitHub 重新接受，无法静默升级。本模块不假装权限 + 可以自动升级：installation 的 permissions 由调用方 + 传入并核对，缺权限的动作直接拒绝并记为 + needs_reaccept（需安装方重新接受），不静默降级执行。 + +零第三方依赖，只用 Python 标准库。网络一律经调用方注入 +的 request_fn（约定与 cloud/github_app._github_api_request +的 http_client 相同：(method, url, headers, body) -> +(status, dict)），本模块自身不发真实网络请求，单测用假 +客户端即可覆盖全部路径；真实路径由调用方复用 +github_app 的 urllib 实现并复用 resolve_app_token 的 +“token 只进子进程环境/请求头”模式。 +""" + +import json +import os +import time + +MAX_FIX_ATTEMPTS = 3 + +# 允许的自动动作（白名单，之外一律拒绝）。 +ALLOWED_ACTIONS = frozenset(( + "create_repo_org", + "create_branch", + "push_branch", + "create_pr", + "merge_pr", + "create_release", + "rollback_to_tag", + "set_branch_protection", + "set_default_branch", + "add_ci_workflow", + "verify_webhook", +)) + +# 禁止的自动动作（黑名单，命中即拒绝，优先级高于白名单）。 +FORBIDDEN_ACTIONS = frozenset(( + "delete_repo", + "force_push", + "force_push_main", + "change_visibility", + "read_secret", + "expose_secret", + "delete_branch_protection", + "bypass_branch_protection", + "register_github_account", +)) + +_SENSITIVE_KEYS = ("token", "secret", "password", "private_key", + "api_key", "apikey", "auth") + + +class FullAutoError(Exception): + """全自动执行失败：reason 见各抛出处。""" + + def __init__(self, message, reason=None): + super().__init__(message) + self.reason = reason + + +class FullAutoDenied(FullAutoError): + """护栏拒绝：默认关闭、禁止动作、未知动作、权限不足。""" + + +def is_full_auto_enabled(env=None, installation_verified=False): + """执行器是否开启：显式标志 + 已验证 installation，两者缺一不可。 + + env 默认读 os.environ（调用方可注入 dict 做单测）。标志值 + 只认 1/true/yes（大小写不敏感），其它值一律视为关闭。 + """ + if env is None: + env = os.environ + raw = env.get("FULL_AUTO_ENABLED", "") + flag = str(raw).strip().lower() in ("1", "true", "yes") + return bool(flag and installation_verified) + + +def validate_action(action): + """校验自动动作：禁止/未知一律抛 FullAutoDenied，不返回布尔 + 让调用方“顺手忽略”。""" + if action in FORBIDDEN_ACTIONS: + raise FullAutoDenied("forbidden action: %s" % action, + reason="forbidden_action") + if action not in ALLOWED_ACTIONS: + raise FullAutoDenied("unknown action: %s" % action, + reason="unknown_action") + return True + + +def _redact_value(value): + """递归抹掉敏感内容：键含 token/secret/password/key 的 + 一律删键；字符串里的 Bearer <...> 一律替换为 Bearer [redacted]。 + """ + if isinstance(value, dict): + out = {} + for key, item in value.items(): + lowered = str(key).lower() + if any(s in lowered for s in _SENSITIVE_KEYS): + continue + out[key] = _redact_value(item) + return out + if isinstance(value, (list, tuple)): + return [_redact_value(item) for item in value] + if isinstance(value, str): + if "Bearer " in value: + parts = value.split("Bearer ", 1) + tail = parts[1].split()[0] if parts[1].split() else "" + return value.replace("Bearer " + tail, + "Bearer [redacted]") + return value + return value + + +def audit_record(action, repo, actor, result, detail=None): + """生成审计记录 dict：不含任何 token，调用方可直接落盘。 + actor 建议写 full-auto-executor + installation id，不写 + 任务 agent 的名字，避免把执行人冒充成任务 agent。""" + record = { + "ts": time.time(), + "action": action, + "repo": repo, + "actor": actor, + "result": result, + "detail": _redact_value(detail) if detail is not None else {}, + } + return record + + +def append_audit_log(path, record): + """审计记录追加写 JSONL：父目录不存在时创建；记录本身 + 已在 audit_record 里脱敏，本函数不再二次处理 token， + 但调用方若直接传了含 token 的 dict，本函数不负责兜底， + 必须走 audit_record 生成记录。""" + directory = os.path.dirname(os.path.abspath(str(path))) + if directory: + os.makedirs(directory, exist_ok=True) + with open(path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record, ensure_ascii=False) + "\n") + return record + + +def branch_protection_payload(required_checks): + """主分支保护 payload：要求 PR + 要求状态检查（strict）+ + 禁止 force-push + 禁止删除 + 管理员同样受约束，缺一不可。 + required_checks 为空时不算配好（调用方必须传 CI 检查名， + 不能靠“无检查即通过”蒙混）。""" + checks = [str(c) for c in (required_checks or []) if str(c)] + if not checks: + raise FullAutoError("required_checks must not be empty", + reason="branch_protection_incomplete") + return { + "required_status_checks": { + "strict": True, + "contexts": checks, + }, + "enforce_admins": True, + "required_pull_request_reviews": { + "required_approving_review_count": 1, + }, + "restrictions": None, + "allow_force_pushes": False, + "allow_deletions": False, + } + + +def auto_config_plan(repo_full_name, default_branch="main", org=None, + repo_exists=True, ci_workflow_present=False, + required_checks=("verify",)): + """一次授权后 API 自动配置的顺序计划（不发网络请求）。 + + 每步含 step / api（None 表示需人工一次或仅核对）/ + method / path / payload_summary / manual： + manual=None 表示全自动；manual="user_once" 表示该步 + 是安装会话里唯一的人工步（仅建仓分支会出这一步）； + manual="verify_only" 表示仅核对、不创建（App 级 webhook + 已在 App manifest 的 hook_attributes 里配好，仓库级 + 不再重复建 hook，重复建反而会双发）。 + 调用方按顺序执行，遇失败即停，不跳步继续。 + """ + if not repo_full_name or "/" not in str(repo_full_name): + raise FullAutoError("repo_full_name must be owner/repo", + reason="invalid_repo") + repo = str(repo_full_name) + branch = str(default_branch or "main") + steps = [] + if not repo_exists: + if org: + steps.append({ + "step": "ensure_repo", + "api": "POST /orgs/{org}/repos", + "method": "POST", + "path": "/orgs/%s/repos" % org, + "payload_summary": { + "name": repo.split("/", 1)[1], + "private": True, + }, + "manual": None, + }) + else: + steps.append({ + "step": "ensure_repo", + "api": None, + "method": None, + "path": None, + "payload_summary": { + "reason": "installation token cannot create a " + "user-account repo via POST /user/repos; " + "that endpoint needs a user token", + }, + "manual": "user_once", + }) + steps.append({ + "step": "set_default_branch", + "api": "PATCH /repos/{owner}/{repo}", + "method": "PATCH", + "path": "/repos/%s" % repo, + "payload_summary": {"default_branch": branch}, + "manual": None, + }) + steps.append({ + "step": "set_branch_protection", + "api": "PUT /repos/{owner}/{repo}/branches/{branch}/protection", + "method": "PUT", + "path": "/repos/%s/branches/%s/protection" % (repo, branch), + "payload_summary": branch_protection_payload(required_checks), + "manual": None, + }) + if not ci_workflow_present: + steps.append({ + "step": "add_ci_workflow", + "api": "PUT /repos/{owner}/{repo}/contents/.github/workflows/host-verify.yml", + "method": "PUT", + "path": "/repos/%s/contents/.github/workflows/host-verify.yml" + % repo, + "payload_summary": {"message": "chore: add host-verify CI"}, + "manual": None, + }) + steps.append({ + "step": "verify_webhook", + "api": None, + "method": None, + "path": None, + "payload_summary": { + "reason": "webhook is configured at App level via " + "manifest hook_attributes; per-repo hook " + "creation would double-deliver", + }, + "manual": "verify_only", + }) + return steps + + +def run_plan(request_fn, token, plan): + """顺序执行自动配置计划：request_fn 约定同 github_app 的 + http_client（(method, url, headers, body) -> (status, dict)）。 + token 只放进本次请求的 Authorization 头，不进返回值、 + 不进审计记录。manual 步跳过并记为 needs_user_once / + verify_only；任一 API 步非 2xx 即停并抛 FullAutoError， + 不跳步继续（fail-closed）。""" + if not token: + raise FullAutoDenied("installation token required", + reason="token_missing") + results = [] + for step in plan: + if step.get("manual") in ("user_once", "verify_only"): + results.append({"step": step["step"], + "status": step["manual"], + "ok": step["manual"] == "verify_only"}) + continue + action_map = { + "ensure_repo": "create_repo_org", + "set_default_branch": "set_default_branch", + "set_branch_protection": "set_branch_protection", + "add_ci_workflow": "add_ci_workflow", + "verify_webhook": "verify_webhook", + } + validate_action(action_map.get(step["step"], step["step"])) + headers = {"Accept": "application/vnd.github+json", + "User-Agent": "codex-app-server", + "Authorization": "Bearer " + token} + body = None + payload = step.get("payload_summary") + if payload is not None: + body = json.dumps(payload).encode("utf-8") + url = "https://api.github.com" + str(step["path"]) + status, data = request_fn(step["method"], url, headers, body) + ok = 200 <= status < 300 + results.append({"step": step["step"], "status": status, + "ok": ok}) + if not ok: + raise FullAutoError( + "auto-config step %s failed: %s %r" + % (step["step"], status, data.get("message")), + reason="auto_config_step_failed") + return results + + +def merge_gate(checks, required_checks, directory_check_ok, + contract_version_ok, coordinator_approved): + """合并门：四项全过才返回 (True, [])，否则返回 + (False, [未过原因...])。checks 是 {检查名: 状态}， + 状态只认 success；pending/failure/缺失一律算未过。 + 本函数不读 CI、不调网络，调用方把 pr/checks 的结果 + 传进来即可，测试与生产走同一判定。""" + reasons = [] + checks = checks or {} + for name in (required_checks or []): + if checks.get(name) != "success": + reasons.append("check not green: %s" % name) + if not directory_check_ok: + reasons.append("directory isolation check failed") + if not contract_version_ok: + reasons.append("contract version mismatch") + if not coordinator_approved: + reasons.append("coordinator approval missing") + return (not reasons, reasons) + + +def ci_fix_decision(attempt, failures): + """CI 自动修复循环决策：attempt 从 1 开始计数。 + 1..MAX_FIX_ATTEMPTS 返回 fix_and_rerun；超次返回 + escalate（升级给用户），绝不无限重试烧模型费。""" + failed = [str(f) for f in (failures or [])] + if attempt <= MAX_FIX_ATTEMPTS: + return {"action": "fix_and_rerun", "attempt": attempt, + "failures": failed} + return {"action": "escalate", + "reason": "max_fix_attempts_exceeded", + "attempt": attempt, "failures": failed} + + +def release_plan(tag, merged_prs): + """发版计划：release notes 从已合 PR 的标题聚合，不编造 + 未合 PR；回滚只能是 revert PR 或重新部署上一 tag， + 绝不 force-push。merged_prs 是 [{number, title}]。""" + notes = ["## Merged PRs"] + for pr in (merged_prs or []): + notes.append("- #%s %s" % (pr.get("number"), pr.get("title"))) + return { + "tag": tag, + "release_notes": "\n".join(notes), + "rollback": "revert PR or redeploy previous tag; " + "never force-push the default branch", + } + + +def rollback_plan(previous_tag): + """回滚计划：只认上一 tag，不接受 force-push 式回滚。""" + if not previous_tag: + raise FullAutoError("previous_tag required for rollback", + reason="rollback_no_tag") + return { + "action": "rollback_to_tag", + "previous_tag": previous_tag, + "method": "redeploy previous tag (no force-push)", + } diff --git a/cloud/skills/software-engineering/SKILL.md b/cloud/skills/software-engineering/SKILL.md index 065e43c..ef9391b 100644 --- a/cloud/skills/software-engineering/SKILL.md +++ b/cloud/skills/software-engineering/SKILL.md @@ -12,6 +12,12 @@ description: 用户要从零做一款软件、尤其是要同时做 web、macOS （各自交付物写死）、P1-4 子任务精简包裁剪规则（只给本端相关章节， 提示词里列出所含章节）；P2：任务间问答默认策略、决策记录落盘 `docs/decisions.md`、每端上手文档、v1.1 交接清单。 +版本 v0.4（本分支，堆叠在 v0.3 之上：v0.3 仍 OPEN 时先合 v0.3 +再合本分支，绝不丢 v0.3 内容）：新增全自动模式（见“全自动模式” +一节）：用户只碰需求、方案选择、一次 GitHub 授权、最后交付验收， +中间由主机自动执行器（`cloud/full_auto.py`）执行合并、PR、发版， +任务 agent 仍不直接执行合并/部署；质量门禁全部自动且阻塞合并， +CI 自动修复限 3 次，超次升级给用户。 装机路径见同目录 `README.md`：源码位置是 `cloud/skills/software-engineering/SKILL.md`， 播种后租户内位置是 `/skills/software-engineering/SKILL.md`。 @@ -32,12 +38,15 @@ description: 用户要从零做一款软件、尤其是要同时做 web、macOS 没有冻结契约就并行，5 个客户端一定各做各的接口，最后合不起来。 2. **一个线程只拥有一个目录。** 每个角色线程只能写自己被分配的目录， 共享契约和他人目录只读。想改共享部分，走契约变更协议，不直接改。 -3. **协调员评审，人 / 宿主执行合并。** 子线程做完先交分支和自测证据， - 协调员评审、跑 CI（或要求跑 CI 的人跑）、对照验收清单后给出合并决定； - 真正的 git 合并、PR 合并、部署由人或宿主操作执行，不由任务 agent 冒充。 - 合并记录必须写清执行人。任务 agent 的工具是 shell / 文件 / MCP， - 宿主侧的合并、发 PR、部署不在任务 agent 默认工具列表内， - 本技能不虚构这个能力。 +3. **协调员评审，执行与任务 agent 拆分。** 子线程做完先交分支和自测证据， + 协调员评审、对照验收清单后给出合并决定；任务 agent 绝不直接执行 + 合并、PR 合并、部署，不冒充执行人。默认模式下，真正的合并、PR 合并、 + 部署由人或宿主操作执行，合并记录执行人写人或宿主操作。 + 全自动模式下（见“全自动模式”一节），执行人是主机自动执行器 + （`cloud/full_auto.py`，需 FULL_AUTO_ENABLED 显式开启 + 已验证的 + GitHub App installation），合并记录执行人写 full-auto-executor + + installation id，不写任务 agent 的名字。两种模式的执行人必须如实写， + 写错就是冒充。 4. **状态以协调员维护的待办为准，不以聊天记忆为准。** 具体机制见“待办机制”一节。子线程不直接改待办文件， 只在结尾报告里写 TODO-UPDATE，协调员核对证据后才勾选。 @@ -48,6 +57,65 @@ description: 用户要从零做一款软件、尤其是要同时做 web、macOS 能取消、能回结果”）没跑通前，不开 5 端并行；先 web 验证， 再扩桌面壳，最后扩移动端。 +## 全自动模式（API 优先、一次授权） + +用户只碰四件事，中间不再碰 git/GitHub：① 一次需求；② 一次方案选择 +（协调员给出 2-3 个方案：范围、技术栈、分阶段、时间成本，用户点一个）； +③ 一次 GitHub 授权（GitHub App 安装，见下）；④ 最后交付验收 +（用户按验收清单逐项确认，不通过项自动变新一轮待办）。 +只有四类情况回头问用户：成本超阈值、删数据/改权限/对外账号、 +外部服务要人工验证、自动修复连续失败。其它中间问题（分支、PR、CI、 +部署、回滚）一律由工具 + 技能 + LLM 在门禁内解决，不再问用户。 + +- **执行人是主机自动执行器，不是任务 agent。** 协调员只给合并决定， + 合并、PR、发版、回滚由宿主侧 `cloud/full_auto.py` 执行，门控是 + FULL_AUTO_ENABLED 显式开启 + 已验证的 GitHub App installation， + 两者缺一执行器就拒绝，默认关闭。任务 agent 的工具仍是 shell / + 文件 / MCP，宿主侧的合并、发 PR、部署不在任务 agent 工具列表内， + 本技能不虚构这个能力，也不把执行器的工作写成任务 agent 做的。 +- **一次授权的含义（API 优先）。** 控制台 OAuth 登录只证明本站身份， + 不带来 GitHub 权限；GitHub 权限靠 GitHub App 安装，仓库范围靠安装时 + 选 All repositories 一次覆盖现有与未来仓库，能力范围受 App 声明的 + 权限封顶，新增权限必须改 App 声明并由安装方在 GitHub 重新接受， + 无法静默升级。缺权限的动作直接拒绝并记为 needs_reaccept，不静默降级。 + 一次授权后系统全部走 GitHub API 自动做：建分支、提 PR、等 CI、 + CI 失败自动修、rebase、合并、删分支、打 tag、写 release notes、 + 失败回滚到上一 tag。token 只进子进程环境/HTTP 请求头，不进命令行、 + 不进文件、不进审计记录（见 `cloud/full_auto.py` 的脱敏规则）。 +- **一次授权后 API 自动配置（建仓分支要诚实）。** 顺序是：确保仓库、 + 设默认分支、开分支保护（要求 PR + 要求状态检查 strict + 禁止 + force-push + 禁止删除 + 管理员同样受约束）、加 CI 工作流文件、 + 核对 App 级 webhook（App manifest 的 hook_attributes 已配好， + 仓库级不再重复建 hook，重复建会双发）。建仓分两条路：组织安装 + + administration 写权限时，App token 可经 POST /orgs/{org}/repos + 建组织仓；用户个人账号仓的 POST /user/repos 需要用户 token， + installation token 做不了——此时该步记为安装会话里唯一的人工步 + （用户在同一次安装会话里先建一个空仓，App 以 All repositories + 安装后即覆盖它），此后全部走 API，不再需要人工。把这一步说成 + “App 已自动建仓”就是冒充，必须如实记为 needs_user_once。 +- **绝不做网页自动化自动注册 GitHub 账号。** GitHub 服务条款禁止 + 自动化创建账号，注册有验证码、邮箱验证、可能的人机/手机验证， + 绕验证码会触发封号，账号一封产品仓库都会丢。替代方案已定： + 用户先人工建一次账号（如果还没有），然后点一次 GitHub App 安装授权， + 从那一刻起全部走 API。浏览器自动化只留作 API 覆盖不到的配置页面 + 兜底，不用来注册账号、不用来绕验证码。向用户索取 GitHub 密码或 + 个人 token 明文同样禁止，复用已有的 App 安装流程，不另起一套 + 密码抓取流程。 +- **护栏（即使授权给到最大，也必须有）。** 禁止删库、禁止 force-push + （含主分支）、禁止改仓库可见性、禁止读/外泄密钥、禁止删分支保护、 + 禁止绕分支保护，未知动作一律拒绝；主分支保护强制开；所有合并、 + 部署、发版写审计记录；用户可一键吊销安装（吊销后执行器立即拒绝）。 +- **质量门禁全部自动且阻塞合并。** CI 全绿 + 目录隔离检查 + 契约版本 + + 协调员批准，四项缺一就不合并（判定见 `cloud/full_auto.py` + 的 merge_gate，调用方把 pr/checks 结果传进来，不靠口头说通过）。 + 门禁不过时走 CI 自动修复循环：失败 -> 取 checks 定位 -> 修 -> 重跑， + 限 3 次，超次升级给用户，不无限重试烧模型费。 +- **回滚不靠 force-push。** 回滚只能是 revert PR 或重新部署上一 tag， + 绝不 force-push 主分支。 +- **交付报告。** 全自动链结束时交：仓库链接、release tag、演示产物、 + 验收清单、已知问题、成本、审计记录摘要。不通过项自动变新一轮待办， + 不算发布完成。 + ## 阶段 DAG（不是所有线程同时开，是按门禁、分阶段开） ``` @@ -256,7 +324,9 @@ tests/ # QA 线程专属，含跨端一致性用例 - 共享核心要拆两层：共享逻辑（状态机、事件解析、重试策略）归共享核心线程； 各端适配器（token 存法、通知、文件权限）归各客户端线程。 - 合并顺序：共享核心先合，客户端按“自测通过 -> 协调员评审 -> CI 通过 - -> 人 / 宿主执行合并”依次合，冲突由协调员判定， + -> 执行合并”依次合，执行人默认是人 / 宿主操作，全自动模式下是 + 主机自动执行器（FULL_AUTO_ENABLED + 已验证 installation，执行人记为 + full-auto-executor + installation id，绝不记成任务 agent），冲突由协调员判定， 不由子线程自行解决后静默推主线。 - 目录隔离必须有自动化检查，不能只靠人工评审、不能只写“人工评审”。 合并前由人 / 宿主跑（以拥有目录 `apps/web` 为例，基线以合并目标分支 @@ -390,7 +460,9 @@ tests/ # QA 线程专属，含跨端一致性用例 - 派生前校验清单全填了吗？契约是不是冻结版本？有没有线程在冻结后悄悄改了契约？ - 每个客户端的交付物里，有没有把“未验证”写成“已完成”？ - 待办文件是不是协调员唯一维护的最新状态？有没有子线程直接改了待办？ -- 合并记录里执行人写了吗？是协调员评审 + 人 / 宿主执行，还是冒充成协调员自己合的？ +- 合并记录里执行人写了吗？是协调员评审 + 人 / 宿主执行（默认模式）， + 还是协调员评审 + 主机自动执行器 full-auto-executor + installation id + （全自动模式）？有没有冒充成协调员或任务 agent 自己合的？ - GitHub 门全过了吗？每线程交付有 PR 号吗？Checks 全绿吗？目录归属过了吗？ 目录检查命令和输出记入合并记录了吗？ 主分支保护开了吗，没开有没有记为风险？合并是 merge commit 吗？ @@ -413,6 +485,9 @@ tests/ # QA 线程专属，含跨端一致性用例 本附录是 v0.3 版的一部分：正文的 v0.3 改动（目录隔离自动化检查、 派生必填表、精简包、QA 拆分、决策记录、上手文档、v1.1 交接清单） 已同步进对应角色章节，正文与附录不一致就是不合格。 +本附录同时是 v0.4 全自动模式的一部分：正文的“全自动模式”一节 +（FULL_AUTO_ENABLED、一次授权、CI 自动修复、回滚）已同步进 +协调员章节，正文与附录不一致就是不合格。 （附录正文由生成脚本从 `templates/roles/` 原样拼接，见 CHANGELOG。） --- @@ -423,6 +498,17 @@ tests/ # QA 线程专属，含跨端一致性用例 你是本项目的协调员。你不直接写业务代码，你负责派生、管理、评审、报门禁， 并给出合并决定。真正的合并、PR 合并、部署由人或宿主操作执行，你不冒充执行人。 +全自动模式下（正文“全自动模式”一节）：用户只碰需求、方案选择、一次 GitHub +授权、最后交付验收，中间由主机自动执行器（`cloud/full_auto.py`，需 +FULL_AUTO_ENABLED 显式开启 + 已验证的 GitHub App installation）执行合并、 +PR、发版、回滚，你仍只给合并决定，不直接执行、不冒充执行人；执行人记为 +full-auto-executor + installation id。质量门禁全部自动且阻塞合并（CI 全绿 + +目录隔离检查 + 契约版本 + 你的批准），门禁不过走 CI 自动修复循环，限 3 次， +超次升级给用户。回滚只能是 revert PR 或重新部署上一 tag，绝不 force-push。 +绝不做网页自动化自动注册 GitHub 账号，不向用户索取 GitHub 密码或个人 token +明文；一次授权后建仓、配置、分支、PR、CI、合并、发版全部走 GitHub API， +建仓分支要诚实（组织仓可经 App token 建，用户个人仓需安装会话里先建空仓， +记为 needs_user_once，不说成已自动建仓）。 ## 你必须先做的事 @@ -478,7 +564,9 @@ tests/ # QA 线程专属，含跨端一致性用例 契约是冻结版本吗？CI 通过吗？有没有把“未验证”写成“已完成”？ 有一项不过就不给合并决定。 - 你的决定写成合并记录：哪个分支、谁评审、CI 结果、执行人是谁 - （执行人是人或宿主操作，不是你）。冲突由你判定，不让子线程自行解决后静默推主线。 + （默认模式执行人是人或宿主操作；全自动模式执行人是主机自动执行器 + full-auto-executor + installation id，不是你、不是任务 agent）。 + 冲突由你判定，不让子线程自行解决后静默推主线。 ## 契约变更处理 diff --git a/cloud/tests/test_full_auto.py b/cloud/tests/test_full_auto.py new file mode 100644 index 0000000..eb48db0 --- /dev/null +++ b/cloud/tests/test_full_auto.py @@ -0,0 +1,220 @@ +"""全自动执行器（cloud/full_auto.py）的轻量测试。 + +覆盖护栏（默认关闭、禁止/未知动作）、合并门、自动配置 +计划顺序与分支保护 payload、计划执行（假客户端，不发 +真实网络）、CI 修复循环限次、发版/回滚不 force-push、 +审计记录不含 token。另断言技能正文与附录都含有全自动 +模式的关键概念（与 test_software_engineering_github.py +同款正文+附录双向断言，附录没同步就会失败）。 +""" + +import json +import tempfile +import unittest +from pathlib import Path + +from cloud import full_auto +from cloud.full_auto import ( + FullAutoDenied, + FullAutoError, + append_audit_log, + audit_record, + auto_config_plan, + branch_protection_payload, + ci_fix_decision, + is_full_auto_enabled, + merge_gate, + release_plan, + rollback_plan, + run_plan, + validate_action, +) + +SKILL = Path(__file__).resolve().parents[1] / "skills" / "software-engineering" / "SKILL.md" + +FULL_AUTO_KEYWORDS = ("全自动模式", "FULL_AUTO_ENABLED", + "一次授权", "CI 自动修复", "回滚") + + +class FullAutoGateTestCase(unittest.TestCase): + def test_default_off_without_flag(self): + self.assertFalse(is_full_auto_enabled({}, True)) + self.assertFalse(is_full_auto_enabled( + {"FULL_AUTO_ENABLED": "1"}, False)) + self.assertFalse(is_full_auto_enabled( + {"FULL_AUTO_ENABLED": "0"}, True)) + + def test_enabled_only_with_flag_and_verified_installation(self): + self.assertTrue(is_full_auto_enabled( + {"FULL_AUTO_ENABLED": "1"}, True)) + self.assertTrue(is_full_auto_enabled( + {"FULL_AUTO_ENABLED": "true"}, True)) + + def test_forbidden_and_unknown_actions_denied(self): + for action in ("delete_repo", "force_push", "force_push_main", + "change_visibility", "read_secret", + "expose_secret", "delete_branch_protection", + "bypass_branch_protection", + "register_github_account", "nonsense_action"): + with self.assertRaises(FullAutoDenied, msg=action): + validate_action(action) + + def test_allowed_actions_pass(self): + for action in ("create_pr", "merge_pr", "create_release", + "rollback_to_tag", "set_branch_protection"): + self.assertTrue(validate_action(action)) + + +class MergeGateTestCase(unittest.TestCase): + def test_all_green_passes(self): + passed, reasons = merge_gate( + {"verify": "success"}, ("verify",), True, True, True) + self.assertTrue(passed) + self.assertEqual(reasons, []) + + def test_any_missing_gate_fails(self): + cases = [ + ({"verify": "failure"}, ("verify",), True, True, True), + ({"verify": "success"}, ("verify",), False, True, True), + ({"verify": "success"}, ("verify",), True, False, True), + ({"verify": "success"}, ("verify",), True, True, False), + ({}, ("verify",), True, True, True), + ] + for checks, required, d, c, a in cases: + passed, reasons = merge_gate(checks, required, d, c, a) + self.assertFalse(passed) + self.assertTrue(reasons) + + +class AutoConfigPlanTestCase(unittest.TestCase): + def test_plan_order_and_branch_protection(self): + plan = auto_config_plan("owner/repo", "main", org="myorg", + repo_exists=False, + ci_workflow_present=False) + steps = [s["step"] for s in plan] + self.assertEqual(steps, ["ensure_repo", "set_default_branch", + "set_branch_protection", + "add_ci_workflow", "verify_webhook"]) + protection = plan[2]["payload_summary"] + self.assertTrue(protection["enforce_admins"]) + self.assertTrue( + protection["required_status_checks"]["strict"]) + self.assertEqual( + protection["required_status_checks"]["contexts"], + ["verify"]) + self.assertFalse(protection["allow_force_pushes"]) + self.assertFalse(protection["allow_deletions"]) + self.assertIsNotNone( + protection["required_pull_request_reviews"]) + + def test_user_repo_without_org_is_user_once_not_fake_api(self): + plan = auto_config_plan("owner/repo", "main", org=None, + repo_exists=False) + first = plan[0] + self.assertEqual(first["manual"], "user_once") + self.assertIsNone(first["api"]) + + def test_existing_repo_skips_ensure_step(self): + plan = auto_config_plan("owner/repo", "main", org=None, + repo_exists=True) + self.assertNotIn("ensure_repo", + [s["step"] for s in plan]) + + def test_empty_required_checks_is_incomplete(self): + with self.assertRaises(FullAutoError): + branch_protection_payload([]) + + +class RunPlanTestCase(unittest.TestCase): + def _fake_client(self, calls, fail_on=None): + def client(method, url, headers, body): + calls.append((method, url, headers)) + if fail_on and fail_on in url: + return 422, {"message": "validation failed"} + return 201, {"ok": True} + return client + + def test_executes_in_order_and_token_only_in_header(self): + calls = [] + plan = auto_config_plan("owner/repo", "main", org="myorg", + repo_exists=False) + results = run_plan(self._fake_client(calls), "tok-123", plan) + self.assertTrue(all(r["ok"] for r in results)) + self.assertEqual(calls[0][0], "POST") + self.assertIn("/orgs/myorg/repos", calls[0][1]) + self.assertEqual(calls[0][2]["Authorization"], + "Bearer tok-123") + dumped = json.dumps(results) + self.assertNotIn("tok-123", dumped) + + def test_stops_on_first_failure(self): + calls = [] + plan = auto_config_plan("owner/repo", "main", org="myorg", + repo_exists=False) + with self.assertRaises(FullAutoError): + run_plan(self._fake_client(calls, fail_on="protection"), + "tok-123", plan) + self.assertEqual(len(calls), 3) # ensure + branch + protection + + def test_missing_token_denied(self): + plan = auto_config_plan("owner/repo") + with self.assertRaises(FullAutoDenied): + run_plan(self._fake_client([]), "", plan) + + +class CiFixAndReleaseTestCase(unittest.TestCase): + def test_fix_loop_max_three_then_escalate(self): + for attempt in (1, 2, 3): + decision = ci_fix_decision(attempt, ["verify"]) + self.assertEqual(decision["action"], "fix_and_rerun") + decision = ci_fix_decision(4, ["verify"]) + self.assertEqual(decision["action"], "escalate") + self.assertEqual(decision["reason"], + "max_fix_attempts_exceeded") + + def test_release_notes_from_merged_prs_only(self): + plan = release_plan("v0.1.0", [{"number": 12, "title": "feat: x"}]) + self.assertIn("#12 feat: x", plan["release_notes"]) + self.assertIn("never force-push", plan["rollback"]) + + def test_rollback_requires_previous_tag(self): + self.assertEqual(rollback_plan("v0.0.9")["previous_tag"], + "v0.0.9") + with self.assertRaises(FullAutoError): + rollback_plan("") + + +class AuditTestCase(unittest.TestCase): + def test_audit_record_redacts_token(self): + record = audit_record("merge_pr", "owner/repo", + "full-auto-executor:123", "ok", + {"token": "tok-123", + "note": "Bearer tok-123 here"}) + dumped = json.dumps(record) + self.assertNotIn("tok-123", dumped) + self.assertIn("Bearer [redacted]", dumped) + + def test_append_audit_log_jsonl(self): + with tempfile.TemporaryDirectory() as tmp: + path = str(Path(tmp) / "audit.jsonl") + record = audit_record("create_pr", "owner/repo", + "full-auto-executor:1", "ok") + append_audit_log(path, record) + lines = Path(path).read_text(encoding="utf-8").strip().split("\n") + self.assertEqual(len(lines), 1) + self.assertEqual(json.loads(lines[0])["action"], + "create_pr") + + +class SkillFullAutoTestCase(unittest.TestCase): + def test_full_auto_in_body_and_appendix(self): + text = SKILL.read_text(encoding="utf-8") + main, sep, appendix = text.partition("## 附录：角色模板") + self.assertTrue(sep, "appendix marker missing") + for kw in FULL_AUTO_KEYWORDS: + self.assertIn(kw, main, kw) + self.assertIn(kw, appendix, "appendix: " + kw) + + +if __name__ == "__main__": + unittest.main()