feat(cheap): 便宜档九门判据 judge_cheap_verdict + run_cheap_gates 同步门跑(端口透传) (切片一 阶段一②/T1)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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 {}

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"

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…」

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)