feat(studio): A11 切片三 M3 确定性类执行(零 LLM 改规范工程一处→重建→九门)

两段式第二段的 deterministic 类:用户确认后的改动落到规范工程那一处,零 LLM。

cheap_modify.py:
- apply_deterministic_modify:config/level→core.js 等 L3 的 export const 值替换;
  asset→assets.js IMAGE_FILES 键值或 baseUrl;no-op 检测(无 found&&old≠new→ok=false,M5 断言①地基)
- materialize_source_project:系统级落盘全 files(含 L1,绕 write_file 的 L1 守卫)
- execute_deterministic_modify:apply→(缺 plumbing 才 scaffold)→materialize→cheap_run build/stage/smoke/play 九门

worker_service.py:
- process_job 最前 modify 分流(_process_modify_job);deterministic 走执行 + manifest 进 trace.modifyManifest
  + best-effort 带新版 sourceProject(链式改);regenerate-module 显式 failed 留 M4;create/生成路一字未动

测试:test_a11_m3_deterministic 18/18 + 回归全绿。
九门 smoke(独立复跑,本机 Chrome 免 LLM):amgen-bake-click-score-0 base 源 → ROUND_MS 30000→19000
→ status=succeeded、verdictPass=true、manifest found=true、落盘 core.js=19000、运行时 remainMs=19000
(改的那一处真在游戏里生效并过九门)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili 2026-06-29 01:49:10 -07:00
parent 938219d7a5
commit 51230f1347
4 changed files with 724 additions and 3 deletions

View File

@ -0,0 +1,326 @@
"""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 ensure_play_spec 九门 play,**复用 cheap_run 现有九门,不另造**
apply no-op/unsupported 直接 failed( manifest + error),不重建( esbuild/Chrome)
best-effort 铁律:apply 失败 / 重建失败 failed + 诊断,绝不抛崩 worker只改 L3(core/assets/game-logic),
不碰 L1 plumbing
"""
import json
import re
from pathlib import Path
import cheap_run
# 调数值(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 且实际值变化(oldnew) 整体 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,
*, 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 ensure_play_spec 九门 play
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}
"""
# 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 → ensure_play_spec → 循环外九门 play(复用 cheap_studio 收口同一套)。
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)
cheap_run.ensure_play_spec(game_id, sm.get("state")) # 已存在不覆盖;缺则据 state 形态补 driver(防裸跑假绿)
pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port)
verdict = pr.get("verdict")
gates_pass = isinstance(verdict, dict) and verdict.get("pass") is True
status = "succeeded" if gates_pass else "failed"
log(f"确定性改重建+九门完成 game_id={game_id} status={status} smokeOk={sm.get('ok')} playOk={pr.get('ok')}")
return {"status": status, "manifest": manifest, "verdict": verdict,
"stage": "play", "error": None if status == "succeeded" else "九门未通过",
"smokeOk": sm.get("ok"), "playOk": pr.get("ok")}

View File

@ -0,0 +1,295 @@
"""test_a11_m3_deterministic.py — A11 切片三 M3「确定性类纠错执行」单测(execution 段零 LLM)。
守的不变量:
· apply_deterministic_modify(纯函数):config core.js 一处常量 / asset assets.js 一处资产键 /
多常量一次改 / 字符串·数组值 / 多行常量(Object.freeze)/ **no-op 检测**(目标找不到 found=falseok=false,
断言①地基)/ 坏输入容错不抛
· materialize_source_project:base 源全 files( L1 plumbing)系统级落盘到 game_dir( write_file L1 守卫)
· worker 接线:modifyMode=deterministic 走确定性执行(manifest 附进 trace成功带新版 sourceProject);
regenerate-module 显式 failed(不静默吞);create ( modifyMode)零副作用不调 modify_fn
:cheap-worker/.venv/bin/python cheap-worker/tests/test_a11_m3_deterministic.py
"""
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
import cheap_run # noqa: E402
import cheap_modify as M # noqa: E402
import worker_service as W # noqa: E402
import dedup # noqa: E402
# 把 game_dir 重定向到临时目录:materialize / worker 接线测试落盘免污染真 games/(与 cheap_modify、worker_service
# 同引用一个 cheap_run 模块对象 → patch 模块属性即对二者生效)。
_TMP_GAMES = Path(tempfile.mkdtemp(prefix="a11-m3-games-"))
cheap_run.game_dir = lambda gid: _TMP_GAMES / f"amgen-{gid}" # noqa: E731
# create 路会调 D9 dedup.check_similarity(写 FS 登记表)→ 重定向防污染真登记表(与 test_worker_service 同范式)。
dedup.DEDUP_REGISTRY = Path(tempfile.mkdtemp(prefix="a11m3-dedup-")) / "_dedup_registry.jsonl"
_GOOD_BUNDLE = "var __GameBundle=(function(){return{bootGameHost(){}}})();"
# ───────────────────────── ① apply_deterministic_modify ─────────────────────────
def test_config_change_round_ms():
"""config:ROUND_MS 30000→20000,core.js 那一处变、别处不动、manifest 命中、ok=true。"""
sp = {"files": {"src/core.js": "// core\nexport const ROUND_MS = 30000;\nexport const PAD = 4;\n"}}
patch = {"target": {"kind": "config", "path": "core.js#ROUND_MS"}, "payload": {"value": {"ROUND_MS": 20000}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is True, out
assert "export const ROUND_MS = 20000;" in out["files"]["src/core.js"]
assert "export const PAD = 4;" in out["files"]["src/core.js"], "别的常量不应被动"
rec = out["manifest"][0]
assert rec["found"] is True and rec["key"] == "ROUND_MS"
assert rec["old"] == "30000" and rec["new"] == "20000"
assert rec["file"].endswith("core.js") and rec["kind"] == "config"
def test_asset_change_image_file():
"""asset:assets.js IMAGE_FILES 的 heroSprite 换文件名,manifest 命中、别的资产不动。"""
sp = {"files": {"src/assets.js": "export const IMAGE_FILES = { heroSprite: 'hero.png', bg: 'bg.jpg' };\n"}}
patch = {"target": {"kind": "asset", "path": "assets.js#heroSprite"},
"payload": {"value": {"heroSprite": "hero-green.png"}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is True, out
assert "hero-green.png" in out["files"]["src/assets.js"]
assert "bg.jpg" in out["files"]["src/assets.js"], "别的资产键不应被动"
rec = out["manifest"][0]
assert rec["found"] is True and rec["old"] == "hero.png" and rec["new"] == "hero-green.png"
assert rec["kind"] == "asset"
def test_asset_base_url_change():
"""asset:baseUrl 特例 → 换 loadAssets 默认资产根。"""
sp = {"files": {"src/assets.js": "const root = baseUrl == null ? './assets/' : baseUrl;\n"}}
patch = {"target": {"kind": "asset"}, "payload": {"value": {"baseUrl": "https://cdn.x/g/"}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is True, out
assert "https://cdn.x/g/" in out["files"]["src/assets.js"]
assert out["manifest"][0]["old"] == "./assets/"
def test_noop_const_not_found():
"""no-op(断言①地基):目标常量不存在 → found=false、整体 ok=false、带 error。"""
sp = {"files": {"src/core.js": "export const ROUND_MS = 30000;\n"}}
patch = {"target": {"kind": "config"}, "payload": {"value": {"NOT_THERE": 5}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is False, "no-op 不应算真改动"
assert out["manifest"][0]["found"] is False
assert out["error"]
# 文本未被改动。
assert out["files"]["src/core.js"] == "export const ROUND_MS = 30000;\n"
def test_noop_same_value():
"""no-op:新值与旧值相同(常量命中但无变化)→ ok=false。"""
sp = {"files": {"src/core.js": "export const SPEED = 5;\n"}}
patch = {"target": {"kind": "config"}, "payload": {"value": {"SPEED": 5}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["manifest"][0]["found"] is True, "常量命中"
assert out["ok"] is False, "命中但值未变 = no-op"
def test_string_array_and_multi_const():
"""字符串值 / 数组值 / 多常量一次改 —— 都支持,manifest 三条全命中。"""
sp = {"files": {"src/core.js": (
"export const SPEED = 5;\n"
"export const LABEL = 'old';\n"
"export const KEYS = ['a','b'];\n"
)}}
patch = {"target": {"kind": "config"},
"payload": {"value": {"SPEED": 8, "LABEL": "new", "KEYS": ["x", "y", "z"]}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is True, out
txt = out["files"]["src/core.js"]
assert "export const SPEED = 8;" in txt
assert 'export const LABEL = "new";' in txt, "字符串值 → JS 字符串字面量(json 双引号合法)"
assert 'export const KEYS = ["x", "y", "z"];' in txt, "数组值 → JS 数组字面量"
assert len(out["manifest"]) == 3 and all(r["found"] for r in out["manifest"])
def test_config_searches_other_l3_when_not_in_core():
"""config 常量不在 core.js 时,按优先序搜其它 L3(game-logic.js)并命中。"""
sp = {"files": {
"src/core.js": "export const A = 1;\n",
"src/game-logic.js": "export const SPAWN_RATE = 2;\n",
}}
patch = {"target": {"kind": "config"}, "payload": {"value": {"SPAWN_RATE": 9}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is True, out
assert "export const SPAWN_RATE = 9;" in out["files"]["src/game-logic.js"]
assert out["manifest"][0]["file"].endswith("game-logic.js")
def test_multiline_const_replace():
"""多行常量(Object.freeze 跨行)→ 多行兜底正则替换,且不吞掉其后的常量。"""
sp = {"files": {"src/core.js": (
"export const PALETTE = Object.freeze([\n"
" { fill: '#fff', score: 1 },\n"
" { fill: '#000', score: 2 },\n"
"]);\n"
"export const NEXT = 7;\n"
)}}
patch = {"target": {"kind": "config"}, "payload": {"value": {"PALETTE": [1, 2, 3]}}}
out = M.apply_deterministic_modify(sp, patch)
assert out["ok"] is True, out
txt = out["files"]["src/core.js"]
assert "export const PALETTE = [1, 2, 3];" in txt
assert "export const NEXT = 7;" in txt, "多行替换必须锚到本常量行尾分号,不能吞掉后续常量"
def test_level_complex_layout_unsupported():
"""level:复杂布局重排(value 非 {常量名:新值})→ unsupported、ok=false。"""
sp = {"files": {"src/core.js": "export const X = 1;\n"}}
out = M.apply_deterministic_modify(sp, {"target": {"kind": "level"}, "payload": {"layout": [[1, 0], [0, 1]]}})
assert out["ok"] is False and "level" in (out["error"] or "")
def test_level_const_text_replace_supported():
"""level:关卡常量文本替换(payload.value={常量名:新值})与 config 同路,支持。"""
sp = {"files": {"src/core.js": "export const LEVEL_ROWS = 8;\n"}}
out = M.apply_deterministic_modify(sp, {"target": {"kind": "level"}, "payload": {"value": {"LEVEL_ROWS": 12}}})
assert out["ok"] is True
assert "export const LEVEL_ROWS = 12;" in out["files"]["src/core.js"]
assert out["manifest"][0]["kind"] == "level"
def test_tolerant_bad_inputs():
"""坏 source_project / 坏 patch / 非确定性 kind —— 容错返 ok=false,绝不抛。"""
assert M.apply_deterministic_modify("{not json", {"target": {"kind": "config"},
"payload": {"value": {"X": 1}}})["ok"] is False
assert M.apply_deterministic_modify(None, {})["ok"] is False
assert M.apply_deterministic_modify({"files": {"src/core.js": "export const X = 1;\n"}}, "{bad")["ok"] is False
assert M.apply_deterministic_modify({"files": {"src/core.js": "x"}}, None)["ok"] is False
out = M.apply_deterministic_modify({"files": {"src/core.js": "x"}},
{"target": {"kind": "behavior"}, "payload": {"value": {"X": 1}}})
assert out["ok"] is False and "behavior" in (out["error"] or "")
def test_apply_does_not_mutate_input():
"""apply 不原地改入参 files(返回的是改后拷贝)。"""
sp = {"files": {"src/core.js": "export const A = 1;\n"}}
M.apply_deterministic_modify(sp, {"target": {"kind": "config"}, "payload": {"value": {"A": 9}}})
assert sp["files"]["src/core.js"] == "export const A = 1;\n", "入参 files 不应被原地改动"
# ───────────────────────── ② materialize_source_project ─────────────────────────
def test_materialize_writes_all_files_incl_l1():
"""全 files(含 L1 plumbing host-config/game/index/entry)落盘到 game_dir(系统落盘绕 L1 守卫)。"""
files = {
"src/core.js": "export const A = 1;\n",
"src/host-config.js": "// L1 plumbing host-config\n",
"src/game.js": "// L1 wrapper game\n",
"index.html": "<html><body>g</body></html>",
"entry-bundle.js": "// L1 entry\n",
}
gd = M.materialize_source_project("mat1", files)
assert (gd / "src" / "core.js").read_text(encoding="utf-8") == "export const A = 1;\n"
assert (gd / "src" / "host-config.js").exists(), "L1 host-config 也应落盘(系统落盘不拦 L1)"
assert (gd / "src" / "game.js").exists(), "L1 game.js 也应落盘"
assert (gd / "index.html").read_text(encoding="utf-8") == "<html><body>g</body></html>"
assert (gd / "entry-bundle.js").exists(), "L1 entry-bundle 也应落盘"
def test_materialize_skips_escape_path():
"""越界路径(../ 逃逸)被跳过、不写到 game_dir 外,且不抛。"""
gd = M.materialize_source_project("mat2", {"../escape.js": "x", "src/ok.js": "y"})
assert (gd / "src" / "ok.js").exists()
assert not (gd.parent / "escape.js").exists(), "越界路径必须被跳过"
# ───────────────────────── ④ worker 接线 ─────────────────────────
def test_worker_deterministic_route_attaches_manifest_on_failed():
"""modifyMode=deterministic → 走确定性执行;失败(no-op)路也把 manifest/error 附进 trace 回调。"""
captured = {}
fake = {"status": "failed",
"manifest": [{"file": "src/core.js", "kind": "config", "key": "X", "old": None, "new": "5", "found": False}],
"verdict": None, "error": "no-op:常量未找到", "stage": "apply"}
state = W.WorkerState(modify_fn=lambda gid, sp, patch: fake,
send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""),
profile_fn=lambda j, g: None)
job = {"job_id": "m1", "traceId": "m1", "modifyMode": "deterministic", "gameId": "nope-x",
"sourceProject": "{}", "modifyPatch": "{}", "brief": "x", "callback": {"target": "http://cb"}}
W.process_job(state, job)
p = captured["payload"]
assert p["status"] == "failed", "无 bundle + verdict 失败 → failed"
assert p["trace"]["modifyManifest"][0]["key"] == "X"
assert p["trace"]["modifyError"] == "no-op:常量未找到"
def test_worker_deterministic_route_success_carries_bundle_and_source():
"""deterministic 成功 → result-out 带 engineBundle(重建产物)+ manifest + 新版 sourceProject(链式改)。"""
gid = "m3ok"
gd = cheap_run.game_dir(gid)
(gd / "src").mkdir(parents=True, exist_ok=True)
(gd / "bundle.iife.js").write_text(_GOOD_BUNDLE, encoding="utf-8")
(gd / "src" / "core.js").write_text("export const A = 1;\n", encoding="utf-8")
captured = {}
fake = {"status": "succeeded",
"manifest": [{"file": "src/core.js", "kind": "config", "key": "ROUND_MS",
"old": "30000", "new": "20000", "found": True}],
"verdict": {"pass": True, "guards": {}}, "error": None, "stage": "play"}
profile = {"tickModel": "realtime", "inputModel": "discrete-choice", "progressModel": "metric"}
state = W.WorkerState(modify_fn=lambda *a: fake,
send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""),
profile_fn=lambda j, g: profile)
job = {"job_id": "m2", "traceId": "m2", "modifyMode": "deterministic", "gameId": gid,
"sourceProject": "{}", "modifyPatch": "{}", "brief": "点点乐", "callback": {"target": "http://cb"}}
W.process_job(state, job)
p = captured["payload"]
assert p["status"] == "succeeded"
assert "__GameBundle" in p["engineBundle"], "成功路携重建后的 bundle"
assert p["trace"]["modifyManifest"][0]["key"] == "ROUND_MS"
assert "sourceProject" in p, "成功路 best-effort 带新版源(供下一次链式改)"
def test_worker_regenerate_module_explicit_failed():
"""modifyMode=regenerate-module(M4 未实现)→ 显式 failed + 发 failed 回调,绝不静默吞。"""
captured = {}
state = W.WorkerState(send_fn=lambda u, p, s: captured.update(payload=p) or (200, ""))
job = {"job_id": "rm1", "traceId": "rm1", "modifyMode": "regenerate-module", "gameId": "g",
"brief": "x", "callback": {"target": "http://cb"}}
out = W.process_job(state, job)
assert out["status"] == "failed" and out["mode"] == "regenerate-module"
assert captured["payload"]["status"] == "failed", "应发 failed 回调(任务不挂 RUNNING)"
def _boom_modify(*a):
raise AssertionError("create 路不应调用 modify_fn")
def test_create_path_unaffected_by_modify_wiring():
"""无 modifyMode 的 create 路:走 run_fn、不触 modify_fn(modify 接线对 create 零副作用)。"""
gid = "create1"
gd = cheap_run.game_dir(gid)
(gd / "src").mkdir(parents=True, exist_ok=True)
(gd / "bundle.iife.js").write_text(_GOOD_BUNDLE, encoding="utf-8")
called = {}
state = W.WorkerState(
run_fn=lambda job: (called.update(ran=True) or {"verdict": {"pass": True}}, gd),
send_fn=lambda u, p, s: called.update(payload=p) or (200, ""),
profile_fn=lambda j, g: None,
modify_fn=_boom_modify,
)
job = {"job_id": "c1", "traceId": "c1", "brief": "x", "callback": {"target": "http://cb"}} # 无 modifyMode
W.process_job(state, job)
assert called.get("ran") is True, "create 路必须走 run_fn"
assert called["payload"]["status"] == "succeeded"
if __name__ == "__main__":
_fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
_failed = 0
for _fn in _fns:
try:
_fn()
print(f" PASS {_fn.__name__}")
except Exception as e: # noqa: BLE001
_failed += 1
print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}")
print(f"\n[test_a11_m3_deterministic] {len(_fns) - _failed}/{len(_fns)} passed")
sys.exit(1 if _failed else 0)

View File

@ -149,13 +149,16 @@ class WorkerState:
"""进程级状态:配置 + 有界队列 + 可注入依赖 + job_id 去重。"""
def __init__(self, *, callback_secret: str = "", queue_maxsize: int = DEFAULT_QUEUE_MAXSIZE,
run_fn=None, send_fn=None, profile_fn=None, classify_fn=None):
run_fn=None, send_fn=None, profile_fn=None, classify_fn=None, modify_fn=None):
self.callback_secret = callback_secret
self.queue: queue.Queue = queue.Queue(maxsize=queue_maxsize)
self.run_fn = run_fn or _default_run_fn # job -> (summary, game_dir)
self.send_fn = send_fn or post_callback # (url, payload, secret) -> (status, body)
self.profile_fn = profile_fn or derive_profile # (job, game_dir) -> profile | None
self.classify_fn = classify_fn or cheap_classify.classify # (rawText, sourceProject) -> 建议改动(A11 M2 判意图)
# A11 切片三 M3:确定性类执行器(game_id, sourceProject, modifyPatch) -> 执行结果。
# None=懒绑 cheap_modify.execute_deterministic_modify(单测注入桩免真 esbuild/Chrome)。
self.modify_fn = modify_fn
self._seen: set = set() # job_id 去重(防同 job 重投重跑)
self._lock = threading.Lock()
@ -176,8 +179,19 @@ def try_enqueue(state: WorkerState, job: dict) -> tuple[bool, int]:
def process_job(state: WorkerState, job: dict) -> dict:
"""处理一个 job:run_fn 生成 → 组 §6.1 result-out → 向 callback.target 发(带 HMAC)。返回 payload。"""
"""处理一个 job:run_fn 生成 → 组 §6.1 result-out → 向 callback.target 发(带 HMAC)。返回 payload。
A11 切片三: modifyMode job 走调整路(_process_modify_job),**create/生成路一字不动**
modify 路在最前分流,下方生成路对无 modifyMode job 零副作用
"""
trace_id = job.get("traceId") or job.get("job_id")
# ── A11 调整路分流(modify):有 modifyMode → deterministic 确定性执行 / regenerate-module 留 M4。──
modify = parse_modify(job)
if modify is not None:
return _process_modify_job(state, job, modify)
# ↓↓↓ 以下 create/生成路保持原样、零改动 ↓↓↓
summary, game_dir = state.run_fn(job)
# ── M3b U3:D9 反同质化(vendored dedup)。撞重只告警不阻断(status 仍按九门判);
@ -209,6 +223,84 @@ def process_job(state: WorkerState, job: dict) -> dict:
return payload
# ---------- A11 切片三:调整路(modify)处理 ----------
def _modify_summary(result: dict, game_id: str) -> dict:
"""把 execute_deterministic_modify 的执行结果转成 build_result_out 吃的 run-summary 形态。
确定性改一次成型( resume ) attempts=1;verdict.pass 取执行结果 status;verdictFull 透传九门 verdict
trace.sevenGateVerdict(后端 D11)manifest/error summary 带出, _process_modify_job 附进 trace
"""
pass_ = result.get("status") == "succeeded"
verdict_full = result.get("verdict") if isinstance(result.get("verdict"), dict) else None
failed = []
if verdict_full:
guards = verdict_full.get("guards") or {}
failed = [k for k, g in guards.items() if isinstance(g, dict) and g.get("pass") is False]
return {
"ok": pass_, "gameId": game_id, "stage": result.get("stage", "play"), "attempts": 1,
"verdict": {"pass": pass_, "failedGates": failed},
"verdictFull": verdict_full,
}
def _process_modify_job(state: WorkerState, job: dict, modify: dict) -> dict:
"""A11 调整路:deterministic 走确定性执行(改→重建→九门);regenerate-module 留 M4;未知 mode → failed。
deterministic:execute_deterministic_modify(注入 modify_fn 可桩) result-out(manifest 附进 trace
best-effort 带新版 sourceProject 供链式改) 回调regenerate-module/未知:显式 failed + 日志(不静默吞)
"""
trace_id = job.get("traceId") or job.get("job_id")
mode = modify.get("mode")
game_id = job.get("gameId") or job.get("job_id")
if mode != "deterministic":
# regenerate-module 属 M4(模块重生成,要 LLM)、未知 mode 都显式 failed,绝不静默吞(错误路红线)。
why = "regenerate-module 属 M4、未实现" if mode == "regenerate-module" else f"未知 modifyMode={mode}"
log(f"调整路不支持的 mode → 发兜底 failed trace_id={trace_id}: {why}")
_send_failed(state, job, why)
return {"status": "failed", "mode": mode, "error": why}
# deterministic:懒绑执行器(单测注入桩免真 esbuild/Chrome)。
import cheap_run
modify_fn = state.modify_fn
if modify_fn is None:
import cheap_modify
modify_fn = cheap_modify.execute_deterministic_modify
result = modify_fn(game_id, modify.get("sourceProject"), modify.get("modifyPatch"))
summary = _modify_summary(result, game_id)
game_dir = cheap_run.game_dir(game_id)
# best-effort:重建成功则把新版源回传(供下一次链式改用 base);派生失败只记日志、不阻断。
source_project = None
if result.get("status") == "succeeded":
try:
profile = state.profile_fn(job, game_dir)
if profile:
source_project = result_out.build_source_project(game_dir, profile)
except Exception as e: # noqa: BLE001 — sourceProject 非承重,派生失败不阻断回调
log(f"调整路 sourceProject 派生失败(省略,不阻断)trace_id={trace_id}: {type(e).__name__}: {e}")
payload = result_out.build_result_out(job, summary, game_dir, source_project=source_project)
# 改动清单 manifest 附进 trace(后端 trace 是开放 Map、忽略未知键;此键供前端展示「改了哪几处」+ 排障)。
manifest = result.get("manifest")
if manifest is not None:
payload.setdefault("trace", {})["modifyManifest"] = manifest
if result.get("error"):
payload.setdefault("trace", {})["modifyError"] = result["error"]
callback_url = (job.get("callback") or {}).get("target")
if callback_url:
http_status, body = state.send_fn(callback_url, payload, state.callback_secret)
log(f"调整路回调完成 trace_id={trace_id}, status={payload['status']}, "
f"callbackHttp={http_status}, manifest={len(manifest) if manifest else 0} 处, body={str(body)[:160]}")
else:
log(f"调整路 job 无 callback.target,跳过回调 trace_id={trace_id}, status={payload['status']}")
return payload
def _send_failed(state: WorkerState, job: dict, reason: str) -> None:
"""生成异常兜底:发 failed result-out,避免任务挂 RUNNING(对执行器 callbackFailed 语义)。"""
trace_id = job.get("traceId") or job.get("job_id")

View File

@ -1,6 +1,6 @@
---
title: A11 对话式调整回路(试玩后纠错)细化执行 plan
status: 已批(创始人 2026-06-29· 执行中M1✅ M2✅ → M3
status: 已批(创始人 2026-06-29· 执行中M1✅ M2✅ M3✅ → M4
date: 2026-06-28
topic: studio-A11-调整回路
canonical: false # 切片三统领在 plan①本档是「切片三·A11」的可执行细化不另立 SoT评审后把 A11 设计面蒸馏回 docs/architecture 生成引擎域(兑现 plan① 收口 TODO⑥
@ -289,6 +289,14 @@ stateDiagram-v2
- **Java 侧studio `/modify/plan` → aigc `classifyModify` → worker `/classify` 同步对接)完成、绿**契约先行studio.yaml +`/modify/plan` + ReqVO/RespVO`AppStudioController.planModify``StudioServiceImpl.planModify``validateCreator` 白名单门[核实与既有 createDraft/generate/modify L99/139/313 一致] + `resolveBaseVersionGameWithOwnership` 归属校验防越权 + 委托 aigc + 兜底);`AigcApi.classifyModify``WorkerClassifyClient`(据 baseVersionId 反查 base 源 → NO_PROXY JDK HttpClient POST worker `/classify` → best-effort 兜底 unclear+needsConfirm软注入 `ObjectProvider<WorkerClassifyClient>`executor 关闭时 Bean 缺席→兜底、不致启动失败,镜像 WorkerDispatchClient。子代理本地 maven 43 测试全绿;**主会话独立复跑 WorkerClassifyClient/StudioServiceImpl/AigcApiImpl = 43/43 绿、BUILD SUCCESS**。
- **M2 收口**:两段式第一段全链路通——前端 → studio `/modify/plan` → aigc 反查 base 源 + 同步 worker `/classify` → 判意图建议 + 风险回前端needsConfirm=true危险/大改/改玩法/意图不明)待用户确认后调 `/modify`M3/M4 执行)。真连 worker 的判意图 e2e 待 M5mini-desktop。前端 Vue 确认 UI = 前端 lane消费 `/modify/plan`,本切片 backend 承接已就位、非孤儿。worker URL 派生 = `getWorkerUrl()` 尾段 `/generate``/classify`;判意图同步 HTTP 超时 30s 常量极端慢→兜底降级MVP 可接受)。
**M3 确定性类执行完成、绿2026-06-29· 零 LLM**
- **`cheap_modify.py`**`apply_deterministic_modify`改规范工程那一处——config/level→core.js 等 L3 的 `export const NAME=` 值替换 / asset→assets.js `IMAGE_FILES` 键值或 baseUrl**no-op 检测**=无任一条 `found 且 old≠new``ok=false`,是 M5 断言①地基);`materialize_source_project`(系统级落盘全 files 含 L1、绕 write_file 的 L1 守卫);`execute_deterministic_modify`apply→缺 plumbing 才 scaffold→materialize→`cheap_run` build/stage/smoke/play 九门,零另造)。
- **worker 接线**`process_job` 最前 modify 分流(`_process_modify_job`deterministic 走执行 + manifest 进 `trace.modifyManifest` + best-effort 带新版 sourceProject链式改`regenerate-module` 显式 failed 留 M4不静默吞**create/生成路一字未动**。
- **测试**`test_a11_m3_deterministic` 18/18 + 回归a11_m1 5/5、a11_m2_classify 9/9、worker_service 15/15全绿。
- **九门 smoke主会话独立复跑本机 Chrome 免 LLM**:取 `amgen-bake-click-score-0` base 源 → config 改 `ROUND_MS 30000→19000`**status=succeeded、stage=play、verdictPass=true、manifest found=true、落盘 core.js=19000、运行时 remainMs=19000**(改的那一处真在游戏里生效并过九门)。坐实"改规范工程一处→重建→过九门"全链路。
- **M3 收口**:确定性类执行端到端通(零 LLM改动清单为 M5 断言①就位。
## 状态
双评审已过(发现全处置)→ 待创始人批 → 执行M1→M5 顺序M1 为 plan① 所称"首个里程碑")。