diff --git a/cheap-worker/cheap_gates.py b/cheap-worker/cheap_gates.py new file mode 100644 index 00000000..220b0faa --- /dev/null +++ b/cheap-worker/cheap_gates.py @@ -0,0 +1,27 @@ +"""cheap_gates.py — 便宜档九门收口的同步包装(供 Service 续修 check 闭包经 to_thread 调用)。 + +归并后便宜档收敛判据从「check+build 绿」升为「九门绿」:RepairMiddleware 的 check 闭包在 finish 点 +把工程收口跑一遍九门(stage→smoke→ensure_play_spec→play),经 worker.gate_judge.judge_cheap_verdict 归一。 +本模块只做「同步 pipeline + 端口透传」,不做判据归一(那在 gate_judge,与 tier2 并列)。 +端口由调用方按 session 派生传入(决策③ 并发避撞);shell-out/subprocess 阻塞,故封在同步函数里给 to_thread 跑。 +""" + +import cheap_run + + +def run_cheap_gates(game_id: str, port: int, cdp_port: int) -> dict: + """同步跑便宜档九门收口 stage→smoke→ensure_play_spec→play,返回九门 verdict({pass, guards})或 {}(未跑出)。 + + 任一段失败(stage 打包不出 / play 没产 verdict)→ 返 {}(judge_cheap_verdict 判未过、feedback 落 harness 说明, + 不伪造门绿)。端口按 session 派生传入,避免并发多局撞固定 4320/9222(决策③)。 + """ + st = cheap_run.stage(game_id) + if not st["ok"]: + print(f"[cheap-gates] game={game_id} stage 失败(判未过):{st['output'][:300]}", flush=True) + return {} + sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port) + # ensure_play_spec 据 smoke 抓的 state 产 driver(已存在不覆盖),让九门 driven=true、E_live/H_progress 由 advisory 升致命, + # 根治「裸 harness 无 driver → 假绿」。smoke 失败(state=None)时按保守 key-cycle 薄 spec,play 仍照跑。 + cheap_run.ensure_play_spec(game_id, sm.get("state")) + pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port) + return pr.get("verdict") or {} diff --git a/cheap-worker/tests/test_cheap_gates.py b/cheap-worker/tests/test_cheap_gates.py new file mode 100644 index 00000000..f34b1fa3 --- /dev/null +++ b/cheap-worker/tests/test_cheap_gates.py @@ -0,0 +1,33 @@ +"""cheap_gates.run_cheap_gates 单测:同步跑九门收口的端口透传 + 逐段桩(monkeypatch cheap_run,零真 chrome/esbuild)。 +跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_gates.py -v +""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/ +import _bootstrap # noqa: E402,F401 仅加 sys.path(import 时不取 key、不触网) +import cheap_gates # noqa: E402 +import cheap_run # noqa: E402 + + +def test_run_cheap_gates_threads_ports_and_returns_verdict(monkeypatch): + seen = {} + monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": True, "output": ""}) + monkeypatch.setattr(cheap_run, "smoke", + lambda gid, port, cdp_port: seen.update(smoke=(port, cdp_port)) or {"ok": True, "state": {"targets": []}}) + monkeypatch.setattr(cheap_run, "ensure_play_spec", lambda gid, state: {"ok": True, "wrote": True, "driverType": "tap-targets"}) + monkeypatch.setattr(cheap_run, "play", + lambda gid, port, cdp_port: seen.update(play=(port, cdp_port)) or {"ok": True, "verdict": {"pass": True, "guards": {}}}) + out = cheap_gates.run_cheap_gates("g1", 4407, 9407) + assert out == {"pass": True, "guards": {}} + assert seen["smoke"] == (4407, 9407), "smoke 应收到派生端口" + assert seen["play"] == (4407, 9407), "play 应收到派生端口" + + +def test_run_cheap_gates_stage_fail_returns_empty(monkeypatch): + monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": False, "output": "esbuild bundle 缺失"}) + called = {"smoke": 0} + monkeypatch.setattr(cheap_run, "smoke", lambda *a, **k: called.update(smoke=called["smoke"] + 1) or {"ok": True}) + out = cheap_gates.run_cheap_gates("g1", 4407, 9407) + assert out == {}, "stage 失败应短路返 {}(judge 判未过)" + assert called["smoke"] == 0, "stage 失败不应再跑 smoke" diff --git a/tier2/gen-worker/tests/test_gate_judge_cheap.py b/tier2/gen-worker/tests/test_gate_judge_cheap.py new file mode 100644 index 00000000..f7de2aa7 --- /dev/null +++ b/tier2/gen-worker/tests/test_gate_judge_cheap.py @@ -0,0 +1,50 @@ +"""gate_judge.judge_cheap_verdict 单测:便宜档九门 verdict 归一成 GateJudgment(绿 / 红列门+feedback / 缺失 三态)。 +跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge_cheap.py -v +""" +from worker.gate_judge import GateJudgment, judge_cheap_verdict + + +def test_cheap_green_passes(): + # 顶层 pass=True 且各门 pass=True → passed=True、无失败门。 + v = {"pass": True, "guards": {"A_boot": {"pass": True}, "H_progress": {"pass": True}}} + j = judge_cheap_verdict(v) + assert isinstance(j, GateJudgment) + assert j.passed is True + assert j.failed_gates == [] + + +def test_cheap_pass_but_empty_guards_not_passed(): + # I3 防假绿:顶层 pass=True 但 guards 空(play 没真跑出逐门,或被伪造 pass)→ passed=False。 + # 放行判据 = 顶层 pass **且** guards 非空;单看 pass 会把「空 verdict 强标 pass」当绿。 + for v in ({"pass": True, "guards": {}}, {"pass": True}): + j = judge_cheap_verdict(v) + assert j.passed is False, "pass=True 但无逐门 guards 不放行" + assert j.failed_gates == [] + assert j.feedback and "verdict" in j.feedback # 落 harness 说明(未跑出逐门) + + +def test_cheap_red_lists_failed_gates_and_feedback(): + # 顶层 pass=False + 有未过门 → passed=False、失败门入列、feedback 非空且含门中文提示与 detail。 + v = {"pass": False, "guards": { + "A_boot": {"pass": True}, + "F_wiring": {"pass": False, "callCount": 0, "expected": ["sessionScore."], "sample": []}, + "H_progress": {"pass": False, "checks": [ + {"path": "score", "op": "increased", "before": 0, "after": 0, "pass": False, "why": "真玩应加分"}], + "latch": {"pass": True}}, + }} + j = judge_cheap_verdict(v) + assert j.passed is False + assert "F_wiring" in j.failed_gates and "H_progress" in j.failed_gates + assert "A_boot" not in j.failed_gates + assert j.feedback and isinstance(j.feedback, str) + assert "F_wiring" in j.feedback and "H_progress" in j.feedback + assert "score" in j.feedback # H_progress detail 抽到未过 check 的 path + + +def test_cheap_missing_verdict_is_not_passed(): + # play 没跑出 verdict(None / 无 guards)→ passed=False、失败门空、feedback 落 harness 说明。 + for v in (None, {}, {"pass": False}): + j = judge_cheap_verdict(v) + assert j.passed is False + assert j.failed_gates == [] + assert "verdict" in j.feedback # 「九门真玩没有跑出 verdict…」 diff --git a/tier2/gen-worker/worker/gate_judge.py b/tier2/gen-worker/worker/gate_judge.py index edddb599..7001ae47 100644 --- a/tier2/gen-worker/worker/gate_judge.py +++ b/tier2/gen-worker/worker/gate_judge.py @@ -53,3 +53,91 @@ def judge_tier2_verdict(gate_result: dict) -> GateJudgment: failed_gates.append(f"finding:{_f.get('severity')}") feedback = verdict_feedback(v, gate_result.get("log") or "") return GateJudgment(passed=passed, failed_gates=failed_gates, feedback=feedback) + + +# ── 便宜档九门中文提示(门名 → 未过时给 agent 的修复方向;各门 detail 由 _cheap_gate_detail 补)── +# 便宜档九门与 tier2 是同一套引擎无关门(serve-and-play.sh),门名散见 verdict.json guards: +# A_boot / B_uncaught / C_frame / D_render / E_live / F_wiring / G_input / H_progress / I_control。 +_CHEAP_GATE_HINTS = { + "A_boot": "游戏没能正常启动(A_boot):检查 index.html / bundle.iife.js 能否加载、__GameBundle 是否正确导出。", + "B_uncaught": "运行时抛了未捕获异常(B_uncaught):按下方 uncaught 信息定位并修掉抛错的代码。", + "C_frame": "引擎主循环没有推进(C_frame,帧未变化):确认每帧 update/render 被调用、有真实状态变更。", + "D_render": "画面没有有效渲染(D_render,过暗或单一色):确认每帧真的画了东西(精灵/文字/形状)。", + "E_live": "运行态不鲜活(E_live,状态无区分度):画面/状态要随时间与输入真实变化,不能是静止定帧。", + "F_wiring": "引擎能力没接线(F_wiring,期望的引擎调用没发生):按 play-spec 的 expectedEngineCallPrefixes," + "用 plugins.<能力> 真正调用得分/音频/juice 等能力。", + "G_input": "输入没有驱动游戏(G_input,控制无响应):确认按键/点击真正改变了游戏状态。", + "H_progress": "没有真实进展(H_progress):真玩后关键指标未按预期变化,或没到达终局 latch;" + "核对得分/进度随真玩上升、终局态可达。", + "I_control": "控制门未通过(I_control)。", +} + + +def _cheap_gate_detail(name: str, g: dict) -> str: + """据门名从 guard detail 抽一句人读补充(各门字段不同,见 play.cdp.cjs 产出;缺字段返空,绝不抛)。""" + try: + if name == "B_uncaught": + u = g.get("uncaught") + return f" 未捕获:{str(u)[:200]}" if u else "" + if name == "F_wiring": + return (f" 期望前缀={g.get('expected')} 实际 callCount={g.get('callCount')}" + f" 样例={str(g.get('sample'))[:120]}") + if name == "E_live": + return f" 去重态数={g.get('distinctStates')}" + if name == "C_frame": + return f" f0={g.get('f0')} f1={g.get('f1')} delta={g.get('delta')}" + if name == "D_render": + return f" 亮度={g.get('bright')} 最大通道={g.get('maxCh')}" + if name == "H_progress": + parts = [] + for c in (g.get("checks") or []): + if isinstance(c, dict) and c.get("pass") is False: + parts.append(f"{c.get('path')} {c.get('op')}(before={c.get('before')} " + f"after={c.get('after')};{(c.get('why') or '')[:80]})") + latch = g.get("latch") or {} + if isinstance(latch, dict) and latch.get("pass") is False: + parts.append(f"终局 latch 未达(after={latch.get('after')} {(latch.get('reason') or '')[:80]})") + return (" " + "; ".join(parts)) if parts else "" + except Exception: # noqa: BLE001 —— 抽 detail 失败不连累 feedback 主体 + return "" + return "" + + +def _cheap_verdict_feedback(v: dict) -> str: + """把便宜档九门 verdict 的未过门拼成给 agent 的中文续修回喂。 + + tier2 的 verdict_feedback 吃 layerResults 形状,cheap verdict 无 layerResults(只有 guards map),故新写。 + verdict 缺失 / 无 guards(play 没跑出 verdict)→ 落一句 harness 说明(不伪造门绿)。 + """ + v = v or {} + guards = v.get("guards") or {} + if not guards: + return ("九门真玩没有跑出 verdict(可能 stage/build/smoke 失败或工程跑不起来):" + "请确认 check 与 build 都绿、游戏能被引擎启动后再交付。") + lines = [] + for name, g in guards.items(): + if isinstance(g, dict) and g.get("pass") is False: + hint = _CHEAP_GATE_HINTS.get(name, f"{name} 门未通过。") + lines.append("· " + hint + _cheap_gate_detail(name, g)) + if not lines: + return "九门整体未放行,但未定位到具体未过门;请检查 check/build 与真玩可运行性。" + return "以下真玩验收门未通过,请据每条修复后再交付,不要直接结束:\n" + "\n".join(lines) + + +def judge_cheap_verdict(play_result: dict) -> GateJudgment: + """把便宜档 cheap_run.play 的九门 verdict({pass, guards{门:{pass,...}}})归一成 GateJudgment(与 judge_tier2_verdict 并列)。 + + 放行判据 = 顶层 pass **且 guards 非空**(便宜档 verdict.pass 已是九门 AND 后的总放行,等价 gen.mjs done 门;判据不放松)。 + **I3 防假绿**:只看 `bool(v.get("pass"))` 会把「play 没跑出逐门却被强标 pass」当绿——故要求 `guards` 非空 + (九门真跑必产逐门 map);空 guards + pass=True → passed=False、feedback 落 harness 说明。 + failed_gates 复用 _verdict_brief 口径(guards 里 pass is False 的门名;H_progress 嵌套已在 g.pass 层判)。 + feedback 由 _cheap_verdict_feedback 从各门 detail 拼中文(tier2 的 verdict_feedback 吃 layerResults,cheap 无,不复用)。 + play_result 为 None / 空 / 无 guards(play 没产出 verdict)→ passed=False、失败门空、feedback 落 harness 说明。 + """ + v = play_result or {} + guards = v.get("guards") or {} + # 放行必须「顶层 pass 且有逐门 guards」:空 guards 无从证明九门真跑过,不放行(防空 verdict 假绿)。 + passed = bool(v.get("pass")) and bool(guards) + failed_gates = [k for k, g in guards.items() if isinstance(g, dict) and g.get("pass") is False] + feedback = _cheap_verdict_feedback(v) + return GateJudgment(passed=passed, failed_gates=failed_gates, feedback=feedback)