ffe3a5c993
M2 auto-vs-golden 门抓到的生成契约漂移修复(Codex 实现 + 主代理诊断/受控验证)。 根因:~20% click-score 款 _forensicsView 暴露单数 target(对象、菜单态为 null) 而非复数 targets(数组),tap-targets harness 只认复数数组:① _build_play_spec 推断漏判→误推 key-cycle;② 驱动器 targetsPath 找不到单数 target。 两处修(认两种合法形态、既有数组行为零变): - cheap_run.py:_build_play_spec 以【键存在性】判 tap-targets(targets/target 键 在即可,菜单态 null/空数组仍归 tap-targets)——非值类型(空数组过/null 不过的 非对称是漏判根因)。 - play.cdp.cjs:runTapTargets 取值非数组时回退读单数 target、包成 1 元素数组。 验证:退化款 avg-click-score-14 auto 路 E_live/G_input/H_progress 0.667→1.0 (3/3 确定性复现);test_ensure_play_spec 8→13 绿。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
288 lines
15 KiB
Python
288 lines
15 KiB
Python
"""
|
||
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}"}
|
||
|
||
|
||
# 品类 → 通用 L2 插件前缀期望(held 期望,外生常量;对齐 fixtures/golden-specs/*.play-spec.json 三份金标的 expectedEngineCallPrefixes)。
|
||
# 防自证铁律(gen-path-parity skill):此表是【预置常量】,F_wiring 期望基准绝不从该局实际 __engineCalls 反推——
|
||
# 一旦反推,期望恒等于实际、F_wiring 退化自证、必过、假绿。tap-targets occupied 族三品类(点击得分/打地鼠/经营点客)共用此前缀集。
|
||
_TAP_TARGETS_CALL_PREFIXES = ("sessionScore.", "audioMusic.", "juice.")
|
||
|
||
|
||
def _build_play_spec(state: Optional[dict]) -> dict:
|
||
"""据 _forensicsView().state() 形态产 play-spec dict(纯函数、不写盘,便于单测)。
|
||
|
||
tap-targets occupied 族(state 有 targets 数组):加厚到金标同质——补 F_wiring 的 expectedEngineCallPrefixes
|
||
语义期望(外生 held 期望、预置常量,防自证);key-cycle 族(按键类):薄版,期望前缀待输入键契约(WU-C 5.4)。
|
||
"""
|
||
# 以键存在性判定「可点目标」类游戏:游戏暴露 targets/target 键即声明有可点目标,菜单态 null/空数组仍视为 tap-targets
|
||
has_targets = isinstance(state, dict) and ("targets" in state or "target" in state)
|
||
if has_targets:
|
||
driver = {"type": "tap-targets", "targetMode": "occupied", "targetsPath": "targets", "steps": 50, "stepMs": 220}
|
||
return {
|
||
"exportState": ["phase", "score", "targets"],
|
||
"driver": driver,
|
||
# U2 加厚到金标同质:F_wiring 语义期望前缀(外生 held 期望,绝不从实际调用反推 → 防自证假绿)。
|
||
"expectedEngineCallPrefixes": list(_TAP_TARGETS_CALL_PREFIXES),
|
||
"assertAfterPlay": [{"path": "score", "op": "increased", "why": "真玩应加分"}],
|
||
"expectLatch": True, # latch advisory 兜底:限时类玩不到终局但有真进展→降 advisory 不致命
|
||
}
|
||
driver = {"type": "key-cycle", "keys": ["ArrowLeft", "ArrowRight", "ArrowUp", "ArrowDown"], "steps": 60, "stepMs": 180, "downMs": 60}
|
||
return {
|
||
"exportState": ["phase", "score"],
|
||
"driver": driver,
|
||
"assertAfterPlay": [{"path": "score", "op": "increased", "why": "真玩应加分"}],
|
||
"expectLatch": True,
|
||
# key-cycle 薄版:expectedEngineCallPrefixes 待『两路绑同一套输入键』契约(WU-C 5.4),M1 不加厚成可能驱不动的厚 spec。
|
||
"_note": "key-cycle 自动 spec 薄版:F_wiring 期望前缀待输入键契约(WU-C 5.4)、M1 不加厚",
|
||
}
|
||
|
||
|
||
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": "已存在不覆盖"}
|
||
spec = _build_play_spec(state)
|
||
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": spec["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}"}
|