cbfd4d871b
固化 Match-3 生产者、视觉、音频与双 Judge 证据闭包。 将《山海行纪》r1.1 绑定新的不可变 release，并以生产预检现场核验 bundle、Registry/2 和 25 项 Writer 快照。 同步地图1平衡锁值、跨游戏回归修复、验收契约与 SoT 证据。
683 lines
38 KiB
Python
683 lines
38 KiB
Python
"""cheap_modify.py — A11 切片三 M3 便宜档「确定性类纠错执行」(execution 段零 LLM)。
|
||
|
||
两段式回路第二段。第一段(cheap_classify,M2)用 LLM 判用户原话的意图 → 建议改动
|
||
{mode,target,payload};用户确认后前端调已有 /modify 执行。本模块是 deterministic 类
|
||
(换资产 / 调数值 / 改关卡)的执行核心:**改"规范工程那一处" → 落盘 → 重建(esbuild)→ 九门 → 改动清单**。
|
||
判意图的 LLM 在第一段已做完,故 execution 段是纯文本定位替换、**零 LLM**。
|
||
|
||
工程规范性(创始人 2026-06-28 定,本模块寻址的地基):便宜档产物已规范——
|
||
· 资产统一在 src/assets.js 的 IMAGE_FILES(键→文件名);
|
||
· 核心数值集中在 src/core.js 的 `export const NAME = VALUE`。
|
||
确定性执行 = 在这"一处"做文本替换,而不是 LLM 在散落代码里定位。
|
||
|
||
三件:
|
||
· apply_deterministic_modify(纯函数·TDD 核心):patch → 改后 files + 改动清单 manifest + ok/no-op 判定。
|
||
· materialize_source_project(系统级落盘):把 base 源 files 直接写到 game_dir——**系统恢复已知良源**,
|
||
绕 cheap_run.write_file 的 L1_FIXED 守卫(那守卫是防 *agent* 写坏 wiring;这里是系统落盘,只仍校验不越界)。
|
||
· execute_deterministic_modify(集成):apply → scaffold 已知良 plumbing → materialize 覆盖改后源 →
|
||
build → stage → smoke → 九门 play(undriven),**复用 cheap_run 现有九门,不另造**。
|
||
apply no-op/unsupported → 直接 failed(带 manifest + error),不重建(省 esbuild/Chrome)。
|
||
|
||
best-effort 铁律:apply 失败 / 重建失败 → 返 failed + 诊断,绝不抛崩 worker。只改 L3(core/assets/game-logic),
|
||
不碰 L1 plumbing。
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import cheap_assert
|
||
import cheap_run
|
||
import cheap_verify # W-AXIS-V2 波1:modify 两档改调统一验收编排器(create/modify 同标准)
|
||
|
||
|
||
def _read_game_brief(game_id: str) -> str:
|
||
"""读原游戏 brief(evidence/brief.json;modify 验收给测试员上下文——改后仍应是同一款可玩游戏)。缺/坏 → 空串。"""
|
||
try:
|
||
p = cheap_run.game_dir(game_id) / "evidence" / "brief.json"
|
||
if p.is_file():
|
||
d = json.loads(p.read_text(encoding="utf-8"))
|
||
return (d.get("brief") if isinstance(d, dict) else "") or ""
|
||
except Exception: # noqa: BLE001 brief 读失败不致命,测试员用空 brief 仍可判 broken/hollow
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _resolve_a11_template_route(source_project) -> tuple[str | None, str | None]:
|
||
"""只从源工程可信元数据恢复 templateRoute/genre；v3 禁止用 brief 或单独 genre 反推 profile。"""
|
||
sp = _as_dict(source_project)
|
||
template = sp.get("scaffoldTemplate")
|
||
genre = cheap_verify.GENRE_BY_TEMPLATE.get(template) if template else None
|
||
if template and genre in cheap_verify._V3_GENRES:
|
||
return str(template), str(genre)
|
||
return None, None
|
||
|
||
|
||
def _resolve_a11_origin_brief(source_project) -> tuple[str | None, str | None]:
|
||
"""从 base SourceProject 恢复不可变原题面；active v3 禁止依赖本机 evidence 残留。"""
|
||
sp = _as_dict(source_project)
|
||
brief = sp.get("originBrief")
|
||
brief_hash = sp.get("originBriefHash")
|
||
if (not isinstance(brief, str) or not brief.strip() or not isinstance(brief_hash, str)
|
||
or hashlib.sha256(brief.encode("utf-8")).hexdigest() != brief_hash):
|
||
return None, None
|
||
return brief, brief_hash
|
||
|
||
# A11 M4 模块重生成：behavior 在便宜档映射到玩法文件（KTD4）——只重写它，其余文件保持不变。
|
||
_BEHAVIOR_TARGET = "game-logic.js"
|
||
_TARGET_REL = "src/game-logic.js"
|
||
# 断言②非目标稳定：game_dir 根下要 track 稳定性的 L1 入口/页面（src/*.js 用 glob，build 衍生物 bundle.iife.js 不 track）。
|
||
_ROOT_STABLE = ("index.html", "entry-bundle.js", "entry.js")
|
||
|
||
# 调数值(config/level)寻址的 L3 文件优先序:核心数值集中在 core.js,找不到再搜其它 L3。
|
||
_CONFIG_FILE_ORDER = ("core.js", "game-logic.js", "render.js", "balance.js")
|
||
# 换资产寻址的 L3 文件:资产清单统一在 assets.js。
|
||
_ASSET_FILE = "assets.js"
|
||
|
||
|
||
def log(msg: str) -> None:
|
||
"""带前缀日志(stdout,便于 nohup/setsid 落盘排障)。"""
|
||
print(f"[cheap-modify] {msg}", flush=True)
|
||
|
||
|
||
# ───────────────────────── 容错解析 / JS 字面量序列化 ─────────────────────────
|
||
|
||
def _as_dict(obj) -> dict:
|
||
"""obj 容错成 dict(后端经 HTTP 传的是 JSON 串;也容忍已是 dict)。坏输入返 {}。"""
|
||
if isinstance(obj, dict):
|
||
return obj
|
||
if isinstance(obj, str) and obj.strip():
|
||
try:
|
||
parsed = json.loads(obj)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except Exception: # noqa: BLE001 — 坏 JSON 容错返空,由上层判 no-op,不抛
|
||
return {}
|
||
return {}
|
||
|
||
|
||
def _to_js_literal(value) -> str:
|
||
"""把 Python 值序列化成一个合法 JS 字面量(数值/字符串/数组/对象/布尔)。
|
||
|
||
用 json.dumps:20000→"20000"、0.9→"0.9"、"#000"→双引号串、[1,2]→"[1, 2]"、True→"true"——
|
||
JSON 子集恒是合法 JS。既有源用单引号,这里出双引号亦合法(esbuild 照吃),不强求引号风格一致。
|
||
非 JSON 可序列化(罕见)兜底为 str()。
|
||
"""
|
||
try:
|
||
return json.dumps(value, ensure_ascii=False)
|
||
except (TypeError, ValueError):
|
||
return str(value)
|
||
|
||
|
||
def _match_file_key(files: dict, basename: str):
|
||
"""在 files(路径→文本)里找 basename 对应的键(容忍 "src/core.js" 或 "core.js" 两种键形)。无 → None。"""
|
||
for key in files:
|
||
if key == basename or key.endswith("/" + basename):
|
||
return key
|
||
return None
|
||
|
||
|
||
# ───────────────────────── 文本定位替换:常量 / 资产键 ─────────────────────────
|
||
|
||
def _replace_const(text: str, name: str, new_literal: str):
|
||
"""把 `export const <name> = <旧值>;` 的值替换成 new_literal。返回 (新文本, 旧值字符串|None)。
|
||
|
||
锚 `export const NAME =` 到行尾 `;`,两层稳健:
|
||
· 单行优先:值不含换行与分号(数值 / 字符串 / 单行数组),`^[ \\t]*` 锚行首防误匹配注释/字符串内文本;
|
||
· 多行兜底:`Object.freeze([...])` 跨行、值内无行尾分号 → DOTALL 非贪到行尾分号。
|
||
只替第一处(`export const` 同名应唯一)。找不到 → (原文, None)(由上层记 found=false=no-op 地基)。
|
||
"""
|
||
# 单行:值在一行内,不跨换行、不含分号。
|
||
pat_single = re.compile(
|
||
r"(^[ \t]*export\s+const\s+" + re.escape(name) + r"\s*=\s*)([^\n;]*)(\s*;)",
|
||
re.MULTILINE,
|
||
)
|
||
m = pat_single.search(text)
|
||
if m is None:
|
||
# 多行兜底:跨行值(如 Object.freeze 多行数组),锚到「行尾的分号」。
|
||
pat_multi = re.compile(
|
||
r"(export\s+const\s+" + re.escape(name) + r"\s*=\s*)(.*?)(;[ \t]*$)",
|
||
re.DOTALL | re.MULTILINE,
|
||
)
|
||
m = pat_multi.search(text)
|
||
if m is None:
|
||
return text, None
|
||
old = m.group(2).strip()
|
||
new_text = text[: m.start(2)] + new_literal + text[m.end(2):]
|
||
return new_text, old
|
||
|
||
|
||
def _replace_asset_value(text: str, key: str, new_value: str):
|
||
"""把 IMAGE_FILES 里 `<key>: '<旧文件名>'` 的值替换成 new_value(保留原引号风格)。
|
||
|
||
容忍键带/不带引号(heroSprite: / 'heroSprite': / "heroSprite":);值引号用反引用配对(单/双引号都行)。
|
||
返回 (新文本, 旧值|None)。空 IMAGE_FILES = {} 或键不存在 → (原文, None)=no-op 地基。
|
||
"""
|
||
key_pat = r"['\"]?" + re.escape(key) + r"['\"]?"
|
||
pat = re.compile(r"(" + key_pat + r"\s*:\s*)(['\"])(.*?)\2")
|
||
m = pat.search(text)
|
||
if m is None:
|
||
return text, None
|
||
old = m.group(3)
|
||
# 只换值(group3),保留原引号 group2;new_value 是裸文件名/URL,引号由原文提供。
|
||
new_text = text[: m.start(3)] + str(new_value) + text[m.end(3):]
|
||
return new_text, old
|
||
|
||
|
||
def _replace_base_url(text: str, new_value: str):
|
||
"""把 loadAssets 的默认资产根 `baseUrl == null ? '<旧>' : baseUrl` 中的 <旧> 替换成 new_value。
|
||
|
||
返回 (新文本, 旧值|None)。匹配不到(无该默认表达式)→ (原文, None)。
|
||
"""
|
||
pat = re.compile(r"(baseUrl\s*==\s*null\s*\?\s*)(['\"])(.*?)\2")
|
||
m = pat.search(text)
|
||
if m is None:
|
||
return text, None
|
||
old = m.group(3)
|
||
new_text = text[: m.start(3)] + str(new_value) + text[m.end(3):]
|
||
return new_text, old
|
||
|
||
|
||
# ───────────────────────── ① apply_deterministic_modify(纯函数) ─────────────────────────
|
||
|
||
def _apply_config(files: dict, value: dict, kind: str) -> list:
|
||
"""config/level:对 value 的每个 {常量名:新值},在 L3 文件优先序里替换 `export const 名 = 值`。返回 manifest。"""
|
||
manifest = []
|
||
for name, new_val in value.items():
|
||
new_lit = _to_js_literal(new_val)
|
||
found = False
|
||
target_key = None
|
||
old = None
|
||
for base in _CONFIG_FILE_ORDER:
|
||
fk = _match_file_key(files, base)
|
||
if fk is None:
|
||
continue
|
||
new_text, old_val = _replace_const(files[fk], name, new_lit)
|
||
if old_val is not None:
|
||
files[fk] = new_text
|
||
found, target_key, old = True, fk, old_val
|
||
break
|
||
manifest.append({
|
||
"file": target_key or ("src/" + _CONFIG_FILE_ORDER[0]),
|
||
"kind": kind, "key": name,
|
||
"old": old, "new": new_lit, "found": found,
|
||
})
|
||
return manifest
|
||
|
||
|
||
def _apply_asset(files: dict, value: dict) -> list:
|
||
"""asset:对 value 的每个 {资产键:新文件名/URL}(或 baseUrl),在 assets.js 替换。返回 manifest。"""
|
||
manifest = []
|
||
fk = _match_file_key(files, _ASSET_FILE)
|
||
for key, new_val in value.items():
|
||
found = False
|
||
old = None
|
||
if fk is not None:
|
||
if key == "baseUrl":
|
||
new_text, old_val = _replace_base_url(files[fk], str(new_val))
|
||
else:
|
||
new_text, old_val = _replace_asset_value(files[fk], key, str(new_val))
|
||
if old_val is not None:
|
||
files[fk] = new_text
|
||
found, old = True, old_val
|
||
manifest.append({
|
||
"file": fk or ("src/" + _ASSET_FILE),
|
||
"kind": "asset", "key": key,
|
||
"old": old, "new": str(new_val), "found": found,
|
||
})
|
||
return manifest
|
||
|
||
|
||
def apply_deterministic_modify(source_project, modify_patch) -> dict:
|
||
"""确定性改一处(纯函数):据 patch 的 target.kind 改 base 源的规范一处,产改后 files + 改动清单。
|
||
|
||
Args:
|
||
source_project: base 源工程(dict 或 2.0 JSON 串,有 files=路径→文本)。
|
||
modify_patch: 改动意图(dict 或 JSON 串),形如
|
||
{"target":{"kind":"config|asset|level","path":"..."},"payload":{"value":{"键":"新值",...}}}。
|
||
|
||
Returns:
|
||
{"files": 改后 files dict, "manifest": [...], "ok": bool, "error": str|None}。
|
||
manifest 每条 {"file","kind","key","old","new","found"};**no-op 检测**:目标常量/键找不到 → found=false;
|
||
若无任何一条 found=true 且实际值变化(old≠new)→ 整体 ok=false(诚实判 no-op,不是真改动)。
|
||
"""
|
||
sp = _as_dict(source_project)
|
||
patch = _as_dict(modify_patch)
|
||
raw_files = sp.get("files")
|
||
if not isinstance(raw_files, dict) or not raw_files:
|
||
return {"files": {}, "manifest": [], "ok": False, "error": "源工程无 files(无可改源)"}
|
||
files = dict(raw_files) # 浅拷贝:整串替换、不原地改入参
|
||
|
||
target = patch.get("target") if isinstance(patch.get("target"), dict) else {}
|
||
kind = target.get("kind")
|
||
payload = patch.get("payload") if isinstance(patch.get("payload"), dict) else {}
|
||
value = payload.get("value")
|
||
|
||
if kind not in ("config", "asset", "level"):
|
||
return {"files": files, "manifest": [], "ok": False,
|
||
"error": f"非确定性类 kind={kind}(M3 只处理 config/asset/level)"}
|
||
if not isinstance(value, dict) or not value:
|
||
# level 复杂布局重排(value 非 {常量名:新值} 映射)→ unsupported;其余类同样要求映射。
|
||
hint = "level 复杂布局重排不在 M3(仅支持关卡常量文本替换)" if kind == "level" else "payload.value 须为 {键:新值} 非空映射"
|
||
return {"files": files, "manifest": [], "ok": False, "error": hint}
|
||
|
||
if kind in ("config", "level"):
|
||
manifest = _apply_config(files, value, kind)
|
||
else: # asset
|
||
manifest = _apply_asset(files, value)
|
||
|
||
# no-op 判定:至少一条「命中且实际变化」才算真改动。
|
||
ok = any(r["found"] and r["old"] != r["new"] for r in manifest)
|
||
error = None if ok else "no-op:无任何目标命中且实际变化(常量/键未找到,或新值与旧值相同)"
|
||
return {"files": files, "manifest": manifest, "ok": ok, "error": error}
|
||
|
||
|
||
# ───────────────────────── ② materialize_source_project(系统级落盘) ─────────────────────────
|
||
|
||
def materialize_source_project(game_id: str, files) -> Path:
|
||
"""把 base 源工程的全部 files 直接写到 cheap_run.game_dir(game_id)(系统恢复已知良源)。
|
||
|
||
**绕 cheap_run.write_file 的 L1_FIXED 守卫**——那守卫是防 *agent* 把 wiring 写坏(host-config/game/index 等);
|
||
这里是 *系统* 落盘已知良源(L1 plumbing + L3 都照写),故不拦 L1。仍校验路径不逃出 game_dir(防越界写)。
|
||
这是「在既有产物上只重建+重过门」轻入口的前半;后半(build+九门)由 execute_deterministic_modify 接。
|
||
|
||
Args:
|
||
game_id: 本 run 的 gameId(写 games/amgen-<id>/)。
|
||
files: 路径(如 "src/core.js")→ 文本 的 dict。
|
||
|
||
Returns:
|
||
game_dir(Path)。
|
||
"""
|
||
gd = cheap_run.game_dir(game_id)
|
||
gd_real = gd.resolve()
|
||
gd.mkdir(parents=True, exist_ok=True)
|
||
if not isinstance(files, dict):
|
||
return gd
|
||
for rel_path, content in files.items():
|
||
abs_p = (gd / rel_path).resolve()
|
||
# 越界守卫:解析后必须仍落在 game_dir 内(防 ../ 或绝对路径逃逸)。
|
||
if abs_p != gd_real and gd_real not in abs_p.parents:
|
||
log(f"materialize 跳过越界路径(不在 game_dir 内):{rel_path}")
|
||
continue
|
||
abs_p.parent.mkdir(parents=True, exist_ok=True)
|
||
abs_p.write_text(content if isinstance(content, str) else str(content), encoding="utf-8")
|
||
return gd
|
||
|
||
|
||
# ───────────────────────── ③ execute_deterministic_modify(集成) ─────────────────────────
|
||
|
||
def execute_deterministic_modify(game_id: str, source_project, modify_patch,
|
||
*, base_version_id=None, task_trace_id=None,
|
||
port: int = 4320, cdp_port: int = 9222) -> dict:
|
||
"""确定性执行集成:apply 改一处 → 落盘已知良源 → 重建 → 九门。复用 cheap_run 九门,不另造。
|
||
|
||
流程:apply_deterministic_modify → 若 ok:(缺 plumbing 才 scaffold 已知良 entry-bundle/index.html)→
|
||
materialize 覆盖改后 L3 源 → build(esbuild)→ stage → smoke → 九门 play(undriven)。
|
||
apply no-op/unsupported → 直接 failed(带 manifest + error),**不重建**(省 esbuild/Chrome)。
|
||
任一步失败 → failed + 诊断(stage 标到哪步断的),绝不抛崩。
|
||
|
||
Returns:
|
||
{"status":"succeeded|failed", "manifest":[...], "verdict":{...}|None, "stage":"apply|scaffold|build|stage|play",
|
||
"error":str|None, "smokeOk":bool|None, "playOk":bool|None}。
|
||
"""
|
||
mode = str(cheap_verify._acceptance_v3_cfg().get("mode") or "v3_shadow")
|
||
task_binding_hash = None
|
||
if mode in ("v3", "v3_shadow"):
|
||
try:
|
||
task_binding_hash = cheap_verify.task_binding_hash_v3(task_trace_id)
|
||
except ValueError as exc:
|
||
return {"status": "failed", "manifest": [], "verdict": None,
|
||
"stage": "apply", "error": str(exc), "smokeOk": None, "playOk": None}
|
||
|
||
# 1. apply:确定性改一处(纯文本定位替换,零 LLM)。
|
||
applied = apply_deterministic_modify(source_project, modify_patch)
|
||
manifest = applied["manifest"]
|
||
if not applied["ok"]:
|
||
# no-op / unsupported:不重建,直接 failed + 带 manifest + error。
|
||
log(f"apply 未产生真改动(no-op/unsupported)game_id={game_id}: {applied['error']}")
|
||
return {"status": "failed", "manifest": manifest, "verdict": None,
|
||
"stage": "apply", "error": applied["error"], "smokeOk": None, "playOk": None}
|
||
|
||
# 2. 确保可构建的 game_dir:已存在 plumbing(既有产物)→ 复用(保留 assets/);缺 → scaffold 已知良 plumbing。
|
||
gd = cheap_run.game_dir(game_id)
|
||
if not (gd / "entry-bundle.js").exists():
|
||
sc = cheap_run.scaffold(game_id) # scaffold-saa:克隆 _template + SAA 信封 entry-bundle.js/index.html
|
||
if not sc["ok"]:
|
||
log(f"scaffold 失败 game_id={game_id}: {sc.get('output', '')[:300]}")
|
||
return {"status": "failed", "manifest": manifest, "verdict": None,
|
||
"stage": "scaffold", "error": "scaffold 失败:" + sc.get("output", ""),
|
||
"smokeOk": None, "playOk": None}
|
||
|
||
# 3. 落盘改后源(覆盖 scaffold 的默认 L3 src/*,把 base 源 + 改动那一处写进去)。
|
||
materialize_source_project(game_id, applied["files"])
|
||
|
||
# 4. 重建 esbuild。
|
||
bd = cheap_run.build(game_id)
|
||
if not bd["ok"]:
|
||
log(f"build 失败 game_id={game_id}: {bd.get('output', '')[:300]}")
|
||
return {"status": "failed", "manifest": manifest, "verdict": None,
|
||
"stage": "build", "error": "build 失败:" + bd.get("output", ""),
|
||
"smokeOk": None, "playOk": None}
|
||
|
||
# 5. stage → smoke → 循环外九门 play(undriven;复用 cheap_studio 收口同一套;ensure_play_spec 波2 已摘)。
|
||
st = cheap_run.stage(game_id)
|
||
if not st["ok"]:
|
||
log(f"stage 失败 game_id={game_id}: {st.get('output', '')[:300]}")
|
||
return {"status": "failed", "manifest": manifest, "verdict": None,
|
||
"stage": "stage", "error": "stage 失败:" + st.get("output", ""),
|
||
"smokeOk": None, "playOk": None}
|
||
sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port) # boot 预检(smokeOk 供返回);其 state 不再喂 ensure_play_spec
|
||
pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port)
|
||
verdict = pr.get("verdict")
|
||
# create 与 modify 不允许两套验收标准：确定性改也调用 v3 唯一入口；仅显式历史模式回放 v1/v2。
|
||
# brief 恒取原游戏题面；若 v3 最终给 verified reject + repairEligible，才调用一次受限 writer，并沿 parentRun
|
||
# 复验同一 brief。同步执行路径用 asyncio.run 驱动异步验收。
|
||
import asyncio # noqa: PLC0415
|
||
template_route, genre = _resolve_a11_template_route(source_project)
|
||
origin_brief, origin_brief_hash = _resolve_a11_origin_brief(source_project)
|
||
brief = origin_brief if mode in ("v3", "v3_shadow") else _read_game_brief(game_id)
|
||
if mode in ("v3", "v3_shadow") and (not template_route or genre not in cheap_verify._V3_GENRES
|
||
or not origin_brief or not origin_brief_hash):
|
||
import cheap_studio # noqa: PLC0415
|
||
|
||
_acc = cheap_studio.apply_v3_entry_failure(
|
||
{"verdict": verdict},
|
||
"A11 sourceProject 缺可信 templateRoute/originBrief，禁止依赖本机 evidence 或修改指令猜原题面",
|
||
mode=mode)
|
||
gates_pass = False
|
||
elif mode in ("v3", "v3_shadow"):
|
||
import cheap_studio # noqa: PLC0415
|
||
|
||
acceptance_identity = cheap_verify.build_acceptance_v3_identity(
|
||
game_id, brief, genre=genre, template_route=template_route, repair_ordinal=0,
|
||
task_binding_hash=task_binding_hash)
|
||
_acc_v3 = asyncio.run(cheap_verify.run_acceptance_v3(cheap_studio.build_acceptance_v3_request(
|
||
game_id, brief, verdict, acceptance_identity=acceptance_identity,
|
||
idempotency_key=f"{game_id}:a11-deterministic-v3")))
|
||
_acc = cheap_studio.apply_acceptance_v3({"verdict": verdict}, _acc_v3)
|
||
decision = _acc_v3.get("decision") or {}
|
||
if cheap_verify.is_v3_repair_authorized(_acc_v3):
|
||
# 确定性改本身零 LLM；只有 v3 最终硬证拒绝时，才把同一产物交受限 writer 修一次。
|
||
try:
|
||
repair_identity = cheap_verify.build_acceptance_v3_identity(
|
||
game_id, brief, genre=acceptance_identity["genre"],
|
||
template_route=acceptance_identity["templateRoute"],
|
||
source_artifact_hash=_acc_v3["artifactHash"],
|
||
parent_acceptance_request_hash=acceptance_identity["acceptanceRequestHash"],
|
||
repair_ordinal=1, proof_profile_id=acceptance_identity["proofProfileId"],
|
||
proof_registry_version=acceptance_identity["proofRegistryVersion"],
|
||
task_binding_hash=acceptance_identity["taskBindingHash"],
|
||
)
|
||
repair_summary = _default_rewrite(
|
||
game_id, decision.get("repairFeedback") or "按 v3 硬证修复玩法闭环",
|
||
port=port, cdp_port=cdp_port,
|
||
acceptance_parent_run_id=_acc_v3.get("runId"), acceptance_repair_count=1,
|
||
acceptance_brief=brief, genre=genre, acceptance_identity=repair_identity,
|
||
acceptance_template_route=template_route,
|
||
acceptance_task_trace_id=task_trace_id)
|
||
except Exception as e: # noqa: BLE001 —— A11 执行器必须返回 failed 摘要，不能把 worker 主循环打崩
|
||
repair_summary = None
|
||
log(f"v3 单次 writer 修复异常(保留首轮 reject):game_id={game_id} {type(e).__name__}: {e}")
|
||
if isinstance(repair_summary, dict):
|
||
final_acceptance = repair_summary.get("acceptanceV3")
|
||
_acc = (cheap_studio.apply_acceptance_v3(repair_summary, final_acceptance, first_pass=_acc_v3)
|
||
if isinstance(final_acceptance, dict) else repair_summary)
|
||
verdict = repair_summary.get("verdictFull") or verdict
|
||
gates_pass = cheap_verify.is_v3_publishable(_acc.get("acceptanceV3") or {})
|
||
else:
|
||
_acc = asyncio.run(cheap_verify.run_acceptance(
|
||
{"verdict": verdict}, game_id=game_id, brief=brief, verdict=verdict))
|
||
gates_pass = bool(_acc.get("accepted"))
|
||
|
||
# M5 三结构断言:base→改后 真比对(断言①改动真生效 / 断言②非目标稳)+ 血缘(③ baseVersionId)。
|
||
# 断言②对 deterministic 是兜 apply bug(理应只改目标文件,真比对抓"是否顺手碰了非目标")。
|
||
base_files = _as_dict(source_project).get("files") if isinstance(_as_dict(source_project).get("files"), dict) else {}
|
||
target_files = {m["file"] for m in manifest if m.get("file")}
|
||
assertions = cheap_assert.three_assertions(manifest, base_files, applied["files"], target_files, base_version_id)
|
||
a12_pass = assertions["a1_changeApplied"]["pass"] and assertions["a2_nontargetStable"]["pass"]
|
||
status = "succeeded" if (gates_pass and a12_pass) else "failed"
|
||
err = None
|
||
if status != "succeeded":
|
||
err = "验收未通过(统一验收入口)" if not gates_pass else "三断言未通过:" + json.dumps(
|
||
{k: v for k, v in assertions.items() if isinstance(v, dict) and not v.get("pass")}, ensure_ascii=False)
|
||
log(f"确定性改重建+统一验收+三断言完成 game_id={game_id} status={status} gates={gates_pass} a12={a12_pass} "
|
||
f"mode={_acc.get('acceptanceVersion')}")
|
||
return {"status": status, "manifest": manifest, "verdict": verdict, "assertions": assertions,
|
||
"stage": "play", "error": err, "smokeOk": sm.get("ok"), "playOk": pr.get("ok"),
|
||
# 统一验收摘要 additive；v3 完整封存对象保留在 acceptanceV3。
|
||
"acceptance": {k: _acc.get(k) for k in ("acceptanceVersion", "floor", "playtest", "judge", "accepted",
|
||
"acceptanceV3", "acceptanceV3FirstPass", "publishFrozen",
|
||
"firstPassAccepted", "repairAttempted", "acceptedAfterRepair")}}
|
||
|
||
|
||
# ───────────────────────── ④ execute_regenerate_modify（模块重生成 · 改玩法）─────────────────────────
|
||
|
||
def _hash_file(p: Path) -> str:
|
||
"""文件内容 sha256（断言②的稳定性指纹）。"""
|
||
return hashlib.sha256(p.read_bytes()).hexdigest()
|
||
|
||
|
||
def _snapshot_nontarget_hashes(game_id: str) -> dict:
|
||
"""对 game_dir 内**非目标源文件**算 hash 快照（断言②地基）。返回 {rel_path: sha256}。
|
||
|
||
非目标 = 除 game-logic.js 外的全部源文件：src/ 下其余 .js（core/render/assets/balance + L1 的
|
||
host-config/game）+ 根下 L1 入口/页面（index.html/entry-bundle.js/entry.js）。**不 track build 衍生物**
|
||
（bundle.iife.js 每次 build 重生成，track 它会假性不稳）。重写前后各拍一次、逐键比对即知非目标有没有被误伤。
|
||
"""
|
||
gd = cheap_run.game_dir(game_id)
|
||
out = {}
|
||
src = gd / "src"
|
||
if src.is_dir():
|
||
for p in sorted(src.glob("*.js")):
|
||
if p.name == _BEHAVIOR_TARGET:
|
||
continue # 目标文件本身不算「非目标」
|
||
out[f"src/{p.name}"] = _hash_file(p)
|
||
for name in _ROOT_STABLE:
|
||
p = gd / name
|
||
if p.is_file():
|
||
out[name] = _hash_file(p)
|
||
return out
|
||
|
||
|
||
def _regen_manifest(intent: str, *, changed, stable) -> dict:
|
||
"""模块重生成改动清单（单条 dict，区别确定性类的多条 list）：哪个文件 / 哪类 / 意图 / 变没变 / 非目标稳没稳。"""
|
||
return {"file": _TARGET_REL, "kind": "behavior", "intent": intent,
|
||
"changed": changed, "untouchedStable": stable}
|
||
|
||
|
||
def _regen_why(gates_pass: bool, untouched_stable: bool, changed: bool = True) -> str:
|
||
"""失败诊断文案：断言②泄漏优先（误伤非目标比九门不过更严重）→ 断言①no-op → 九门。"""
|
||
if not untouched_stable:
|
||
return "断言②失败：模块重生成误伤了非目标文件（core/render/assets/host-config 等应保持不变）"
|
||
if not changed:
|
||
return "断言①失败：模块重生成未改变 game-logic.js（no-op，目标文件没动）"
|
||
if not gates_pass:
|
||
return "九门未通过"
|
||
return "模块重生成失败"
|
||
|
||
|
||
def _noop_prepare(game_id: str) -> dict:
|
||
"""run_studio 的 prepare 注入位：game_dir 已由 execute_regenerate_modify 预备（scaffold+materialize base 源），
|
||
不要再 scaffold 覆盖掉 base 源。返回 scaffold 同形契约 {ok,output}。"""
|
||
return {"ok": True, "output": "game_dir 已由 execute_regenerate_modify 预备（scaffold+materialize base 源）"}
|
||
|
||
|
||
def _default_rewrite(game_id: str, intent: str, *, port: int = 4320, cdp_port: int = 9222,
|
||
acceptance_parent_run_id=None, acceptance_repair_count=0,
|
||
acceptance_brief=None, genre=None, acceptance_identity=None,
|
||
acceptance_template_route=None,
|
||
acceptance_task_trace_id=None) -> dict:
|
||
"""默认重写器：跑 cheap_studio.run_studio 的 modify 态（自定义 system prompt + kick + 写边界白名单 +
|
||
prepare=noop[game_dir 已就绪]），复用同一套 ReAct + resume 熔断 + 三层校验收口。返回 run-summary（含九门 verdict）。
|
||
|
||
懒导入 cheap_studio/cheap_roles：只在真运行时装配 tier2 框架 + key，单测注入 rewrite_fn 不触发本函数。
|
||
"""
|
||
import asyncio
|
||
|
||
import cheap_roles
|
||
import cheap_studio
|
||
|
||
sys_prompt = cheap_roles.build_modify_system_prompt(game_id, intent)
|
||
g = f"game-runtime/games/amgen-{game_id}"
|
||
kick = (f"{g}/src/game-logic.js 现在是一款能跑能玩的游戏。请**只重写它**来实现这个调整：「{intent}」。"
|
||
f"先 read_file 读当前的 {g}/src/game-logic.js 看现状，再据调整意图最小改写；"
|
||
"core/render/assets/host-config 等其余文件保持不变、别写它们（写会被拒）。"
|
||
"check 与 build 都绿了就立即 finish。")
|
||
# 复用既有 max_iters(40)/max_resumes(6) resume 熔断，不新造。
|
||
return asyncio.run(cheap_studio.run_studio(
|
||
game_id, intent, system_prompt=sys_prompt, initial_kick=kick,
|
||
write_whitelist={_BEHAVIOR_TARGET}, prepare=_noop_prepare,
|
||
port=port, cdp_port=cdp_port,
|
||
acceptance_parent_run_id=acceptance_parent_run_id,
|
||
acceptance_repair_count=acceptance_repair_count,
|
||
acceptance_brief=acceptance_brief, genre=genre,
|
||
acceptance_identity=acceptance_identity,
|
||
acceptance_template_route=acceptance_template_route,
|
||
acceptance_task_trace_id=acceptance_task_trace_id))
|
||
|
||
|
||
def execute_regenerate_modify(game_id: str, source_project, modify_patch,
|
||
*, base_version_id=None, task_trace_id=None, rewrite_fn=None,
|
||
port: int = 4320, cdp_port: int = 9222) -> dict:
|
||
"""模块重生成执行（改玩法）：取 base 源 + intent → 有界单文件重写 game-logic.js → 九门 → 断言②非目标稳定 + 改动清单。
|
||
|
||
流程：取 intent（空→failed 不重写）→ 取 base 源 files（无源→failed）→ scaffold-if-needed + materialize
|
||
base 源（同 deterministic 口径）→ 重写前对目标+非目标算 hash → 有界重写 game-logic.js（复用 cheap_studio
|
||
ReAct+resume+三层校验，写边界收窄到只许写 game-logic.js）→ 重写后再算 hash 判 changed / untouchedStable →
|
||
据九门 verdict + 断言②判 status。best-effort：任一步异常→failed+诊断，绝不抛崩 worker。
|
||
|
||
Args:
|
||
game_id: 本 run gameId。
|
||
source_project: base 源工程（dict 或 2.0 JSON 串，有 files=路径→文本）。
|
||
modify_patch: 改动意图，形如 {"target":{"kind":"behavior"},"payload":{"intent":"<改写意图>"}}。
|
||
rewrite_fn: (game_id, intent, *, port, cdp_port)->run_summary 的重写器注入位（单测桩免真 LLM/esbuild/Chrome）；
|
||
默认 _default_rewrite（真跑 cheap_studio modify 态）。
|
||
port/cdp_port: serve/CDP 端口（透传 run_studio 收口）。
|
||
|
||
Returns:
|
||
{"status":"succeeded|failed", "stage":..., "manifest":{...}, "verdict":九门verdict|None,
|
||
"summary":run-summary|None, "error":str|None}。status=succeeded ⟺ 九门过 ∧ 断言②非目标稳定。
|
||
manifest = {"file":"src/game-logic.js","kind":"behavior","intent":intent,"changed":bool,"untouchedStable":bool}。
|
||
"""
|
||
mode = str(cheap_verify._acceptance_v3_cfg().get("mode") or "v3_shadow")
|
||
task_binding_hash = None
|
||
if mode in ("v3", "v3_shadow"):
|
||
try:
|
||
task_binding_hash = cheap_verify.task_binding_hash_v3(task_trace_id)
|
||
except ValueError as exc:
|
||
return {"status": "failed", "stage": "intent",
|
||
"manifest": _regen_manifest("", changed=False, stable=None),
|
||
"verdict": None, "summary": None, "error": str(exc)}
|
||
|
||
# 1. 取 intent（改写意图）；空 → failed，不重写（省一切重活）。
|
||
patch = _as_dict(modify_patch)
|
||
payload = patch.get("payload") if isinstance(patch.get("payload"), dict) else {}
|
||
intent = payload.get("intent")
|
||
intent = intent.strip() if isinstance(intent, str) else ""
|
||
if not intent:
|
||
log(f"模块重生成 intent 为空 → failed 不重写 game_id={game_id}")
|
||
return {"status": "failed", "stage": "intent",
|
||
"manifest": _regen_manifest("", changed=False, stable=None),
|
||
"verdict": None, "summary": None,
|
||
"error": "modifyPatch.payload.intent 为空（改写意图缺失），不重写"}
|
||
|
||
# 2. 取 base 源 files；无源 → failed（无可改源）。
|
||
sp = _as_dict(source_project)
|
||
base_files = sp.get("files") if isinstance(sp.get("files"), dict) else None
|
||
if not base_files:
|
||
log(f"模块重生成无 base 源 files → failed game_id={game_id}")
|
||
return {"status": "failed", "stage": "intent",
|
||
"manifest": _regen_manifest(intent, changed=False, stable=None),
|
||
"verdict": None, "summary": None,
|
||
"error": "base 源工程无 files（无可改源）"}
|
||
template_route, genre = _resolve_a11_template_route(sp)
|
||
origin_brief, origin_brief_hash = _resolve_a11_origin_brief(sp)
|
||
original_brief = origin_brief if mode in ("v3", "v3_shadow") else _read_game_brief(game_id)
|
||
if rewrite_fn is None and mode in ("v3", "v3_shadow") \
|
||
and (not template_route or genre not in cheap_verify._V3_GENRES
|
||
or not origin_brief or not origin_brief_hash):
|
||
return {"status": "failed", "stage": "intent",
|
||
"manifest": _regen_manifest(intent, changed=False, stable=None),
|
||
"verdict": None, "summary": None,
|
||
"error": "A11 sourceProject 缺可信 templateRoute/originBrief，禁止依赖本机 evidence 或修改指令猜原题面，不执行重生成"}
|
||
|
||
try:
|
||
# 3. prep：缺 plumbing 才 scaffold（补 L1 entry-bundle/index.html）→ materialize base 源覆盖 src/（同 deterministic 口径）。
|
||
gd = cheap_run.game_dir(game_id)
|
||
if not (gd / "entry-bundle.js").exists():
|
||
sc = cheap_run.scaffold(game_id)
|
||
if not sc["ok"]:
|
||
log(f"模块重生成 scaffold 失败 game_id={game_id}: {sc.get('output', '')[:300]}")
|
||
return {"status": "failed", "stage": "scaffold",
|
||
"manifest": _regen_manifest(intent, changed=False, stable=None),
|
||
"verdict": None, "summary": None,
|
||
"error": "scaffold 失败：" + sc.get("output", "")}
|
||
materialize_source_project(game_id, base_files)
|
||
|
||
# 4. 断言②地基：重写前对目标 + 非目标算 hash 快照。
|
||
target_p = gd / _TARGET_REL
|
||
target_before = _hash_file(target_p) if target_p.is_file() else None
|
||
nontarget_before = _snapshot_nontarget_hashes(game_id)
|
||
|
||
# 5. 有界单文件重写 game-logic.js（默认 cheap_studio modify 态；rewrite_fn 可注入桩）。
|
||
fn = rewrite_fn or _default_rewrite
|
||
identity = None
|
||
if rewrite_fn is None and mode in ("v3", "v3_shadow"):
|
||
# 在默认 Writer 启动前冻结 ordinal=0 身份；后续 run_studio 只验证和透传，不重选 profile。
|
||
identity = cheap_verify.build_acceptance_v3_identity(
|
||
game_id, original_brief, genre=genre, template_route=template_route, repair_ordinal=0,
|
||
task_binding_hash=task_binding_hash)
|
||
run_summary = fn(
|
||
game_id, intent, port=port, cdp_port=cdp_port, genre=genre,
|
||
acceptance_brief=original_brief, acceptance_identity=identity,
|
||
acceptance_template_route=template_route,
|
||
acceptance_task_trace_id=task_trace_id,
|
||
)
|
||
run_summary = run_summary if isinstance(run_summary, dict) else {}
|
||
|
||
# 6. 断言②：重写后再算 hash，判目标变没变 + 非目标稳没稳。
|
||
target_after = _hash_file(target_p) if target_p.is_file() else None
|
||
nontarget_after = _snapshot_nontarget_hashes(game_id)
|
||
changed = target_before != target_after
|
||
untouched_stable = nontarget_before == nontarget_after
|
||
manifest = _regen_manifest(intent, changed=changed, stable=untouched_stable)
|
||
except Exception as e: # noqa: BLE001 — 重写/落盘/hash 任一异常 best-effort 兜底，不抛崩 worker
|
||
log(f"模块重生成异常 game_id={game_id}: {type(e).__name__}: {e}")
|
||
return {"status": "failed", "stage": "code",
|
||
"manifest": _regen_manifest(intent, changed=False, stable=None),
|
||
"verdict": None, "summary": None,
|
||
"error": f"模块重生成异常：{type(e).__name__}: {e}"}
|
||
|
||
# 7. 据九门 verdict + 三断言判 status：九门过 ∧ 改动真生效(①) ∧ 非目标稳(②) 才算成功
|
||
# （误伤非目标=边界泄漏、目标没变=no-op，即便九门过也判失败）。
|
||
verdict_full = run_summary.get("verdictFull") if isinstance(run_summary.get("verdictFull"), dict) else None
|
||
vb = run_summary.get("verdict") if isinstance(run_summary.get("verdict"), dict) else None
|
||
# W-AXIS-V2 波1:_default_rewrite 走 run_studio(run_gates=True),收口已跑统一验收编排器,验收权威 = accepted
|
||
# (create/modify 同标准);run_summary 缺 accepted(注入桩/老路)→ 回落四门投影,不回落九门 verdict.pass(退役口径)。
|
||
v3_payload = run_summary.get("acceptanceV3")
|
||
v3_marker = isinstance(v3_payload, dict) or run_summary.get("acceptanceVersion") in ("v3", "v3_shadow")
|
||
if v3_marker:
|
||
# v3 标记存在但完整 payload 缺失时必须 fail-closed，禁止回落旧 accepted/四门。
|
||
gates_pass = cheap_verify.is_v3_publishable(v3_payload if isinstance(v3_payload, dict) else {})
|
||
else:
|
||
legacy_accepted = run_summary.get("accepted")
|
||
gates_pass = (bool(legacy_accepted) if legacy_accepted is not None
|
||
else cheap_verify.project_floor(verdict_full)["pass"])
|
||
# 三断言 verdict（统一形态，与 deterministic 路一致）：①changed ②非目标稳（含误伤清单）③血缘。
|
||
nontarget_changed = sorted(k for k in set(nontarget_before) | set(nontarget_after)
|
||
if nontarget_before.get(k) != nontarget_after.get(k))
|
||
assertions = {
|
||
"a1_changeApplied": {"pass": changed, "reason": "" if changed else "目标文件未变(no-op)"},
|
||
"a2_nontargetStable": {"pass": untouched_stable, "nontargetChanged": nontarget_changed},
|
||
"a3_lineageQueryable": cheap_assert.assert_lineage(base_version_id),
|
||
"allPass": bool(changed and untouched_stable and base_version_id is not None),
|
||
}
|
||
a12_pass = changed and untouched_stable
|
||
status = "succeeded" if (gates_pass and a12_pass) else "failed"
|
||
stage = run_summary.get("stage") or "code"
|
||
log(f"模块重生成完成 game_id={game_id} status={status} changed={changed} "
|
||
f"untouchedStable={untouched_stable} gatesPass={gates_pass} stage={stage}")
|
||
return {"status": status, "stage": stage, "manifest": manifest,
|
||
"verdict": verdict_full, "summary": run_summary, "assertions": assertions,
|
||
"error": None if status == "succeeded" else _regen_why(gates_pass, untouched_stable, changed)}
|