# -*- coding: utf-8 -*- """SOP-10 · B7 分批准入（每批 ≤5，累计 ≤15，四硬闸 + 人工签字）。 用法： # 出票：跑四硬闸 + 生成准入票据 + 写台账（待签字） python scripts/admit_batch.py --batch B7-1 --out data/B7_准入票据_B7-1.md # 复核：只读重算（不跑闸），落票据 python scripts/admit_batch.py --check --json reports/b7_admission_check.json # 人工签字：缺签字票据不作废，但不得进执行层 python scripts/admit_batch.py --sign --sign-by "姓名" --signed-at 2026-11-02 设计约束： 1. 候选只从 B5 池取（`data/semantic_family.json` 的 bridge_candidates + members）； 不在池内 → 直接拒绝出票。 2. 条数不补齐：票上几条就写几条，凑 5 条视为违规（A9）。 3. 四硬闸：v3_gate / unittest / verify_skill_matrix / redline_scan。 4. 签字是人工动作：脚本只记录「谁、何时」，不自称签字。 """ from pathlib import Path import argparse import json import os import re import subprocess import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from holoaeon.fusion_upgrade import admission as ADM # noqa: E402 REPO = ROOT RULES = os.path.join(REPO, "data", "admission_rules.json") LEDGER = os.path.join(REPO, "data", "admission_ledger.csv") FAMILY = os.path.join(REPO, "data", "semantic_family.json") GATES_JSON = os.path.join(REPO, "reports", "b7_gates.json") V3_JSON = os.path.join(REPO, "reports", "gate_fusion_v4.json") CHECK_JSON = os.path.join(REPO, "reports", "b7_admission_check.json") def read_json(path): with open(path, encoding="utf-8") as f: return json.load(f) def write_json(path, doc): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8", newline="\n") as f: f.write(json.dumps(doc, ensure_ascii=False, indent=2) + "\n") def run(cmd): env = dict(os.environ, PYTHONIOENCODING="utf-8") p = subprocess.run(cmd, cwd=REPO, capture_output=True, text=True, encoding="utf-8", errors="replace", env=env) return p.returncode, (p.stdout or "") + (p.stderr or "") def run_gates(): """跑四道硬闸并解析成 {gate_id: {passed,total,ok,cmd,raw_tail}}。""" out = {} for gid, cmd in ADM.HARD_GATES: rc, text = run(cmd.split()) tail = text.strip().splitlines()[-1] if text.strip() else "" passed = total = None if gid == "v3_gate": try: d = read_json(V3_JSON) passed, total = d.get("passed"), d.get("total") except Exception: pass elif gid == "unittest": m = re.search(r"Ran (\d+) tests", text) if m: passed = total = int(m.group(1)) elif gid == "skill_matrix": try: d = json.loads(text[text.index("{"):]) st = d.get("selftest", {}) passed = int(st.get("passed") or 0) total = len(st.get("detail") or []) except Exception: pass elif gid == "redline": m = re.search(r"扫描文件数：(\d+)", text) if m: passed = total = int(m.group(1)) ok = rc == 0 and bool(passed) and passed == total out[gid] = {"passed": passed, "total": total, "ok": ok, "cmd": cmd, "raw_tail": tail[:200]} return out def load_pool(): fam = read_json(FAMILY) skills = {c["skill"] for c in fam.get("bridge_candidates", [])} skills |= {m.get("skill") for m in fam.get("members", []) if m.get("skill")} return fam, skills def build(batch, out_path, skills_arg=None): fam, pool = load_pool() if skills_arg: chosen = [s.strip() for s in skills_arg.split(",") if s.strip()] else: chosen = [c["skill"] for c in fam.get("bridge_candidates", [])] outside = [s for s in chosen if s not in pool] if outside: print(json.dumps({"verdict": "REFUSED", "reason": "outside_pool", "skills": outside}, ensure_ascii=False)) return 2 if len(chosen) > ADM.MAX_PER_BATCH: print(json.dumps({"verdict": "REFUSED", "reason": "over_per_batch", "count": len(chosen)}, ensure_ascii=False)) return 2 cand = {c["skill"]: c for c in fam.get("bridge_candidates", [])} for m in fam.get("members", []): cand.setdefault(m.get("skill"), m) gates = run_gates() write_json(GATES_JSON, {"gates_id": "SG-B7-HARD-GATES-V1", "batch": batch, "evidence_class": ADM.EVIDENCE_CLASS, "gates": gates}) gate_cells = {gid: ("%s/%s" % (gates[gid]["passed"], gates[gid]["total"]) if gates[gid]["passed"] else "FAIL") for gid in ADM.GATE_IDS} rows = [] for s in chosen: c = cand.get(s, {}) rows.append({ "batch": batch, "ticket": out_path, "skill": s, "target_mechanism": c.get("target_mechanism", "semantic_bus"), "module": c.get("module", ""), "usage_class": c.get("label", ""), "gate_v3_gate": gate_cells["v3_gate"], "gate_unittest": gate_cells["unittest"], "gate_skill_matrix": gate_cells["skill_matrix"], "gate_redline": gate_cells["redline"], "signature_status": "PENDING", "signer": "", "signed_at": "", "decision": ADM.DECISION_WAITING, "note": "B5 candidate_bridge；未签字不进执行层", }) # 幂等：同批行整段替换，其它批保留 old = [] if os.path.isfile(LEDGER): old = ADM.parse_ledger(Path(LEDGER).read_text(encoding="utf-8")) kept = [r for r in old if r.get("batch") != batch] allrows = kept + rows os.makedirs(os.path.dirname(LEDGER), exist_ok=True) open(LEDGER, "w", encoding="utf-8", newline="\n").write(ADM.dump_ledger(allrows)) res = ADM.validate(allrows, pool=pool, gate_results=gates) write_ticket(out_path, batch, rows, gates, res) print(json.dumps({"batch": batch, "ticket": out_path, "rows": len(rows), "verdict": res["verdict"], "gates_ok": all(g["ok"] for g in gates.values()), "admitted": res["counts"]["admitted"], "remaining": res["counts"]["remaining"]}, ensure_ascii=False)) return 0 def write_ticket(path, batch, rows, gates, res, signature=None): g = gates ok_all = all(v["ok"] for v in g.values()) lines = [] lines.append("# B7 分批准入票据 · %s" % batch) lines.append("") lines.append("> 规则 `%s` · 每批 ≤%d · 累计 ≤%d · 四硬闸 + 人工签字" % (ADM.RULES_ID, ADM.MAX_PER_BATCH, ADM.MAX_CUMULATIVE)) lines.append("> evidence_class=`%s`（仿真/离线证据，不等于治疗证据）" % ADM.EVIDENCE_CLASS) lines.append("") lines.append("## 一、判据与来源") lines.append("") lines.append("判据取自 `holoaeon/fusion_upgrade/plan.py` B7 段与 SOP-10：") lines.append("") lines.append("1. 每批 ≤%d 个技能；" % ADM.MAX_PER_BATCH) lines.append("2. 四道硬闸全绿（v3_gate / unittest / verify_skill_matrix / redline_scan）；") lines.append("3. 每批有签字人姓名与日期；缺签字 → `WAITING_SIGNATURE`，不进执行层；") lines.append("4. 累计准入 ≤%d；" % ADM.MAX_CUMULATIVE) lines.append("5. 准入票据存在（本文件）。") lines.append("") lines.append("## 二、本批候选（只从 B5 池取，不扩池、不补齐）") lines.append("") lines.append("| # | 技能 | 目标机制 | 模块 | B3 标签 | 来源状态 |") lines.append("| --- | --- | --- | --- | --- | --- |") for i, r in enumerate(rows, 1): lines.append("| %d | `%s` | %s | %s | %s | candidate_bridge |" % (i, r["skill"], r["target_mechanism"], r["module"], r["usage_class"])) lines.append("") k_count = sum(1 for r in rows if r["usage_class"] == "K") lines.append("本批 **%d 条**（≤%d，未按 5 补齐）。全部取自 B5 登记池（bridge_candidates + 族成员），" "其中 B3 标签 K（知识源）%d 条、非 K %d 条 —— 非 K 条目须人工复核后才可进执行层。" % (len(rows), ADM.MAX_PER_BATCH, k_count, len(rows) - k_count)) lines.append("") lines.append("## 三、四道硬闸实测") lines.append("") lines.append("| 闸 | 命令 | 结果 | 判定 |") lines.append("| --- | --- | --- | --- |") for gid, _cmd in ADM.HARD_GATES: v = g[gid] lines.append("| %s | `%s` | %s/%s | %s |" % (gid, v["cmd"], v["passed"], v["total"], "PASS" if v["ok"] else "FAIL")) lines.append("") lines.append("闸全绿：**%s**；闸票据 `reports/b7_gates.json`。" % ("是" if ok_all else "否")) lines.append("") lines.append("## 四、准入结论") lines.append("") lines.append("- 判定：`%s`" % (signature["decision"] if signature else res["verdict"])) lines.append("- 已准入累计：**%d / %d**（剩余 %d）" % (res["counts"]["admitted"], ADM.MAX_CUMULATIVE, res["counts"]["remaining"])) lines.append("- 未签字前，本批 %d 条技能**不进执行层**，`kernel.skills` 恒 47。" % len(rows)) lines.append("") lines.append("## 五、人工签字") lines.append("") if signature: lines.append("- 签字人：**%s**" % signature["signer"]) lines.append("- 签字日期：**%s**" % signature["signed_at"]) lines.append("- 签字后结论：`ADMITTED`（仍不计入 `kernel.skills`，只解除执行层禁令）") else: lines.append("待签字（签名与日期两项缺一不可；缺签字票据不作废但不得进执行层）。") lines.append("") lines.append("```") lines.append("签字人：____________________ 日期：____________________") lines.append("```") lines.append("") lines.append("## 六、可复现") lines.append("") lines.append("```bash") lines.append("python scripts/admit_batch.py --batch %s --out %s # 出票" % (batch, path)) lines.append("python scripts/admit_batch.py --check --json reports/b7_admission_check.json") lines.append("python scripts/admit_batch.py --sign --sign-by \"姓名\" --signed-at 2026-11-02") lines.append("python -m unittest discover -s tests_contract # 旁挂全部单测") lines.append("```") lines.append("") lines.append("## 七、维度不变自证") lines.append("") lines.append("| 维度 | 锁值 | 本批后 | 结论 |") lines.append("| --- | --- | --- | --- |") lines.append("| agents | 25 | 25 | 未动 |") lines.append("| skills | 47 | 47 | 未动（准入只改登记态，不进 `kernel.skills`） |") lines.append("| packs | 9 | 9 | 未动（未新增 pack） |") lines.append("| spec.VERSION | 0.9.1 | 0.9.1 | 未动 |") lines.append("") os.makedirs(os.path.dirname(path), exist_ok=True) open(path, "w", encoding="utf-8", newline="\n").write("\n".join(lines) + "\n") def check(json_out=None): rules_sha_ok = True if os.path.isfile(RULES): rules_sha_ok = read_json(RULES).get("sha256") == ADM.rules_doc()["sha256"] if not os.path.isfile(LEDGER): print(json.dumps({"verdict": "FAIL", "reason": "ledger_missing"}, ensure_ascii=False)) return 1 rows = ADM.parse_ledger(Path(LEDGER).read_text(encoding="utf-8")) fam, pool = load_pool() gates = read_json(GATES_JSON)["gates"] if os.path.isfile(GATES_JSON) else None tickets = {} for b in sorted({r["batch"] for r in rows}): p = next(r["ticket"] for r in rows if r["batch"] == b) if os.path.isfile(p): tickets[b] = Path(p).read_text(encoding="utf-8") res = ADM.validate(rows, pool=pool, ticket_texts=tickets, gate_results=gates) extra = [] if not rules_sha_ok: extra.append("rules_sha256_drift") payload = { "check_id": "SG-B7-ADMISSION-CHECK-V1", "rules_id": ADM.RULES_ID, "evidence_class": ADM.EVIDENCE_CLASS, "verdict": "FAIL" if extra else res["verdict"], "failed": extra + res["violations"], "checks": res["checks"], "counts": res["counts"], "rules_sha256_match": rules_sha_ok, "gates": {k: {"passed": v["passed"], "total": v["total"], "ok": v["ok"]} for k, v in (gates or {}).items()}, } if json_out: write_json(json_out, payload) print(json.dumps({k: payload[k] for k in ("verdict", "failed", "counts")}, ensure_ascii=False)) return 0 if payload["verdict"] in ("PASS", ADM.DECISION_WAITING) else 1 def sign(batch, signer, signed_at): if not signer or not signed_at: print(json.dumps({"verdict": "REFUSED", "reason": "signature_incomplete"}, ensure_ascii=False)) return 2 if not os.path.isfile(LEDGER): print(json.dumps({"verdict": "REFUSED", "reason": "ledger_missing"}, ensure_ascii=False)) return 2 rows = ADM.parse_ledger(Path(LEDGER).read_text(encoding="utf-8")) hit = [r for r in rows if r["batch"] == batch] if not hit: print(json.dumps({"verdict": "REFUSED", "reason": "batch_not_found"}, ensure_ascii=False)) return 2 if ADM.admitted_count(rows) + len(hit) > ADM.MAX_CUMULATIVE: print(json.dumps({"verdict": "REFUSED", "reason": "over_cumulative", "admitted": ADM.admitted_count(rows), "batch": len(hit)}, ensure_ascii=False)) return 2 gates = read_json(GATES_JSON)["gates"] if os.path.isfile(GATES_JSON) else None if gates and not all(v["ok"] for v in gates.values()): print(json.dumps({"verdict": "REFUSED", "reason": "hard_gate_red"}, ensure_ascii=False)) return 2 for r in rows: if r["batch"] == batch: r["signature_status"] = "SIGNED" r["signer"] = signer r["signed_at"] = signed_at r["decision"] = ADM.DECISION_ADMITTED open(LEDGER, "w", encoding="utf-8", newline="\n").write(ADM.dump_ledger(rows)) ticket = hit[0]["ticket"] fam, pool = load_pool() tickets = {} for b in sorted({r["batch"] for r in rows}): p = next(r["ticket"] for r in rows if r["batch"] == b) if os.path.isfile(p): tickets[b] = Path(p).read_text(encoding="utf-8") res = ADM.validate(rows, pool=pool, ticket_texts=tickets, gate_results=gates) batch_only = [r for r in rows if r["batch"] == batch] write_ticket(ticket, batch, batch_only, gates or {}, res, signature={"signer": signer, "signed_at": signed_at, "decision": "ADMITTED"}) print(json.dumps({"batch": batch, "signed_by": signer, "signed_at": signed_at, "admitted": res["counts"]["admitted"], "remaining": res["counts"]["remaining"]}, ensure_ascii=False)) return 0 def main(): ap = argparse.ArgumentParser() ap.add_argument("--batch", default="B7-1") ap.add_argument("--skills", default="") ap.add_argument("--out", default="") ap.add_argument("--check", action="store_true") ap.add_argument("--json", default="") ap.add_argument("--sign", action="store_true") ap.add_argument("--sign-by", default="") ap.add_argument("--signed-at", default="") a = ap.parse_args() if a.sign: return sign(a.batch, a.sign_by, a.signed_at) if a.check: return check(a.json or None) out = a.out or os.path.join(REPO, "data", "B7_准入票据_%s.md" % a.batch) return build(a.batch, out, a.skills or None) if __name__ == "__main__": sys.exit(main())