d00938ccbd
便宜档生成切到 cheap-worker 新路后,result_out 只产 trace.cost、缺 9d trace 七项,致后端 D11 ReadinessScorer 三维恒中性、退化成 45/55 双峰近恒定分,D9 反同质失效。
U1 cheap_studio 富化:build_trace_source + _furthest_stage,additive 捕获 verdictFull/driverType/models/stage。U2 result_out._build_trace:七项(repairs=max(0,attempts-1))+ sevenGateVerdict{pass,guards} + gatespec{driver} + cost + similarity,camelCase 镜像 _extract_trace、None 容错。U3 D9 dedup vendored 进 cheap-worker/dedup.py(registry 指自己 results/)、worker_service 接线非阻断。
Java 跨语言切片测加逐维反假绿门(反射调真 ReadinessScorer:三维脱离中性 + 合成分 100 vs 退化 ≤55 拉开 ≥35)。顺带修预存 M3a 红测:worker 异常兜底 failureReason 改传真枚举 llm_error。
验证:cheap-worker 全 16 测绿 + Java 切片 5/5 绿 BUILD SUCCESS。范围仅 cheap-worker/ + 2 fixture + 1 Java 测 + plan,零 live 触碰。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
87 lines
4.0 KiB
Python
87 lines
4.0 KiB
Python
"""test_trace_source.py — M3b U1：build_trace_source 9d-trace 源维度富化单测。
|
||
|
||
守的不变量：
|
||
· 纯函数 build_trace_source(verdict, driver_type, attempts, stage, model) 产 additive 4 键
|
||
verdictFull/driverType/models/stage，供 result_out 组 payload["trace"]、后端 D11 算分；
|
||
· verdictFull 保留完整 verdict（guards.H_progress 仍是嵌套对象，含 pass）；play 没跑 → None；
|
||
· driverType 透传（没产 play-spec → None）；models = {code: model} map；
|
||
· 不回写既有键 attempts（守「不动既有键」，repairs 由 result_out 据 summary.attempts 派生）；
|
||
· _furthest_stage 据收口实际到达步判最远阶段。
|
||
|
||
不跑真 gen（纯函数，合成 verdict dict）。
|
||
跑：cheap-worker/.venv/bin/python cheap-worker/tests/test_trace_source.py
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
||
import cheap_studio as S # noqa: E402
|
||
|
||
# 合成完整九门 verdict（含 guards.H_progress 嵌套对象，对齐真 verdict.json 形态；带 url/ts 噪声供 U2 验去噪）。
|
||
_FULL_VERDICT = {
|
||
"pass": True,
|
||
"url": "http://localhost:4320/?cheap",
|
||
"ts": 1700000000,
|
||
"guards": {
|
||
"A_boot": {"pass": True},
|
||
"H_progress": {"pass": True, "checks": ["score+"], "latch": True, "state0": 0, "state1": 3},
|
||
},
|
||
}
|
||
|
||
|
||
def test_build_trace_source_success_shape():
|
||
"""成功路：verdictFull 含完整 guards（H_progress 嵌套保 pass）、driverType 透传、models map、stage。"""
|
||
out = S.build_trace_source(_FULL_VERDICT, "tap-targets", 1, "play", "MiniMax-M3")
|
||
assert out["verdictFull"]["pass"] is True
|
||
# H_progress 保持嵌套对象（不是裸 bool），后端 ReadinessScorer.guardPass 读其 .pass
|
||
assert isinstance(out["verdictFull"]["guards"]["H_progress"], dict)
|
||
assert out["verdictFull"]["guards"]["H_progress"]["pass"] is True
|
||
assert out["driverType"] == "tap-targets"
|
||
assert out["models"] == {"code": "MiniMax-M3"}
|
||
assert out["stage"] == "play"
|
||
|
||
|
||
def test_build_trace_source_none_verdict():
|
||
"""play 没跑（verdict=None）→ verdictFull=None；driverType=None（不崩）。"""
|
||
out = S.build_trace_source(None, None, 1, "smoke", "MiniMax-M3")
|
||
assert out["verdictFull"] is None
|
||
assert out["driverType"] is None
|
||
assert out["models"] == {"code": "MiniMax-M3"}
|
||
assert out["stage"] == "smoke"
|
||
|
||
|
||
def test_build_trace_source_does_not_emit_attempts():
|
||
"""不回写既有键 attempts（守「不动既有键」；repairs 由 result_out 据 summary.attempts 派生）。"""
|
||
out = S.build_trace_source(_FULL_VERDICT, "tap-targets", 3, "play", "MiniMax-M3")
|
||
assert "attempts" not in out
|
||
|
||
|
||
def test_build_trace_source_non_dict_verdict_to_none():
|
||
"""verdict 非 dict（异常形态）→ verdictFull=None（防御）。"""
|
||
out = S.build_trace_source("not-a-dict", None, 1, "code", "MiniMax-M3")
|
||
assert out["verdictFull"] is None
|
||
|
||
|
||
def test_furthest_stage_ordering():
|
||
"""_furthest_stage 据收口实际到达步取最远：play>smoke>stage>code>scaffold。"""
|
||
assert S._furthest_stage(finished=True, staged=True, smoke_ran=True, played=True) == "play"
|
||
assert S._furthest_stage(finished=True, staged=True, smoke_ran=True, played=False) == "smoke"
|
||
assert S._furthest_stage(finished=True, staged=True, smoke_ran=False, played=False) == "stage"
|
||
assert S._furthest_stage(finished=True, staged=False, smoke_ran=False, played=False) == "code"
|
||
assert S._furthest_stage(finished=False, staged=False, smoke_ran=False, played=False) == "scaffold"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
||
_failed = 0
|
||
for _fn in _fns:
|
||
try:
|
||
_fn()
|
||
print(f" PASS {_fn.__name__}")
|
||
except Exception as e: # noqa: BLE001
|
||
_failed += 1
|
||
print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}")
|
||
print(f"\n{len(_fns) - _failed}/{len(_fns)} passed")
|
||
sys.exit(1 if _failed else 0)
|