games-development-ai/cheap-worker/judge_golden.py

"""judge_golden.py — 玩法地板判定的金标校准包读取器 + 复验脚本(W-AXIS 波2 · 裁定三校准纪律)。

这份解决什么:判定器自己会假阳假阴,裁定三③钉死靠【每品类金标正反例复验 + 创始人抽玩定标】治。本模块是那套
校准纪律的工程落点——把金标语料(fixtures/judge-golden/manifest.json)读进来,对【已标注】的金标跑一次判定、
报与标注不符的漂移(rubric 规范三同构:漂移超线先复采样再回退)。

现状(2026-07-09):五份审计在册的 score-only pass 样本登记为【待标注种子】,expectReject 全为 null——
标注归创始人/主会话亲玩定标,本模块【不预设它们是反例】(双评审明令)。故 check 现在会报「0 已标注金标、
5 待标注」;等亲玩把 expectReject 填上,check 才真正复验漂移。

用法:
  cheap-worker/.venv/bin/python cheap-worker/judge_golden.py list      # 列金标清单(标注状态)
  cheap-worker/.venv/bin/python cheap-worker/judge_golden.py check     # 对已标注金标真跑判定、报漂移(真花判定¥)

load_golden() 是纯函数(读 manifest、解析、把 evidenceDir 解析成绝对路径),可单测、零网络。
"""

import json
import sys
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[1]
_MANIFEST = Path(__file__).resolve().parent / "fixtures" / "judge-golden" / "manifest.json"


def load_golden(manifest_path=None) -> dict:
    """读金标 manifest → {schemaVersion, samples:[{gid, brief, evidenceDir(绝对), expectReject, rejectClasses, ...}]}。

    纯函数:把每条 evidenceDir(相对仓根)解析成绝对路径 + 附 evidenceExists 标记(证据在不在盘上)。
    manifest 缺失/坏 → 返回 {"schemaVersion": None, "samples": []}(绝不抛,列不出就是空清单)。
    """
    p = Path(manifest_path) if manifest_path else _MANIFEST
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
    except Exception as e:  # noqa: BLE001  读不到/坏 → 空清单
        print(f"[judge-golden] manifest 读取失败(按空清单):{type(e).__name__}: {e}", file=sys.stderr)
        return {"schemaVersion": None, "samples": []}
    out = {"schemaVersion": data.get("schemaVersion"), "_note": data.get("_note"), "samples": []}
    for s in (data.get("samples") or []):
        if not isinstance(s, dict) or not s.get("gid"):
            continue
        ev_rel = s.get("evidenceDir") or ""
        ev_abs = (_REPO_ROOT / ev_rel) if ev_rel else None
        out["samples"].append({
            "gid": s.get("gid"),
            "brief": s.get("brief") or "",
            "evidenceDir": str(ev_abs) if ev_abs else None,
            "evidenceExists": bool(ev_abs and ev_abs.is_dir()),
            "expectReject": s.get("expectReject"),            # True/False/None(None=未标注)
            "rejectClasses": s.get("rejectClasses") or [],    # 标注为拒时应命中的类
            "labeledBy": s.get("labeledBy"),
            "labelNote": s.get("labelNote"),
        })
    return out


def labeled_samples(golden: dict) -> list:
    """只取【已标注】(expectReject 非 None)的金标——金标复验只对已定标者跑,未标注种子不参与漂移判定。"""
    return [s for s in (golden.get("samples") or []) if s.get("expectReject") is not None]


def _cmd_list() -> int:
    g = load_golden()
    samples = g.get("samples") or []
    labeled = labeled_samples(g)
    print(f"金标清单(schemaVersion={g.get('schemaVersion')}):{len(samples)} 份,已标注 {len(labeled)} 份,"
          f"待标注 {len(samples) - len(labeled)} 份\n")
    for s in samples:
        state = "待标注" if s["expectReject"] is None else (
            f"应拒[{','.join(s['rejectClasses']) or '?'}]" if s["expectReject"] else "应过")
        ev = "证据在盘" if s["evidenceExists"] else "⚠证据缺失"
        print(f"· {s['gid']:<16} [{state}] {ev}  brief={s['brief'][:36]}")
    if not labeled:
        print("\n(尚无已标注金标——五份为待标注种子,标注归创始人/主会话亲玩定标;check 暂无可复验项。)")
    return 0


def _cmd_check() -> int:
    """对已标注金标真跑判定、报漂移(判定器判定 ≠ 金标标注即漂移)。真花判定¥。"""
    import asyncio

    sys.path.insert(0, str(Path(__file__).resolve().parent))
    import cheap_verify  # noqa: PLC0415

    g = load_golden()
    labeled = labeled_samples(g)
    if not labeled:
        print("无已标注金标可复验(五份待标注种子的 expectReject 均为 null;先由创始人亲玩定标再 check)。")
        return 0

    async def _run():
        drift, total_cost = [], 0.0
        for s in labeled:
            r = await cheap_verify.judge_gameplay_floor(
                s["gid"], brief=s["brief"], evidence_dir=s["evidenceDir"])
            got_reject = not bool(r.get("accepted"))  # 判定拒 = 未 accept
            expect_reject = bool(s["expectReject"])
            cost = r.get("costRmb")
            if isinstance(cost, (int, float)):
                total_cost += cost
            ok = (got_reject == expect_reject)
            tag = "一致" if ok else "★漂移"
            print(f"· {s['gid']:<16} 金标={'拒' if expect_reject else '过'} 判定={'拒' if got_reject else '过'}"
                  f" rejectClasses={r.get('rejectClasses')} degraded={r.get('degraded')} ¥{cost} [{tag}]")
            if not ok:
                drift.append(s["gid"])
        print(f"\n复验完成:{len(labeled)} 份已标注金标,漂移 {len(drift)} 份{('（'+','.join(drift)+'）') if drift else ''},"
              f"判定总成本 ≈¥{round(total_cost, 4)}")
        return 1 if drift else 0

    return asyncio.run(_run())


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "list"
    if cmd == "check":
        sys.exit(_cmd_check())
    sys.exit(_cmd_list())