diff --git a/governance/transitions.yaml b/governance/transitions.yaml index 1f5a1ca..9bfa06a 100644 --- a/governance/transitions.yaml +++ b/governance/transitions.yaml @@ -50,25 +50,42 @@ transitions: to_state: ready action: noop guard: "sender_role in ['owner', 'agent']" - - id: T5 # spec PR 合并 + suite 就绪后进入红队阶段(ISSUE-263 AC-12 / ADR-0049) - # suite 就绪谓词(conductor 侧重新断言,不信任 dispatch 载荷): - # suite/ 存在 + 含非空测试文件(test_*.py)+ ast.parse 可解析。 - # 不满足=转移拒绝并告警(AC-12 / AC-17 的 T-14 卡绑定测试回路)。 + # T5/T6 suite 就绪谓词(ISSUE-263 AC-12 / W4-C1 .github#282): + # suite 就绪 = 确定性谓词,conductor 在转移前重新断言(不信任 dispatch 载荷): + # 1. suite/ 目录存在; + # 2. suite/ 含至少一个非空测试文件(字节数 >0 且含有效断言语法); + # 3. 测试文件可解析(python -m py_compile / node --check / 等语言原生语法校验通过)。 + # 任一不满足 → 转移拒绝并告警(fail-closed:不放行不确定的红队/波次规划)。 + # 该谓词由 conductor 在运行时对目标 spec 仓做确定性检查,非仅标签面断言。 + - id: T5 # spec PR 合并 + suite 就绪后进入红队阶段(ISSUE-263 AC-12 / ADR-0049 / W4-C1) from_state: spec event: label:state:redteam to_state: redteam action: noop guard: "sender_role in ['owner', 'agent'] and 'type:intent' in label_set" - - id: T6 # redteam 产出 survived 审计记录后进入波次规划(ISSUE-263 AC-12 / ADR-0079) - # 三元组校验(conductor 侧重新断言):进入 wave-planned 必须存在该卡本次 - # 生命周期内、卡 ID + specVersion + 审计 run ID 三元组对应的 survived 审计 - # 记录(禁止跨卡/历史记录短路)。adversary verdict 必须 = survived。 - # 不满足=转移拒绝并告警(AC-12)。 + suite_ready_required: true # conductor 在转移前执行 suite 就绪谓词(存在+非空+可解析) + # T6 三元组校验(ISSUE-263 AC-12 / W4-C1 .github#282): + # 进入 wave-planned 必须存在该卡本次生命周期内、卡 ID + specVersion + 审计 run ID + # 三元组对应的 survived 审计记录。禁止跨卡/历史记录短路: + # - 卡 ID 必须与当前 issue 匹配; + # - specVersion 必须与当前 spec PR 合并后的 spec 版本一致; + # - 审计 run ID 必须来自本次生命周期(非历史 run)。 + # 三元组任一缺失或不匹配 → 转移拒绝并告警。 + - id: T6 # redteam 产出 survived 审计记录后进入波次规划(ISSUE-263 AC-12 / ADR-0079 / W4-C1) from_state: redteam event: label:state:wave-planned to_state: wave-planned action: noop guard: "sender_role == 'agent' and 'adversary:survived' in label_set" + triple_check: # conductor 在转移前校验三元组 survived 记录 + card_id: issue # 卡 ID 键 = 当前 issue number + spec_version: spec # specVersion 键 = spec 文件版本(SHA 或 semver) + audit_run_id: adversary-run # 审计 run ID 键 = adversary 报告 run_id + source_label: "adversary:survived" # 触发本转移的 survived 标签 + # needs-human 不可直跳 wave-planned(ISSUE-263 AC-12 / AC-14 / W4-C1): + # needs-human → wave-planned 转移不在本表列 = 禁止(conductor no-op)。 + # 该断言作为 required check 在 CI 常驻强制运行(conductor-negtest 模式), + # 确保没有任何路径可从 needs-human 直跳 wave-planned。 # ---------- Bug 流(W3-C1 .github#218 / ADR-0064,宪法 §3)---------- # 执行者:CI-Workflows pipeline/bugflow(bug-reproduce.yml 以 App 身份写标签)。 # conductor 对 label:bug/label:reproduced 等非 state 标签事件静默 no-op—— diff --git a/scripts/escalate-test-issue.py b/scripts/escalate-test-issue.py index b28ce03..de6593c 100644 --- a/scripts/escalate-test-issue.py +++ b/scripts/escalate-test-issue.py @@ -1,28 +1,32 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""escalate-test-issue.py —— agent 发现测试/AC 有误时结构化上报 issue(W2-C4 .github#276,AC-18) - -实现 agent 发现验收测试/AC 语义有误时经结构化上报 issue(引用 test/AC 编号) -路由 owner,TTL 内裁决(dead-man 提醒),裁决前暂停该卡相关合并——这是"发现 -错误"的合法上报通道,与防篡改阻断互补。 - -上报结构(JSON schema: escalation-issue/v1): - - card_ref: 卡编号(如 Cloudbird-Software/.github#276) - - test_ref: 有误的测试/AC 编号(如 AC-17、specs/ISSUE-263/suite/test_holdout.py) - - finding_type: test-error / ac-error / false-positive / false-negative - - evidence: 证据描述(引用 file:line) - - ttl_hours: 裁决 TTL(默认 72h) - - pause_merge: 是否暂停该卡相关合并(默认 true) - -用法: - python escalate-test-issue.py --card --test-ref \ - --finding --evidence [--ttl ] \ - [--pause/--no-pause] [--output ] [--create-issue] - 环境变量:GH_TOKEN / APP_TOKEN - -退出码: 0=上报成功/结构化记录已生成 - 1=参数错误 - 2=infra 错误(API 调用失败) +"""escalate-test-issue.py —— 实现 agent 发现测试/AC 有误时结构化上报 issue(W2-C4 .github#276 / AC-18) + +当实现 agent 发现验收测试/AC 语义有误(测试本身写错、AC 理解偏差、断言与 spec 不一致), +经本脚本结构化上报 issue: + - 引用 test/AC 编号(机器可追溯); + - 路由 owner 裁决(TTL 内 + dead-man 提醒); + - 裁决前暂停该卡相关合并(防错误测试放行实现)。 + +这是"发现错误"的合法上报通道,与 g060 防篡改阻断互补: + - g060 防的是"开发 agent 篡改测试以通过"(阻断+开 issue); + - 本脚本是"agent 发现测试本身有误"(上报+暂停合并+等裁决)。 + +终态落盘:.artest/.json(machine-verifiable final state) + status: adopted | rejected | expired + evidence: 上报评论 body + 引用的 test/AC 编号 + ttl_hours: 48(裁决窗口,短于 g060 的 72h——测试错误阻塞实现,须更快响应) + +用法: + python3 scripts/escalate-test-issue.py create \ + --card ISSUE-263 --repo Cloudbird-Software/.github \ + --test-ref "specs/ISSUE-263/suite/test_tax.py::test_brackets" \ + --ac-ref AC-3 \ + --reason "断言期望值与 spec §AC-3 不一致(spec 要求 ≥0.7,断言写 0.5)" \ + --actor cloudbrid-agent + python3 scripts/escalate-test-issue.py resolve --issue --repo [--owner ] + +退出码:0=已闭环/已创建 | 1=参数错误 | 2=issue 不存在/无权限 """ from __future__ import annotations @@ -34,174 +38,200 @@ import sys import urllib.error import urllib.request -from typing import Any +from pathlib import Path -DEFAULT_TTL_HOURS = 72 +TTL_HOURS = 48 +ADOPT_RE = re.compile(r"/test-adopt\b", re.IGNORECASE) +REJECT_RE = re.compile(r"/test-reject\b", re.IGNORECASE) +COMMENT_RE = re.compile(r"/test-comment\b", re.IGNORECASE) def now_iso() -> str: return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") -def ttl_deadline(ttl_hours: int) -> str: - return (dt.datetime.now(dt.timezone.utc) + dt.timedelta(hours=ttl_hours)).strftime("%Y-%m-%dT%H:%M:%SZ") - - -def err(msg: str) -> None: - prefix = "::error::" if os.environ.get("CI") else "FATAL: " - print(prefix + msg, file=sys.stderr) - - -def die(code: int, msg: str) -> None: - err(msg) - sys.exit(code) - - -# --------------------------------------------------------------------------- -# 结构化上报记录 -# --------------------------------------------------------------------------- - -def build_escalation_record(card_ref: str, test_ref: str, finding_type: str, - evidence: str, ttl_hours: int, - pause_merge: bool) -> dict: - """构建 escalation-issue/v1 结构化记录。""" - now = now_iso() - deadline = ttl_deadline(ttl_hours) - return { - "schema": "escalation-issue/v1", - "card_ref": card_ref, - "test_ref": test_ref, - "finding_type": finding_type, - "evidence": evidence, - "reported_at": now, - "ttl_hours": ttl_hours, - "deadline": deadline, - "pause_merge": pause_merge, - "status": "open", - "resolution": None, - "resolved_at": None, - "resolution_actor": None, - } - - -def format_issue_body(record: dict) -> str: - """将结构化记录格式化为 issue body。""" - lines = [ - "## 测试/AC 语义有误上报(W2-C4 / AC-18)", - "", - f"- **卡编号**: {record['card_ref']}", - f"- **有误的测试/AC**: {record['test_ref']}", - f"- **发现类型**: {record['finding_type']}", - f"- **证据**: {record['evidence']}", - f"- **上报时间**: {record['reported_at']}", - f"- **裁决 TTL**: {record['ttl_hours']}h(截止 {record['deadline']})", - f"- **暂停合并**: {'是' if record['pause_merge'] else '否'}", - "", - "### 裁决闭环(机器可核)", - "", - "终态:采纳 / 驳回 + 证据引用。", - "TTL 内未处置 → dead-man 提醒(conductor 转介 arbiter / owner 通知)。", - "裁决前该卡相关合并暂停(与防篡改阻断互补——这是'发现错误'的合法上报通道)。", - "", - f"", - ] - return "\n".join(lines) - - -# --------------------------------------------------------------------------- -# GitHub API:创建 issue -# --------------------------------------------------------------------------- - -def create_issue(token: str, repo: str, title: str, body: str, - labels: list[str] | None = None) -> dict: - """经 GitHub API 创建 issue。""" - data = {"title": title, "body": body} - if labels: - data["labels"] = labels - req_body = json.dumps(data).encode("utf-8") +def gh_api(token: str, path: str, method: str = "GET", data: dict | None = None): + api = os.environ.get("CB_GITHUB_API", "https://api.github.com") + body = json.dumps(data).encode("utf-8") if data is not None else None req = urllib.request.Request( - f"https://api.github.com/repos/{repo}/issues", - data=req_body, - headers={"Authorization": f"Bearer {token}", - "Accept": "application/vnd.github+json", - "User-Agent": "escalate-test-issue"}, - method="POST", + f"{api}{path}", data=body, + headers={ + "Authorization": f"Bearer {token}", + "Accept": "application/vnd.github+json", + "Content-Type": "application/json", + "User-Agent": "cloudbrid-agent", + }, + method=method, ) try: with urllib.request.urlopen(req, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) + raw = resp.read().decode("utf-8") + return resp.status, (json.loads(raw) if raw.strip() else {}) except urllib.error.HTTPError as e: - err_body = e.read().decode("utf-8", errors="replace")[:500] - die(2, f"创建 issue 失败 HTTP {e.code}: {err_body}") - except Exception as e: # noqa: BLE001 - die(2, f"创建 issue 异常: {e}") - - -# --------------------------------------------------------------------------- -# CLI -# --------------------------------------------------------------------------- - -def main() -> int: - ap = argparse.ArgumentParser(prog="escalate-test-issue.py", description="agent 发现测试/AC 有误时结构化上报 issue(W2-C4 / AC-18)") - ap.add_argument("--card", required=True, help="卡编号(如 Cloudbird-Software/.github#276)") - ap.add_argument("--test-ref", required=True, help="有误的测试/AC 编号") - ap.add_argument("--finding", required=True, - choices=["test-error", "ac-error", "false-positive", "false-negative", "other"], - help="发现类型") - ap.add_argument("--evidence", required=True, help="证据描述(引用 file:line)") - ap.add_argument("--ttl", type=int, default=DEFAULT_TTL_HOURS, help=f"裁决 TTL 小时(默认 {DEFAULT_TTL_HOURS})") - ap.add_argument("--pause", dest="pause", action="store_true", default=True, help="暂停该卡相关合并(默认开启)") - ap.add_argument("--no-pause", dest="pause", action="store_false", help="不暂停合并") - ap.add_argument("--output", default=None, help="结构化记录输出路径(JSON)") - ap.add_argument("--create-issue", action="store_true", help="经 GitHub API 创建 issue(需 GH_TOKEN)") - ap.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY", "Cloudbird-Software/.github"), - help="issue 目标仓库") - a = ap.parse_args() - - record = build_escalation_record( - card_ref=a.card, - test_ref=a.test_ref, - finding_type=a.finding, - evidence=a.evidence, - ttl_hours=a.ttl, - pause_merge=a.pause, + err = e.read().decode("utf-8", errors="replace")[:500] + print(f"API {method} {path} 失败 rc={e.code}: {err}", file=sys.stderr) + raise + + +def create_issue(token: str, repo: str, card: str, test_ref: str, ac_ref: str, + reason: str, actor: str) -> int: + """创建结构化上报 issue,返回 issue number。""" + title = f"[test-ac-error] {card} 测试/AC 语义有误({ac_ref})" + body = ( + f"## 测试/AC 语义有误上报(AC-18 正向上报通道)\n\n" + f"- **卡**:`{card}`\n" + f"- **引用测试**:`{test_ref}`\n" + f"- **引用 AC**:`{ac_ref}`\n" + f"- **上报者**:`{actor}`\n" + f"- **上报时间**:{now_iso()}\n\n" + f"### 问题描述\n{reason}\n\n" + f"### 影响与处置\n" + f"- 裁决前该卡相关合并暂停(防错误测试放行实现)\n" + f"- owner 请在 TTL {TTL_HOURS}h 内裁决:\n" + f" - `/test-adopt` — 确认测试有误,接受修复方案\n" + f" - `/test-reject` — 驳回(测试正确,实现须符合现有测试)\n" + f" - `/test-comment <内容>` — 补充信息\n" + f"- TTL 无裁决 → dead-man 提醒 + 终态=expired(维持暂停)\n\n" + f"---\n" + f"*本 issue 由 agent 经 escalate-test-issue.py 自动创建(W2-C4 / AC-18)。" + f"与 g060 防篡改阻断互补:本通道是"发现错误"的合法上报,非篡改测试。*" ) + status, data = gh_api(token, f"/repos/{repo}/issues", "POST", { + "title": title, + "body": body, + "labels": ["test-ac-error", f"card:{card}"], + }) + if status not in (200, 201): + print(f"创建 issue 失败 rc={status}: {data}", file=sys.stderr) + sys.exit(2) + number = data.get("number", 0) + print(f"已创建上报 issue #{number}({repo})") + return number + + +def resolve_issue(token: str, repo: str, issue: int, owner: str, final_dir: str) -> int: + """读取 issue 评论,解析 owner 裁决指令,写终态。""" + try: + iss, _ = gh_api(token, f"/repos/{repo}/issues/{issue}") + except Exception: + print(f"错误:无法读取 issue #{issue}", file=sys.stderr) + return 2 + + title = iss.get("title", "") + if "test-ac-error" not in title: + print(f"警告:issue #{issue} 标题不含 'test-ac-error',仍尝试处理", file=sys.stderr) + + final_path = Path(final_dir) / f"{issue}.json" + final_path.parent.mkdir(parents=True, exist_ok=True) + + # 已落盘则幂等返回 + if final_path.is_file(): + existing = json.loads(final_path.read_text(encoding="utf-8")) + print(f"终态已落盘 {final_path}: status={existing.get('status')}") + return 0 + + verdict: str | None = None + evidence: dict | None = None + try: + status, comments = gh_api(token, f"/repos/{repo}/issues/{issue}/comments?per_page=100") + comments = comments if isinstance(comments, list) else [] + except Exception: + comments = [] + + for c in comments: + author = (c.get("user") or {}).get("login", "") + if author != owner: + continue + body = c.get("body") or "" + if ADOPT_RE.search(body): + verdict = "adopted" + evidence = {"comment_body": body, "created_at": c.get("created_at", ""), "author": author} + break + if REJECT_RE.search(body): + verdict = "rejected" + evidence = {"comment_body": body, "created_at": c.get("created_at", ""), "author": author} + break + + # TTL 判定 + created_at = iss.get("created_at", "") + if verdict is None and created_at: + try: + created = dt.datetime.strptime(created_at, "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=dt.timezone.utc) + deadline = created + dt.timedelta(hours=TTL_HOURS) + if dt.datetime.now(dt.timezone.utc) > deadline: + verdict = "expired" + evidence = {"reason": f"TTL {TTL_HOURS}h 已过", "created_at": created_at, + "deadline": deadline.strftime("%Y-%m-%dT%H:%M:%SZ"), + "dead_man": True} + except ValueError: + pass + + if verdict is None: + print(f"issue #{issue} 尚未裁决且 TTL 未过 → 待决(退出 0,等待下次轮询)") + return 0 + + final = { + "schema": "test-ac-error-final-state/v1", + "issue": issue, + "repo": repo, + "status": verdict, + "evidence": evidence, + "ttl_hours": TTL_HOURS, + "resolved_at": now_iso(), + } + final_path.write_text(json.dumps(final, ensure_ascii=False, indent=2), + encoding="utf-8", newline="\n") + print(f"终态落盘 {final_path}: status={verdict}") - body = format_issue_body(record) - print("== 测试/AC 语义有误上报(W2-C4 / AC-18)==") - print(f"卡编号: {record['card_ref']}") - print(f"有误的测试/AC: {record['test_ref']}") - print(f"发现类型: {record['finding_type']}") - print(f"证据: {record['evidence']}") - print(f"TTL: {record['ttl_hours']}h(截止 {record['deadline']})") - print(f"暂停合并: {record['pause_merge']}") - - # 结构化记录落盘 - if a.output: - with open(a.output, "w", encoding="utf-8", newline="\n") as f: - json.dump(record, f, ensure_ascii=False, indent=2) - print(f"结构化记录已写入: {a.output}") - - # 创建 issue - if a.create_issue: - token = os.environ.get("GH_TOKEN") or os.environ.get("APP_TOKEN") or "" - if not token: - die(2, "创建 issue 需要 GH_TOKEN 或 APP_TOKEN") - title = f"[test-issue] {a.card} {a.test_ref} {a.finding}" - labels = ["test-issue", "needs-human", f"finding:{a.finding}"] + if verdict == "expired": try: - issue = create_issue(token, a.repo, title, body, labels) - print(f"Issue 已创建: #{issue.get('number')} {issue.get('html_url')}") - record["issue_number"] = issue.get("number") - record["issue_url"] = issue.get("html_url") - if a.output: - with open(a.output, "w", encoding="utf-8", newline="\n") as f: - json.dump(record, f, ensure_ascii=False, indent=2) - except SystemExit: - raise + gh_api(token, f"/repos/{repo}/issues/{issue}/comments", "POST", { + "body": ( + f"@${owner} 测试/AC 错误上报 TTL {TTL_HOURS}h 已过,无裁决 → " + f"终态 **expired**(维持合并暂停)。请尽快裁决:" + f"`/test-adopt` 或 `/test-reject`。\ndead-man 提醒时间:{now_iso()}" + ), + }) + print("已追加 dead-man 提醒评论") + except Exception as e: + print(f"追加 dead-man 评论失败(终态已落盘,不阻断): {e}", file=sys.stderr) return 0 +def main() -> int: + ap = argparse.ArgumentParser(prog="escalate-test-issue.py", + description="agent 发现测试/AC 有误时结构化上报") + sub = ap.add_subparsers(dest="cmd", required=True) + + p_create = sub.add_parser("create", help="创建测试/AC 错误上报 issue") + p_create.add_argument("--card", required=True, help="卡 ID") + p_create.add_argument("--repo", required=True, help="owner/repo") + p_create.add_argument("--test-ref", required=True, help="引用测试(文件::用例)") + p_create.add_argument("--ac-ref", required=True, help="引用 AC 编号") + p_create.add_argument("--reason", required=True, help="问题描述") + p_create.add_argument("--actor", default="cloudbrid-agent", help="上报者") + + p_resolve = sub.add_parser("resolve", help="解析 owner 裁决,写终态") + p_resolve.add_argument("--issue", required=True, type=int) + p_resolve.add_argument("--repo", required=True) + p_resolve.add_argument("--owner", default="randypanding") + p_resolve.add_argument("--final-dir", default=".artest") + + args = ap.parse_args() + token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN", "") + if not token: + print("错误:GH_TOKEN / GITHUB_TOKEN 未设置", file=sys.stderr) + return 1 + + if args.cmd == "create": + num = create_issue(token, args.repo, args.card, args.test_ref, args.ac_ref, + args.reason, args.actor) + print(json.dumps({"issue": num, "repo": args.repo, "card": args.card}, ensure_ascii=False)) + return 0 + return resolve_issue(token, args.repo, args.issue, args.owner, args.final_dir) + + if __name__ == "__main__": sys.exit(main()) diff --git a/scripts/holdout-register.sh b/scripts/holdout-register.sh index d436451..5709b34 100644 --- a/scripts/holdout-register.sh +++ b/scripts/holdout-register.sh @@ -1,74 +1,84 @@ #!/usr/bin/env bash -# holdout-register.sh —— holdout 注册入口脚本(W2-C4 .github#276,AC-17 / ADR-0068) +# holdout-register.sh —— holdout 注册入口(W2-C4 .github#276 / ADR-0068 / AC-17 / AC-18) # -# 由验证者 APP 令牌调用(非验证者 APP 写入被拒,覆盖跨仓场景)。 -# 接线 CI-Workflows pipeline/adversary/holdout_registry.py。 +# 由验证者 APP 执行(IFACE-01 时序闸:验证者 APP 未就绪时本脚本 dry-run 占位)。 +# 经 verifier APP 令牌调用 pipeline/adversary/holdout_registry.py register, +# 将 holdout 条目注册到 registry(hash 落盘,供 PR 引用一致性检查消费)。 # -# 用法: -# bash scripts/holdout-register.sh -# 环境变量: -# GH_TOKEN / APP_TOKEN = 验证者 APP 令牌(workflow 注入) -# VERIFIER_APP_SLUG = 验证者 APP slug(缺省 verifier-app[bot]) +# 凭据: +# - 验证者 APP 私钥:VERIFIER_APP_SECRET_FILE(PEM 路径)或 VERIFIER_APP_SECRET(PEM 字面量) +# - 验证者 APP_ID:VERIFIER_APP_ID(从 expected-state.json#verifier_app.id 读取) +# - 未就绪时 HOLDOUT_REGISTRY_MODE=dry-run(占位注册,不落 push) # -# 退出码: 0=注册成功 | 1=校验失败/拒绝 | 2=infra 错误 - +# 用法: +# VERIFIER_APP_ID= VERIFIER_APP_SECRET_FILE= \ +# bash scripts/holdout-register.sh +# +# 退出码:0=注册成功 | 1=参数错误 | 2=registry/环境错误 | 3=非验证者 APP 写入被拒 set -euo pipefail -REPO="${1:?用法: holdout-register.sh }" -ENTRY_PATH="${2:?用法: holdout-register.sh }" -VERIFIER_SLUG="${VERIFIER_APP_SLUG:-verifier-app[bot]}" +ORG="${ORG:-Cloudbird-Software}" +CARD="${1:?用法: holdout-register.sh }" +ENTRY="${2:?用法: holdout-register.sh }" +SHA256="${3:?用法: holdout-register.sh }" +REGISTRY="${4:?用法: holdout-register.sh }" -if [[ ! -f "$ENTRY_PATH" ]]; then - echo "::error::条目文件不存在: $ENTRY_PATH" - exit 2 -fi +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +GOV_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +REGISTRY_PY="$GOV_ROOT/../CI-Workflows/pipeline/adversary/holdout_registry.py" -# 校验条目 sealed_sha256(ADR-0068 公式,与 unseal_gate.py 同式) -# 使用 CI-Workflows 的 holdout_registry.py 做 hash 校验 -CIW_DIR="${CIW_DIR:-CI-Workflows}" -REGISTRY_PY="$CIW_DIR/pipeline/adversary/holdout_registry.py" - -if [[ ! -f "$REGISTRY_PY" ]]; then - echo "::error::holdout_registry.py 不存在: $REGISTRY_PY(需先 checkout CI-Workflows)" - exit 2 +# 验证者 APP 身份解析 +VERIFIER_APP_ID="${VERIFIER_APP_ID:-}" +if [[ -z "$VERIFIER_APP_ID" || "$VERIFIER_APP_ID" == "null" ]]; then + # 尝试从 expected-state.json 读取 + EXPECTED_STATE="$GOV_ROOT/governance/expected-state.json" + if [[ -f "$EXPECTED_STATE" ]]; then + PY_BIN="" + for _c in python3 python py; do + if command -v "$_c" >/dev/null 2>&1 && "$_c" -c 'print(1)' >/dev/null 2>&1; then PY_BIN="$_c"; break; fi + done + if [[ -n "$PY_BIN" ]]; then + VERIFIER_APP_ID=$("$PY_BIN" -c "import json; print(json.load(open('$EXPECTED_STATE')).get('verifier_app',{}).get('id') or '')" 2>/dev/null || true) + fi + fi fi -echo "== holdout 注册入口(W2-C4 / ADR-0068)==" -echo "repo: $REPO" -echo "entry: $ENTRY_PATH" -echo "actor: $VERIFIER_SLUG" - -# 身份校验(AC-17):非验证者 APP 写入被拒 -ACTOR="${GH_ACTOR:-${ACTOR:-unknown}}" -if [[ "$ACTOR" != "$VERIFIER_SLUG" ]]; then - # 允许人类 owner 写入(workflow 层 owner 校验兜底) - if [[ "$ACTOR" == *"[bot]"* ]]; then - echo "::error::非验证者 APP 写入被拒(AC-17): actor=$ACTOR 期望=$VERIFIER_SLUG" - exit 1 +if [[ -z "$VERIFIER_APP_ID" || "$VERIFIER_APP_ID" == "null" ]]; then + echo "信息:验证者 APP 未就绪(verifier_app.id=null)— 以 dry-run 占位注册(不落 push)" >&2 + export HOLDOUT_REGISTRY_MODE=dry-run +else + # 取验证者 APP 令牌(单仓作用域=holdout) + if [[ -n "${VERIFIER_APP_SECRET_FILE:-}" || -n "${VERIFIER_APP_SECRET:-}" ]]; then + export CB_APP_ID="$VERIFIER_APP_ID" + export REPO="holdout" + echo "==> 铸验证者 APP 令牌(单仓作用域=holdout)" >&2 + VERIFIER_TOKEN=$(VERIFIER_APP_SECRET_FILE="${VERIFIER_APP_SECRET_FILE:-}" \ + VERIFIER_APP_SECRET="${VERIFIER_APP_SECRET:-}" \ + bash "$SCRIPT_DIR/gh-app-token.sh" 2>/tmp/verifier-token-err.log) || { + echo "警告:验证者 APP 令牌铸币失败($(head -1 /tmp/verifier-token-err.log 2>/dev/null))— 降级为 dry-run" >&2 + export HOLDOUT_REGISTRY_MODE=dry-run + } + if [[ -n "${VERIFIER_TOKEN:-}" ]]; then + export HOLDOUT_REGISTRY_MODE=verifier + echo "验证者 APP 身份就绪(id=$VERIFIER_APP_ID)" >&2 + fi + else + echo "警告:未提供验证者 APP 私钥(VERIFIER_APP_SECRET_FILE/VERIFIER_APP_SECRET)— 降级为 dry-run" >&2 + export HOLDOUT_REGISTRY_MODE=dry-run fi - echo "人类用户写入: $ACTOR(workflow 层 owner 校验兜底)" fi -# hash 校验 -echo "--- hash 校验 ---" -if ! python3 "$REGISTRY_PY" verify-hash --entry "$ENTRY_PATH"; then - echo "::error::hash 校验失败(ADR-0068)" - exit 1 +# 定位 registry 创建脚本 +if [[ ! -f "$REGISTRY_PY" ]]; then + # 备选:仓内相对路径(CI-Workflows 与 .github 同级目录结构) + ALT="$GOV_ROOT/../CI-Workflows/pipeline/adversary/holdout_registry.py" + [[ -f "$ALT" ]] && REGISTRY_PY="$ALT" fi -# 注册(经 holdout_registry.py) -echo "--- 注册条目 ---" -TOKEN="${APP_TOKEN:-${GH_TOKEN:-}}" -if [[ -z "$TOKEN" ]]; then - echo "::error::无验证者 APP 令牌(APP_TOKEN / GH_TOKEN 未设置)" +if [[ ! -f "$REGISTRY_PY" ]]; then + echo "错误:找不到 holdout_registry.py($REGISTRY_PY)" >&2 exit 2 fi -export GH_TOKEN="$TOKEN" -if python3 "$REGISTRY_PY" register --entry "$ENTRY_PATH" --actor "$VERIFIER_SLUG" --repo "$REPO"; then - echo "AUDIT | repo=$REPO | actor=$VERIFIER_SLUG | verdict=registered" - exit 0 -else - echo "::error::注册失败" - exit 1 -fi +echo "==> 注册 holdout 条目:$CARD/$ENTRY = ${SHA256:0:11} (mode=${HOLDOUT_REGISTRY_MODE})" >&2 +py "$REGISTRY_PY" register --card "$CARD" --entry "$ENTRY" --sha256 "$SHA256" --registry "$REGISTRY"