lili 231b17194a fix(cheap-gen): new-api 额度耗尽真机探测揪出误判——401+「额度已用尽」修为 quota_exhausted
内测开放项真环境探测(2026-07-08,dev 栈恢复后):建 remain_quota=0 测试 token 打 new-api
/v1/chat/completions,坐实 new-api 对额度耗尽返 HTTP 401 + message「该令牌额度已用尽
!token.UnlimitedQuota && token.RemainQuota = 0」——非代码假设的 402/403,且关键词只认「额度不足/
insufficient」漏了「额度已用尽」。旧 classify_newapi_failure_reason:401→llm_error(可重试)、
关键词不匹配 → 把耗尽误判成可重试 → 反复重试到 15min watchdog 超时,而非干净"额度用尽"。

修:message 额度关键词优先(含实测「额度已用尽」「RemainQuota = 0」),401 不再无脑判 llm_error
(401 与凭据失效重叠,按 message 分辨:耗尽→quota_exhausted、真凭据失效→llm_error 可重试);
402/403 保守保留判耗尽。删代码 §3.9【阶段2 待验】TODO(已真机坐实)。新增真机 401 回归用例,
26 测过。正是代码自带 TODO 与「真环境>桩」要抓的:桩测永远发现不了 new-api 用 401 不用 402/403。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-08 07:52:57 -07:00

409 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""test_result_out.py — §6.1 result-out 三字段组装单测(M3a U1)。
守的不变量(KTD2 BLOCKER 修正,镜像 SaaGraphDispatcher.buildCallbackReqVO + service.py build_callback_payload):
· gameConfig = 最小占位 Map(templateId/title/theme/engineDriven),绝不塞源工程;
· engineBundle = bundle.iife.js 全文(含 __GameBundle)= 运行时承重产物;
· 成功门 = verdict.pass ∧ engineBundle 含 __GameBundle,否则 failed + 七值 failureReason;
· sourceProject = best-effort 可选(profile 派生不出即省略,不伪造、不阻断成功);
· traceId 取 job.traceId 或 job_id;纯函数无网络无 LLM、同输入恒同输出。
跑:cheap-worker/.venv/bin/python cheap-worker/tests/test_result_out.py
"""
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
import result_out as R # noqa: E402
_GOOD_BUNDLE = "var __GameBundle=(function(){return{bootGameHost(){}}})();"
def _tmp() -> Path:
return Path(tempfile.mkdtemp(prefix="ro-test-"))
def _make_game_dir(tmp: Path, *, bundle=_GOOD_BUNDLE, src_files=None) -> Path:
"""造 amgen-<id>/ 形态产物目录(bundle.iife.js + src/)。"""
gd = tmp / "amgen-t1"
(gd / "src").mkdir(parents=True)
if bundle is not None:
(gd / "bundle.iife.js").write_text(bundle, encoding="utf-8")
for name, content in (src_files or {"game-logic.js": "// logic\n", "core.js": "// core\n"}).items():
(gd / "src" / name).write_text(content, encoding="utf-8")
return gd
def test_succeeded_shape():
"""成功路:verdict.pass ∧ bundle 含 __GameBundle → status=succeeded + gameConfig 占位 + engineBundle 承重。"""
gd = _make_game_dir(_tmp())
job = {"job_id": "trace-abc", "traceId": "trace-abc", "templateId": "generic", "brief": "点点乐"}
summary = {"ok": True, "finished": True, "verdict": {"pass": True, "failedGates": []}, "costRmb": 0.05}
out = R.build_result_out(job, summary, gd)
assert out["traceId"] == "trace-abc"
assert out["status"] == "succeeded"
assert out["templateId"] == "generic"
# gameConfig 占位,绝不含源工程(BLOCKER 修正)
assert out["gameConfig"]["templateId"] == "generic"
assert out["gameConfig"]["theme"] == "generic"
assert out["gameConfig"]["engineDriven"] is True
assert "files" not in out["gameConfig"] and "entry" not in out["gameConfig"]
# engineBundle 承重
assert "__GameBundle" in out["engineBundle"]
assert out.get("failureReason") is None
def test_failed_when_gates_not_pass():
"""verdict.pass=False → status=failed + 七值 failureReason,不带 engineBundle。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "t2", "templateId": "generic", "brief": "x"}
summary = {"finished": True, "verdict": {"pass": False, "failedGates": ["E_live"]}}
out = R.build_result_out(job, summary, gd)
assert out["status"] == "failed"
assert out.get("engineBundle") is None
assert out["failureReason"] in R._FAILURE_REASONS # 对齐后端 FailureReasonEnum 真七值
def test_failed_when_engine_bundle_missing___gamebundle():
"""即使 verdict.pass,bundle 无 __GameBundle → status=failed(承重门:engineBundle 含 __GameBundle)。"""
gd = _make_game_dir(_tmp(), bundle="var notTheGlobal=1;")
job = {"traceId": "t3", "templateId": "generic", "brief": "x"}
summary = {"finished": True, "verdict": {"pass": True, "failedGates": []}}
out = R.build_result_out(job, summary, gd)
assert out["status"] == "failed"
assert out.get("engineBundle") is None
def test_failed_breaker_maps_llm_error():
"""预算熔断 / 内部真因不在七值内 → failureReason=llm_error(契约#6 catch-all;枚举无 budget 专项值)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "t4", "templateId": "generic", "brief": "x"}
summary = {"finished": False, "breaker": True, "verdict": {"pass": None, "failedGates": None}}
out = R.build_result_out(job, summary, gd)
assert out["status"] == "failed"
assert out["failureReason"] == "llm_error"
def test_explicit_failure_reason_wins():
"""显式 failure_reason(在七值内)优先于推断。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "t5", "templateId": "generic", "brief": "x"}
summary = {"finished": True, "verdict": {"pass": False, "failedGates": ["A_boot"]}}
out = R.build_result_out(job, summary, gd, failure_reason="timeout")
assert out["status"] == "failed"
assert out["failureReason"] == "timeout"
# ── WU2 §3.9:额度耗尽干净失败 + one-api 惯例分辨 ──
def test_quota_exhausted_in_enum():
"""quota_exhausted 已进 failureReason 枚举(跨契约共享枚举同批新增,与 aigc.yaml/dify-workflow-io.json 一致)。"""
assert "quota_exhausted" in R._FAILURE_REASONS
def test_summary_failure_reason_quota_maps_through():
"""WU2 §3.9:summary.failureReason=quota_exhausted(Service 分辨→driver 透传)→ 落 failureReason=quota_exhausted。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "tq", "templateId": "generic", "brief": "x"}
summary = {"finished": False, "verdict": {"pass": False, "failedGates": None},
"failureReason": "quota_exhausted"}
out = R.build_result_out(job, summary, gd)
assert out["status"] == "failed"
assert out["failureReason"] == "quota_exhausted"
def test_summary_failure_reason_ignored_when_not_in_enum():
"""summary.failureReason 非枚举值 → 不透传,回落 catch-all llm_error(防脏值污染契约)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "tq2", "templateId": "generic", "brief": "x"}
summary = {"finished": False, "verdict": {"pass": False}, "failureReason": "some_garbage"}
out = R.build_result_out(job, summary, gd)
assert out["failureReason"] == "llm_error"
def test_explicit_failure_reason_wins_over_summary():
"""显式入参优先级高于 summary.failureReason。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "tq3", "templateId": "generic", "brief": "x"}
summary = {"verdict": {"pass": False}, "failureReason": "quota_exhausted"}
out = R.build_result_out(job, summary, gd, failure_reason="timeout")
assert out["failureReason"] == "timeout"
def test_classify_newapi_status_codes():
"""状态码分辨:402/403(one-api 惯例额度码)→ quota_exhausted;裸 401 / 500 / None(无额度关键词)→ llm_error 可重试。"""
assert R.classify_newapi_failure_reason(402) == "quota_exhausted"
assert R.classify_newapi_failure_reason(403) == "quota_exhausted"
assert R.classify_newapi_failure_reason(401) == "llm_error" # 裸 401 无 message = 真凭据失效
assert R.classify_newapi_failure_reason(500) == "llm_error"
assert R.classify_newapi_failure_reason(None) == "llm_error"
def test_classify_newapi_real_quota_exhausted_401():
"""2026-07-08 真机耗尽探测坐实:new-api 对额度耗尽返 HTTP 401 + 「该令牌额度已用尽 …RemainQuota = 0」。
401 与凭据失效状态码重叠,故必须按 message 分辨——耗尽判 quota_exhausted(干净失败),真凭据失效仍 llm_error(可重试)。
这是关键回归:旧代码 401→llm_error 会把耗尽误判成可重试 → 反复重试到 watchdog 超时。"""
real = "[sk-air***0AS] 该令牌额度已用尽 !token.UnlimitedQuota && token.RemainQuota = 0 (request id: 20260708...)"
assert R.classify_newapi_failure_reason(401, real) == "quota_exhausted"
assert R.classify_newapi_failure_reason(401, "无效的令牌,请检查") == "llm_error" # 真凭据失效仍可重试、不塌成耗尽
def test_classify_newapi_keyword_fallback():
"""message 关键词分辨额度耗尽(状态码取不到 / 被框架包装时同样生效,优先于状态码)。"""
assert R.classify_newapi_failure_reason(None, "error: insufficient user quota") == "quota_exhausted"
assert R.classify_newapi_failure_reason(None, "当前分组余额不足,请充值") == "quota_exhausted"
assert R.classify_newapi_failure_reason(None, "connection reset by peer") == "llm_error"
def test_newapi_error_status_text_extraction():
"""从带 status_code/response 的异常对象抽 (status, text);抽不到 status 返 None、文本兜底 str(exc)。"""
class _Err(Exception):
status_code = 402
message = "insufficient user quota"
st, txt = R.newapi_error_status_text(_Err("boom"))
assert st == 402 and "insufficient" in txt
class _Resp:
status_code = 403
class _Wrapped(Exception):
response = _Resp()
st2, _ = R.newapi_error_status_text(_Wrapped("x"))
assert st2 == 403
st3, txt3 = R.newapi_error_status_text(ValueError("plain error"))
assert st3 is None and "plain error" in txt3
def test_traceid_fallback_to_job_id():
"""traceId 缺 → 用 job_id(贯穿键同值)。"""
gd = _make_game_dir(_tmp())
job = {"job_id": "jid-9", "templateId": "generic", "brief": "x"}
summary = {"verdict": {"pass": True}}
out = R.build_result_out(job, summary, gd)
assert out["traceId"] == "jid-9"
def test_cost_goes_into_trace_not_toplevel():
"""成本落 trace.cost.totalRmb——顶层不得有 costRmb(DifyCallbackReqVO 无此字段、后端严格拒未知)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "t6", "templateId": "generic", "brief": "x"}
summary = {"verdict": {"pass": True}, "costRmb": 0.07}
out = R.build_result_out(job, summary, gd)
assert "costRmb" not in out, "顶层不得有 costRmb(会被后端 ObjectMapper 拒)"
assert out["trace"]["cost"]["totalRmb"] == 0.07
def test_result_out_keys_within_vo_known_set():
"""result-out 顶层键严格 ⊆ DifyCallbackReqVO 已知字段集(后端拒未知字段的硬约束)。"""
vo_known = {"traceId", "status", "templateId", "gameConfig", "assets",
"engineBundle", "failureReason", "qualityScore", "sourceProject", "trace"}
gd = _make_game_dir(_tmp())
job = {"traceId": "t9", "templateId": "generic", "brief": "x"}
out_ok = R.build_result_out(job, {"verdict": {"pass": True}, "costRmb": 0.07}, gd,
source_project='{"schemaVersion":"2.0"}')
out_fail = R.build_result_out(job, {"verdict": {"pass": False}}, gd)
assert set(out_ok) <= vo_known, f"成功路含未知字段:{set(out_ok) - vo_known}"
assert set(out_fail) <= vo_known, f"失败路含未知字段:{set(out_fail) - vo_known}"
def test_build_source_project_valid_2_0():
"""profile 可派生 + src/ 有文件 → sourceProject 是合法 2.0(必填齐 + sourceHash 64-hex)。"""
import json
import re
gd = _make_game_dir(_tmp(), src_files={"game-logic.js": "// l\n", "core.js": "// c\n"})
profile = {"tickModel": "frame", "inputModel": "tap-targets", "progressModel": "score"}
sp_str = R.build_source_project(gd, profile, entry="entry.js")
assert sp_str is not None
sp = json.loads(sp_str)
assert sp["schemaVersion"] == "2.0"
assert sp["globalName"] == "__GameBundle"
assert sp["entry"] == "entry.js"
assert sp["profile"] == profile
assert "src/game-logic.js" in sp["files"] and "src/core.js" in sp["files"]
assert re.fullmatch(r"[a-f0-9]{64}", sp["sourceHash"])
def test_build_source_project_omitted_when_profile_incomplete():
"""profile 派生不出(None / 三枚举不全)→ 返 None(省略、绝不伪造)。"""
gd = _make_game_dir(_tmp())
assert R.build_source_project(gd, None, entry="entry.js") is None
assert R.build_source_project(gd, {"tickModel": "frame"}, entry="entry.js") is None
def test_build_source_project_omitted_when_no_src():
"""src/ 无文件 → 返 None。"""
empty = _tmp() / "amgen-e"
(empty / "src").mkdir(parents=True)
profile = {"tickModel": "frame", "inputModel": "tap-targets", "progressModel": "score"}
assert R.build_source_project(empty, profile, entry="entry.js") is None
def test_result_out_source_project_additive():
"""build_result_out 收到 source_project(JSON str)→ payload 带;缺省不带(additive)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "t7", "templateId": "generic", "brief": "x"}
summary = {"verdict": {"pass": True}}
out_with = R.build_result_out(job, summary, gd, source_project='{"schemaVersion":"2.0"}')
assert out_with["sourceProject"] == '{"schemaVersion":"2.0"}'
out_without = R.build_result_out(job, summary, gd)
assert "sourceProject" not in out_without
def test_deterministic_zero_llm():
"""纯函数:同输入恒同输出(无网络无 LLM)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "t8", "templateId": "generic", "brief": "稳定"}
summary = {"verdict": {"pass": True}, "costRmb": 0.01}
assert R.build_result_out(job, summary, gd) == R.build_result_out(job, summary, gd)
# ──────────────────────────────────────────────────────────────────────────────
# M3b U2:trace 9d 七项 + D11 首局维(sevenGateVerdict/gatespec)+ repairs 口径 + None 容错
# 守的不变量(镜像 wg1 _extract_trace + 后端 ReadinessScorer 读取口径):
# · 必填七项 camelCase:pass/repairs/wallS/models/attempts/gameId/stage;
# · sevenGateVerdict 只取 {pass,guards}(去 url/ts 噪声,guards.H_progress 保嵌套含 .pass);
# · gatespec 键名必须是 `driver`(落 driverType 则后端 hasDriver=false、firstPlay 永中性);
# · repairs = max(0, attempts-1)(首轮 attempts=1→repairs=0);
# · 成功/失败路都抽 trace;verdictFull/driverType=None 时省略 sevenGateVerdict/gatespec(不塞 None、不崩)。
# ──────────────────────────────────────────────────────────────────────────────
# U1 富化后的成功路 summary(合成;含完整 verdictFull 带 url/ts 噪声 + driverType + stage + models)。
_SUMMARY_SUCCESS = {
"ok": True, "finished": True, "gameId": "amgen-t1", "attempts": 1,
"verdict": {"pass": True, "failedGates": []}, # brief(既有键)
"verdictFull": { # U1 新增:完整 verdict(带噪声 + guards.H_progress 嵌套)
"pass": True, "url": "http://localhost:4320/?x", "ts": 1700000000,
"guards": {
"A_boot": {"pass": True},
"H_progress": {"pass": True, "checks": ["score+"], "latch": True, "state0": 0, "state1": 3},
},
},
"driverType": "tap-targets",
"models": {"code": "MiniMax-M3"},
"stage": "play",
"wallSec": 102.3,
"costRmb": 0.0431,
}
def test_trace_seven_required_fields_and_d11_dims_success():
"""成功路 trace:七项齐 + sevenGateVerdict.guards.H_progress.pass + gatespec.driver;噪声 url/ts 已去。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "u2-ok", "templateId": "generic", "brief": "点击得分小游戏"}
out = R.build_result_out(job, _SUMMARY_SUCCESS, gd)
tr = out["trace"]
# 必填七项(camelCase 逐字镜像 _extract_trace)
assert tr["pass"] is True
assert tr["repairs"] == 0 # attempts=1 → max(0,0)=0(首轮 0)
assert tr["wallS"] == 102.3
assert tr["models"] == {"code": "MiniMax-M3"}
assert tr["attempts"] == 1
assert tr["gameId"] == "amgen-t1"
assert tr["stage"] == "play"
# D11 首局维:sevenGateVerdict 只 {pass,guards},guards.H_progress 嵌套含 .pass(后端 guardPass 读它)
assert set(tr["sevenGateVerdict"].keys()) == {"pass", "guards"}
assert "url" not in tr["sevenGateVerdict"] and "ts" not in tr["sevenGateVerdict"]
assert tr["sevenGateVerdict"]["guards"]["H_progress"]["pass"] is True
# gatespec 键名必须是 driver(落 driverType 会让后端 hasDriver=false)
assert tr["gatespec"] == {"driver": "tap-targets"}
assert "driverType" not in tr["gatespec"]
# cost 维持现状
assert tr["cost"]["totalRmb"] == 0.0431
def test_trace_repairs_equals_max0_attempts_minus_1():
"""repairs = max(0, attempts-1):attempts=1→0、attempts=3→2(对齐 wg1 len(attempts)-1)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "u2-rep", "templateId": "generic", "brief": "x"}
s1 = dict(_SUMMARY_SUCCESS, attempts=1)
assert R.build_result_out(job, s1, gd)["trace"]["repairs"] == 0
s3 = dict(_SUMMARY_SUCCESS, attempts=3)
assert R.build_result_out(job, s3, gd)["trace"]["repairs"] == 2
def test_trace_failed_path_still_has_seven_required():
"""失败路(verdictFull/driverType=None)trace 仍含七项(repairs/attempts/stage/wallS 排障价值)。"""
gd = _make_game_dir(_tmp(), bundle="var notTheGlobal=1;") # 无 __GameBundle → status=failed
job = {"traceId": "u2-fail", "templateId": "generic", "brief": "x"}
summary = {
"ok": False, "finished": False, "gameId": "amgen-t2", "attempts": 6,
"verdict": None, "verdictFull": None, "driverType": None,
"models": {"code": "MiniMax-M3"}, "stage": "code", "wallSec": 50.1, "costRmb": 0.02,
}
out = R.build_result_out(job, summary, gd)
assert out["status"] == "failed"
tr = out["trace"]
for k in ("pass", "repairs", "wallS", "models", "attempts", "gameId", "stage"):
assert k in tr, f"失败路 trace 缺必填项 {k}"
assert tr["repairs"] == 5 # attempts=6 → 5
assert tr["stage"] == "code"
assert tr["pass"] is None # play 没跑
def test_trace_none_tolerance_omits_d11_dims():
"""verdictFull/driverType=None(对照路)→ 省略 sevenGateVerdict/gatespec(不塞 None、不崩)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "u2-none", "templateId": "generic", "brief": "x"}
summary = {
"gameId": "amgen-t3", "attempts": 2, "verdict": None,
"verdictFull": None, "driverType": None,
"models": {"code": "MiniMax-M3"}, "stage": "smoke", "wallSec": 10.0,
}
out = R.build_result_out(job, summary, gd)
tr = out["trace"]
assert "sevenGateVerdict" not in tr # verdictFull=None → 省略
assert "gatespec" not in tr # driverType=None → 省略
assert tr["repairs"] == 1 # attempts=2 → 1
def test_trace_keys_camelcase_match_readiness_scorer():
"""trace 键名逐字镜像 ReadinessScorer 读取口径(拼错会被后端静默接收 → 假绿)。"""
gd = _make_game_dir(_tmp())
job = {"traceId": "u2-keys", "templateId": "generic", "brief": "x"}
tr = R.build_result_out(job, _SUMMARY_SUCCESS, gd)["trace"]
# ReadinessScorer 读取的确切键(camelCase):trace.pass / gatespec.driver /
# sevenGateVerdict.guards.H_progress.pass / repairs / cost.totalRmb
assert "pass" in tr and "gatespec" in tr and "sevenGateVerdict" in tr
assert "wallS" in tr and "gameId" in tr # 非 wall_s / game_id(snake 会被后端读不到)
assert "driver" in tr["gatespec"]
assert "H_progress" in tr["sevenGateVerdict"]["guards"]
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)