games-development-ai/wg1/gen-worker/gamedef_quickcheck.py

"""gamedef_quickcheck.py —— plan 2026-06-18-001 U3 gamedef 路 lili-mac 快走查（cutover 门早读）。

run.py 的 gamedef 版:把 generate(iife) → validate(静态) 换成
  generate(GAMEDEF_SYSTEM) → build-from-source.mjs(校验 schema/引用/静态扫描 + 装配出工厂)，
其余(scaffold/build/play 九门)复用 run.py（harness 零改）。

GAMEDEF_SYSTEM **从后端 SaaPrompts.java 抽取**（drift-free：跑的就是后端真用的 prompt）。
模型/网关/代理旁路复用 _client；play-spec/brief 复用现有 briefs/*.json（gatespec driver 同口径）。

用法：.venv/bin/python gamedef_quickcheck.py runner flappy whack pong   # 默认这 4
仅 lili-mac 快走查（小样早读）；authoritative ≥60% cutover 门在 mini-desktop（SaaFullGraphE2eTest）。
密钥经 _client 读 .env，绝不打印。
"""
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from worker import _client, run  # noqa: E402  复用 chat + scaffold/build/play

REPO_ROOT = Path(__file__).resolve().parents[2]
SAA_PROMPTS = (REPO_ROOT / "game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/"
               "com/wanxiang/huijing/game/module/aigc/saa/SaaPrompts.java")
GAME_RUNTIME = REPO_ROOT / "game-runtime"
GEN_DIR = GAME_RUNTIME / "games" / "_wg1-gen"
BRIEFS = Path(__file__).resolve().parent / "briefs"
import os
# 救场阶梯（mirror 真图 SaaStudioGraph）：stage1 连续 STAGE1_REPAIRS 次失败 → 升档 stage2 再 STAGE2_EXTRA 次。
STAGE1_MODEL = os.environ.get("QC_STAGE1", "deepseek-v4-flash")
STAGE2_MODEL = os.environ.get("QC_STAGE2", "deepseek-v4-pro")
STAGE1_REPAIRS = int(os.environ.get("QC_STAGE1_REPAIRS", "5"))   # 真图 maxRepairs 默认 5
STAGE2_EXTRA = int(os.environ.get("QC_STAGE2_EXTRA", "3"))       # 真图 stage2ExtraRepairs 默认 3


def extract_block(name):
    """从 SaaPrompts.java 抽指定 SYSTEM 文本块（去 Java 文本块 12 空格缩进）。drift-free=跑后端真 prompt。"""
    java = SAA_PROMPTS.read_text(encoding="utf-8")
    m = re.search(name + r' = """\n(.*?)\n\s*""";', java, re.DOTALL)
    if not m:
        raise RuntimeError("未能从 SaaPrompts.java 抽取 " + name)
    lines = [(ln[12:] if ln.startswith(" " * 12) else ln) for ln in m.group(1).split("\n")]
    return "\n".join(lines)


def extract_gatespec(design_text):
    """复刻 SaaPrompts.extractGatespec：抠 ```gatespec JSON 块 → 补 expectLatch=true + ballPath .y→.x。无则 None。"""
    m = re.search(r"```gatespec\s*(.+?)```", design_text or "", re.DOTALL)
    if not m:
        return None
    gs = extract_json_obj(m.group(1))
    if not isinstance(gs, dict):
        return None
    if "assertAfterPlay" in gs and not isinstance(gs["assertAfterPlay"], list):
        gs["assertAfterPlay"] = []
    gs.setdefault("expectLatch", True)
    # gamedef 取证只暴露 score/remaining/progress/实体位置——把 assertAfterPlay 的顶层自定义标量字段(moves/totalRound 等)强制到 score
    # (设计 agent 常造取证读不到的字段→H_progress 必挂;不动含 '.' 的实体位置断言如 ball.x)。
    EXPOSED = ("score", "remaining", "progress")
    for a in (gs.get("assertAfterPlay") or []):
        if isinstance(a, dict) and isinstance(a.get("path"), str):
            p = a["path"].lstrip("/")
            top = p.split(".")[0].split("[")[0]
            if "." not in p and top not in EXPOSED:
                a["path"] = "score"
                a["op"] = a.get("op", "increased")
    drv = gs.get("driver")
    if isinstance(drv, dict) and drv.get("type") == "paddle-intercept":
        bp = drv.get("ballPath")
        if isinstance(bp, str) and bp.endswith(".y"):
            drv["ballPath"] = bp[:-2] + ".x"
    return gs


def extract_json_obj(content):
    """从模型输出抠最外层 {...}（去 ```json 围栏/前后说明）。"""
    s = content.strip()
    s = re.sub(r"^```(?:json)?\s*", "", s)
    s = re.sub(r"\s*```$", "", s)
    lo, hi = s.find("{"), s.rfind("}")
    if lo < 0 or hi <= lo:
        return None
    try:
        return json.loads(s[lo:hi + 1])
    except Exception:
        return None


def assemble_via_cli(game_id, gamedef_obj):
    """写 source.json → 跑 build-from-source.mjs 校验+装配 → generated-factory.js。返回 (ok, err)。"""
    gdir = GEN_DIR / game_id
    gdir.mkdir(parents=True, exist_ok=True)
    src = gdir / "source.json"
    src.write_text(json.dumps(gamedef_obj, ensure_ascii=False), encoding="utf-8")
    r = subprocess.run(
        ["node", "src/host/build-from-source.mjs", str(src), str(gdir / "generated-factory.js")],
        cwd=str(GAME_RUNTIME), capture_output=True, text=True, timeout=60,
    )
    return r.returncode == 0, (r.stdout + r.stderr).strip()


def run_one(game_id, gamedef_system, design_system):
    brief_text, play_spec = run._load_brief(game_id)
    gid = "gd-" + game_id
    out = {"game_id": gid, "stage": None, "pass": False, "guards": {}, "err": "", "driver": None, "model": STAGE1_MODEL}
    # ① design 步（mirror 真图 design 节点）：产 gatespec(driver/controlCheck/assert) 覆写 play_spec + enriched。
    enriched = brief_text
    try:
        dresp = _client.chat(STAGE1_MODEL, design_system, brief_text, max_tokens=16000)
        enriched = brief_text + "\n\n## 设计稿\n" + dresp["content"]
        gs = extract_gatespec(dresp["content"])
        if gs:
            for k in ("exportState", "driver", "controlCheck", "assertAfterPlay", "expectLatch"):
                if k in gs and gs[k] is not None:
                    play_spec[k] = gs[k]
            out["driver"] = (gs.get("driver") or {}).get("type")
    except Exception as e:
        out["err"] = "design 降级:" + str(e)[:80]   # design 失败不致命，凭 brief 继续 generate
    # ② generate→assemble→build→play **repair 回环**（mirror 真图：失败→verdict/错误回喂→重生成,≤max_retries 次）。
    #    这是 ≥60% cutover 门的真机制（单发首攻 ≠ 终率）；design 只跑一次,repair 只重 generate。
    gdir = GEN_DIR / gid
    feedback = None
    total = STAGE1_REPAIRS + STAGE2_EXTRA
    for attempt in range(total):
        gen_model = STAGE2_MODEL if attempt >= STAGE1_REPAIRS else STAGE1_MODEL   # 救场升档:stage1 耗尽→stage2 强档
        out["attempts"] = attempt + 1
        out["model"] = gen_model
        user = "请为下面这款游戏产出一份 gameDefinition（只输出一个 JSON 对象，不要 ```代码块、不要解释）：\n\n" + enriched
        if feedback:
            user += "\n\n———\n上一次生成【未通过】，原因如下，请针对性修正后重新产出完整 gameDefinition JSON：\n" + feedback
        try:
            resp = _client.chat(gen_model, gamedef_system, user, max_tokens=16000)
        except Exception as e:
            out["stage"] = "generate"; feedback = "模型调用失败：" + str(e)[:160]; continue
        gd = extract_json_obj(resp["content"])
        if gd is None:
            out["stage"] = "parse"; feedback = "未能解析出 gameDefinition JSON。请只输出一个 JSON 对象（顶层 entities/components/behaviors/scenes/rules）。"; continue
        ok, err = assemble_via_cli(gid, gd)
        if not ok:
            out["stage"] = "validate/assemble"; feedback = "校验/装配未过：\n" + err[:500]; continue
        shutil.copyfile(GEN_DIR / "_shared" / "entry-bundle.template.js", gdir / "entry-bundle.js")
        shutil.copyfile(GEN_DIR / "_shared" / "index.template.html", gdir / "index.html")
        (gdir / "play-spec.json").write_text(json.dumps(play_spec, ensure_ascii=False, indent=2), encoding="utf-8")
        bok, blog = run.build(gid)
        if not bok:
            out["stage"] = "build"; feedback = "打包失败（esbuild）：\n" + blog[:500]; continue
        rc, _, verdict = run.play(gid)
        out["stage"] = "play"
        out["guards"] = {k: g.get("pass") for k, g in ((verdict or {}).get("guards") or {}).items()}
        if rc == 0 and verdict and verdict.get("pass"):
            out["pass"] = True
            break
        feedback = run._verdict_feedback(verdict)
        out["err"] = feedback[:300]
    return out


def main():
    games = sys.argv[1:] or ["runner", "flappy", "whack", "pong"]
    gds = extract_block("GAMEDEF_SYSTEM")
    dsys = extract_block("DESIGN_SYSTEM")
    print(f"[gamedef-qc] stage1={STAGE1_MODEL}({STAGE1_REPAIRS})→stage2={STAGE2_MODEL}({STAGE2_EXTRA})  games={games}  (全图镜像+救场升档)\n")
    results = []
    for g in games:
        r = run_one(g, gds, dsys)
        results.append(r)
        mark = "✅ PASS" if r["pass"] else f"❌ {r['stage']}"
        gpass = sum(1 for v in r["guards"].values() if v)
        gtot = len(r["guards"])
        drv = r.get("driver") or "-"
        att = r.get("attempts", 1)
        print(f"  {r['game_id']:14s} {mark:16s} drv={drv:14s} att={att} gates={gpass}/{gtot}  {('' if r['pass'] else r['err'][:100])}")
    npass = sum(1 for r in results if r["pass"])
    print(f"\n[gamedef-qc] 过门 {npass}/{len(results)}  (cutover 门基线=iife 60%; 本为 lili-mac 小样早读, authoritative=mini-desktop)")
    Path(__file__).resolve().parent.joinpath("results", "gamedef-quickcheck.json").write_text(
        json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
    sys.exit(0)


if __name__ == "__main__":
    main()