87777975c7
actor/judge prompt eval 闭合: actor 3.0.5 正文强化+baseline(7轮); judge 3.0.14 正文强化+anyTerms同义词补全+baseline(judge-a 13轮/judge-b 33轮); registry/checker 版本对齐+顺序锚点 fail-closed 双保险 W-GOLD-LIVE live_prompt: play-loop 契约开口(request/3+provenance/3+ReferenceAssetRecord/1+Registry/1+迁移清单15条); cheap_verify 切v3+消费对账六闸(只消费active)+生成prompt注入接线(0 active不注入); Node runner playtest-v3.cdp.cjs 升/3+绊线恢复; 真模型回归验向后兼容 full_gate+三批基线: full_gate.py 集成runner串六子门+降级+22测; baseline_gates.py fresh25阈值+historical11预期表(2 needs_human交创始人定标)+shadow20框架+38测 注: registry cheap-system 登记 1.8.0→1.8.1 同步此前在途 cheap-system.md frontmatter 升版(修 pre-commit 版本漂移); cheap_verify/cheap_studio/validate/test_acceptance_v3/test_cheap_service_driver 为在途M含本会话叠加+此前W-AXIS在途(同一线无法hunk分离); 3红测试根因与raw审计未入本提交
285 lines
19 KiB
Python
285 lines
19 KiB
Python
"""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)
|