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
# 跨局并发数：仅跨不同游戏并行(局内 repair 链恒串行——第 N+1 次需第 N 次失败回喂);
# 每局独立 (port,cdpPort) 端口对(serve/Chrome/user-data-dir 全隔离,见 serve-and-play.sh)→互不撞;
# 每局过门判定不变→聚合过门率与串行等价。authoritative ≥60% cutover 仍在 mini-desktop。
CONCURRENCY = int(os.environ.get("QC_CONCURRENCY", "4"))


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)
    # assertAfterPlay 处理两步：① 有进展断言则只留进展断言(挡终局态断言)；② 取证读不到的自定义标量归一到 score。
    PROGRESS_OPS = ("increased", "decreased", "changed", ">", ">=", "<", "<=")
    KNOWN = ("score", "remaining", "progress", "result", "phase")  # 取证暴露的合法顶层字段:不强转 score(result/phase 是真字段)

    # ① 终局态断言(phase/result=="gameover"/"win"、result in[...]、误编码 score=="gameover" 等 op 非进展类)不应否决
    #    「进展门」H_progress:只要≥1 条进展断言(op∈PROGRESS_OPS)存在,就只留进展断言、丢弃所有非进展(==/in/!=…)断言——
    #    bot 限时真玩常到不了终局,但 score 已涨=确有进展(井字棋 0→9 实证)。无任何进展断言(如规避类只给 result=="win")
    #    则全留(终局即唯一进展信号)。镜像后端 SaaPrompts.isProgressOp 过滤(drift-free);终局完成度另由 expectLatch 把关。
    aap = [a for a in (gs.get("assertAfterPlay") or []) if isinstance(a, dict)]
    if any(a.get("op") in PROGRESS_OPS for a in aap):
        aap = [a for a in aap if a.get("op") in PROGRESS_OPS]
    gs["assertAfterPlay"] = aap
    # ② 顶层自定义标量字段(moves/totalRound 取证读不到)强制到 score;result/phase 等合法字段与含 '.' 的实体位置断言(ball.x)不动。
    for a in aap:
        if isinstance(a.get("path"), str):
            p = a["path"].lstrip("/")
            top = p.split(".")[0].split("[")[0]
            if "." not in p and top not in KNOWN:
                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):
    """从模型输出抠最外层 {...}（先剥 <think> 推理块、再去 ```json 围栏/前后说明）。"""
    # 推理模型(M3)即便关了 thinking 偶仍内联 <think>…</think>;先整块剥除,避免抽到思维链里的花括号。
    s = re.sub(r"<think>.*?</think>", "", content or "", flags=re.DOTALL).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
    body = s[lo:hi + 1]
    try:
        return json.loads(body)               # ① 严格优先
    except Exception:
        pass
    # ② 宽松回退：便宜模型偶产轻微非法 JSON(如多余引号 "h":130")，json_repair 修复——镜像后端 looseParse 宽松级,
    #    跑的就是产线真解析口径(drift-free);避免快走查因严格解析假性 parse 失败、低估真过门率。
    try:
        import json_repair
        obj = json_repair.loads(body)
        return obj if isinstance(obj, dict) else None
    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, port=4320, cdp_port=9222):
    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, port=port, cdp_port=cdp_port)
        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 _fmt_result(r):
    """单局结果行（含 game_id，并发完成序打印不串）。"""
    mark = "✅ PASS" if r["pass"] else f"❌ {r['stage']}"
    gpass = sum(1 for v in (r.get("guards") or {}).values() if v)
    gtot = len(r.get("guards") or {})
    drv = r.get("driver") or "-"
    att = r.get("attempts", 1)
    return f"  {r['game_id']:14s} {mark:16s} drv={str(drv):14s} att={att} gates={gpass}/{gtot}  {('' if r['pass'] else r.get('err','')[:100])}"


def main():
    import concurrent.futures
    games = sys.argv[1:] or ["runner", "flappy", "whack", "pong"]
    gds = extract_block("GAMEDEF_SYSTEM")
    dsys = extract_block("DESIGN_SYSTEM")
    conc = max(1, CONCURRENCY)
    # flush=True：stdout 重定向到文件时 Python 默认块缓冲,加 flush 让后台日志逐局可读(便于进度观察)。
    print(f"[gamedef-qc] stage1={STAGE1_MODEL}({STAGE1_REPAIRS})→stage2={STAGE2_MODEL}({STAGE2_EXTRA})  games={games}  并发={conc}  (全图镜像+救场升档)\n", flush=True)
    # 跨局并发：每局分到独立端口对(port=4400+i / cdpPort=9300+i,与串行默认 4320/9222 错开,逐局唯一→任意调度都不撞)。
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=conc) as ex:
        futs = {ex.submit(run_one, g, gds, dsys, 4400 + i, 9300 + i): g for i, g in enumerate(games)}
        for fut in concurrent.futures.as_completed(futs):
            try:
                r = fut.result()
            except Exception as e:   # harness 级异常兜底,不让一局炸掉整轮
                r = {"game_id": "gd-" + futs[fut], "stage": "harness", "pass": False, "guards": {}, "err": "harness 异常:" + str(e)[:160]}
            results.append(r)
            print(_fmt_result(r), flush=True)
    # as_completed 是完成序,按输入顺序重排再落盘/汇总。
    order = {("gd-" + g): i for i, g in enumerate(games)}
    results.sort(key=lambda r: order.get(r.get("game_id"), 999))
    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)", flush=True)
    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()