games-development-ai/tier2/gen-worker/run_engine.py

#!/usr/bin/env python3
"""run_engine.py —— tier2 富游戏自治生成引擎 · 端到端可跑入口(一句话 brief → 可玩富游戏)。

这是把 6 模块接成一条线的【总入口】。给一句话题面,它驱动:
  阶段 1(可选):设计 agent 把题面 → 富游戏设计稿;
  阶段 2:单写 ReAct agent 在 mini-肥鹅 scaffold 上自治循环——
    scaffold_init(铺平台预建三系统骨架 + 留空数据表)
    → write_source(填 data/datatable.gold.json 真值 + play-scene.js 表现层)
    → build(esbuild 多文件打 Phaser iife)
    → run_gates(真浏览器跑 L1 九门 + 富游戏三门,judge 纯代码产 verdict)
    → read_verdict(看哪门没过)→ 针对性改源 → 再 run_gates …… 直到门全绿调 finish,或四道熔断停。

M3 路:走 new-api 网关的 Anthropic 原生 /v1/messages(thinking 分离);凭据从
  docs/内网凭据与端点.md(NEWAPI_KEY)/ env 读,绝不硬编码(见 worker/client.py)。

运行环境纪律(创始人铁律):
  - 6c6g 禁 chrome:本入口的 run_gates 段需真浏览器,**只在 mini-desktop 真跑**(已装 chrome + esbuild)。
    6c6g 上跑会在 build/run_gates 段返回结构化失败(脚本/工具链缺)而非静默成功——这是诚实失败,不是 bug。
  - 端到端真跑步骤见 tier2/RUN-ON-MINI-DESKTOP.md。

用法:
  # mini-desktop(真跑):
  python run_engine.py feie-001 --brief "做一个面包店经营合成游戏,合成原料做成品交订单赚金币"
  # 6c6g(只验装配,不真跑门;门段会诚实失败):
  python run_engine.py feie-001 --brief "..." --dry-run-note

输出:打印结果摘要 + 落 results/<game_id>.json(含 source_project / verdict / 熔断 / token)。
退出码:0 = 已 finish 交付源工程;1 = 未 finish(门没过到收敛 / 熔断 / 异常)。
"""

import argparse
import asyncio
import json
import sys
from pathlib import Path

# 包内/直跑兼容:本文件在 tier2/gen-worker/run_engine.py;把 gen-worker/ 加进 sys.path 使顶层包 `worker` 可解析。
sys.path.insert(0, str(Path(__file__).resolve().parent))

from worker import archetypes, client, config  # noqa: E402  —— client 先 import(装代理旁路)
from worker.agent_loop.studio import run_studio  # noqa: E402

# 结果落点(承袭 wg1 results/ 惯例;tier2 独立目录)。
RESULTS_DIR = Path(__file__).resolve().parent / "results"


# ── 品类相关物(play_spec / 提示)已收敛进品类注册表 worker/archetypes.py(U2 单一事实源)──
#   下面两个符号保留为**向后兼容薄壳**,直接委托注册表,避免别处旧引用失效、且与注册表零漂移。
#   richGame.{mergeChains,orders} 故意不写死:run_gates 会从 agent 真填的 data/datatable.gold.json 派生
#   (见 worker/run.py _derive_richgame_from_datatable),故三联动门静态校验查的是 agent 的实际表。
def default_business_sim_spec(game_id: str) -> dict:
    """构造 business-sim 品类的默认真玩规格(driver + 经济门双路 + latch + 机制断言)。

    【兼容薄壳】实体已搬进 archetypes.ARCHETYPES['business-sim'].gate_spec_factory;此处委托它,
    保持与注册表逐字节同构(回归安全)。与 tier2/harness/mini-feie.play-spec.example.json 对齐。
    """
    return archetypes.gate_spec_for("business-sim", game_id)


# ── mini-肥鹅 品类约定提示(喂单写 agent;兼容薄壳,实体在注册表 promptHint)──
FIXTURE_HINT_BUSINESS_SIM = archetypes.prompt_hint_for("business-sim")


async def _run(game_id: str, brief: str, *, model_name, max_tokens, thinking_budget,
               max_iters, do_design, template) -> dict:
    """异步主流程:据品类注册表取 play_spec + 提示 → run_studio(U2 泛化:不再硬编码 business-sim)。

    未注册品类 → play_spec=None + 提示空串(诚实跳过富游戏门规格,只走通用九门),不静默当成 business-sim。
    """
    # 据品类注册表取真玩门规格(gateSpec)与品类约定提示(promptHint);未注册返 None/空串。
    play_spec = archetypes.gate_spec_for(template, game_id)
    fixture_hint = archetypes.prompt_hint_for(template)
    return await run_studio(
        game_id, brief, play_spec=play_spec,
        model_name=model_name, max_tokens=max_tokens, thinking_budget=thinking_budget,
        writer_max_iters=max_iters, fixture_hint=fixture_hint, do_design=do_design,
    )


def main() -> None:
    ap = argparse.ArgumentParser(description="tier2 富游戏自治生成引擎 · 端到端入口")
    ap.add_argument("game_id", help="工程标识(决定 workdir = game-runtime/games/_tier2-gen/<game_id>)")
    ap.add_argument("--brief", required=True, help="一句话题面")
    ap.add_argument("--template", default="business-sim",
                    help="品类模板(默认 business-sim = mini-肥鹅 经营/合成富游戏)")
    ap.add_argument("--model", default=None, help="M3 模型名(默认 env TIER2_MODEL / MiniMax-M3)")
    ap.add_argument("--max-tokens", type=int, default=16000, help="输出上限(须 > thinking-budget)")
    ap.add_argument("--thinking-budget", type=int, default=8000, help="thinking 预算")
    ap.add_argument("--max-iters", type=int, default=None,
                    help="单写 ReAct 最大轮数;缺省 None=读 generation.yaml iteration.writer_max_iters(单源)。"
                         "2026-07-04 实证:旧 default=40 显式透传会短路 genconfig(阈值提到 60 而 writer 仍 40)")
    ap.add_argument("--no-design", action="store_true", help="跳过阶段 1 设计")
    ap.add_argument("--dry-run-note", action="store_true",
                    help="只提示:6c6g 上 build/run_gates 段会诚实失败(无 chrome/esbuild),真跑去 mini-desktop")
    args = ap.parse_args()

    if args.dry_run_note:
        print("[run_engine] 注:6c6g 禁 chrome、无 esbuild → build/run_gates 段会返回结构化失败(非静默成功)。"
              "端到端真跑去 mini-desktop,见 tier2/RUN-ON-MINI-DESKTOP.md。", file=sys.stderr)

    # 提前校验 key 可读(缺 key 即抛,尽早暴露而非跑到模型调用才崩;权威来源 docs/内网凭据与端点.md)。
    try:
        client.get_api_key()
    except RuntimeError as e:
        print(f"[run_engine] ✗ {e}", file=sys.stderr)
        sys.exit(2)
    print(f"[run_engine] 模型出口 = {config.client.resolve_base_url()} (Anthropic 原生 /v1/messages)",
          file=sys.stderr)

    res = asyncio.run(_run(
        args.game_id, args.brief,
        model_name=args.model, max_tokens=args.max_tokens, thinking_budget=args.thinking_budget,
        max_iters=args.max_iters, do_design=not args.no_design, template=args.template,
    ))

    # 落结果(含 source_project / verdict / 熔断 / token;供事后对账 + 成本台账)。
    RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    out = RESULTS_DIR / f"{args.game_id}.json"
    out.write_text(json.dumps(res, ensure_ascii=False, indent=2), encoding="utf-8")

    # 摘要(创始人要求紧凑;细节落 results/)。
    cb = res.get("circuit_break")
    v = res.get("last_verdict") or {}
    l1 = ((v.get("layerResults") or {}).get("L1") or {})
    print(f"\n=== tier2 引擎 [{args.game_id}] finished={res['finished']} ===")
    print(f"  model={res['model']} wall={res['wall_s']}s "
          f"tokens(in/out/cached)={res['tokens']['prompt']}/{res['tokens']['completion']}/{res['tokens']['cached']}")
    print(f"  files={len(res.get('file_tree') or [])} verdict.decision={v.get('decision')} "
          f"L1.passed={l1.get('passed')}")
    if cb:
        print(f"  熔断 kind={cb.get('kind')} reason={cb.get('reason')}")
    print(f"  结果 → {out}")
    sys.exit(0 if res["finished"] else 1)


if __name__ == "__main__":
    main()