596ea13666
死圈根因(双评审坐实):产物带小红线违规→check FAIL→agent 想修但只有整文件覆盖的 write_file→改3行变重写22KB→MiniMax 漏 path 连撞→stuck 熔断→九门没跑→ok=False。 fix①-A:cheap_run.edit_file(读现文件→校验 old 唯一命中→精确替换→经 write_file 回写继承 L1_FIXED/game_dir 保护)+ cheap_toolkit edit_file 工具(whitelist 把关)注册进 七工具面;edit_file 加入软停 blocked-tools(middleware.py,越软停线同 write_file 拦)。 edit 自身失败面(old 未找到/不唯一)给结构化提示。6 单测(命中/未找到/不唯一/L1拒改/ 越界/缺文件)+ 现有 toolkit 8 测全绿。 fix②-prompt(cheap_roles.py):① 工具清单纠正为七工具、明确 edit_file 小改优先(修 之前"没有 edit_file"的错误引导——否则 agent 不会用新工具);② 克隆纪律"rng/nowMs 照 _template 逐字克隆、禁自造 Math.random/Date.now 防御回退、check 函数内裸调照拦"; ③ check-early"先写最小核心→立即 check→edit_file 修红线→再扩玩法"。 待续(本计划剩项):fix②-check 报错行号(tools.mjs)、fix③ brief 落 evidence、 fix①-B 共享 tier2 熔断窄口径纠偏(最险另做)、n=5 收敛环(ok=True 判据)。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
529 lines
30 KiB
Python
529 lines
30 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 hashlib
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import time
|
||
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 路径。
|
||
# 只在该路径真实存在(=Mac)才注入;Linux(mini-desktop)上注入它会把 smoke 全打成 spawn ENOENT →
|
||
# state=None → 考卷全体回退 key-cycle(2026-07-04 生产实证:80007 族错配盲修 ¥14.9、80012 gatespec
|
||
# driver=key-cycle 的共同上游根因)——Linux 交 play.mjs:22 自身默认 /usr/bin/google-chrome。
|
||
_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
|
||
|
||
|
||
# 便宜档 Service 会话注册表目录（C2：driver 按 session_id 写 session→后端 gameId 映射 + 本 session 配置，
|
||
# Service 两工厂读它把框架分配的 session_id 解析回后端 gameId）。_ 前缀 = 基建目录（同 _template/_wg1-gen），
|
||
# 非游戏目录，不会被 scaffold/list 当游戏，避免在 games/ 堆 amgen-<session uuid> 空壳。
|
||
_CHEAP_SESSIONS_DIR = _GAMES_DIR / "_cheap-sessions"
|
||
|
||
|
||
def session_cfg_path(session_id: str) -> Path:
|
||
"""本 session 的 cheap 配置 sidecar 路径（按 session_id 键；driver 在 /chat 前写、Service 工厂读）。
|
||
|
||
内容 {external_game_id, write_whitelist, scaffold_template, restricted}：因 AgentScope 工厂签名固定
|
||
(user_id, agent_id, session_id)、拿不到后端 gameId 也拿不到 session 记录，故用同机共享 FS sidecar 把
|
||
「session_id → 后端 gameId + 本 session 配置」跨进程传给工厂（C2 收口）。
|
||
"""
|
||
return _CHEAP_SESSIONS_DIR / f"{session_id}.json"
|
||
|
||
|
||
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
|
||
|
||
|
||
# ───────────────────────── 知识件外置 shadow（W-CFG-KB 切片一）─────────────────────────
|
||
# read_file 根解析：先查「知识根」（TIER2_KB_ROOT，externalized 知识件的物化落点，在 git 检出之外、
|
||
# deploy-safe），命中同相对路径的文件即读它；未命中回落「仓根」（_REPO_ROOT，api.d.ts 等代码类不外置、
|
||
# 仍读仓）。TIER2_KB_ROOT 未配置时整条 shadow 短路 —— read_file 行为与外置前逐字节一致（默认关，纯增量，
|
||
# 不碰生成主链）。写侧（kb_store.materialize_active 物化到此根）与本读侧读同一个 env TIER2_KB_ROOT。
|
||
|
||
def _kb_root() -> Optional[Path]:
|
||
"""知识根：env TIER2_KB_ROOT（未配 / 空 → None，read_file 退回只读仓根、行为字节不变）。"""
|
||
raw = os.environ.get("TIER2_KB_ROOT")
|
||
if not raw or not raw.strip():
|
||
return None
|
||
try:
|
||
return Path(raw).expanduser().resolve()
|
||
except Exception: # noqa: BLE001 —— 路径异常按未配处理（退回仓根，绝不因知识根拖垮读）
|
||
return None
|
||
|
||
|
||
def _resolve_in_root(rel: str, root: Path) -> Path:
|
||
"""把相对路径解析到给定根内，校验不逃出该根（知识根 shadow 用；与 _resolve_in_repo 同边界语义）。"""
|
||
abs_p = (root / rel).resolve()
|
||
if abs_p != root and root not in abs_p.parents:
|
||
raise ValueError(f"路径越界（必须在根内）：{rel}")
|
||
return abs_p
|
||
|
||
|
||
def _read_bounded(abs_p: Path) -> dict:
|
||
"""读一个已解析的文件：200KB 截断（对 tools.mjs MAX_READ_BYTES）。返回 {ok, content, truncated}。"""
|
||
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}
|
||
|
||
|
||
# ───────────────────────── read / list / write（Python 直接实现）─────────────────────────
|
||
|
||
def read_file(path: str) -> dict:
|
||
"""read_file：知识根 shadow 优先、回落仓根，两根均只读 + 200KB 截断 + 路径不逃逸（对 tools.mjs readFileTool）。
|
||
|
||
根解析（W-CFG-KB 切片一，纯增量）：TIER2_KB_ROOT 配置了就先查知识根同相对路径的文件，命中即读
|
||
（externalized 知识件热换即生效）；未命中（如插件 api.d.ts 等未外置的代码类）回落仓根。TIER2_KB_ROOT
|
||
未配时整条 shadow 短路，read_file 行为与外置前逐字节一致（默认关）。
|
||
"""
|
||
try:
|
||
kb_root = _kb_root()
|
||
if kb_root is not None:
|
||
# 知识根内解析（越界即拒；两根同相对路径，越界对仓根同样越界，不静默回落绕过边界）。
|
||
kb_abs = _resolve_in_root(path, kb_root)
|
||
if kb_abs.is_file():
|
||
return _read_bounded(kb_abs) # 知识根命中：读 externalized 知识件
|
||
# 知识根内无此文件（未外置的代码类）→ 回落仓根（下面）。
|
||
abs_p = _resolve_in_repo(path)
|
||
if abs_p.is_dir():
|
||
return {"ok": False, "error": f"是目录，不是文件：{path}（请用 list_dir）"}
|
||
return _read_bounded(abs_p)
|
||
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)}
|
||
|
||
|
||
def edit_file(game_id: str, path: str, old: str, new: str) -> dict:
|
||
"""edit_file：增量修改——把 game 目录内某 L3 文件里出现一次的 old 串替换成 new，避免"改三行→重写整文件"。
|
||
|
||
死圈修复 fix①-A（2026-07-08 双评审）：便宜档六工具原本只有整文件覆盖的 write_file，
|
||
"改一个 rng 小函数"被迫变成"重写 22KB"，MiniMax 在超长 tool-call 参数上反复漏 path 触发熔断。
|
||
edit_file 让小修表达为短参数替换，天然绕开漏参高发区。
|
||
|
||
边界继承：读现文件走 game_dir 内解析（不走 read_file 的知识根 shadow，要改的是工程盘上的真文件），
|
||
回写复用 write_file（继承 L1_FIXED 拒写 + games/amgen-<id>/ 越界拒写）；write_whitelist 由 toolkit 层把守。
|
||
old 必须在文件里唯一命中（0 次=没找到、>1 次=不唯一），否则拒改并给结构化提示（edit 自身失败面，纳入纠偏）。
|
||
"""
|
||
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"edit 越界：只许改 games/amgen-{game_id}/ 内（你给的是 {path}）"}
|
||
if not abs_p.is_file():
|
||
return {"ok": False, "error": f"文件不存在，无法 edit：{path}（先用 write_file 建它，再 edit 增量改）"}
|
||
content = abs_p.read_text(encoding="utf-8")
|
||
cnt = content.count(old)
|
||
if cnt == 0:
|
||
return {"ok": False, "error": (
|
||
"edit 未找到待替换串 old（逐字节精确匹配，含缩进/换行）。"
|
||
"请复制文件里的原文当 old（可 read_file 核对），或改用 write_file 整体覆盖。")}
|
||
if cnt > 1:
|
||
return {"ok": False, "error": (
|
||
f"edit 的 old 串在文件里出现 {cnt} 次、不唯一，无法定位。"
|
||
"请把 old 加长到含唯一上下文（多带几行），锚定到唯一位置。")}
|
||
new_content = content.replace(old, new, 1)
|
||
# 回写经 write_file：继承 L1_FIXED 拒写 + game_dir 越界拒写（edit 不新开绕过面）。
|
||
r = write_file(game_id, path, new_content)
|
||
if r.get("ok"):
|
||
r["message"] = "edit 已应用（替换 1 处）"
|
||
return r
|
||
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)
|
||
if os.path.exists(_DEFAULT_CHROME): # 仅 Mac 注入;Linux 不注入交 play.mjs 自身默认(见 _DEFAULT_CHROME 注)
|
||
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, template: Optional[str] = None) -> dict:
|
||
"""scaffold：shell-out `node tools.mjs scaffold-saa <id> [template]`（clone 模板 + SAA 信封 __GameBundle）。返回 {ok, output}。
|
||
|
||
template=None（默认）→ 通用 _template（点圆得分起点）；扩模板：传 per-genre 黄金骨架目录名
|
||
（经营=_template-shop 等），让难品类 AI 从预置经营循环起步、少写。模板不存在则 node 侧回落 _template。
|
||
"""
|
||
try:
|
||
argv = ["node", str(_AMODEL_GEN / "tools.mjs"), "scaffold-saa", game_id]
|
||
if template:
|
||
argv.append(template)
|
||
r = subprocess.run(argv, cwd=str(_GAME_RUNTIME), capture_output=True, text=True,
|
||
timeout=90.0, check=False)
|
||
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.")
|
||
|
||
|
||
# C5 契约(contracts/play-loop/play-spec.schema.json)接线常量:
|
||
# 本产线自动 spec 的 generator 标识——ensure_play_spec 据它区分「本产线自动 spec」与「手工金标/外部注入 spec」:
|
||
# 只有本产线自动 spec 才做 sourceHash 陈旧重生;金标考卷(fixtures/golden-specs,经 compare_node.inject_golden
|
||
# 强制覆写注入)不带此标识,held-constant 永不被 hash 检查覆盖(对照公平的落点,金标不误杀)。
|
||
_SPEC_GENERATOR = "cheap_run.ensure_play_spec"
|
||
# 源工程 hash 不可算(src/ 缺失 / IO 异常)时写进 derivedFrom.sourceHash 的占位值(C5 required 非空):
|
||
# 语义=「产 spec 时源不可读」;下次源可读时真 hash ≠ 占位 → 自然触发重生,不留恒等假绑定。
|
||
_SOURCE_HASH_UNAVAILABLE = "unavailable"
|
||
|
||
|
||
def _source_hash(game_id: str) -> Optional[str]:
|
||
"""算源工程内容 hash(games/amgen-<id>/src/ 全部文件,按相对路径排序聚合 sha256)。
|
||
|
||
C5 derivedFrom.sourceHash 的取数源:agent 只写 src/ 下 L3 文件(write_file 边界),故 src/ 内容
|
||
即「源工程会变的全部」;续修轮 agent 改了任一 src 文件 → hash 变 → 旧考卷判陈旧、重生。
|
||
读不到(目录缺失 / IO 异常)返回 None——调用方回落保守行为(不据 hash 重生),绝不抛。
|
||
"""
|
||
try:
|
||
src = game_dir(game_id) / "src"
|
||
if not src.is_dir():
|
||
return None
|
||
h = hashlib.sha256()
|
||
for f in sorted(src.rglob("*")):
|
||
if f.is_file():
|
||
# 路径与内容都进 hash(文件改名/增删也算源变),NUL 分隔防拼接歧义。
|
||
h.update(f.relative_to(src).as_posix().encode("utf-8"))
|
||
h.update(b"\0")
|
||
h.update(f.read_bytes())
|
||
h.update(b"\0")
|
||
return h.hexdigest()
|
||
except Exception: # noqa: BLE001 —— hash 失败按「不可算」回落,不连累门流水线
|
||
return None
|
||
|
||
|
||
def _derived_from(source_hash: Optional[str]) -> dict:
|
||
"""组 C5 derivedFrom 派生绑定段(spec↔源工程绑定锚;hash 不可算时写占位、不留恒等假绑定)。"""
|
||
return {
|
||
"sourceHash": source_hash or _SOURCE_HASH_UNAVAILABLE,
|
||
"generatedAt": str(int(time.time() * 1000)), # 毫秒时间戳字符串(C5 允许 string|integer,取 string 稳妥)
|
||
"generator": _SPEC_GENERATOR,
|
||
"regenerateOnSourceChange": True,
|
||
}
|
||
|
||
|
||
def _build_play_spec(state: Optional[dict], *, source_hash: Optional[str] = None) -> dict:
|
||
"""据 _forensicsView().state() 形态产 play-spec dict(纯函数、不写盘,便于单测;产物须过 C5 契约校验)。
|
||
|
||
tap-targets occupied 族(state 有 targets 数组):加厚到金标同质——补 F_wiring 的 expectedEngineCallPrefixes
|
||
语义期望(外生 held 期望、预置常量,防自证);key-cycle 族(按键类):薄版,期望前缀待输入键契约(WU-C 5.4)。
|
||
C5 接线(W-S1 单①):driver 带显式 selectionBasis(把「state 有无 targets/target 键定族」这条约定写进
|
||
考卷自解释,不再是只活在本函数里的隐形开关);key-cycle 族带 startRitual 默认起局仪式(菜单/开始页游戏
|
||
先起局再按键,治「driver 只按键、不会起局 → 卡 menu」的确定性漂移);两族都带 derivedFrom 派生绑定。
|
||
"""
|
||
# 以键存在性判定「可点目标」类游戏:游戏暴露 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,
|
||
# C5:驱动器族选择依据显式进考卷(自解释,取代读代码才知道的隐形约定)。
|
||
"selectionBasis": "state 暴露 targets/target 键 → 可点目标类 → tap-targets occupied 族",
|
||
}
|
||
return {
|
||
"schemaVersion": "play-spec/1",
|
||
"derivedFrom": _derived_from(source_hash),
|
||
"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 不致命
|
||
# tap-targets occupied 族不声明 startRitual:runTapTargets 已内建「非游玩态→中线扫点起局」
|
||
# (play.cdp.cjs 场景推进段),再声明会重复动作;key-cycle 族才需要显式仪式。
|
||
}
|
||
driver = {
|
||
"type": "key-cycle", "keys": ["ArrowLeft", "ArrowRight", "ArrowUp", "ArrowDown"],
|
||
"steps": 60, "stepMs": 180, "downMs": 60,
|
||
"selectionBasis": "state 无 targets/target 键 → 按键类回退 → key-cycle 族",
|
||
}
|
||
return {
|
||
"schemaVersion": "play-spec/1",
|
||
"derivedFrom": _derived_from(source_hash),
|
||
"exportState": ["phase", "score"],
|
||
"driver": driver,
|
||
# C5 startRitual(W-S1 单①):key-cycle 游戏停在菜单/开始页时按键循环驱不动(fx4c 卡 menu 实证),
|
||
# 先依序试 Enter / Space / 画布中心点起局(harness 每步后查 phase,进游玩态即止)。
|
||
"startRitual": {
|
||
"kind": "input-sequence",
|
||
"inputs": [
|
||
{"t": "key", "code": "Enter"},
|
||
{"t": "key", "code": "Space"},
|
||
{"t": "tap", "x": 195, "y": 422},
|
||
],
|
||
"why": "菜单/开始页游戏需先起局,按键循环才驱得动(未进游玩态则 G_input/H_progress 必挂)",
|
||
},
|
||
"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 _validate_c5_best_effort(spec: dict) -> None:
|
||
"""产出的 play-spec 对 C5 契约(contracts/play-loop)做 best-effort 自校验:失败只告警、绝不阻断门流水线。
|
||
|
||
校验器 = contracts/play-loop/validate.py(零依赖 stdlib),经 importlib 按路径加载(它不是包)。
|
||
为什么只告警不阻断:校验挂生产路径是为了漂移可见(响亮日志),阻断会让契约演进期的新字段直接打红生成
|
||
主链——强校验在单测层(tests/test_play_spec_contract.py 对两族产物全量过 C5)。
|
||
"""
|
||
try:
|
||
import importlib.util # noqa: PLC0415
|
||
|
||
vp = _REPO_ROOT / "contracts" / "play-loop" / "validate.py"
|
||
sp = _REPO_ROOT / "contracts" / "play-loop" / "play-spec.schema.json"
|
||
if not (vp.exists() and sp.exists()):
|
||
return # 契约目录不在(裁剪部署)→ 静默跳过
|
||
mod_spec = importlib.util.spec_from_file_location("_playloop_validate", vp)
|
||
mod = importlib.util.module_from_spec(mod_spec)
|
||
mod_spec.loader.exec_module(mod)
|
||
schema = json.loads(sp.read_text(encoding="utf-8"))
|
||
errors = mod.validate(schema, spec, schema)
|
||
if errors:
|
||
print(f"[cheap-run] ⚠ 自动 play-spec 未过 C5 契约校验(不阻断,请修 _build_play_spec):{errors[:3]}",
|
||
flush=True)
|
||
except Exception as e: # noqa: BLE001 —— 校验器自身异常绝不连累生成
|
||
print(f"[cheap-run] C5 自校验异常(忽略):{type(e).__name__}: {e}", flush=True)
|
||
|
||
|
||
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 → 假绿"。
|
||
|
||
C5 陈旧可判定(W-S1 单①,取代旧「已存在一律不覆盖」):
|
||
· 已存在且是【本产线自动 spec】(derivedFrom.generator 以 cheap_run.ensure_play_spec 开头):
|
||
比对 derivedFrom.sourceHash 与当前源工程 hash——一致=源未变,复用;不一致=源已变(续修轮 agent
|
||
改过代码),旧考卷陈旧、重生。这是「拿旧考卷判新工程 → 假绿/错判」的封口点。
|
||
· 已存在但【非本产线出品】(无 derivedFrom / generator 异,即手工金标或外部注入,如
|
||
compare_node.inject_golden 的金标考卷):held-constant 保持不覆盖——金标是对照公平的锚,
|
||
绝不被 hash 检查误杀;旧世代自动 spec(无 derivedFrom)同样落此保守支,新链路产的 spec 都带
|
||
绑定、增量收敛。
|
||
· 源 hash 不可算(src/ 缺失):保守不覆盖已有 spec(判不了陈旧就不动)。
|
||
|
||
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"
|
||
cur_hash = _source_hash(game_id)
|
||
if spec_path.exists():
|
||
old_generator = None
|
||
old_hash = None
|
||
try:
|
||
old = json.loads(spec_path.read_text(encoding="utf-8"))
|
||
df = old.get("derivedFrom") if isinstance(old, dict) else None
|
||
if isinstance(df, dict):
|
||
old_generator = df.get("generator")
|
||
old_hash = df.get("sourceHash")
|
||
except Exception: # noqa: BLE001 —— 坏 JSON:按非本产线处理(保守不覆盖,由人工清理)
|
||
pass
|
||
is_auto_spec = isinstance(old_generator, str) and old_generator.startswith(_SPEC_GENERATOR)
|
||
if not is_auto_spec:
|
||
# 手工金标 / 外部注入 / 旧世代无绑定 spec:held-constant,不覆盖(金标不误杀)。
|
||
return {"ok": True, "wrote": False,
|
||
"reason": "已存在且非本产线自动 spec(手工金标/外部注入/旧世代),held-constant 不覆盖"}
|
||
if cur_hash is None:
|
||
return {"ok": True, "wrote": False, "reason": "已存在;源工程 hash 不可算,保守不覆盖"}
|
||
if old_hash == cur_hash:
|
||
return {"ok": True, "wrote": False, "reason": "已存在且 sourceHash 一致(源工程未变),复用"}
|
||
print(f"[cheap-run] play-spec 陈旧(sourceHash 已变):game={game_id} 旧={str(old_hash)[:12]}… "
|
||
f"新={cur_hash[:12]}… → 重生考卷", flush=True)
|
||
spec = _build_play_spec(state, source_hash=cur_hash)
|
||
_validate_c5_best_effort(spec) # 生产路径 C5 自校验(告警不阻断;强校验在单测)
|
||
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}"}
|