diff --git a/cheap-worker/cheap_service_app.py b/cheap-worker/cheap_service_app.py index c9e156a8..990e3787 100644 --- a/cheap-worker/cheap_service_app.py +++ b/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, diff --git a/cheap-worker/cheap_service_driver.py b/cheap-worker/cheap_service_driver.py index 999d409f..8f5c75ab 100644 --- a/cheap-worker/cheap_service_driver.py +++ b/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") diff --git a/cheap-worker/cheap_studio.py b/cheap-worker/cheap_studio.py index 034b1647..f39d640b 100644 --- a/cheap-worker/cheap_studio.py +++ b/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 = { diff --git a/tier2/gen-worker/tests/test_feedback_c6.py b/tier2/gen-worker/tests/test_feedback_c6.py new file mode 100644 index 00000000..0dc91ca6 --- /dev/null +++ b/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"))) diff --git a/tier2/gen-worker/worker/gate_judge.py b/tier2/gen-worker/worker/gate_judge.py index 2cced61e..50110334 100644 --- a/tier2/gen-worker/worker/gate_judge.py +++ b/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//(读 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) diff --git a/tier2/gen-worker/worker/run.py b/tier2/gen-worker/worker/run.py index 8d68dc48..10b28569 100644 --- a/tier2/gen-worker/worker/run.py +++ b/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:)。 + 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)