"""full_gate.py — 生成线验收 full_gate 集成 runner（W-AXIS 收口 R1）。 把六个已成熟的子门部件串成单一 full_gate decision，只消费子门既有产物/函数，不重写任何子门逻辑： ① 九门机械预筛 ← Node tools.mjs check + play.cdp.cjs 驱动器产出的 verdict（pass/guards） ② 视觉地板 ← cheap_verify.judge_gameplay_floor 的裁决（parse_floor_judgment 结构：accepted/degraded） ③ playtest/3 真玩 ← cheap_verify.run_acceptance_v3 封存 payload（双 Judge + rollGuard + merge + finalPostguard） ④ prompt 四门 ← eval_gate.py 真模型闸台账记录（gate1 schema / gate2 成功率 / gate3 回归 / gate4 成本延迟 + gate5 稳定） ⑤ schema 语义 ← contracts/play-loop/validate.py（经 cheap_verify.validate_acceptance_v3_payload 调 canonical 校验） ⑥ W-GOLD-LIVE 对账 ← cheap_verify.reconcile_v3_reference_asset_consumption（参照资产消费六项闸） 降级口径（设计档 §3.2 结果四态 + §5.2 失败归因，焊死）： · 全过 → outcome=accept、pass=True（唯一可接受路径）； · 任一子门 tester_error / 缺产物 / degraded / 自相矛盾 → 整体 tester_error（仪器异常不得伪装成 accept， 也不得伪装成 gameplay reject——§5.2「Actor/runner/Judge/schema/图像/环境异常 100% 归 tester_error」）； · 任一子门 reject → 整体 reject（硬证已证的产物缺陷）； · 任一子门 inconclusive → 整体 inconclusive（合法证据不足或矛盾）； · 优先级 tester_error > reject > inconclusive > accept：仪器异常时证据不可信，不能 claiming 已证缺陷。 本步为纯本地代码 + 罐头单测；真跑（真模型真浏览器）留 mini-desktop，经 run_full_gate_live 编排。 """ import sys from pathlib import Path # 与 cheap-worker 其余模块同惯例：本目录直挂 sys.path，import cheap_verify 复用子门函数。 sys.path.insert(0, str(Path(__file__).resolve().parent)) import cheap_verify as V # noqa: E402 子门函数单一来源：地板判定/v3 编排/canonical 校验/金标对账 # 设计档 §3.2 结果四态；full_gate 与子门共用同一取值集，未知取值一律 fail-closed 归 tester_error。 _OUTCOMES = ("accept", "reject", "inconclusive", "tester_error") # 子门状态合并优先级：数值越小越严重。仪器异常（tester_error）压过已证缺陷（reject）压过证据不足（inconclusive）。 _SEVERITY = {"tester_error": 0, "reject": 1, "inconclusive": 2, "pass": 3} # ────────────────────────── 六个子门消费函数（只消费产物，不重写逻辑）────────────────────────── def _eval_mechanical_nine(verdict) -> dict: """① 九门机械预筛：消费 Node harness verdict（A..I guards + pass AND）。 缺产物 → tester_error（无机械证据，fail-closed）；guards 与 pass 自相矛盾 → tester_error（harness 异常）； 明确未过 → reject（结构性/装载/渲染/接线机械缺陷）。 """ if not isinstance(verdict, dict): return {"state": "tester_error", "reason": "九门 verdict 产物缺失，无机械预筛证据", "failedGates": []} # check 级结构失败（如 src/ 不存在）：verdict.ok=False 携带 errors，属机械缺陷。 if verdict.get("ok") is False: errors = [str(e) for e in (verdict.get("errors") or [])][:6] return {"state": "reject", "reason": "机械结构检查未过:" + ";".join(errors), "failedGates": []} guards = verdict.get("guards") if isinstance(verdict.get("guards"), dict) else {} failed = sorted(name for name, g in guards.items() if isinstance(g, dict) and g.get("pass") is False) passed = verdict.get("pass") if passed is True and failed: # pass=True 却存在 pass=False 的门：harness 自报与逐门证据矛盾 → 仪器异常，不信任一自报。 return {"state": "tester_error", "reason": f"verdict.pass=True 与 guards 矛盾（未过门:{failed}）", "failedGates": failed} if passed is True: return {"state": "pass", "reason": "九门机械预筛全绿", "failedGates": []} if passed is False: return {"state": "reject", "reason": f"九门机械预筛未过（{','.join(failed) or 'pass=False'}）", "failedGates": failed} return {"state": "tester_error", "reason": "verdict.pass 缺失或非布尔，机械预筛无结论", "failedGates": failed} def _eval_visual_floor(floor_judgment) -> dict: """② 视觉地板：消费独立模型玩法地板判定（judge_gameplay_floor / parse_floor_judgment 结构）。 degraded（评不出/无证据/调用失败）按 §5.2 归仪器异常 tester_error——不伪装 accept，也不伪装 gameplay reject； accepted=False 且非 degraded → reject（独立模型判定的 broken/hollow/off_brief）。 """ if not isinstance(floor_judgment, dict): return {"state": "tester_error", "reason": "视觉地板判定产物缺失", "rejectClasses": []} if floor_judgment.get("degraded") is True: return {"state": "tester_error", "reason": f"视觉地板降级（fail-closed）:{floor_judgment.get('reason') or '未知'}", "rejectClasses": []} accepted = floor_judgment.get("accepted") if accepted is True: return {"state": "pass", "reason": "视觉地板接受", "rejectClasses": []} if accepted is False: classes = [str(c) for c in (floor_judgment.get("rejectClasses") or [])] return {"state": "reject", "reason": f"视觉地板拒绝（{','.join(classes) or '地板未过'}）", "rejectClasses": classes} return {"state": "tester_error", "reason": "视觉地板判定 accepted 字段缺失或非布尔", "rejectClasses": []} def _eval_playtest_v3(v3_payload) -> dict: """③ playtest/3 真玩：消费 run_acceptance_v3 封存 payload 的 outcome/decision/finalPostguard。 不复算子门内部逻辑，但做零信任自洽复核：accepted=True 必须与 outcome=accept 且 finalPostguard.pass=True 同真，任一自相矛盾 → tester_error（封存产物不一致 = 仪器异常）。shadow 模式 accepted 仅供校准， publishFrozen 原样带出，发布判定留给 is_v3_publishable。 """ if not isinstance(v3_payload, dict): return {"state": "tester_error", "reason": "playtest/3 封存产物缺失", "outcome": None} outcome = v3_payload.get("outcome") decision = v3_payload.get("decision") if isinstance(v3_payload.get("decision"), dict) else {} final_guard = v3_payload.get("finalPostguard") if isinstance(v3_payload.get("finalPostguard"), dict) else {} compatibility = v3_payload.get("compatibility") if isinstance(v3_payload.get("compatibility"), dict) else {} detail = { "outcome": outcome, "accepted": decision.get("accepted"), "acceptanceMode": v3_payload.get("acceptanceMode"), "publishFrozen": decision.get("publishFrozen"), "rescuedByRoll": decision.get("rescuedByRoll"), "failure": decision.get("failure"), } if outcome not in _OUTCOMES: return {"state": "tester_error", "reason": f"playtest/3 outcome 未知:{outcome!r}", **detail} # 零信任自洽复核：三个写权字段必须同真，矛盾即封存产物异常。 if decision.get("accepted") is True and (outcome != "accept" or final_guard.get("pass") is not True): return {"state": "tester_error", "reason": f"decision.accepted=True 与 outcome={outcome}/finalPostguard.pass=" f"{final_guard.get('pass')} 矛盾", **detail} if outcome == "accept": if decision.get("accepted") is not True or final_guard.get("pass") is not True: return {"state": "tester_error", "reason": "outcome=accept 但 decision/finalPostguard 未同时确认", **detail} return {"state": "pass", "reason": "playtest/3 真玩接受（硬证完整 + 双 Judge 共识 + finalPostguard 全绿）", "compatibilityAccepted": compatibility.get("accepted"), **detail} if outcome == "reject": return {"state": "reject", "reason": "playtest/3 真玩拒绝（硬证已证可复现产物缺陷）", **detail} if outcome == "inconclusive": return {"state": "inconclusive", "reason": "playtest/3 证据不足或矛盾", **detail} return {"state": "tester_error", "reason": "playtest/3 仪器异常（Actor/runner/Judge/环境）", **detail} def _eval_prompt_eval(prompt_eval_record) -> dict: """④ prompt 四门：消费 eval_gate.py 真模型闸台账记录（单条 dict 或 Actor/Judge 多条 list）。 每条记录必须 infrastructureComplete 且 allGreen（= gate1 schema ∧ gate2 成功率 ∧ gate3 回归 ∧ gate4 成本延迟 ∧ gate5 稳定 全绿）。校准未过 = Actor/Judge 判读不可信，按 §5.2「Judge 异常 100% tester_error」归仪器异常， 绝不放行也绝不归 gameplay。 """ if prompt_eval_record is None: return {"state": "tester_error", "reason": "prompt 四门台账记录缺失（校准闸未跑）", "failedRecords": []} records = prompt_eval_record if isinstance(prompt_eval_record, list) else [prompt_eval_record] if not records: return {"state": "tester_error", "reason": "prompt 四门台账记录为空", "failedRecords": []} failed = [] for i, record in enumerate(records): if not isinstance(record, dict): failed.append({"index": i, "promptId": None, "reason": "记录不是对象"}) continue prompt_id = record.get("promptId") or f"#{i}" if record.get("infrastructureComplete") is not True: failed.append({"index": i, "promptId": prompt_id, "reason": "基础设施不完整（图像/审计/请求哈希不可信）"}) continue # allGreen 缺失时退回逐门 AND（gate1..gate5），两者皆无 → fail-closed。 all_green = record.get("allGreen") if all_green is None: gate_keys = ("gate1_schema", "gate2_success", "gate3_regression", "gate4_cost_latency", "gate5_stability") gates = {key: record.get(key) for key in gate_keys} if any(value is None for value in gates.values()): failed.append({"index": i, "promptId": prompt_id, "reason": "缺 allGreen 且逐门结果不全"}) continue all_green = all(value is True for value in gates.values()) if all_green is not True: red = [key for key in ("gate1_schema", "gate2_success", "gate3_regression", "gate4_cost_latency", "gate5_stability") if record.get(key) is False] failed.append({"index": i, "promptId": prompt_id, "reason": f"闸门红:{','.join(red) or 'allGreen=False'}"}) if failed: return {"state": "tester_error", "reason": f"prompt 四门未全绿（{len(failed)}/{len(records)} 条）", "failedRecords": failed} return {"state": "pass", "reason": f"prompt 四门全绿（{len(records)} 条记录）", "failedRecords": []} def _eval_schema_semantics(v3_payload, schema_validator) -> dict: """⑤ schema 语义：对 playtest/3 封存产物跑 canonical validate.py（schema + 语义）。 经 cheap_verify.validate_acceptance_v3_payload 调权威校验器（运行时与测试共用同一规则）； 任何错误按 §5.2「schema 异常 100% tester_error」归仪器异常。 """ if not isinstance(v3_payload, dict): return {"state": "tester_error", "reason": "无 playtest/3 产物可校验", "errors": []} try: errors = [str(e) for e in (schema_validator(v3_payload) or [])] except Exception as exc: # noqa: BLE001 —— 校验器自身异常必须 fail-closed 归仪器异常 return {"state": "tester_error", "reason": f"canonical 校验器异常:{type(exc).__name__}: {exc}", "errors": []} if errors: return {"state": "tester_error", "reason": f"playtest/3 schema/语义校验失败（{len(errors)} 项）", "errors": errors[:8]} return {"state": "pass", "reason": "canonical schema + 语义校验通过", "errors": []} def _eval_gold_reconcile(consumptions, consumer_ref, gold_registry, gold_reconciler) -> dict: """⑥ W-GOLD-LIVE 消费对账：对 identity 声明的参照资产消费跑六项闸（存在/激活/role/consumerRef/版本/缺维度）。 未声明消费 ≡ 旧路径，vacuous 放行（与 _v3_check_declared_reference_assets 语义一致）； 声明消费而对账失败 → reject（设计语义：声明消费无 active 匹配 = verified reject）。 """ items = list(consumptions or []) if not items: return {"state": "pass", "reason": "未声明参照资产消费（旧路径，对账 vacuous 通过）", "vacuous": True, "consumed": [], "errors": []} try: reconciled = gold_reconciler(items, consumer_ref=consumer_ref, registry=gold_registry) except Exception as exc: # noqa: BLE001 —— 注册表不可读按 fail-closed 拒绝消费 return {"state": "reject", "reason": f"参照资产消费对账失败:{type(exc).__name__}: {exc}", "vacuous": False, "consumed": [], "errors": [str(exc)]} if reconciled.get("ok"): return {"state": "pass", "reason": f"参照资产消费对账通过（{len(reconciled.get('consumed') or [])} 条）", "vacuous": False, "consumed": reconciled.get("consumed") or [], "errors": []} return {"state": "reject", "reason": "参照资产消费对账拒绝:" + ";".join((reconciled.get("errors") or [])[:4]), "vacuous": False, "consumed": [], "errors": reconciled.get("errors") or []} # ────────────────────────── 单一 decision 组合（确定性，零模型）────────────────────────── def _combine(gates: dict) -> dict: """按严重度优先级合并六个子门状态为单一 full_gate decision（tester_error > reject > inconclusive > accept）。""" states = [g["state"] for g in gates.values()] worst = min(states, key=lambda s: _SEVERITY.get(s, -1)) if worst not in _SEVERITY: # 出现未知状态本身即仪器异常，绝不放行。 worst = "tester_error" outcome = "accept" if worst == "pass" else worst reasons = [f"[{name}] {g['reason']}" for name, g in gates.items() if g["state"] != "pass"] return {"pass": worst == "pass", "outcome": outcome, "reasons": reasons} def run_full_gate(*, verdict=None, floor_judgment=None, v3_payload=None, prompt_eval_record=None, reference_consumptions=(), consumer_ref=None, gold_registry=None, schema_validator=None, gold_reconciler=None) -> dict: """full_gate 纯函数核心：消费六个子门既有产物，产出单一 decision（确定性、零模型、零 I/O）。 Args: verdict: Node 九门产物 {ok?, pass, guards, errors?}；None=未跑 → fail-closed tester_error。 floor_judgment: judge_gameplay_floor 裁决 {accepted, degraded, rejectClasses, reason?}。 v3_payload: run_acceptance_v3 封存 payload（outcome/decision/finalPostguard/compatibility）。 prompt_eval_record: eval_gate 台账记录 dict 或 list（Actor + Judge A/B 多条）。 reference_consumptions: identity 声明的参照资产消费（recordId 字符串或对象列表）；空=vacuous 放行。 consumer_ref / gold_registry: 金标对账的消费方身份与注册表（None → 默认注册表）。 schema_validator / gold_reconciler: 可注入的子门函数（默认绑 cheap_verify 真函数；单测注入罐头）。 Returns: {pass, outcome, publishable, gates: {六门各自 state/reason/细节}, reasons: [未过原因]} """ validator = schema_validator if schema_validator is not None else V.validate_acceptance_v3_payload reconciler = gold_reconciler if gold_reconciler is not None else V.reconcile_v3_reference_asset_consumption gates = { "mechanicalNine": _eval_mechanical_nine(verdict), "visualFloor": _eval_visual_floor(floor_judgment), "playtestV3": _eval_playtest_v3(v3_payload), "promptEval": _eval_prompt_eval(prompt_eval_record), "schemaSemantics": _eval_schema_semantics(v3_payload, validator), "goldReconcile": _eval_gold_reconcile(reference_consumptions, consumer_ref, gold_registry, reconciler), } decision = _combine(gates) # 发布谓词只在全过时才有意义：active v3 + 未冻结 + 兼容层同真（is_v3_publishable 的字段口径，消费不重写）。 publishable = False if decision["pass"] and isinstance(v3_payload, dict): decision_node = v3_payload.get("decision") if isinstance(v3_payload.get("decision"), dict) else {} compatibility = v3_payload.get("compatibility") if isinstance(v3_payload.get("compatibility"), dict) else {} publishable = bool( v3_payload.get("acceptanceMode") == "v3" and decision_node.get("publishFrozen") is False and compatibility.get("accepted") is True and compatibility.get("ok") is True and compatibility.get("publishFrozen") is False) decision["publishable"] = publishable decision["gates"] = gates return decision async def run_full_gate_live(*, game_id: str, brief: str, verdict: dict, v3_request: dict, floor_model=None, floor_model_name: str = None, prompt_eval_record=None, reference_consumptions=(), consumer_ref=None, gold_registry=None) -> dict: """真编排入口（mini-desktop 真跑用）：调视觉地板判定 + run_acceptance_v3 真玩，再交纯函数核心出单一 decision。 本函数是唯一会触达真模型的编排层；九门 verdict 与 prompt 四门台账由上游生成/CI 产物传入（消费既有产物）。 本地单测只覆盖 run_full_gate 纯核心，不跑本函数。 """ # ② 视觉地板：独立模型看真玩截图/日志（fail-closed 语义由 judge_gameplay_floor 内部保证）。 floor_judgment = await V.judge_gameplay_floor( game_id, brief=brief, model=floor_model, model_name=floor_model_name) # ③ playtest/3 真玩：唯一 v3 编排入口（含四门投影/双 Judge/二掷/merge/finalPostguard/幂等封存）。 v3_payload = await V.run_acceptance_v3(v3_request) return run_full_gate(verdict=verdict, floor_judgment=floor_judgment, v3_payload=v3_payload, prompt_eval_record=prompt_eval_record, reference_consumptions=reference_consumptions, consumer_ref=consumer_ref, gold_registry=gold_registry)