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

"""
cheap_run.py — 便宜档生成的工具底座实现（对照源：Node amodel-gen 的 tools.mjs + play.mjs）。

分工（KTD3）：
  · read_file / list_dir / write_file —— Python 直接实现（对齐 tools.mjs 的 readFileTool/listDirTool/writeFileTool：
    repo 根只读 + 200KB 截断 / L1_FIXED 拦截 + 写边界），逻辑简单、无引擎重活。
  · check / build / scaffold / stage / play —— shell-out 现有 node/cjs，不 Python 重写 esbuild/CDP/形状门：
    check=`node tools.mjs check <id>`（含 node --check + 全部静态门 + 两条形状门）、build=`node tools.mjs build-saa <id>`、
    scaffold/stage 同 tools.mjs CLI、九门 play=`bash serve-and-play.sh <id>`（U4 落地）。
    Node 已验过不误伤的形状门，便宜档 shell-out 同一份代码自动继承，零口径漂移。

本文件 U2 先落 read/list/write/check/build；scaffold/stage/play/ensure_play_spec 由 U4 补全。
"""

import json
import os
import subprocess
from pathlib import Path
from typing import Optional

# 路径基准：cheap-worker/cheap_run.py → parents[1]=repo 根。
_REPO_ROOT = Path(__file__).resolve().parents[1]
_GAME_RUNTIME = _REPO_ROOT / "game-runtime"
_GAMES_DIR = _GAME_RUNTIME / "games"
_AMODEL_GEN = _GAME_RUNTIME / "tools" / "amodel-gen"
_TEMPLATE_DIR = _GAMES_DIR / "_template"
_WG1_DIR = _GAMES_DIR / "_wg1-gen"
_SERVE_AND_PLAY = _WG1_DIR / "_shared" / "serve-and-play.sh"  # 九门一键（cwd=game-runtime）
# 本机 Mac Chrome：play.mjs smokeBoot 默认 /usr/bin/google-chrome 在 Mac 不存在，须显式 Mac 路径。
_DEFAULT_CHROME = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"

_MAX_READ_BYTES = 200 * 1024  # 单文件读上限，对 tools.mjs MAX_READ_BYTES，防撑爆 agent context
# L1 固定 plumbing（boot 链/插件装配/viewport/工厂 wiring），agent 不许写，对 tools.mjs:89 L1_FIXED。
_L1_FIXED = {"host-config.js", "game.js", "index.html", "entry-bundle.js", "entry.js", "main.js"}


def game_dir(game_id: str) -> Path:
    """本 run 的 game 目录（= 写边界），对 tools.mjs gameDir：games/amgen-<id>/。"""
    return _GAMES_DIR / f"amgen-{game_id}"


def wg1_game_dir(game_id: str) -> Path:
    """SAA stage 目标目录，对 tools.mjs saaGameDir：games/_wg1-gen/<id>/。"""
    return _WG1_DIR / game_id


def _resolve_in_repo(rel: str) -> Path:
    """把 repo 相对路径解析为绝对路径，校验不逃出 repo 根（对 tools.mjs resolveInRepo）。"""
    abs_p = (_REPO_ROOT / rel).resolve()
    if abs_p != _REPO_ROOT and _REPO_ROOT not in abs_p.parents:
        raise ValueError(f"路径越界（必须在 repo 根内）：{rel}")
    return abs_p


# ───────────────────────── read / list / write（Python 直接实现）─────────────────────────

def read_file(path: str) -> dict:
    """read_file：repo 根只读。返回 {ok, content?, truncated?, error?}（对 tools.mjs readFileTool）。"""
    try:
        abs_p = _resolve_in_repo(path)
        if abs_p.is_dir():
            return {"ok": False, "error": f"是目录，不是文件：{path}（请用 list_dir）"}
        raw = abs_p.read_bytes()
        truncated = False
        if len(raw) > _MAX_READ_BYTES:
            raw = raw[:_MAX_READ_BYTES]  # 截断：超大文件只给前 200KB
            truncated = True
        return {"ok": True, "content": raw.decode("utf-8", "ignore"), "truncated": truncated}
    except Exception as e:  # noqa: BLE001  工具层统一软失败返回给 agent，不抛断循环
        return {"ok": False, "error": str(e)}


def list_dir(path: str) -> dict:
    """list_dir：repo 根内列目录（目录名带尾 /）。返回 {ok, entries?, error?}（对 tools.mjs listDirTool）。"""
    try:
        abs_p = _resolve_in_repo(path)
        entries = sorted((e.name + "/" if e.is_dir() else e.name) for e in abs_p.iterdir())
        return {"ok": True, "entries": entries}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "error": str(e)}


def write_file(game_id: str, path: str, content: str) -> dict:
    """write_file：仅许写本 run game 目录的 L3 文件。返回 {ok, bytes?, error?}（对 tools.mjs writeFileTool）。

    边界：abs 必须落在 games/amgen-<id>/ 内；basename 命中 L1_FIXED 则拒写（根除把 wiring 写坏类错误）。
    """
    try:
        abs_p = _resolve_in_repo(path)
        gd = game_dir(game_id).resolve()
        if abs_p != gd and gd not in abs_p.parents:
            return {"ok": False, "error": f"write 越界：只许写 games/amgen-{game_id}/ 内（你给的是 {path}）"}
        base = abs_p.name
        if base in _L1_FIXED:
            return {"ok": False, "error": (
                f"{base} 是 L1 固定 plumbing（boot 链/插件装配/viewport/工厂 wiring），不要写它。"
                "你只写 L3 游戏本体：src/{game-logic,core,render,balance,assets}.js。"
                "游戏逻辑写在 game-logic.js 的 createGame({plugins,bundle,viewport})；插件经 plugins.<键> 直接用（已替你注入）。")}
        abs_p.parent.mkdir(parents=True, exist_ok=True)
        abs_p.write_text(content, encoding="utf-8")
        return {"ok": True, "bytes": len(content.encode("utf-8"))}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "error": str(e)}


# ───────────────────────── check / build（shell-out node tools.mjs）─────────────────────────

def _run_node_tools(cmd: str, game_id: str, timeout: float = 90.0) -> subprocess.CompletedProcess:
    """shell-out `node tools.mjs <cmd> <id>`（cwd=game-runtime，对 tools.mjs 的路径基准）。"""
    return subprocess.run(
        ["node", str(_AMODEL_GEN / "tools.mjs"), cmd, game_id],
        cwd=str(_GAME_RUNTIME), capture_output=True, text=True, timeout=timeout,
        check=False,
    )


def check(game_id: str) -> dict:
    """check：循环内快反馈门，shell-out `node tools.mjs check <id>`（node --check + 全部静态门 + 两条形状门）。

    退出码 0=PASS；非 0=FAIL，stdout 含逐条错误清单（直接喂 agent 自纠）。返回 {ok, output}。
    """
    try:
        r = _run_node_tools("check", game_id)
        return {"ok": r.returncode == 0, "output": (r.stdout + r.stderr).strip()}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "output": f"check shell-out 异常：{e}"}


def build(game_id: str) -> dict:
    """build：循环内 esbuild 打包，shell-out `node tools.mjs build-saa <id>`（SAA 信封 __GameBundle）。

    返回 {ok, output}。
    """
    try:
        r = _run_node_tools("build-saa", game_id)
        return {"ok": r.returncode == 0, "output": (r.stdout + r.stderr).strip()}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "output": f"build shell-out 异常：{e}"}


# ───────────────────────── U4：scaffold / stage / smoke / play / ensure_play_spec（shell-out node/cjs）─────────────────────────

def _shell_env() -> dict:
    """shell-out node/bash 用的 env：本机 Mac Chrome 路径 + NO_PROXY（serve 是 localhost，play 不连内网，但对齐本机跑法）。"""
    env = dict(os.environ)
    env.setdefault("CHROME_BIN", _DEFAULT_CHROME)
    no_proxy = env.get("NO_PROXY", "")
    for h in ("localhost", "127.0.0.1", "100.64.0.8"):
        if h not in no_proxy:
            no_proxy = (no_proxy + "," + h) if no_proxy else h
    env["NO_PROXY"] = no_proxy
    env["no_proxy"] = no_proxy
    return env


def scaffold(game_id: str) -> dict:
    """scaffold：shell-out `node tools.mjs scaffold-saa <id>`（clone _template + SAA 信封 __GameBundle）。返回 {ok, output}。"""
    try:
        r = _run_node_tools("scaffold-saa", game_id)
        return {"ok": r.returncode == 0, "output": (r.stdout + r.stderr).strip()}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "output": f"scaffold shell-out 异常：{e}"}


def stage(game_id: str) -> dict:
    """stage：shell-out `node tools.mjs stage <id>`（原子拷 bundle.iife.js + index.html → _wg1-gen/<id>/）。返回 {ok, output}。"""
    try:
        r = _run_node_tools("stage", game_id)
        return {"ok": r.returncode == 0, "output": (r.stdout + r.stderr).strip()}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "output": f"stage shell-out 异常：{e}"}


def smoke(game_id: str, port: int = 4320, cdp_port: int = 9222) -> dict:
    """smoke：shell-out `node play.mjs smoke <id>`（SAA 信封 boot + 帧推进 + 抓 _forensicsView state）。

    解析 play.mjs 末行 `[smoke] PASS/FAIL {json}` 拿 ok + state（state 供 ensure_play_spec 推断 driver 形态）。
    返回 {ok, state, raw}。
    """
    try:
        r = subprocess.run(
            ["node", str(_AMODEL_GEN / "play.mjs"), "smoke", game_id, str(port), str(cdp_port)],
            cwd=str(_GAME_RUNTIME), capture_output=True, text=True, timeout=120,
            env=_shell_env(), check=False,
        )
        out = r.stdout or ""
        ok = r.returncode == 0
        state = None
        idx = out.rfind("[smoke]")
        if idx >= 0:
            line = out[idx:].splitlines()[0]
            brace = line.find("{")
            if brace >= 0:
                try:
                    o = json.loads(line[brace:])
                    ok = bool(o.get("ok"))
                    state = o.get("state")
                except json.JSONDecodeError:
                    pass
        return {"ok": ok, "state": state, "raw": (out + (r.stderr or ""))[-2000:]}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "state": None, "raw": f"smoke shell-out 异常：{e}"}


def ensure_play_spec(game_id: str, state: Optional[dict]) -> dict:
    """据 _forensicsView().state() 形态自动产 _wg1-gen/<id>/play-spec.json（对 tools.mjs ensurePlaySpec）。

    driver 推断（照 play.cdp.cjs 的 driver 家族）：有 targets 数组（点击类）→ tap-targets occupied；
    否则（按键类）→ key-cycle。**已存在不覆盖**。让九门 driven=true、E_live/H_progress 由 advisory 升为致命，
    根治"纯 harness 裸跑无 driver → 假绿"（本周链路三修的便宜档 Python 侧对应）。

    Args:
        game_id: stage 后的 gameId（写 _wg1-gen/<id>/play-spec.json）。
        state: smoke 抓到的 state 快照（None/非 dict → 保守按键类）。
    """
    try:
        dst = wg1_game_dir(game_id)
        spec_path = dst / "play-spec.json"
        if spec_path.exists():
            return {"ok": True, "wrote": False, "reason": "已存在不覆盖"}
        has_targets = isinstance(state, dict) and isinstance(state.get("targets"), list)
        if has_targets:
            driver = {"type": "tap-targets", "targetMode": "occupied", "targetsPath": "targets", "steps": 50, "stepMs": 220}
            export_state = ["phase", "score", "targets"]
        else:
            driver = {"type": "key-cycle", "keys": ["ArrowLeft", "ArrowRight", "ArrowUp", "ArrowDown"], "steps": 60, "stepMs": 180, "downMs": 60}
            export_state = ["phase", "score"]
        spec = {
            "exportState": export_state,
            "driver": driver,
            "assertAfterPlay": [{"path": "score", "op": "increased", "why": "真玩应加分"}],
            "expectLatch": True,  # latch 现有 advisory 兜底：限时类玩不到终局但有真进展→降 advisory 不致命
        }
        dst.mkdir(parents=True, exist_ok=True)
        tmp = dst / "play-spec.json.tmp"
        tmp.write_text(json.dumps(spec, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp.replace(spec_path)  # 原子写，防 play 读半成品
        return {"ok": True, "wrote": True, "driverType": driver["type"]}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "wrote": False, "reason": str(e)}


def play(game_id: str, port: int = 4320, cdp_port: int = 9222) -> dict:
    """play：循环外九门真玩，shell-out `bash serve-and-play.sh <id>`（CDP、引擎无关九门）。

    读 _wg1-gen/<id>/evidence/verdict.json 拿九门逐门 verdict。返回 {ok, verdict, exitCode, raw}。
    """
    try:
        r = subprocess.run(
            ["bash", str(_SERVE_AND_PLAY), game_id, str(port), str(cdp_port)],
            cwd=str(_GAME_RUNTIME), capture_output=True, text=True, timeout=180,
            env=_shell_env(), check=False,
        )
        verdict = None
        vp = wg1_game_dir(game_id) / "evidence" / "verdict.json"
        if vp.exists():
            try:
                verdict = json.loads(vp.read_text(encoding="utf-8"))
            except json.JSONDecodeError:
                pass
        return {"ok": r.returncode == 0, "verdict": verdict, "exitCode": r.returncode,
                "raw": (r.stdout + r.stderr)[-2000:]}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "verdict": None, "exitCode": -1, "raw": f"play shell-out 异常：{e}"}