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 兜底）。
本模块**不含任何丰富度的代码断言**——8 条好玩清单只是【喂给 LLM 的评分尺】，命中与否全由 judge 读码后裁。
（注：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 cheap_run  # 仅 stdlib 依赖（读产物 src 文件）；不 import agentscope/_bootstrap，故纯函数可在无 agentscope 环境单测。

# ── sim-business §10 反「无趣」8 条好玩清单（judge 的评分 rubric）──
#   这是【喂给 LLM judge 的打分尺】，不是代码校验：丰富不丰富全由 judge 读源码后逐条裁。
#   与 .agents/skills/sim-business-game-design.md §10 同源（那边是 design 自检，这里是产物 judge 评分）。
#   每项 = (标准名, 含义)；judge 按标准名逐条回命中与否 + 理由。
RICHNESS_CHECKLIST = [
    ("即时反馈", "每个主要操作有「飘字 +N + 音效 + 粒子」即时回报"),
    ("可见成长", "有数字/规模肉眼可见地变大（分数飙升/店铺扩张/等级上涨）"),
    ("下一个解锁", "任意时刻玩家眼前都有「再攒一点就解锁 X」的钩子"),
    ("30秒爽点", "开局 30 秒内有第一次升级/解锁"),
    ("数值滚雪球", "成长有「越来越快」的暴富段，而非平淡线性"),
    ("情感锚", "有萌角色/拥有物让玩家「想养大它」"),
    ("放置回归", "离线回来有惊喜（离线收益弹窗等）"),
    ("音反馈", "收益/升级/解锁有声音（哪怕程序化）"),
]

_MAX = len(RICHNESS_CHECKLIST)  # = 8

# 只读 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 = 3000  # judge 输出小（8 条短理由 + 一句点评），3000 足够、M3 无 thinking。

# judge 的 system / user 框定——只评分、不改码、要看源码真实现、输出严格 JSON。
_JUDGE_SYSTEM = (
    "你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，"
    "只评判它「作为一款游戏够不够丰富、好玩」，逐条给出 8 条好玩清单的命中与否 + 一句中文理由，"
    "最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。"
    "评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈与音效，别被空壳或注释骗。"
)


def _degraded(reason: str) -> dict:
    """降级结果（非阻塞铁律）：score=None + degraded=True；score 为 None 表示「这次没评出来」，绝不参与达标。"""
    return {"score": None, "max": _MAX, "hits": [], "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) -> dict:
    """纯函数：把 judge 的 LLM 文本输出解析成结构化丰富度评分（含 degraded 兜底）。可单测、零网络、零 agentscope。

    成功 → {score:int(命中数 0..8), max:8, hits:[{name,hit,why}*8], notes:str, degraded:False}
    解析失败（空/非 JSON 对象/缺 checks 数组）→ _degraded(...)（score=None, degraded=True）。

    对齐策略：优先按 name 对齐 judge 的 checks 到 8 条标准项；name 对不上的按位置兜底（容忍模型改名/英文名）。
    本函数**不含任何丰富度的代码判断**——只搬运 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

    hits = []
    score = 0
    for i, (name, _meaning) in enumerate(checklist):
        c = by_name.get(name)
        if c is None and i < len(checks_raw) and isinstance(checks_raw[i], dict):
            c = checks_raw[i]  # 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 ""
        hits.append({"name": name, "hit": hit, "why": why})
        if hit:
            score += 1
    notes = str(data.get("notes", "")).strip()[:500]
    return {"score": score, "max": len(checklist), "hits": hits, "notes": notes, "degraded": False}


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 = "") -> str:
    """拼 judge 的 user 消息：brief（可选）+ 8 条清单 + 产物源码 + 严格 JSON 输出契约（纯确定性字符串拼接）。"""
    checklist_lines = "\n".join(
        f"{i + 1}. {name}：{meaning}" for i, (name, meaning) in enumerate(RICHNESS_CHECKLIST)
    )
    brief_block = f"这款游戏的 brief（玩家想要的）：{brief}\n\n" if brief else ""
    return (
        f"{brief_block}"
        "下面是这款便宜档小游戏的 L3 源码。请逐条评判它是否命中这 8 条好玩清单，"
        "据**源码里真实实现了的玩法/数值/音效/解锁**判断（别被注释或空壳骗：比如只 import 了 audioMusic "
        "但收益处没真调 playSfx，则「音反馈」不算命中）：\n\n"
        f"{checklist_lines}\n\n"
        "=== 源码开始 ===\n"
        f"{src_text}\n"
        "=== 源码结束 ===\n\n"
        "严格只输出以下 JSON（不要任何额外文字、不要 markdown 围栏）：\n"
        '{"checks":[{"name":"即时反馈","hit":true,"why":"一句中文理由"}, … 共 8 条，'
        'name 用上面 8 条的中文名、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) -> 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 的源码上限。

    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)

        sys_msg = SystemMsg(name="system", content=_JUDGE_SYSTEM)
        user_msg = UserMsg(name="user", content=_build_judge_user(src_text, brief))
        # 硬超时包裹真模型调用：judge 卡住绝不能拖死收口（非阻塞铁律）。
        resp = await asyncio.wait_for(m([sys_msg, user_msg]), timeout=timeout)
        result = parse_judge_output(_extract_text(resp))

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