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

# §6.1 result-out / 契约#6 output.failureReason 七值枚举(对齐后端 FailureReasonEnum,真验改正:
#   原稿用了 content_violation/budget_exceeded/generation_failed 三个不在枚举内的值 → 真 e2e 被后端
#   DifyCallbackTxService L165 拒「failureReason 不在七值内」、回调事务回滚。真七值见 FailureReasonEnum)。
_FAILURE_REASONS = {
    "unsafe_prompt", "intent_unclear", "no_template_match", "config_invalid",
    "llm_error", "timeout", "asset_gen_failed",
}

# 源工程 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 枚举。

    显式入参(在七值内)优先;否则统一归 `llm_error`——契约#6 catch-all(SaaGraphDispatcher 同口径:
    图内部真因[九门不过 / 未收敛 / 预算熔断 / engineBundle 缺]不在七值内时归「生成执行链异常」最近桶,
    真因已落 summary/日志)。注:枚举无 budget/generation 专项值,故不强造、统一 llm_error。
    """
    if explicit and explicit in _FAILURE_REASONS:
        return explicit
    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
    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 {}
    gates_pass = verdict.get("pass") is True
    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