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

"""
cheap_studio.py — 便宜档 ReAct 主编排（spike 核心；对照源：Node gen.mjs 主循环 + tier2 studio.py:run_studio）。

组装：内置 Agent + ReActConfig + 外层有界 resume 循环（照 studio.py:404-458）。done 由 cheap_toolkit 的 finish
工具触发（工具内复核 check+build 绿）；finish 收敛后收口 stage → smoke → ensure_play_spec → 循环外九门 play。
熔断（CircuitBreakerMiddleware）/ trace（Tier2TraceMiddleware）/ 历史压缩（ContextConfig）经 import 复用 tier2，零重写。

为什么要外层 resume：AgentScope 2.0.2 内置 ReAct 在"模型产出无 tool_call 的纯文本回合"即退出（看一次门没绿就停）。
外层在 agent 停下后，若【没真 finish】且【门没绿】且【还有预算】就带反馈再 reply()——跨 reply 保留 memory = 原地续修。
这一圈正对应 Node gen.mjs 的 while 主循环。

CLI：cheap-worker/.venv/bin/python cheap-worker/cheap_studio.py --id cheap-run1 --brief "一个简单的点击得分小游戏"
"""

import json
import sys
import time
from pathlib import Path

# 跨包 import 兜底 + key 注入（_bootstrap 模块级把 tier2/gen-worker 加进 sys.path）。
sys.path.insert(0, str(Path(__file__).resolve().parent))  # → cheap-worker/（CLI 直跑兼容）
import _bootstrap  # noqa: E402,F401
from cheap_roles import build_system_prompt  # noqa: E402
from cheap_toolkit import CheapSession, build_toolkit  # noqa: E402
import cheap_budget  # noqa: E402  预算两段式同源工厂(CLI 与生产 Service 读同一配置源)
import cheap_run  # noqa: E402
import cheap_verify  # noqa: E402  U-A2：便宜档「丰富度」LLM 验证 agent（非阻塞·只报告）

# tier2 框架层（import config 触发代理旁路 + 加载 agentscope，必须在裸 import agentscope/openai 之前）。
from worker import config  # noqa: E402
from worker.gate_judge import judge_cheap_verdict  # noqa: E402  CLI 回喂:九门判据与生产 RepairMiddleware 同源
from worker.middleware import (  # noqa: E402  breaker 本体经 cheap_budget 同源工厂构造,此处只用 trace + 熔断异常
    Tier2TraceMiddleware, Tier2CircuitBreak,
)
# C1b trace 落库：复用 tier2 observability 的 JSONL sink（cheap-worker 不另造 sink）。
# 非阻塞铁律：导入失败只告警，回落到无 sink 模式，绝不阻断生成主链。
try:
    from observability.trace import make_jsonl_sink as _make_jsonl_sink  # noqa: E402
    _TRACE_SINK_AVAILABLE = True
except Exception as _import_err:
    _TRACE_SINK_AVAILABLE = False
    print(f"[cheap_studio] trace sink import 失败（回落无 sink 模式，不影响生成）：{_import_err}",
          flush=True)
# T5 压缩崩封口(CLI 路同样吃到:压缩失败 fail-open 跳过、不杀生成主链;钉 2.0.2,纪律同 m3_stream_patch)。
from ctx_compress_patch import apply_ctx_compress_patch  # noqa: E402
apply_ctx_compress_patch()
# AgentScope（此时 agentscope 已由 config import 链加载、代理旁路已装）。
from agentscope.agent import Agent, ReActConfig  # noqa: E402
from agentscope.message import UserMsg  # noqa: E402
from agentscope.state import AgentState  # noqa: E402
from agentscope.permission import PermissionContext, PermissionMode  # noqa: E402


def _rec(msg: str) -> None:
    """进度日志走 stderr（保 CLI stdout 末行可作机读 JSON）。"""
    print(f"[cheap_studio] {msg}", file=sys.stderr, flush=True)


def _bypass_state() -> AgentState:
    """无人值守跳过工具 ASK（否则 FunctionTool 默认 check_permissions 返回 ASK 卡死；对 studio.py:_bypass_state）。"""
    return AgentState(permission_context=PermissionContext(mode=PermissionMode.BYPASS))


def _user_msg(text: str) -> UserMsg:
    """2.0.1+ UserMsg 必须带 name（对 studio.py:user_msg）。"""
    return UserMsg(name="user", content=text)


def _verdict_brief(v) -> dict:
    """从九门 verdict.json 抽 {pass, failedGates}。"""
    if not isinstance(v, dict):
        return {"pass": None, "failedGates": None}
    guards = v.get("guards") or {}
    failed = [k for k, g in guards.items() if isinstance(g, dict) and g.get("pass") is False]
    return {"pass": v.get("pass"), "failedGates": failed}


def _closeout_gates(game_id, *, port, cdp_port) -> dict:
    """CLI 收口门流水线 stage → smoke → ensure_play_spec → 九门 play(一次跑齐;回喂循环与收口段共用)。

    与 Service 路 cheap_gates.run_cheap_gates 同序,多回传 staged/smoke_ok/driver_type 供 run-summary 组装
    (Service 路这些由 collector/driver 另采)。起 play 前清残留 verdict(红线③,镜像 cheap_gates.py:28):
    serve-and-play.sh 早退/超时不写新 verdict 也不删旧的,不清则读回上一轮绿 verdict → 假绿。
    """
    out = {"staged": False, "smoke_ok": None, "driver_type": None, "verdict": None}
    st = cheap_run.stage(game_id)
    out["staged"] = st["ok"]
    if not st["ok"]:
        _rec(f"stage FAIL: {st['output'][:300]}")
        return out
    sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port)
    out["smoke_ok"] = sm["ok"]
    _rec(f"smoke {'PASS' if sm['ok'] else 'FAIL'}(抓 state 供 play-spec)")
    ps = cheap_run.ensure_play_spec(game_id, sm.get("state"))
    # driver 类型:本轮新产的直接带;复用旧 spec(hash 一致)时读盘补(spec 是判卷单一来源)。
    dt = ps.get("driverType")
    if not dt:
        try:
            spec = json.loads((cheap_run.wg1_game_dir(game_id) / "play-spec.json").read_text(encoding="utf-8"))
            dt = (spec.get("driver") or {}).get("type")
        except Exception:  # noqa: BLE001 —— 读不出只降级 None
            dt = None
    out["driver_type"] = dt
    _rec(f"play-spec {'已产 driver=' + str(dt) if ps.get('wrote') else ps.get('reason', '?')}")
    (cheap_run.wg1_game_dir(game_id) / "evidence" / "verdict.json").unlink(missing_ok=True)  # 红线③:清陈旧 verdict
    pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port)
    out["verdict"] = pr["verdict"]
    return out


def _furthest_stage(*, finished, staged, smoke_ran, played) -> str:
    """据收口流程实际到达的步，判本 run 走到的最远阶段（scaffold/code/stage/smoke/play）。

    取最远到达者：scaffold 恒成功（失败提前返回、不到此）；finished=code 已收敛；staged=esbuild 打包成功；
    smoke_ran=跑过冒烟（无论过否）；played=跑过九门 play（无论过否）。供 9d trace.stage（后端审核台读）。
    """
    if played:
        return "play"
    if smoke_ran:
        return "smoke"
    if staged:
        return "stage"
    if finished:
        return "code"
    return "scaffold"


def build_trace_source(verdict, driver_type, attempts, stage, model) -> dict:
    """组 run-summary 的 9d-trace 源维度（additive 纯函数·可单测；run_studio 收口段调）。

    把收口段已算出但没存进 summary 的维度结构化产出，供 result_out 组 payload["trace"]、
    后端 D11 ReadinessScorer 据此算就绪分。纯字典构造、零外部依赖、同输入恒同输出。

    Args:
        verdict: 完整九门 verdict（含 guards，其中 guards.H_progress 是嵌套对象 {pass,checks,latch,...}）；
                 play 没跑（finished=False / run_gates=False / smoke 失败）时传 None。
        driver_type: ensure_play_spec 产的 driver 类型（ps.get("driverType")）；没产 play-spec 时 None。
        attempts: resume 轮数（int，=summary 既有字段）。本函数接收以构成完整 trace-源契约，但 repairs 由
                  result_out 据 summary.attempts 派生（max(0,attempts-1)），故此处**不回写 attempts**
                  （守「不动既有键」）。
        stage: 本 run 走到的最远阶段（scaffold/code/stage/smoke/play 之一，_furthest_stage 产）。
        model: 终态模型名（便宜档单模型 = _bootstrap.SPIKE_MODEL）。

    Returns:
        dict（仅新增键，供 summary.update）：verdictFull / driverType / models / stage。
    """
    return {
        # 完整 verdict（保留 guards 嵌套，含 H_progress.pass）；play 没跑 → None。U2 在 result_out 内削成 {pass,guards}。
        "verdictFull": verdict if isinstance(verdict, dict) else None,
        # driver 类型（ensure_play_spec 产）；没产 play-spec（对照路 run_gates=False）→ None。
        "driverType": driver_type,
        # 便宜档单模型 → {code: SPIKE_MODEL} map（镜像 wg1 models 多模型 map 形态）。
        "models": {"code": model},
        # 最远阶段。
        "stage": stage,
    }


async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens=16000,
                     port=4320, cdp_port=9222, run_gates=True,
                     system_prompt=None, initial_kick=None, write_whitelist=None, prepare=None,
                     scaffold_template=None, scaffold_desc=None, verify_richness_enabled=True,
                     genre=None):
    """跑便宜档一局生成（scaffold → ReAct 写 src/ → done 门 → 收口 stage+smoke+九门）。返回 run-summary dict。

    genre（W-GENRE 件④，缺省 None=原行为）：品类键（如 'puzzle'）显式透传给丰富度 LLM 评分——同一次
    judge 调用 additive 追加品类扩展条目（分母独立小计落 richness.genre）；None → 按 scaffold_template
    查 GENRE_BY_TEMPLATE 推断；两路都没有/条目缺 → 自动退化为纯通用底座评分。

    run_gates=True（默认，spike 单局）：收口跑全套 stage→smoke→ensure_play_spec→九门 play。
    run_gates=False（generation-only，给 compare_node 对照用）：收口只到 stage+smoke 为止，**不**自动产 play-spec、
        **不**跑内部九门 play —— 由调用方在生成与 play 之间注入金标 spec 再单独 play，把驱动器从对照变量里摘掉
        （Codex C1：run_studio 内部已 play，对照需把生成与 play 拆开）。

    A11 M4 模块重生成复用本编排（同一 ReAct + resume + 熔断 + 三层校验收口，不另造），靠四个可选参切到 modify 态，
    都默认 None=create 原行为（create 路零改动）：
      · system_prompt：None=create 的 build_system_prompt；modify 传 build_modify_system_prompt（只改玩法）。
      · initial_kick：None=create 的「按 brief 造游戏」首条 kick；modify 传「只重写 game-logic.js 实现 intent」。
      · write_whitelist：None=create 不收窄；modify 传 {"game-logic.js"} 把写边界收窄到只许写玩法文件。
      · prepare：None=create 的 scaffold（clone _template）；modify 传「已由上游 scaffold+materialize base 源」
        的 noop（game_dir 已就绪、不要再 scaffold 覆盖掉 base 源）。签名同 scaffold：(game_id)->{ok,output}。
    """
    t0 = time.time()
    _rec(f"model={_bootstrap.SPIKE_MODEL} id={game_id} brief=「{brief}」max_iters={max_iters} max_resumes={max_resumes}")

    # create=scaffold clone 模板(扩模板:scaffold_template 传 per-genre 黄金骨架名,如经营=_template-shop)；modify=上游已预备的 noop
    prep = prepare or (lambda gid: cheap_run.scaffold(gid, scaffold_template))
    sc = prep(game_id)
    if not sc["ok"]:
        return {"ok": False, "gameId": game_id, "fail": "准备失败：" + sc["output"]}
    _rec(f"准备就绪 amgen-{game_id}（{'scaffold clone _template + SAA 信封' if prepare is None else 'modify: base 源已预备'}）")

    session = CheapSession(game_id=game_id)
    toolkit = build_toolkit(session, write_whitelist=write_whitelist)  # modify 收窄到只许写 game-logic.js
    model = _bootstrap.build_cheap_model(max_tokens=max_tokens)
    # 熔断:便宜档预算两段式(W-ARCH②,创始人 2026-07-03 裁决)——软停线 ¥10(越线只许收尾类动作,
    #   on_acting 拦新增生成调用)+ RMB 硬地板 ¥15(×1.5,fail-closed 封死上界)。经 cheap_budget 同源工厂,
    #   与生产 Service(cheap_service_app)读同一 genconfig 配置源(取代旧 hard 单段 ¥10,CLI/生产口径统一)。
    #   取价不可达时 _ensure_pricing 降级为次数闸(既有行为,不静默超支、不静默阻断)。步数/超时仍用 tier2 默认。
    breaker = cheap_budget.build_cheap_breaker()
    # C1b trace 落库：把生成轨迹按五字段（traceId/step/cost/verdict/timestamp + ext）
    #   追加写到产物目录 games/amgen-<id>/trace.jsonl，供成本对账 / replay 用。
    #   非阻塞铁律：接线异常只告警 + 回落无 sink 模式，绝不阻断生成主链。
    #   为什么落盘路径选产物目录：与 run-summary.json / evidence/ 同目录，随产物一起留存，运维按 gameId 定位直观。
    _trace_sink = None
    if _TRACE_SINK_AVAILABLE:
        try:
            _trace_path = cheap_run.game_dir(game_id) / "trace.jsonl"
            _trace_sink = _make_jsonl_sink(_trace_path)
            _rec(f"trace sink 接线 → {_trace_path}")
        except Exception as _sink_err:
            _rec(f"trace sink 构造异常（只告警，回落无 sink，不阻断生成）：{_sink_err}")
    tracer = Tier2TraceMiddleware(trace_id=game_id, sink=_trace_sink)

    writer = Agent(
        name="cheap-writer",
        system_prompt=system_prompt or build_system_prompt(game_id, scaffold_desc),  # modify 传 build_modify_system_prompt；扩模板传 scaffold_desc
        model=model,
        toolkit=toolkit,
        middlewares=[tracer, breaker],  # trace 外层、熔断内层（列表序=洋葱序）
        state=_bypass_state(),
        react_config=ReActConfig(max_iters=max_iters),
        context_config=config.build_context_config(),  # 历史压缩
    )

    kick = initial_kick or (
        f"请按这个 brief 造一款游戏：「{brief}」。先 read_file 读手册（.agents/skills/littlejs-game-dev.md）"
        "和你的起点 game-logic.js 再动手；核心玩法实现完、check 与 build 都绿了就立即 finish。")
    breaker_tripped = None
    attempts = 0
    closeout = None  # CLI 回喂:最近一次九门收口结果(回喂循环产出,收口段复用、不重复跑门)
    try:
        for attempt in range(max_resumes + 1):
            attempts = attempt + 1
            _rec(f"resume attempt {attempts}/{max_resumes + 1} → writer.reply …")
            await writer.reply(_user_msg(kick))
            # ① 真 finish:CLI 回喂对齐(W-S1 单③,对账发现六④)——收敛判据从「check+build 绿」升为
            #    「九门绿」:finish 后跑九门收口,门未全绿且轮数/预算有余时带 C6 反馈 resume 续修,
            #    使 lab 复验口径与生产 RepairMiddleware(finish 点拦截→跑门→judge→回喂)行为同型。
            if session.finished is not None:
                if not run_gates:
                    _rec("finish 已接受(check+build 绿)→ 收敛(generation-only,不跑九门回喂)")
                    break
                _rec("finish 已接受(check+build 绿)→ 九门收口判(CLI 回喂对齐)")
                closeout = _closeout_gates(game_id, port=port, cdp_port=cdp_port)
                j = judge_cheap_verdict(closeout["verdict"], game_id=game_id,
                                        staged_dir=cheap_run.wg1_game_dir(game_id))
                vb = _verdict_brief(closeout["verdict"])
                _rec(f"九门 play pass={vb['pass']} failedGates={vb['failedGates']}")
                if j.passed:
                    break
                if attempt >= max_resumes:
                    _rec("九门未绿,但 resume 轮数耗尽 → 交尽力产物收口")
                    break
                # 预算判定与生产同口径:软停已触发 / 实测已花越软停线 → 只收尾、不再回喂(裁决:越线只许收尾)。
                budget_out = bool(getattr(breaker, "budget_soft_tripped", False)) or (
                    breaker._rmb_gate_active and breaker.spent_rmb >= breaker.rmb_hard_limit)
                if budget_out:
                    _rec(f"九门未绿,但 ¥ 软停线已到(已花 ¥{breaker.spent_rmb:.4f})→ 只收尾、不再回喂")
                    break
                session.finished = None  # 重置收敛标记,带门反馈续修(跨 reply memory 保留=原地续修)
                _rec(f"九门未绿(failedGates={j.failed_gates})→ 带 C6 反馈回喂续修")
                kick = j.feedback  # C6 结构渲染全文(含 phaseNow/driverType/game-log 信号;与 RepairMiddleware 注入同源)
                continue
            # ② 门已绿但没 finish → 踹一脚让它 finish
            c_ok = bool(session.last_check and session.last_check.get("ok"))
            b_ok = bool(session.last_build and session.last_build.get("ok"))
            if c_ok and b_ok:
                _rec("门绿但未 finish → 踹 finish")
                kick = "check 与 build 都已 PASS。现在直接调 finish 交付（summary 一句话），不要再改。"
                continue
            # ③ resume 预算耗尽 → 停
            if attempt >= max_resumes:
                _rec("resume 预算耗尽，停")
                break
            # ④ 门没绿 + agent 自停 → 带失败反馈再 reply（跨 reply memory 保留=原地续修）
            fb = "尚未跑出绿 check/build。"
            if session.last_check and not session.last_check.get("ok"):
                fb = "上次 check 失败：\n" + (session.last_check.get("output") or "")[:1500]
            elif session.last_build and not session.last_build.get("ok"):
                fb = "上次 build 失败：\n" + (session.last_build.get("output") or "")[:1500]
            _rec(f"门没绿 + agent 自停 → 带反馈再 reply（attempt {attempts}）")
            kick = (f"你刚才停下了，但 check/build 还没全绿——不要放弃。{fb}\n"
                    "请据失败 write_file 针对性修，再 check → build，直到都绿再 finish。先修最关键的错。")
    except Tier2CircuitBreak as e:
        breaker_tripped = {"kind": e.kind, "reason": e.reason}
        _rec(f"熔断 CircuitBreak：kind={e.kind} reason={e.reason}")
    except Exception as e:  # noqa: BLE001  顶层兜底，落 breaker 信息便于诊断
        breaker_tripped = {"kind": None, "reason": f"{type(e).__name__}: {e}"}
        _rec(f"未捕获异常：{type(e).__name__}: {e}")

    # ── 收口:finish 后 stage → smoke → ensure_play_spec → 循环外九门 play ──
    #   CLI 回喂对齐后,run_gates 路的九门收口已在循环内跑过(closeout),此处直接复用最后一轮结果,
    #   不重复跑门;对照路(run_gates=False)保持原「只 stage+smoke」行为零改动。
    finished = session.finished is not None
    staged = False
    smoke_ok = None
    verdict = None
    driver_type = None  # M3b U1:ensure_play_spec 产的 driver 类型(对照路 run_gates=False 不产 → 保持 None)
    if finished:
        if run_gates:
            cg = closeout or _closeout_gates(game_id, port=port, cdp_port=cdp_port)  # 防御:正常路循环内已跑
            staged = cg["staged"]
            smoke_ok = cg["smoke_ok"]
            driver_type = cg["driver_type"]  # M3b U1:driver 类型存进 trace.gatespec.driver(后端 D11 firstPlay 维)
            verdict = cg["verdict"]
        else:
            st = cheap_run.stage(game_id)
            staged = st["ok"]
            _rec(f"stage {'OK' if staged else 'FAIL: ' + st['output']}")
            if staged:
                sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port)
                smoke_ok = sm["ok"]
                _rec(f"smoke {'PASS' if smoke_ok else 'FAIL'}(抓 state 供 play-spec)")
                _rec("generation-only(run_gates=False):跳过 ensure_play_spec + 九门 play,交对照方注入金标 spec 后单独 play")

    tok_in, tok_out = model.usage_sum()  # 成本落盘(R4)：token 台账始终可取(RecordingOpenAIChatModel.records 累计)。
    summary = {
        "ok": finished,
        "gameId": game_id,
        "brief": brief,
        "model": _bootstrap.SPIKE_MODEL,
        "finished": finished,
        "attempts": attempts,
        "check": {"ok": bool(session.last_check and session.last_check.get("ok"))} if session.last_check else None,
        "build": {"ok": bool(session.last_build and session.last_build.get("ok"))} if session.last_build else None,
        "staged": staged,
        "smoke": {"ok": smoke_ok} if smoke_ok is not None else None,
        "verdict": _verdict_brief(verdict) if verdict is not None else None,
        "breaker": breaker_tripped,
        # 成本落盘(R4)：¥ 走 breaker 同口径(new-api quota 折价 cost.compute)；取价不可达时 spent_rmb=0 且
        #   rmbGate=degraded 标识(_ensure_pricing 已打降级日志可查)——不静默超支、不静默阻断。
        "costRmb": round(breaker.spent_rmb, 4),
        "rmbGate": "active" if breaker._rmb_gate_active else "degraded",
        "tokens": {"in": tok_in, "out": tok_out, "total": tok_in + tok_out},
        "doneSummary": session.finished.get("summary") if session.finished else None,
        "wallSec": round(time.time() - t0, 1),
        # C1b trace 收口摘要：{traceId, steps, dropped}，供编排器对账（trace.jsonl 落盘路径已在接线处 _rec）。
        "trace": tracer.summary(),
    }
    # ── M3b U1：additive 富化 9d trace 源维度（verdictFull/driverType/models/stage）──
    #   收口段已算出但没存的维度结构化进 summary，供 result_out 组 payload["trace"]、后端 D11 算分；
    #   全 additive（新增键、不动既有键），对照路（run_gates=False）verdict/driver_type 为 None 时各键自然降级。
    stage = _furthest_stage(finished=finished, staged=staged,
                            smoke_ran=(smoke_ok is not None), played=(verdict is not None))
    summary.update(build_trace_source(verdict, driver_type, attempts, stage, _bootstrap.SPIKE_MODEL))

    # ── U-A2：便宜档「丰富度」LLM 评分（非阻塞·只报告·不进 verdict / 不改 ok 达标）──
    #   架构红线：玩法「丰富不丰富」的校验需大模型能力，绝不写成 code-presence/正则/断言（违反创始人红线）。
    #   故纯走 LLM judge（cheap_verify.verify_richness），结果 additive 写 summary["richness"]：
    #   · 不进 verdict、不改 summary["ok"]/达标判定、不动任何既有键（result_out._build_trace 只读 named 字段，不读 richness）；
    #   · judge 用独立 model 实例，token 不污染生成成本台账（costRmb 仍只算生成）；
    #   · 非阻塞双保险：verify_richness 内部已 try/except 兜底 degraded，这里再包一层，verify 失败绝不影响 run。
    #   只在 run_gates（完整 spike 收口）且 finished（有真产物）时跑；对照路（run_gates=False）零改动、不触发 LLM。
    if verify_richness_enabled and run_gates and finished:
        try:
            # 品类扩展 rubric(W-GENRE 件④):显式 genre 参数优先,否则按 scaffold_template 查
            #   GENRE_BY_TEMPLATE 推断;命中时同一次评分 additive 追加品类条目(分母独立小计落
            #   richness.genre);两路都没有/未登记 → 行为与既有完全一致。
            _genre = genre or (cheap_verify.GENRE_BY_TEMPLATE.get(scaffold_template) if scaffold_template else None)
            summary["richness"] = await cheap_verify.verify_richness(game_id, brief=brief, genre=_genre)
            rv = summary["richness"]
            _rec(f"丰富度 LLM 评分 score={rv.get('score')}/{rv.get('max')} groups={rv.get('groups')} "
                 f"genre={(rv.get('genre') or {}).get('key')}:{(rv.get('genre') or {}).get('score')} degraded={rv.get('degraded')}")
        except Exception as e:  # noqa: BLE001  非阻塞铁律：richness 绝不影响生成主链
            summary["richness"] = {"score": None, "degraded": True,
                                   "reason": f"richness 接线异常：{type(e).__name__}: {e}"}
            _rec(f"丰富度评分接线异常（已降级，不影响 run）：{type(e).__name__}: {e}")

    ev_dir = cheap_run.game_dir(game_id) / "evidence"
    ev_dir.mkdir(parents=True, exist_ok=True)
    (ev_dir / "run-summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    _rec(f"run-summary 落盘 → {ev_dir / 'run-summary.json'}  wallSec={summary['wallSec']}")
    # 注：n=5 收敛环批跑（RunRecord/batch_run）是 plan deferred；spike 单局 run-summary 已含等价字段。
    return summary


if __name__ == "__main__":
    import argparse
    import asyncio

    ap = argparse.ArgumentParser(description="便宜档 ReAct 主编排（spike 单局）")
    ap.add_argument("--id", default="cheap-run1")
    ap.add_argument("--brief", default="一个简单的点击得分小游戏")
    ap.add_argument("--max-iters", type=int, default=40)
    ap.add_argument("--max-resumes", type=int, default=6)
    ap.add_argument("--max-tokens", type=int, default=16000)
    a = ap.parse_args()
    result = asyncio.run(run_studio(a.id, a.brief, max_iters=a.max_iters,
                                    max_resumes=a.max_resumes, max_tokens=a.max_tokens))
    # stdout 末行单行 JSON（U6 对照 shell-out 据此解析）。
    print(json.dumps(result, ensure_ascii=False))
    sys.exit(0 if result.get("ok") else 1)