games-development-ai/cheap-worker/cheap_verify.py

"""
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 json
from pathlib import Path

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 预填做同款"),
]

_MAX = len(RICHNESS_CHECKLIST)  # = 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/未登记 → 不评品类扩展）。
GENRE_BY_TEMPLATE = {
    "_template-trpg": "trpg",
    "_template-puzzle": "puzzle",
}


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:
    """各层条目数（分母）。通用底座 v2 = L2:6 / L3:3 / L4:2。"""
    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。
_JUDGE_SYSTEM = (
    "你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
    "只评判它「作为一款游戏够不够丰富、好玩、耐玩、想传出去」，逐条给出 11 条丰富度清单的命中与否 + 一句中文理由，"
    "最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。"
    "这 11 条分三层：L2 内容丰富（有料耐玩）、L3 留存结构（想再玩的结构前提）、L4 传播钩子（想让别人看/做同款的前提）；"
    "评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈、音效、结算画面与品类元数据，别被空壳或注释骗。"
)

# 品类扩展评分时的 system 变体（W-GENRE 件④）：仅把「11 条」的量词放宽到「通用 11 条 + 品类扩展若干条」，
# 其余判据措辞一字不动（锚定纪律：无品类路走上面原版 _JUDGE_SYSTEM，逐字节不变）。
_JUDGE_SYSTEM_GENRE = (
    "你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
    "只评判它「作为一款游戏够不够丰富、好玩、耐玩、想传出去」，逐条给出丰富度清单（通用 11 条 + 品类扩展若干条）的命中与否 + 一句中文理由，"
    "最后输出严格 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} 品类扩展清单】（分母独立、不与上面 11 条混合，同样逐条独立判命中）：\n"
            f"{genre_lines}\n"
        )
        total_note = f"共 {len(RICHNESS_CHECKLIST) + n_genre} 条（通用 11 条在前、品类 {n_genre} 条紧随其后）"
        name_note = "name 用上面各条的中文名"
    else:
        genre_block = ""
        total_note = "共 11 条"
        name_note = "name 用上面 11 条的中文名"
    return (
        f"{brief_block}"
        "下面是这款便宜档小游戏的 L3 源码。请逐条评判它是否命中这 11 条丰富度清单（每条前的 [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}")


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