fix(cheap): 收口采集重复打印幂等(工单 f)+ bake_off brief 全缺 fail-fast(工单 i)
工单 f:_CheapRunCollector._flush 在 on_reply 的 REPLY_END / except / finally 三条降级路
会被调 ≥2 次,落盘每次覆盖是幂等的(采集语义不变、driver 恒读到最新),但「收口采集落盘」
成功行重复打印纯观感噪声。加实例标记 _flush_logged 只在首次成功落盘时打,后续静默覆盖
(失败行不受此门、每次都报便于诊断)。
工单 i:run_bakeoff 对无 brief 品类 skip+continue,全缺时 valid 空 → aggregate(genre_runs,[])
的 overall=(not [])and(not [])=True → 在 0 款上打「达标 ✅」exit 0(S5b 第一跑生产实录的
空过陷阱 / 无声截断)。抽 _resolve_briefs 纯解析 valid/missing;采样列表为空 → 立即
SystemExit(2)、stderr 打印缺失清单(品类 + 期望 run-summary 路径)、绝不进入判定;有样本
但部分缺 → 逐条告警后按既有逻辑继续(不升级为全停,judge_genre/aggregate 判定口径不改)。
测试 +6:工单 f 直接三次 _flush + 崩溃路端到端各验只打一行(capsys);工单 i 全缺 fail-fast
+缺失清单 / 正常路径行为不变 / 部分缺继续 / _resolve_briefs 拆分。全套 335 passed(329+6)。
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
lili
parent
2e4dd56b76
commit
eb95441dd6
@ -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()
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user