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

"""result_out.py — 把 run_studio 产物组成 §6.1 result-out(DifyCallbackReqVO 形态)。

便宜档 M3a U1。严格对齐仓内权威 `SaaGraphDispatcher.buildCallbackReqVO`(game-cloud)
与 Python 参考 `wg1/gen-worker/worker/service.py:build_callback_payload`——三字段语义分离:

  · gameConfig  = 最小占位 Map {templateId,title,theme:"generic",engineDriven:true}
                  (engine 路游戏逻辑全在 engineBundle,gameConfig 仍不可空:GamePackage required);
  · engineBundle = bundle.iife.js 全文(含 __GameBundle)= 运行时承重产物
                   (后端 resolveEngineBundleText 取值落 GamePackage.engineBundle 入 feed);
  · sourceProject = 源工程 2.0 JSON 字符串 = best-effort 可选源血缘(additive,缺则后端源落库旁路)。

成功门:verdict.pass ∧ engineBundle 含 __GameBundle,否则 status=failed + 七值 failureReason。
纯函数(吃 job + run-summary + 产物目录,出 dict),无网络无 LLM、同输入恒同输出——可单测。
"""

import hashlib
import json
from pathlib import Path

import cheap_verify  # W-AXIS-V2 波1:四门投影(project_floor)= 便宜档预筛权威的唯一产地

# §6.1 result-out / 契约#6 output.failureReason 枚举(对齐后端 FailureReasonEnum,真验改正:
#   原稿用了 content_violation/budget_exceeded/generation_failed 三个不在枚举内的值 → 真 e2e 被后端
#   DifyCallbackTxService L165 拒「failureReason 不在枚举内」、回调事务回滚。真枚举见 FailureReasonEnum)。
#   WU2 §3.9 起新增 quota_exhausted(new-api per-user 额度耗尽/池空,网关 402/403;跨契约共享枚举同批落
#   aigc.yaml/dify-workflow-io.json/FailureReasonEnum 三处)。
_FAILURE_REASONS = {
    "unsafe_prompt", "intent_unclear", "no_template_match", "config_invalid",
    "llm_error", "timeout", "asset_gen_failed", "quota_exhausted",
}


def newapi_error_status_text(exc) -> tuple:
    """从一个 new-api 调用异常里尽力抽出 (http_status, 错误体文本),供 classify_newapi_failure_reason 分辨(§3.9)。

    兼容 openai SDK 的 APIStatusError(.status_code / .response.status_code / .body / .message)与被上层框架
    (AgentScope)包装后的普通异常(只剩 str)。抽不到 status 返 None,文本恒为 str(exc) 兜底关键词匹配。
    """
    status = getattr(exc, "status_code", None)
    if status is None:
        status = getattr(exc, "code", None)
    if status is None:
        resp = getattr(exc, "response", None)
        status = getattr(resp, "status_code", None) if resp is not None else None
    parts = []
    for attr in ("message", "body"):
        v = getattr(exc, attr, None)
        if v:
            parts.append(str(v))
    parts.append(str(exc))
    return status, " ".join(parts)


def classify_newapi_failure_reason(status=None, text: str = "") -> str:
    """据 new-api 真实响应把网关非 2xx 分辨成失败因(WU2 §3.9)。

    返回 'quota_exhausted'(额度耗尽,干净失败、不可重试)或 'llm_error'(凭据失效/其他,可重试、不塌成额度耗尽)。

    判据(2026-07-08 真机耗尽探测坐实,取代旧的纯状态码假设):new-api 对**额度耗尽返 HTTP 401**、
    message 含「该令牌额度已用尽 !token.UnlimitedQuota && token.RemainQuota = 0」——**401 与凭据失效状态码重叠**,
    故必须按错误体 message 分辨、不能只看状态码(旧代码 401→llm_error 会把耗尽误判成可重试 → 反复重试到
    watchdog 超时,而非干净"额度用尽")。所以:① message 命中额度关键词 → quota_exhausted(优先,不看状态码);
    ② 否则 HTTP 402/403(one-api 惯例额度码,不与凭据失效重叠)→ quota_exhausted;③ 其余(401 无额度关键词
    = 真凭据失效 / 其他)→ llm_error 可重试。
    """
    try:
        s = int(status) if status is not None else None
    except (TypeError, ValueError):
        s = None
    t = (text or "").lower()
    # 额度耗尽关键词(实测 new-api 中文 "该令牌额度已用尽" + 错误体内联判定条件 "RemainQuota = 0";兼容 one-api 英文/通用)。
    quota_markers = (
        "额度已用尽", "额度已用完", "额度不足", "令牌额度", "余额不足",   # new-api/one-api 中文
        "remainquota = 0", "remainquota=0",                            # new-api 错误体内联条件文本
        "insufficient user quota", "insufficient quota", "quota not enough",  # one-api 英文
    )
    if any(m in t for m in quota_markers):
        return "quota_exhausted"          # ① message 优先:catch new-api 真实的 401+额度耗尽
    if s in (402, 403):
        return "quota_exhausted"          # ② one-api 惯例额度码(与凭据失效 401 不重叠,保守判耗尽)
    return "llm_error"                    # ③ 401 无额度关键词=真凭据失效 / 其他 → 可重试

# 源工程 2.0 profile 必填三枚举(contracts/agent-loop/source-project.schema.json:39)。
_REQUIRED_PROFILE_KEYS = ("tickModel", "inputModel", "progressModel")


def _valid_profile(profile) -> bool:
    """profile 是否含齐三枚举且非空——派生不出即视为无效(省略 sourceProject、不伪造)。"""
    return isinstance(profile, dict) and all(profile.get(k) for k in _REQUIRED_PROFILE_KEYS)


def _source_hash(files: dict) -> str:
    """规范化 files(按路径排序的紧凑 JSON)的 sha256——源可寻址/幂等键(schema 2.0 sourceHash)。"""
    canon = json.dumps(files, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canon.encode("utf-8")).hexdigest()


def build_source_project(game_dir, profile, entry: str = "entry.js") -> str | None:
    """best-effort 组源工程 2.0 JSON 字符串(additive 可选源血缘)。

    读 amgen-<id>/src/*.js 组 files、算 sourceHash、填 profile;**profile 三枚举不全 / src 空 → 返 None**
    (省略 sourceProject,后端源落库旁路、不阻断成功,绝不为凑 schema 伪造 profile)。

    Args:
        game_dir: amgen-<id>/ 目录(读其下 src/)。
        profile: {tickModel,inputModel,progressModel};上游从 play-spec/品类派生,派生不出传 None。
        entry: esbuild 入口相对路径(默认 entry.js)。

    Returns:
        合法 2.0 源工程 JSON 字符串,或 None(无法组出合法 2.0 时)。
    """
    src_dir = Path(game_dir) / "src"
    if not src_dir.is_dir():
        return None
    files = {}
    for p in sorted(src_dir.glob("*.js")):
        files[f"src/{p.name}"] = p.read_text(encoding="utf-8")
    if not files or not _valid_profile(profile):
        return None
    source_project = {
        "schemaVersion": "2.0",
        "sourceHash": _source_hash(files),
        "profile": profile,
        "files": files,
        "entry": entry,
        "globalName": "__GameBundle",
    }
    return json.dumps(source_project, ensure_ascii=False)


def _derive_title(brief: str) -> str:
    """从一句话创意派生标题(供后端 meta 段派生 cover/summary);截断防超 META_TITLE_MAX。"""
    brief = (brief or "").strip()
    return brief[:20] if brief else "便宜档小游戏"


def _read_bundle(game_dir: Path) -> str | None:
    """读 amgen-<id>/bundle.iife.js 全文(engineBundle 真值);缺失返 None。"""
    p = Path(game_dir) / "bundle.iife.js"
    return p.read_text(encoding="utf-8") if p.is_file() else None


def _map_failure_reason(summary: dict, explicit: str | None) -> str:
    """把 run-summary 的失败信号映射到契约#6 failureReason 枚举。

    优先级:显式入参(在枚举内)> summary.failureReason(在枚举内)> catch-all `llm_error`。
    summary.failureReason 是 WU2 §3.9 的桥:Service 侧在模型调用抛错时按 one-api 惯例分辨额度耗尽,
    经收口 sidecar → driver._build_summary 透传到 summary,这里据它落成 quota_exhausted(而非含糊 llm_error)。
    其余图内部真因[九门不过 / 未收敛 / 预算熔断 / engineBundle 缺]不在枚举内 → 归「生成执行链异常」最近桶
    (SaaGraphDispatcher 同口径,真因已落 summary/日志);枚举无 budget/generation 专项值,不强造、统一 llm_error。
    """
    if explicit and explicit in _FAILURE_REASONS:
        return explicit
    svc_reason = (summary or {}).get("failureReason")
    if svc_reason and svc_reason in _FAILURE_REASONS:
        return svc_reason
    return "llm_error"


def _build_trace(summary: dict) -> dict | None:
    """从(U1 富化的)run-summary 组 9d trace(camelCase,镜像 wg1 service._extract_trace;后端 D11
    ReadinessScorer 据此算就绪分)。成功/失败路都抽(失败路 repairs/attempts/stage/wallS 仍有排障价值);
    summary 空(兜底路)→ None(不塞 trace)。

    ⚠️ trace 是后端开放 Map(DifyCallbackReqVO.trace),Jackson **不拒 trace 内层未知键**——键名拼错
       (如 seven_gate_verdict / driverType)会被静默接收、ReadinessScorer 读不到 → 悄悄落中性(假绿)。
       故内层键名必须逐字镜像 ReadinessScorer 读取口径。

    必填七项:pass/repairs/wallS/models/attempts/gameId/stage
      · pass = summary.verdict.pass(brief 既有键);repairs = max(0, attempts-1)(首轮 attempts=1→0,
        对齐 wg1 len(attempts)-1);wallS = summary.wallSec;models/attempts/gameId/stage = U1 富化字段直透。
    D11 首局维(None 容错:verdictFull/driverType 为 None 的对照路 → 省略键,不塞 None):
      · sevenGateVerdict 只取 {pass,guards}(从 verdictFull 去 url/ts 噪声;guards.H_progress 保嵌套含 .pass);
      · gatespec 键名必须是 `driver`(值=driverType;落成 `driverType` 会让后端 hasDriver=false、firstPlay 永中性)。
    similarity:U3 D9 查重产,有则带、缺则省。
    """
    if not summary:
        return None
    attempts = summary.get("attempts")
    # repairs = 自修复轮数 = max(0, attempts-1)(首轮 attempts=1 → repairs=0);attempts 非 int 时降 None。
    repairs = max(0, attempts - 1) if isinstance(attempts, int) else None
    verdict_brief = summary.get("verdict") or {}
    trace = {
        "pass": verdict_brief.get("pass"),
        "repairs": repairs,
        "wallS": summary.get("wallSec"),
        "models": summary.get("models"),
        "attempts": attempts,
        "gameId": summary.get("gameId"),
        "stage": summary.get("stage"),
    }
    # sevenGateVerdict:只 {pass,guards}(去 url/ts 噪声,对齐 _extract_trace:148-154);verdictFull=None → 省略。
    vf = summary.get("verdictFull")
    if isinstance(vf, dict):
        trace["sevenGateVerdict"] = {"pass": vf.get("pass"), "guards": vf.get("guards")}
    # gatespec:键名必须 driver(driverType=None 的对照路 → 省略,不塞 None)。
    driver_type = summary.get("driverType")
    if driver_type is not None:
        trace["gatespec"] = {"driver": driver_type}
    # cost.totalRmb(维持现状,additive 成本台账;顶层不可有 costRmb,会被后端 ObjectMapper 拒)。
    if "costRmb" in summary:
        trace.setdefault("cost", {})["totalRmb"] = summary["costRmb"]
    # similarity(U3 D9 查重产;additive,有则带)。
    similarity = summary.get("similarity")
    if similarity is not None:
        trace["similarity"] = similarity
    # gameplayJudge(W-AXIS 波2):additive 透传玩法地板判定裁决,给观测线区分「玩法判定失败」类目
    #   (后端 GenMetrics/OTLP 消费前 Jackson 静默接收、ReadinessScorer 不读 → 无害;消费后据它补 gate 归因)。
    #   只带轻量裁决字段(不塞完整 checks/截图):accepted 是最终验收权威,rejectClasses 供 gate 桶归因。
    judge = summary.get("judge")
    if isinstance(judge, dict):
        trace["gameplayJudge"] = {
            "accepted": summary.get("accepted"),
            "verdict": judge.get("verdict"),
            "rejectClasses": judge.get("rejectClasses"),
            "degraded": judge.get("degraded"),
            "blocking": judge.get("blocking"),
            "ran": judge.get("ran"),
        }
    # playtest(W-AXIS-V2 波1):additive 透传测试员真玩裁决 + firstPlay 三字段(§5 首局门语义并入,后端
    #   ReadinessScorer.scoreFirstPlay 改读 trace.playtest.firstPlay)。trace.gameplayJudge 双写窗口保留(旧消费面)。
    playtest = summary.get("playtest")
    if isinstance(playtest, dict):
        fp = playtest.get("firstPlay") if isinstance(playtest.get("firstPlay"), dict) else {}
        trace["playtest"] = {
            "accepted": playtest.get("accepted"),
            "verdict": playtest.get("verdict"),
            "degraded": playtest.get("degraded"),
            "rollCount": playtest.get("rollCount"),
            # firstPlay 键名逐字镜像 ReadinessScorer 读取口径(playableAtMs/firstFeedbackMs/loopClosed)。
            "firstPlay": {"playableAtMs": fp.get("playableAtMs"), "firstFeedbackMs": fp.get("firstFeedbackMs"),
                          "loopClosed": fp.get("loopClosed")},
        }
    # floor 四门投影(预筛权威)与 acceptanceVersion:additive,后端可据它区分预筛与最终验收、对账 shadow/v2。
    floor = summary.get("floor")
    if isinstance(floor, dict):
        trace["floor"] = {"pass": floor.get("pass"), "gates": floor.get("gates")}
    av = summary.get("acceptanceVersion")
    if av is not None:
        trace["acceptanceVersion"] = av
    return trace


def build_result_out(job: dict, summary: dict, game_dir, *,
                     failure_reason: str | None = None, source_project: str | None = None) -> dict:
    """据 §6.1 job-in + run-summary + 产物目录组 result-out(DifyCallbackReqVO 形态)。

    Args:
        job: §6.1 job-in(取 traceId/job_id、templateId、brief)。
        summary: run_studio 返回的 run-summary dict(取 verdict.pass 判成败、costRmb 透传)。
        game_dir: 本 run 的 amgen-<id>/ 目录(读 bundle.iife.js)。
        failure_reason: 失败路显式失败原因(可空,缺则从 summary 推)。
        source_project: best-effort 源工程 2.0 JSON 字符串(可空=省略;由 build_source_project 预组)。

    Returns:
        result-out dict(traceId/status/templateId/gameConfig/assets[+engineBundle|+failureReason]
        [+trace.cost][+sourceProject];字段严格限 DifyCallbackReqVO 已知集,后端拒未知字段)。
    """
    trace_id = job.get("traceId") or job.get("job_id")
    template_id = job.get("templateId") or "generic"
    brief = job.get("brief") or ""

    bundle_text = _read_bundle(game_dir)
    verdict = summary.get("verdict") or {}
    vf = summary.get("verdictFull") if isinstance(summary.get("verdictFull"), dict) else verdict
    # W-AXIS-V2 波1:预筛权威 = 四门投影(floor)。三级取值(注意 summary.verdict 是 brief 版、无 guards):
    #   ① summary.floor(run_acceptance 落,生产路)→ 直接用;
    #   ② verdictFull 带 guards(老产物 / W-AXIS 波2 路)→ 现算 project_floor(四门投影,**不复用**九门 verdict.pass);
    #   ③ 完全无 guards(极老产物 / 契约桩只给 verdict.pass)→ 回落 verdict.pass 作预筛(向后兼容、不误拒;无 guards
    #      的假绿在上游 gate_judge/run_acceptance 已拦,build_result_out 是最终契约组装、非验收门)。
    if isinstance(summary.get("floor"), dict):
        floor = summary["floor"]
    elif vf.get("guards"):
        floor = cheap_verify.project_floor(vf)
    else:
        floor = {"pass": bool(verdict.get("pass")), "gates": {}}
    prefilter_pass = bool(floor.get("pass"))
    # W-AXIS 波2:验收权威 = accepted（= 预筛 ∧ 测试 agent 真玩裁决）。summary 无 accepted 键（旧产物 / 对照路 /
    #   验收未跑）→ 回落四门投影（向后兼容，不误拒）；测试员拒 / degraded 时 accepted=False → status=failed。
    accepted = summary.get("accepted")
    gates_pass = (accepted is True) if accepted is not None else prefilter_pass
    engine_ok = bool(bundle_text) and "__GameBundle" in bundle_text
    status = "succeeded" if (gates_pass and engine_ok) else "failed"

    payload = {
        "traceId": trace_id,
        "status": status,
        "templateId": template_id,
        # gameConfig 最小合法占位(engine 路逻辑全在 engineBundle,gameConfig 仍不可空)。
        "gameConfig": {
            "templateId": template_id,
            "title": _derive_title(brief),
            "theme": "generic",
            "engineDriven": True,
        },
        "assets": [],
    }
    if status == "succeeded":
        # 成功路携 bundle 全文 → 后端落 GamePackage.engineBundle 入 feed。
        payload["engineBundle"] = bundle_text
    else:
        # 失败路:后端据 status=failed 把任务置 FAILED;failureReason 供排障(七值枚举)。
        payload["failureReason"] = _map_failure_reason(summary, failure_reason)

    # ── M3b U2:组 9d trace(必填七项 + D11 首局维 sevenGateVerdict/gatespec + cost + similarity)──
    #   camelCase 逐字镜像 wg1 _extract_trace + 后端 ReadinessScorer 读取口径;成功/失败路都抽,summary 空则省略。
    trace = _build_trace(summary)
    if trace:
        payload["trace"] = trace
    # sourceProject best-effort(additive:有则带、缺则省,后端源落库旁路不阻断)。
    if source_project is not None:
        payload["sourceProject"] = source_project
    return payload