cbfd4d871b
固化 Match-3 生产者、视觉、音频与双 Judge 证据闭包。 将《山海行纪》r1.1 绑定新的不可变 release，并以生产预检现场核验 bundle、Registry/2 和 25 项 Writer 快照。 同步地图1平衡锁值、跨游戏回归修复、验收契约与 SoT 证据。
4176 lines
245 KiB
Python
4176 lines
245 KiB
Python
"""
|
||
cheap_verify.py — 便宜档「丰富度」LLM 验证 agent（非阻塞·只报告）。
|
||
|
||
架构红线（创始人多轮强调，绝不可破）：**项目代码只做机械确定的事**；玩法/美术/音乐的「丰富」——
|
||
无论生成还是校验——都需大模型能力，**绝不写成代码校验**（code-presence/静态扫描/正则/断言一律禁）。
|
||
故本模块只做两件 100% 机械确定的事，「丰富不丰富」的判断本身完全交 LLM judge：
|
||
① 读产物 L3 源码（确定性文件 I/O）+ 拼 judge prompt（确定性字符串）；
|
||
② 解析 LLM 返回的 JSON 成结构化评分（确定性解析 + degraded 兜底）。
|
||
本模块**不含任何丰富度的代码断言**——11 条丰富度清单只是【喂给 LLM 的评分尺】，命中与否全由 judge 读码后裁；
|
||
评分按层标 L2/L3/L4 聚合成三分组小计（L2 内容丰富 / L3 留存结构 / L4 传播钩子，不设单一总分），规范见质量模型 SoT §3.3 计分制。
|
||
（注：sim-business-game-design.md §10 写了「4 条可机检 code-presence」子集，那与本红线冲突——本模块刻意不照它写代码校验。）
|
||
|
||
非阻塞铁律：LLM 调失败/超时/解析失败 → 返回 {score:None, degraded:True, reason:...}，
|
||
**绝不抛、绝不阻断生成、绝不进 verdict / 不改达标判定**。接线方（cheap_studio）再包一层 try/except 双保险。
|
||
|
||
被 cheap_studio.run_studio 收口段调用（九门 play 之后、写 run-summary 之前），结果 additive 写 summary["richness"]。
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import fcntl
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import os
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
import secrets
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
||
from pathlib import Path, PurePosixPath
|
||
|
||
import artifact_snapshot
|
||
import cheap_run # 仅 stdlib 依赖（读产物 src 文件）；不 import agentscope/_bootstrap，故纯函数可在无 agentscope 环境单测。
|
||
|
||
# ── 通用底座 v2 = 11 条丰富度清单（judge 的评分 rubric）──
|
||
# 这是【喂给 LLM judge 的打分尺】，不是代码校验：丰富不丰富全由 judge 读源码后逐条裁。
|
||
# 规范来源 = 质量模型 SoT《游戏质量与爆火能力·设计》§3.3 计分制（每条 0/1 + 层标 L2/L3/L4，产三分组小计、不设单一总分）。
|
||
# v2 = 现行 8 条（与 .agents/skills/sim-business-game-design.md §10 同源）+ 新增 3 条（首 3 分钟脚本 / 炫耀时刻 / 同款钩子）。
|
||
# 每项 = (标准名, 层标, 含义)；judge 按标准名逐条回命中与否 + 理由，命中数按层聚合成三分组小计。
|
||
# ★锚定纪律（§3.3 第 3 条）：原 8 条判据文本与顺序一字不改（评分尺变更须金标复验、原 8 条判定漂移 ≤±1），
|
||
# 新增 3 条追加于末尾；层标只是聚合归属，不改原条目的判据本身。
|
||
# 分层含义：L2 内容丰富（有料耐玩）/ L3 留存结构（想再玩的结构前提）/ L4 传播钩子（想传出去的结构前提）。
|
||
RICHNESS_CHECKLIST = [
|
||
# 原 8 条（文本/顺序锁定，仅补层标）
|
||
("即时反馈", "L2", "每个主要操作有「飘字 +N + 音效 + 粒子」即时回报"),
|
||
("可见成长", "L2", "有数字/规模肉眼可见地变大（分数飙升/店铺扩张/等级上涨）"),
|
||
("下一个解锁", "L3", "任意时刻玩家眼前都有「再攒一点就解锁 X」的钩子"),
|
||
("30秒爽点", "L2", "开局 30 秒内有第一次升级/解锁"),
|
||
("数值滚雪球", "L2", "成长有「越来越快」的暴富段，而非平淡线性"),
|
||
("情感锚", "L2", "有萌角色/拥有物让玩家「想养大它」"),
|
||
("放置回归", "L3", "离线回来有惊喜（离线收益弹窗等）"),
|
||
("音反馈", "L2", "收益/升级/解锁有声音（哪怕程序化）"),
|
||
# 新增 3 条（v2 草案，随首个品类件落地验证；判据文本源自 §3.3 第 1 条）
|
||
("首3分钟脚本", "L3", "首 3 分钟脚本成立：0–10s 零阅读上手 / 10–60s 首次升级 / 1–3min 露出 2–3 个后续锁"),
|
||
("炫耀时刻", "L4", "有可截图分享的结算/成就画面（打完有结算展示场景，不是直接黑屏）"),
|
||
("同款钩子", "L4", "有清晰的品类原型 + 主题标签元数据，可供 remix 预填做同款"),
|
||
# 第 12 条（2026-07-03 增补，质量 SoT §4 规范一「通用底座 12 条口径」；创始人授权默认项拍板采纳）：
|
||
# 源自便宜档主 prompt 自检第 ⑨ 条（策划知识包 v2 否决项判定句式），升为评分尺条目。
|
||
# 锚定纪律：追加于末尾、前 11 条一字不动；L2 分母随之 6→7，档位观测线百分比不变、分母随升；
|
||
# 落地与金标复验同批（复验漂移超线按规范三复采样定性再判回退）。
|
||
("核心操作非无脑", "L2", "每次主操作含真实决策或技巧含量、判错有真代价（如补错货压库存），不是点哪都加分/点了就自动结算"),
|
||
]
|
||
|
||
_MAX = len(RICHNESS_CHECKLIST) # = 12（2026-07-03 ⑨升第 12 条；此前 v2 = 11）
|
||
_LAYERS = ("L2", "L3", "L4") # 三分组固定顺序（消费面按分组小计各自取阈，见 §3.6 档位表）
|
||
|
||
# ── 品类扩展 rubric（W-GENRE 品类件④；质量模型 SoT §4 规范二）──
|
||
# 数据源唯一 = fixtures/genre-rubrics/<genre>.json（items 键，条目 {name,layer,meaning,positive,negative}），
|
||
# 经 load_genre_checklist 加载。2026-07-02 rubric 挂点归一：原内联 GENRE_CHECKLISTS 字典退役，
|
||
# TRPG 条目一字不动迁入 trpg.json——品类条目自此只有 fixture 一路，不再留内联第二源。
|
||
# 每品类另加 ≥4 条品类特有维度：同一字段形状 (标准名, 层标, 含义)、同样标层，
|
||
# **分母独立小计**（与通用底座分开、不混合平均——评分输出落 richness.genre.groups，通用 groups 不变）。
|
||
# 这仍是【喂给 LLM judge 的评分尺】数据、不是代码校验（红线：不进九门/不进 verdict/不进脚手架）；
|
||
# 设计侧同源对齐 = .agents/skills/<genre>-game-design.md §10 品类自检。
|
||
# 锚定纪律（规范三）：品类条目文本变更须金标复验（金标正例 + 薄反例，分组小计漂移 ≤±1 即回退）。
|
||
_GENRE_RUBRICS_DIR = Path(__file__).resolve().parent / "fixtures" / "genre-rubrics"
|
||
|
||
# 品类脚手架目录名 → 品类 key（run_studio 据 scaffold_template 透传 genre；None/未登记 → 不评品类扩展）。
|
||
# 非遗品类键收敛 = "heritage"（与 rubric 文件名/散文 skill heritage-game-design.md 同名）：
|
||
# 模板目录 _template-feiyi 是资产目录名、金标 spec 的 _genre="heritage-craft" 是金标名（driver 承重不动），
|
||
# 三者同一品类——映射依据详见 fixtures/genre-rubrics/heritage.json 的 _note。
|
||
GENRE_BY_TEMPLATE = {
|
||
"_template-trpg": "trpg",
|
||
"_template-puzzle": "puzzle",
|
||
"_template-feiyi": "heritage",
|
||
# 2026-07-04 数据面补全(策划知识包 v2):剧情/经营两品类 fixture 落地补线,
|
||
# 条目源见各 fixture _note(narrative=skill §10 表逐字迁入;sim-business=质量 SoT §4 规范二名目展开)。
|
||
"_template-story": "narrative",
|
||
"_template-shop": "sim-business",
|
||
}
|
||
|
||
|
||
def load_genre_checklist(genre) -> list:
|
||
"""读品类 rubric fixture（品类条目唯一数据源）→ [(标准名, 层标, 判据)*]；genre 空/文件缺/坏条目 → []（绝不抛）。
|
||
|
||
条目形状随 fixture（name/layer/meaning，正反例 positive/negative 只作金标锚材料、不进 judge prompt
|
||
——prompt 里逐条只给一句可判定判据，与通用底座条目同形）。
|
||
"""
|
||
if not genre or not isinstance(genre, str):
|
||
return []
|
||
try:
|
||
p = _GENRE_RUBRICS_DIR / (genre + ".json")
|
||
if not p.is_file():
|
||
return []
|
||
data = json.loads(p.read_text(encoding="utf-8"))
|
||
out = []
|
||
for it in (data.get("items") or []):
|
||
if not isinstance(it, dict):
|
||
continue
|
||
name, layer, meaning = it.get("name"), it.get("layer"), it.get("meaning")
|
||
if isinstance(name, str) and name and layer in _LAYERS and isinstance(meaning, str) and meaning:
|
||
out.append((name, layer, meaning))
|
||
return out
|
||
except Exception: # noqa: BLE001 fixture 坏 → 扩展为空，评分退化为纯通用底座，绝不阻断
|
||
return []
|
||
|
||
|
||
def _layer_maxes(checklist=RICHNESS_CHECKLIST) -> dict:
|
||
"""各层条目数（分母）。通用底座 12 条口径 = L2:7 / L3:3 / L4:2（2026-07-03 第 12 条入 L2）。"""
|
||
m = {ly: 0 for ly in _LAYERS}
|
||
for _name, layer, _meaning in checklist:
|
||
m[layer] = m.get(layer, 0) + 1
|
||
return m
|
||
|
||
|
||
def _group_subtotals(hits: list, checklist=RICHNESS_CHECKLIST) -> dict:
|
||
"""把逐条命中按层标聚合成三分组小计 {L2:{score,max}, L3:..., L4:...}。
|
||
|
||
这是 §3.3 计分制的落库主形状——三层各自对得上自己的消费点（品类件验收/批次观测/回流分析），
|
||
**不做单一总分平均**（单一总分诱导 Goodhart，见 §3.3 第 1 条）。
|
||
"""
|
||
maxes = _layer_maxes(checklist)
|
||
scores = {ly: 0 for ly in _LAYERS}
|
||
for h in hits:
|
||
ly = h.get("layer")
|
||
if h.get("hit") and ly in scores:
|
||
scores[ly] += 1
|
||
return {ly: {"score": scores[ly], "max": maxes.get(ly, 0)} for ly in _LAYERS}
|
||
|
||
# 只读 L3 游戏本体（玩法/画面/数值/资产）；跳过 L1 plumbing（host-config/game.js/index.html——
|
||
# 固定脚手架、与「游戏丰富不丰富」无关，喂给 judge 只会稀释信号）。
|
||
_L3_SOURCE_FILES = ("game-logic.js", "core.js", "render.js", "balance.js", "assets.js")
|
||
_MAX_SRC_BYTES = 24000 # 喂 judge 的源码总量上限（控 token / 成本；超限按字节截断）。
|
||
_JUDGE_MAX_TOKENS = 4000 # judge 输出小（11 条短理由 + 一句点评），4000 足够、M3 无 thinking。
|
||
|
||
# judge 的 system / user 框定——只评分、不改码、要看源码真实现、输出严格 JSON。
|
||
# 计数用 len(RICHNESS_CHECKLIST) 动态拼(2026-07-03 ⑨升第 12 条时改:此前「11 条」写死在文案里,
|
||
# 清单变更即漂;动态取数后条数永远与评分尺同源。评分尺变更本身仍受锚定纪律约束=金标复验)。
|
||
_JUDGE_SYSTEM = (
|
||
"你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
|
||
f"只评判它「作为一款游戏够不够丰富、好玩、耐玩、想传出去」，逐条给出 {len(RICHNESS_CHECKLIST)} 条丰富度清单的命中与否 + 一句中文理由，"
|
||
"最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。"
|
||
f"这 {len(RICHNESS_CHECKLIST)} 条分三层：L2 内容丰富（有料耐玩）、L3 留存结构（想再玩的结构前提）、L4 传播钩子（想让别人看/做同款的前提）；"
|
||
"评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈、音效、结算画面与品类元数据，别被空壳或注释骗。"
|
||
)
|
||
|
||
# 品类扩展评分时的 system 变体（W-GENRE 件④）：仅把量词放宽到「通用 N 条 + 品类扩展若干条」，
|
||
# 其余判据措辞一字不动（锚定纪律：无品类路走上面原版 _JUDGE_SYSTEM）。
|
||
_JUDGE_SYSTEM_GENRE = (
|
||
"你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
|
||
f"只评判它「作为一款游戏够不够丰富、好玩、耐玩、想传出去」，逐条给出丰富度清单（通用 {len(RICHNESS_CHECKLIST)} 条 + 品类扩展若干条）的命中与否 + 一句中文理由，"
|
||
"最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。"
|
||
"条目分三层：L2 内容丰富（有料耐玩）、L3 留存结构（想再玩的结构前提）、L4 传播钩子（想让别人看/做同款的前提）；"
|
||
"评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈、音效、结算画面与品类元数据，别被空壳或注释骗。"
|
||
)
|
||
|
||
|
||
def _degraded(reason: str) -> dict:
|
||
"""降级结果（非阻塞铁律）：score=None + degraded=True；score 为 None 表示「这次没评出来」，绝不参与达标。
|
||
|
||
groups 同置 None：三分组小计的消费面（品类件验收/回流分析）遇 None 即「本次未评出」，与 score=None 同义。
|
||
"""
|
||
return {"score": None, "max": _MAX, "hits": [], "groups": None, "notes": None, "degraded": True, "reason": reason}
|
||
|
||
|
||
def _coerce_bool(v) -> bool:
|
||
"""把 judge 可能给的多形态命中值归一为 bool（true/1/"是"/"命中" 等 → True；其余 → False）。"""
|
||
if isinstance(v, bool):
|
||
return v
|
||
if isinstance(v, (int, float)):
|
||
return v > 0
|
||
if isinstance(v, str):
|
||
return v.strip().lower() in ("true", "1", "yes", "y", "是", "命中", "hit", "对", "✓", "√")
|
||
return False
|
||
|
||
|
||
def parse_judge_output(text, *, checklist=RICHNESS_CHECKLIST, genre_checklist=None, genre_key=None) -> dict:
|
||
"""纯函数：把 judge 的 LLM 文本输出解析成结构化丰富度评分（含 degraded 兜底）。可单测、零网络、零 agentscope。
|
||
|
||
成功 → {
|
||
score:int(11 条命中总数 0..11), # ★向后兼容字段：仅供 bake_off.richness_dist 读；非「质量总分」。
|
||
max:11, # 设计不设单一总分（诱导 Goodhart），主消费形状 = groups 三分组小计。
|
||
hits:[{name,layer,hit,why}*11], # 逐条：加 layer 层标（L2/L3/L4）
|
||
groups:{L2:{score,max}, L3:..., L4:...}, # ★三分组小计（§3.3 计分制主形状）
|
||
notes:str, degraded:False}
|
||
传 genre_checklist（品类扩展条目，W-GENRE 件④）时 additive 加：
|
||
genre:{key, hits:[…*N], score, max:N, groups:{…}} # ★品类分母独立小计（§4 规范二：不与通用混合平均）
|
||
解析失败（空/非 JSON 对象/缺 checks 数组）→ _degraded(...)（score=None, groups=None, degraded=True）。
|
||
|
||
对齐策略：优先按 name 对齐 judge 的 checks；name 对不上的按位置兜底（容忍模型改名/英文名）——
|
||
checks 数组约定 = 通用条目在前、品类条目紧随其后，位置兜底按段内偏移算。
|
||
layer 层标取自 checklist（不依赖 LLM 回传），故分组小计对模型改名鲁棒。
|
||
本函数**不含任何丰富度的代码判断**——只搬运 judge 的逐条裁决并按层聚合命中数。
|
||
"""
|
||
if not isinstance(text, str) or not text.strip():
|
||
return _degraded("judge 输出为空")
|
||
try:
|
||
# json_repair 容忍 markdown 围栏 / 前后赘语 / 尾逗号等 LLM 常见脏输出。
|
||
import json_repair
|
||
data = json_repair.loads(text)
|
||
except Exception as e: # noqa: BLE001 解析层任何异常都降级，绝不抛
|
||
return _degraded(f"judge 输出解析失败：{type(e).__name__}")
|
||
if not isinstance(data, dict):
|
||
return _degraded("judge 输出不是 JSON 对象")
|
||
checks_raw = data.get("checks")
|
||
if not isinstance(checks_raw, list) or not checks_raw:
|
||
return _degraded("judge 输出缺 checks 数组")
|
||
|
||
by_name = {}
|
||
for c in checks_raw:
|
||
if isinstance(c, dict):
|
||
nm = str(c.get("name", "")).strip()
|
||
if nm:
|
||
by_name[nm] = c
|
||
|
||
def _align(items, offset):
|
||
"""按 name 对齐（对不上按位置兜底，位置 = offset + 段内下标）→ [{name,layer,hit,why}]。"""
|
||
out = []
|
||
for i, (name, layer, _meaning) in enumerate(items):
|
||
c = by_name.get(name)
|
||
pos = offset + i
|
||
if c is None and pos < len(checks_raw) and isinstance(checks_raw[pos], dict):
|
||
c = checks_raw[pos] # name 对不上 → 位置兜底
|
||
hit = _coerce_bool(c.get("hit")) if isinstance(c, dict) else False
|
||
why = (str(c.get("why", "")).strip()[:200]) if isinstance(c, dict) else ""
|
||
out.append({"name": name, "layer": layer, "hit": hit, "why": why}) # layer 取自 checklist（权威）
|
||
return out
|
||
|
||
hits = _align(checklist, 0)
|
||
score = sum(1 for h in hits if h["hit"])
|
||
notes = str(data.get("notes", "")).strip()[:500]
|
||
result = {"score": score, "max": len(checklist), "hits": hits,
|
||
"groups": _group_subtotals(hits, checklist), "notes": notes, "degraded": False}
|
||
|
||
# 品类扩展段（W-GENRE 件④）：checks 数组的通用段之后是品类条目；**分母独立小计**（不并进通用 groups/score）。
|
||
if genre_checklist:
|
||
ghits = _align(genre_checklist, len(checklist))
|
||
result["genre"] = {
|
||
"key": genre_key,
|
||
"hits": ghits,
|
||
"score": sum(1 for h in ghits if h["hit"]),
|
||
"max": len(genre_checklist),
|
||
"groups": _group_subtotals(ghits, genre_checklist),
|
||
}
|
||
return result
|
||
|
||
|
||
def _collect_sources(game_id: str, *, max_bytes: int = _MAX_SRC_BYTES) -> str:
|
||
"""读产物 L3 源码（game-logic/core/render/balance/assets）拼成带文件头的一段文本，总量截断到 max_bytes。
|
||
|
||
只读 L3 游戏本体；读不到的文件跳过；全空 → 返回空串（上层据此降级、跳过 LLM、不白烧钱）。
|
||
"""
|
||
src_dir = cheap_run.game_dir(game_id) / "src"
|
||
parts = []
|
||
total = 0
|
||
for name in _L3_SOURCE_FILES:
|
||
p = src_dir / name
|
||
try:
|
||
if not p.is_file():
|
||
continue
|
||
txt = p.read_text(encoding="utf-8", errors="ignore")
|
||
except OSError:
|
||
continue # 单文件读失败不致命，跳过即可
|
||
chunk = f"\n// ===== {name} =====\n{txt}\n"
|
||
parts.append(chunk)
|
||
total += len(chunk.encode("utf-8"))
|
||
if total >= max_bytes:
|
||
break
|
||
joined = "".join(parts)
|
||
enc = joined.encode("utf-8")
|
||
if len(enc) > max_bytes: # 字节级截断兜底（防单个大文件超限撑爆 token）
|
||
joined = enc[:max_bytes].decode("utf-8", "ignore") + "\n// [源码已截断]"
|
||
return joined
|
||
|
||
|
||
def _build_judge_user(src_text: str, brief: str = "", genre_key: str = None, genre_checklist=None) -> str:
|
||
"""拼 judge 的 user 消息：brief（可选）+ 11 条清单（带层标）[+ 品类扩展清单] + 产物源码 + 严格 JSON 输出契约。
|
||
|
||
纯确定性字符串拼接。**不传品类时输出与既有版本逐字节一致**（金标复验基线稳，规范三锚定纪律）；
|
||
传品类时通用清单后 additive 追加品类段、输出契约条数 = 11+N（checks 通用在前、品类紧随）。
|
||
"""
|
||
checklist_lines = "\n".join(
|
||
f"{i + 1}. [{layer}] {name}：{meaning}" for i, (name, layer, meaning) in enumerate(RICHNESS_CHECKLIST)
|
||
)
|
||
brief_block = f"这款游戏的 brief（玩家想要的）：{brief}\n\n" if brief else ""
|
||
n_genre = len(genre_checklist) if genre_checklist else 0
|
||
if n_genre:
|
||
genre_lines = "\n".join(
|
||
f"{len(RICHNESS_CHECKLIST) + i + 1}. [{layer}] {name}：{meaning}"
|
||
for i, (name, layer, meaning) in enumerate(genre_checklist)
|
||
)
|
||
genre_block = (
|
||
f"\n另有 {n_genre} 条【{genre_key} 品类扩展清单】（分母独立、不与上面 {len(RICHNESS_CHECKLIST)} 条混合，同样逐条独立判命中）：\n"
|
||
f"{genre_lines}\n"
|
||
)
|
||
total_note = f"共 {len(RICHNESS_CHECKLIST) + n_genre} 条（通用 {len(RICHNESS_CHECKLIST)} 条在前、品类 {n_genre} 条紧随其后）"
|
||
name_note = "name 用上面各条的中文名"
|
||
else:
|
||
genre_block = ""
|
||
total_note = f"共 {len(RICHNESS_CHECKLIST)} 条"
|
||
name_note = f"name 用上面 {len(RICHNESS_CHECKLIST)} 条的中文名"
|
||
return (
|
||
f"{brief_block}"
|
||
f"下面是这款便宜档小游戏的 L3 源码。请逐条评判它是否命中这 {len(RICHNESS_CHECKLIST)} 条丰富度清单（每条前的 [L2]/[L3]/[L4] 是分层标注——"
|
||
"L2 内容丰富、L3 留存结构、L4 传播钩子——仅用于分组，不改变你对每条的独立判命中），"
|
||
"据**源码里真实实现了的玩法/数值/音效/解锁/结算画面/品类元数据**判断（别被注释或空壳骗：比如只 import 了 audioMusic "
|
||
"但收益处没真调 playSfx，则「音反馈」不算命中）：\n\n"
|
||
f"{checklist_lines}\n"
|
||
f"{genre_block}\n"
|
||
"=== 源码开始 ===\n"
|
||
f"{src_text}\n"
|
||
"=== 源码结束 ===\n\n"
|
||
"严格只输出以下 JSON（不要任何额外文字、不要 markdown 围栏）：\n"
|
||
f'{{"checks":[{{"name":"即时反馈","hit":true,"why":"一句中文理由"}}, … {total_note}，'
|
||
f'{name_note}、hit 为 true/false], "notes":"整体一句话点评"}}'
|
||
)
|
||
|
||
|
||
def _extract_text(resp) -> str:
|
||
"""从 ChatResponse 抽纯文本：content 是 TextBlock/ToolCall... 序列，只取 text 块（兼容 pydantic 块与 dict 块）。"""
|
||
content = getattr(resp, "content", None)
|
||
if content is None and isinstance(resp, dict):
|
||
content = resp.get("content")
|
||
if isinstance(content, str):
|
||
return content
|
||
parts = []
|
||
for b in (content or []):
|
||
if isinstance(b, dict):
|
||
if b.get("type") == "text" and isinstance(b.get("text"), str):
|
||
parts.append(b["text"])
|
||
elif getattr(b, "type", None) == "text":
|
||
t = getattr(b, "text", None)
|
||
if isinstance(t, str):
|
||
parts.append(t)
|
||
return "".join(parts)
|
||
|
||
|
||
async def verify_richness(game_id: str, *, brief: str = "", model=None, sources=None,
|
||
timeout: float = 90.0, max_src_bytes: int = _MAX_SRC_BYTES,
|
||
genre: str = None) -> dict:
|
||
"""生成完成后跑一次 LLM 丰富度评分（非阻塞·只报告）。**绝不抛、绝不阻断、绝不进 verdict / 不改达标**。
|
||
|
||
Args:
|
||
game_id: 本 run 游戏 id（读 games/amgen-<id>/src/ 下 L3 源码）。
|
||
brief: 本局 brief（喂 judge 做上下文，提升「情感锚/解锁」等判断质量）；缺省空串。
|
||
model: 可选注入的 LLM 客户端（单测用 fake model 注入、零网络）；None → 用 _bootstrap.build_cheap_model
|
||
新建**独立** M3 实例（judge token 不污染生成成本台账 costRmb——richness 是非阻塞 add-on）。
|
||
sources: 可选直接注入的源码文本（单测用，绕过文件 I/O）；None → 从 game_id 收集。
|
||
timeout: judge LLM 调用超时秒数；超时 → degraded（绝不卡死收口）。
|
||
max_src_bytes: 喂 judge 的源码上限。
|
||
genre: 品类 key（W-GENRE 件④）；命中 fixtures/genre-rubrics/<genre>.json（唯一数据源）→
|
||
同一次评分 additive 追加品类扩展条目（分母独立小计落 result["genre"]）；
|
||
None/fixture 缺 → 行为与既有版本完全一致（prompt 逐字节不变）。
|
||
|
||
Returns:
|
||
成功 → parse_judge_output 的结构化评分（含 judgeTokens 观测，best-effort）；
|
||
任何失败（src 空/LLM 异常/超时/解析失败）→ _degraded(...)（score=None, degraded=True, reason）。
|
||
"""
|
||
try:
|
||
src_text = sources if sources is not None else _collect_sources(game_id, max_bytes=max_src_bytes)
|
||
if not src_text or not src_text.strip():
|
||
return _degraded("产物 src 为空/读不到，跳过 LLM 评分")
|
||
|
||
# 惰性 import：agentscope / _bootstrap 较重，且让上面的 parse_judge_output 等纯函数能在无 agentscope 环境单测。
|
||
from agentscope.message import SystemMsg, UserMsg
|
||
m = model
|
||
if m is None:
|
||
import _bootstrap # 代理旁路 + key 注入由 _bootstrap.build_cheap_model 内部处理
|
||
m = _bootstrap.build_cheap_model(max_tokens=_JUDGE_MAX_TOKENS)
|
||
|
||
# 品类扩展（唯一数据源 = fixtures/genre-rubrics/<genre>.json，经 load_genre_checklist；
|
||
# 没有 fixture → 静默按 None——评分尺没有它的条目，别让评分半路造尺）。
|
||
genre_checklist = (load_genre_checklist(genre) or None) if genre else None
|
||
genre_key = genre if genre_checklist else None
|
||
|
||
sys_msg = SystemMsg(name="system", content=_JUDGE_SYSTEM_GENRE if genre_checklist else _JUDGE_SYSTEM)
|
||
user_msg = UserMsg(name="user", content=_build_judge_user(src_text, brief,
|
||
genre_key=genre_key, genre_checklist=genre_checklist))
|
||
# 硬超时包裹真模型调用：judge 卡住绝不能拖死收口（非阻塞铁律）。
|
||
resp = await asyncio.wait_for(m([sys_msg, user_msg]), timeout=timeout)
|
||
result = parse_judge_output(_extract_text(resp), genre_checklist=genre_checklist, genre_key=genre_key)
|
||
|
||
# best-effort 附 judge token（外部 LLM 调用可观测；与生成成本台账隔离——judge 用独立 model 实例）。
|
||
try:
|
||
ti, to = m.usage_sum()
|
||
result["judgeTokens"] = {"in": ti, "out": to, "total": ti + to}
|
||
except Exception: # noqa: BLE001 观测字段失败不影响评分主体
|
||
pass
|
||
return result
|
||
except (asyncio.TimeoutError, TimeoutError):
|
||
return _degraded(f"judge LLM 超时（>{timeout}s）")
|
||
except Exception as e: # noqa: BLE001 非阻塞铁律：任何异常（网络/import/解析）都降级，绝不抛断生成主链
|
||
return _degraded(f"judge 异常：{type(e).__name__}: {e}")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 玩法地板判定(W-AXIS 波2 · 质量模型 SoT 裁定三的代码兑现)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# 与上面的丰富度评分是两物,别混:
|
||
# · 丰富度(verify_richness):评「有多丰富/多耐玩」的【程度】——非阻塞软信号,只观测、永不拒发(L2)。
|
||
# · 玩法地板(judge_gameplay_floor):裁「作为游戏成不成立」的【地板布尔】——阻断放行(L1 的判定半)。
|
||
# 裁定三钉死的分界:判定只裁三类【拒绝】的有无,不判程度高低:
|
||
# · broken —— 真人玩不通 / 到不了终局 / 起不来 / 卡死一屏 / 核心操作无响应;
|
||
# · hollow —— 空壳:核心循环没有决策层(点哪都一样、无输赢取舍、纯自动结算的挂机点击),判「有没有」不判「多好」;
|
||
# · off_brief —— 不切题:画面/玩法与 brief 要的游戏是两回事。
|
||
# 三条铁律(裁定三):
|
||
# ① 出题≠被考(2026-07-10 创始人修正边界):判定只吃运行证据不吃自报;判定模型允许 M3(生成同源)——
|
||
# 地板只裁「有无」不裁品味、同源盲区风险最小,且 07-09 spike 实证 M3 当场抓出 M3 自己写的渲染 bug;
|
||
# 同源残余风险由金标正反例+创始人抽玩定标兜底。修正动因=glm-5.2 唯一通道(闲鱼二手中转)图像支持
|
||
# 按池轮换、07-09 整日全盲,判定层可用性不能押注二手供给。SoT 修订随验收 v2 双评审落档。
|
||
# ② fail-closed:判定失败/评不出/无证据 → 不放行、标 degraded(绝不静默降级成自动通过)。
|
||
# ③ 校准与阻断开关:金标正反例复验 + 创始人抽玩定标;judge.blocking 开关默认 true,可一键整体回退到修订前口径。
|
||
# 判定模型走 new-api:生产档由 generation.yaml judge.model 定(现 MiniMax-M3,一手直连通道);
|
||
# 单局判定成本目标 ≤¥0.3(实测 4-6 帧约 ¥0.10,记账进 judge.costRmb)。
|
||
|
||
# 判定默认档(genconfig 不可达时的编译回落,生产以 YAML 为准):2026-07-10 随创始人决策切 M3。
|
||
# 07-09 探针实证 glm-5.2 与 MiniMax-M3 均能读图;glm-5.2 因二手供给整日全盲弃用。
|
||
_JUDGE_DEFAULT_MODEL = "MiniMax-M3"
|
||
# 三类拒绝的机读键 → 对外类名(off_brief 用下划线,与 LLM 输出的 camelCase offBrief 解耦)。
|
||
_FLOOR_CLASSES = (("broken", "broken"), ("hollow", "hollow"), ("offBrief", "off_brief"))
|
||
|
||
_FLOOR_SYSTEM = (
|
||
"你是轻量小游戏的【玩法地板判定 agent】。你不是打分器——只判「这款游戏作为一个游戏,地板成不成立」,"
|
||
"只裁【有没有】、绝不裁【做得多好】(丰富/耐玩/好玩的程度问题不归你,归另一套软评分)。"
|
||
"你只看运行证据(真玩截图 / 运行日志 / 取证状态时间线),绝不采信任何自报、承诺或源码里写了什么。"
|
||
"逐条判这三类【拒绝】是否成立:\n"
|
||
"· broken(坏死):真人根本玩不通 —— 起不来 / 一直卡在同一屏 / 核心操作点了没反应 / 到不了任何终局或结算。\n"
|
||
"· hollow(空壳):核心循环没有决策层 —— 点哪都一样、没有输赢、没有取舍、判错没有代价,纯自动结算的挂机点击。"
|
||
"只判「有没有决策层」,不判「决策做得多有层次」。\n"
|
||
"· off_brief(不切题):画面与玩法跟 brief 要的游戏是两回事(主题漂移 / 品类不符)。\n"
|
||
"任一类成立 → 整体判 reject;三类都不成立 → 判 accept。"
|
||
"证据不足以断定某一类是否成立时,该类按【成立(有问题)】算——地板判定 fail-closed,宁可错拦、不可放过坏游戏"
|
||
"(后面有人工复核与金标校准兜)。严格只输出 JSON,不要任何额外文字、不要 markdown 围栏。"
|
||
)
|
||
|
||
_FLOOR_OUTPUT_CONTRACT = (
|
||
"严格只输出以下 JSON(problem=true 表示「这一类有问题、应拒绝」,why 给一句中文依据;"
|
||
"imagesSeen=你在本条消息里【真实看到】的截图张数,整数——一张都看不到就如实填 0,"
|
||
"这个字段用于检测图像传输故障,绝不据它奖惩你):\n"
|
||
'{"imagesSeen":0,"broken":{"problem":false,"why":"…"},"hollow":{"problem":false,"why":"…"},'
|
||
'"offBrief":{"problem":false,"why":"…"},"verdict":"accept","notes":"整体一句话"}'
|
||
)
|
||
|
||
|
||
def _degraded_floor(reason: str) -> dict:
|
||
"""fail-closed 降级结果(裁定三②):评不出/无证据/调用失败 → 不放行、标 degraded。
|
||
|
||
accepted=False 是【地板判定的 fail-closed 语义】,与丰富度的 score=None(非阻塞观测)完全不同:
|
||
这里 degraded 直接构成「拒绝」(rejectClasses=['degraded']),阻断放行,由人工复核兜。
|
||
"""
|
||
return {"accepted": False, "verdict": "reject", "rejectClasses": ["degraded"],
|
||
"checks": [], "notes": None, "degraded": True, "reason": reason}
|
||
|
||
|
||
def parse_floor_judgment(text) -> dict:
|
||
"""纯函数:把判定 LLM 的 JSON 输出解析成结构化地板裁决(含 fail-closed 兜底)。可单测、零网络、零 agentscope。
|
||
|
||
成功 → {
|
||
accepted: bool, # = 三类均无问题 且 LLM verdict 未判 reject(两者取交,任一拒即拒)
|
||
verdict: "accept"|"reject",
|
||
rejectClasses: [str], # 命中的拒绝类子集(broken/hollow/off_brief)
|
||
checks: [{class, problem, why}*3],
|
||
notes: str, degraded: False}
|
||
解析失败 / 非对象 / 缺任一类判定字段 → _degraded_floor(...)(accepted=False、fail-closed)。
|
||
|
||
每一类容忍两种形状:{problem,why} 对象,或裸 bool(模型偷懒直接给 true/false);缺字段一律 fail-closed
|
||
(无法确认地板 → 保守拒绝),绝不把「没判出来」静默当成通过。
|
||
"""
|
||
if not isinstance(text, str) or not text.strip():
|
||
return _degraded_floor("判定输出为空")
|
||
try:
|
||
import json_repair # 容忍 markdown 围栏 / 前后赘语 / 尾逗号等 LLM 常见脏输出
|
||
data = json_repair.loads(text)
|
||
except Exception as e: # noqa: BLE001 解析层任何异常都 fail-closed,绝不抛
|
||
return _degraded_floor(f"判定输出解析失败：{type(e).__name__}")
|
||
if not isinstance(data, dict):
|
||
return _degraded_floor("判定输出不是 JSON 对象")
|
||
checks = []
|
||
for raw_key, cname in _FLOOR_CLASSES:
|
||
node = data.get(raw_key)
|
||
if isinstance(node, dict):
|
||
problem = _coerce_bool(node.get("problem"))
|
||
why = str(node.get("why", "")).strip()[:200]
|
||
elif node is not None: # 裸 bool/字符串:模型没按对象给,也接住
|
||
problem = _coerce_bool(node)
|
||
why = ""
|
||
else: # 缺这一类判定 → 无法确认地板 → fail-closed(不静默判过)
|
||
return _degraded_floor(f"判定输出缺 {raw_key} 字段(无法确认地板,fail-closed)")
|
||
checks.append({"class": cname, "problem": problem, "why": why})
|
||
reject_classes = [c["class"] for c in checks if c["problem"]]
|
||
llm_verdict = str(data.get("verdict", "")).strip().lower()
|
||
# accepted 需两者一致取 accept:三类均无问题 且 LLM 自评 verdict 不是 reject(任一给拒绝信号 → 拒,fail-closed 偏严)。
|
||
accepted = (not reject_classes) and (llm_verdict != "reject")
|
||
# imagesSeen:模型自报真实看到的截图张数(图像盲检测的结构化信号;缺失/非数字 → None=未知,不触发盲重掷)。
|
||
# 背景(2026-07-09 基线终审实锤):网关对 glm-5.2 按请求体确定性路由,~15% 载荷永远落在丢图通道——
|
||
# 同载荷重试必盲、任何字节微扰即换通道。靠措辞 grep 判盲太脆,改为契约字段模型自报。
|
||
images_seen = None
|
||
try:
|
||
raw_seen = data.get("imagesSeen")
|
||
if raw_seen is not None and not isinstance(raw_seen, bool):
|
||
images_seen = max(0, int(raw_seen))
|
||
except (TypeError, ValueError):
|
||
images_seen = None
|
||
return {"accepted": accepted, "verdict": "accept" if accepted else "reject",
|
||
"rejectClasses": reject_classes, "checks": checks, "imagesSeen": images_seen,
|
||
"notes": str(data.get("notes", "")).strip()[:300], "degraded": False}
|
||
|
||
|
||
class _JudgeResp:
|
||
"""判定多模态调用的极简响应载体(content=纯文本);与丰富度 _extract_text 的 str 分支兼容。"""
|
||
|
||
def __init__(self, text: str):
|
||
self.content = text if isinstance(text, str) else ""
|
||
|
||
|
||
class _NewapiVisionModel:
|
||
"""判定用的薄多模态客户端:直连 new-api /v1/chat/completions(OpenAI 兼容,支持 image_url 图像块)。
|
||
|
||
为什么不复用 build_cheap_model(agentscope M3):判定层要的是薄而可控的多模态 HTTP 客户端——
|
||
直连网关对图像块形状可控(image_url data-URI 已 probe 坐实)、finish_reason/reasoning 元信息可留痕,
|
||
与生成栈(agentscope 全家桶)解耦;仍走 new-api 统一出口(base+/v1 + NEWAPI_KEY),不违统一网关铁律。
|
||
(判定模型 2026-07-10 起=MiniMax-M3,由 YAML judge.model 定;出题≠被考边界修正见 §判定地板头注。)
|
||
trust_env=False:判定目标在内网 Tailscale(new-api 100.64.0.8),彻底绕开本机 clash 代理(fake-ip 拦本地/内网,
|
||
§7 内网直连必绕代理)。records 记 (in,out,cached) 供 judge 段成本记账(与生成成本台账隔离——判定用独立档)。
|
||
"""
|
||
|
||
def __init__(self, model_name: str, base_url: str, api_key: str, *,
|
||
max_tokens: int = 1500, timeout: float = 120.0):
|
||
self.model_name = model_name
|
||
self.base_url = base_url.rstrip("/")
|
||
self.api_key = api_key
|
||
self.max_tokens = max_tokens
|
||
self.timeout = timeout
|
||
self.records = [] # [(input_tokens, output_tokens, cached_tokens)]
|
||
self.last_meta = {} # 最近一次响应的诊断元信息(finishReason/reasoningChars)——content 空时据此归因
|
||
|
||
async def __call__(self, messages):
|
||
import httpx # noqa: PLC0415 惰性 import(顶层零重依赖;与 cheap_verify 纯函数可无 httpx 环境单测同源)
|
||
body = {"model": self.model_name, "max_tokens": self.max_tokens, "messages": messages}
|
||
async with httpx.AsyncClient(trust_env=False, timeout=self.timeout) as h:
|
||
r = await h.post(
|
||
f"{self.base_url}/chat/completions",
|
||
headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
|
||
json=body,
|
||
)
|
||
r.raise_for_status()
|
||
d = r.json()
|
||
u = d.get("usage") or {}
|
||
pi = int(u.get("prompt_tokens") or u.get("input_tokens") or 0)
|
||
po = int(u.get("completion_tokens") or u.get("output_tokens") or 0)
|
||
cached = 0
|
||
det = u.get("prompt_tokens_details") or {}
|
||
if isinstance(det, dict):
|
||
cached = int(det.get("cached_tokens") or 0)
|
||
self.records.append((pi, po, cached))
|
||
ch = (d.get("choices") or [{}])[0]
|
||
msg = ch.get("message") or {}
|
||
text = msg.get("content", "")
|
||
# 判定器自身的真相层:留 finish_reason 与 reasoning 长度。glm-5.2 带思考(reasoning_content),思考长度是
|
||
# 随机变量——吃光 max_tokens 时 content 为空、finish_reason=length(w2b-verify 2026-07-09 实测案),据此可诊断。
|
||
self.last_meta = {"finishReason": ch.get("finish_reason"),
|
||
"reasoningChars": len(msg.get("reasoning_content") or "")}
|
||
return _JudgeResp(text if isinstance(text, str) else "")
|
||
|
||
def usage_sum(self):
|
||
return (sum(r[0] for r in self.records), sum(r[1] for r in self.records))
|
||
|
||
|
||
def _wg1_evidence_dir(game_id: str) -> Path:
|
||
"""判卷证据目录 = _wg1-gen/<id>/evidence/(九门 play 落 first-paint/after-play/midplay/game-log/verdict 处)。"""
|
||
return cheap_run.wg1_game_dir(game_id) / "evidence"
|
||
|
||
|
||
def _natural_key(name: str):
|
||
"""midplay-2 排在 midplay-10 前:按名字里的数字段自然排序(纯字母段保原序)。"""
|
||
import re # noqa: PLC0415
|
||
return [int(t) if t.isdigit() else t for t in re.split(r"(\d+)", name)]
|
||
|
||
|
||
def _collect_frames(evidence_dir, *, max_frames: int = 6) -> list:
|
||
"""收 evidence 目录的真玩截图 → [data:image/png;base64,...],顺序 first-paint → 局中连拍 midplay-* → after-play。
|
||
|
||
超 max_frames 时保头尾(首帧+局末)、中间连拍均匀采样(判定需要「局前→局中→局末」的时间跨度,不能只留一头)。
|
||
读不到目录 / 无截图 → []（上层据此 fail-closed:真玩过的游戏必有 first-paint,缺失即证据异常）。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir)
|
||
if not ev.is_dir():
|
||
return []
|
||
ordered = []
|
||
fp = ev / "first-paint.png"
|
||
if fp.is_file():
|
||
ordered.append(fp)
|
||
ordered.extend(sorted(ev.glob("midplay-*.png"), key=lambda p: _natural_key(p.name)))
|
||
ap = ev / "after-play.png"
|
||
if ap.is_file():
|
||
ordered.append(ap)
|
||
if len(ordered) > max_frames:
|
||
head, tail, mid = ordered[0], ordered[-1], ordered[1:-1]
|
||
keep = max_frames - 2
|
||
if keep <= 0:
|
||
ordered = [head, tail]
|
||
else:
|
||
step = len(mid) / keep
|
||
ordered = [head] + [mid[int(i * step)] for i in range(keep)] + [tail]
|
||
out = []
|
||
for p in ordered:
|
||
try:
|
||
out.append("data:image/png;base64," + base64.b64encode(p.read_bytes()).decode())
|
||
except OSError:
|
||
continue
|
||
return out
|
||
except Exception: # noqa: BLE001 证据装配失败 → 返回空(上层 fail-closed),绝不抛
|
||
return []
|
||
|
||
|
||
def _read_state_timeline(evidence_dir) -> str:
|
||
"""从九门 verdict.json 抽一段取证状态时间线(判定的文字证据,补截图看不清的数值/终态信号)。
|
||
|
||
只搬运机械门已测出的客观信号(不加解读):H_progress 的 score 玩前/玩后、E_live 的 played/phase/活性、
|
||
latch 的终局 phase 与是否驻留、D_render 的亮度。读不到/坏 → 空串(判定仍靠截图,不伪造)。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir)
|
||
p = ev / "verdict.json"
|
||
if not p.is_file():
|
||
return ""
|
||
v = json.loads(p.read_text(encoding="utf-8"))
|
||
guards = (v or {}).get("guards") or {}
|
||
lines = []
|
||
hg = guards.get("H_progress") or {}
|
||
for c in (hg.get("checks") or []):
|
||
if isinstance(c, dict):
|
||
lines.append(f"进展[{c.get('path')}] {c.get('op')}: 玩前={c.get('before')} 玩后={c.get('after')}"
|
||
f"（机械门判 {'达成' if c.get('pass') else '未达成'}）")
|
||
latch = hg.get("latch") or {}
|
||
if isinstance(latch, dict) and latch:
|
||
lines.append(f"终局 latch: 到达终态={bool(latch.get('pass'))} 此刻 phase={latch.get('phaseNow') or latch.get('after')}"
|
||
f" {(latch.get('reason') or '')[:60]}")
|
||
el = guards.get("E_live") or {}
|
||
if isinstance(el, dict) and el:
|
||
lines.append(f"活性 E_live: 真进过游玩={el.get('played')} 玩后 phase={el.get('phase1')}"
|
||
f" 去重画面态数={el.get('distinctStates')} 状态真变={el.get('liveByState')}")
|
||
dr = guards.get("D_render") or {}
|
||
if isinstance(dr, dict) and dr:
|
||
lines.append(f"渲染 D_render: 亮度={dr.get('bright')} 最大通道={dr.get('maxCh')}")
|
||
return "\n".join(lines)
|
||
except Exception: # noqa: BLE001 取证时间线 best-effort,读不出降级空串
|
||
return ""
|
||
|
||
|
||
def _read_game_log_text(evidence_dir, *, max_entries: int = 12, max_chars: int = 800) -> str:
|
||
"""从 evidence/game-log.json 摘运行时日志(游戏语义 + 插件告警),给判定补运行期信号。读不到/坏 → 空串。"""
|
||
try:
|
||
ev = Path(evidence_dir)
|
||
p = ev / "game-log.json"
|
||
if not p.is_file():
|
||
return ""
|
||
arr = json.loads(p.read_text(encoding="utf-8"))
|
||
if not isinstance(arr, list) or not arr:
|
||
return ""
|
||
|
||
def _fmt(e):
|
||
if not isinstance(e, dict):
|
||
return str(e)[:100]
|
||
return f"[{e.get('scope')}:{e.get('tag')}] {e.get('msg') or ''}"[:100]
|
||
|
||
return "\n".join(_fmt(e) for e in arr[-max_entries:])[:max_chars]
|
||
except Exception: # noqa: BLE001 日志摘要 best-effort
|
||
return ""
|
||
|
||
|
||
def _build_floor_messages(brief: str, state_text: str, log_text: str, data_uris: list) -> list:
|
||
"""拼判定的 OpenAI 多模态消息:brief + 取证时间线 + 日志 + 「首帧/局中/局末」截图序列 + 输出契约。"""
|
||
head = (
|
||
f"这款游戏的 brief（玩家想要的）：{brief or '(未提供)'}\n\n"
|
||
"下面给你这一局【真玩】留下的运行证据。请只据这些运行证据判三类拒绝是否成立。\n\n"
|
||
f"【取证状态时间线（机械门客观测得，未加解读）】\n{state_text or '（无）'}\n\n"
|
||
f"【运行时日志（节选）】\n{log_text or '（无）'}\n\n"
|
||
f"【真玩截图】接下来 {len(data_uris)} 张图按时间顺序给出：第 1 张=首帧（刚进入），"
|
||
"中间=局中连拍（真玩过程），最后 1 张=局末（真玩结束时）。请对照 brief 看这些画面是不是一款玩得通、"
|
||
"有决策层、且切题的游戏：\n"
|
||
)
|
||
parts = [{"type": "text", "text": head}]
|
||
for uri in data_uris:
|
||
parts.append({"type": "image_url", "image_url": {"url": uri}})
|
||
parts.append({"type": "text", "text": "\n" + _FLOOR_OUTPUT_CONTRACT})
|
||
return [{"role": "system", "content": _FLOOR_SYSTEM},
|
||
{"role": "user", "content": parts}]
|
||
|
||
|
||
def _build_judge_model(model_name: str, max_tokens: int, timeout: float):
|
||
"""装判定多模态客户端(直连 new-api /v1)。key 从 env/凭据档解析(内网阶段单一事实源)。"""
|
||
import _bootstrap # noqa: PLC0415 代理旁路时序与 key 注入由 _bootstrap/client 处理
|
||
_bootstrap.ensure_api_key_env()
|
||
from worker import client # noqa: PLC0415
|
||
base = client.resolve_base_url().rstrip("/")
|
||
if not base.endswith("/v1"):
|
||
base = base + "/v1"
|
||
return _NewapiVisionModel(model_name, base, client.get_api_key(),
|
||
max_tokens=max_tokens, timeout=timeout)
|
||
|
||
|
||
def _estimate_judge_cost(model_name: str, ti: int, to: int, cached: int):
|
||
"""据 new-api /api/pricing 权威倍率把判定 token 折成 ¥(与生成成本台账同口径 observability.cost.compute)。
|
||
|
||
best-effort:取价失败(网关不可达 / 无 httpx)→ None（judge 段记 tokens、costRmb 留空,不伪造成本）。
|
||
"""
|
||
try:
|
||
from observability import cost as _cost, newapi_pricing as _np # noqa: PLC0415
|
||
params = _np.fetch_pricing_params()
|
||
if not params:
|
||
return None
|
||
d = _cost.compute(model_name, ti, to, params["pricing"], params["qpu"],
|
||
params["usd_rate"], group_ratio=1.0, cached_tokens=cached)
|
||
return round(d["rmb"], 5)
|
||
except Exception: # noqa: BLE001 成本记账 best-effort,绝不连累判定主体
|
||
return None
|
||
|
||
|
||
def _persist_judge_json(evidence_dir, result: dict, raw_text: str) -> None:
|
||
"""判定真相层落盘(best-effort,绝不抛):evidence/judge.json = 完整地板裁决 + LLM 原始输出头(截断 2000 字)。
|
||
|
||
动机(W-AXIS 波2 验收发现,2026-07-09):判定结果此前只进 run-summary → 后端 trace,本地证据目录零痕迹——
|
||
verdict.json 孤零零 pass=true,盘上看不出这局判定发生过什么;degraded(如「判定输出为空」)更无从诊断。
|
||
判定器自己也要有真相层:每次判定(含 degraded)都在证据目录留完整裁决与原始输出。目录不存在则静默跳过(单测 fake gid)。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir) if evidence_dir else None
|
||
if ev is None or not ev.is_dir():
|
||
return
|
||
payload = dict(result)
|
||
payload["rawTextHead"] = (raw_text or "")[:2000]
|
||
payload["ts"] = int(time.time() * 1000)
|
||
(ev / "judge.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception: # noqa: BLE001 落盘失败绝不连累判定主体
|
||
pass
|
||
|
||
|
||
async def judge_gameplay_floor(game_id: str, *, brief: str = "", model=None, frames=None,
|
||
state_text=None, log_text=None, evidence_dir=None,
|
||
model_name: str = None, max_tokens: int = 1500,
|
||
timeout: float = 120.0, max_frames: int = 6) -> dict:
|
||
"""对一局真玩证据跑独立多模态玩法地板判定(阻断权威;裁定三)。**fail-closed**:任何失败/无证据 → degraded（=拒绝）。
|
||
|
||
与 verify_richness 的非阻塞铁律相反——这里失败不是「静默降级放行」,而是「fail-closed 拒绝」:
|
||
评不出的游戏不许自动过,标 degraded 交人工复核。
|
||
|
||
Args:
|
||
game_id: 本局 id（读 _wg1-gen/<id>/evidence/ 下截图/日志/verdict）。
|
||
brief: 本局 brief（off_brief 判定必需）。
|
||
model: 可注入的判定 LLM 客户端（单测用 fake，零网络）；None → 建 glm-5.2 多模态客户端。
|
||
frames/state_text/log_text: 可直接注入的证据（单测用，绕文件 I/O）；None → 从 evidence_dir 收集。
|
||
evidence_dir: 证据目录；None → _wg1-gen/<id>/evidence/。
|
||
model_name/max_tokens/timeout/max_frames: 判定档旋钮（缺省走 glm-5.2 / 1500 / 120s / 6 帧）。
|
||
|
||
Returns:
|
||
parse_floor_judgment 的结构（附 judgeTokens/costRmb/model/frames 观测），或 _degraded_floor(...)（fail-closed）。
|
||
"""
|
||
mn = model_name or _JUDGE_DEFAULT_MODEL
|
||
ev = None
|
||
try:
|
||
ev = evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id)
|
||
imgs = frames if frames is not None else _collect_frames(ev, max_frames=max_frames)
|
||
if not imgs:
|
||
# 无真玩截图:真玩过的游戏必落 first-paint.png,缺失即证据异常 → fail-closed（绝不无证据放行）。
|
||
r = _degraded_floor("真玩截图证据缺失（无 first-paint/midplay/after-play），fail-closed 不放行")
|
||
r["model"] = mn
|
||
_persist_judge_json(ev, r, "")
|
||
return r
|
||
st = state_text if state_text is not None else _read_state_timeline(ev)
|
||
lg = log_text if log_text is not None else _read_game_log_text(ev)
|
||
m = model if model is not None else _build_judge_model(mn, max_tokens, timeout)
|
||
|
||
async def _roll(msgs):
|
||
"""一掷判定 = 一次调用 + 空输出有界重试一次(2026-07-09 w2b-verify 实测案):glm-5.2 带思考,
|
||
思考长度是随机变量——吃光 max_tokens 时 content 为空;同证据复跑即可得答案(主会话复现坐实)。
|
||
重试前放大 max_tokens 对症;绝不无限重试(判定成本有账)。放大取 max(原值, ×2封顶6000):
|
||
base 已调大(YAML 202752)时绝不反向缩小——旧式 min(×2,6000) 在大 base 下会把上限缩回 6000。"""
|
||
t = ""
|
||
used = 0
|
||
for attempt in range(2):
|
||
resp = await asyncio.wait_for(m(msgs), timeout=timeout)
|
||
used = attempt + 1
|
||
t = _extract_text(resp)
|
||
if isinstance(t, str) and t.strip():
|
||
break
|
||
if attempt == 0 and hasattr(m, "max_tokens"):
|
||
cur = int(getattr(m, "max_tokens", 1500))
|
||
m.max_tokens = max(cur, min(cur * 2, 6000))
|
||
return t, used
|
||
|
||
text, attempts_used = await _roll(_build_floor_messages(brief, st, lg, imgs))
|
||
result = parse_floor_judgment(text)
|
||
# 图像盲微扰重掷(2026-07-09 基线终审实锤):网关对 glm-5.2 按请求体【确定性】路由,部分载荷永远落在
|
||
# 丢图通道——同载荷重试必盲(3 例假阴复判 3/3 仍盲),而 brief 尾加一个空格微扰载荷即换路由、当场看见。
|
||
# 故:送了帧而模型自报 imagesSeen=0 → 微扰重掷一次;重掷仍自报 0 → 这是判定仪器故障不是游戏证据缺失,
|
||
# fail-closed degraded 归因「图像通道盲」交人工,不再把好游戏错杀成 hollow/off_brief(基线 3/20 假阴案)。
|
||
# imagesSeen=None(模型没报/老 fake)不触发,向后兼容。
|
||
if imgs and not result.get("degraded") and result.get("imagesSeen") == 0:
|
||
text2, used2 = await _roll(_build_floor_messages(brief + " ", st, lg, imgs))
|
||
attempts_used += used2
|
||
second = parse_floor_judgment(text2)
|
||
if not second.get("degraded") and second.get("imagesSeen") == 0:
|
||
second = _degraded_floor("判定模型两掷均自报看不见截图(网关同载荷确定性丢图)——判定仪器故障 fail-closed,待人工复核")
|
||
second["imageBlindRetried"] = True
|
||
result, text = second, text2
|
||
result["model"] = mn
|
||
result["frames"] = len(imgs)
|
||
if attempts_used > 1:
|
||
result["retries"] = attempts_used - 1
|
||
# 判定器真相层:附最近响应的 finish_reason(空输出归因的第一手证据;fake 模型无此属性则省略)。
|
||
meta = getattr(m, "last_meta", None)
|
||
if isinstance(meta, dict) and meta.get("finishReason") is not None:
|
||
result["finishReason"] = meta.get("finishReason")
|
||
# 判定成本记账（独立档，不污染生成 costRmb）——best-effort。
|
||
try:
|
||
ti, to = m.usage_sum()
|
||
cached = sum(r[2] for r in getattr(m, "records", []) if len(r) > 2)
|
||
result["judgeTokens"] = {"in": ti, "out": to, "total": ti + to}
|
||
result["costRmb"] = _estimate_judge_cost(mn, ti, to, cached)
|
||
except Exception: # noqa: BLE001 成本/观测字段失败不影响判定主体
|
||
pass
|
||
_persist_judge_json(ev, result, text)
|
||
return result
|
||
except (asyncio.TimeoutError, TimeoutError):
|
||
r = _degraded_floor(f"判定 LLM 超时（>{timeout}s），fail-closed 不放行")
|
||
r["model"] = mn
|
||
_persist_judge_json(ev, r, "")
|
||
return r
|
||
except Exception as e: # noqa: BLE001 fail-closed:任何异常（网络/import/解析）都判 degraded=拒绝,绝不放行
|
||
r = _degraded_floor(f"判定异常：{type(e).__name__}: {e}")
|
||
r["model"] = mn
|
||
_persist_judge_json(ev, r, "")
|
||
return r
|
||
|
||
|
||
def _judge_cfg() -> dict:
|
||
"""读判定档配置（genconfig judge.*；worker 不可达时回落编译默认，保证无 agentscope 环境也能取值/单测）。"""
|
||
try:
|
||
from worker import genconfig # noqa: PLC0415
|
||
g = genconfig.get
|
||
except Exception: # noqa: BLE001 无 worker（纯单测环境）→ 用编译默认
|
||
def g(_area, _key, default):
|
||
return default
|
||
return {
|
||
"blocking": bool(g("judge", "blocking", True)),
|
||
"model": g("judge", "model", _JUDGE_DEFAULT_MODEL),
|
||
"max_tokens": int(g("judge", "max_tokens", 1500)),
|
||
"timeout_s": float(g("judge", "timeout_s", 120.0)),
|
||
"max_frames": int(g("judge", "max_frames", 6)),
|
||
"cost_target_rmb": float(g("judge", "cost_target_rmb", 0.3)),
|
||
}
|
||
|
||
|
||
def _writeback_verdict_json(evidence_dir, accepted: bool, judge_summary: dict) -> None:
|
||
"""把最终验收权威增量写回盘上 evidence/verdict.json(best-effort,绝不抛)。
|
||
|
||
字段只加不减:`pass` 物理保留(语义=预筛/机械九门),新增 `accepted`(=预筛∧玩法判定,最终权威)与
|
||
`judge` 摘要段。动机(W-AXIS 波2 验收发现):此前判定只进 run-summary→后端 trace,盘上 verdict.json
|
||
孤零零 pass=true——证据链与真实验收结论脱节(§6.13 亲眼验收看的就是盘上证据)。verdict.json 不存在则跳过。
|
||
"""
|
||
try:
|
||
p = (Path(evidence_dir) / "verdict.json") if evidence_dir else None
|
||
if p is None or not p.is_file():
|
||
return
|
||
v = json.loads(p.read_text(encoding="utf-8"))
|
||
if not isinstance(v, dict):
|
||
return
|
||
v["accepted"] = bool(accepted)
|
||
v["judge"] = judge_summary
|
||
p.write_text(json.dumps(v, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception: # noqa: BLE001 写回失败绝不连累判定主链
|
||
pass
|
||
|
||
|
||
async def apply_gameplay_judge(summary: dict, *, game_id: str, brief: str = "",
|
||
model=None, blocking=None, evidence_dir=None,
|
||
prefilter=None) -> dict:
|
||
"""把玩法地板判定接到 run-summary，落 ok 语义切换（W-AXIS 波2 · SoT 裁定三代码兑现）。
|
||
|
||
ok 语义（裁定三）:`ok = 预筛 ∧ 判定`。verdict 的 `pass` 字段物理保留、语义收窄为「预筛（机械九门）通过」,
|
||
新增 `accepted`（= 预筛 ∧ 玩法地板判定）与 `judge` 段（裁决/三类理由/degraded/成本）——字段只加不减。
|
||
|
||
· 预筛 = 显式 `prefilter` 入参;缺省(None)回落 summary 现有的 ok。**预筛必须是机械九门口径**(裁定一:
|
||
driven 时九门全量 AND)——Service 路 summary.ok=九门 judge.passed 天然合规可走缺省;CLI(cheap_studio)路
|
||
summary.ok=finished(模型自称收敛),**不是**九门信号,必须显式传 prefilter=finished∧verdict.pass
|
||
(2026-07-09 W3 真跑暴露:三门挂的局按 finished 预筛被判定放行成 accepted=True,违裁定三,已修)。
|
||
· 判定只对【过筛者】跑（省 ¥:预筛没过的游戏已被机械门拒,不再花判定钱;对齐诊断档 §5「预筛先挡明显死」）。
|
||
· blocking=True（默认）→ ok=accepted（判定阻断放行）;blocking=False → ok=预筛（判定观测不阻断,整体回退位）。
|
||
· fail-closed:证据缺/调用失败/解析失败 → judge.degraded=True → floor 不成立 → (blocking) accepted=False。
|
||
|
||
绝不抛（顶层兜底）:additive 写 summary['accepted'] 与 summary['judge']，并按上式更新 summary['ok']。
|
||
blocking/evidence_dir 缺省 None：blocking→读 genconfig；evidence_dir→按 game_id 推 _wg1-gen/<id>/evidence/。
|
||
"""
|
||
try:
|
||
cfg = _judge_cfg()
|
||
blk = cfg["blocking"] if blocking is None else bool(blocking)
|
||
ev = evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id)
|
||
prefilter = bool(summary.get("ok")) if prefilter is None else bool(prefilter)
|
||
if not prefilter:
|
||
# 预筛未过:机械门已拒,不花判定钱。accepted=预筛(False);ok 维持 False。
|
||
summary["judge"] = {"ran": False, "blocking": blk,
|
||
"reason": "预筛（机械九门）未过，不跑玩法判定"}
|
||
summary["accepted"] = False
|
||
summary["ok"] = False
|
||
_writeback_verdict_json(ev, False, summary["judge"])
|
||
return summary
|
||
floor = await judge_gameplay_floor(
|
||
game_id, brief=brief, model=model, evidence_dir=ev,
|
||
model_name=cfg["model"], max_tokens=cfg["max_tokens"],
|
||
timeout=cfg["timeout_s"], max_frames=cfg["max_frames"])
|
||
floor_ok = bool(floor.get("accepted")) and not bool(floor.get("degraded"))
|
||
accepted = prefilter and floor_ok
|
||
summary["judge"] = {
|
||
"ran": True, "blocking": blk,
|
||
"verdict": floor.get("verdict"),
|
||
"accepted": bool(floor.get("accepted")),
|
||
"rejectClasses": floor.get("rejectClasses"),
|
||
"checks": floor.get("checks"),
|
||
"degraded": bool(floor.get("degraded")),
|
||
"reason": floor.get("reason"),
|
||
"notes": floor.get("notes"),
|
||
"model": floor.get("model"),
|
||
"frames": floor.get("frames"),
|
||
"costRmb": floor.get("costRmb"),
|
||
"judgeTokens": floor.get("judgeTokens"),
|
||
}
|
||
summary["accepted"] = accepted
|
||
summary["ok"] = accepted if blk else prefilter
|
||
# 阻断路且未过:degraded 走可重试类失败因（result_out 缺省 llm_error 即可重试；不覆盖已有 quota_exhausted）。
|
||
if blk and not accepted and floor.get("degraded"):
|
||
summary.setdefault("failureReason", "llm_error")
|
||
# 盘上证据链与最终验收结论对齐(judge.json 由 judge_gameplay_floor 已落;这里补 verdict.json 增量段)。
|
||
_writeback_verdict_json(ev, accepted, summary["judge"])
|
||
return summary
|
||
except Exception as e: # noqa: BLE001 判定接线级异常:fail-closed（blocking→不放行）,observe 模式保预筛。
|
||
blk = True if blocking is None else bool(blocking)
|
||
try:
|
||
blk = _judge_cfg()["blocking"] if blocking is None else bool(blocking)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
prefilter = bool(summary.get("ok")) if prefilter is None else bool(prefilter)
|
||
summary["judge"] = {"ran": False, "blocking": blk, "degraded": True,
|
||
"reason": f"判定接线异常（fail-closed）：{type(e).__name__}: {e}"}
|
||
summary["accepted"] = False if blk else prefilter
|
||
summary["ok"] = False if blk else prefilter
|
||
try: # 接线异常路也尽力对齐盘上证据(ev 推导自身可能失败,再兜一层)
|
||
_writeback_verdict_json(
|
||
evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id),
|
||
summary["accepted"], summary["judge"])
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return summary
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 验收 v2:四门投影 + 测试 agent 真玩 + 统一编排器(W-AXIS-V2 波1 · 「拆着杀」)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# W-AXIS 三波把验收权威交给独立模型判定,但证据仍由「一份固定代码玩几十种游戏」的取证契约(play-spec /
|
||
# tap-targets 驱动器 / _forensicsView)生产——n=5 基线逐局破案坐实这条证据生产线本身就是最大失败源。
|
||
# 本段兑现创始人 2026-07-09「拆着杀」:契约无关的四门(A/B/C/D)降为地板保留,依赖契约的 E/G/H/I/F 与
|
||
# 驱动器退役,验收证据的生产与裁决交测试 agent(playtest.cdp.cjs:视觉引导真玩 + 玩法地板判定合一)。
|
||
# 三路(CLI/Service/modify)共用一个编排器 run_acceptance;acceptance.mode 三态 v2/shadow/v1 控阻断与灰度。
|
||
|
||
# 四门 = 契约无关:读装载标志 / 未捕获异常 / 引擎帧号 / canvas 像素,不需游戏配合任何自报接口,
|
||
# 零成本、确定性、抓真死(基线里 boot 空指针正是 A 门拦的)。F_wiring 的期望前缀集源自 play-spec、
|
||
# 随契约退役降观测,不入地板(裁定三 2026-07-10 修订)。
|
||
_FLOOR_GATES = ("A_boot", "B_uncaught", "C_frame", "D_render")
|
||
|
||
|
||
def project_floor(verdict) -> dict:
|
||
"""四门投影(floor 段的**唯一产地**):从 verdict.guards 显式抽取 A_boot/B_uncaught/C_frame/D_render 求 AND。
|
||
|
||
这是消费层函数,**绝不复用 harness 原生 verdict.pass**——那是九门(driven)/七门(undriven)AND、随驱动器语义,
|
||
拆契约后会静默坍缩;四门投影独立于它,是便宜档预筛的权威取值(裁定一 2026-07-10 修订)。
|
||
任一门缺失(guards 无该门 dict)或 pass 非 True → floor.pass=False(fail-closed:无四门证据不放行)。
|
||
返回 {pass: bool, gates: {A,B,C,D: bool}}(短键 A/B/C/D = plan §4 字段协议)。
|
||
"""
|
||
guards = (verdict or {}).get("guards") or {}
|
||
gates = {}
|
||
all_ok = True
|
||
for full in _FLOOR_GATES:
|
||
short = full.split("_")[0] # A_boot → A
|
||
node = guards.get(full)
|
||
ok = isinstance(node, dict) and node.get("pass") is True
|
||
gates[short] = ok
|
||
all_ok = all_ok and ok
|
||
return {"pass": all_ok, "gates": gates}
|
||
|
||
|
||
def apply_acceptance_failure_attribution(summary: dict) -> dict:
|
||
"""按最终 v2 验收权威重算失败归因，阻止旧 E/G/H 等降观测门污染批账。
|
||
|
||
v2 的成败权威只有 ``accepted = floor ∧ playtest``：
|
||
· accepted=True → 无失败，归因固定为 none，原因置空；
|
||
· 四门地板未过 → mechanical，只列 A/B/C/D 中未过或缺失的门；
|
||
· 测试员 degraded → tester_degraded，保留测试员给出的故障原因；
|
||
· 四门全过但测试员明确拒绝 → gameplay，原因只取测试员裁决与客观现象。
|
||
|
||
shadow/v1 仍以旧验收器为权威，原有三层归因保持不变。函数原地更新 summary 并返回它，供
|
||
CLI、Service、modify 三路共用的 ``run_acceptance`` 在最终裁决后统一调用。
|
||
"""
|
||
if (summary or {}).get("acceptanceVersion") != "v2":
|
||
return summary
|
||
|
||
if summary.get("accepted") is True:
|
||
summary["failureLayer"] = {"layer": "none", "reason": None, "failedGates": []}
|
||
return summary
|
||
|
||
floor = summary.get("floor") if isinstance(summary.get("floor"), dict) else {}
|
||
playtest = summary.get("playtest") if isinstance(summary.get("playtest"), dict) else {}
|
||
judge = summary.get("judge") if isinstance(summary.get("judge"), dict) else {}
|
||
|
||
# 四门是测试员之前的机械地板；即使后续编排也异常，已知地板失败仍是更靠前、更确定的根因。
|
||
# floor/gates 缺字段按 fail-closed 视作对应门未通过，避免无证据时误落 gameplay。
|
||
if floor.get("pass") is not True:
|
||
gate_names = {"A": "A_boot", "B": "B_uncaught", "C": "C_frame", "D": "D_render"}
|
||
gates = floor.get("gates") if isinstance(floor.get("gates"), dict) else {}
|
||
failed = [full for short, full in gate_names.items() if gates.get(short) is not True]
|
||
summary["failureLayer"] = {
|
||
"layer": "mechanical",
|
||
"reason": f"四门地板失败({'/'.join(failed)}):装载/异常/掌帧/渲染证据未全部通过",
|
||
"failedGates": failed,
|
||
}
|
||
return summary
|
||
|
||
# 四门全绿后测试员自身无法完成裁决，失败属于验收仪器，不可误归为游戏玩法缺陷。
|
||
# 编排器/写回层异常会在保留已有 playtest 的同时把 judge 标成 degraded；此时必须优先归因验收器，
|
||
# 不能因为测试员先前已产出 accept/reject 就把基础设施故障误算成 gameplay。
|
||
if playtest.get("degraded") or judge.get("degraded"):
|
||
reason = playtest.get("reason") or judge.get("reason") or "测试员未能产出有效裁决，fail-closed 待人工复核"
|
||
summary["failureLayer"] = {
|
||
"layer": "tester_degraded",
|
||
"reason": f"测试员降级:{reason}",
|
||
"failedGates": [],
|
||
}
|
||
return summary
|
||
|
||
# 测试员明确拒绝时，只引用真玩裁决与去掉“推测”分句后的客观现象，不再读取旧 E/G/H 门结果。
|
||
import re # noqa: PLC0415
|
||
|
||
observations = []
|
||
for problem in playtest.get("problems") or []:
|
||
phenomenon = "；".join(
|
||
segment.strip() for segment in re.split(r"[;；]", str(problem))
|
||
if segment.strip() and "推测" not in segment
|
||
)
|
||
if phenomenon:
|
||
observations.append(phenomenon)
|
||
# gameplay 归因只引用测试员记录的客观现象；summary/reason 可能夹带推测，不作为有现象时的根因文案。
|
||
if observations:
|
||
suffix = "；".join(observations[:3])
|
||
else:
|
||
suffix = str(playtest.get("summary") or playtest.get("reason") or "测试员真玩后明确判定未通过")
|
||
summary["failureLayer"] = {
|
||
"layer": "gameplay",
|
||
"reason": f"测试员真玩拒绝:{suffix}",
|
||
"failedGates": [],
|
||
}
|
||
return summary
|
||
|
||
|
||
def _acceptance_cfg() -> dict:
|
||
"""读验收 v2 配置(genconfig acceptance.* + playtest.* + judge.*;worker 不可达回落编译默认,保单测)。"""
|
||
try:
|
||
from worker import genconfig # noqa: PLC0415
|
||
g = genconfig.get
|
||
except Exception: # noqa: BLE001 无 worker(纯单测)→ 编译默认
|
||
def g(_area, _key, default):
|
||
return default
|
||
configured_mode = str(g("acceptance", "mode", "v3_shadow"))
|
||
return {
|
||
# v2 helper 保留旧三态返回值，避免历史回放测试和显式调用漂移；configured_mode 记录生产总开关。
|
||
"mode": configured_mode if configured_mode in ("v2", "shadow", "v1") else "v2",
|
||
"configured_mode": configured_mode,
|
||
"model": g("judge", "model", _JUDGE_DEFAULT_MODEL), # 测试员 = 判定档同一配置(§2;现 MiniMax-M3)
|
||
"steps_base": int(g("playtest", "steps_base", 14)), # 基础步数预算(§3 第6件)
|
||
"steps_max": int(g("playtest", "steps_max", 24)), # 有进展扩展上限
|
||
"second_roll": bool(g("playtest", "second_roll", True)),# fail 二掷确认(§3 第7件)
|
||
"timeout_s": float(g("playtest", "timeout_s", 600.0)), # 单掷子进程墙钟超时
|
||
"cost_cap_rmb": float(g("playtest", "cost_cap_rmb", 1.5)), # 单局验收成本上限(含二掷)
|
||
"blocking": bool(g("judge", "blocking", True)), # 测试员裁决是否阻断放行
|
||
}
|
||
|
||
|
||
def _playtest_script() -> Path:
|
||
"""serve-and-playtest.sh 路径(与 serve-and-play.sh 同套起服编排,末步换调 playtest.cdp.cjs)。"""
|
||
return cheap_run._WG1_DIR / "_shared" / "serve-and-playtest.sh"
|
||
|
||
|
||
def _playtest_env() -> dict:
|
||
"""subprocess env:复用 cheap_run 起服 env(CHROME_BIN + NO_PROXY)+ 注入 NEWAPI_KEY/BASE(测试员连 new-api)。
|
||
|
||
playtest.cdp.cjs 用 Node http 模块直连 new-api、天然绕系统代理;key/base 由本函数从凭据档解析后注入 env。
|
||
"""
|
||
env = dict(cheap_run._shell_env())
|
||
# ActorGuide renderer 依赖 cheap-worker 已安装的 Pillow；强制复用当前解释器，禁止漂到系统 python3。
|
||
env["PYTHON"] = sys.executable
|
||
try:
|
||
import _bootstrap # noqa: PLC0415 代理旁路时序与 key 注入
|
||
_bootstrap.ensure_api_key_env()
|
||
from worker import client # noqa: PLC0415
|
||
env["NEWAPI_KEY"] = client.get_api_key()
|
||
env["NEWAPI_BASE_URL"] = client.resolve_base_url()
|
||
except Exception: # noqa: BLE001 凭据解析失败 → 交 cjs 自身的 env 兜底(仍可能从进程环境拿到)
|
||
pass
|
||
return env
|
||
|
||
|
||
# Chrome unsafe ports(net::ERR_UNSAFE_PORT):派生范围 4998..5097 内被 Chrome 封锁的页面加载端口。
|
||
# 2026-07-10 考卷真跑实锤:hard-sim-business-r2 派生到 5060(SIP),curl 就绪检查 200 而 Chrome navigate
|
||
# 直接拒(NAT slipstreaming 防护),__genBooted 永不置位 → 假 boot-timeout。派生必须跳过。
|
||
_CHROME_UNSAFE_PORTS = {5060, 5061}
|
||
|
||
|
||
def _derive_playtest_ports(game_id) -> tuple:
|
||
"""按 game_id 确定性派生测试员起服端口(避并发多局撞:Service 异步队列 / modify / 批跑可能同时收口)。
|
||
|
||
基址 4998/9331(spike 同段,与九门 play 的 4320/9222 分离)+ game_id 稳定 hash % 100(md5,禁 Python
|
||
随机化 hash)。同一 game_id 恒得同端口(可复现);不同 game_id 大概率不撞(并发 ≤15 远低于 100 槽)。
|
||
静态服务端口撞 Chrome unsafe 清单(5060/5061=SIP,页面加载被 ERR_UNSAFE_PORT 拒)→ +2 避开
|
||
(CDP 端口无此限制——unsafe 清单只管页面加载,不管 Chrome 自身监听)。
|
||
"""
|
||
import hashlib # noqa: PLC0415
|
||
h = int(hashlib.md5(str(game_id).encode("utf-8")).hexdigest()[:4], 16) % 100
|
||
port = 4998 + h
|
||
if port in _CHROME_UNSAFE_PORTS:
|
||
port += 2 # 5060→5062 / 5061→5063(5062/5063 不在清单;仍确定性可复现)
|
||
return port, 9331 + h
|
||
|
||
|
||
_V3_PORT_SLOT_COUNT = 20_000
|
||
_V3_HTTP_PORT_BASE = 20_000
|
||
_V3_CDP_PORT_BASE = 40_000
|
||
|
||
|
||
def _derive_playtest_v3_ports(run_id, attempt: int = 0) -> tuple:
|
||
"""只给 v3 派生高端端口；attempt 进入 hash，基础设施重试不会原样复撞同一端口。
|
||
|
||
HTTP 使用 20000..39999，避开 Chrome 已知 restricted ports；CDP 使用 40000..59999，与静态服务
|
||
完全分段。20,000 个稳定槽让生产并发 15 局的生日碰撞概率降到约 0.5%。
|
||
"""
|
||
attempt = int(attempt)
|
||
if attempt < 0:
|
||
raise ValueError("v3 port attempt 不得为负数")
|
||
key = f"{run_id}:attempt:{attempt}".encode("utf-8")
|
||
slot = int.from_bytes(hashlib.sha256(key).digest()[:8], "big") % _V3_PORT_SLOT_COUNT
|
||
return _V3_HTTP_PORT_BASE + slot, _V3_CDP_PORT_BASE + slot
|
||
|
||
|
||
def _run_playtest_roll_sync(game_id, brief_file, *, model_name, roll, seed, steps_base, steps_max,
|
||
evidence_dir, port, cdp_port, timeout) -> dict:
|
||
"""同步跑一掷(subprocess serve-and-playtest.sh → playtest.cdp.cjs);解析 stdout 末行 JSON 裁决。
|
||
|
||
绝不抛——超时/异常/未产裁决一律归 degraded 裁决(fail-closed)。退出码语义(playtest/1 协议):
|
||
0=正常裁决、2=运行错误/装载失败(degraded)、3=图像通道故障(degraded,归因测试员图像通道非游戏缺陷)。
|
||
"""
|
||
argv = ["bash", str(_playtest_script()), str(game_id), str(port), str(cdp_port), "--",
|
||
f"--model={model_name}", f"--roll={roll}", f"--seed={seed}",
|
||
f"--steps-base={steps_base}", f"--steps-max={steps_max}",
|
||
f"--brief-file={brief_file}", f"--evidence-dir={evidence_dir}"]
|
||
try:
|
||
r = subprocess.run(argv, cwd=str(cheap_run._GAME_RUNTIME), capture_output=True, text=True,
|
||
timeout=timeout, env=_playtest_env(), check=False)
|
||
except subprocess.TimeoutExpired:
|
||
return {"degraded": True, "reason": f"测试员子进程超时(>{timeout}s),fail-closed", "exitCode": -1, "roll": roll}
|
||
except Exception as e: # noqa: BLE001 subprocess 层异常一律 fail-closed
|
||
return {"degraded": True, "reason": f"测试员子进程异常:{type(e).__name__}: {e}", "exitCode": -1, "roll": roll}
|
||
# 解析 stdout 末行单行 JSON(playtest.cdp.cjs 末行 = 裁决)。
|
||
out = None
|
||
for line in reversed((r.stdout or "").splitlines()):
|
||
line = line.strip()
|
||
if line.startswith("{") and line.endswith("}"):
|
||
try:
|
||
out = json.loads(line)
|
||
break
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if out is None:
|
||
return {"degraded": True, "reason": "测试员未产出裁决 JSON(见 raw)", "exitCode": r.returncode,
|
||
"raw": (r.stdout + r.stderr)[-1000:], "roll": roll}
|
||
out["exitCode"] = r.returncode
|
||
# 退出码/自报标志 → degraded(fail-closed):运行错误、装载失败、图像通道故障都不放行。
|
||
if r.returncode == 2 or out.get("runnerError") or out.get("bootDead"):
|
||
out["degraded"] = True
|
||
out.setdefault("reason", "测试员运行错误/装载失败,fail-closed")
|
||
if r.returncode == 3 or out.get("imageBlind"):
|
||
out["degraded"] = True
|
||
out["imageBlind"] = True
|
||
out.setdefault("reason", "测试员图像通道故障(自报看不到截图),fail-closed 归因测试员非游戏")
|
||
return out
|
||
|
||
|
||
def _roll_brief(roll) -> dict:
|
||
"""把单掷裁决压成轻量摘要(落 playtest.json 的 rolls 列表 / 批次账,不塞完整转写)。"""
|
||
return {"roll": roll.get("roll"), "seed": roll.get("seed"), "pass": roll.get("pass"),
|
||
"degraded": bool(roll.get("degraded")), "canSee": roll.get("canSee"),
|
||
"steps": roll.get("steps"), "tapPoints": roll.get("tapPoints"),
|
||
"summary": roll.get("summary"), "reason": roll.get("reason"),
|
||
"tokens": roll.get("tokens")}
|
||
|
||
|
||
def _playtest_roll_cost(model_name, roll) -> float:
|
||
"""据 new-api 权威倍率把一掷 token 折成 ¥(与判定成本同口径);取价失败 → 0.0(不伪造成本、不误停)。"""
|
||
tok = roll.get("tokens") or {}
|
||
ti, to = int(tok.get("in") or 0), int(tok.get("out") or 0)
|
||
c = _estimate_judge_cost(model_name, ti, to, 0)
|
||
return float(c) if isinstance(c, (int, float)) else 0.0
|
||
|
||
|
||
def _persist_playtest_json(evidence_dir, result: dict) -> None:
|
||
"""测试员验收真相层落盘(best-effort,绝不抛):evidence/playtest/playtest.json = 汇总裁决 + 二掷账。
|
||
|
||
与 judge.json 同一动机(W-AXIS 波2):验收结论必须在盘上留痕,§6.13 亲眼验收看的就是它 + roll-N 截图序列。
|
||
"""
|
||
try:
|
||
ev = Path(evidence_dir) if evidence_dir else None
|
||
if ev is None:
|
||
return
|
||
pdir = ev / "playtest"
|
||
pdir.mkdir(parents=True, exist_ok=True)
|
||
payload = dict(result)
|
||
payload["ts"] = int(time.time() * 1000)
|
||
(pdir / "playtest.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception: # noqa: BLE001 落盘失败绝不连累验收主链
|
||
pass
|
||
|
||
|
||
async def run_playtest(game_id, *, brief="", model_name=None, evidence_dir=None,
|
||
port=None, cdp_port=None, steps_base=None, steps_max=None,
|
||
second_roll=None, timeout=None, cost_cap_rmb=None, seed=None) -> dict:
|
||
"""测试 agent 真玩验收(阻断权威;裁定三 v2)。**fail-closed**:超时/图像故障/未裁决 → degraded=拒绝。
|
||
|
||
二掷(§3 第7件):roll-1 判 fail 且非 degraded、second_roll 开、成本未撞上限 → roll-2(不同 seed、清
|
||
localStorage 独立);两掷同 fail 才落 fail,roll-2 pass 即翻案(治 puzzle 类坐标失手假阴)。pass 一掷即过。
|
||
|
||
Returns:
|
||
playtest 段 dict:{accepted, verdict, pass, degraded, reason, problems, feedback, summary,
|
||
firstPlay, rolls, rollCount, costRmb, model, imageBlind}。落 evidence/playtest/playtest.json。
|
||
"""
|
||
cfg = _acceptance_cfg()
|
||
mn = model_name or cfg["model"]
|
||
sb = steps_base if steps_base is not None else cfg["steps_base"]
|
||
smax = steps_max if steps_max is not None else cfg["steps_max"]
|
||
sr = cfg["second_roll"] if second_roll is None else bool(second_roll)
|
||
to = timeout if timeout is not None else cfg["timeout_s"]
|
||
cap = cost_cap_rmb if cost_cap_rmb is not None else cfg["cost_cap_rmb"]
|
||
ev = Path(evidence_dir) if evidence_dir is not None else _wg1_evidence_dir(game_id)
|
||
base_seed = seed if seed is not None else int(time.time() * 1000) % 1_000_000
|
||
if port is None or cdp_port is None: # 缺省 → 按 game_id 派生 per-game 端口(并发避撞)
|
||
dp, dc = _derive_playtest_ports(game_id)
|
||
port = dp if port is None else port
|
||
cdp_port = dc if cdp_port is None else cdp_port
|
||
_NULL_FIRSTPLAY = {"playableAtMs": None, "firstFeedbackMs": None, "loopClosed": False}
|
||
|
||
result = {"model": mn, "rolls": [], "degraded": False}
|
||
brief_file = None
|
||
try:
|
||
import tempfile # noqa: PLC0415 brief 写临时文件(诚实来源,避免命令行转义/长度)
|
||
with tempfile.NamedTemporaryFile("w", suffix=".brieftxt", delete=False, encoding="utf-8") as bf:
|
||
bf.write(brief or "")
|
||
brief_file = bf.name
|
||
|
||
async def _one(roll, s):
|
||
# subprocess 阻塞 → to_thread 让出事件循环(超时由 subprocess timeout 兜)。
|
||
return await asyncio.to_thread(
|
||
_run_playtest_roll_sync, game_id, brief_file,
|
||
model_name=mn, roll=roll, seed=s, steps_base=sb, steps_max=smax,
|
||
evidence_dir=str(ev), port=port, cdp_port=cdp_port, timeout=to)
|
||
|
||
# roll-1
|
||
r1 = await _one(1, base_seed)
|
||
result["rolls"].append(_roll_brief(r1))
|
||
cost = _playtest_roll_cost(mn, r1)
|
||
final = r1
|
||
# 二掷:仅当 roll-1 是【确定的 fail】(非 degraded)、开关开、成本未撞线时;degraded 不二掷(图像/运行故障重掷无益)。
|
||
if (not r1.get("degraded")) and (r1.get("pass") is False) and sr and cost < cap:
|
||
r2 = await _one(2, base_seed + 7919) # 素数偏移换 seed(禁隐式时钟种子)
|
||
result["rolls"].append(_roll_brief(r2))
|
||
cost += _playtest_roll_cost(mn, r2)
|
||
if r2.get("pass") is True:
|
||
final = r2 # roll-2 翻案 → pass
|
||
elif r2.get("degraded"):
|
||
final = r2 # roll-2 degraded → fail-closed
|
||
else:
|
||
final = r1 # 两掷同 fail → 维持 fail
|
||
except Exception as e: # noqa: BLE001 编排层任何异常 → fail-closed degraded
|
||
result.update({"accepted": False, "verdict": "reject", "pass": False, "degraded": True,
|
||
"reason": f"测试员编排异常:{type(e).__name__}: {e}", "problems": [], "feedback": "",
|
||
"summary": "orchestration-error", "firstPlay": dict(_NULL_FIRSTPLAY),
|
||
"rollCount": len(result["rolls"]), "costRmb": 0.0, "imageBlind": False})
|
||
_persist_playtest_json(ev, result)
|
||
return result
|
||
finally:
|
||
if brief_file:
|
||
try:
|
||
import os as _os # noqa: PLC0415
|
||
_os.unlink(brief_file)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
final_pass = final.get("pass") is True
|
||
final_degraded = bool(final.get("degraded"))
|
||
result.update({
|
||
"accepted": final_pass and not final_degraded,
|
||
"verdict": "accept" if (final_pass and not final_degraded) else "reject",
|
||
"pass": final_pass,
|
||
"degraded": final_degraded,
|
||
"reason": final.get("reason"),
|
||
"problems": final.get("problems") or [], # 现象+推测(修复反馈拼装只引现象,见 _build_repair_feedback)
|
||
"feedback": final.get("feedback") or "",
|
||
"summary": final.get("summary") or "",
|
||
"firstPlay": final.get("firstPlay") or dict(_NULL_FIRSTPLAY),
|
||
"rollCount": len(result["rolls"]),
|
||
"costRmb": round(cost, 5),
|
||
"imageBlind": bool(final.get("imageBlind")),
|
||
})
|
||
_persist_playtest_json(ev, result)
|
||
return result
|
||
|
||
|
||
def _build_repair_feedback(floor: dict, playtest: dict) -> str:
|
||
"""修复反馈拼装(现象/推测强制分离,续修**只引现象段**)——给 writer resume 的续修来源。
|
||
|
||
治 spike 三次「测试员失手后错误归因」失效模式:测试员 problems 里凡含「推测」的分句剔除,只留带步号+落点的
|
||
客观现象进反馈;degraded 则不派续修(fail-closed 待人工)。附四门地板逐门结果作接线上下文。
|
||
"""
|
||
import re # noqa: PLC0415
|
||
if playtest.get("degraded"):
|
||
return f"测试员未能给出有效裁决(degraded:{playtest.get('reason')}),fail-closed 待人工复核,不派续修。"
|
||
g = (floor or {}).get("gates") or {}
|
||
gate_str = " ".join(f"{k}={'✓' if g.get(k) else '✗'}" for k in ("A", "B", "C", "D"))
|
||
lines = []
|
||
for p in (playtest.get("problems") or []):
|
||
# 现象/推测分离:按分句剔除含「推测」的段,只留客观现象。
|
||
phenom = "；".join(seg.strip() for seg in re.split(r"[;；]", str(p)) if seg.strip() and "推测" not in seg)
|
||
if phenom:
|
||
lines.append(phenom)
|
||
head = (f"[测试员真玩现象·续修依据] 四门地板 {gate_str};测试员判不通过。以下仅列客观现象(带步号与落点坐标,"
|
||
f"推测已剔除),据此定位修复:\n")
|
||
if not lines:
|
||
lines = [playtest.get("summary") or "测试员判不通过但未列具体现象"]
|
||
return head + "\n".join(f"- {ln}" for ln in lines)
|
||
|
||
|
||
async def run_acceptance(summary: dict, *, game_id: str, brief: str = "", verdict=None,
|
||
evidence_dir=None, mode=None, model=None,
|
||
port=None, cdp_port=None) -> dict:
|
||
"""统一验收编排器(CLI / Service / modify 三路共用)。accepted = floor ∧ 测试员(mode=v2)/旧口径(shadow/v1)。
|
||
|
||
· floor = project_floor(verdict):四门投影(预筛权威,唯一产地),写 summary['floor']。
|
||
· mode=v2(目标态):测试员真玩阻断,accepted = floor.pass ∧ playtest.accepted。
|
||
· mode=shadow(波1 灰度):测试员照跑照落证据(对照),accepted 仍取旧口径(判定器 apply_gameplay_judge);
|
||
并写 summary['shadowV2Accepted'] 供批次账新旧口径对照表。
|
||
· mode=v1(回退态):W-AXIS 旧口径(apply_gameplay_judge:driven 九门预筛→四门投影 + 判定器读驱动器证据)。
|
||
|
||
additive 写 summary:floor 段、acceptanceVersion、playtest 段、judge 段(§4 双写窗口)、ok/accepted。
|
||
绝不抛(顶层兜底 fail-closed:异常 → accepted=False)。verdict 缺 → 取 summary.verdict/verdictFull。
|
||
"""
|
||
try:
|
||
cfg = _acceptance_cfg()
|
||
# v2 只允许显式历史回放。生产总开关进入 v3_shadow/v3 后，旧入口不得静默把新模式当 v2 执行。
|
||
if mode is None and cfg.get("configured_mode") not in ("v2", "shadow", "v1"):
|
||
summary["acceptanceVersion"] = cfg.get("configured_mode")
|
||
summary["accepted"] = False
|
||
summary["ok"] = False
|
||
summary["judge"] = {
|
||
"ran": False, "degraded": True,
|
||
"reason": "v2 验收已退出生产自动路径；历史回放必须显式传 mode=v2/shadow/v1",
|
||
}
|
||
return summary
|
||
mode = mode or cfg["mode"]
|
||
ev = evidence_dir if evidence_dir is not None else _wg1_evidence_dir(game_id)
|
||
v = verdict if verdict is not None else (summary.get("verdict") or summary.get("verdictFull") or {})
|
||
floor = project_floor(v)
|
||
summary["floor"] = floor
|
||
summary["acceptanceVersion"] = mode
|
||
|
||
if mode == "v1":
|
||
# 回退态:预筛权威统一为四门投影(floor.pass),判定仍走驱动器证据(apply_gameplay_judge→judge_gameplay_floor)。
|
||
# 时序约束:v1 仅在 play-spec/驱动器仍挂载的波1~波2 前有效(波2 拆除后不可达)。
|
||
return await apply_gameplay_judge(summary, game_id=game_id, brief=brief,
|
||
evidence_dir=ev, prefilter=floor["pass"])
|
||
|
||
# v2 / shadow:跑测试员——仅预筛(四门)过时跑,省 ¥(预筛没过已被机械门拒,不花测试员钱)。
|
||
if floor["pass"]:
|
||
playtest = await run_playtest(game_id, brief=brief, evidence_dir=ev,
|
||
model_name=model or cfg["model"], port=port, cdp_port=cdp_port)
|
||
else:
|
||
playtest = {"accepted": False, "verdict": "reject", "pass": False, "degraded": False,
|
||
"reason": "四门地板未过,不跑测试员(省 ¥)", "problems": [], "feedback": "",
|
||
"summary": "floor-fail", "rolls": [], "rollCount": 0, "costRmb": 0.0, "model": cfg["model"],
|
||
"firstPlay": {"playableAtMs": None, "firstFeedbackMs": None, "loopClosed": False}}
|
||
summary["playtest"] = playtest
|
||
v2_accepted = bool(floor["pass"]) and bool(playtest.get("accepted")) and not bool(playtest.get("degraded"))
|
||
|
||
if mode == "shadow":
|
||
# 灰度:accepted 取旧口径(判定器 apply_gameplay_judge 写 judge 段);测试员只观测、并列落对照。
|
||
summary = await apply_gameplay_judge(summary, game_id=game_id, brief=brief,
|
||
evidence_dir=ev, prefilter=floor["pass"])
|
||
summary["shadowV2Accepted"] = v2_accepted # 对照:v2 生效时会是什么(批次账新旧口径对照表)
|
||
return summary
|
||
|
||
# mode == v2:测试员阻断放行。
|
||
blk = cfg["blocking"]
|
||
# rejectClasses:v2 测试员不产 broken/hollow/off_brief 细类,归一为 playtest_reject / tester_degraded(供 gate 桶归因)。
|
||
if playtest.get("degraded"):
|
||
reject_classes = ["tester_degraded"]
|
||
elif not v2_accepted:
|
||
reject_classes = ["playtest_reject"]
|
||
else:
|
||
reject_classes = []
|
||
summary["judge"] = { # §4 双写窗口:v2 下 judge 段镜像 playtest 裁决,喂 trace.gameplayJudge 与旧消费面
|
||
"ran": True, "blocking": blk, "verdict": playtest.get("verdict"),
|
||
"accepted": bool(playtest.get("accepted")), "rejectClasses": reject_classes,
|
||
"degraded": bool(playtest.get("degraded")), "reason": playtest.get("reason"),
|
||
"notes": playtest.get("summary"), "model": playtest.get("model"),
|
||
"costRmb": playtest.get("costRmb"), "acceptanceVersion": "v2",
|
||
}
|
||
summary["accepted"] = v2_accepted
|
||
summary["ok"] = v2_accepted if blk else floor["pass"]
|
||
if not v2_accepted: # 续修回喂来源(只引现象段);accepted 时无失败现象、不拼装(E2E 实测曾误拼自相矛盾文案)
|
||
summary["repairFeedback"] = _build_repair_feedback(floor, playtest)
|
||
if blk and not v2_accepted and playtest.get("degraded"):
|
||
summary.setdefault("failureReason", "tester_degraded") # §4 新增归因层
|
||
_writeback_verdict_json(ev, v2_accepted, summary["judge"])
|
||
# 最终验收后统一覆盖预先按旧九门生成的 failureLayer；CLI/Service/modify 三路由此共享同一权威归因。
|
||
return apply_acceptance_failure_attribution(summary)
|
||
except Exception as e: # noqa: BLE001 编排接线级异常:fail-closed(v2/shadow 不放行)。
|
||
summary["acceptanceVersion"] = summary.get("acceptanceVersion") or (mode or "v2")
|
||
summary["judge"] = {"ran": False, "degraded": True,
|
||
"reason": f"验收编排异常(fail-closed):{type(e).__name__}: {e}"}
|
||
summary["accepted"] = False
|
||
summary["ok"] = False
|
||
# v2 编排异常属于测试员/验收器降级，不得回落到旧 E/G/H 归因；shadow/v1 仍由 helper 保持兼容。
|
||
return apply_acceptance_failure_attribution(summary)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# 验收 v3 Python 编排核心(playtest/3 · 结构化目标 + 双 Judge 可信证据闭环)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
_PLAYTEST_V3_SCHEMA = "playtest/3"
|
||
_V3_OUTCOMES = {"accept", "reject", "inconclusive", "tester_error"}
|
||
_V3_INFRA_SUBTYPES = {
|
||
"environment_error", "runner_error", "runner_timeout", "image_error", "judge_error",
|
||
"browser_error", "port_error", "schema_error", "seed_mismatch", "unknown_schema",
|
||
"profile_contract_error", "provenance_error", "event_contract_error",
|
||
}
|
||
_V3_CONTRACT_DIR = Path(__file__).resolve().parents[1] / "contracts" / "play-loop"
|
||
# W-GOLD-LIVE 检查点 2：Python v3 编排产出 acceptance-request/3 —— 即 v2 字段集之上开口三个参照资产
|
||
# 可选字段（designRef/referenceAssetRecordIds/consumerRef，全 optional，旧调用方不传即旧行为）。canonical
|
||
# 校验（validate.py 语义层）钉死 /3 与 /2 同一条 proof registry 版本线（2026-07-15.v3 → proof-obligations.v2.json），
|
||
# 注册表版本与 schemaVersion 错配时 Writer 前身份必然校验失败，二者必须成对切换。普通路径的
|
||
# interactionBinding 保持 null；只有可信调用方显式选择 v3_shadow 校准 profile 时，才生成并贯通
|
||
# InteractionBinding/1。canonical hash 口径不变：仍是 canonical 字段集的稳定 JSON SHA-256。
|
||
# Node runner（playtest-v3.cdp.cjs validateProfileProvenance）已接受 /1、/2、/3 请求；非空 binding
|
||
# 通过独立文件与五项显式参数进入真浏览器路径，普通路径仍以 null 维持原行为。
|
||
_V3_OBLIGATIONS_FILE = _V3_CONTRACT_DIR / "proof-obligations.v2.json"
|
||
_V3_OBLIGATIONS_SCHEMA = _V3_CONTRACT_DIR / "proof-obligation-registry.schema.json"
|
||
_V3_SINGLE_ROLL_SCHEMA = _V3_CONTRACT_DIR / "single-roll-fact-v3.schema.json"
|
||
_V3_EVIDENCE_SCHEMA = _V3_CONTRACT_DIR / "playtest-evidence-v3.schema.json"
|
||
_V3_ACCEPTANCE_REQUEST_SCHEMA = _V3_CONTRACT_DIR / "acceptance-request-v3.schema.json"
|
||
_V3_ACCEPTANCE_PROVENANCE_SCHEMA = _V3_CONTRACT_DIR / "acceptance-provenance-v3.schema.json"
|
||
_V4_ACCEPTANCE_PROVENANCE_SCHEMA = _V3_CONTRACT_DIR / "acceptance-provenance-v4.schema.json"
|
||
_REFERENCE_ASSET_RECEIPT_SCHEMA = _V3_CONTRACT_DIR / "reference-asset-verification-receipt.schema.json"
|
||
_INTERACTION_PROFILE_REGISTRY_FILE = _V3_CONTRACT_DIR / "interaction-profiles.v2.json"
|
||
_INTERACTION_PROFILE_REGISTRY_SCHEMA = _V3_CONTRACT_DIR / "interaction-profile-registry-v2.schema.json"
|
||
_INTERACTION_BINDING_SCHEMA = _V3_CONTRACT_DIR / "interaction-binding.schema.json"
|
||
# Registry/1 仅供旧 acceptance 身份与 provenance 对账兼容；新生成入口的冻结消费统一走
|
||
# ReferenceAssetRegistry/2 + release 锚定预检，不得把本兼容器当作生产资产读取入口。
|
||
_V3_REFERENCE_ASSET_REGISTRY_FILE = _V3_CONTRACT_DIR / "reference-asset-registry.initial.json"
|
||
|
||
# 参照资产 v2 生产消费锚点。调用方只可选择 policyId；仓根、release 路径、release SHA 与冻结模式
|
||
# 全由本模块固定，避免 job/CLI 自报路径或 hash 绕过可信边界。
|
||
_REFERENCE_ASSET_POLICY_ID = "survivor-gold-v1"
|
||
_REFERENCE_ASSET_POLICY_MODE = "frozen_preflight"
|
||
_REFERENCE_ASSET_RELEASE_REF = "contracts/play-loop/reference-asset-release.initial.json"
|
||
_REFERENCE_ASSET_RELEASE_SHA256 = "506e181f688f082ea34bd7515e4b1c9952b30a1caad1a5799972ba2d7a3d27a5"
|
||
_REFERENCE_ASSET_TRUSTED_ROOT = Path(__file__).resolve().parents[1]
|
||
_V3_RESULTS_DIR = Path(__file__).resolve().parent / "results" / "acceptance-v3"
|
||
_V3_GENRES = {"narrative", "trpg", "heritage", "puzzle", "sim-business"}
|
||
_V3_GUARD_CHECKS = (
|
||
"floorPass", "requiredObligationsSatisfied", "referencesValid", "sameActionEvidence",
|
||
"noContradictions", "noBlockingProblems", "actorHealthy", "runnerHealthy", "judgeHealthy",
|
||
"judgeAccept",
|
||
)
|
||
_V3_RMB_QUANTUM = Decimal("0.00001")
|
||
|
||
|
||
def _finite_nonnegative_rmb_v3(value, field: str) -> float:
|
||
"""把成本入口收敛为有限非负浮点；布尔、NaN、Infinity 和负数一律拒绝。"""
|
||
if (not isinstance(value, (int, float)) or isinstance(value, bool)
|
||
or not math.isfinite(value) or value < 0):
|
||
raise ValueError(f"{field} 必须是有限非负金额")
|
||
return float(value)
|
||
|
||
|
||
def _finite_positive_rmb_v3(value, field: str) -> float:
|
||
"""成本硬帽必须为正有限数，零值不能把所有调用伪装成正常配置。"""
|
||
numeric = _finite_nonnegative_rmb_v3(value, field)
|
||
if numeric <= 0:
|
||
raise ValueError(f"{field} 必须是正有限金额")
|
||
return numeric
|
||
|
||
|
||
def _quantize_rmb_v3(value, field: str = "costRmb") -> float:
|
||
"""按十进制 ROUND_HALF_UP 统一保留五位，和 Node/contract 使用同一金额口径。"""
|
||
numeric = _finite_nonnegative_rmb_v3(value, field)
|
||
try:
|
||
return float(Decimal(str(numeric)).quantize(_V3_RMB_QUANTUM, rounding=ROUND_HALF_UP))
|
||
except (InvalidOperation, ValueError) as exc:
|
||
raise ValueError(f"{field} 无法按五位金额量化") from exc
|
||
|
||
|
||
def _acceptance_v3_cfg() -> dict:
|
||
"""读取 v3 编排旋钮；YAML 未落或纯单测环境时使用批准设计中的保守默认值。"""
|
||
try:
|
||
from worker import genconfig # noqa: PLC0415
|
||
g = genconfig.get
|
||
except Exception: # noqa: BLE001
|
||
def g(_area, _key, default):
|
||
return default
|
||
return {
|
||
"mode": str(g("acceptance", "mode", "v3_shadow")),
|
||
"model": str(g("acceptance_v3", "model", g("judge", "model", _JUDGE_DEFAULT_MODEL))),
|
||
"timeout_s": float(g("acceptance_v3", "timeout_s", 600.0)),
|
||
"cost_cap_rmb": float(g("acceptance_v3", "cost_cap_rmb", 1.5)),
|
||
"parent_chain_cost_cap_rmb": float(g("acceptance_v3", "parent_chain_cost_cap_rmb", 15.0)),
|
||
"second_roll": bool(g("acceptance_v3", "second_roll", True)),
|
||
"infra_retry": int(g("acceptance_v3", "infra_retry", 1)),
|
||
"run_lock_timeout_s": float(g("acceptance_v3", "run_lock_timeout_s", 30.0)),
|
||
"action_quantum_ms": int(g("acceptance_v3", "action_quantum_ms", 600)),
|
||
"wait_min_ms": int(g("acceptance_v3", "wait_min_ms", 100)),
|
||
"wait_max_ms": int(g("acceptance_v3", "wait_max_ms", 600)),
|
||
}
|
||
|
||
|
||
def _sha256_text(value: str) -> str:
|
||
"""生成协议指纹；统一使用十六进制 SHA-256，便于跨 Python/Node/Java 对账。"""
|
||
return hashlib.sha256((value or "").encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _is_sha256_v3(value) -> bool:
|
||
"""机械校验十六进制 SHA-256；证据引用不能只凭长度过关。"""
|
||
text = str(value or "")
|
||
return len(text) == 64 and all(ch in "0123456789abcdef" for ch in text)
|
||
|
||
|
||
def task_binding_hash_v3(trace_id) -> str:
|
||
"""把受信任务 traceId 冻结为验收身份；空白或非字符串一律拒绝。"""
|
||
if not isinstance(trace_id, str) or not trace_id.strip():
|
||
raise ValueError("v3 task traceId 必须是非空字符串")
|
||
return _sha256_text(trace_id)
|
||
|
||
|
||
def _is_safe_game_id_v3(value) -> bool:
|
||
"""限制证据目录分段与 runner 一致，阻止 gameId 路径穿越外部结果根。"""
|
||
text = str(value or "")
|
||
ascii_alnum = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
|
||
return bool(text and text[0] in ascii_alnum
|
||
and all(ch in ascii_alnum or ch in "._-" for ch in text))
|
||
|
||
|
||
# recordId 字符集（与 ReferenceAssetRecord/1.recordId 及 acceptance-request/3.$defs.recordId 同形）；
|
||
# 首字符额外允许下划线以兼容 _fewshot-*/_template-* 既有资产名。
|
||
_V3_RECORD_ID_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._-"
|
||
|
||
|
||
def _is_v3_record_id(value) -> bool:
|
||
"""机械校验参照资产 recordId 形状；非字符串、空串或含非法字符一律拒绝。"""
|
||
return isinstance(value, str) and bool(value) and all(ch in _V3_RECORD_ID_CHARS for ch in value)
|
||
|
||
|
||
_V3_LOCAL_RUN_LOCKS: dict[str, threading.Lock] = {}
|
||
_V3_LOCAL_RUN_LOCKS_GUARD = threading.Lock()
|
||
|
||
|
||
def _assert_no_symlink_components(path: Path, *, require_exists: bool) -> Path:
|
||
"""逐段拒绝用户 symlink，仅兼容 macOS 等系统的根目录直属受管别名。"""
|
||
lexical = Path(os.path.abspath(os.fspath(path)))
|
||
parts = lexical.parts[1:]
|
||
current = Path(lexical.anchor)
|
||
for index, part in enumerate(parts):
|
||
candidate = current / part
|
||
try:
|
||
info = candidate.lstat()
|
||
except FileNotFoundError:
|
||
if require_exists:
|
||
raise ValueError(f"证据路径不存在:{candidate}")
|
||
# 首个缺失分量之后不可能已有真实子项，保留剩余词法路径交 mkdir 创建。
|
||
for rest in parts[index:]:
|
||
current /= rest
|
||
return current
|
||
if stat.S_ISLNK(info.st_mode):
|
||
is_root_managed_alias = (current == Path(lexical.anchor)
|
||
and index < len(parts) - 1 and info.st_uid == 0)
|
||
if not is_root_managed_alias:
|
||
raise ValueError(f"证据路径禁止 symlink:{candidate}")
|
||
current = candidate.resolve(strict=True)
|
||
else:
|
||
current = candidate
|
||
return current
|
||
|
||
|
||
def _local_v3_run_lock(key: str) -> threading.Lock:
|
||
"""同进程协程/线程先经本地锁排队；跨进程再由 flock 仲裁。"""
|
||
with _V3_LOCAL_RUN_LOCKS_GUARD:
|
||
return _V3_LOCAL_RUN_LOCKS.setdefault(key, threading.Lock())
|
||
|
||
|
||
def _acquire_v3_run_lock(evidence_root: Path, run_dir: Path, timeout_s: float):
|
||
"""创建 run 目录并获取本地锁 + 跨进程 flock；超时返回空 claim 和可审计原因。"""
|
||
timeout_s = max(0.0, float(timeout_s or 0.0))
|
||
evidence_root = _assert_no_symlink_components(evidence_root, require_exists=False)
|
||
evidence_root.mkdir(parents=True, exist_ok=True)
|
||
evidence_root = _assert_no_symlink_components(evidence_root, require_exists=True)
|
||
run_dir = _assert_no_symlink_components(run_dir, require_exists=False)
|
||
if run_dir.parent != evidence_root:
|
||
raise ValueError("v3 runDir 必须是 evidenceRoot 的直接子目录")
|
||
run_dir.mkdir(parents=True, exist_ok=True)
|
||
run_dir = _assert_no_symlink_components(run_dir, require_exists=True)
|
||
local_lock = _local_v3_run_lock(str(run_dir))
|
||
started = time.monotonic()
|
||
if not local_lock.acquire(timeout=timeout_s):
|
||
return None, f"v3 run 本地锁等待超过 {timeout_s:.3f}s"
|
||
|
||
lock_fd = None
|
||
try:
|
||
flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
|
||
lock_fd = os.open(run_dir / ".run.lock", flags, 0o600)
|
||
if not stat.S_ISREG(os.fstat(lock_fd).st_mode):
|
||
raise ValueError("v3 run 锁文件不是普通文件")
|
||
deadline = started + timeout_s
|
||
while True:
|
||
try:
|
||
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
return (lock_fd, local_lock), None
|
||
except BlockingIOError:
|
||
if time.monotonic() >= deadline:
|
||
os.close(lock_fd)
|
||
lock_fd = None
|
||
local_lock.release()
|
||
return None, f"v3 run 跨进程锁等待超过 {timeout_s:.3f}s"
|
||
time.sleep(0.05)
|
||
except Exception:
|
||
if lock_fd is not None:
|
||
os.close(lock_fd)
|
||
local_lock.release()
|
||
raise
|
||
|
||
|
||
def _release_v3_run_lock(claim) -> None:
|
||
"""释放跨进程与同进程两层锁；锁文件保留作稳定 claim 点。"""
|
||
if not claim:
|
||
return
|
||
lock_fd, local_lock = claim
|
||
try:
|
||
fcntl.flock(lock_fd, fcntl.LOCK_UN)
|
||
finally:
|
||
os.close(lock_fd)
|
||
local_lock.release()
|
||
|
||
|
||
def _stable_json_hash_v3(value) -> str:
|
||
"""仅用于不含浮点的 Python 侧身份对象；game-event 必须使用 payloadCanonical。"""
|
||
encoded = json.dumps(
|
||
value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False,
|
||
).encode("utf-8")
|
||
return hashlib.sha256(encoded).hexdigest()
|
||
|
||
|
||
def _stable_json_bytes_v3(value) -> bytes:
|
||
"""生成 request/manifest 的稳定 JSON；这些身份对象协议禁止浮点。"""
|
||
return json.dumps(
|
||
value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False,
|
||
).encode("utf-8")
|
||
|
||
|
||
def _load_v3_validator():
|
||
"""加载仓内 canonical 校验器；不可加载属于 tester_error，调用方不得降级自判。"""
|
||
import importlib.util # noqa: PLC0415
|
||
|
||
validator_path = _V3_CONTRACT_DIR / "validate.py"
|
||
spec = importlib.util.spec_from_file_location("play_loop_validate_runtime", validator_path)
|
||
if spec is None or spec.loader is None:
|
||
raise RuntimeError("无法加载 contracts/play-loop/validate.py")
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
def _validate_v3_instance(schema_file: Path, instance: dict) -> list:
|
||
"""复用 canonical schema + semantic validator；返回空数组表示结构与跨引用均成立。"""
|
||
try:
|
||
module = _load_v3_validator()
|
||
schema = json.loads(Path(schema_file).read_text(encoding="utf-8"))
|
||
errors = module.validate(schema, instance, schema)
|
||
errors += module._semantic_validate(schema, instance)
|
||
return errors
|
||
except Exception as exc: # noqa: BLE001
|
||
return [f"canonical validator 不可用:{type(exc).__name__}: {exc}"]
|
||
|
||
|
||
def _validate_v3_schema_only(schema_file: Path, instance: dict) -> list:
|
||
"""只执行 canonical 结构校验；运行目录内 ref 的语义闭包由可信边界自行复算。"""
|
||
try:
|
||
module = _load_v3_validator()
|
||
schema = json.loads(Path(schema_file).read_text(encoding="utf-8"))
|
||
return module.validate(schema, instance, schema)
|
||
except Exception as exc: # noqa: BLE001
|
||
return [f"canonical validator 不可用:{type(exc).__name__}: {exc}"]
|
||
|
||
|
||
def _validate_v3_instance_parts(schema_file: Path, instance: dict) -> tuple[list, list]:
|
||
"""分别返回 schema 与 semantic 错误；guard 需据语义错误类型区分 reject/inconclusive/tester_error。"""
|
||
try:
|
||
module = _load_v3_validator()
|
||
schema = json.loads(Path(schema_file).read_text(encoding="utf-8"))
|
||
return module.validate(schema, instance, schema), module._semantic_validate(schema, instance)
|
||
except Exception as exc: # noqa: BLE001
|
||
return [f"canonical validator 不可用:{type(exc).__name__}: {exc}"], []
|
||
|
||
|
||
def _canonical_acceptance_request(identity: dict) -> dict:
|
||
"""返回写盘 request 的固定字段；acceptanceRequestHash 本身不入包，避免自引用 hash。
|
||
|
||
v3 字段集 = v2 十四字段（含 interactionBinding，普通路径为 null、显式 shadow 校准可为对象）+
|
||
三个参照资产可选字段
|
||
（designRef/referenceAssetRecordIds/consumerRef，未声明时分别为 null/[]/null）。字段集只增不改，
|
||
canonical hash 口径不变（稳定 JSON SHA-256）。
|
||
"""
|
||
fields = (
|
||
"schemaVersion", "gameId", "briefHash", "genre", "templateRoute", "proofProfileId",
|
||
"proofRegistryVersion", "taskBindingHash", "interactionBinding",
|
||
"sourceArtifactHash", "parentAcceptanceRequestHash", "repairOrdinal",
|
||
"designRef", "referenceAssetRecordIds", "consumerRef",
|
||
)
|
||
return {field: identity.get(field) for field in fields}
|
||
|
||
|
||
def _resolve_interaction_profile_v3(interaction_profile_id, resolved_proof: dict) -> dict:
|
||
"""读取 canonical 交互注册表，并把显式 profile 与当前 proof 路由逐项交叉核对。"""
|
||
if not isinstance(interaction_profile_id, str) or not interaction_profile_id:
|
||
raise ValueError("interactionProfileId 必须是非空字符串")
|
||
try:
|
||
registry = json.loads(_INTERACTION_PROFILE_REGISTRY_FILE.read_text(encoding="utf-8"))
|
||
except Exception as exc: # noqa: BLE001 注册表不可读时不得降级成普通 puzzle
|
||
raise ValueError(f"interaction profile 注册表不可读:{type(exc).__name__}: {exc}") from exc
|
||
registry_errors = _validate_v3_instance(_INTERACTION_PROFILE_REGISTRY_SCHEMA, registry)
|
||
if registry_errors:
|
||
raise ValueError("interaction profile 注册表非法:" + ";".join(registry_errors[:4]))
|
||
profiles = registry.get("profiles") if isinstance(registry.get("profiles"), dict) else {}
|
||
profile = profiles.get(interaction_profile_id)
|
||
if not isinstance(profile, dict) or profile.get("id") != interaction_profile_id:
|
||
raise ValueError(f"interactionProfileId 未在可信注册表中唯一登记:{interaction_profile_id}")
|
||
cross_checks = {
|
||
"genre": resolved_proof.get("genre"),
|
||
"templateRoute": resolved_proof.get("templateRoute"),
|
||
"proofProfileId": resolved_proof.get("proofProfileId"),
|
||
}
|
||
for field, expected in cross_checks.items():
|
||
if profile.get(field) != expected:
|
||
raise ValueError(
|
||
f"interactionProfileId 与当前 {field} 不一致:{profile.get(field)} != {expected}")
|
||
prompt = ((profile.get("actorProtocol") or {}).get("prompt")
|
||
if isinstance(profile.get("actorProtocol"), dict) else None)
|
||
if not isinstance(prompt, dict) or prompt.get("runtimeEligible") is not False:
|
||
raise ValueError("Match-3 shadow profile 必须保持 runtimeEligible=false")
|
||
return {"registry": registry, "profile": profile}
|
||
|
||
|
||
def _build_interaction_binding_v3(interaction_profile_id, task_binding_hash: str,
|
||
resolved_proof: dict) -> dict:
|
||
"""按 InteractionBinding/1 canonical 前缀与稳定 JSON 生成任务绑定。"""
|
||
resolved_interaction = _resolve_interaction_profile_v3(interaction_profile_id, resolved_proof)
|
||
content = {
|
||
"interactionProfileId": interaction_profile_id,
|
||
"interactionRegistryVersion": resolved_interaction["registry"]["registryVersion"],
|
||
"taskBindingHash": task_binding_hash,
|
||
}
|
||
binding = {
|
||
"schemaVersion": "InteractionBinding/1",
|
||
**content,
|
||
"interactionBindingHash": hashlib.sha256(
|
||
b"interaction-binding/1\n" + _stable_json_bytes_v3(content)
|
||
).hexdigest(),
|
||
}
|
||
binding_errors = _validate_v3_schema_only(_INTERACTION_BINDING_SCHEMA, binding)
|
||
if binding_errors:
|
||
raise ValueError("interactionBinding 非法:" + ";".join(binding_errors[:4]))
|
||
return binding
|
||
|
||
|
||
def preflight_interaction_profile_v3(interaction_profile_id, acceptance_mode, *, genre: str,
|
||
template_route: str, proof_profile_id=None) -> dict | None:
|
||
"""在 scaffold/Writer 前验证显式 profile；仅 v3_shadow 校准路径允许启用。"""
|
||
if interaction_profile_id is None:
|
||
return None
|
||
if acceptance_mode != "v3_shadow":
|
||
raise ValueError("显式 interactionProfileId 仅允许 v3_shadow 校准路径")
|
||
resolved_proof = _v3_resolve_registry(
|
||
genre, "", template_route=template_route, proof_profile_id=proof_profile_id)
|
||
resolved_interaction = _resolve_interaction_profile_v3(interaction_profile_id, resolved_proof)
|
||
return {
|
||
"interactionProfileId": interaction_profile_id,
|
||
"interactionRegistryVersion": resolved_interaction["registry"]["registryVersion"],
|
||
"proofProfileId": resolved_proof["proofProfileId"],
|
||
"genre": resolved_proof["genre"],
|
||
"templateRoute": resolved_proof["templateRoute"],
|
||
}
|
||
|
||
|
||
def build_acceptance_v3_identity(game_id: str, brief: str, *, genre: str, template_route: str,
|
||
source_artifact_hash=None, parent_acceptance_request_hash=None,
|
||
repair_ordinal: int = 0, proof_profile_id=None,
|
||
proof_registry_version=None, task_binding_hash=None,
|
||
interaction_profile_id=None, interaction_binding=None,
|
||
design_ref=None, reference_asset_record_ids=None,
|
||
consumer_ref=None) -> dict:
|
||
"""Writer 前冻结验收身份；profile 只能由可信 templateRoute 选中，修复链不得重选。
|
||
|
||
W-GOLD-LIVE 检查点 2 起产出 acceptance-request/3：v2 字段集之上新增三个可选参照资产字段——
|
||
design_ref（已批准 designIntent 引用）、reference_asset_record_ids（声明消费的参照资产 recordId
|
||
列表）、consumer_ref（消费者身份）。三者全部可选：旧调用方不传即 null/[]/null，与 v2 行为一致；
|
||
是否真能消费由消费对账（reconcile_v3_reference_asset_consumption）在编排接线处强制。
|
||
"""
|
||
repair_ordinal = int(repair_ordinal)
|
||
if repair_ordinal not in (0, 1):
|
||
raise ValueError("repairOrdinal 只能是 0 或 1")
|
||
if source_artifact_hash is not None and not _is_sha256_v3(source_artifact_hash):
|
||
raise ValueError("sourceArtifactHash 必须是 SHA-256 或 null")
|
||
if not _is_sha256_v3(task_binding_hash):
|
||
raise ValueError("taskBindingHash 必须是当前任务 traceId 的 SHA-256")
|
||
if repair_ordinal == 0:
|
||
if source_artifact_hash is not None or parent_acceptance_request_hash is not None:
|
||
raise ValueError("首轮 acceptance identity 的 sourceArtifactHash/parentAcceptanceRequestHash 必须为 null")
|
||
elif not _is_sha256_v3(parent_acceptance_request_hash) or not _is_sha256_v3(source_artifact_hash):
|
||
raise ValueError("修复 identity 必须固定 parentAcceptanceRequestHash 与上一产物 sourceArtifactHash")
|
||
# 三个可选参照资产字段归一化：缺省 ≡ 未声明（向后兼容旧路径）；声明则必须形状合法。
|
||
if design_ref is not None and (not isinstance(design_ref, str) or not design_ref):
|
||
raise ValueError("designRef 必须是非空字符串或 null")
|
||
if consumer_ref is not None and (not isinstance(consumer_ref, str) or not consumer_ref):
|
||
raise ValueError("consumerRef 必须是非空字符串或 null")
|
||
if reference_asset_record_ids is None:
|
||
reference_asset_record_ids = []
|
||
if not isinstance(reference_asset_record_ids, list):
|
||
raise ValueError("referenceAssetRecordIds 必须是 recordId 列表或 null")
|
||
for record_id in reference_asset_record_ids:
|
||
if not _is_v3_record_id(record_id):
|
||
raise ValueError(f"referenceAssetRecordIds 含非法 recordId:{record_id!r}")
|
||
|
||
resolved = _v3_resolve_registry(
|
||
genre, brief, template_route=template_route, proof_profile_id=proof_profile_id,
|
||
proof_registry_version=proof_registry_version,
|
||
)
|
||
if interaction_profile_id is not None and interaction_binding is not None:
|
||
raise ValueError("interaction_profile_id 与 interaction_binding 不能同时提供")
|
||
if interaction_profile_id is not None:
|
||
canonical_interaction_binding = _build_interaction_binding_v3(
|
||
interaction_profile_id, task_binding_hash, resolved)
|
||
elif interaction_binding is not None:
|
||
if not isinstance(interaction_binding, dict):
|
||
raise ValueError("interactionBinding 必须是对象或 null")
|
||
canonical_interaction_binding = _build_interaction_binding_v3(
|
||
interaction_binding.get("interactionProfileId"), task_binding_hash, resolved)
|
||
if interaction_binding != canonical_interaction_binding:
|
||
raise ValueError("interactionBinding 与当前 task/profile canonical 内容不一致")
|
||
else:
|
||
canonical_interaction_binding = None
|
||
core = {
|
||
"schemaVersion": "acceptance-request/3",
|
||
"gameId": str(game_id),
|
||
"briefHash": _sha256_text(brief or ""),
|
||
"genre": resolved["genre"],
|
||
"templateRoute": resolved["templateRoute"],
|
||
"proofProfileId": resolved["proofProfileId"],
|
||
"proofRegistryVersion": resolved["version"],
|
||
"taskBindingHash": task_binding_hash,
|
||
# 普通 puzzle 保持显式 null；仅可信调用方显式选择 shadow profile 时写入 canonical binding。
|
||
"interactionBinding": canonical_interaction_binding,
|
||
"sourceArtifactHash": source_artifact_hash,
|
||
"parentAcceptanceRequestHash": parent_acceptance_request_hash,
|
||
"repairOrdinal": repair_ordinal,
|
||
"designRef": design_ref,
|
||
"referenceAssetRecordIds": list(reference_asset_record_ids),
|
||
"consumerRef": consumer_ref,
|
||
}
|
||
request_errors = _validate_v3_instance(_V3_ACCEPTANCE_REQUEST_SCHEMA, core)
|
||
if request_errors:
|
||
raise ValueError("acceptance request 非法:" + ";".join(request_errors[:4]))
|
||
return {**core, "acceptanceRequestHash": hashlib.sha256(_stable_json_bytes_v3(core)).hexdigest()}
|
||
|
||
|
||
def _write_json_bytes_atomic(path: Path, payload: dict) -> str:
|
||
"""原子写 canonical JSON 并返回字节 hash；用于可信 request 与 provenance manifest。"""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
data = _stable_json_bytes_v3(payload)
|
||
tmp = path.with_name(path.name + ".tmp")
|
||
tmp.write_bytes(data)
|
||
tmp.replace(path)
|
||
return hashlib.sha256(data).hexdigest()
|
||
|
||
|
||
def _write_receipt_json_atomic(path: Path, data: bytes) -> None:
|
||
"""用不可预测的 O_EXCL/no-follow 临时普通文件原子发布验收回执。"""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
no_follow = getattr(os, "O_NOFOLLOW", None)
|
||
if no_follow is None:
|
||
raise ValueError("当前平台不支持 no-follow 回执临时文件")
|
||
temp_path = None
|
||
temp_fd = None
|
||
try:
|
||
# 临时名不复用目标名，避免攻击者预置固定 .tmp symlink 诱导写到 run 外部。
|
||
for _ in range(8):
|
||
candidate = path.parent / f".{path.name}.{secrets.token_hex(16)}.tmp"
|
||
try:
|
||
temp_fd = os.open(
|
||
candidate,
|
||
os.O_WRONLY | os.O_CREAT | os.O_EXCL | no_follow,
|
||
0o600,
|
||
)
|
||
except FileExistsError:
|
||
continue
|
||
temp_path = candidate
|
||
break
|
||
if temp_fd is None or temp_path is None:
|
||
raise ValueError("无法创建不可预测的回执临时文件")
|
||
if not stat.S_ISREG(os.fstat(temp_fd).st_mode):
|
||
raise ValueError("回执临时文件不是普通文件")
|
||
with os.fdopen(temp_fd, "wb") as handle:
|
||
temp_fd = None
|
||
handle.write(data)
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
# 同目录 rename 保证回执对 runner 只呈现完整 canonical 字节。
|
||
os.replace(temp_path, path)
|
||
temp_path = None
|
||
finally:
|
||
if temp_fd is not None:
|
||
os.close(temp_fd)
|
||
if temp_path is not None:
|
||
try:
|
||
temp_path.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def write_acceptance_v3_provenance(game_id: str, identity: dict, *, artifact_hash: str,
|
||
evidence_root, artifact_path=None,
|
||
consumed_reference_assets=None,
|
||
reference_asset_verification_receipts=None) -> dict:
|
||
"""stage 后在 gameDir 外写 request/manifest；证据不得成为浏览器可加载却不入 artifactHash 的旁路。
|
||
|
||
W-GOLD-LIVE 检查点 2 起产出 acceptance-provenance/3：逐字段镜像 v3 request（含 interactionBinding
|
||
与三个参照资产可选字段），并落 consumed_reference_assets——本次实际消费的参照资产对账快照
|
||
（[{recordId,role,artifactHash}]，冻结消费时刻的 role/hash，注册表后续改动不覆盖历史 provenance）。
|
||
快照只许是请求侧 referenceAssetRecordIds 的子集：声明外消费即拒绝，保证请求/来源对账对应。
|
||
"""
|
||
if not _is_sha256_v3(artifact_hash):
|
||
raise ValueError("artifactHash 必须是 SHA-256")
|
||
artifact_root = (Path(artifact_path) if artifact_path else cheap_run.wg1_game_dir(game_id)).resolve()
|
||
if not artifact_root.is_dir():
|
||
raise ValueError(f"staged 产物不存在:{artifact_root}")
|
||
evidence_dir = Path(evidence_root).resolve()
|
||
if evidence_dir == artifact_root or artifact_root in evidence_dir.parents:
|
||
raise ValueError("v3 evidence/provenance 必须位于 staged gameDir 外")
|
||
request_payload = _canonical_acceptance_request(identity)
|
||
expected_request_hash = str(identity.get("acceptanceRequestHash") or "")
|
||
actual_request_hash = hashlib.sha256(_stable_json_bytes_v3(request_payload)).hexdigest()
|
||
if actual_request_hash != expected_request_hash:
|
||
raise ValueError("acceptance identity 在 Writer 后发生漂移")
|
||
request_file = evidence_dir / f"acceptance-request-{expected_request_hash}.json"
|
||
written_request_hash = _write_json_bytes_atomic(request_file, request_payload)
|
||
if written_request_hash != expected_request_hash:
|
||
raise ValueError("acceptance request 写盘 hash 不一致")
|
||
interaction_binding = request_payload.get("interactionBinding")
|
||
interaction_binding_file = None
|
||
interaction_binding_file_hash = None
|
||
if interaction_binding is not None:
|
||
interaction_binding_file = evidence_dir / "interaction-binding.json"
|
||
interaction_binding_file_hash = _write_json_bytes_atomic(
|
||
interaction_binding_file, interaction_binding)
|
||
# 消费对账快照形状先做声明子集闸；role/artifactHash 的机械形状由下方 canonical provenance schema 兜底。
|
||
consumed = list(consumed_reference_assets or [])
|
||
declared_ids = set(request_payload.get("referenceAssetRecordIds") or [])
|
||
for entry in consumed:
|
||
if (not isinstance(entry, dict) or not _is_v3_record_id(entry.get("recordId"))
|
||
or entry["recordId"] not in declared_ids):
|
||
raise ValueError(f"consumedReferenceAssets 含声明外消费:{entry!r}")
|
||
receipt_refs = None
|
||
if reference_asset_verification_receipts is not None:
|
||
receipts = list(reference_asset_verification_receipts)
|
||
if not receipts:
|
||
raise ValueError("acceptance /4 必须带至少一份验收自产回执")
|
||
receipt_ids = []
|
||
for index, receipt in enumerate(receipts):
|
||
shape_errors = _validate_v3_schema_only(_REFERENCE_ASSET_RECEIPT_SCHEMA, receipt)
|
||
if shape_errors:
|
||
raise ValueError(
|
||
f"验收自产回执[{index}] 形状非法:" + ";".join(shape_errors[:4]))
|
||
receipt_id = receipt.get("receiptId")
|
||
if not _is_v3_record_id(receipt_id) or receipt_id in receipt_ids:
|
||
raise ValueError("验收自产回执 receiptId 非法或重复")
|
||
receipt_ids.append(receipt_id)
|
||
receipt_records = {receipt["recordId"] for receipt in receipts}
|
||
consumed_records = {entry.get("recordId") for entry in consumed if isinstance(entry, dict)}
|
||
if receipt_records != declared_ids or receipt_records != consumed_records:
|
||
raise ValueError("验收自产回执、声明与消费快照 recordId 集合不闭合")
|
||
receipts_by_record = {receipt["recordId"]: receipt for receipt in receipts}
|
||
if len({receipt["finalSnapshotHash"] for receipt in receipts}) != 1:
|
||
raise ValueError("验收自产回执 finalSnapshotHash 不一致")
|
||
for entry in consumed:
|
||
receipt = receipts_by_record[entry["recordId"]]
|
||
if (receipt["consumerRef"] != request_payload.get("consumerRef")
|
||
or entry.get("role") != receipt["role"]
|
||
or entry.get("artifactHash") != receipt["expected"]["artifactHash"]):
|
||
raise ValueError("验收自产回执与 consumerRef/消费快照不一致")
|
||
|
||
receipt_dir = evidence_dir / "reference-asset-verification-receipts"
|
||
if os.path.lexists(receipt_dir):
|
||
receipt_stat = os.lstat(receipt_dir)
|
||
if stat.S_ISLNK(receipt_stat.st_mode) or not stat.S_ISDIR(receipt_stat.st_mode):
|
||
raise ValueError("验收回执目录必须是当前 run 内普通目录且不得为 symlink")
|
||
else:
|
||
receipt_dir.mkdir(parents=True)
|
||
if receipt_dir.resolve().parent != evidence_dir:
|
||
raise ValueError("验收回执目录越出当前 run")
|
||
|
||
receipt_refs = []
|
||
for receipt in receipts:
|
||
receipt_file = receipt_dir / f"{receipt['receiptId']}.json"
|
||
receipt_bytes = _stable_json_bytes_v3(receipt)
|
||
_write_receipt_json_atomic(receipt_file, receipt_bytes)
|
||
receipt_refs.append({
|
||
"ref": receipt_file.relative_to(evidence_dir).as_posix(),
|
||
"hash": hashlib.sha256(receipt_bytes).hexdigest(),
|
||
})
|
||
|
||
manifest = {
|
||
"schemaVersion": ("acceptance-provenance/4" if receipt_refs is not None
|
||
else "acceptance-provenance/3"),
|
||
**{key: request_payload[key] for key in (
|
||
"gameId", "briefHash", "genre", "templateRoute", "proofProfileId", "proofRegistryVersion",
|
||
"taskBindingHash", "interactionBinding",
|
||
"sourceArtifactHash", "parentAcceptanceRequestHash", "repairOrdinal",
|
||
"designRef", "referenceAssetRecordIds", "consumerRef",
|
||
)},
|
||
"acceptanceRequestHash": expected_request_hash,
|
||
"artifactHash": artifact_hash,
|
||
"consumedReferenceAssets": consumed,
|
||
**({"referenceAssetVerificationReceipts": receipt_refs}
|
||
if receipt_refs is not None else {}),
|
||
}
|
||
manifest_errors = (_validate_v3_schema_only(_V4_ACCEPTANCE_PROVENANCE_SCHEMA, manifest)
|
||
if receipt_refs is not None
|
||
else _validate_v3_instance(_V3_ACCEPTANCE_PROVENANCE_SCHEMA, manifest))
|
||
if manifest_errors:
|
||
raise ValueError("acceptance provenance 非法:" + ";".join(manifest_errors[:4]))
|
||
manifest_file = evidence_dir / "acceptance-provenance.json"
|
||
manifest_hash = _write_json_bytes_atomic(manifest_file, manifest)
|
||
return {
|
||
"acceptanceRequestFile": str(request_file),
|
||
"acceptanceRequestHash": expected_request_hash,
|
||
"provenanceManifestFile": str(manifest_file),
|
||
"provenanceManifestHash": manifest_hash,
|
||
"interactionBindingFile": (str(interaction_binding_file)
|
||
if interaction_binding_file is not None else None),
|
||
"interactionBindingFileHash": interaction_binding_file_hash,
|
||
"manifest": manifest,
|
||
}
|
||
|
||
|
||
# 参照资产消费角色枚举，与 ReferenceAssetRecord/1.role 一致；role 与 lifecycleStatus 正交，
|
||
# candidate/migration_pending 不是第五种角色。
|
||
_V3_REFERENCE_ASSET_ROLES = frozenset(
|
||
("harness_fixture", "prompt_eval_gold", "generation_exemplar", "game_content_gold"))
|
||
|
||
|
||
def _load_v3_reference_asset_registry(registry_file=None) -> dict:
|
||
"""读取参照资产持久注册表（当前阶段为 migration-list 初始快照）。
|
||
|
||
注册表不可读必须响亮失败，绝不能被当成『无记录』放行或拒绝消费——读不到注册表时
|
||
任何消费判定都失去依据，交由调用方按 fail-closed 处理。
|
||
"""
|
||
path = Path(registry_file) if registry_file else _V3_REFERENCE_ASSET_REGISTRY_FILE
|
||
try:
|
||
registry = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception as exc: # noqa: BLE001 —— 注册表缺失/损坏是接线错误，不得静默
|
||
raise ValueError(f"参照资产注册表不可读:{type(exc).__name__}: {exc}") from exc
|
||
if not isinstance(registry, dict) or not isinstance(registry.get("records"), list):
|
||
raise ValueError("参照资产注册表缺 records 列表")
|
||
return registry
|
||
|
||
|
||
def reconcile_v3_reference_asset_consumption(consumptions, *, consumer_ref=None,
|
||
registry=None, registry_file=None) -> dict:
|
||
"""参照资产消费对账（纯本地，金标 SoT §7）：逐条消费声明过六项闸，全过才允许读取资产。
|
||
|
||
六项闸（口径 = 金标 SoT『消费者按 recordId 对账 role + consumerRef + artifactHash 后才能读取资产』）：
|
||
① recordId 在注册表存在；
|
||
② lifecycleStatus==active——migration_pending/candidate/retired 一律拒绝（参照资产未激活不得消费）；
|
||
③ role 与消费场景匹配（声明了期望 role 时，记录登记 role 必须一致）；
|
||
④ consumerRef 登记闸：记录必须已登记 consumerRef；调用方报上 consumer_ref 时还必须与登记一致；
|
||
⑤ 版本门：调用方报上消费时实际制品 hash 时，必须与记录 artifactHash 一致（hash 漂移拒绝）；
|
||
⑥ 缺维度门：role==game_content_gold 的记录必须 designRef 非空（缺已批准 designIntent 拒绝）。
|
||
|
||
consumptions：消费声明列表，元素为 recordId 字符串或 {recordId, role?, artifactHash?}；
|
||
consumed 快照冻结消费时刻记录里的 role/artifactHash，注册表后续改动不覆盖历史对账留痕。
|
||
返回 {ok, consumed:[{recordId,role,artifactHash}], errors:[...]}。
|
||
"""
|
||
if registry is None:
|
||
registry = _load_v3_reference_asset_registry(registry_file)
|
||
by_id = {}
|
||
for record in registry.get("records") or []:
|
||
if isinstance(record, dict):
|
||
record_id = record.get("recordId")
|
||
if _is_v3_record_id(record_id) and record_id not in by_id:
|
||
by_id[record_id] = record
|
||
consumed = []
|
||
errors = []
|
||
for item in consumptions or []:
|
||
if isinstance(item, str):
|
||
item = {"recordId": item}
|
||
if not isinstance(item, dict):
|
||
errors.append(f"消费声明必须是 recordId 字符串或对象:{item!r}")
|
||
continue
|
||
record_id = item.get("recordId")
|
||
if not _is_v3_record_id(record_id):
|
||
errors.append(f"消费声明 recordId 非法:{record_id!r}")
|
||
continue
|
||
record = by_id.get(record_id)
|
||
# ① 存在闸
|
||
if record is None:
|
||
errors.append(f"参照资产记录不存在:{record_id}")
|
||
continue
|
||
# ② 激活闸：只有 active 可消费（未激活记录的占位 hash 不得进入任何消费快照）
|
||
status = record.get("lifecycleStatus")
|
||
if status != "active":
|
||
errors.append(f"参照资产未激活不得消费:{record_id}(lifecycleStatus={status or 'unknown'})")
|
||
continue
|
||
# ③ role 闸：角色非法或与消费场景声明不符均拒绝
|
||
role = record.get("role")
|
||
if role not in _V3_REFERENCE_ASSET_ROLES:
|
||
errors.append(f"参照资产 role 非法:{record_id}(role={role or 'unknown'})")
|
||
continue
|
||
expected_role = item.get("role")
|
||
if expected_role is not None and expected_role != role:
|
||
errors.append(
|
||
f"参照资产 role 与消费场景不匹配:{record_id}(登记={role} 声明={expected_role})")
|
||
continue
|
||
# ④ consumerRef 闸：active 记录必须已登记消费方；调用方报上身份时必须与登记一致
|
||
record_consumer = record.get("consumerRef")
|
||
if not (isinstance(record_consumer, str) and record_consumer):
|
||
errors.append(f"参照资产 consumerRef 未登记:{record_id}")
|
||
continue
|
||
if consumer_ref is not None and record_consumer != consumer_ref:
|
||
errors.append(
|
||
f"参照资产 consumerRef 与登记不符:{record_id}(登记={record_consumer} 消费方={consumer_ref})")
|
||
continue
|
||
# ⑤ 版本门：冻结消费时刻制品身份，登记 hash 与实际 hash 漂移即拒绝
|
||
record_hash = record.get("artifactHash")
|
||
actual_hash = item.get("artifactHash")
|
||
if actual_hash is not None and actual_hash != record_hash:
|
||
errors.append(f"参照资产制品 hash 漂移拒绝:{record_id}(登记={record_hash} 消费时={actual_hash})")
|
||
continue
|
||
# ⑥ 缺维度门：完整内容金标必须指向已批准 designIntent，缺维度不得被消费
|
||
if role == "game_content_gold":
|
||
design_refs = record.get("designRef")
|
||
if not (isinstance(design_refs, list) and design_refs):
|
||
errors.append(f"game_content_gold 缺 designRef 维度拒绝:{record_id}")
|
||
continue
|
||
consumed.append({"recordId": record_id, "role": role, "artifactHash": str(record_hash or "")})
|
||
return {"ok": not errors, "consumed": consumed, "errors": errors}
|
||
|
||
|
||
def _v3_check_declared_reference_assets(identity: dict) -> tuple:
|
||
"""编排接线处的消费对账闸：返回 (拒绝原因, 消费快照)；identity 声明消费必须全部过六项对账。
|
||
|
||
Registry/1 对账仅供旧 acceptance 请求与 provenance 兼容。未声明消费时返回 (None, []) 正常放行；
|
||
对账成功时同时返回消费时刻快照 [{recordId,role,artifactHash}]，由调用方写入
|
||
provenance manifest 的 consumedReferenceAssets 完成消费溯源闭环（快照出自验收时点对账，零信任，
|
||
不接受生成侧自报）。新生成入口不得用本兼容器读取资产，统一走 release 锚定冻结预检。
|
||
"""
|
||
declared = identity.get("referenceAssetRecordIds") or []
|
||
if not declared:
|
||
return None, []
|
||
try:
|
||
reconciled = reconcile_v3_reference_asset_consumption(
|
||
declared, consumer_ref=identity.get("consumerRef"))
|
||
except ValueError as exc: # noqa: BLE001 —— 注册表不可读按 fail-closed 拒绝消费
|
||
return f"参照资产消费对账失败:{exc}", []
|
||
if reconciled["ok"]:
|
||
return None, reconciled["consumed"]
|
||
return "参照资产消费对账拒绝:" + ";".join(reconciled["errors"][:4]), []
|
||
|
||
|
||
def preflight_reference_asset_policy(policy_id, acceptance_mode):
|
||
"""在 Writer 前按生产内置锚点验证显式冻结策略；未选择策略时返回 ``None``。
|
||
|
||
调用方唯一可控输入是 ``policy_id`` 与当前 acceptance mode。可信仓根、release 路径、release SHA、
|
||
gate mode 均由本模块固定；任何未知策略、非 shadow 模式或资产漂移直接抛错，不做降级回落。
|
||
"""
|
||
if policy_id is None:
|
||
return None
|
||
if acceptance_mode == "v3":
|
||
raise ValueError("v3 live 禁止使用冻结参照资产 policy")
|
||
if acceptance_mode != "v3_shadow":
|
||
raise ValueError("冻结参照资产 policy 仅允许 v3_shadow")
|
||
|
||
import reference_asset_gate # noqa: PLC0415 仅显式策略路径加载可信消费门
|
||
|
||
return reference_asset_gate.verify_policy(
|
||
policy_id,
|
||
release_ref=_REFERENCE_ASSET_RELEASE_REF,
|
||
expected_release_hash=_REFERENCE_ASSET_RELEASE_SHA256,
|
||
trusted_root=_REFERENCE_ASSET_TRUSTED_ROOT,
|
||
mode=_REFERENCE_ASSET_POLICY_MODE,
|
||
)
|
||
|
||
|
||
def _reverify_acceptance_reference_asset_policy(
|
||
request: dict, identity: dict, acceptance_mode: str) -> tuple:
|
||
"""验收侧独立复验 policy，并与 Writer 前 generation receipts 做 canonical 全等对账。"""
|
||
import reference_asset_gate # noqa: PLC0415 验收只消费统一 gate 的公开结果与 canonical 编码
|
||
|
||
policy_present = "referenceAssetPolicyId" in request
|
||
receipts_present = "referenceAssetGenerationReceipts" in request
|
||
if any(key.startswith("referenceAssetVerificationReceipt") for key in request):
|
||
return "请求不得自报验收回执 ref/hash", None, []
|
||
if policy_present != receipts_present:
|
||
return "referenceAssetPolicyId/referenceAssetGenerationReceipts 必须成对出现", None, []
|
||
if not policy_present:
|
||
return None, None, []
|
||
|
||
policy_id = request.get("referenceAssetPolicyId")
|
||
generation_receipts = request.get("referenceAssetGenerationReceipts")
|
||
if not isinstance(policy_id, str) or not policy_id:
|
||
return "referenceAssetPolicyId 必须是非空字符串", None, []
|
||
if (not isinstance(generation_receipts, list) or not generation_receipts
|
||
or any(not isinstance(receipt, dict) for receipt in generation_receipts)):
|
||
return "referenceAssetGenerationReceipts 必须是非空 JSON 对象数组", None, []
|
||
for index, receipt in enumerate(generation_receipts):
|
||
shape_errors = _validate_v3_schema_only(_REFERENCE_ASSET_RECEIPT_SCHEMA, receipt)
|
||
if shape_errors:
|
||
return (f"referenceAssetGenerationReceipts[{index}] 形状非法:"
|
||
+ ";".join(shape_errors[:4])), None, []
|
||
try:
|
||
generation_bytes = reference_asset_gate.canonical_json_bytes(generation_receipts)
|
||
verified = preflight_reference_asset_policy(policy_id, acceptance_mode)
|
||
acceptance_receipts = reference_asset_gate.to_json_value(verified.receipts)
|
||
acceptance_bytes = reference_asset_gate.canonical_json_bytes(acceptance_receipts)
|
||
except Exception as exc: # noqa: BLE001 任一锚点或资产漂移都必须转为稳定 provenance_error
|
||
return f"验收侧参照资产独立复验失败:{type(exc).__name__}: {exc}", None, []
|
||
if generation_bytes != acceptance_bytes:
|
||
return "generation receipts 与验收独立复验 canonical 回执不一致", None, []
|
||
|
||
record_ids = [receipt["recordId"] for receipt in acceptance_receipts]
|
||
consumers = {receipt["consumerRef"] for receipt in acceptance_receipts}
|
||
snapshots = {receipt["finalSnapshotHash"] for receipt in acceptance_receipts}
|
||
declared = identity.get("referenceAssetRecordIds") or []
|
||
if len(record_ids) != len(set(record_ids)) or set(declared) != set(record_ids):
|
||
return "acceptance identity 的 referenceAssetRecordIds 与验收回执集合不一致", None, []
|
||
if len(consumers) != 1 or identity.get("consumerRef") not in consumers:
|
||
return "acceptance identity 的 consumerRef 与验收回执不一致", None, []
|
||
if len(snapshots) != 1:
|
||
return "验收回执 finalSnapshotHash 不一致", None, []
|
||
consumed = [{
|
||
"recordId": receipt["recordId"],
|
||
"role": receipt["role"],
|
||
"artifactHash": receipt["expected"]["artifactHash"],
|
||
} for receipt in acceptance_receipts]
|
||
return None, acceptance_receipts, consumed
|
||
|
||
|
||
def build_frozen_reference_asset_constraint_block(verified) -> str:
|
||
"""只用验证器冻结结果生成 Writer 约束，不接收 identity 自报字段。"""
|
||
import reference_asset_gate # noqa: PLC0415 与预检结果类型保持单一来源
|
||
|
||
if not isinstance(verified, reference_asset_gate.VerifiedReferenceAssets):
|
||
raise TypeError("参照资产约束只接受 VerifiedReferenceAssets")
|
||
|
||
lines = [
|
||
"【冻结参照资产约束】",
|
||
f"policyId={_REFERENCE_ASSET_POLICY_ID}",
|
||
f"mode={_REFERENCE_ASSET_POLICY_MODE}",
|
||
f"snapshotHash={verified.snapshot_hash}",
|
||
"以下文件是本次 Writer 唯一可读的参照资产快照；只能 read_file/list_dir，禁止写入或读取活目录：",
|
||
]
|
||
for record in verified.records:
|
||
lines.append(f"record={record.get('recordId')} role={record.get('role')}")
|
||
for receipt in verified.receipts:
|
||
lines.append(
|
||
f"receipt={receipt.get('receiptId')} finalSnapshotHash={receipt.get('finalSnapshotHash')}"
|
||
)
|
||
for record_id in sorted(verified.reference_roots):
|
||
roots = ",".join(verified.reference_roots[record_id])
|
||
lines.append(f"roots[{record_id}]={roots}")
|
||
for path in sorted(verified.reference_files, key=lambda item: item.encode("utf-8")):
|
||
content = verified.reference_files[path]
|
||
lines.append(f"- {path} size={len(content)} sha256={hashlib.sha256(content).hexdigest()}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def build_v3_reference_asset_generation_constraints(identity: dict, registry=None) -> dict | None:
|
||
"""旧 acceptance 兼容注入门：Registry/1 对账通过 → 约束块；未声明 → None；失败 → ValueError。
|
||
|
||
本函数只保留给旧 acceptance identity 回放，不是 v2 生产消费入口：
|
||
· referenceAssetRecordIds 为空 ≡ 未声明消费 ≡ 旧路径，返回 None，生成行为逐字节不变；
|
||
· 声明消费 → 逐条过 reconcile_v3_reference_asset_consumption 六项闸，任一失败抛 ValueError（调用方在
|
||
生成入口响亮拒绝）；
|
||
· 对账通过（仅记录签认 active 后才可能）→ 生成约束文本：recordId + role + 制品 hash + assetRef +
|
||
designRef + 参照义务（内容完整度/玩法义务须达到参照资产水准），供调用方注入生成 prompt 约束块。
|
||
返回 {constraint_block, consumed, records}；consumed 为消费时刻快照，records 为注册表完整登记（审计用）。
|
||
注：provenance 的 consumedReferenceAssets 不由本函数落——验收侧以验收时点对账零信任重产快照，两侧独立成证。
|
||
"""
|
||
declared = (identity or {}).get("referenceAssetRecordIds") or []
|
||
if not declared:
|
||
return None
|
||
reconciled = reconcile_v3_reference_asset_consumption(
|
||
declared, consumer_ref=(identity or {}).get("consumerRef"), registry=registry)
|
||
if not reconciled["ok"]:
|
||
raise ValueError("参照资产消费对账拒绝:" + ";".join(reconciled["errors"][:4]))
|
||
# 取完整登记记录（assetRef/designRef）渲染约束文本；对账已保证记录存在且 active。by_id 与对账函数同口径首条优先。
|
||
if registry is None:
|
||
registry = _load_v3_reference_asset_registry()
|
||
by_id = {}
|
||
for record in registry.get("records") or []:
|
||
if isinstance(record, dict):
|
||
record_id = record.get("recordId")
|
||
if _is_v3_record_id(record_id) and record_id not in by_id:
|
||
by_id[record_id] = record
|
||
lines = [
|
||
"【参照资产消费约束（已对账 active）】",
|
||
"本局生成声明消费以下参照资产并通过对账（recordId + role + consumerRef + artifactHash 逐条核验）。"
|
||
"生成必须达到参照义务，未达标即验收失败：",
|
||
]
|
||
records = []
|
||
for item in reconciled["consumed"]:
|
||
record = by_id.get(item["recordId"]) or {}
|
||
records.append(record)
|
||
design_refs = record.get("designRef")
|
||
design_text = ";".join(design_refs) if isinstance(design_refs, list) and design_refs else ""
|
||
lines.append(
|
||
f"- recordId={item['recordId']}（role={item['role']}，制品 hash={item['artifactHash']}）："
|
||
f"资产={record.get('assetRef') or '-'}"
|
||
+ (f"；设计参照（designRef）={design_text}" if design_text else ""))
|
||
lines.append(
|
||
f" 参照义务：本局游戏的内容完整度与玩法义务须达到参照资产 {item['recordId']} 的水准；"
|
||
f"开工先 read 上述 assetRef 源工程，以它为该品类的最低验收基线，不得降级照抄。")
|
||
return {"constraint_block": "\n".join(lines), "consumed": reconciled["consumed"], "records": records}
|
||
|
||
|
||
def _artifact_hash_v3(game_id: str, artifact_path=None) -> str:
|
||
"""从受限不可变快照计算 staged 指纹；验收与发布不再维护两套读取边界。"""
|
||
root = Path(artifact_path) if artifact_path else cheap_run.wg1_game_dir(game_id)
|
||
try:
|
||
root.lstat()
|
||
except FileNotFoundError:
|
||
return _sha256_text(f"missing:{root}")
|
||
return artifact_snapshot.capture_artifact_snapshot(root).artifact_hash
|
||
|
||
|
||
def _v3_run_id(request: dict, artifact_hash: str, brief_hash: str) -> str:
|
||
"""由幂等键和不可变输入生成稳定 runId；同输入重入只会命中同一封存结果。"""
|
||
idem = str(request.get("idempotencyKey") or request.get("idempotency_key") or request.get("gameId") or "")
|
||
parent = str(request.get("parentRunId") or "")
|
||
mode = str(request.get("acceptanceMode") or _acceptance_v3_cfg()["mode"])
|
||
repair_count = int(request.get("repairCountAcrossParentChain") or 0)
|
||
identity = request.get("acceptanceIdentity") if isinstance(request.get("acceptanceIdentity"), dict) else request
|
||
acceptance_request_hash = str(identity.get("acceptanceRequestHash") or "")
|
||
proof_profile_id = str(identity.get("proofProfileId") or "")
|
||
proof_registry_version = str(identity.get("proofRegistryVersion") or "")
|
||
raw = (
|
||
f"{idem}|{artifact_hash}|{brief_hash}|{mode}|{parent}|{repair_count}|"
|
||
f"{acceptance_request_hash}|{proof_profile_id}|{proof_registry_version}"
|
||
)
|
||
return "v3-" + _sha256_text(raw)[:24]
|
||
|
||
|
||
def _playtest_v3_script() -> Path:
|
||
"""返回 playtest/3 runner 的唯一入口；runner 未部署时同步封装会诚实返回 tester_error。"""
|
||
return cheap_run._WG1_DIR / "_shared" / "serve-and-playtest-v3.sh"
|
||
|
||
|
||
def _trusted_runner_cost_v3(payload: dict) -> float | None:
|
||
"""只接受与 Node ``createCostLedger`` 同构且可复算的成本；否则按未知成本处理。
|
||
|
||
Node 每个 entry 的金额键是 ``rmb``，顶层与 ledger.costRmb 都是 ``sum(entry.rmb).toFixed(5)``。
|
||
这里只信可逐项复算的有限非负金额；看错键、只信顶层或不核总和都会让伪造的低成本绕过硬帽。
|
||
"""
|
||
if not isinstance(payload, dict):
|
||
return None
|
||
top = payload.get("costRmb")
|
||
ledger = payload.get("cost")
|
||
ledger_top = ledger.get("costRmb") if isinstance(ledger, dict) else None
|
||
if (not isinstance(top, (int, float)) or isinstance(top, bool) or not math.isfinite(top) or top < 0
|
||
or not isinstance(ledger_top, (int, float)) or isinstance(ledger_top, bool)
|
||
or not math.isfinite(ledger_top) or ledger_top < 0
|
||
or not isinstance(ledger.get("entries"), list)):
|
||
return None
|
||
normalized_top = _quantize_rmb_v3(top, "runner.costRmb")
|
||
if float(top) != normalized_top or float(ledger_top) != normalized_top:
|
||
return None
|
||
total = 0.0
|
||
entry_fields = {
|
||
"role", "model", "promptTokens", "completionTokens", "cachedTokens",
|
||
"modelRatio", "completionRatio", "cacheRatio", "quota", "rmb",
|
||
}
|
||
for entry in ledger["entries"]:
|
||
if not isinstance(entry, dict) or set(entry) != entry_fields:
|
||
return None
|
||
role = entry.get("role")
|
||
model = entry.get("model")
|
||
value = entry.get("rmb")
|
||
if role not in ("Actor", "JudgeA", "JudgeB") or not isinstance(model, str) or not model.strip():
|
||
return None
|
||
for key in ("promptTokens", "completionTokens", "cachedTokens"):
|
||
token_value = entry.get(key)
|
||
if not isinstance(token_value, int) or isinstance(token_value, bool) or token_value < 0:
|
||
return None
|
||
for key in ("modelRatio", "completionRatio", "cacheRatio", "quota"):
|
||
numeric = entry.get(key)
|
||
if (not isinstance(numeric, (int, float)) or isinstance(numeric, bool)
|
||
or not math.isfinite(numeric) or numeric < 0):
|
||
return None
|
||
if (not isinstance(value, (int, float)) or isinstance(value, bool)
|
||
or not math.isfinite(value) or value < 0):
|
||
return None
|
||
total += float(value)
|
||
if _quantize_rmb_v3(total, "runner.cost.entries") != normalized_top:
|
||
return None
|
||
return normalized_top
|
||
|
||
|
||
def _run_playtest_v3_roll_sync(game_id: str, *, run_id: str, roll: int, requested_seed: int,
|
||
policy_seed: int, brief_hash: str, artifact_hash: str,
|
||
brief_file: str, genre: str, template_route: str,
|
||
proof_profile_id: str, proof_registry_version: str,
|
||
task_binding_hash: str, acceptance_request_hash: str, acceptance_request_file: str,
|
||
provenance_manifest_file: str, provenance_manifest_hash: str,
|
||
acceptance_mode: str, evidence_mode: str, evidence_dir, model_name: str,
|
||
port: int, cdp_port: int, timeout: float,
|
||
remaining_cost_rmb: float, cfg: dict, legacy_mapping_file=None,
|
||
interaction_binding_file=None, interaction_binding_file_hash=None,
|
||
interaction_profile_id=None, interaction_registry_version=None,
|
||
interaction_binding_hash=None) -> dict:
|
||
"""调用新 serve 脚本执行一掷；只解析 playtest/3 JSON，不把进程错误伪装成 gameplay。"""
|
||
unknown_cost = max(0.0, float(remaining_cost_rmb or 0.0))
|
||
script = _playtest_v3_script()
|
||
if not script.is_file():
|
||
return {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "roll": roll, "outcome": "tester_error",
|
||
"failureSubtype": "runner_error", "reason": f"v3 runner 不存在:{script}",
|
||
"infraRetryable": False, "costRmb": 0.0,
|
||
}
|
||
argv = [
|
||
# 与既有 serve-and-playtest.sh 保持同形：gameId/port/cdpPort 三个位置参数，随后用 -- 透传 runner 参数。
|
||
"bash", str(script), str(game_id), str(port), str(cdp_port), "--",
|
||
f"--run-id={run_id}", f"--roll={roll}", f"--requested-seed={requested_seed}",
|
||
f"--policy-seed={policy_seed}", f"--brief-hash={brief_hash}",
|
||
f"--artifact-hash={artifact_hash}", f"--evidence-dir={evidence_dir}",
|
||
f"--brief-file={brief_file}", f"--obligations-file={_V3_OBLIGATIONS_FILE}", f"--genre={genre}",
|
||
f"--template-route={template_route}", f"--proof-profile-id={proof_profile_id}",
|
||
f"--proof-registry-version={proof_registry_version}",
|
||
f"--task-binding-hash={task_binding_hash}",
|
||
f"--acceptance-request-hash={acceptance_request_hash}",
|
||
f"--acceptance-request-file={acceptance_request_file}",
|
||
f"--provenance-manifest={provenance_manifest_file}",
|
||
f"--provenance-manifest-hash={provenance_manifest_hash}",
|
||
f"--acceptance-mode={acceptance_mode}", f"--evidence-mode={evidence_mode}",
|
||
f"--model={model_name}", f"--cost-cap-rmb={max(0.0, remaining_cost_rmb):.6f}",
|
||
f"--action-quantum-ms={cfg['action_quantum_ms']}",
|
||
f"--wait-min-ms={cfg['wait_min_ms']}", f"--wait-max-ms={cfg['wait_max_ms']}",
|
||
]
|
||
if legacy_mapping_file:
|
||
argv.append(f"--legacy-mapping-file={legacy_mapping_file}")
|
||
interaction_values = (
|
||
interaction_binding_file, interaction_binding_file_hash, interaction_profile_id,
|
||
interaction_registry_version, interaction_binding_hash,
|
||
)
|
||
if any(value is not None for value in interaction_values):
|
||
if not all(value is not None for value in interaction_values):
|
||
return {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "roll": roll, "outcome": "tester_error",
|
||
"failureSubtype": "profile_contract_error",
|
||
"reason": "interaction binding runner 参数必须五项全有或全无",
|
||
"infraRetryable": False, "costRmb": 0.0,
|
||
}
|
||
# 新编排使用语义明确的 --interaction-binding-file；Node 同时兼容历史 --interaction-binding。
|
||
argv.extend((
|
||
f"--interaction-binding-file={interaction_binding_file}",
|
||
f"--interaction-binding-file-hash={interaction_binding_file_hash}",
|
||
f"--interaction-profile-id={interaction_profile_id}",
|
||
f"--interaction-registry-version={interaction_registry_version}",
|
||
f"--interaction-binding-hash={interaction_binding_hash}",
|
||
))
|
||
try:
|
||
proc = subprocess.run(argv, cwd=str(cheap_run._GAME_RUNTIME), capture_output=True, text=True,
|
||
timeout=timeout, env=_playtest_env(), check=False)
|
||
except subprocess.TimeoutExpired:
|
||
return {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "roll": roll, "outcome": "tester_error",
|
||
"failureSubtype": "runner_timeout", "reason": f"v3 runner 超时(>{timeout}s)",
|
||
"infraRetryable": True, "costRmb": unknown_cost,
|
||
}
|
||
except Exception as e: # noqa: BLE001
|
||
return {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "roll": roll, "outcome": "tester_error",
|
||
"failureSubtype": "runner_error", "reason": f"v3 runner 异常:{type(e).__name__}: {e}",
|
||
"infraRetryable": True, "costRmb": unknown_cost,
|
||
}
|
||
payload = None
|
||
for line in reversed((proc.stdout or "").splitlines()):
|
||
line = line.strip()
|
||
if not (line.startswith("{") and line.endswith("}")):
|
||
continue
|
||
try:
|
||
payload = json.loads(line)
|
||
break
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if not isinstance(payload, dict):
|
||
return {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "roll": roll, "outcome": "tester_error",
|
||
"failureSubtype": "schema_error", "reason": "v3 runner 未产出 JSON 对象",
|
||
"infraRetryable": True, "raw": ((proc.stdout or "") + (proc.stderr or ""))[-1000:],
|
||
"costRmb": unknown_cost,
|
||
}
|
||
package_type = payload.get("packageType")
|
||
trusted_payload_cost = _trusted_runner_cost_v3(payload)
|
||
if package_type == "SingleRollRunnerError":
|
||
error = payload.get("error") if isinstance(payload.get("error"), dict) else {}
|
||
payload["outcome"] = "tester_error"
|
||
payload.setdefault("failureSubtype", "runner_error")
|
||
payload.setdefault("infraRetryable", False)
|
||
payload.setdefault("reason", error.get("message") or f"v3 runner 退出码 {proc.returncode}")
|
||
payload["costRmb"] = trusted_payload_cost if trusted_payload_cost is not None else unknown_cost
|
||
return payload
|
||
if package_type != "SingleRollFact":
|
||
return {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "packageType": "PythonBoundaryError", "roll": roll,
|
||
"outcome": "tester_error", "failureSubtype": "schema_error", "infraRetryable": True,
|
||
"reason": f"runner stdout packageType 非法:{package_type or '-'}", "costRmb": unknown_cost,
|
||
}
|
||
if proc.returncode != 0:
|
||
return {
|
||
**payload, "outcome": "tester_error", "failureSubtype": "runner_error", "infraRetryable": True,
|
||
"reason": f"runner 返回 SingleRollFact 但退出码为 {proc.returncode}",
|
||
"costRmb": trusted_payload_cost if trusted_payload_cost is not None else unknown_cost,
|
||
}
|
||
|
||
fact_schema_errors, _ = _validate_v3_instance_parts(_V3_SINGLE_ROLL_SCHEMA, payload)
|
||
if fact_schema_errors:
|
||
return {
|
||
**payload, "outcome": "tester_error", "failureSubtype": "schema_error",
|
||
"infraRetryable": False, "reason": "SingleRollFact schema 非法:" + ";".join(fact_schema_errors[:6]),
|
||
"costRmb": trusted_payload_cost if trusted_payload_cost is not None else unknown_cost,
|
||
}
|
||
|
||
# stdout 只是单掷索引；可信动作、帧、事件和义务必须从已封存 proofRef 回读，不能信 stdout 复制件。
|
||
runner = payload.get("runner") if isinstance(payload.get("runner"), dict) else {}
|
||
proof_ref = payload.get("proofPackageRef") or runner.get("proofRef")
|
||
proof_path = (Path(evidence_dir) / str(proof_ref or "")).resolve()
|
||
evidence_root = Path(evidence_dir).resolve()
|
||
try:
|
||
if not proof_ref or (proof_path != evidence_root and evidence_root not in proof_path.parents):
|
||
raise ValueError("proofRef 缺失或越界")
|
||
proof_bytes = proof_path.read_bytes()
|
||
actual_proof_hash = hashlib.sha256(proof_bytes).hexdigest()
|
||
expected_proof_hash = str(payload.get("proofPackageHash") or runner.get("proofHash") or "")
|
||
if not _is_sha256_v3(expected_proof_hash) or actual_proof_hash != expected_proof_hash:
|
||
raise ValueError("proofPackageHash 与封存文件字节不一致")
|
||
proof = json.loads(proof_bytes.decode("utf-8"))
|
||
if proof.get("schemaVersion") != _PLAYTEST_V3_SCHEMA or proof.get("packageType") != "SingleRollProof":
|
||
raise ValueError(f"proof packageType/schema 非法:{proof.get('packageType')}/{proof.get('schemaVersion')}")
|
||
identities = {
|
||
"runId": run_id, "roll": roll, "gameId": str(game_id), "genre": genre,
|
||
"templateRoute": template_route, "proofProfileId": proof_profile_id,
|
||
"proofRegistryVersion": proof_registry_version,
|
||
"taskBindingHash": task_binding_hash,
|
||
"acceptanceRequestHash": acceptance_request_hash, "evidenceMode": evidence_mode,
|
||
"artifactHash": artifact_hash, "briefHash": brief_hash,
|
||
}
|
||
for key, expected in identities.items():
|
||
if payload.get(key) != expected:
|
||
raise ValueError(f"stdout {key} 与请求不一致:{payload.get(key)} != {expected}")
|
||
if proof.get(key) != expected:
|
||
raise ValueError(f"proof {key} 与请求不一致:{proof.get(key)} != {expected}")
|
||
# stdout 只作索引；所有可影响 guard 的事实必须与 hash 封存 proof 逐字段一致。
|
||
mirror_fields = (
|
||
"requestedSeed", "actualSeed", "policySeed", "environment", "actor", "actions", "frames",
|
||
"events", "briefRuleMatches", "proofObligations", "firstPlay", "judgePackageRef",
|
||
"judgePackageHash", "judgeARef", "judgeAHash", "judgeBRef", "judgeBHash",
|
||
"judgeConsensusRef", "judgeConsensusHash", "costReservationRef", "costReservationHash",
|
||
"judge", "errors", "costRmb", "cost",
|
||
)
|
||
for key in mirror_fields:
|
||
if payload.get(key) != proof.get(key):
|
||
raise ValueError(f"stdout {key} 与封存 proof 不一致")
|
||
payload["_proofPackage"] = proof
|
||
payload["_proofPackageRef"] = str(proof_ref)
|
||
payload["_evidenceRoot"] = str(evidence_root)
|
||
judge_ref = proof.get("judgePackageRef") or payload.get("judgePackageRef")
|
||
judge_hash = str(proof.get("judgePackageHash") or payload.get("judgePackageHash") or "")
|
||
if judge_ref:
|
||
judge_path = (evidence_root / str(judge_ref)).resolve()
|
||
if evidence_root not in judge_path.parents or not _is_sha256_v3(judge_hash):
|
||
raise ValueError("judgePackageRef 越界或 judgePackageHash 非法")
|
||
if hashlib.sha256(judge_path.read_bytes()).hexdigest() != judge_hash:
|
||
raise ValueError("judgePackageHash 与封存文件字节不一致")
|
||
for slot_name in ("judgeARef", "judgeBRef"):
|
||
slot_ref = proof.get(slot_name) if isinstance(proof.get(slot_name), dict) else {}
|
||
for ref_field, hash_field in (("rawRef", "rawHash"), ("parsedRef", "parsedHash")):
|
||
target = (evidence_root / str(slot_ref.get(ref_field) or "")).resolve()
|
||
expected = str(slot_ref.get(hash_field) or "")
|
||
if (evidence_root not in target.parents or not target.is_file()
|
||
or not _is_sha256_v3(expected)
|
||
or hashlib.sha256(target.read_bytes()).hexdigest() != expected):
|
||
raise ValueError(f"{slot_name}.{ref_field}/{hash_field} 无法回读复核")
|
||
normalized_path = (evidence_root / str(slot_ref.get("normalizedRef") or "")).resolve()
|
||
top_hash_field = "judgeAHash" if slot_name == "judgeARef" else "judgeBHash"
|
||
normalized_hash = str(proof.get(top_hash_field) or "")
|
||
if (evidence_root not in normalized_path.parents or not normalized_path.is_file()
|
||
or not _is_sha256_v3(normalized_hash)
|
||
or hashlib.sha256(normalized_path.read_bytes()).hexdigest() != normalized_hash):
|
||
raise ValueError(f"{slot_name}.normalizedRef/{top_hash_field} 无法回读复核")
|
||
consensus_ref = str(proof.get("judgeConsensusRef") or "")
|
||
consensus_path = (evidence_root / consensus_ref).resolve()
|
||
consensus_hash = str(proof.get("judgeConsensusHash") or "")
|
||
if (evidence_root not in consensus_path.parents or not consensus_path.is_file()
|
||
or not _is_sha256_v3(consensus_hash)
|
||
or hashlib.sha256(consensus_path.read_bytes()).hexdigest() != consensus_hash):
|
||
raise ValueError("JudgeConsensus ref/hash 无法回读复核")
|
||
consensus = json.loads(consensus_path.read_text(encoding="utf-8"))
|
||
if consensus.get("strategyVersion") != "JudgeConsensus/1" \
|
||
or consensus.get("judgePackageHash") != judge_hash:
|
||
raise ValueError("JudgeConsensus 版本或 packageHash 不一致")
|
||
evaluation_hashes = consensus.get("evaluationHashes") \
|
||
if isinstance(consensus.get("evaluationHashes"), dict) else {}
|
||
if (evaluation_hashes.get("judgeA") != (proof.get("judgeARef") or {}).get("resultHash")
|
||
or evaluation_hashes.get("judgeB") != (proof.get("judgeBRef") or {}).get("resultHash")):
|
||
raise ValueError("JudgeConsensus evaluationHashes 与 A/B 槽结果不一致")
|
||
reservation_ref = str(proof.get("costReservationRef") or "")
|
||
reservation_path = (evidence_root / reservation_ref).resolve()
|
||
reservation_hash = str(proof.get("costReservationHash") or "")
|
||
if (evidence_root not in reservation_path.parents or not reservation_path.is_file()
|
||
or not _is_sha256_v3(reservation_hash)
|
||
or hashlib.sha256(reservation_path.read_bytes()).hexdigest() != reservation_hash):
|
||
raise ValueError("CostReservation ref/hash 无法回读复核")
|
||
reservation = json.loads(reservation_path.read_text(encoding="utf-8"))
|
||
reservation_errors, reservation_semantic_errors = _validate_v3_instance_parts(
|
||
_V3_CONTRACT_DIR / "cost-reservation.schema.json", reservation)
|
||
if reservation_errors or reservation_semantic_errors \
|
||
or reservation.get("judgePackageHash") != judge_hash:
|
||
raise ValueError("CostReservation schema、语义或 judgePackageHash 不一致")
|
||
payload["_judgeConsensus"] = consensus
|
||
payload["_costReservation"] = reservation
|
||
except Exception as e: # noqa: BLE001 —— 封存 proof 不可读属于测试器错误，绝不降成 gameplay reject
|
||
return {
|
||
**payload, "outcome": "tester_error", "failureSubtype": "schema_error", "infraRetryable": True,
|
||
"reason": f"封存 proof 回读失败:{type(e).__name__}: {e}",
|
||
"costRmb": trusted_payload_cost if trusted_payload_cost is not None else unknown_cost,
|
||
}
|
||
payload.setdefault("costRmb", 0.0)
|
||
return payload
|
||
|
||
|
||
|
||
|
||
def _v3_resolve_registry(genre: str, brief: str, *, template_route: str,
|
||
proof_profile_id=None, proof_registry_version=None) -> dict:
|
||
"""按可信模板路由选择唯一 profile，并交叉核对 genre/profile/registryVersion。"""
|
||
registry = json.loads(_V3_OBLIGATIONS_FILE.read_text(encoding="utf-8"))
|
||
registry_errors = _validate_v3_instance(_V3_OBLIGATIONS_SCHEMA, registry)
|
||
if registry_errors:
|
||
raise ValueError("proof obligation 注册表非法:" + ";".join(registry_errors[:4]))
|
||
registry_version = str(registry.get("registryVersion") or "")
|
||
if proof_registry_version is not None and str(proof_registry_version) != registry_version:
|
||
raise ValueError("proofRegistryVersion 与 canonical 注册表不一致")
|
||
profiles = registry.get("profiles") if isinstance(registry.get("profiles"), dict) else {}
|
||
matches = [profile for profile in profiles.values()
|
||
if isinstance(profile, dict) and profile.get("templateRoute") == template_route]
|
||
if len(matches) != 1:
|
||
raise ValueError(f"templateRoute 必须唯一命中可信 profile:{template_route or '-'}")
|
||
profile = matches[0]
|
||
if profile.get("genre") != genre:
|
||
raise ValueError(f"genre/templateRoute 不一致:{genre or '-'} != {profile.get('genre') or '-'}")
|
||
if proof_profile_id is not None and profile.get("id") != proof_profile_id:
|
||
raise ValueError(f"proofProfileId/templateRoute 不一致:{proof_profile_id} != {profile.get('id')}")
|
||
normalized = (brief or "").lower()
|
||
promoted = set()
|
||
brief_matches = []
|
||
for rule in profile.get("briefOptionalRules") or []:
|
||
keywords = ((rule.get("when") or {}).get("anyKeywords") or [])
|
||
if any(str(keyword).lower() in normalized for keyword in keywords):
|
||
brief_matches.append(str(rule.get("id")))
|
||
promoted.update(rule.get("requireObligationIds") or [])
|
||
obligations = []
|
||
for item in profile.get("obligations") or []:
|
||
required = bool(item.get("defaultRequired") or item.get("id") in promoted)
|
||
obligations.append({
|
||
"id": str(item.get("id") or ""),
|
||
"required": required,
|
||
"evidence": item.get("evidence") if isinstance(item.get("evidence"), dict) else {},
|
||
"status": "missing" if required else "not-required",
|
||
"sequenceRefs": [], "actionRefs": [], "postFrameRefs": [], "eventRefs": [],
|
||
"contradictions": [], "blockingProblems": [],
|
||
})
|
||
return {
|
||
"schemaVersion": str(registry.get("schemaVersion") or ""),
|
||
"version": registry_version,
|
||
"proofProfileId": str(profile.get("id") or ""),
|
||
"genre": str(profile.get("genre") or ""),
|
||
"templateRoute": str(profile.get("templateRoute") or ""),
|
||
"briefRuleMatches": brief_matches,
|
||
"obligations": obligations,
|
||
}
|
||
|
||
|
||
|
||
|
||
def _v3_judge_contract(raw: dict, *, evidence_dir: str) -> dict | None:
|
||
"""读取 playtest/3 双 Judge consensus 投影；不再把单槽身份冒充最终 Judge。"""
|
||
judge = raw.get("judge") if isinstance(raw.get("judge"), dict) else {}
|
||
if (judge.get("schemaVersion") == "JudgeConsensus/1"
|
||
and isinstance(judge.get("obligationResults"), list)
|
||
and judge.get("decision") in _V3_OUTCOMES and "degraded" in judge):
|
||
return dict(judge)
|
||
return None
|
||
|
||
|
||
def _v3_actor_contract(raw: dict, *, evidence_dir: str) -> dict | None:
|
||
"""投影 Actor 身份；Actor 错误时仍保留身份用于审计，但 guard 会判 actorHealthy=false。"""
|
||
actor = raw.get("actor") if isinstance(raw.get("actor"), dict) else {}
|
||
if not actor:
|
||
return None
|
||
prompt = actor.get("prompt") if isinstance(actor.get("prompt"), dict) else {}
|
||
return {
|
||
"model": str(actor.get("model") or "unknown"),
|
||
"apiClientId": str(actor.get("apiClientId") or "playtest-v3-actor-client"),
|
||
"sessionId": str(actor.get("sessionId") or "unknown-actor-session"),
|
||
"systemPromptVersion": str(actor.get("systemPromptVersion") or prompt.get("version") or "playtest.actor@unknown"),
|
||
"contextProjectionVersion": str(actor.get("contextProjectionVersion") or "ActorView/1"),
|
||
"evidenceDir": str(actor.get("evidenceDir") or f"{evidence_dir}/actor-output"),
|
||
"requestedGameSeed": int(raw.get("requestedSeed") or 0), "actualGameSeed": int(raw.get("actualSeed") or 0),
|
||
"policySeed": int(raw.get("policySeed") or 0),
|
||
}
|
||
|
||
|
||
_V3_RELATION_ERROR_MARKERS = (
|
||
"/sameValue", "所列步骤的值必须逐字相等", "缺少可比较的 JSON scalar",
|
||
"orderedSeries", "工序数必须在", "动态 stepId 必须", "stepIndex 必须", "totalSteps 必须",
|
||
"judgement 必须", "所有工序必须属于", "工序 stepId 必须唯一", "每道工序必须使用",
|
||
"只允许与最后一道工序", "共享最后动作时", "workId 必须与工序链一致",
|
||
"orderedStepIds 必须", "allStepsCompleted 必须", "score 必须是有限数", "works 必须大于零",
|
||
)
|
||
|
||
|
||
def _v3_relation_contradiction_errors(errors: list) -> list:
|
||
"""筛出游戏自报完成态与有序关系冲突；这类是玩法硬证 reject，不是 harness tester_error。"""
|
||
return [str(error) for error in errors
|
||
if any(marker in str(error) for marker in _V3_RELATION_ERROR_MARKERS)]
|
||
|
||
|
||
def _v3_judge_text_reference_errors(roll: dict) -> list:
|
||
"""独立复核 Judge 自由文本只能引用本次单掷真实 action/frame/event。"""
|
||
import re # noqa: PLC0415
|
||
|
||
judge = roll.get("judge") if isinstance(roll.get("judge"), dict) else {}
|
||
# playtest/3 的最终投影不携带自由文本；A/B 原文已在各槽 parser 与 consensus 中封存。
|
||
if judge.get("schemaVersion") == "JudgeConsensus/1":
|
||
return []
|
||
if judge.get("decision") == "tester_error":
|
||
return []
|
||
action_refs = {str(row.get("actionId")) for row in roll.get("actions") or []
|
||
if isinstance(row, dict) and row.get("actionId")}
|
||
frame_refs = {str(row.get("ref")) for row in roll.get("frames") or []
|
||
if isinstance(row, dict) and row.get("ref")}
|
||
event_refs = {f"event:{row.get('seq')}" for row in roll.get("events") or []
|
||
if isinstance(row, dict) and isinstance(row.get("seq"), int)
|
||
and not isinstance(row.get("seq"), bool) and row.get("seq") >= 1}
|
||
allowed = action_refs | frame_refs | event_refs
|
||
pattern = re.compile(
|
||
r"event:\d+|action-\d+|(?:[A-Za-z0-9._-]+/)+[A-Za-z0-9._-]+[.](?:png|jpe?g)"
|
||
)
|
||
groups = [
|
||
("problems", [str(value) for value in judge.get("problems") or []]),
|
||
("contradictions", [str(value) for value in judge.get("contradictions") or []]),
|
||
("summary", [str(judge.get("summary") or "")]),
|
||
]
|
||
used = set()
|
||
errors = []
|
||
for label, values in groups:
|
||
non_empty = [text for text in values if text.strip()]
|
||
if label == "summary" and len(non_empty) != 1:
|
||
errors.append("Judge summary 不得为空")
|
||
continue
|
||
refs = {ref for text in non_empty for ref in pattern.findall(text)}
|
||
if non_empty and not refs:
|
||
errors.append(f"Judge {label} 整体缺 action/frame/event 原始引用")
|
||
continue
|
||
unknown = sorted(refs - allowed)
|
||
if unknown:
|
||
errors.append(f"Judge {label} 含跨包或未知引用:{','.join(unknown)}")
|
||
used.update(refs)
|
||
if judge.get("decision") == "reject" \
|
||
and (not (used & event_refs) or not (used & frame_refs)):
|
||
errors.append("Judge reject 未同时绑定本次事件和截图")
|
||
return errors
|
||
|
||
|
||
def _v3_relation_reject_facts(roll: dict, relation_errors: list) -> tuple[dict, list]:
|
||
"""把关系冲突投影成 schema 合法的 contradicted 义务，避免坏自报关系污染最终契约。"""
|
||
import copy # noqa: PLC0415
|
||
import re # noqa: PLC0415
|
||
|
||
facts = {
|
||
"environment": copy.deepcopy(roll.get("environment")),
|
||
"actions": copy.deepcopy(roll.get("actions") or []),
|
||
"events": copy.deepcopy(roll.get("events") or []),
|
||
"proofObligations": copy.deepcopy(roll.get("proofObligations") or []),
|
||
"firstPlay": copy.deepcopy(roll.get("firstPlay") or {}),
|
||
"proofPackageRef": str(roll.get("proofPackageRef") or ""),
|
||
"judgePackageRef": roll.get("judgePackageRef"),
|
||
"proofRegistryVersion": roll.get("proofRegistryVersion"),
|
||
"briefRuleMatches": list(roll.get("briefRuleMatches") or []),
|
||
}
|
||
affected = set()
|
||
for error in relation_errors:
|
||
match = re.search(r"#/proofObligations/([^/]+)/", error)
|
||
if match:
|
||
affected.add(match.group(1))
|
||
if not affected:
|
||
affected = {row.get("id") for row in facts["proofObligations"] if row.get("required") is True}
|
||
for row in facts["proofObligations"]:
|
||
if row.get("id") not in affected:
|
||
continue
|
||
row.update({"status": "contradicted", "sequenceRefs": [], "actionRefs": [],
|
||
"postFrameRefs": [], "eventRefs": [],
|
||
"contradictions": list(relation_errors), "blockingProblems": []})
|
||
judge = _v3_judge_contract(roll, evidence_dir=str(roll.get("_evidenceDir") or "evidence/playtest-v3"))
|
||
# 关系硬证由 Python guard 单独拒绝；不得改写已 hash 封存的 JudgeConsensus 投影。
|
||
facts["judge"] = judge
|
||
facts["actor"] = _v3_actor_contract(
|
||
roll, evidence_dir=str(roll.get("_evidenceDir") or "evidence/playtest-v3"))
|
||
return facts, sorted(str(value) for value in affected if value)
|
||
|
||
|
||
def _roll_guard_v3(roll: dict, *, artifact_hash: str, brief_hash: str, floor_pass: bool = True,
|
||
registry: dict | None = None, cfg: dict | None = None,
|
||
task_binding_hash=None, acceptance_request_hash=None, evidence_mode="native") -> dict:
|
||
"""复核 canonical SingleRollFact、封存字节和可信身份；只产单掷四态。"""
|
||
del cfg # 时间、sameValue、orderedSeries 等全部由 canonical validator 统一校验。
|
||
empty_checks = {key: False for key in _V3_GUARD_CHECKS}
|
||
checks = dict(empty_checks)
|
||
checks["floorPass"] = bool(floor_pass)
|
||
|
||
def tester(reason, subtype="schema_error", *, retryable=True):
|
||
return {"pass": False, "checks": checks, "blockingProblems": [reason], "contradictions": [],
|
||
"reason": reason, "rollOutcome": "tester_error", "subtype": subtype,
|
||
"infraRetryable": retryable, "errors": [reason]}
|
||
|
||
if not isinstance(roll, dict):
|
||
return tester("单掷结果不是对象")
|
||
if roll.get("schemaVersion") != _PLAYTEST_V3_SCHEMA:
|
||
return tester("未知 playtest schema", "unknown_schema")
|
||
if roll.get("outcome") == "tester_error" or roll.get("packageType") in ("SingleRollRunnerError", "PythonBoundaryError"):
|
||
reason = str(roll.get("reason") or "runner_error")
|
||
return tester(reason, roll.get("failureSubtype") or "runner_error",
|
||
retryable=bool(roll.get("infraRetryable", True)))
|
||
|
||
schema_errors, semantic_errors = _validate_v3_instance_parts(_V3_SINGLE_ROLL_SCHEMA, roll)
|
||
if schema_errors:
|
||
return tester("SingleRollFact schema 非法:" + ";".join(schema_errors[:6]),
|
||
"schema_error", retryable=False)
|
||
relation_errors = _v3_relation_contradiction_errors(semantic_errors)
|
||
other_semantic_errors = [error for error in semantic_errors if error not in relation_errors]
|
||
if other_semantic_errors:
|
||
return tester("SingleRollFact contract/semantic 非法:" + ";".join(other_semantic_errors[:6]),
|
||
"schema_error", retryable=False)
|
||
|
||
registry = registry or {}
|
||
identities = {
|
||
"artifactHash": artifact_hash,
|
||
"briefHash": brief_hash,
|
||
"templateRoute": registry.get("templateRoute"),
|
||
"proofProfileId": registry.get("proofProfileId"),
|
||
"proofRegistryVersion": registry.get("version"),
|
||
"genre": registry.get("genre"),
|
||
"evidenceMode": evidence_mode,
|
||
}
|
||
if task_binding_hash is not None:
|
||
identities["taskBindingHash"] = task_binding_hash
|
||
if acceptance_request_hash is not None:
|
||
identities["acceptanceRequestHash"] = acceptance_request_hash
|
||
for field, expected in identities.items():
|
||
if expected is not None and roll.get(field) != expected:
|
||
return tester(f"SingleRollFact {field} 与可信请求不一致", "profile_contract_error", retryable=False)
|
||
if list(roll.get("briefRuleMatches") or []) != list(registry.get("briefRuleMatches") or []):
|
||
return tester("briefRuleMatches 与可信 profile 解析不一致", "profile_contract_error", retryable=False)
|
||
|
||
actions = list(roll.get("actions") or [])
|
||
events = list(roll.get("events") or [])
|
||
obligations = list(roll.get("proofObligations") or [])
|
||
required = [row for row in obligations if row["required"] is True]
|
||
actor_raw = roll.get("actor") if isinstance(roll.get("actor"), dict) else {}
|
||
judge_raw = roll.get("judge") if isinstance(roll.get("judge"), dict) else {}
|
||
consensus_raw = roll.get("_judgeConsensus") if isinstance(roll.get("_judgeConsensus"), dict) else {}
|
||
judge = _v3_judge_contract(roll, evidence_dir=str(roll.get("_evidenceDir") or "evidence/playtest-v3"))
|
||
actor = _v3_actor_contract(roll, evidence_dir=str(roll.get("_evidenceDir") or "evidence/playtest-v3"))
|
||
|
||
evidence_hashes = set()
|
||
evidence_root_value = roll.get("_evidenceRoot")
|
||
evidence_root = Path(evidence_root_value).resolve() if evidence_root_value else None
|
||
required_sequences = [ref for row in required for ref in row.get("sequenceRefs") or []]
|
||
referenced_frames = {ref.get("postFrameRef") for ref in required_sequences}
|
||
referenced_events = {ref.get("eventRef") for ref in required_sequences}
|
||
frame_by_ref = {row.get("ref"): row for row in roll.get("frames") or [] if isinstance(row, dict)}
|
||
if referenced_frames and evidence_root is None:
|
||
return tester("缺 evidenceRoot，无法复核动作后截图字节", "image_error")
|
||
for frame_ref in referenced_frames:
|
||
frame = frame_by_ref.get(frame_ref) or {}
|
||
path = (evidence_root / str(frame_ref or "")).resolve() if evidence_root else None
|
||
if (not frame_ref or not path or evidence_root not in path.parents
|
||
or not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != frame.get("hash")):
|
||
return tester(f"动作后截图缺失、越界或 hash 不一致:{frame_ref}", "image_error")
|
||
evidence_hashes.add(str(frame.get("hash")))
|
||
for event in events:
|
||
payload_canonical = event.get("payloadCanonical")
|
||
canonical_hash = (hashlib.sha256(payload_canonical.encode("utf-8")).hexdigest()
|
||
if isinstance(payload_canonical, str) else None)
|
||
if canonical_hash != event.get("payloadHash"):
|
||
return tester(f"game-event payloadCanonical hash 无法复核:seq={event.get('seq')}",
|
||
"event_contract_error", retryable=False)
|
||
if event.get("seq") in referenced_events:
|
||
evidence_hashes.add(str(event.get("payloadHash")))
|
||
|
||
if relation_errors:
|
||
facts, affected = _v3_relation_reject_facts(roll, relation_errors)
|
||
checks.update({
|
||
"requiredObligationsSatisfied": False, "referencesValid": True,
|
||
"sameActionEvidence": True, "noContradictions": False,
|
||
"noBlockingProblems": True, "actorHealthy": facts.get("actor") is not None,
|
||
"runnerHealthy": not roll.get("errors"), "judgeHealthy": facts.get("judge") is not None,
|
||
"judgeAccept": False,
|
||
})
|
||
evidence_refs = {
|
||
"actions": list(dict.fromkeys(v for row in required for v in row.get("actionRefs") or [])),
|
||
"frames": list(dict.fromkeys(v for row in required for v in row.get("postFrameRefs") or [])),
|
||
"events": [str(v) for v in dict.fromkeys(v for row in required for v in row.get("eventRefs") or [])],
|
||
}
|
||
return {
|
||
"pass": False, "checks": checks, "blockingProblems": [],
|
||
"contradictions": list(relation_errors),
|
||
"reason": "游戏自报完成态与同值/有序关系硬证矛盾",
|
||
"rollOutcome": "reject", "subtype": "completion_relation_contradiction",
|
||
"failedObligations": affected,
|
||
"failureSignature": "completion_relation_contradiction|" + "|".join(affected),
|
||
"verified": True, "repairEvidenceComplete": False, "facts": facts,
|
||
"evidenceHashes": sorted(evidence_hashes), "evidenceRefs": evidence_refs,
|
||
}
|
||
|
||
required_satisfied = bool(required) and all(row["status"] == "satisfied" for row in required)
|
||
consensus_reason = str(consensus_raw.get("reasonCode") or "")
|
||
consensus_conflicts = [str(v) for v in consensus_raw.get("conflicts") or []]
|
||
contradictions = ([text for row in obligations for text in row["contradictions"]]
|
||
+ (consensus_conflicts if consensus_reason in {
|
||
"judge_semantic_conflict", "evidence_contradiction"} else []))
|
||
blocking = ([text for row in obligations for text in row["blockingProblems"]]
|
||
+ ([consensus_reason] if judge_raw.get("decision") == "reject" else []))
|
||
actor_healthy = actor is not None
|
||
runner_healthy = not roll.get("errors")
|
||
judge_healthy = judge is not None and judge_raw.get("degraded") is not True
|
||
judge_accept = bool(judge_healthy and judge_raw.get("decision") == "accept")
|
||
checks.update({
|
||
"requiredObligationsSatisfied": required_satisfied,
|
||
"referencesValid": True, "sameActionEvidence": True,
|
||
"noContradictions": not contradictions, "noBlockingProblems": not blocking,
|
||
"actorHealthy": actor_healthy, "runnerHealthy": runner_healthy,
|
||
"judgeHealthy": judge_healthy, "judgeAccept": judge_accept,
|
||
})
|
||
facts = {"environment": roll.get("environment"), "actor": actor,
|
||
"actions": actions, "events": events, "proofObligations": obligations, "judge": judge,
|
||
"firstPlay": roll.get("firstPlay"), "proofPackageRef": str(roll.get("proofPackageRef") or ""),
|
||
"judgePackageRef": roll.get("judgePackageRef"), "proofRegistryVersion": roll.get("proofRegistryVersion"),
|
||
"briefRuleMatches": list(roll.get("briefRuleMatches") or [])}
|
||
internal = {"facts": facts, "evidenceHashes": sorted(evidence_hashes),
|
||
"evidenceRefs": {"actions": list(dict.fromkeys(v for row in required for v in row["actionRefs"])),
|
||
"frames": list(dict.fromkeys(v for row in required for v in row["postFrameRefs"])),
|
||
"events": [str(v) for v in dict.fromkeys(v for row in required for v in row["eventRefs"])]}}
|
||
if not runner_healthy:
|
||
return {"pass": False, "checks": checks, "blockingProblems": list(roll.get("errors") or []),
|
||
"contradictions": contradictions, "reason": ";".join(map(str, roll.get("errors") or [])),
|
||
"rollOutcome": "tester_error", "subtype": "runner_error", "infraRetryable": True,
|
||
"errors": list(roll.get("errors") or []), **internal}
|
||
if not judge_healthy or judge_raw.get("decision") == "tester_error":
|
||
return {"pass": False, "checks": checks,
|
||
"blockingProblems": [str(judge_raw.get("reason") or "Judge 未产出健康裁决")],
|
||
"contradictions": contradictions,
|
||
"reason": consensus_reason or "Judge 未产出健康裁决",
|
||
"rollOutcome": "tester_error", "subtype": "judge_error",
|
||
"infraRetryable": True, "errors": ["judge_error"], **internal}
|
||
judge_reference_errors = _v3_judge_text_reference_errors(roll)
|
||
if judge_reference_errors:
|
||
checks["referencesValid"] = False
|
||
checks["judgeHealthy"] = False
|
||
reason = ";".join(judge_reference_errors)
|
||
return {"pass": False, "checks": checks, "blockingProblems": judge_reference_errors,
|
||
"contradictions": contradictions, "reason": reason,
|
||
"rollOutcome": "tester_error", "subtype": "judge_reference_error",
|
||
"infraRetryable": False, "errors": judge_reference_errors, **internal}
|
||
if contradictions:
|
||
return {"pass": False, "checks": checks, "blockingProblems": blocking,
|
||
"contradictions": contradictions, "reason": "截图/事件/Judge 证据互相矛盾",
|
||
"rollOutcome": "inconclusive", "subtype": "evidence_contradiction", **internal}
|
||
failed = [row for row in required if row["status"] == "failed"]
|
||
if failed and judge_raw.get("decision") == "reject":
|
||
failed_ids = [row["id"] for row in failed]
|
||
signature = str(judge_raw.get("failureSignature") or consensus_raw.get("failureClass")
|
||
or "proof_obligation_failed") + "|" + "|".join(failed_ids)
|
||
return {"pass": False, "checks": checks, "blockingProblems": blocking, "contradictions": [],
|
||
"reason": consensus_reason or f"硬证义务失败:{','.join(failed_ids)}",
|
||
"rollOutcome": "reject", "subtype": str(consensus_raw.get("failureClass") or "proof_obligation_failed"),
|
||
"failedObligations": failed_ids, "failureSignature": signature, "verified": True,
|
||
"repairEvidenceComplete": bool(all(row.get("actionRefs") and row.get("postFrameRefs")
|
||
and row.get("eventRefs") for row in failed)), **internal}
|
||
# broken/hollow/off_brief 是 Judge 对整局的全局拒绝，不依赖某条 proof obligation 失败。
|
||
# 若把它降成 inconclusive，第二掷只需补一张新图就可能错误翻案为 accept。
|
||
failure_class = str(consensus_raw.get("failureClass") or "")
|
||
if judge_raw.get("decision") == "reject" and failure_class in {
|
||
"broken", "hollow", "off_brief"}:
|
||
signature = str(judge_raw.get("failureSignature") or f"{failure_class}|global")
|
||
return {"pass": False, "checks": checks, "blockingProblems": blocking,
|
||
"contradictions": [],
|
||
"reason": consensus_reason or f"Judge 全局拒绝:{failure_class}",
|
||
"rollOutcome": "reject", "subtype": failure_class,
|
||
"failedObligations": [], "failureSignature": signature, "verified": True,
|
||
"repairEvidenceComplete": False, **internal}
|
||
if all(checks.values()):
|
||
return {"pass": True, "checks": checks, "blockingProblems": [], "contradictions": [],
|
||
"reason": "单掷硬证完整", "rollOutcome": "accept", "subtype": None, "verified": True, **internal}
|
||
return {"pass": False, "checks": checks, "blockingProblems": blocking, "contradictions": contradictions,
|
||
"reason": consensus_reason or "合法证据不足",
|
||
"rollOutcome": "inconclusive", "subtype": "proof_incomplete", **internal}
|
||
|
||
|
||
|
||
|
||
def _merge_candidate_v3(guards: list) -> dict:
|
||
"""按批准矩阵合并一至两掷；返回 contract merge 字段和下划线内部复核信息。"""
|
||
roll_refs = [f"roll-{index + 1}" for index in range(len(guards))]
|
||
def result(candidate, reason, *, conflicts=None, rescued=None, source=None, **internal):
|
||
return {"triggered": len(guards) > 1, "rollRefs": roll_refs, "mergeCandidate": candidate,
|
||
"conflicts": list(conflicts or []), "rescuedByRoll": rescued, "reason": reason,
|
||
"_sourceGuard": source, **{f"_{k}": v for k, v in internal.items()}}
|
||
if not guards:
|
||
return result("tester_error", "没有单掷结果", subtype="schema_error")
|
||
first = guards[0]
|
||
if len(guards) == 1:
|
||
return result(first["rollOutcome"], first.get("reason") or "第一掷直接形成候选", source=first,
|
||
subtype=first.get("subtype"),
|
||
verifiedReject=bool(first.get("verified") and first["rollOutcome"] == "reject"
|
||
and first.get("repairEvidenceComplete") is True),
|
||
failedObligations=first.get("failedObligations") or [])
|
||
second = guards[1]
|
||
a, b = first["rollOutcome"], second["rollOutcome"]
|
||
if "tester_error" in (a, b):
|
||
bad = first if a == "tester_error" else second
|
||
return result("tester_error", bad.get("reason") or "测试器错误不进入 gameplay 合并", source=bad,
|
||
subtype=bad.get("subtype"))
|
||
if a == "reject" and b == "accept":
|
||
return result("inconclusive", "一掷证实缺陷、另一掷接受", conflicts=["reject+accept"],
|
||
subtype="roll_conflict")
|
||
if a == "reject" and b == "reject":
|
||
if first.get("failureSignature") == second.get("failureSignature"):
|
||
return result("reject", first.get("reason") or "两掷硬证一致拒绝", source=second,
|
||
subtype=first.get("subtype"),
|
||
verifiedReject=bool(first.get("verified") and second.get("verified")
|
||
and first.get("repairEvidenceComplete") is True
|
||
and second.get("repairEvidenceComplete") is True),
|
||
failedObligations=first.get("failedObligations") or [])
|
||
return result("inconclusive", "两掷拒绝原因冲突", conflicts=["reject-signature-conflict"],
|
||
subtype="roll_conflict")
|
||
if a == "inconclusive" and b == "accept":
|
||
if first.get("subtype") in ("evidence_contradiction", "roll_conflict"):
|
||
return result("inconclusive", first.get("reason") or "第一掷证据矛盾，禁止自动翻案",
|
||
conflicts=["evidence-contradiction"], subtype=first.get("subtype"))
|
||
new_evidence = set(second.get("evidenceHashes") or []) - set(first.get("evidenceHashes") or [])
|
||
if new_evidence:
|
||
return result("accept", "第二掷以新增硬证补齐第一掷缺口", rescued=2, source=second,
|
||
newEvidenceHashes=sorted(new_evidence))
|
||
return result("inconclusive", "第二掷未新增硬证", subtype="proof_not_new")
|
||
if a == "inconclusive" and b == "reject":
|
||
return result("inconclusive", "第一掷证据不足、第二掷拒绝，单掷拒绝不得覆盖不确定性",
|
||
conflicts=["inconclusive+reject"], subtype="roll_conflict")
|
||
if a == "inconclusive" and b == "inconclusive":
|
||
return result("inconclusive", "两掷均未形成完整证据", subtype="proof_incomplete")
|
||
if a == "accept" and b == "accept":
|
||
return result("accept", "两掷均完整证明闭环", source=second)
|
||
return result("inconclusive", f"未定义合并:{a}+{b}", conflicts=[f"{a}+{b}"], subtype="roll_conflict")
|
||
|
||
|
||
def _v3_failure_projection(outcome: str, *, floor: dict, merge: dict) -> tuple:
|
||
"""把 v3 outcome 单向投影到旧 failureLayer/failureReason；旧字段不再自行判定。"""
|
||
if outcome == "accept":
|
||
return "none", None
|
||
if merge.get("_subtype") == "floor_gate":
|
||
failed = [name for name, result in (floor.get("gates") or {}).items() if result.get("pass") is not True]
|
||
reason = merge.get("reason") or f"四门失败:{'/'.join(failed)}"
|
||
return "mechanical", reason
|
||
if outcome == "reject":
|
||
reason = merge.get("reason") or "硬证证明玩法闭环失败"
|
||
return "gameplay", reason
|
||
reason = merge.get("reason") or ("证据不足或冲突" if outcome == "inconclusive" else "验收基础设施错误")
|
||
return "tester_degraded", reason
|
||
|
||
|
||
def _repair_feedback_v3(merge: dict, rolls: list) -> str:
|
||
"""生成可审计 repair 契约；只引用失败义务及其 action/frame/event 硬证。"""
|
||
failed = merge.get("_failedObligations") or []
|
||
source = merge.get("_sourceGuard") or {}
|
||
facts = source.get("facts") or {}
|
||
obligations = [row for row in facts.get("proofObligations") or [] if row.get("id") in failed]
|
||
action_refs = sorted({ref for row in obligations for ref in row.get("actionRefs") or []})
|
||
frame_refs = sorted({ref for row in obligations for ref in row.get("postFrameRefs") or []})
|
||
event_refs = sorted({str(ref) for row in obligations for ref in row.get("eventRefs") or []})
|
||
failed_gates = [str(v) for v in merge.get("_failedGates") or []]
|
||
floor_refs = [f"floor/verdict.json#/guards/{name}" for name in failed_gates]
|
||
return ("[v3 已验证拒绝·仅一次修复契约] "
|
||
f"失败门={','.join(failed_gates) or '-'}；floorRefs={','.join(floor_refs) or '-'}；"
|
||
f"失败义务={','.join(map(str, failed)) or '-'}；actionRefs={','.join(action_refs) or '-'}；"
|
||
f"postFrameRefs={','.join(frame_refs) or '-'}；eventRefs={','.join(event_refs) or '-'}；"
|
||
f"事实={merge.get('reason') or '硬证证明闭环失败'}")
|
||
|
||
|
||
def _v3_guard_contract(guard: dict | None) -> dict:
|
||
"""删除 Python 内部复核字段，只留下 schema guard 五字段。"""
|
||
guard = guard or {}
|
||
checks = guard.get("checks") if isinstance(guard.get("checks"), dict) else {}
|
||
return {"pass": bool(guard.get("pass")), "checks": {key: bool(checks.get(key)) for key in _V3_GUARD_CHECKS},
|
||
"blockingProblems": [str(v) for v in guard.get("blockingProblems") or []],
|
||
"contradictions": [str(v) for v in guard.get("contradictions") or []],
|
||
"reason": str(guard.get("reason") or "")}
|
||
|
||
|
||
def _v3_merge_contract(merge: dict) -> dict:
|
||
"""删除 merge 下划线内部字段，保证封存对象只含 contract 字段。"""
|
||
return {"triggered": bool(merge.get("triggered")), "rollRefs": list(merge.get("rollRefs") or []),
|
||
"mergeCandidate": str(merge.get("mergeCandidate") or "tester_error"),
|
||
"conflicts": [str(v) for v in merge.get("conflicts") or []],
|
||
"rescuedByRoll": merge.get("rescuedByRoll"), "reason": str(merge.get("reason") or "")}
|
||
|
||
|
||
def _v3_first_play(facts: dict, *, accepted: bool) -> dict:
|
||
"""保留 runner 已通过 semantic 校验的冷启动实测；终裁只收紧闭环结论，不重算时间。"""
|
||
source = facts.get("firstPlay") if isinstance(facts.get("firstPlay"), dict) else {}
|
||
loop_closed = bool(accepted and source.get("loopClosed") is True)
|
||
return {
|
||
"interactiveAtVirtualMs": source.get("interactiveAtVirtualMs"),
|
||
"firstFeedbackAtVirtualMs": source.get("firstFeedbackAtVirtualMs"),
|
||
"loopClosedAtVirtualMs": source.get("loopClosedAtVirtualMs") if loop_closed else None,
|
||
"loopClosed": loop_closed,
|
||
"proofObligationRefs": list(source.get("proofObligationRefs") or []) if loop_closed else [],
|
||
}
|
||
|
||
|
||
def _v3_empty_judge_slot_ref(roll_no: int, suffix: str) -> dict:
|
||
"""测试器错误尚未产出 Judge 槽时填充显式占位；最终 outcome 仍为 tester_error。"""
|
||
base = f"runner-error/roll-{roll_no}/judge-{suffix}"
|
||
return {"rawRef": f"{base}/raw", "rawHash": "0" * 64,
|
||
"parsedRef": f"{base}/parsed", "parsedHash": "0" * 64,
|
||
"normalizedRef": f"{base}/normalized", "resultHash": "0" * 64}
|
||
|
||
|
||
def _v3_roll_contract(record: dict) -> dict:
|
||
"""把内部 raw/attempts/guard 记录投影成 schema roll。"""
|
||
raw = record.get("raw") if isinstance(record.get("raw"), dict) else {}
|
||
guard = record.get("rollGuard") or {}
|
||
facts = guard.get("facts") or {}
|
||
actor = facts.get("actor") or {}
|
||
roll_no = int(record.get("roll") or raw.get("roll") or 1)
|
||
errors = [str(v) for v in guard.get("errors") or []]
|
||
attempt_cost = sum(float(item.get("costRmb") or 0.0) for item in record.get("attempts") or []
|
||
if isinstance(item, dict))
|
||
return {"rollId": f"roll-{roll_no}", "roll": roll_no,
|
||
"requestedGameSeed": int(raw.get("requestedSeed") or actor.get("requestedGameSeed") or 0),
|
||
"actualGameSeed": int(raw.get("actualSeed") or actor.get("actualGameSeed") or 0),
|
||
"policySeed": int(raw.get("policySeed") or actor.get("policySeed") or 0),
|
||
"actorSessionId": str(actor.get("sessionId") or f"runner-error-roll-{roll_no}"),
|
||
"proofPackageRef": str(raw.get("proofPackageRef") or facts.get("proofPackageRef")
|
||
or f"playtest-v3/roll-{roll_no}/runner-error.json"),
|
||
"proofPackageHash": str(raw.get("proofPackageHash") or "0" * 64),
|
||
"judgePackageRef": str(raw.get("judgePackageRef") or facts.get("judgePackageRef") or "runner-error"),
|
||
"judgePackageHash": str(raw.get("judgePackageHash") or "0" * 64),
|
||
"judgeARef": raw.get("judgeARef") or _v3_empty_judge_slot_ref(roll_no, "a"),
|
||
"judgeAHash": str(raw.get("judgeAHash") or "0" * 64),
|
||
"judgeBRef": raw.get("judgeBRef") or _v3_empty_judge_slot_ref(roll_no, "b"),
|
||
"judgeBHash": str(raw.get("judgeBHash") or "0" * 64),
|
||
"judgeConsensusRef": str(raw.get("judgeConsensusRef") or "runner-error"),
|
||
"judgeConsensusHash": str(raw.get("judgeConsensusHash") or "0" * 64),
|
||
"costReservationRef": str(raw.get("costReservationRef") or "runner-error"),
|
||
"costReservationHash": str(raw.get("costReservationHash") or "0" * 64),
|
||
"rollGuard": _v3_guard_contract(guard),
|
||
"rollOutcome": str(guard.get("rollOutcome") or "tester_error"), "errors": errors,
|
||
"costRmb": _quantize_rmb_v3(
|
||
attempt_cost if record.get("attempts") else float(raw.get("costRmb") or 0.0),
|
||
f"roll-{roll_no}.costRmb",
|
||
)}
|
||
|
||
|
||
def _final_postguard_v3(*, mode: str, floor: dict, merge: dict, rolls: list,
|
||
repair_count: int, cost_rmb: float, cost_cap_rmb: float,
|
||
parent_chain_cost_rmb: float = 0.0, parent_chain_cost_cap_rmb: float = 15.0) -> dict:
|
||
"""v3 唯一 decision 写入口；从原始 guard/facts 重算接受条件，不信 merge 自报。"""
|
||
cost_rmb = _finite_nonnegative_rmb_v3(cost_rmb, "costRmb")
|
||
cost_cap_rmb = _finite_positive_rmb_v3(cost_cap_rmb, "costCapRmb")
|
||
parent_chain_cost_rmb = _finite_nonnegative_rmb_v3(
|
||
parent_chain_cost_rmb, "preAcceptanceChainCostRmb")
|
||
parent_chain_cost_cap_rmb = _finite_positive_rmb_v3(
|
||
parent_chain_cost_cap_rmb, "parentChainCostCapRmb")
|
||
outcome = str(merge.get("mergeCandidate") or "tester_error")
|
||
if outcome not in _V3_OUTCOMES:
|
||
merge = _merge_candidate_v3([]); merge["reason"] = "未知 mergeCandidate"; merge["_subtype"] = "unknown_schema"
|
||
outcome = "tester_error"
|
||
if cost_rmb > cost_cap_rmb + 1e-9:
|
||
outcome = "tester_error"
|
||
merge.update({"mergeCandidate": outcome, "reason": f"验收成本 ¥{cost_rmb:.5f} 超硬帽 ¥{cost_cap_rmb:.2f}",
|
||
"_subtype": "cost_cap_exceeded", "_sourceGuard": None})
|
||
chain_cost = float(parent_chain_cost_rmb or 0.0) + float(cost_rmb or 0.0)
|
||
if chain_cost > parent_chain_cost_cap_rmb + 1e-9:
|
||
outcome = "tester_error"
|
||
merge.update({"mergeCandidate": outcome,
|
||
"reason": f"parentRun 全链成本 ¥{chain_cost:.5f} 超硬帽 ¥{parent_chain_cost_cap_rmb:.2f}",
|
||
"_subtype": "parent_chain_cost_cap_exceeded", "_sourceGuard": None})
|
||
if floor.get("pass") is not True and merge.get("_subtype") != "floor_gate":
|
||
outcome = "tester_error"
|
||
merge.update({"mergeCandidate": outcome, "reason": "四门证据不完整",
|
||
"_subtype": "floor_evidence_missing", "_sourceGuard": None})
|
||
|
||
verified_accept = [record.get("rollGuard") or {} for record in rolls
|
||
if (record.get("rollGuard") or {}).get("rollOutcome") == "accept"
|
||
and (record.get("rollGuard") or {}).get("verified") is True]
|
||
source_guard = merge.get("_sourceGuard") or (verified_accept[-1] if verified_accept else {})
|
||
facts = source_guard.get("facts") or {}
|
||
source_checks = source_guard.get("checks") if isinstance(source_guard.get("checks"), dict) else {}
|
||
final_checks = {key: bool(source_checks.get(key)) for key in _V3_GUARD_CHECKS}
|
||
final_checks["floorPass"] = floor.get("pass") is True
|
||
accept_recomputed = bool(outcome == "accept" and verified_accept and all(final_checks.values()))
|
||
accept_recomputed = accept_recomputed and not merge.get("conflicts")
|
||
if outcome == "accept" and merge.get("rescuedByRoll") == 2:
|
||
second = next(((record.get("rollGuard") or {}) for record in rolls if record.get("roll") == 2), {})
|
||
first = next(((record.get("rollGuard") or {}) for record in rolls if record.get("roll") == 1), {})
|
||
new_evidence = set(second.get("evidenceHashes") or []) - set(first.get("evidenceHashes") or [])
|
||
accept_recomputed = accept_recomputed and bool(new_evidence) and new_evidence == set(merge.get("_newEvidenceHashes") or [])
|
||
if outcome == "accept" and not accept_recomputed:
|
||
outcome = "tester_error"
|
||
merge.update({"mergeCandidate": outcome, "reason": "finalPostguard 无法从原始硬证复算 accept",
|
||
"_subtype": "final_guard_failed", "_sourceGuard": source_guard})
|
||
|
||
shadow = mode == "v3_shadow"
|
||
decision_accepted = outcome == "accept"
|
||
authoritative_accepted = decision_accepted and not shadow
|
||
repair_evidence_complete = bool(
|
||
merge.get("_subtype") == "floor_gate" or source_guard.get("repairEvidenceComplete") is True)
|
||
repair_eligible = bool(outcome == "reject" and merge.get("_verifiedReject")
|
||
and repair_evidence_complete and repair_count == 0)
|
||
failure_layer, failure_reason = _v3_failure_projection(outcome, floor=floor, merge=merge)
|
||
first_play = _v3_first_play(facts, accepted=decision_accepted)
|
||
final_guard = {"pass": decision_accepted, "mergeCandidate": outcome, "checks": final_checks,
|
||
"blockingProblems": [str(v) for v in source_guard.get("blockingProblems") or []],
|
||
"contradictions": [str(v) for v in source_guard.get("contradictions") or []],
|
||
"writeAuthority": True, "reason": merge.get("reason") or ("硬证完整" if decision_accepted else failure_reason or "")}
|
||
repair_feedback = _repair_feedback_v3(merge, rolls) if repair_eligible else None
|
||
decision = {
|
||
"outcome": outcome, "accepted": decision_accepted,
|
||
"shadowAccepted": decision_accepted if shadow else None,
|
||
"publishFrozen": shadow or not decision_accepted,
|
||
"failure": {"layer": failure_layer, "subtype": merge.get("_subtype"), "reason": failure_reason},
|
||
"repairCountAcrossParentChain": repair_count, "repairEligible": repair_eligible,
|
||
"repairFeedback": repair_feedback, "rescuedByRoll": merge.get("rescuedByRoll"),
|
||
"costRmb": _quantize_rmb_v3(cost_rmb),
|
||
"parentChainCostRmb": _quantize_rmb_v3(chain_cost, "parentChainCostRmb"),
|
||
}
|
||
repair = {"countAcrossParentChain": repair_count, "eligible": repair_eligible,
|
||
"attempted": repair_count > 0, "reason": repair_feedback or ("已接受" if decision_accepted else failure_reason or "不可修复")}
|
||
judge_contract = facts.get("judge") if isinstance(facts.get("judge"), dict) else None
|
||
failed_gates = [name for name, result in (floor.get("gates") or {}).items() if result.get("pass") is not True]
|
||
failure_projection = {"layer": failure_layer, "reason": failure_reason, "failedGates": failed_gates}
|
||
judge_projection = {
|
||
"ran": bool(rolls), "verdict": outcome, "accepted": outcome == "accept",
|
||
"degraded": outcome == "tester_error", "reason": failure_reason,
|
||
"acceptanceVersion": mode, "schemaVersion": _PLAYTEST_V3_SCHEMA,
|
||
}
|
||
playtest_projection = {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "outcome": outcome, "accepted": outcome == "accept",
|
||
"rollCount": len(rolls), "rescuedByRoll": merge.get("rescuedByRoll"),
|
||
"costRmb": _quantize_rmb_v3(cost_rmb), "proofComplete": decision_accepted,
|
||
"firstPlay": {"playableAtMs": first_play["interactiveAtVirtualMs"],
|
||
"firstFeedbackMs": first_play["firstFeedbackAtVirtualMs"],
|
||
"loopClosed": first_play["loopClosed"], "proofRefs": first_play["proofObligationRefs"]},
|
||
"evidenceRefs": source_guard.get("evidenceRefs") or {"actions": [], "frames": [], "events": []},
|
||
}
|
||
compatibility = {
|
||
"accepted": authoritative_accepted,
|
||
"ok": authoritative_accepted,
|
||
"acceptanceVersion": mode,
|
||
"publishFrozen": shadow or not decision_accepted,
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA,
|
||
"sourceRollId": (judge_contract or {}).get("sourceRollId"),
|
||
}
|
||
return {"outcome": outcome, "merge": merge, "decision": decision, "repair": repair,
|
||
"compatibility": compatibility, "finalPostguard": final_guard, "firstPlay": first_play, "facts": facts}
|
||
|
||
|
||
async def run_playtest_v3(request: dict, *, run_dir, artifact_hash: str, brief_hash: str,
|
||
cfg=None) -> dict:
|
||
"""执行 v3 一至两掷、基础设施原条件重试和成本硬帽；返回 rollGuard 与 mergeCandidate。"""
|
||
cfg = cfg or _acceptance_v3_cfg()
|
||
game_id = str(request.get("gameId") or "")
|
||
brief = str(request.get("brief") or "")
|
||
identity = request.get("acceptanceIdentity") if isinstance(request.get("acceptanceIdentity"), dict) else {}
|
||
provenance = request.get("acceptanceProvenance") if isinstance(request.get("acceptanceProvenance"), dict) else {}
|
||
genre = str(identity.get("genre") or "unknown")
|
||
template_route = str(identity.get("templateRoute") or "")
|
||
proof_profile_id = str(identity.get("proofProfileId") or "")
|
||
proof_registry_version = str(identity.get("proofRegistryVersion") or "")
|
||
task_binding_hash = str(identity.get("taskBindingHash") or "")
|
||
interaction_binding = (identity.get("interactionBinding")
|
||
if isinstance(identity.get("interactionBinding"), dict) else None)
|
||
acceptance_request_hash = str(identity.get("acceptanceRequestHash") or "")
|
||
acceptance_mode = str(request.get("acceptanceMode") or cfg.get("mode") or "v3_shadow")
|
||
evidence_mode = str(request.get("evidenceMode") or "native")
|
||
run_id = str(request.get("runId") or Path(run_dir).name)
|
||
seed_value = request.get("gameSeed") if request.get("gameSeed") is not None else time.time_ns() % 1_000_000_000
|
||
if request.get("policySeed") is not None:
|
||
policy_seed_value = request.get("policySeed")
|
||
elif isinstance(seed_value, int) and not isinstance(seed_value, bool):
|
||
policy_seed_value = seed_value ^ 0x5F3759DF
|
||
else:
|
||
policy_seed_value = None
|
||
valid_seed = isinstance(seed_value, int) and not isinstance(seed_value, bool) and 0 <= seed_value <= 0xFFFFFFFF
|
||
valid_policy_seed = (isinstance(policy_seed_value, int) and not isinstance(policy_seed_value, bool)
|
||
and 0 <= policy_seed_value <= 0xFFFFFFFF)
|
||
seed = int(seed_value) if valid_seed else 0
|
||
policy_seed = int(policy_seed_value) if valid_policy_seed else 0
|
||
rolls = []
|
||
guards = []
|
||
total_cost = 0.0
|
||
try:
|
||
registry = _v3_resolve_registry(
|
||
genre, brief, template_route=template_route, proof_profile_id=proof_profile_id,
|
||
proof_registry_version=proof_registry_version,
|
||
)
|
||
required_provenance = (
|
||
"acceptanceRequestFile", "provenanceManifestFile", "provenanceManifestHash",
|
||
)
|
||
if any(not provenance.get(field) for field in required_provenance):
|
||
raise ValueError("acceptance provenance 缺 request/manifest 路径或 manifest hash")
|
||
binding_provenance_fields = ("interactionBindingFile", "interactionBindingFileHash")
|
||
if interaction_binding is None:
|
||
if any(provenance.get(field) is not None for field in binding_provenance_fields):
|
||
raise ValueError("interactionBinding=null 时不得携带独立 binding 文件或 hash")
|
||
elif any(not provenance.get(field) for field in binding_provenance_fields):
|
||
raise ValueError("非空 interactionBinding 缺独立 binding 文件或 hash")
|
||
except Exception as e: # noqa: BLE001 —— 未知品类/坏注册表是 tester_error，禁止猜测义务后继续
|
||
return {"rolls": [], "rollGuards": [],
|
||
"merge": {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error",
|
||
"conflicts": [], "rescuedByRoll": None,
|
||
"reason": f"proof registry 解析失败:{type(e).__name__}: {e}",
|
||
"_subtype": "schema_error", "_sourceGuard": None},
|
||
"costRmb": 0.0, "registry": {"version": "unknown", "briefRuleMatches": [], "obligations": []}}
|
||
|
||
if not valid_seed or not valid_policy_seed:
|
||
return {"rolls": [], "rollGuards": [],
|
||
"merge": {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error",
|
||
"conflicts": [], "rescuedByRoll": None,
|
||
"reason": "gameSeed/policySeed 必须是 uint32 整数",
|
||
"_subtype": "seed_mismatch", "_sourceGuard": None},
|
||
"costRmb": 0.0, "registry": registry}
|
||
|
||
# brief 走临时文件，避免命令行转义和长度截断；两掷及 infra retry 读取同一真实题面。
|
||
brief_file = None
|
||
with tempfile.NamedTemporaryFile("w", suffix=".brieftxt", delete=False, encoding="utf-8") as bf:
|
||
bf.write(brief)
|
||
brief_file = bf.name
|
||
explicit_port = request.get("port")
|
||
explicit_cdp_port = request.get("cdpPort")
|
||
|
||
def ports_for_attempt(attempt_no: int) -> tuple:
|
||
derived_port, derived_cdp = _derive_playtest_v3_ports(run_id, attempt_no)
|
||
# 显式端口只约束首尝试；若它冲突，infra retry 必须换到确定性的备用槽。
|
||
if attempt_no == 0:
|
||
return (int(explicit_port) if explicit_port is not None else derived_port,
|
||
int(explicit_cdp_port) if explicit_cdp_port is not None else derived_cdp)
|
||
return derived_port, derived_cdp
|
||
|
||
async def _execute(roll_no: int, game_seed: int, actor_seed: int, attempt_no: int):
|
||
remaining = max(0.0, cfg["cost_cap_rmb"] - total_cost)
|
||
port, cdp_port = ports_for_attempt(attempt_no)
|
||
raw = await asyncio.to_thread(
|
||
_run_playtest_v3_roll_sync, game_id, run_id=run_id, roll=roll_no,
|
||
requested_seed=game_seed, policy_seed=actor_seed, brief_hash=brief_hash,
|
||
artifact_hash=artifact_hash, brief_file=brief_file, genre=genre,
|
||
template_route=template_route, proof_profile_id=proof_profile_id,
|
||
proof_registry_version=proof_registry_version,
|
||
task_binding_hash=task_binding_hash,
|
||
acceptance_request_hash=acceptance_request_hash,
|
||
acceptance_request_file=str(provenance["acceptanceRequestFile"]),
|
||
provenance_manifest_file=str(provenance["provenanceManifestFile"]),
|
||
provenance_manifest_hash=str(provenance["provenanceManifestHash"]),
|
||
acceptance_mode=acceptance_mode, evidence_mode=evidence_mode,
|
||
evidence_dir=str(run_dir), model_name=cfg["model"], port=port, cdp_port=cdp_port,
|
||
timeout=cfg["timeout_s"], remaining_cost_rmb=remaining, cfg=cfg,
|
||
legacy_mapping_file=request.get("legacyMappingFile"),
|
||
interaction_binding_file=provenance.get("interactionBindingFile"),
|
||
interaction_binding_file_hash=provenance.get("interactionBindingFileHash"),
|
||
interaction_profile_id=(interaction_binding or {}).get("interactionProfileId"),
|
||
interaction_registry_version=(interaction_binding or {}).get("interactionRegistryVersion"),
|
||
interaction_binding_hash=(interaction_binding or {}).get("interactionBindingHash"))
|
||
raw["_evidenceDir"] = str(Path(run_dir) / f"roll-{roll_no}")
|
||
return raw
|
||
|
||
async def _with_infra_retry(roll_no: int, game_seed: int, actor_seed: int):
|
||
nonlocal total_cost
|
||
attempts = []
|
||
raw = await _execute(roll_no, game_seed, actor_seed, 0)
|
||
attempts.append(raw)
|
||
total_cost += float(raw.get("costRmb") or 0.0)
|
||
guard = _roll_guard_v3(raw, artifact_hash=artifact_hash, brief_hash=brief_hash,
|
||
floor_pass=True, registry=registry, cfg=cfg,
|
||
task_binding_hash=task_binding_hash,
|
||
acceptance_request_hash=acceptance_request_hash,
|
||
evidence_mode=evidence_mode)
|
||
if (guard.get("rollOutcome") == "tester_error" and guard.get("infraRetryable")
|
||
and cfg.get("infra_retry", 1) > 0 and total_cost < cfg["cost_cap_rmb"]):
|
||
# seed/policy/run/产物保持不变，仅端口槽按 attempt=1 切换，避免端口冲突原样复撞。
|
||
raw = await _execute(roll_no, game_seed, actor_seed, 1)
|
||
attempts.append(raw)
|
||
total_cost += float(raw.get("costRmb") or 0.0)
|
||
guard = _roll_guard_v3(raw, artifact_hash=artifact_hash, brief_hash=brief_hash,
|
||
floor_pass=True, registry=registry, cfg=cfg,
|
||
task_binding_hash=task_binding_hash,
|
||
acceptance_request_hash=acceptance_request_hash,
|
||
evidence_mode=evidence_mode)
|
||
return raw, guard, attempts
|
||
|
||
try:
|
||
raw1, guard1, attempts1 = await _with_infra_retry(1, seed, policy_seed)
|
||
rolls.append({"roll": 1, "raw": raw1, "attempts": attempts1, "rollGuard": guard1})
|
||
guards.append(guard1)
|
||
if (guard1["rollOutcome"] in ("reject", "inconclusive") and cfg.get("second_roll", True)
|
||
and total_cost < cfg["cost_cap_rmb"]):
|
||
second_seed = (seed + 7919) & 0xFFFFFFFF
|
||
second_policy_seed = (policy_seed + 104729) & 0xFFFFFFFF
|
||
assert second_seed != seed and second_policy_seed != policy_seed
|
||
raw2, guard2, attempts2 = await _with_infra_retry(2, second_seed, second_policy_seed)
|
||
rolls.append({"roll": 2, "raw": raw2, "attempts": attempts2, "rollGuard": guard2})
|
||
guards.append(guard2)
|
||
merge = _merge_candidate_v3(guards)
|
||
if total_cost > cfg["cost_cap_rmb"]:
|
||
merge.update({"mergeCandidate": "tester_error", "_subtype": "cost_cap_exceeded",
|
||
"reason": f"验收成本 ¥{total_cost:.5f} 超硬帽 ¥{cfg['cost_cap_rmb']:.2f}",
|
||
"_sourceGuard": None})
|
||
return {"rolls": rolls, "rollGuards": guards, "merge": merge,
|
||
"costRmb": _quantize_rmb_v3(total_cost), "registry": registry}
|
||
finally:
|
||
try:
|
||
Path(brief_file).unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _v3_floor_contract(verdict) -> tuple:
|
||
"""把旧九门中的 A/B/C/D 投影成 contract gateResult，并区分明确失败与证据缺失。"""
|
||
guards = verdict.get("guards") if isinstance(verdict, dict) and isinstance(verdict.get("guards"), dict) else {}
|
||
gates = {}
|
||
missing = []
|
||
for name in _FLOOR_GATES:
|
||
source = guards.get(name)
|
||
if not isinstance(source, dict) or "pass" not in source:
|
||
missing.append(name)
|
||
source = {}
|
||
gates[name] = {"pass": source.get("pass") is True,
|
||
"evidenceRef": f"floor/verdict.json#/guards/{name}",
|
||
"measurement": source}
|
||
return {"pass": not missing and all(result["pass"] for result in gates.values()), "gates": gates}, missing
|
||
|
||
|
||
def _v3_default_environment(artifact_hash: str, cfg: dict) -> dict:
|
||
"""无 Actor 的机械拒绝/测试器错误仍需完整环境结构；unknown 值明确表示未取到，不冒充健康证据。"""
|
||
return {"host": "unknown-runner", "chromeVersion": "unknown",
|
||
"viewport": {"width": 390, "height": 844, "dpr": 1}, "buildRef": artifact_hash,
|
||
"virtualTime": {"mode": "cdp", "actionQuantumMs": cfg["action_quantum_ms"],
|
||
"waitMinMs": cfg["wait_min_ms"], "waitMaxMs": cfg["wait_max_ms"]}}
|
||
|
||
|
||
def _v3_assemble_payload(base: dict, *, floor: dict, play: dict, final: dict,
|
||
registry: dict, cfg: dict) -> dict:
|
||
"""组 DECIDED 完整封存对象；所有字段都来自 floor、选中单掷 facts 或显式 unknown 降级。"""
|
||
facts = final.get("facts") if isinstance(final.get("facts"), dict) else {}
|
||
rolls = [_v3_roll_contract(record) for record in play.get("rolls") or []]
|
||
top_cost = _quantize_rmb_v3(
|
||
sum(float(record.get("costRmb") or 0.0) for record in rolls), "playtest.costRmb")
|
||
decision = dict(final["decision"]); decision["costRmb"] = top_cost
|
||
compatibility = dict(final["compatibility"])
|
||
proof_obligations = list(facts.get("proofObligations") or (
|
||
registry.get("obligations") if floor.get("pass") is True else []))
|
||
# registry 内的 evidence selector 只供 Python guard 复核，不属于顶层 proofObligation 契约字段。
|
||
proof_obligations = [{key: value for key, value in row.items() if key != "evidence"}
|
||
for row in proof_obligations if isinstance(row, dict)]
|
||
return {
|
||
**base, "outcome": final["outcome"],
|
||
"environment": facts.get("environment") or _v3_default_environment(base["artifactHash"], cfg),
|
||
"floor": floor, "actor": facts.get("actor"), "actions": list(facts.get("actions") or []),
|
||
"events": list(facts.get("events") or []),
|
||
"proofRegistryVersion": str(facts.get("proofRegistryVersion") or registry.get("version") or "unknown"),
|
||
"briefRuleMatches": list(facts.get("briefRuleMatches") or registry.get("briefRuleMatches") or []),
|
||
"proofObligations": proof_obligations,
|
||
"firstPlay": final["firstPlay"], "rolls": rolls, "merge": _v3_merge_contract(final["merge"]),
|
||
"sourceRollId": (facts.get("judge") or {}).get("sourceRollId"),
|
||
"judge": facts.get("judge"), "repair": final["repair"],
|
||
"costRmb": top_cost, "finalPostguard": final["finalPostguard"], "decision": decision,
|
||
"compatibility": compatibility,
|
||
}
|
||
|
||
|
||
def validate_acceptance_v3_payload(payload: dict) -> list:
|
||
"""调用 canonical validate.py 的 schema + semantic 校验；运行时与测试共用同一权威规则。"""
|
||
import importlib.util # noqa: PLC0415 —— 避免模块顶层把 contracts 变成 Python 包依赖
|
||
try:
|
||
validator_path = _V3_CONTRACT_DIR / "validate.py"
|
||
spec = importlib.util.spec_from_file_location("play_loop_validate_runtime", validator_path)
|
||
if spec is None or spec.loader is None:
|
||
return ["无法加载 canonical validate.py"]
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
schema = json.loads(_V3_EVIDENCE_SCHEMA.read_text(encoding="utf-8"))
|
||
errors = module.validate(schema, payload, schema)
|
||
if not errors:
|
||
errors += module._semantic_validate(schema, payload)
|
||
return errors
|
||
except Exception as e: # noqa: BLE001 —— 校验器自身不可用必须 fail-closed，消费方不应因异常放行
|
||
return [f"canonical v3 validator 不可用:{type(e).__name__}: {e}"]
|
||
|
||
|
||
def is_v3_publishable(payload: dict) -> bool:
|
||
"""active v3 唯一发布谓词；缺失、shadow、四态非 accept 或语义脏对象一律 false。"""
|
||
if validate_acceptance_v3_payload(payload):
|
||
return False
|
||
decision = payload.get("decision") or {}
|
||
compatibility = payload.get("compatibility") or {}
|
||
return bool(payload.get("acceptanceMode") == "v3" and payload.get("outcome") == "accept"
|
||
and decision.get("accepted") is True and decision.get("publishFrozen") is False
|
||
and compatibility.get("accepted") is True and compatibility.get("ok") is True
|
||
and compatibility.get("publishFrozen") is False)
|
||
|
||
|
||
def is_v3_repair_authorized(payload: dict) -> bool:
|
||
"""唯一修复授权谓词；除 contract 合规外，还复核 mechanical/gameplay 硬证与 repair 镜像。"""
|
||
if validate_acceptance_v3_payload(payload):
|
||
return False
|
||
decision = payload.get("decision") or {}
|
||
repair = payload.get("repair") or {}
|
||
merge = payload.get("merge") or {}
|
||
final_guard = payload.get("finalPostguard") or {}
|
||
feedback = decision.get("repairFeedback")
|
||
mirrors = (payload.get("outcome") == "reject" and merge.get("mergeCandidate") == "reject"
|
||
and final_guard.get("mergeCandidate") == "reject" and final_guard.get("pass") is False
|
||
and decision.get("repairEligible") is True and repair.get("eligible") is True
|
||
and decision.get("repairCountAcrossParentChain") == 0
|
||
and repair.get("countAcrossParentChain") == 0 and repair.get("attempted") is False
|
||
and isinstance(feedback, str) and feedback and repair.get("reason") == feedback)
|
||
if not mirrors:
|
||
return False
|
||
floor = payload.get("floor") or {}
|
||
failed_gates = [name for name, result in (floor.get("gates") or {}).items() if result.get("pass") is not True]
|
||
if floor.get("pass") is not True:
|
||
return bool(failed_gates and all(f"floor/verdict.json#/guards/{name}" in feedback for name in failed_gates))
|
||
failed = [row for row in payload.get("proofObligations") or []
|
||
if row.get("required") is True and row.get("status") == "failed"]
|
||
return bool(failed and all(row.get("actionRefs") and row.get("postFrameRefs") and row.get("eventRefs")
|
||
for row in failed))
|
||
|
||
|
||
def _safe_v3_frame_ref(value) -> str:
|
||
"""把契约内截图引用收敛为 run 目录内的规范 POSIX 相对路径。"""
|
||
if not isinstance(value, str) or not value or "\\" in value or "\x00" in value:
|
||
raise ValueError(f"截图引用不是安全相对路径:{value!r}")
|
||
path = PurePosixPath(value)
|
||
if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts):
|
||
raise ValueError(f"截图引用越界:{value}")
|
||
return path.as_posix()
|
||
|
||
|
||
def _verify_v3_sealed_frames(payload: dict, snapshot: artifact_snapshot.ArtifactSnapshot) -> None:
|
||
"""复核封存 decision 中全部 action pre/post 截图的真实字节与 SHA-256。"""
|
||
roll_ids = []
|
||
for roll in payload.get("rolls") or []:
|
||
roll_id = roll.get("rollId") if isinstance(roll, dict) else None
|
||
if not isinstance(roll_id, str) or not _is_safe_game_id_v3(roll_id):
|
||
raise ValueError(f"封存结果 rollId 非法:{roll_id!r}")
|
||
roll_ids.append(roll_id)
|
||
|
||
post_refs = {}
|
||
for action in payload.get("actions") or []:
|
||
if not isinstance(action, dict):
|
||
raise ValueError("封存结果 action 不是对象")
|
||
action_id = str(action.get("actionId") or "")
|
||
for field in ("preFrameRef", "postFrameRef"):
|
||
frame = action.get(field) if isinstance(action.get(field), dict) else {}
|
||
relative = _safe_v3_frame_ref(frame.get("path"))
|
||
expected_hash = frame.get("hash")
|
||
if not _is_sha256_v3(expected_hash):
|
||
raise ValueError(f"封存截图 hash 非法:{action_id}/{field}")
|
||
matched = False
|
||
for roll_id in roll_ids:
|
||
data = snapshot.files.get(f"{roll_id}/{relative}")
|
||
if data is not None and hashlib.sha256(data).hexdigest() == expected_hash:
|
||
matched = True
|
||
break
|
||
if not matched:
|
||
raise ValueError(f"封存截图缺失、越界或 hash 不一致:{action_id}/{field}:{relative}")
|
||
if field == "postFrameRef":
|
||
post_refs[relative] = expected_hash
|
||
|
||
# proof obligation 只允许引用已经逐字节复核过的 action post frame，不能另指一条未审计路径。
|
||
for obligation in payload.get("proofObligations") or []:
|
||
if not isinstance(obligation, dict):
|
||
continue
|
||
for value in obligation.get("postFrameRefs") or []:
|
||
relative = _safe_v3_frame_ref(value)
|
||
if relative not in post_refs:
|
||
raise ValueError(f"proof obligation 引用了未复核截图:{relative}")
|
||
|
||
|
||
def _load_verified_v3_sealed(evidence_root: Path, run_dir: Path) -> dict:
|
||
"""从不可变 run 快照加载 decision，并复核契约、目录边界和全部截图证据。"""
|
||
evidence_root = _assert_no_symlink_components(evidence_root, require_exists=True)
|
||
run_dir = _assert_no_symlink_components(run_dir, require_exists=True)
|
||
if run_dir.parent != evidence_root:
|
||
raise ValueError("decision runDir 不在声明的 evidenceRoot 直接子目录")
|
||
snapshot = artifact_snapshot.capture_artifact_snapshot(run_dir)
|
||
decision_bytes = snapshot.files.get("decision.json")
|
||
if decision_bytes is None:
|
||
raise ValueError("decision.json 不存在或不是普通文件")
|
||
try:
|
||
payload = json.loads(decision_bytes.decode("utf-8"))
|
||
except (UnicodeError, ValueError) as exc:
|
||
raise ValueError(f"decision.json 不可解析:{type(exc).__name__}: {exc}") from exc
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("decision.json 顶层不是对象")
|
||
errors = validate_acceptance_v3_payload(payload)
|
||
if errors:
|
||
raise ValueError("封存结果 contract/semantic 非法:" + ";".join(errors[:6]))
|
||
if payload.get("runId") != run_dir.name:
|
||
raise ValueError("封存结果 runId 与目录不一致")
|
||
_verify_v3_sealed_frames(payload, snapshot)
|
||
return payload
|
||
|
||
|
||
def _verify_v3_sealed_reference_provenance(
|
||
run_dir: Path, identity: dict, artifact_hash: str,
|
||
expected_receipts: list[dict] | None) -> None:
|
||
"""复核 sealed run 的 /3 或 /4 provenance，并把 /4 回执闭包与本次独立复验全等对账。"""
|
||
run_dir = _assert_no_symlink_components(run_dir, require_exists=True)
|
||
snapshot = artifact_snapshot.capture_artifact_snapshot(run_dir)
|
||
manifest_bytes = snapshot.files.get("acceptance-provenance.json")
|
||
if manifest_bytes is None:
|
||
raise ValueError("sealed run 缺 acceptance-provenance.json")
|
||
try:
|
||
manifest = json.loads(manifest_bytes.decode("utf-8"))
|
||
except (UnicodeError, ValueError) as exc:
|
||
raise ValueError(f"sealed acceptance provenance 不可解析:{type(exc).__name__}: {exc}") from exc
|
||
if not isinstance(manifest, dict):
|
||
raise ValueError("sealed acceptance provenance 顶层不是对象")
|
||
try:
|
||
canonical_manifest = _stable_json_bytes_v3(manifest)
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError(f"sealed acceptance provenance 不是 canonical JSON:{exc}") from exc
|
||
if canonical_manifest != manifest_bytes:
|
||
raise ValueError("sealed acceptance provenance 不是 canonical JSON")
|
||
schema = manifest.get("schemaVersion")
|
||
if schema == "acceptance-provenance/4":
|
||
schema_errors = _validate_v3_schema_only(_V4_ACCEPTANCE_PROVENANCE_SCHEMA, manifest)
|
||
elif schema == "acceptance-provenance/3":
|
||
schema_errors = _validate_v3_instance(_V3_ACCEPTANCE_PROVENANCE_SCHEMA, manifest)
|
||
else:
|
||
raise ValueError(f"sealed acceptance provenance schema 未知:{schema!r}")
|
||
if schema_errors:
|
||
raise ValueError("sealed acceptance provenance contract 非法:" + ";".join(schema_errors[:6]))
|
||
|
||
for field in (
|
||
"gameId", "briefHash", "genre", "templateRoute", "proofProfileId",
|
||
"proofRegistryVersion", "taskBindingHash", "interactionBinding", "acceptanceRequestHash",
|
||
"sourceArtifactHash", "parentAcceptanceRequestHash", "repairOrdinal",
|
||
"designRef", "referenceAssetRecordIds", "consumerRef",
|
||
):
|
||
if manifest.get(field) != identity.get(field):
|
||
raise ValueError(f"sealed provenance {field} 与当前 acceptance identity 不一致")
|
||
expected_binding = identity.get("interactionBinding")
|
||
binding_bytes = snapshot.files.get("interaction-binding.json")
|
||
if expected_binding is None:
|
||
if binding_bytes is not None:
|
||
raise ValueError("sealed provenance 的 null interactionBinding 不得残留独立 binding 文件")
|
||
elif binding_bytes != _stable_json_bytes_v3(expected_binding):
|
||
raise ValueError("sealed provenance 的独立 interaction binding 缺失或漂移")
|
||
if manifest.get("artifactHash") != artifact_hash:
|
||
raise ValueError("sealed provenance artifactHash 与当前产物不一致")
|
||
|
||
if schema == "acceptance-provenance/3":
|
||
if expected_receipts is not None:
|
||
raise ValueError("当前请求要求 /4，但 sealed provenance 仍是 /3")
|
||
return
|
||
if expected_receipts is None:
|
||
raise ValueError("当前请求未提供 policy，但 sealed provenance 含 /4 回执")
|
||
|
||
links = manifest.get("referenceAssetVerificationReceipts")
|
||
if not isinstance(links, list) or not links:
|
||
raise ValueError("sealed provenance /4 缺非空回执链接")
|
||
persisted_receipts = []
|
||
seen_receipt_ids = set()
|
||
for index, link in enumerate(links):
|
||
if not isinstance(link, dict) or set(link) != {"ref", "hash"}:
|
||
raise ValueError(f"sealed provenance /4 回执链接[{index}] 形状非法")
|
||
ref = link.get("ref")
|
||
if not isinstance(ref, str):
|
||
raise ValueError(f"sealed provenance /4 回执链接[{index}] ref 非法")
|
||
try:
|
||
relative = _safe_v3_frame_ref(ref)
|
||
except ValueError as exc:
|
||
raise ValueError(f"sealed provenance /4 回执链接[{index}] 越界:{exc}") from exc
|
||
receipt_bytes = snapshot.files.get(relative)
|
||
if receipt_bytes is None:
|
||
raise ValueError(f"sealed provenance /4 回执缺失:{relative}")
|
||
if not _is_sha256_v3(link.get("hash")):
|
||
raise ValueError(f"sealed provenance /4 回执 hash 非法:{relative}")
|
||
if hashlib.sha256(receipt_bytes).hexdigest() != link["hash"]:
|
||
raise ValueError(f"sealed provenance /4 回执 hash 漂移:{relative}")
|
||
try:
|
||
receipt = json.loads(receipt_bytes.decode("utf-8"))
|
||
except (UnicodeError, ValueError) as exc:
|
||
raise ValueError(f"sealed provenance /4 回执不可解析:{relative}") from exc
|
||
if not isinstance(receipt, dict):
|
||
raise ValueError(f"sealed provenance /4 回执顶层不是对象:{relative}")
|
||
receipt_errors = _validate_v3_schema_only(_REFERENCE_ASSET_RECEIPT_SCHEMA, receipt)
|
||
if receipt_errors:
|
||
raise ValueError(f"sealed provenance /4 回执 contract 非法:{relative}")
|
||
if _stable_json_bytes_v3(receipt) != receipt_bytes:
|
||
raise ValueError(f"sealed provenance /4 回执不是 canonical JSON:{relative}")
|
||
receipt_id = receipt.get("receiptId")
|
||
expected_ref = f"reference-asset-verification-receipts/{receipt_id}.json"
|
||
if not _is_v3_record_id(receipt_id) or ref != expected_ref:
|
||
raise ValueError(f"sealed provenance /4 回执 ref 与 receiptId 不一致:{relative}")
|
||
if receipt_id in seen_receipt_ids:
|
||
raise ValueError(f"sealed provenance /4 receiptId 重复:{receipt_id}")
|
||
seen_receipt_ids.add(receipt_id)
|
||
persisted_receipts.append(receipt)
|
||
|
||
if _stable_json_bytes_v3(persisted_receipts) != _stable_json_bytes_v3(expected_receipts):
|
||
raise ValueError("sealed provenance /4 回执与当前验收独立复验结果不一致")
|
||
|
||
|
||
def _v3_tester_error_payload(base: dict, floor: dict, registry: dict, cfg: dict,
|
||
request: dict, *, reason: str, subtype: str,
|
||
pre_acceptance_chain_cost_rmb: float = 0.0) -> dict:
|
||
"""构造不落盘的 canonical tester_error，用于坏封存、证据越界和 run 锁超时。"""
|
||
merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error", "conflicts": [],
|
||
"rescuedByRoll": None, "reason": reason, "_subtype": subtype, "_sourceGuard": None}
|
||
play = {"rolls": [], "merge": merge, "costRmb": 0.0, "registry": registry}
|
||
final = _final_postguard_v3(
|
||
mode=str(base.get("acceptanceMode") or "v3_shadow"), floor=floor, merge=merge, rolls=[],
|
||
repair_count=int(request.get("repairCountAcrossParentChain") or 0),
|
||
cost_rmb=0.0, cost_cap_rmb=cfg["cost_cap_rmb"],
|
||
parent_chain_cost_rmb=pre_acceptance_chain_cost_rmb,
|
||
parent_chain_cost_cap_rmb=cfg["parent_chain_cost_cap_rmb"])
|
||
payload = _v3_assemble_payload(base, floor=floor, play=play, final=final,
|
||
registry=registry, cfg=cfg)
|
||
errors = validate_acceptance_v3_payload(payload)
|
||
if errors:
|
||
raise ValueError("tester_error 降级结果仍不合规:" + ";".join(errors[:6]))
|
||
return payload
|
||
|
||
|
||
def _persist_v3_decision(run_dir: Path, payload: dict) -> None:
|
||
"""原子封存 v3 decision；已存在结果不覆盖，保证证据只读和幂等。"""
|
||
errors = validate_acceptance_v3_payload(payload)
|
||
if errors:
|
||
raise ValueError("playtest/3 封存前 contract 校验失败:" + ";".join(errors[:6]))
|
||
run_dir.mkdir(parents=True, exist_ok=True)
|
||
target = run_dir / "decision.json"
|
||
if target.exists():
|
||
return
|
||
tmp = run_dir / "decision.json.tmp"
|
||
tmp.write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False), encoding="utf-8")
|
||
tmp.replace(target)
|
||
|
||
|
||
def _validate_v3_repair_parent(request: dict, identity: dict,
|
||
evidence_root: Path) -> tuple[str | None, float]:
|
||
"""复核唯一修回父决策，并返回封存父链成本；调用方不得自报或重置历史成本。"""
|
||
ordinal = int(identity.get("repairOrdinal") or 0)
|
||
parent_run_id = request.get("parentRunId")
|
||
if ordinal == 0:
|
||
return ((None, 0.0) if parent_run_id is None
|
||
else ("首轮 acceptance 不得携带 parentRunId", 0.0))
|
||
if not isinstance(parent_run_id, str) or not _is_safe_game_id_v3(parent_run_id):
|
||
return "repairOrdinal=1 必须携带安全 parentRunId", 0.0
|
||
try:
|
||
parent = _load_verified_v3_sealed(
|
||
Path(evidence_root), Path(evidence_root) / parent_run_id)
|
||
except Exception as exc: # noqa: BLE001
|
||
return f"修回父 decision/截图证据不可复核:{type(exc).__name__}: {exc}", 0.0
|
||
if not is_v3_repair_authorized(parent):
|
||
return "父决策未授权唯一修回", 0.0
|
||
parent_decision = parent.get("decision") or {}
|
||
checks = {
|
||
"sourceArtifactHash": (identity.get("sourceArtifactHash"), parent.get("artifactHash")),
|
||
"parentAcceptanceRequestHash": (
|
||
identity.get("parentAcceptanceRequestHash"), parent.get("acceptanceRequestHash")),
|
||
"gameId": (identity.get("gameId"), parent.get("gameId")),
|
||
"briefHash": (identity.get("briefHash"), parent.get("briefHash")),
|
||
"genre": (identity.get("genre"), parent.get("genre")),
|
||
"templateRoute": (identity.get("templateRoute"), parent.get("templateRoute")),
|
||
"proofProfileId": (identity.get("proofProfileId"), parent.get("proofProfileId")),
|
||
"proofRegistryVersion": (
|
||
identity.get("proofRegistryVersion"), parent.get("proofRegistryVersion")),
|
||
"taskBindingHash": (identity.get("taskBindingHash"), parent.get("taskBindingHash")),
|
||
}
|
||
try:
|
||
parent_snapshot = artifact_snapshot.capture_artifact_snapshot(Path(evidence_root) / parent_run_id)
|
||
parent_request_ref = f"acceptance-request-{parent.get('acceptanceRequestHash')}.json"
|
||
parent_request_bytes = parent_snapshot.files.get(parent_request_ref)
|
||
if (parent_request_bytes is None
|
||
or hashlib.sha256(parent_request_bytes).hexdigest() != parent.get("acceptanceRequestHash")):
|
||
raise ValueError("父 acceptance request 缺失或 hash 漂移")
|
||
parent_request = json.loads(parent_request_bytes.decode("utf-8"))
|
||
if (_stable_json_bytes_v3(parent_request) != parent_request_bytes
|
||
or _validate_v3_instance(_V3_ACCEPTANCE_REQUEST_SCHEMA, parent_request)):
|
||
raise ValueError("父 acceptance request 不是 canonical 合法请求")
|
||
except Exception as exc: # noqa: BLE001 interaction binding 血缘只能信封存父 request
|
||
return f"修回父 acceptance request 不可复核:{type(exc).__name__}: {exc}", 0.0
|
||
checks["interactionBinding"] = (
|
||
identity.get("interactionBinding"), parent_request.get("interactionBinding"))
|
||
for field, (actual, expected) in checks.items():
|
||
if actual != expected:
|
||
return f"修回身份 {field} 未锁定父决策", 0.0
|
||
if parent_decision.get("repairCountAcrossParentChain") != 0:
|
||
return "父决策 repairCountAcrossParentChain 必须为 0", 0.0
|
||
try:
|
||
parent_cost = _finite_nonnegative_rmb_v3(
|
||
parent_decision.get("parentChainCostRmb"), "父决策 parentChainCostRmb")
|
||
except ValueError as exc:
|
||
return str(exc), 0.0
|
||
return None, parent_cost
|
||
|
||
|
||
async def _run_acceptance_v3_locked(request: dict) -> dict:
|
||
"""持有当前 run claim 后执行 v3 编排；公开入口负责跨协程/跨进程幂等锁。"""
|
||
request = dict(request or {})
|
||
cfg = _acceptance_v3_cfg()
|
||
config_cost_error = None
|
||
try:
|
||
cfg["cost_cap_rmb"] = _finite_positive_rmb_v3(cfg["cost_cap_rmb"], "costCapRmb")
|
||
cfg["parent_chain_cost_cap_rmb"] = _finite_positive_rmb_v3(
|
||
cfg["parent_chain_cost_cap_rmb"], "parentChainCostCapRmb")
|
||
except ValueError as exc:
|
||
# 错误配置也要能产出 canonical tester_error，故回落批准默认帽，但保留错误原因阻断执行。
|
||
config_cost_error = str(exc)
|
||
cfg["cost_cap_rmb"] = 1.5
|
||
cfg["parent_chain_cost_cap_rmb"] = 15.0
|
||
mode = str(request.get("acceptanceMode") or cfg["mode"])
|
||
game_id = str(request.get("gameId") or "")
|
||
if not _is_safe_game_id_v3(game_id):
|
||
raise ValueError("v3 gameId 非法")
|
||
brief = str(request.get("brief") or "")
|
||
identity = request.get("acceptanceIdentity") if isinstance(request.get("acceptanceIdentity"), dict) else None
|
||
if identity is None:
|
||
raise ValueError("v3 请求缺 Writer 前冻结的 acceptanceIdentity")
|
||
identity = dict(identity)
|
||
identity_fields = set(_canonical_acceptance_request(identity)) | {"acceptanceRequestHash"}
|
||
if set(identity) != identity_fields:
|
||
raise ValueError("acceptanceIdentity 字段集合非法")
|
||
if str(identity.get("gameId") or "") != game_id:
|
||
raise ValueError("acceptanceIdentity.gameId 与入口请求不一致")
|
||
expected_identity = build_acceptance_v3_identity(
|
||
game_id, brief, genre=str(identity.get("genre") or ""),
|
||
template_route=str(identity.get("templateRoute") or ""),
|
||
source_artifact_hash=identity.get("sourceArtifactHash"),
|
||
parent_acceptance_request_hash=identity.get("parentAcceptanceRequestHash"),
|
||
repair_ordinal=int(identity.get("repairOrdinal") or 0),
|
||
proof_profile_id=identity.get("proofProfileId"),
|
||
proof_registry_version=identity.get("proofRegistryVersion"),
|
||
task_binding_hash=identity.get("taskBindingHash"),
|
||
interaction_binding=identity.get("interactionBinding"),
|
||
design_ref=identity.get("designRef"),
|
||
reference_asset_record_ids=identity.get("referenceAssetRecordIds"),
|
||
consumer_ref=identity.get("consumerRef"),
|
||
)
|
||
if identity != expected_identity:
|
||
raise ValueError("acceptanceIdentity 与 canonical route/profile/brief 解析不一致")
|
||
genre = identity["genre"]
|
||
brief_hash = identity["briefHash"]
|
||
evidence_mode = str(request.get("evidenceMode") or "native")
|
||
artifact_hash = str(request.get("artifactHash") or _artifact_hash_v3(game_id, request.get("artifactPath")))
|
||
run_id = _v3_run_id(request, artifact_hash, brief_hash)
|
||
request["runId"] = run_id
|
||
evidence_root = Path(request.get("evidenceRoot") or (_V3_RESULTS_DIR / game_id))
|
||
run_dir = evidence_root / run_id
|
||
floor, missing_floor = _v3_floor_contract(request.get("verdict") or {})
|
||
try:
|
||
registry = _v3_resolve_registry(
|
||
genre, brief, template_route=identity["templateRoute"],
|
||
proof_profile_id=identity["proofProfileId"],
|
||
proof_registry_version=identity["proofRegistryVersion"],
|
||
)
|
||
except Exception as e: # noqa: BLE001 —— 未知品类绝不猜测义务；仍以 tester_error 响亮失败
|
||
registry = {"version": "unknown", "briefRuleMatches": [], "obligations": []}
|
||
registry_error = f"proof registry 解析失败:{type(e).__name__}: {e}"
|
||
else:
|
||
registry_error = None
|
||
# 显式 policy 只信验收侧独立重跑 gate 的自产回执；无 policy 才保留 Registry/1 历史兼容对账。
|
||
reference_asset_provenance_error = None
|
||
acceptance_reference_asset_receipts = None
|
||
if ("referenceAssetPolicyId" in request or "referenceAssetGenerationReceipts" in request
|
||
or any(key.startswith("referenceAssetVerificationReceipt") for key in request)):
|
||
(reference_asset_provenance_error, acceptance_reference_asset_receipts,
|
||
consumed_reference_assets) = _reverify_acceptance_reference_asset_policy(
|
||
request, identity, mode)
|
||
reference_asset_error = None
|
||
else:
|
||
reference_asset_error, consumed_reference_assets = _v3_check_declared_reference_assets(identity)
|
||
base = {
|
||
"schemaVersion": _PLAYTEST_V3_SCHEMA, "runId": run_id, "parentRunId": request.get("parentRunId"),
|
||
"gameId": game_id, "genre": genre, "templateRoute": identity["templateRoute"],
|
||
"proofProfileId": identity["proofProfileId"],
|
||
"proofRegistryVersion": identity["proofRegistryVersion"],
|
||
"taskBindingHash": identity["taskBindingHash"],
|
||
"acceptanceRequestHash": identity["acceptanceRequestHash"],
|
||
"artifactHash": artifact_hash, "briefHash": brief_hash,
|
||
"acceptanceMode": mode, "evidenceMode": evidence_mode, "phase": "DECIDED",
|
||
}
|
||
cost_input_error = None
|
||
if "parentChainCostRmb" in request:
|
||
cost_input_error = "请求不得自报 parentChainCostRmb；repair 历史成本只能从封存父决策推导"
|
||
try:
|
||
writer_cost_rmb = _finite_nonnegative_rmb_v3(
|
||
request.get("writerCostRmb", 0.0), "writerCostRmb")
|
||
except ValueError as exc:
|
||
writer_cost_rmb = 0.0
|
||
cost_input_error = str(exc)
|
||
repair_parent_error, sealed_parent_chain_cost = _validate_v3_repair_parent(
|
||
request, identity, evidence_root)
|
||
pre_acceptance_chain_cost = sealed_parent_chain_cost + writer_cost_rmb
|
||
forced_tester_error = request.get("_forcedV3TesterError")
|
||
if isinstance(forced_tester_error, str) and forced_tester_error:
|
||
return _v3_tester_error_payload(
|
||
base, floor, registry, cfg, request, reason=forced_tester_error,
|
||
subtype="environment_error",
|
||
pre_acceptance_chain_cost_rmb=pre_acceptance_chain_cost)
|
||
sealed = run_dir / "decision.json"
|
||
if os.path.lexists(sealed) and repair_parent_error is None:
|
||
if reference_asset_provenance_error:
|
||
return _v3_tester_error_payload(
|
||
base, floor, registry, cfg, request,
|
||
reason=f"sealed 重入参照资产独立复验失败:{reference_asset_provenance_error}",
|
||
subtype="provenance_error",
|
||
pre_acceptance_chain_cost_rmb=pre_acceptance_chain_cost)
|
||
try:
|
||
existing = _load_verified_v3_sealed(evidence_root, run_dir)
|
||
if existing.get("schemaVersion") != _PLAYTEST_V3_SCHEMA:
|
||
raise ValueError("封存结果 schema 未知")
|
||
sealed_identities = {
|
||
"artifactHash": artifact_hash, "briefHash": brief_hash, "genre": genre,
|
||
"templateRoute": identity["templateRoute"],
|
||
"proofProfileId": identity["proofProfileId"],
|
||
"proofRegistryVersion": identity["proofRegistryVersion"],
|
||
"taskBindingHash": identity["taskBindingHash"],
|
||
"acceptanceRequestHash": identity["acceptanceRequestHash"],
|
||
"evidenceMode": evidence_mode,
|
||
}
|
||
for field, expected in sealed_identities.items():
|
||
if existing.get(field) != expected:
|
||
raise ValueError(f"封存结果 {field} 与请求不一致")
|
||
existing_cost = _finite_nonnegative_rmb_v3(existing.get("costRmb"), "封存 costRmb")
|
||
expected_chain_cost = _quantize_rmb_v3(
|
||
pre_acceptance_chain_cost + existing_cost, "封存 parentChainCostRmb")
|
||
if (existing.get("decision") or {}).get("parentChainCostRmb") != expected_chain_cost:
|
||
raise ValueError("封存结果 parentChainCostRmb 与封存父成本和本轮 writer 成本不一致")
|
||
except Exception as e: # noqa: BLE001
|
||
return _v3_tester_error_payload(
|
||
base, floor, registry, cfg, request,
|
||
reason=f"幂等封存/截图证据复核失败:{type(e).__name__}: {e}",
|
||
subtype="image_error")
|
||
try:
|
||
_verify_v3_sealed_reference_provenance(
|
||
run_dir, identity, artifact_hash, acceptance_reference_asset_receipts)
|
||
except Exception as e: # noqa: BLE001 —— sealed /4 缺失或漂移不得返回旧 accept
|
||
return _v3_tester_error_payload(
|
||
base, floor, registry, cfg, request,
|
||
reason=f"sealed provenance/receipt 复核失败:{type(e).__name__}: {e}",
|
||
subtype="provenance_error",
|
||
pre_acceptance_chain_cost_rmb=pre_acceptance_chain_cost)
|
||
return existing
|
||
|
||
repair_count = int(request.get("repairCountAcrossParentChain") or 0)
|
||
invalid_request = (
|
||
mode not in ("v3_shadow", "v3", "historical_replay") or not game_id
|
||
or genre not in _V3_GENRES or repair_count != identity["repairOrdinal"]
|
||
or (mode in ("v3", "v3_shadow") and evidence_mode != "native")
|
||
or (mode == "historical_replay" and evidence_mode != "legacy-adapted")
|
||
or (identity.get("interactionBinding") is not None and mode != "v3_shadow")
|
||
)
|
||
if (invalid_request or registry_error or repair_parent_error or cost_input_error or config_cost_error
|
||
or reference_asset_provenance_error):
|
||
reason = (reference_asset_provenance_error or config_cost_error or cost_input_error
|
||
or registry_error or repair_parent_error
|
||
or f"非法 v3 请求:mode={mode} gameId={game_id or '-'} genre={genre or '-'}")
|
||
merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error", "conflicts": [],
|
||
"rescuedByRoll": None, "reason": reason,
|
||
"_subtype": ("provenance_error" if (repair_parent_error or reference_asset_provenance_error) else
|
||
"profile_contract_error" if registry_error else "cost_contract_error"
|
||
if (cost_input_error or config_cost_error) else "schema_error"),
|
||
"_sourceGuard": None}
|
||
play = {"rolls": [], "merge": merge, "costRmb": 0.0, "registry": registry}
|
||
elif reference_asset_error:
|
||
# 声明消费参照资产但 Registry/1 兼容对账未过 → verified reject + 发布冻结，
|
||
# 不起浏览器、不烧模型；这是身份/策略闸，先于证据评估（floor/playtest）。
|
||
merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "reject", "conflicts": [],
|
||
"rescuedByRoll": None, "reason": reference_asset_error,
|
||
"_subtype": "reference_asset_consumption_rejected",
|
||
"_verifiedReject": True, "_failedObligations": [], "_failedGates": [],
|
||
"_sourceGuard": None}
|
||
play = {"rolls": [], "merge": merge, "costRmb": 0.0, "registry": registry}
|
||
elif missing_floor:
|
||
merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error", "conflicts": [],
|
||
"rescuedByRoll": None, "reason": f"A/B/C/D 四门证据不完整:{','.join(missing_floor)}",
|
||
"_subtype": "floor_evidence_missing", "_sourceGuard": None}
|
||
play = {"rolls": [], "merge": merge, "costRmb": 0.0, "registry": registry}
|
||
elif not floor["pass"]:
|
||
merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "reject", "conflicts": [],
|
||
"rescuedByRoll": None, "reason": "A/B/C/D 明确产物失败", "_subtype": "floor_gate",
|
||
"_verifiedReject": True, "_failedObligations": [],
|
||
"_failedGates": [name for name, result in floor["gates"].items() if result["pass"] is not True],
|
||
"_sourceGuard": None}
|
||
play = {"rolls": [], "merge": merge, "costRmb": 0.0, "registry": registry}
|
||
else:
|
||
try:
|
||
# W-GOLD-LIVE 检查点 3a：验收时点对账的消费快照落 provenance（未声明消费时为 []，旧行为不变）。
|
||
provenance = write_acceptance_v3_provenance(
|
||
game_id, identity, artifact_hash=artifact_hash, evidence_root=run_dir,
|
||
artifact_path=request.get("artifactPath"),
|
||
consumed_reference_assets=consumed_reference_assets,
|
||
reference_asset_verification_receipts=acceptance_reference_asset_receipts)
|
||
except Exception as exc: # noqa: BLE001 —— provenance 不可信时禁止启动浏览器与模型
|
||
merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error", "conflicts": [],
|
||
"rescuedByRoll": None,
|
||
"reason": f"acceptance provenance 写入失败:{type(exc).__name__}: {exc}",
|
||
"_subtype": "provenance_error", "_sourceGuard": None}
|
||
play = {"rolls": [], "merge": merge, "costRmb": 0.0, "registry": registry}
|
||
else:
|
||
request["acceptanceProvenance"] = provenance
|
||
play = await run_playtest_v3(request, run_dir=run_dir, artifact_hash=artifact_hash,
|
||
brief_hash=brief_hash, cfg=cfg)
|
||
merge = play["merge"]
|
||
final = _final_postguard_v3(
|
||
mode=mode, floor=floor, merge=merge, rolls=play["rolls"], repair_count=repair_count,
|
||
cost_rmb=float(play.get("costRmb") or 0.0), cost_cap_rmb=cfg["cost_cap_rmb"],
|
||
parent_chain_cost_rmb=pre_acceptance_chain_cost,
|
||
parent_chain_cost_cap_rmb=cfg["parent_chain_cost_cap_rmb"])
|
||
payload = _v3_assemble_payload(base, floor=floor, play=play, final=final,
|
||
registry=play.get("registry") or registry, cfg=cfg)
|
||
contract_errors = validate_acceptance_v3_payload(payload)
|
||
if contract_errors:
|
||
reason = "Python finalPostguard 产物未通过 canonical contract:" + ";".join(contract_errors[:6])
|
||
fallback_merge = {"triggered": False, "rollRefs": [], "mergeCandidate": "tester_error",
|
||
"conflicts": [], "rescuedByRoll": None, "reason": reason,
|
||
"_subtype": "schema_error", "_sourceGuard": None}
|
||
fallback_play = {"rolls": [], "merge": fallback_merge, "costRmb": 0.0,
|
||
"registry": play.get("registry") or registry}
|
||
fallback_final = _final_postguard_v3(
|
||
mode=mode, floor=floor, merge=fallback_merge, rolls=[], repair_count=repair_count,
|
||
cost_rmb=0.0, cost_cap_rmb=cfg["cost_cap_rmb"],
|
||
parent_chain_cost_rmb=pre_acceptance_chain_cost,
|
||
parent_chain_cost_cap_rmb=cfg["parent_chain_cost_cap_rmb"])
|
||
payload = _v3_assemble_payload(base, floor=floor, play=fallback_play, final=fallback_final,
|
||
registry=fallback_play["registry"], cfg=cfg)
|
||
fallback_errors = validate_acceptance_v3_payload(payload)
|
||
if fallback_errors:
|
||
raise ValueError("playtest/3 fail-closed 结果仍不合规:" + ";".join(fallback_errors[:6]))
|
||
_persist_v3_decision(run_dir, payload)
|
||
# ── Wave0 运行保障:验收 tester_error 事件实时外发告警(best-effort,绝不阻断验收收口)──────
|
||
# sink = cheap_alert_sink(配置见 generation.yaml alert 区;默认 no-op,创始人填 webhook URL +
|
||
# 置 enabled=true 才启用)。只告 tester_error(基础设施/预算/超时类失败——W-GOLD-LIVE 3b Actor
|
||
# playtest 前撞成本硬帽 cost_cap_exceeded、M3 慢尾 model_timeout 即此类);reject(游戏质量
|
||
# 不过)是正常业务结论,不外发。sink 内部已吞全部异常,这里再裹一层 try 兜底,双保险不惊主链。
|
||
try:
|
||
import cheap_alert_sink # noqa: PLC0415 惰性 import:顶层不牵告警依赖
|
||
_decision = final.get("decision") or {}
|
||
if _decision.get("outcome") == "tester_error":
|
||
_failure = _decision.get("failure") or {}
|
||
_subtype = str(_failure.get("subtype") or "unknown")
|
||
# 预算类事件(单局/全链撞硬帽)给 critical;超时及其余 tester_error 给 warning。
|
||
_alert_level = ("critical" if _subtype in ("cost_cap_exceeded", "parent_chain_cost_cap_exceeded")
|
||
else "warning")
|
||
cheap_alert_sink.notify_gen_alert(
|
||
_alert_level, f"acceptance_v3.{_subtype}",
|
||
detail={"gameId": game_id, "runId": run_dir.name,
|
||
"reason": str(_failure.get("reason") or ""),
|
||
"costRmb": str(_decision.get("costRmb") or "")},
|
||
link=str(run_dir))
|
||
except Exception: # noqa: BLE001 —— 告警任何异常都不允许影响验收主链
|
||
pass
|
||
return payload
|
||
|
||
|
||
async def run_acceptance_v3(request: dict) -> dict:
|
||
"""v3 唯一公开入口：同 run 加锁后重查封存，所有等待方只返回同一 decision。"""
|
||
request = dict(request or {})
|
||
game_id = str(request.get("gameId") or "")
|
||
if not _is_safe_game_id_v3(game_id):
|
||
# 保持原入口的参数错误语义，不为非法路径创建 claim 文件。
|
||
return await _run_acceptance_v3_locked(request)
|
||
identity = request.get("acceptanceIdentity") if isinstance(request.get("acceptanceIdentity"), dict) else None
|
||
if identity is None:
|
||
return await _run_acceptance_v3_locked(request)
|
||
|
||
artifact_hash = str(request.get("artifactHash") or _artifact_hash_v3(game_id, request.get("artifactPath")))
|
||
request["artifactHash"] = artifact_hash
|
||
brief_hash = str(identity.get("briefHash") or _sha256_text(str(request.get("brief") or "")))
|
||
run_id = _v3_run_id(request, artifact_hash, brief_hash)
|
||
evidence_root = Path(request.get("evidenceRoot") or (_V3_RESULTS_DIR / game_id))
|
||
run_dir = evidence_root / run_id
|
||
timeout_s = _acceptance_v3_cfg()["run_lock_timeout_s"]
|
||
try:
|
||
claim, lock_error = await asyncio.to_thread(
|
||
_acquire_v3_run_lock, evidence_root, run_dir, timeout_s)
|
||
except Exception as exc: # noqa: BLE001 —— 路径越界/锁文件异常必须形成 tester_error，不能绕锁运行
|
||
claim = None
|
||
lock_error = f"v3 run claim 失败:{type(exc).__name__}: {exc}"
|
||
if claim is None:
|
||
request["_forcedV3TesterError"] = lock_error or "v3 run claim 未取得"
|
||
return await _run_acceptance_v3_locked(request)
|
||
try:
|
||
return await _run_acceptance_v3_locked(request)
|
||
finally:
|
||
_release_v3_run_lock(claim)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 便捷自跑：python cheap-worker/cheap_verify.py <gameId> ["brief"] —— 对已有产物真跑一次评分。
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
gid = sys.argv[1] if len(sys.argv) > 1 else "cheap-smoke1"
|
||
bf = sys.argv[2] if len(sys.argv) > 2 else ""
|
||
out = asyncio.run(verify_richness(gid, brief=bf))
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|