diff --git a/cheap-worker/bake_off.py b/cheap-worker/bake_off.py index 9da0d050..f61a6462 100644 --- a/cheap-worker/bake_off.py +++ b/cheap-worker/bake_off.py @@ -20,6 +20,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) # → cheap-worker/ import _bootstrap # noqa: E402,F401 跨包 sys.path + key 兜底 +import cheap_run # noqa: E402 brief 期望路径构造(缺失清单用,与 compare_node.load_brief 同源) import cheap_studio # noqa: E402 import compare_node as C # noqa: E402 复用 GENRES / load_brief / _port_pool / _next_index / _GENRE_BY_KEY @@ -150,6 +151,35 @@ def run_one_sync(genre_key: str, brief: str, k: int, port: int, cdp: int) -> dic "richness": summary.get("richness")} # U-B1:additive 丰富度评分(run_studio 已写 summary,非阻塞) +def _brief_path(genre: dict) -> Path: + """品类 brief 的期望源路径(与 compare_node.load_brief 同源:base 样本 evidence/run-summary.json)。 + + 仅用于缺失清单里给出「品类 → 期望路径」,便于定位是哪份 run-summary 没生成/为空(工单 i)。 + """ + return cheap_run.game_dir(genre["base"]) / "evidence" / "run-summary.json" + + +def _resolve_briefs(genre_keys: list) -> tuple: + """把品类键解析成 (valid, missing) —— 纯解析,零真跑、零 LLM(可单测)。 + + valid = [(key, brief)]:brief 源存在且非空的品类(保持输入序,与既有 run_bakeoff 收集逻辑一致)。 + missing = [(key, detail)]:未知品类(detail=原因)或 brief 源缺失/为空(detail=期望 run-summary 路径)。 + 调用方据 valid 是否为空决定「空过陷阱兜底」是否触发(见 run_bakeoff);判定阈值/口径不涉。 + """ + valid, missing = [], [] + for key in genre_keys: + genre = C._GENRE_BY_KEY.get(key) + if genre is None: + missing.append((key, "未知品类(不在 GENRES 注册表)")) + continue + brief = C.load_brief(genre) + if not brief: + missing.append((key, str(_brief_path(genre)))) + continue + valid.append((key, brief)) + return valid, missing + + async def run_bakeoff(genre_keys: list, n: int, conc: int = 1, offset: int = 0, base_port: int = 4320, base_cdp: int = 9222) -> dict: """三品类 × n 达标批跑:逐款 run_studio(自动 spec 九门),按品类聚合 ≥80%。 @@ -163,19 +193,22 @@ async def run_bakeoff(genre_keys: list, n: int, conc: int = 1, offset: int = 0, for pp in C._port_pool(conc, base_port, base_cdp): pool.put_nowait(pp) - valid = [] - genre_runs = {} - for key in genre_keys: - genre = C._GENRE_BY_KEY.get(key) - if genre is None: - print(f"[skip] 未知品类 {key}", file=sys.stderr) - continue - brief = C.load_brief(genre) - if not brief: - print(f"[skip] {key} 无 brief(base {genre['base']} run-summary 缺失)", file=sys.stderr) - continue - valid.append((key, brief)) - genre_runs[key] = [] + # brief 源解析(纯逻辑,见 _resolve_briefs):valid=有非空 brief 的品类;missing=未知/缺 brief 的品类+期望路径。 + valid, missing = _resolve_briefs(genre_keys) + for key, detail in missing: # 缺失品类逐条告警(部分缺也打,便于定位是哪份 run-summary 缺;partial 不升级为全停) + print(f"[skip] {key} 无 brief(期望 {detail})", file=sys.stderr) + # 空过陷阱兜底(工单 i · S5b 第一跑生产实录):采样列表为空(所有必需品类都缺 brief)→ 立即非零退出、绝不进入判定。 + # 现状会静默跳过全部品类 → 后面 aggregate(genre_runs, []) 的 overall = (not []) and (not []) = True, + # 于是在 0 款上打「达标 ✅」exit 0(无声截断)。这里在进入 gather/aggregate 前把「无样本可判」变响亮失败, + # 拒绝在 0 款上判达标。有样本但部分缺 → 上面已逐条告警、按既有逻辑继续(judge_genre/aggregate 判定口径不改)。 + if not valid: + print("\n[bake-off] ✗ 无任何可判样本:所有必需品类都缺 brief,拒绝在 0 款上判达标(空过陷阱兜底)。", + file=sys.stderr) + print("[bake-off] 缺失清单(品类 → 期望 brief 源):", file=sys.stderr) + for key, detail in missing: + print(f" · {key} → {detail}", file=sys.stderr) + raise SystemExit(2) # 立即非零退出:绝不进入 aggregate,把「0 款假达标」变永久绊线 + genre_runs = {key: [] for (key, _) in valid} async def one(key, brief, k): port, cdp = await pool.get() diff --git a/cheap-worker/cheap_service_app.py b/cheap-worker/cheap_service_app.py index 55946055..c016f9ed 100644 --- a/cheap-worker/cheap_service_app.py +++ b/cheap-worker/cheap_service_app.py @@ -163,6 +163,7 @@ def _get_collector_cls(): self._breaker = breaker self._repair = repair self._tracer = tracer + self._flush_logged = False # 工单 f:幂等打印标记——多次 _flush 只打一行「收口采集落盘」(落盘覆盖语义不变) async def on_reply(self, agent, input_kwargs, next_handler): # C1:框架 _agent.py:615 先 yield ReplyEndEvent 再 yield finish Msg 再收尾;collector 是最外层 on_reply, @@ -220,8 +221,13 @@ def _get_collector_cls(): ev.mkdir(parents=True, exist_ok=True) (ev / "service-run-summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") - print(f"[cheap-service] 收口采集落盘 game={self._game_id} costRmb={summary['costRmb']} " - f"repairs={summary['repairs']} softTripped={summary['budgetSoftTripped']}", flush=True) + # 工单 f:_flush 在 on_reply 的 REPLY_END / except / finally 三条降级路会被调 ≥2 次,落盘每次覆盖 + # 是幂等的(采集语义不变、driver 恒读到最新),但成功日志只需一行——重复打「收口采集落盘」纯观感噪声。 + # 用实例标记只在首次成功落盘时打,后续静默覆盖(失败行不受此门、每次都报,便于诊断)。 + if not self._flush_logged: + self._flush_logged = True + print(f"[cheap-service] 收口采集落盘 game={self._game_id} costRmb={summary['costRmb']} " + f"repairs={summary['repairs']} softTripped={summary['budgetSoftTripped']}", flush=True) except Exception as e: # noqa: BLE001 —— 采集 best-effort,失败绝不影响生成 print(f"[cheap-service] 收口采集失败(忽略,不影响生成):{type(e).__name__}: {e}", flush=True) diff --git a/cheap-worker/tests/test_bake_off.py b/cheap-worker/tests/test_bake_off.py index ceed95d8..df915485 100644 --- a/cheap-worker/tests/test_bake_off.py +++ b/cheap-worker/tests/test_bake_off.py @@ -13,11 +13,15 @@ test_bake_off.py — 便宜档 ≥80% 达标门聚合/判定逻辑单测(M1 U3,m 跑:cheap-worker/.venv/bin/python cheap-worker/tests/test_bake_off.py """ +import asyncio import sys from pathlib import Path +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/ import bake_off as B # noqa: E402 +import compare_node as C # noqa: E402 工单 i:monkeypatch C.load_brief 造 brief 缺失/齐全 def _p(n): @@ -174,6 +178,66 @@ def test_judge_is_deterministic_zero_llm(): assert r1["perGenre"]["a"]["passRate"] == 0.8 and r1["overallMeets"] is True +# ───────────────────────── brief 源缺失 fail-fast(工单 i · 空过陷阱兜底)───────────────────────── + +def _fake_run_one(genre_key, brief, k, port, cdp): + """mock 掉真跑(run_studio→M3/chrome/build):每款都收敛且过九门,模拟正常生产路的 run dict。""" + return {"gid": f"bake-{genre_key}-{k}", "passed": True, "finished": True, + "failedGates": None, "breakerKind": None, "costRmb": 0.3, "richness": None} + + +def test_run_bakeoff_all_briefs_missing_fails_fast(monkeypatch, capsys): + """全缺:所有必需品类都无 brief → 立即非零退出(SystemExit code≠0),stderr 打印缺失清单(品类名+期望路径), + 绝不进入 aggregate。守住空过陷阱:现状会在 0 款上打「达标 ✅」exit 0(S5b 第一跑生产实录的无声截断)。""" + monkeypatch.setattr(C, "load_brief", lambda genre: "") # 全部品类 brief 源为空 + with pytest.raises(SystemExit) as ei: + asyncio.run(B.run_bakeoff(["click-score", "whack-mole", "shop-serve"], n=1, conc=1)) + assert ei.value.code != 0, "空样本必须非零退出(区别于达标 exit 0)" + err = capsys.readouterr().err + assert "无任何可判样本" in err, "应拒绝在 0 款判达标、响亮失败" + for key in ("click-score", "whack-mole", "shop-serve"): + assert key in err, f"缺失清单应含品类名 {key}" + assert "run-summary.json" in err, "缺失清单应含期望 brief 路径(哪份 run-summary 缺)" + + +def test_run_bakeoff_normal_path_unchanged(tmp_path, monkeypatch): + """正常路径(brief 齐全)行为不变:不 fail-fast、走既有 gather→aggregate→报告→返回, + overallMeets 仍由九门 verdict 决定(判定阈值/口径不改)。mock 掉真跑与报告落盘,只验判定装配。""" + monkeypatch.setattr(C, "load_brief", lambda genre: "一个点击得分小游戏") # brief 齐全 + monkeypatch.setattr(B, "run_one_sync", _fake_run_one) + monkeypatch.setattr(B, "_next_report_path", lambda: tmp_path / "bake-off-test.json") + rep = asyncio.run(B.run_bakeoff(["click-score", "whack-mole"], n=2, conc=1)) # 不抛 SystemExit + assert rep["overallMeets"] is True # 两品类各 2/2 过 → 达标(判定口径不变) + assert rep["missingGenres"] == [] and rep["belowGenres"] == [] + assert set(rep["perGenre"].keys()) == {"click-score", "whack-mole"} + assert rep["perGenre"]["click-score"]["passRate"] == 1.0 + + +def test_run_bakeoff_partial_missing_continues(tmp_path, monkeypatch, capsys): + """有样本但部分缺:click-score 有 brief、whack-mole 无 → 不升级为全停,打印缺失后按既有逻辑只判 click-score + (别把部分缺升级成全停,判定口径不改)。""" + monkeypatch.setattr(C, "load_brief", + lambda genre: "一个点击得分小游戏" if genre["key"] == "click-score" else "") + monkeypatch.setattr(B, "run_one_sync", _fake_run_one) + monkeypatch.setattr(B, "_next_report_path", lambda: tmp_path / "bake-off-partial.json") + rep = asyncio.run(B.run_bakeoff(["click-score", "whack-mole"], n=1, conc=1)) # 不抛 SystemExit + err = capsys.readouterr().err + assert "whack-mole" in err and "run-summary.json" in err, "缺失部分被逐条告警(品类+期望路径)" + assert set(rep["perGenre"].keys()) == {"click-score"}, "只判有 brief 的品类(既有逻辑)" + assert rep["overallMeets"] is True + + +def test_resolve_briefs_splits_valid_and_missing(monkeypatch): + """_resolve_briefs 纯解析:有 brief→valid;无 brief→missing 带期望路径;未知品类→missing 带原因。""" + monkeypatch.setattr(C, "load_brief", + lambda genre: "brief" if genre["key"] == "click-score" else "") + valid, missing = B._resolve_briefs(["click-score", "whack-mole", "no-such-genre"]) + assert [k for k, _ in valid] == ["click-score"] + missing_map = dict(missing) + assert "run-summary.json" in missing_map["whack-mole"] # 缺 brief → 期望路径 + assert "未知品类" in missing_map["no-such-genre"] # 未知品类 → 原因 + + if __name__ == "__main__": _fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] _failed = 0 diff --git a/cheap-worker/tests/test_cheap_service_app.py b/cheap-worker/tests/test_cheap_service_app.py index 433c30bd..bf192121 100644 --- a/cheap-worker/tests/test_cheap_service_app.py +++ b/cheap-worker/tests/test_cheap_service_app.py @@ -145,3 +145,63 @@ def test_reply_end_event_class_name_canary(): # (成功局 costRmb=0),本断言让改名即红、把静默退化变永久绊线。生产码保留字符串判(不引硬 import 依赖、保 6c6g 惰性)。 from agentscope.event import ReplyEndEvent # 已实测可 import assert ReplyEndEvent.__name__ == "ReplyEndEvent" + + +def _make_collector(A_mod, game_id, tmp_path, monkeypatch): + # 工单 f 测试脚手架:mock game_dir 到 tmp,造一个绑好 fake breaker/repair/tracer 的 collector。 + monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}") + + class _FakeBreaker: + spent_rmb, _rmb_gate_active, budget_soft_tripped = 0.5, True, False + + class _FakeRepair: + repairs = 1 + + class _FakeTracer: + def summary(self): + return {"steps": 2} + + Collector = A_mod._get_collector_cls() + return Collector(game_id=game_id, breaker=_FakeBreaker(), repair=_FakeRepair(), tracer=_FakeTracer()) + + +def test_flush_prints_landing_line_once_idempotent(tmp_path, monkeypatch, capsys): + # 工单 f:_flush 在 REPLY_END / except / finally 三条降级路会被调 ≥2 次,落盘每次覆盖是幂等的(语义不变), + # 但「收口采集落盘」成功行只应打一行——重复打印纯观感噪声。此测直接连调三次 _flush,钉住「只打一行」。 + import json + col = _make_collector(A, "70055", tmp_path, monkeypatch) + col._flush() + col._flush() + col._flush() # 三次(模拟 REPLY_END + except + finally 最坏叠加) + out = capsys.readouterr().out + assert out.count("收口采集落盘") == 1, "多次 _flush 只打一行收口采集落盘(幂等打印,工单 f)" + # 采集本身幂等语义不变:多次调用后 sidecar 仍在盘上、内容为最新(costRmb/repairs 可读)。 + summary = json.loads((tmp_path / "amgen-70055" / "evidence" / "service-run-summary.json").read_text(encoding="utf-8")) + assert summary["costRmb"] == 0.5 and summary["repairs"] == 1 + + +def test_flush_landing_line_once_on_crash_path(tmp_path, monkeypatch, capsys): + # 工单 f 端到端:on_reply 崩溃路(next_handler 抛异常)会走 except 支 _flush + finally _flush 两次落盘, + # 修前打两行「收口采集落盘」,修后只打一行;合成 ReplyEndEvent 分支不产成功行、不干扰计数。 + col = _make_collector(A, "70056", tmp_path, monkeypatch) + + class _FakeState: + session_id, reply_id = "s1", "r1" + + class _FakeAgent: + state = _FakeState() + + async def _boom(**_): + raise RuntimeError("run 崩(压缩/熔断模拟)") + yield # 使其成为 async generator(永不到达) + + async def _drive(): + try: + async for _evt in col.on_reply(_FakeAgent(), {}, _boom): + pass # 合成 ReplyEndEvent 流经,不做处理 + except RuntimeError: + pass # 原异常按设计重抛,测试吞掉(不遮真失败) + + asyncio.run(_drive()) + out = capsys.readouterr().out + assert out.count("收口采集落盘") == 1, "崩溃路两次 _flush 也只打一行收口采集落盘(工单 f)"