feat(cheap): T2 续修反馈补厚·C6 判卷契约接线(latch 双支 phaseNow+driverType+game-log 摘要+CLI 九门回喂对齐+tier2 结构化 sidecar) (W-S1 修复三单·单③)

- gate_judge:反馈先构造 C6 结构化对象(build_cheap_feedback_obj)再渲染中文(单一来源,拼串丢字段结构上不可能);
  latch advisory 支(phaseNow)与驻留支(after)都读;每条逐门带 driverType(staged play-spec 取数);
  game-log.json 告警/报错摘要进反馈(截断保护);C6 对象落 evidence/verdict-feedback.json 证据留痕
- cheap_studio:CLI 收敛判据升「九门绿」——finish 后 _closeout_gates 跑门,未绿且轮数/预算有余带 C6 反馈
  resume 续修(与生产 RepairMiddleware finish 点拦截行为同型);软停后只收尾不回喂;收口复用最后一轮门结果;
  对照路 run_gates=False 零改动
- tier2 对齐(有界零行为变更):verdict_feedback 富文本逐字不动,新增 verdict_feedback_obj+write_feedback_sidecar
  结构化 sidecar 落 evidence/(judge_tier2_verdict 未过时 best-effort)
- Service _cheap_check / driver _build_summary 传 game_id+staged_dir 解锁完整反馈口径
- tests/test_feedback_c6.py:phaseNow 两支/负样本被 validate.py 拦/driverType/game-log 截断/sidecar/富文本不变 14 用例

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

cheap-worker/cheap_service_app.py

@@ -282,7 +282,10 @@ async def _cheap_middlewares_factory(user_id: str, agent_id: str, session_id: st
                                     feedback=f"九门执行异常,请确认工程可 check/build/运行后再交付:{type(e).__name__}: {e}")
         finally:
             release_ports(pair)
-        return judge_cheap_verdict(verdict)
+        # C6 接线(W-S1 单③):传 game_id + staged 目录,反馈带 phaseNow/driverType/game-log 摘要,
+        #   并把结构化反馈落 staged evidence/verdict-feedback.json(证据留痕,gate_judge 内 best-effort)。
+        return judge_cheap_verdict(verdict, game_id=game_id,
+                                   staged_dir=cheap_run.wg1_game_dir(game_id))
 
     repair = RepairMiddleware(
         check=_cheap_check,

cheap-worker/cheap_service_driver.py

@@ -143,10 +143,13 @@ def _build_summary(game_id: str, brief: str, verdict, svc: dict, turn: dict, t0:
     costRmb/rmbGate/repairs 来自 Service 收口采集(worker 进程拿不到 Service 内存态,靠 sidecar 跨进程)。
     attempts = repairs+1(result_out 的 trace.repairs = max(0, attempts-1) 会反推回 repairs)。
     """
+    import cheap_run  # noqa: PLC0415
     import cheap_studio  # noqa: PLC0415
     from worker.gate_judge import judge_cheap_verdict  # noqa: PLC0415
 
-    j = judge_cheap_verdict(verdict or {})
+    # C6 接线:带 game_id/staged 目录判(驱动器族/日志摘要口径与 Service 续修一致;此处只为 summary 归一,不再落 sidecar 也无妨——staged_dir 传入使口径同源)。
+    j = judge_cheap_verdict(verdict or {}, game_id=game_id,
+                            staged_dir=cheap_run.wg1_game_dir(game_id))
     vb = cheap_studio._verdict_brief(verdict) if verdict else None
     svc = svc or {}
     repairs = svc.get("repairs")

cheap-worker/cheap_studio.py

@@ -27,6 +27,7 @@ import cheap_verify  # noqa: E402  U-A2：便宜档「丰富度」LLM 验证 age
 
 # tier2 框架层（import config 触发代理旁路 + 加载 agentscope，必须在裸 import agentscope/openai 之前）。
 from worker import config  # noqa: E402
+from worker.gate_judge import judge_cheap_verdict  # noqa: E402  CLI 回喂:九门判据与生产 RepairMiddleware 同源
 from worker.middleware import (  # noqa: E402
     CircuitBreakerMiddleware, Tier2TraceMiddleware, Tier2CircuitBreak,
 )
@@ -70,6 +71,39 @@ def _verdict_brief(v) -> dict:
     return {"pass": v.get("pass"), "failedGates": failed}
 
 
+def _closeout_gates(game_id, *, port, cdp_port) -> dict:
+    """CLI 收口门流水线 stage → smoke → ensure_play_spec → 九门 play(一次跑齐;回喂循环与收口段共用)。
+
+    与 Service 路 cheap_gates.run_cheap_gates 同序,多回传 staged/smoke_ok/driver_type 供 run-summary 组装
+    (Service 路这些由 collector/driver 另采)。起 play 前清残留 verdict(红线③,镜像 cheap_gates.py:28):
+    serve-and-play.sh 早退/超时不写新 verdict 也不删旧的,不清则读回上一轮绿 verdict → 假绿。
+    """
+    out = {"staged": False, "smoke_ok": None, "driver_type": None, "verdict": None}
+    st = cheap_run.stage(game_id)
+    out["staged"] = st["ok"]
+    if not st["ok"]:
+        _rec(f"stage FAIL: {st['output'][:300]}")
+        return out
+    sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port)
+    out["smoke_ok"] = sm["ok"]
+    _rec(f"smoke {'PASS' if sm['ok'] else 'FAIL'}(抓 state 供 play-spec)")
+    ps = cheap_run.ensure_play_spec(game_id, sm.get("state"))
+    # driver 类型:本轮新产的直接带;复用旧 spec(hash 一致)时读盘补(spec 是判卷单一来源)。
+    dt = ps.get("driverType")
+    if not dt:
+        try:
+            spec = json.loads((cheap_run.wg1_game_dir(game_id) / "play-spec.json").read_text(encoding="utf-8"))
+            dt = (spec.get("driver") or {}).get("type")
+        except Exception:  # noqa: BLE001 —— 读不出只降级 None
+            dt = None
+    out["driver_type"] = dt
+    _rec(f"play-spec {'已产 driver=' + str(dt) if ps.get('wrote') else ps.get('reason', '?')}")
+    (cheap_run.wg1_game_dir(game_id) / "evidence" / "verdict.json").unlink(missing_ok=True)  # 红线③:清陈旧 verdict
+    pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port)
+    out["verdict"] = pr["verdict"]
+    return out
+
+
 def _furthest_stage(*, finished, staged, smoke_ran, played) -> str:
     """据收口流程实际到达的步，判本 run 走到的最远阶段（scaffold/code/stage/smoke/play）。
 
@@ -189,15 +223,40 @@ async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens=
         "和你的起点 game-logic.js 再动手；核心玩法实现完、check 与 build 都绿了就立即 finish。")
     breaker_tripped = None
     attempts = 0
+    closeout = None  # CLI 回喂:最近一次九门收口结果(回喂循环产出,收口段复用、不重复跑门)
     try:
         for attempt in range(max_resumes + 1):
             attempts = attempt + 1
             _rec(f"resume attempt {attempts}/{max_resumes + 1} → writer.reply …")
             await writer.reply(_user_msg(kick))
-            # ① 真 finish → 收敛退出
+            # ① 真 finish:CLI 回喂对齐(W-S1 单③,对账发现六④)——收敛判据从「check+build 绿」升为
+            #    「九门绿」:finish 后跑九门收口,门未全绿且轮数/预算有余时带 C6 反馈 resume 续修,
+            #    使 lab 复验口径与生产 RepairMiddleware(finish 点拦截→跑门→judge→回喂)行为同型。
             if session.finished is not None:
-                _rec("finish 已接受（check+build 绿）→ 收敛")
-                break
+                if not run_gates:
+                    _rec("finish 已接受(check+build 绿)→ 收敛(generation-only,不跑九门回喂)")
+                    break
+                _rec("finish 已接受(check+build 绿)→ 九门收口判(CLI 回喂对齐)")
+                closeout = _closeout_gates(game_id, port=port, cdp_port=cdp_port)
+                j = judge_cheap_verdict(closeout["verdict"], game_id=game_id,
+                                        staged_dir=cheap_run.wg1_game_dir(game_id))
+                vb = _verdict_brief(closeout["verdict"])
+                _rec(f"九门 play pass={vb['pass']} failedGates={vb['failedGates']}")
+                if j.passed:
+                    break
+                if attempt >= max_resumes:
+                    _rec("九门未绿,但 resume 轮数耗尽 → 交尽力产物收口")
+                    break
+                # 预算判定与生产同口径:软停已触发 / 实测已花越软停线 → 只收尾、不再回喂(裁决:越线只许收尾)。
+                budget_out = bool(getattr(breaker, "budget_soft_tripped", False)) or (
+                    breaker._rmb_gate_active and breaker.spent_rmb >= breaker.rmb_hard_limit)
+                if budget_out:
+                    _rec(f"九门未绿,但 ¥ 软停线已到(已花 ¥{breaker.spent_rmb:.4f})→ 只收尾、不再回喂")
+                    break
+                session.finished = None  # 重置收敛标记,带门反馈续修(跨 reply memory 保留=原地续修)
+                _rec(f"九门未绿(failedGates={j.failed_gates})→ 带 C6 反馈回喂续修")
+                kick = j.feedback  # C6 结构渲染全文(含 phaseNow/driverType/game-log 信号;与 RepairMiddleware 注入同源)
+                continue
             # ② 门已绿但没 finish → 踹一脚让它 finish
             c_ok = bool(session.last_check and session.last_check.get("ok"))
             b_ok = bool(session.last_build and session.last_build.get("ok"))
@@ -225,30 +284,30 @@ async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens=
         breaker_tripped = {"kind": None, "reason": f"{type(e).__name__}: {e}"}
         _rec(f"未捕获异常：{type(e).__name__}: {e}")
 
-    # ── 收口：finish 后 stage → smoke → ensure_play_spec → 循环外九门 play ──
+    # ── 收口:finish 后 stage → smoke → ensure_play_spec → 循环外九门 play ──
+    #   CLI 回喂对齐后,run_gates 路的九门收口已在循环内跑过(closeout),此处直接复用最后一轮结果,
+    #   不重复跑门;对照路(run_gates=False)保持原「只 stage+smoke」行为零改动。
     finished = session.finished is not None
     staged = False
     smoke_ok = None
     verdict = None
-    driver_type = None  # M3b U1：ensure_play_spec 产的 driver 类型（对照路 run_gates=False 不产 → 保持 None）
+    driver_type = None  # M3b U1:ensure_play_spec 产的 driver 类型(对照路 run_gates=False 不产 → 保持 None)
     if finished:
-        st = cheap_run.stage(game_id)
-        staged = st["ok"]
-        _rec(f"stage {'OK' if staged else 'FAIL: ' + st['output']}")
-        if staged:
-            sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port)
-            smoke_ok = sm["ok"]
-            _rec(f"smoke {'PASS' if smoke_ok else 'FAIL'}（抓 state 供 play-spec）")
-            if run_gates:
-                ps = cheap_run.ensure_play_spec(game_id, sm.get("state"))
-                driver_type = ps.get("driverType")  # M3b U1：driver 类型存进 trace.gatespec.driver（后端 D11 firstPlay 维）
-                _rec(f"play-spec {'已产 driver=' + ps.get('driverType', '?') if ps.get('wrote') else ps.get('reason', '?')}")
-                pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port)
-                verdict = pr["verdict"]
-                vb = _verdict_brief(verdict)
-                _rec(f"九门 play pass={vb['pass']} failedGates={vb['failedGates']}")
-            else:
-                _rec("generation-only（run_gates=False）：跳过 ensure_play_spec + 九门 play，交对照方注入金标 spec 后单独 play")
+        if run_gates:
+            cg = closeout or _closeout_gates(game_id, port=port, cdp_port=cdp_port)  # 防御:正常路循环内已跑
+            staged = cg["staged"]
+            smoke_ok = cg["smoke_ok"]
+            driver_type = cg["driver_type"]  # M3b U1:driver 类型存进 trace.gatespec.driver(后端 D11 firstPlay 维)
+            verdict = cg["verdict"]
+        else:
+            st = cheap_run.stage(game_id)
+            staged = st["ok"]
+            _rec(f"stage {'OK' if staged else 'FAIL: ' + st['output']}")
+            if staged:
+                sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port)
+                smoke_ok = sm["ok"]
+                _rec(f"smoke {'PASS' if smoke_ok else 'FAIL'}(抓 state 供 play-spec)")
+                _rec("generation-only(run_gates=False):跳过 ensure_play_spec + 九门 play,交对照方注入金标 spec 后单独 play")
 
     tok_in, tok_out = model.usage_sum()  # 成本落盘(R4)：token 台账始终可取(RecordingOpenAIChatModel.records 累计)。
     summary = {

tier2/gen-worker/tests/test_feedback_c6.py

@@ -0,0 +1,217 @@
+"""test_feedback_c6.py — C6 判卷反馈契约接线单测(W-S1 修复三单·单③)。
+
+守的不变量:
+  · latch 两条失败支都读:advisory 支(带 phaseNow)与驻留校验支(只带 after)都必须产出 phaseNow——
+    「卡在 menu/playing 三处可见却进不了反馈 → 盲修」的封口点(gate_judge 旧代码只读 after)。
+  · 每条逐门反馈必带 driverType(判卷用的哪族驱动器);staged play-spec.json 是取数源。
+  · game-log.json 的隔离告警/console error 摘要进反馈(截断保护,不整份灌 prompt)。
+  · 反馈先构造为 C6 结构化对象且过 contracts/play-loop/validate.py 强校验;渲染文本由同一结构派生
+    (单一来源,拼串丢字段结构上不可能);负样本(手工去掉 phaseNow)被校验器拦截。
+  · tier2 对齐(有界零行为变更):verdict_feedback 富文本逐字不动,新增 C6 结构化 sidecar 过校验。
+
+跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_feedback_c6.py -v
+"""
+import copy
+import importlib.util
+import json
+from pathlib import Path
+
+from worker.gate_judge import (
+    GateJudgment,
+    _latch_phase_now,
+    _render_cheap_feedback,
+    build_cheap_feedback_obj,
+    judge_cheap_verdict,
+)
+
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+_PLAYLOOP = _REPO_ROOT / "contracts" / "play-loop"
+
+
+def _load_validator():
+    spec = importlib.util.spec_from_file_location("_playloop_validate_c6", _PLAYLOOP / "validate.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_V = _load_validator()
+_C6 = json.loads((_PLAYLOOP / "verdict-feedback.schema.json").read_text(encoding="utf-8"))
+
+
+def _assert_c6(obj):
+    errors = _V.validate(_C6, obj, _C6)
+    assert errors == [], f"反馈对象未过 C6 契约校验:{errors}"
+
+
+# ── 便宜档九门 verdict 样例(latch 两支)──
+_V_ADVISORY = {"pass": False, "guards": {
+    "H_progress": {"pass": False, "checks": [], "latch": {
+        # advisory 失败支(play.cdp.cjs:1083):只带 phaseNow、不带 after。
+        "pass": False, "advisory": False, "reason": "真玩未到达终态且无进展", "phaseNow": "menu"}},
+}}
+_V_DWELL = {"pass": False, "guards": {
+    "H_progress": {"pass": False, "checks": [], "latch": {
+        # 驻留校验支(play.cdp.cjs:1089):带 after、不带 phaseNow(驻留复读时刻的 phase)。
+        "pass": False, "dwelledMs": 600, "after": "playing", "reason": "终态未驻留"}},
+}}
+
+
+# ───────────────────────── phaseNow 两支 ─────────────────────────
+
+def test_phase_now_advisory_branch():
+    """advisory 失败支:phaseNow 直取(旧代码丢它 → 反馈成 after=None,本测封口)。"""
+    obj = build_cheap_feedback_obj(_V_ADVISORY, game_id="g1")
+    item = obj["items"][0]
+    assert item["gate"] == "H_progress"
+    assert item["phaseNow"] == "menu"
+    _assert_c6(obj)
+    # 渲染文本必带 phaseNow 信号(agent 据它知道卡在哪)。
+    assert "menu" in _render_cheap_feedback(obj)
+
+
+def test_phase_now_dwell_branch():
+    """驻留校验支:latch 只带 after → phaseNow 取 after(回弹后的此刻 phase)。"""
+    obj = build_cheap_feedback_obj(_V_DWELL, game_id="g2")
+    item = obj["items"][0]
+    assert item["phaseNow"] == "playing"
+    assert item["latch"]["after"] == "playing"
+    _assert_c6(obj)
+
+
+def test_latch_phase_now_helper():
+    assert _latch_phase_now({"phaseNow": "menu"}) == "menu"
+    assert _latch_phase_now({"after": "over"}) == "over"
+    assert _latch_phase_now({"phaseNow": "menu", "after": "over"}) == "menu"  # phaseNow 优先
+    assert _latch_phase_now({}) is None
+
+
+def test_missing_phase_now_negative_sample_blocked():
+    """负样本:报 latch 失败却丢 phaseNow 的对象必须被 validate.py 拦(结构层封口的回归证明)。"""
+    obj = build_cheap_feedback_obj(_V_ADVISORY, game_id="g3")
+    bad = copy.deepcopy(obj)
+    del bad["items"][0]["phaseNow"]
+    errors = _V.validate(_C6, bad, _C6)
+    assert errors, "丢 phaseNow 的 latch 失败反馈必须被 C6 校验器拦下"
+    assert any("phaseNow" in e for e in errors)
+
+
+# ───────────────────────── driverType ─────────────────────────
+
+def test_driver_type_from_staged_spec(tmp_path):
+    """driverType 从 staged play-spec.json 读(判卷驱动器族);spec 缺 → 'none' 仍满足契约。"""
+    (tmp_path / "play-spec.json").write_text(
+        json.dumps({"driver": {"type": "key-cycle", "keys": ["ArrowLeft"]}}), encoding="utf-8")
+    obj = build_cheap_feedback_obj(_V_ADVISORY, game_id="g4", staged_dir=tmp_path)
+    assert all(it["driverType"] == "key-cycle" for it in obj["items"])
+    _assert_c6(obj)
+    obj2 = build_cheap_feedback_obj(_V_ADVISORY, game_id="g5")  # 无 staged_dir
+    assert all(it["driverType"] == "none" for it in obj2["items"])
+    _assert_c6(obj2)
+
+
+def test_driver_type_explicit_param_wins(tmp_path):
+    obj = build_cheap_feedback_obj(_V_ADVISORY, game_id="g6", staged_dir=tmp_path,
+                                   driver_type="tap-targets")
+    assert obj["items"][0]["driverType"] == "tap-targets"
+
+
+# ───────────────────────── game-log 摘要 ─────────────────────────
+
+def test_game_log_signals_into_feedback(tmp_path):
+    """game-log.json 的隔离告警/报错进反馈(evidence.gameLog + 渲染文本);截断保护。"""
+    ev = tmp_path / "evidence"
+    ev.mkdir()
+    entries = ([{"scope": "plugin", "tag": "info", "msg": f"normal {i}"} for i in range(50)]
+               + [{"scope": "host", "tag": "isolation-warn", "msg": "插件 juice 抛错已隔离"},
+                  {"scope": "game", "tag": "error", "msg": "score NaN"}])
+    (ev / "game-log.json").write_text(json.dumps(entries), encoding="utf-8")
+    obj = build_cheap_feedback_obj(_V_ADVISORY, game_id="g7", staged_dir=tmp_path)
+    log_brief = obj["evidence"]["gameLog"]
+    assert "隔离" in log_brief and "score NaN" in log_brief, "告警类条目应优先入摘要"
+    assert len(log_brief) <= 600, "摘要必须截断,不整份灌 prompt"
+    _assert_c6(obj)
+    assert "score NaN" in _render_cheap_feedback(obj)
+
+
+def test_no_game_log_is_fine(tmp_path):
+    obj = build_cheap_feedback_obj(_V_ADVISORY, game_id="g8", staged_dir=tmp_path)
+    assert "evidence" not in obj
+    _assert_c6(obj)
+
+
+# ───────────────────────── judge_cheap_verdict 接线 ─────────────────────────
+
+def test_judge_writes_c6_sidecar(tmp_path):
+    """门未绿 + 给 staged_dir → C6 结构化反馈落 evidence/verdict-feedback.json(证据留痕,且过校验)。"""
+    j = judge_cheap_verdict(_V_ADVISORY, game_id="g9", staged_dir=tmp_path)
+    assert isinstance(j, GateJudgment) and j.passed is False
+    sidecar = tmp_path / "evidence" / "verdict-feedback.json"
+    assert sidecar.exists()
+    obj = json.loads(sidecar.read_text(encoding="utf-8"))
+    _assert_c6(obj)
+    assert obj["gameId"] == "g9"
+    # feedback 文本 = 同一结构渲染(单一来源)。
+    assert j.feedback == _render_cheap_feedback(obj)
+
+
+def test_judge_backward_compatible_no_params():
+    """老签名(不带 game_id/staged_dir)仍工作:降级口径,判定不变。"""
+    j = judge_cheap_verdict(_V_ADVISORY)
+    assert j.passed is False and j.failed_gates == ["H_progress"]
+    assert "phase=menu" in j.feedback  # phaseNow 修复在降级口径同样生效
+
+
+def test_judge_green_no_sidecar(tmp_path):
+    """门全绿不产反馈对象、不落 sidecar(C6:门绿就不该有反馈)。"""
+    v = {"pass": True, "guards": {"A_boot": {"pass": True}}}
+    j = judge_cheap_verdict(v, game_id="g10", staged_dir=tmp_path)
+    assert j.passed is True
+    assert not (tmp_path / "evidence" / "verdict-feedback.json").exists()
+
+
+# ───────────────────────── tier2 对齐(有界零行为变更)─────────────────────────
+
+_TIER2_VERDICT = {
+    "decision": "reject",
+    "layerResults": {"L1": {"passed": False, "gateResults": [
+        {"gate": "H_progress", "passed": False, "fatal": True, "detail": "progress=false latch=false"},
+        {"gate": "A_boot", "passed": True, "fatal": True, "detail": ""},
+    ], "richGameGates": {"economy": {"passed": False, "checks": [
+        {"name": "profitPathWin", "ok": False, "detail": "coins=42(<100) phase=playing terminal=false"},
+    ]}}}},
+    "findings": [{"severity": "P0", "issue": "订单永远交不了", "suggestion": "修 mergeChains"}],
+}
+
+
+def test_tier2_feedback_obj_passes_c6():
+    from worker.run import verdict_feedback_obj
+
+    obj = verdict_feedback_obj(_TIER2_VERDICT, game_id="t2game")
+    _assert_c6(obj)
+    assert "H_progress" in obj["failedGates"]
+    assert "richGame:economy" in obj["failedGates"]
+    assert "finding:P0" in obj["failedGates"]
+    # 经济门逐子检查证据进 item.ext(Δ9③ 内容不对称)。
+    econ = next(it for it in obj["items"] if it["gate"] == "richGame:economy")
+    assert econ["ext"]["checks"][0]["name"] == "profitPathWin"
+
+
+def test_tier2_feedback_text_unchanged():
+    """tier2 喂模型的富文本保持不动(F-1 已验收):sidecar 是加性旁路,不碰 verdict_feedback 输出。"""
+    from worker.run import verdict_feedback
+
+    text = verdict_feedback(_TIER2_VERDICT, "", game_id=None)
+    assert "致命门 H_progress" in text
+    assert "富游戏门 economy" in text
+    assert "对抗 P0" in text
+
+
+def test_tier2_sidecar_write(tmp_path, monkeypatch):
+    from worker import run as _run
+
+    monkeypatch.setattr(_run, "_workdir", lambda gid: tmp_path)
+    _run.write_feedback_sidecar(_TIER2_VERDICT, game_id="t2sidecar")
+    sidecar = tmp_path / "evidence" / "verdict-feedback.json"
+    assert sidecar.exists()
+    _assert_c6(json.loads(sidecar.read_text(encoding="utf-8")))

tier2/gen-worker/worker/gate_judge.py

@@ -7,7 +7,9 @@
 """
 from __future__ import annotations
 
-from typing import NamedTuple
+import json
+from pathlib import Path
+from typing import NamedTuple, Optional
 
 
 class GateJudgment(NamedTuple):
@@ -55,6 +57,11 @@ def judge_tier2_verdict(gate_result: dict) -> GateJudgment:
     #   经济数值证据(三个数/静态差额/结构性错配);缺省 None 走降级口径(detail+why 仍有),判定不变。
     feedback = verdict_feedback(v, gate_result.get("log") or "",
                                 game_id=gate_result.get("gameId"))
+    # C6 结构化 sidecar(有界·零行为变更):喂模型的富文本 feedback 保持逐字不动(F-1 已验收),
+    #   另把同一 verdict 组成 C6 结构化对象 best-effort 落 evidence/verdict-feedback.json(run.py 实现)。
+    if not passed:
+        from worker.run import write_feedback_sidecar  # noqa: PLC0415 —— 同上惰性 import
+        write_feedback_sidecar(v, game_id=gate_result.get("gameId"))
     return GateJudgment(passed=passed, failed_gates=failed_gates, feedback=feedback)
 
 
@@ -99,17 +106,189 @@ def _cheap_gate_detail(name: str, g: dict) -> str:
                                  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]})")
+                # latch 两条失败支都读(修丢 phaseNow 缺陷,对账发现六①):
+                #   advisory 支(play.cdp.cjs:1083)只带 phaseNow、不带 after;驻留校验支(:1089)带 after 不带
+                #   phaseNow。旧代码只读 after → advisory 支失败时回喂成 after=None,agent 不知卡在哪个 phase 只能盲修。
+                phase_now = _latch_phase_now(latch)
+                parts.append(f"终局 latch 未达(phaseNow={phase_now} advisory={latch.get('advisory')}"
+                             f" {(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 的中文续修回喂。
+def _latch_phase_now(latch: dict):
+    """从 latch 判定细节取「游戏此刻卡在哪个 phase」——两条失败支都认(修 gate_judge 丢 phaseNow 缺陷)。
 
-    tier2 的 verdict_feedback 吃 layerResults 形状,cheap verdict 无 layerResults(只有 guards map),故新写。
+    advisory 失败支(play.cdp.cjs:1083)带 phaseNow;驻留校验支(:1089)只带 after(驻留复读时刻的 phase,
+    即回弹后的此刻 phase,语义上就是「卡在哪」)。两者都缺(异常态)返 None。
+    """
+    if not isinstance(latch, dict):
+        return None
+    return latch.get("phaseNow") if latch.get("phaseNow") is not None else latch.get("after")
+
+
+def _cheap_driver_type(staged_dir: Optional[Path]):
+    """读判卷用的驱动器族(staged 目录 play-spec.json 的 driver.type;C6 feedbackItem.driverType 来源)。
+
+    spec 缺 driver 但有 inputs → 'input-sequence'(固定输入序列判的);读不到/坏 JSON → None(上层落 'none')。
+    """
+    try:
+        if staged_dir is None:
+            return None
+        p = Path(staged_dir) / "play-spec.json"
+        if not p.exists():
+            return None
+        spec = json.loads(p.read_text(encoding="utf-8")) or {}
+        dt = (spec.get("driver") or {}).get("type")
+        if dt:
+            return str(dt)
+        if spec.get("inputs"):
+            return "input-sequence"
+        return None
+    except Exception:  # noqa: BLE001 —— 读不出驱动器族只降级,不连累反馈主体
+        return None
+
+
+def _game_log_brief(staged_dir: Optional[Path], *, max_entries: int = 8, max_chars: int = 600) -> str:
+    """从 staged evidence/game-log.json 摘运行时信号进反馈(截断保护,绝不把整份 log 灌进 prompt)。
+
+    game-log.json = smoke 段抓的 window.__gameLog(ctx.log 游戏语义 + 插件自带日志,含隔离告警),
+    条目形状 {scope, tag, msg}。摘取策略:tag/scope/msg 命中 error/warn/isolat/fail 的告警类条目优先,
+    不足再补末尾常规条目;逐条截 120 字、总长截 max_chars。读不到/坏 JSON → 空串(证据降级,不伪造)。
+    """
+    try:
+        if staged_dir is None:
+            return ""
+        p = Path(staged_dir) / "evidence" / "game-log.json"
+        if not p.exists():
+            return ""
+        arr = json.loads(p.read_text(encoding="utf-8"))
+        if not isinstance(arr, list) or not arr:
+            return ""
+
+        def _fmt(e) -> str:
+            if not isinstance(e, dict):
+                return str(e)[:120]
+            return f"[{e.get('scope')}:{e.get('tag')}] {e.get('msg') or ''}"[:120]
+
+        def _is_alarm(e) -> bool:
+            blob = (str(e.get("scope")) + str(e.get("tag")) + str(e.get("msg"))).lower() \
+                if isinstance(e, dict) else str(e).lower()
+            return any(k in blob for k in ("error", "warn", "isolat", "fail", "uncaught"))
+
+        alarms = [e for e in arr if _is_alarm(e)]
+        picked = alarms[-max_entries:]
+        if len(picked) < max_entries:
+            tail = [e for e in arr[-(max_entries - len(picked)):] if e not in picked]
+            picked = picked + tail
+        text = "\n".join(_fmt(e) for e in picked)
+        return text[:max_chars]
+    except Exception:  # noqa: BLE001 —— 日志摘要失败只降级为空,绝不抛
+        return ""
+
+
+def build_cheap_feedback_obj(v: dict, *, game_id=None, staged_dir=None,
+                             driver_type=None) -> Optional[dict]:
+    """把便宜档九门 verdict 构造成 C6 VerdictFeedback 结构化对象(contracts/play-loop/verdict-feedback.schema.json)。
+
+    单一来源纪律(对账发现六 / Δ1):反馈先成结构、给 agent 读的中文文本由 _render_cheap_feedback 从同一结构
+    渲染——字符串拼装丢字段(如只读 latch.after 丢 phaseNow)从结构上不再可能。
+    返回 None 的情形:verdict 缺失 / 无 guards / 无未过门——C6 契约恒 passed=false 且 items 非空,
+    没有失败面就不产反馈对象(那时该走 harness 说明或放行,不硬造)。
+
+    Args:
+        v: 便宜档九门 verdict({pass, guards{门:{pass,...}}})。
+        game_id: 本局工程 id(C6 required;缺省填 "(unknown)" 仅供渲染,落盘由调用方管)。
+        staged_dir: staged 工程目录 _wg1-gen/<id>/(读 play-spec.json 定 driverType、evidence/game-log.json
+            摘运行时信号);None → 两者降级(driverType 落 'none'、无日志段)。
+        driver_type: 显式驱动器族(优先于从 staged_dir 读;供已知 driver 的调用方少一次读盘)。
+    """
+    guards = (v or {}).get("guards") or {}
+    failed = [(name, g) for name, g in guards.items()
+              if isinstance(g, dict) and g.get("pass") is False]
+    if not failed:
+        return None
+    dt = driver_type or _cheap_driver_type(staged_dir) or "none"
+    items = []
+    for name, g in failed:
+        item = {"gate": name, "driverType": dt,
+                "fixDirection": _CHEAP_GATE_HINTS.get(name, f"{name} 门未通过。")}
+        detail = _cheap_gate_detail(name, g).strip()
+        if detail:
+            item["detail"] = detail
+        if name == "H_progress":
+            latch = g.get("latch")
+            if isinstance(latch, dict):
+                lat = {"pass": bool(latch.get("pass"))}
+                for k in ("advisory", "reason", "after"):
+                    if latch.get(k) is not None:
+                        lat[k] = latch[k]
+                item["latch"] = lat
+                if lat["pass"] is False:
+                    # C6 硬约束:报 latch 失败必带 phaseNow(两支取数见 _latch_phase_now;都缺=异常态,
+                    #   落 'unknown' 仍满足契约、且明示信号缺失而非静默丢字段)。
+                    phase_now = _latch_phase_now(latch)
+                    item["phaseNow"] = str(phase_now) if phase_now is not None else "unknown"
+        if name == "B_uncaught" and g.get("uncaught"):
+            item["evidence"] = {"console": str(g.get("uncaught"))[:300]}
+        items.append(item)
+    obj = {
+        "schemaVersion": "verdict-feedback/1",
+        "gameId": str(game_id) if game_id else "(unknown)",
+        "passed": False,
+        "failedGates": [name for name, _ in failed],
+        "items": items,
+    }
+    log_brief = _game_log_brief(staged_dir)
+    if log_brief:
+        # 信封级证据:游戏运行时日志摘要(C6 evidence 允许扩展键;隔离告警/报错优先、截断保护)。
+        obj["evidence"] = {"gameLog": log_brief}
+    return obj
+
+
+def _render_cheap_feedback(obj: dict) -> str:
+    """C6 结构化对象 → 给 agent 读的中文回喂全文(humanReadable 派生投影)。
+
+    单一来源:渲染只读 obj 里已有的字段、不夹带 obj 外信息——保证结构与文本永不分叉。
+    """
+    items = obj.get("items") or []
+    dt = items[0].get("driverType") if items else "none"
+    lines = []
+    for it in items:
+        line = "· " + (it.get("fixDirection") or f"{it.get('gate')} 门未通过。")
+        if it.get("detail"):
+            line += " " + it["detail"]
+        if it.get("phaseNow"):
+            line += f" 游戏此刻停在 phase={it['phaseNow']}(真玩未驱动到终局)。"
+        console = (it.get("evidence") or {}).get("console")
+        if console:
+            line += f" console:{console}"
+        lines.append(line)
+    text = (f"以下真玩验收门未通过(判卷驱动器={dt}),请据每条修复后再交付,不要直接结束:\n"
+            + "\n".join(lines))
+    log_brief = (obj.get("evidence") or {}).get("gameLog")
+    if log_brief:
+        text += "\n游戏运行时日志信号(节选,含隔离告警/报错):\n" + log_brief
+    return text
+
+
+def _write_cheap_feedback_sidecar(obj: dict, staged_dir) -> None:
+    """把 C6 反馈对象落 staged evidence/verdict-feedback.json(best-effort:证据留痕 + 供 driver/排障回看)。"""
+    try:
+        ev = Path(staged_dir) / "evidence"
+        ev.mkdir(parents=True, exist_ok=True)
+        (ev / "verdict-feedback.json").write_text(
+            json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
+    except Exception as e:  # noqa: BLE001 —— sidecar 落盘失败只告警,不连累续修主链
+        print(f"[gate-judge] C6 反馈 sidecar 落盘失败(忽略):{type(e).__name__}: {e}", flush=True)
+
+
+def _cheap_verdict_feedback(v: dict, *, game_id=None, staged_dir=None) -> str:
+    """把便宜档九门 verdict 的未过门组成给 agent 的中文续修回喂(C6 结构渲染,单一来源)。
+
+    tier2 的 verdict_feedback 吃 layerResults 形状,cheap verdict 无 layerResults(只有 guards map),故独立。
     verdict 缺失 / 无 guards(play 没跑出 verdict)→ 落一句 harness 说明(不伪造门绿)。
     """
     v = v or {}
@@ -117,24 +296,24 @@ def _cheap_verdict_feedback(v: dict) -> str:
     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:
+    obj = build_cheap_feedback_obj(v, game_id=game_id, staged_dir=staged_dir)
+    if obj is None:
         return "九门整体未放行,但未定位到具体未过门;请检查 check/build 与真玩可运行性。"
-    return "以下真玩验收门未通过,请据每条修复后再交付,不要直接结束:\n" + "\n".join(lines)
+    return _render_cheap_feedback(obj)
 
 
-def judge_cheap_verdict(play_result: dict) -> GateJudgment:
+def judge_cheap_verdict(play_result: dict, *, game_id=None, staged_dir=None) -> 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 无,不复用)。
+
+    C6 接线(W-S1 单③):feedback 由 C6 结构化对象渲染(build_cheap_feedback_obj → _render_cheap_feedback,
+    单一来源防拼串丢字段);带 phaseNow(latch 两失败支都读)、driverType(判卷驱动器族)、game-log 摘要。
+    game_id + staged_dir 由调用方传(cheap_service_app._cheap_check / cheap_studio CLI 回喂 / driver 收口),
+    缺省 None = 降级口径(驱动器族落 'none'、无日志段、不落 sidecar)——签名向后兼容,老调用不破。
     play_result 为 None / 空 / 无 guards(play 没产出 verdict)→ passed=False、失败门空、feedback 落 harness 说明。
     """
     v = play_result or {}
@@ -142,5 +321,11 @@ def judge_cheap_verdict(play_result: dict) -> GateJudgment:
     # 放行必须「顶层 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)
+    obj = build_cheap_feedback_obj(v, game_id=game_id, staged_dir=staged_dir) if not passed else None
+    if obj is not None:
+        feedback = _render_cheap_feedback(obj)
+        if staged_dir is not None:
+            _write_cheap_feedback_sidecar(obj, staged_dir)  # C6 证据落盘(best-effort)
+    else:
+        feedback = _cheap_verdict_feedback(v, game_id=game_id, staged_dir=staged_dir)
     return GateJudgment(passed=passed, failed_gates=failed_gates, feedback=feedback)

tier2/gen-worker/worker/run.py

@@ -1115,3 +1115,95 @@ def verdict_feedback(verdict: dict | None, harness_log: str = "", *,
         if f.get("severity") in ("P0", "P1"):
             lines.append(f"- 对抗 {f.get('severity')}:{f.get('issue', '')[:200]}")
     return ("真玩未过:\n" + "\n".join(lines)) if lines else "真玩未过(原因未知,见 verdict 细节)。"
+
+
+def verdict_feedback_obj(verdict: dict | None, *, game_id: str | None = None) -> dict | None:
+    """把 tier2 失败 verdict 组成 C6 VerdictFeedback 结构化对象(contracts/play-loop/verdict-feedback.schema.json)。
+
+    有界·零行为变更(W-S1 单③ tier2 对齐):喂给模型的 verdict_feedback 富文本保持逐字不动(F-1 已验收),
+    本函数只是把同一 verdict 的失败面**另组一份结构化 sidecar**——哪道门 / 驱动器族 / 逐子检查证据(进 ext,
+    Δ9③ 档位差异不进对称核心)。verdict 缺失 / 无失败面 → None(C6 恒 passed=false 且 items 非空,不硬造)。
+    读盘(play-spec 定 driverType)任何失败都降级为 'none',绝不抛。
+    """
+    if not verdict:
+        return None
+    l1 = ((verdict.get("layerResults") or {}).get("L1") or {})
+    # 驱动器族:run_gates 落盘的 play-spec.json driver.type(读不到降级 none;与门无关的门也统一带,契约要求每条有)。
+    driver_type = "none"
+    try:
+        if game_id:
+            spec_path = _workdir(game_id) / "play-spec.json"
+            if spec_path.exists():
+                spec = json.loads(spec_path.read_text(encoding="utf-8")) or {}
+                driver_type = str((spec.get("driver") or {}).get("type") or "none")
+    except Exception:  # noqa: BLE001 —— 驱动器族读不出只降级
+        driver_type = "none"
+
+    failed_gates: list[str] = []
+    items: list[dict] = []
+    # L1 九门(致命未过者;与 verdict_feedback 富文本同口径)。
+    for g in (l1.get("gateResults") or []):
+        if not g.get("passed") and g.get("fatal"):
+            gate = str(g.get("gate") or "?")
+            failed_gates.append(gate)
+            item = {"gate": gate, "driverType": driver_type}
+            detail = str(g.get("detail") or "")[:200]
+            if detail:
+                item["detail"] = detail
+            items.append(item)
+    # 富游戏三门(richGame:<门> 前缀;逐子检查证据进 ext,内容不对称、接口对称)。
+    rich = l1.get("richGameGates") or {}
+    for name in ("tripleLink", "economy", "latch"):
+        gate = rich.get(name)
+        if isinstance(gate, dict) and gate.get("passed") is False:
+            gid_name = f"richGame:{name}"
+            failed_gates.append(gid_name)
+            bad = [c for c in (gate.get("checks") or [])
+                   if isinstance(c, dict) and not c.get("ok")]
+            item = {"gate": gid_name, "driverType": driver_type,
+                    "fixDirection": "按失败子检查逐条修(详见 ext.checks 与富文本反馈)。"}
+            if bad:
+                item["ext"] = {"checks": [
+                    {"name": c.get("name"), "detail": str(c.get("detail") or "")[:200],
+                     "why": _RICH_CHECK_WHY.get(c.get("name") or "", "")}
+                    for c in bad
+                ]}
+            items.append(item)
+    # 对抗发现(finding:<severity>)。
+    for f in (verdict.get("findings") or []):
+        sev = f.get("severity")
+        if sev in ("P0", "P1"):
+            gid_name = f"finding:{sev}"
+            failed_gates.append(gid_name)
+            items.append({"gate": gid_name, "driverType": driver_type,
+                          "detail": str(f.get("issue") or "")[:200],
+                          "ext": {"severity": sev,
+                                  "suggestion": str(f.get("suggestion") or "")[:200]}})
+    if not items:
+        return None
+    return {
+        "schemaVersion": "verdict-feedback/1",
+        "gameId": str(game_id) if game_id else "(unknown)",
+        "passed": False,
+        "failedGates": failed_gates,
+        "items": items,
+    }
+
+
+def write_feedback_sidecar(verdict: dict | None, *, game_id: str | None = None) -> None:
+    """把 C6 结构化反馈落工程 evidence/verdict-feedback.json(best-effort,证据留痕;失败只告警绝不抛)。
+
+    调用点 = gate_judge.judge_tier2_verdict(门未绿时);game_id 缺省 / verdict 无失败面 → 静默跳过。
+    """
+    try:
+        if not game_id:
+            return
+        obj = verdict_feedback_obj(verdict, game_id=game_id)
+        if obj is None:
+            return
+        ev = _workdir(game_id) / "evidence"
+        ev.mkdir(parents=True, exist_ok=True)
+        (ev / "verdict-feedback.json").write_text(
+            json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
+    except Exception as e:  # noqa: BLE001 —— sidecar 是旁路证据,任何异常不连累续修主链
+        print(f"[tier2-run] C6 反馈 sidecar 落盘失败(忽略):{type(e).__name__}: {e}", flush=True)