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

"""
compare_node.py — 便宜档 Python 路 vs Node 旧路对照验证。

【封存(W-AXIS-V2 波2,2026-07-10)】对照范式以金标 play-spec 驱动九门(H_progress 为可玩门);play-spec/
驱动器已随取证契约退出便宜档验收链,Node 旧路亦已退役——两路对照使命完结。保留代码供历史复跑;
生产链不再调用,play-spec 生成器窗口期结束后随其一并清理。

两层用法:
  · 单品类对照(spike U6 起点):compare_logic / build_report / main_async —— 同 brief 各跑一次,纯逻辑见下。
  · 三品类批量对照(WU-A 后续 U2):run_multi —— 三个 tap-targets 代表品类(点击得分/打地鼠/经营点客)各 n=3–5,
    用每品类金标 play-spec 喂两路 held constant、把驱动器从对照变量里摘掉,按两层判据 + 逐品类绝对地板聚合,
    产"框架等价证据(金标 spec 下)"。

对照公平的关键(KTD3 + Codex C1/C4 + Opus O1):
  · 金标 spec 强制覆写到 staged 路径 games/_wg1-gen/<id>/play-spec.json(play.cdp.cjs 真实读取处,非 evidence 目录),
    在「生成」与「九门 play」之间注入 —— 两路同用一份驱动器,过门率差才只反映生成质量。
  · Python 路用 run_studio(run_gates=False) 做 generation-only(内部不 play),Node 路 gen.mjs 本就不 play;
    两路都「生成 → 注入金标 → 单独 play」。
  · 成功定义 = 坐实「Python ≈ Node、范式未回退」(金标 spec 下),≠「过九门」、≠「生产退役许可」(后者另 gate
    在 WU-C 自动 spec 达金标同等过门率)。

CLI:cheap-worker/.venv/bin/python cheap-worker/compare_node.py [--genres click-score,whack-mole,shop-serve] [--n 3]
"""

import asyncio
import json
import statistics
import subprocess
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))  # → cheap-worker/
import _bootstrap  # noqa: E402,F401  跨包 sys.path 兜底
import cheap_run  # noqa: E402
import cheap_studio  # noqa: E402

_GOLDEN_DIR = Path(__file__).resolve().parent / "fixtures" / "golden-specs"
_RESULTS_DIR = Path(__file__).resolve().parent / "results"

# 三个 tap-targets 代表品类(plan KTD2):brief 取自 base1/base2/base4 run-summary,金标 spec 见 fixtures/golden-specs/。
GENRES = [
    {"key": "click-score", "base": "base1", "golden": "click-score.play-spec.json"},
    {"key": "whack-mole", "base": "base2", "golden": "whack-mole.play-spec.json"},
    {"key": "shop-serve", "base": "base4", "golden": "shop-serve.play-spec.json"},
]
_GENRE_BY_KEY = {g["key"]: g for g in GENRES}

# 判据阈值(R3 两层 + 逐品类绝对地板 + 双低标红)。
_EPS = 1e-9
_PLAY_GATE = "H_progress"   # 「真玩起来了」的可玩门 = 有真实游戏进展(driven 后 H 过)。
_DOUBLE_LOW_PLAY = 0.5      # 两路真玩比例都 < 此 → 双低、对照不充分、标红。
_FLOOR_PLAY = 0.6          # 坐实绝对地板:Python 须在 ≥60% 局真玩起来(带余量,不只 ≥Node)。
_MAX_N = 5                  # n>5 小批口径(plan R5),上限夹 5、不滑向 n≥30 统计批跑。


# ───────────────────────── 纯逻辑(可单测,不依赖真跑)─────────────────────────

def gates_from_verdict(verdict) -> dict:
    """从九门 verdict.json 抽 {gate: pass_bool}。"""
    if not isinstance(verdict, dict):
        return {}
    guards = verdict.get("guards") or {}
    return {k: bool(g.get("pass")) for k, g in guards.items() if isinstance(g, dict)}


def src_shape(game_dir: Path) -> dict:
    """产物形态:是否真 src/ 多文件工程(game-logic.js 在 + ≥2 个 .js)。"""
    src = game_dir / "src"
    js = list(src.glob("*.js")) if src.exists() else []
    return {"realSrcMultifile": (src / "game-logic.js").exists() and len(js) >= 2, "jsCount": len(js)}


def _pass_rate(gates: dict) -> float:
    return round(sum(1 for v in gates.values() if v) / len(gates), 3) if gates else 0.0


def compare_logic(py_gates: dict, node_gates: dict, py_real_src: bool) -> dict:
    """单品类对照判定(spike U6 起点):Python 产真 src/ 多文件 且 九门过门率不低于 Node → 坐实。"""
    py_rate = _pass_rate(py_gates)
    node_rate = _pass_rate(node_gates)
    spike_pass = bool(py_real_src and py_rate >= node_rate)
    return {
        "pyPassRate": py_rate,
        "nodePassRate": node_rate,
        "rateDelta": round(py_rate - node_rate, 3),
        "pyRealSrcMultifile": py_real_src,
        "spikePass": spike_pass,
        "verdict": "范式可移植坐实(Python 产真 src/ 多文件且九门过门率不低于 Node)" if spike_pass
                   else "未达标(Python 产物非真 src/ 多文件,或九门过门率低于 Node)",
    }


def _played(gates: dict) -> bool:
    """这一局是否真玩起来了 = H_progress 过(有真实游戏进展;两路同卡菜单时此门为 False)。"""
    return gates.get(_PLAY_GATE) is True


def _mean(xs) -> float:
    xs = list(xs)
    return round(statistics.fmean(xs), 3) if xs else 0.0


def judge_genre(py_runs: list, node_runs: list) -> dict:
    """逐品类两层判据 + 绝对地板 + 双低标红(R3）。

    py_runs/node_runs:每元素 = {"gates": {门: pass_bool}, "realSrc": bool}(一局)。

    status:
      · inconclusive(标红):两路真玩比例都 < 0.5 —— 两路都没真玩起来,对照不充分,不计坐实。
      · equivalent(坐实):Python 每局产真 src/ 且 过门率不低于 Node(相对层)且 Python 真玩比例 ≥0.6(绝对地板)。
      · regressed:Python 明显差(非真 src/ / 过门率低 / 真玩比例不达地板)。
    """
    py_rate = _mean(_pass_rate(r["gates"]) for r in py_runs)
    node_rate = _mean(_pass_rate(r["gates"]) for r in node_runs)
    py_play = _mean(1.0 if _played(r["gates"]) else 0.0 for r in py_runs)
    node_play = _mean(1.0 if _played(r["gates"]) else 0.0 for r in node_runs)
    py_real = bool(py_runs) and all(r.get("realSrc") for r in py_runs)

    relative_ok = py_rate >= node_rate - _EPS
    double_low = py_play < _DOUBLE_LOW_PLAY and node_play < _DOUBLE_LOW_PLAY
    floor_ok = py_play >= _FLOOR_PLAY

    if double_low:
        status, reason = "inconclusive", "两路真玩比例都低(同卡/同低分)→ 对照不充分、标红、不计坐实"
    elif py_real and relative_ok and floor_ok:
        status, reason = "equivalent", "Python 产真 src/ + 过门率不低于 Node + 真玩比例达绝对地板 → 坐实等价"
    else:
        bits = []
        if not py_real:
            bits.append("Python 非每局真 src/")
        if not relative_ok:
            bits.append(f"过门率低于 Node({py_rate}<{node_rate})")
        if not floor_ok:
            bits.append(f"Python 真玩比例 {py_play}<{_FLOOR_PLAY} 地板")
        status, reason = "regressed", "回退:" + "、".join(bits)

    return {
        "pyPassRate": py_rate, "nodePassRate": node_rate, "rateDelta": round(py_rate - node_rate, 3),
        "pyPlayRatio": py_play, "nodePlayRatio": node_play, "pyRealSrcAll": py_real,
        "doubleLowRed": double_low, "status": status, "reason": reason,
    }


def _read_verdict(game_id: str):
    vp = cheap_run.wg1_game_dir(game_id) / "evidence" / "verdict.json"
    if vp.exists():
        try:
            return json.loads(vp.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            return None
    return None


def _clamp_n(n: int) -> int:
    """n>5 小批口径:夹到 [1, 5],不滑向 n≥30 统计批跑(plan R5)。"""
    try:
        n = int(n)
    except (TypeError, ValueError):
        n = 3
    return max(1, min(_MAX_N, n))


# ───────────────────────── 金标 spec 注入(staged 路径、强制覆写)─────────────────────────

def golden_spec_path(genre: dict) -> Path:
    return _GOLDEN_DIR / genre["golden"]


def _write_spec_atomic(dst_path: Path, content: str) -> None:
    """原子写 play-spec.json(写 tmp 后 rename,防 play 读半成品)。"""
    dst_path.parent.mkdir(parents=True, exist_ok=True)
    tmp = dst_path.parent / (dst_path.name + ".tmp")
    tmp.write_text(content, encoding="utf-8")
    tmp.replace(dst_path)


def inject_golden(game_id: str, golden: Path) -> Path:
    """强制覆写金标 spec 到 staged games/_wg1-gen/<id>/play-spec.json(play.cdp.cjs 真实读取处)。

    与 ensure_play_spec 的「不覆盖」不同:这里**强制覆写**两路各自生成时产的自动 spec,把驱动器恒定到金标
    —— 这是对照公平的落点(Codex C1/C4)。返回写入的 staged spec 路径。
    """
    dst = cheap_run.wg1_game_dir(game_id) / "play-spec.json"
    _write_spec_atomic(dst, golden.read_text(encoding="utf-8"))
    return dst


# ───────────────────────── 真跑两路(生成 / play 拆步、金标注入、串行隔离)─────────────────────────

def load_brief(genre: dict) -> str:
    """品类 brief 取自 base 样本 run-summary(单一来源);缺失则空串(由调用方兜底)。"""
    rs = cheap_run.game_dir(genre["base"]) / "evidence" / "run-summary.json"
    if rs.exists():
        try:
            return json.loads(rs.read_text(encoding="utf-8")).get("brief") or ""
        except json.JSONDecodeError:
            pass
    return ""


def _gen_node(brief: str, game_id: str, port: int = 4320, cdp_port: int = 9222,
              timeout: float = 900.0) -> dict:
    """Node 旧路 generation-only:gen.mjs --mode saa(scaffold+stage+done 门 smoke,SAA 不跑九门)。不在此 play。

    传 --port/--cdp:gen.mjs done 门内部 smoke-boot 用这俩端口(行 142-143/249),并发时每槽独立端口防撞。
    """
    r = subprocess.run(
        ["node", str(cheap_run._AMODEL_GEN / "gen.mjs"), "--mode", "saa", "--game-id", game_id,
         "--brief", brief, "--port", str(port), "--cdp", str(cdp_port)],
        cwd=str(cheap_run._GAME_RUNTIME), capture_output=True, text=True, timeout=timeout,
        env=cheap_run._shell_env(), check=False,
    )
    return {"rc": r.returncode, "raw": (r.stdout + r.stderr)[-1200:]}


def run_pair_sync(genre: dict, brief: str, py_id: str, node_id: str,
                  port: int = 4320, cdp_port: int = 9222) -> dict:
    """一局对照(同步、线程内整对跑):两路各 generation-only → 注入同一份金标 spec → 各自单独 play。

    设计为同步,由 run_multi 经 asyncio.to_thread 并发调度(每对一个线程、独立端口);线程内 run_studio 用自带
    event loop(asyncio.run)。返回逐门 + 形态。
    """
    golden = golden_spec_path(genre)

    # Python 路:generation-only(run_gates=False,内部不 play)→ 注入金标 → play。
    asyncio.run(cheap_studio.run_studio(py_id, brief, run_gates=False, port=port, cdp_port=cdp_port))
    inject_golden(py_id, golden)
    py_play = cheap_run.play(py_id, port=port, cdp_port=cdp_port)

    # Node 路:gen.mjs --mode saa(不 play,内部 smoke 用同槽端口)→ 注入金标(覆写自动 spec)→ play。
    node_gen = _gen_node(brief, node_id, port=port, cdp_port=cdp_port)
    inject_golden(node_id, golden)
    node_play = cheap_run.play(node_id, port=port, cdp_port=cdp_port)

    py_gates = gates_from_verdict(py_play.get("verdict"))
    node_gates = gates_from_verdict(node_play.get("verdict"))
    return {
        "pyId": py_id, "nodeId": node_id, "nodeGenRc": node_gen["rc"],
        "py": {"gates": py_gates, "realSrc": src_shape(cheap_run.game_dir(py_id))["realSrcMultifile"]},
        "node": {"gates": node_gates, "realSrc": src_shape(cheap_run.game_dir(node_id))["realSrcMultifile"]},
    }


def aggregate(genre_runs: dict) -> dict:
    """按品类聚合:每品类的多局 runs → judge_genre 结论 + 顶层退役就绪范围。

    genre_runs: {genreKey: [run, ...]}，run = run_pair 返回。
    """
    per_genre = []
    for key, runs in genre_runs.items():
        py_runs = [r["py"] for r in runs]
        node_runs = [r["node"] for r in runs]
        judge = judge_genre(py_runs, node_runs)
        per_genre.append({"genre": key, "n": len(runs), "judge": judge,
                          "runs": [{"pyId": r["pyId"], "nodeId": r["nodeId"],
                                    "pyGates": r["py"]["gates"], "nodeGates": r["node"]["gates"]} for r in runs]})
    equivalent = [g["genre"] for g in per_genre if g["judge"]["status"] == "equivalent"]
    inconclusive = [g["genre"] for g in per_genre if g["judge"]["status"] == "inconclusive"]
    regressed = [g["genre"] for g in per_genre if g["judge"]["status"] == "regressed"]
    return {
        "perGenre": per_genre,
        "summary": {
            "equivalentGenres": equivalent,
            "inconclusiveGenres": inconclusive,
            "regressedGenres": regressed,
            # 「框架等价证据(金标 spec 下)」:坐实等价的品类 = Node 旧路在该品类可退役就绪(金标 spec 下);
            # 面向生产的退役授权另 gate 在 WU-C 自动 spec 达金标同等过门率(auto-vs-golden),不在本计划。
            "retireReadyUnderGolden": equivalent,
            "note": "本对照在金标 spec 下采证 = 框架等价;生产退役授权须另核对 WU-C 自动 spec 达金标同等(plan 对账表)",
        },
    }


def _pair_ids(genre_key: str, k: int) -> tuple:
    """一局两路的 gameId(前缀隔离:cheap-* vs node-*,品类×k 不撞)。"""
    return f"cheap-{genre_key}-{k}", f"node-{genre_key}-{k}"


def _port_pool(conc: int, base_port: int = 4320, base_cdp: int = 9222) -> list:
    """每并发槽一对独立 (server_port, cdp_port):两两不撞、server 段(43xx)与 cdp 段(92xx)不重叠。

    并发的真正约束 = 九门 play/smoke 的固定端口 + Chrome 实例;每槽独立端口才能并发不撞(gameId 前缀已隔离产物)。
    """
    conc = max(1, conc)
    return [(base_port + i * 2, base_cdp + i * 2) for i in range(conc)]


def _next_index(stems: list) -> int:
    """从已有 compare-multi-<N> 文件名算下一个序号(不覆盖历史)。"""
    nums = [int(s.rsplit("-", 1)[-1]) for s in stems if s.rsplit("-", 1)[-1].isdigit()]
    return (max(nums) + 1) if nums else 1


def _next_report_path() -> Path:
    """报告分批次文件、不覆盖历史(plan U2)。"""
    _RESULTS_DIR.mkdir(exist_ok=True)
    idx = _next_index([p.stem for p in _RESULTS_DIR.glob("compare-multi-*.json")])
    return _RESULTS_DIR / f"compare-multi-{idx}.json"


def print_multi_report(report: dict) -> None:
    print("\n=== 三品类对照报告:Python 新路 vs Node 旧路(金标 spec 下)===", file=sys.stderr)
    for g in report["perGenre"]:
        j = g["judge"]
        flag = " ⛔RED" if j["doubleLowRed"] else ""
        print(f"[{g['genre']}] n={g['n']} status={j['status']}{flag} "
              f"py_rate={j['pyPassRate']} node_rate={j['nodePassRate']} "
              f"py_play={j['pyPlayRatio']} node_play={j['nodePlayRatio']}  — {j['reason']}", file=sys.stderr)
    s = report["summary"]
    print(f"\n>>> 坐实等价(金标 spec 下,Node 可退役就绪): {s['equivalentGenres']}", file=sys.stderr)
    print(f">>> 对照不充分(标红): {s['inconclusiveGenres']}   回退: {s['regressedGenres']}", file=sys.stderr)
    print(f">>> 注:{s['note']}\n", file=sys.stderr)


async def run_multi(genre_keys: list, n: int, conc: int = 1,
                    base_port: int = 4320, base_cdp: int = 9222) -> dict:
    """三品类 × n 对照:逐局生成→注入金标→play,按品类聚合。

    conc=1 串行;conc>1 有界并发 —— 端口池(每槽独立 port/cdp)+ 信号量限并发(≤conc 同时跑),每对经
    asyncio.to_thread 在独立线程跑(线程内 run_studio 自带 loop)。前台进程内有界并发,非后台子代理。
    """
    n = _clamp_n(n)
    conc = max(1, conc)
    pool = asyncio.Queue()  # 端口池:size=conc,get 空则阻塞 → 天然限并发到 ≤conc
    for pp in _port_pool(conc, base_port, base_cdp):
        pool.put_nowait(pp)

    # 收集每品类的 runs(线程并发写不同 key 的 list,key 互斥、无共享写)。
    genre_runs = {}
    valid = []
    for key in genre_keys:
        genre = _GENRE_BY_KEY.get(key)
        if genre is None:
            print(f"[skip] 未知品类 {key}", file=sys.stderr)
            continue
        brief = load_brief(genre)
        if not brief:
            print(f"[skip] {key} 无 brief(base {genre['base']} run-summary 缺失)", file=sys.stderr)
            continue
        valid.append((key, genre, brief))
        genre_runs[key] = []

    async def one(key, genre, brief, k):
        py_id, node_id = _pair_ids(key, k)
        port, cdp = await pool.get()  # 取一对独立端口(无空闲则等)
        try:
            print(f"[{key} {k + 1}/{n}] py={py_id} node={node_id} port={port}/{cdp} 生成→注入金标→play …",
                  file=sys.stderr)
            r = await asyncio.to_thread(run_pair_sync, genre, brief, py_id, node_id, port, cdp)
            genre_runs[key].append(r)
        except Exception as e:  # noqa: BLE001  单局失败不拖垮整批,记录后继续
            print(f"[{key} {k + 1}/{n}] 异常:{type(e).__name__}: {e}", file=sys.stderr)
        finally:
            pool.put_nowait((port, cdp))  # 归还端口

    tasks = [one(key, genre, brief, k) for (key, genre, brief) in valid for k in range(n)]
    print(f"[run] {len(valid)} 品类 × n={n} = {len(tasks)} 对,conc={conc}(端口段 {base_port}.. / {base_cdp}..)",
          file=sys.stderr)
    await asyncio.gather(*tasks)
    genre_runs = {k: v for k, v in genre_runs.items() if v}  # 去掉全失败的品类

    report = aggregate(genre_runs)
    out = _next_report_path()
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"[report] → {out}", file=sys.stderr)
    print_multi_report(report)
    return report


# ───────────────────────── 单品类对照(spike U6 起点,保留)─────────────────────────

async def run_python(brief: str, game_id: str) -> dict:
    """Python 新路单局(spike：run_studio 内部已跑九门)。"""
    t0 = time.time()
    summary = await cheap_studio.run_studio(game_id, brief)
    return {"summary": summary, "wallSec": round(time.time() - t0, 1)}


def run_node(brief: str, game_id: str) -> dict:
    """Node 旧路单局(spike：gen.mjs --mode saa,再跑同款九门 play 用自动 spec)。"""
    t0 = time.time()
    gen = _gen_node(brief, game_id)
    pr = cheap_run.play(game_id)
    return {"nodeRC": gen["rc"], "wallSec": round(time.time() - t0, 1), "playVerdict": pr["verdict"], "raw": gen["raw"]}


def build_report(brief: str, py_id: str, node_id: str) -> dict:
    py_v = _read_verdict(py_id)
    node_v = _read_verdict(node_id)
    py_gates = gates_from_verdict(py_v)
    node_gates = gates_from_verdict(node_v)
    py_shape = src_shape(cheap_run.game_dir(py_id))
    node_shape = src_shape(cheap_run.game_dir(node_id))
    cmp = compare_logic(py_gates, node_gates, py_shape["realSrcMultifile"])
    return {
        "brief": brief, "pyId": py_id, "nodeId": node_id,
        "pyGates": py_gates, "nodeGates": node_gates,
        "pyShape": py_shape, "nodeShape": node_shape,
        "pyOverallPass": bool(py_v and py_v.get("pass")),
        "nodeOverallPass": bool(node_v and node_v.get("pass")),
        "compare": cmp,
    }


# ───────────────────────── CLI ─────────────────────────

if __name__ == "__main__":
    import argparse

    ap = argparse.ArgumentParser(description="便宜档 Python 路 vs Node 旧路三品类对照验证(金标 spec 下)")
    ap.add_argument("--genres", default="click-score,whack-mole,shop-serve",
                    help="逗号分隔品类键(默认三 tap-targets 代表品类)")
    ap.add_argument("--n", type=int, default=3, help="每品类重复次数(n>5 小批,上限夹 5)")
    ap.add_argument("--conc", type=int, default=1, help="并发对数(1=串行;>1 端口池+线程有界并发,本机 Chrome 建议 ≤4)")
    a = ap.parse_args()
    keys = [k.strip() for k in a.genres.split(",") if k.strip()]
    rep = asyncio.run(run_multi(keys, a.n, conc=a.conc))
    # 退出码:有任一品类坐实等价即 0(机制验证,非达标判定)。
    sys.exit(0 if rep["summary"]["equivalentGenres"] else 1)