feat(drill): 注入/独立验证/台账引擎（W4-C4 .github#223，ADR-0069）

.github/workflows/gate.yml

@@ -89,7 +89,10 @@ jobs:
           bash -n governance/apply.sh && bash -n governance/drift-check.sh && bash -n scripts/new-repo-init.sh && bash -n scripts/gh-app-token.sh && bash -n scripts/ghcb \
           && bash -n governance/auto-fix-limit.sh && bash -n governance/cost-check.sh \
           && bash -n governance/butler-reconcile.sh \
-          && bash -n governance/deadman-trip.sh && bash -n governance/butler-audit.sh
+          && bash -n governance/deadman-trip.sh && bash -n governance/butler-audit.sh \
+          && bash -n governance/tests/test-drill.sh \
+          && bash -n governance/drill/tests/test-samples.sh && bash -n governance/drill/tests/test-select.sh \
+          && bash -n governance/drill/tests/test-verify.sh
           echo "OK scripts"
       - name: 治理脚本自测（governance/tests）
         run: |

governance/drill/drill.py

@@ -1,16 +1,19 @@
 #!/usr/bin/env python3
-"""drill.py —— 周种子缺陷演习引擎·读面（宪法 §4B/§6 / ADR-0069 / .github#223 W4-C4）
+"""drill.py —— 周种子缺陷演习引擎（宪法 §4B/§6 / ADR-0069 / .github#223 W4-C4）
 
 子命令（全部 fail-closed：任何失败非零退出，绝不静默降级为"通过"）:
   select   随机选缺陷样本 + 随机目标仓（--seed 注入可复现——测试与事后复盘）
+  inject   向目标仓开演习分支（只开分支不开 PR）注入缺陷，输出 head_sha
   decode   owner 审阅用：解码样本缺陷内容（ADR-0069 决策 1 样本库 owner 直管）
+  record   演习记录追加 governance/drill/history.jsonl（append-only 台账=dashboard 数据源）
+  redrate  聚合红率（目标 ≈100%，AC-4）+ 样本难度按周分布（防"为演习写代码"Goodhart）
 
-注入/独立验证/台账（inject/record/redrate）随后续 PR 落地——红绿判定与样本库
-分离（verify_gate.py，ADR-0069 决策 2"注入者与判定者分离"）。注入物只落在
-演习分支（隔离执行，不进 agent 工作区，ADR-0069 风险缓解）；分支验后即删。
+红绿判定不在本文件——verify_gate.py 与样本库分离（独立验证步骤，ADR-0069
+决策 2"注入者与判定者分离"）。注入物只落在演习分支（隔离执行，不进 agent
+工作区，ADR-0069 风险缓解）；分支验后即删。
 """
 import argparse
-import base64  # decode 用（样本 defect_b64）
+import base64
 import json
 import os
 import random
@@ -139,6 +142,60 @@ def cmd_select(a):
                      ensure_ascii=False))
 
 
+def _git(env_extra, *args, cwd=None, check=True):
+    env = dict(os.environ, GIT_TERMINAL_PROMPT="0", **env_extra)
+    p = subprocess.run(["git", *args], cwd=cwd, env=env,
+                       stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    if check and p.returncode != 0:
+        die(f"git {' '.join(args[:3])} 失败(rc={p.returncode}): {p.stdout.strip()[:400]}")
+    return p
+
+
+def cmd_inject(a):
+    samples = {s["id"]: s for s in load_samples(a.samples)}
+    if a.sample_id not in samples:
+        die(f"样本不存在: {a.sample_id}")
+    s = samples[a.sample_id]
+    date = a.date or datetime.now(timezone.utc).strftime("%Y%m%d")
+    prefix = os.environ.get("DRILL_GIT_PREFIX", "https://github.com")
+    push_url = f"https://github.com/{ORG}/{a.repo}.git"  # push 一律直连（镜像仅 fetch）
+    auth = {}
+    tok = os.environ.get("DRILL_TOKEN")
+    if tok:  # CI: GOVERNANCE_TOKEN 经 git env-config 下发 extraheader，不落 remote URL/日志
+        hdr = "Authorization: Basic " + base64.b64encode(f"x-access-token:{tok}".encode()).decode()
+        auth = {"GIT_CONFIG_COUNT": "1",
+                "GIT_CONFIG_KEY_0": "http.https://github.com/.extraheader",
+                "GIT_CONFIG_VALUE_0": hdr}
+    branch = a.branch or f"drill/seed-{date}"
+    with tempfile.TemporaryDirectory(prefix="drill-") as tmp:
+        _git({}, "clone", "--depth", "1", f"{prefix}/{ORG}/{a.repo}.git", tmp)
+        _git({}, "checkout", "-B", branch, cwd=tmp)
+        path = os.path.join(tmp, *s["payload_path"].format(DATE=date).split("/"))
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        if s["payload_kind"] == "file":
+            data = base64.b64decode(s["defect_b64"])
+        else:  # generated: 零填充大文件（hygiene >5MB 禁入规则的靶）
+            data = b"\0" * s["size_bytes"]
+        with open(path, "wb") as f:
+            f.write(data)
+        _git({}, "add", path, cwd=tmp)
+        _git({}, "-c", "user.name=drill-seed-bot", "-c", "user.email=drill-bot@users.noreply.github.com",
+             "commit", "-m", f"drill(seed): 注入演习样本 {s['id']}（ADR-0069 周演习，验后即删）",
+             cwd=tmp)
+        sha = _git({}, "rev-parse", "HEAD", cwd=tmp).stdout.strip()
+        last_rc = 1
+        for _ in range(3):  # push 间歇失败重试（org 网络现实）
+            p = _git(auth, "push", push_url, f"HEAD:refs/heads/{branch}", cwd=tmp, check=False)
+            last_rc = p.returncode
+            if last_rc == 0:
+                break
+        if last_rc != 0:
+            die(f"演习分支 push 失败（3 次重试后）: {p.stdout.strip()[:400]}")
+    print(json.dumps({"branch": branch, "head_sha": sha,
+                      "path": s["payload_path"].format(DATE=date), "sample_id": s["id"]},
+                     ensure_ascii=False))
+
+
 def cmd_decode(a):
     samples = {s["id"]: s for s in load_samples(a.samples)}
     if a.id not in samples:
@@ -150,6 +207,59 @@ def cmd_decode(a):
     sys.stdout.write(base64.b64decode(s["defect_b64"]).decode("utf-8"))
 
 
+def cmd_record(a):
+    """append-only 台账：时间戳严格递增 + 同 run 不重复（AC-4 聚合的数据完整性前提）。"""
+    try:
+        rec = json.loads(a.json)
+    except Exception as e:
+        die(f"--json 解析失败: {e}")
+    if not rec.get("ts") or not rec.get("kind"):
+        die("记录缺 ts/kind 字段")
+    lines = []
+    if os.path.exists(a.history):
+        lines = [l for l in open(a.history, encoding="utf-8").read().splitlines() if l.strip()]
+        for l in lines:
+            prev = json.loads(l)  # 既有行畸形=台账已损坏，拒绝追加（fail-closed）
+            if prev["ts"] >= rec["ts"]:
+                die(f"append-only 破坏: 新 ts {rec['ts']} 不晚于末行 {prev['ts']}")
+            if (prev.get("kind"), prev.get("run_id")) == (rec["kind"], rec.get("run_id")) \
+                    and rec.get("run_id") is not None:
+                die(f"同一 run 的 {rec['kind']} 记录已存在（run_id={rec['run_id']}）")
+    with open(a.history, "a", encoding="utf-8", newline="\n") as f:
+        f.write(json.dumps(rec, ensure_ascii=False, sort_keys=True) + "\n")
+    print(f"OK append 1 行（现有 {len(lines) + 1} 行）")
+
+
+def cmd_redrate(a):
+    """红率（目标 ≈100%）+ 难度按周分布（AC-4；Goodhart 防护=趋势可见而非打分）。"""
+    if not os.path.exists(a.history):
+        die(f"台账不存在: {a.history}")
+    rows = [json.loads(l) for l in open(a.history, encoding="utf-8").read().splitlines() if l.strip()]
+    seeds = [r for r in rows if r.get("kind") == "seed-drill"]
+    verdicts = {}
+    for r in seeds:
+        verdicts[r.get("verdict", "?")] = verdicts.get(r.get("verdict", "?"), 0) + 1
+    red, green = verdicts.get("red", 0), verdicts.get("green", 0)
+    denom = red + green
+    # 零分母 → null（诚实口径：不除零不出假 100%，同 dashboard-update SLI 原则）
+    rate = round(red / denom, 4) if denom else None
+    weeks = {}
+    for r in seeds:
+        wk = r["ts"][:10]
+        d = r.get("difficulty", "?")
+        weeks.setdefault(wk, {}).setdefault(d, 0)
+        weeks[wk][d] += 1
+    fails = [r for r in rows if r.get("kind") == "failclose-drill"]
+    out = {"total_records": len(rows), "seed_drills": len(seeds), "verdicts": verdicts,
+           "red_rate": rate, "red_rate_note": None if denom else "N/A（尚无可判定演习）",
+           "difficulty_trend_by_day": {k: dict(sorted(v.items())) for k, v in sorted(weeks.items())},
+           "failclose_drills": len(fails),
+           "failclose_last": fails[-1].get("outcome") if fails else None}
+    print(json.dumps(out, ensure_ascii=False, indent=2))
+    if a.fail_unhealthy and denom and rate < 0.999:
+        die(f"红率 {rate} < 100%——存在关卡失灵（green={green}），按 AC-4 告警")
+
+
 def main():
     ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     sub = ap.add_subparsers(dest="cmd", required=True)
@@ -163,11 +273,29 @@ def main():
     p.add_argument("--target-repo", help="跳过随机，指定目标仓（复盘/首演用）")
     p.set_defaults(func=cmd_select)
 
+    p = sub.add_parser("inject", help="开演习分支注入缺陷")
+    p.add_argument("--samples", default=os.path.join(here, "samples", "registry.yaml"))
+    p.add_argument("--sample-id", required=True)
+    p.add_argument("--repo", required=True)
+    p.add_argument("--branch", default=None, help="缺省 drill/seed-YYYYMMDD")
+    p.add_argument("--date", default=None, help="payload {DATE} 占位值（缺省今天）")
+    p.set_defaults(func=cmd_inject)
+
     p = sub.add_parser("decode", help="owner 审阅：解码样本内容")
     p.add_argument("--samples", default=os.path.join(here, "samples", "registry.yaml"))
     p.add_argument("--id", required=True)
     p.set_defaults(func=cmd_decode)
 
+    p = sub.add_parser("record", help="追加 history.jsonl（append-only）")
+    p.add_argument("--history", default=os.path.join(here, "history.jsonl"))
+    p.add_argument("--json", required=True)
+    p.set_defaults(func=cmd_record)
+
+    p = sub.add_parser("redrate", help="红率+难度趋势聚合（AC-4）")
+    p.add_argument("--history", default=os.path.join(here, "history.jsonl"))
+    p.add_argument("--fail-unhealthy", action="store_true",
+                   help="红率<100% 时非零退出（workflow 告警用）")
+    p.set_defaults(func=cmd_redrate)
 
     a = ap.parse_args()
     if a.cmd == "select":

governance/drill/tests/test-select.sh

@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+# test-select.sh —— 选样随机性自测（W4-C4）：seed 可注入（确定性复盘）+ 目标池约束
+set -uo pipefail
+HERE="$(cd "$(dirname "$0")" && pwd)"
+ROOT="$(cd "$HERE/.." && pwd)"
+source "$(cd "$(dirname "$0")" && pwd)/lib.sh"
+PYTHON="$(pick_py)" || { echo "::error::无可用 python（含 pyyaml）"; exit 2; }
+PASS=0; FAIL=0
+TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
+
+sel() { "$PYTHON" "$ROOT/drill.py" select --samples "$ROOT/samples/registry.yaml" \
+          --repos "$ROOT/../REPOS.yaml" "$@"; }
+
+echo "== 1) 同 seed 确定性（随机可注入=可复盘可审计）"
+sel --seed 20260822 > "$TMP/a.json"; sel --seed 20260822 > "$TMP/b.json"
+if cmp -s "$TMP/a.json" "$TMP/b.json"; then PASS=$((PASS+1)); echo "ok   同 seed 输出一致"
+else FAIL=$((FAIL+1)); echo "FAIL 同 seed 输出漂移"; cat "$TMP/a.json" "$TMP/b.json"; fi
+
+echo "== 2) 目标池约束（holdout 永不入池；github-scope 样本只打治理总仓）"
+BAD=0
+for s in 1 2 3 5 8 13 21 34 55; do
+  R=$(sel --seed "$s" | "$PYTHON" -c 'import json,sys;print(json.load(sys.stdin)["target_repo"])')
+  [[ "$R" == "holdout" ]] && { echo "FAIL seed=$s 选中 holdout（隔离面污染）"; BAD=1; }
+done
+[[ $BAD -eq 0 ]] && { PASS=$((PASS+1)); echo "ok   多 seed 采样 holdout 零命中（owner 直管封存面隔离）"; } || FAIL=$((FAIL+1))
+
+G=$(sel --seed 7 --sample-id gate-yaml-parse-corrupt | "$PYTHON" -c 'import json,sys;print(json.load(sys.stdin)["target_repo"])')
+[[ "$G" == ".github" ]] && { PASS=$((PASS+1)); echo "ok   github-scope 样本固定目标=.github"; } \
+  || { FAIL=$((FAIL+1)); echo "FAIL github-scope 样本目标=$G"; }
+
+echo "== 3) 固定样本/目标（首演与复盘通道）"
+O=$(sel --seed 99 --sample-id hygiene-gitleaks-aws-key --target-repo .github)
+echo "$O" | grep -q '"sample_id": "hygiene-gitleaks-aws-key"' && echo "$O" | grep -q '"target_repo": ".github"' \
+  && { PASS=$((PASS+1)); echo "ok   --sample-id/--target-repo 钉选生效"; } \
+  || { FAIL=$((FAIL+1)); echo "FAIL 钉选失效: $O"; }
+
+echo "== 4) 选样输出字段完整（AC-2 关卡 ID + AC-4 难度随记录可溯）"
+sel --seed 3 | "$PYTHON" -c '
+import json, sys
+d = json.load(sys.stdin)
+assert d["sample_id"] and d["gate"] and d["difficulty"] in ("easy", "medium", "hard") and d["target_repo"], d
+' && { PASS=$((PASS+1)); echo "ok   字段齐全"; } || { FAIL=$((FAIL+1)); echo "FAIL 字段缺失"; }
+
+echo "== 5) 非法钉选被拒（fail-closed，不静默回退随机）"
+sel --seed 1 --sample-id no-such-sample >/dev/null 2>&1; [[ $? -ne 0 ]] \
+  && { PASS=$((PASS+1)); echo "ok   未知样本 id 非零退出"; } || { FAIL=$((FAIL+1)); echo "FAIL 未知样本被放行"; }
+
+echo "选样自测: pass=$PASS fail=$FAIL"
+[[ $FAIL -eq 0 ]]

governance/drill/tests/test-verify.sh

@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# test-verify.sh —— 红绿判定断言自测（W4-C4 AC-1 判定核）：离线注入 check-runs，
+# 红=演习成功(0) / 绿=演习失败(1) / 中间态(3)——"没看到"绝不装绿。
+set -uo pipefail
+HERE="$(cd "$(dirname "$0")" && pwd)"
+ROOT="$(cd "$HERE/.." && pwd)"
+source "$(cd "$(dirname "$0")" && pwd)/lib.sh"
+PYTHON="$(pick_py)" || { echo "::error::无可用 python（含 pyyaml）"; exit 2; }
+export DRILL_VERIFY_GRACE_S=0 # 静默期压零——离线模式立即判定（默认 120s 仅在线用）
+PASS=0; FAIL=0
+TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
+
+vf() { "$PYTHON" "$ROOT/verify_gate.py" --repo dot --sha deadbeef "$@"; }
+verdict_of() { "$PYTHON" -c 'import json,sys;print(json.load(sys.stdin)["verdict"])'; }
+
+mkcr() { printf '{"check_runs": [%s]}' "$1" > "$TMP/$2"; }
+
+echo "== 1) 红=演习成功（关卡 conclusion=failure → 退出码 0）"
+mkcr '{"name":"org-hygiene","status":"completed","conclusion":"failure"}' red.json
+OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/red.json"); RC=$?
+V=$(verdict_of <<<"$OUT")
+if [[ "$V" == "RED" && $RC -eq 0 ]]; then PASS=$((PASS+1)); echo "ok   failure→RED/exit0"
+else FAIL=$((FAIL+1)); echo "FAIL 期望 RED/0 实得 $V/$RC"; fi
+
+echo "== 2) 绿=演习失败（关卡 success → 退出码 1，绝不装绿）"
+mkcr '{"name":"org-hygiene","status":"completed","conclusion":"success"}' green.json
+OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/green.json"); RC=$?
+V=$(verdict_of <<<"$OUT")
+if [[ "$V" == "GREEN" && $RC -eq 1 ]]; then PASS=$((PASS+1)); echo "ok   success→GREEN/exit1"
+else FAIL=$((FAIL+1)); echo "FAIL 期望 GREEN/1 实得 $V/$RC"; fi
+
+echo "== 3) NO-SURFACE（空 check 面=push 分支无 CI 局限，如实记录不装绿）"
+mkcr '' empty.json
+OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/empty.json"); RC=$?
+V=$(verdict_of <<<"$OUT")
+if [[ "$V" == "NO-SURFACE" && $RC -eq 3 ]]; then PASS=$((PASS+1)); echo "ok   空→NO-SURFACE/exit3"
+else FAIL=$((FAIL+1)); echo "FAIL 期望 NO-SURFACE/3 实得 $V/$RC"; fi
+
+echo "== 4) MISSING-GATE（有别的 check、无目标关卡=名字漂移，不猜）"
+mkcr '{"name":"lint","status":"completed","conclusion":"success"}' other.json
+OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/other.json"); RC=$?
+V=$(verdict_of <<<"$OUT")
+if [[ "$V" == "MISSING-GATE" && $RC -eq 3 ]]; then PASS=$((PASS+1)); echo "ok   缺关卡→MISSING-GATE/exit3"
+else FAIL=$((FAIL+1)); echo "FAIL 期望 MISSING-GATE/3 实得 $V/$RC"; fi
+
+echo "== 5) 子串唯一兜底 + 进行中不算结论（等不到=TIMEOUT）"
+mkcr '{"name":"org-hygiene / gitleaks","status":"completed","conclusion":"failure"}' substr.json
+OUT=$(vf --gate gitleaks --checkruns-file "$TMP/substr.json"); RC=$?
+V=$(verdict_of <<<"$OUT")
+[[ "$V" == "RED" && $RC -eq 0 ]] && { PASS=$((PASS+1)); echo "ok   精确名缺席时子串唯一命中"; } \
+  || { FAIL=$((FAIL+1)); echo "FAIL 子串匹配: $V/$RC"; }
+mkcr '{"name":"org-hygiene","status":"in_progress","conclusion":null}' wip.json
+OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/wip.json" --timeout 0); RC=$?
+V=$(verdict_of <<<"$OUT")
+[[ "$V" == "TIMEOUT" && $RC -eq 3 ]] && { PASS=$((PASS+1)); echo "ok   in_progress 不判结论→TIMEOUT"; } \
+  || { FAIL=$((FAIL+1)); echo "FAIL 进行中语义: $V/$RC"; }
+
+echo "== 6) 歧义子串不猜（两个候选=MISSING-GATE）"
+mkcr '{"name":"org-hygiene-a","status":"completed","conclusion":"failure"},{"name":"org-hygiene-b","status":"completed","conclusion":"failure"}' amb.json
+OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/amb.json"); RC=$?
+V=$(verdict_of <<<"$OUT")
+[[ "$V" == "MISSING-GATE" ]] && { PASS=$((PASS+1)); echo "ok   歧义拒绝猜测"; } \
+  || { FAIL=$((FAIL+1)); echo "FAIL 歧义应 MISSING-GATE 实得 $V"; }
+
+echo "== 7) API 模式无凭据 fail-closed（不静默跳过）"
+unset GH_TOKEN || true
+vf --gate org-hygiene >/dev/null 2>&1; RC=$?
+[[ $RC -eq 2 ]] && { PASS=$((PASS+1)); echo "ok   无 GH_TOKEN 拒跑（exit2）"; } \
+  || { FAIL=$((FAIL+1)); echo "FAIL 无凭据 rc=$RC"; }
+
+echo "验证核自测: pass=$PASS fail=$FAIL"
+[[ $FAIL -eq 0 ]]