| |
| """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 |
|
|
| 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()) |
|
|
