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

#!/usr/bin/env python3
"""batch_run.py —— tier2 0号 spike · 批跑编排(model × brief_variant 矩阵 → JSONL run-records)。

【这份在 spike runbook 里的位置(权威 = G 族图说)】
docs/architecture/架构/生成引擎/tier2细节图说-G-spike-runbook.md:
  - 图 G3「模型矩阵 + 跑序」:0号 spike 不是笼统拿「便宜模型」跑,而是跑一张点名到具体模型的矩阵——
    主力便宜档 deepseek-v4-flash / 强便宜档 deepseek-v4-pro / 中等 agentic 档 MiniMax-M3 / 强基线 Opus·Fable;
    题面用 5 个一句话变体(美食/水果/咖啡/面包/糖水店),每变体每档跑 6 次,5×6 凑够 n≥30;
    **跑序 = 先 M3 证路、再便宜档比成本**(2026-06-21 创始人定:别一上来把整张矩阵 × n≥30 全铺开,
    先用 M3 全流程自治产一款、创始人亲玩判肥鹅味,路走通了再用便宜档各跑 n≥30 比成本)。
  - 图 G4「过门阈值 + 采集字段」:每款一行的 run 级采集字段(RunRecord,见 worker/run_record.py),
    汇成矩阵级三张图(过门率 / ¥/成功款 / 收敛中位数 + fail_system 分布)当 go/no-go 直接输入。

【本模块只做编排,不 re-implement 生成逻辑】
  - 单格生成 = 直接调 run_engine._run(...)(那是把 6 模块接成一条线的总入口的异步函数),
    本模块绝不复制那条生成链;它只负责「矩阵展开 + 跑序 + 落 RunRecord + 断点续跑 + 失败隔离」。
  - 真发 LLM / 真跑 chrome 的部分只在 mini-desktop 跑(6c6g 禁 chrome、无 esbuild、无 agentscope wheel)。
    本机(6c6g)用 --dry-run 走桩 run_one,只校验编排骨架(矩阵展开 / 跑序 / 续跑 / 隔离 / 落盘),不碰网络。

【result → RunRecord 接线(observe-only,不改 run 主链裁决)】
  task A3 的 run_record.py 注释里说「接线交 Phase2 B1/B2」——本模块就是那个接线方:
  从 run_studio 返回的 result(见 agent_loop/studio.py 末尾的 result dict)抽字段,
  按 G4 字段表填一条 RunRecord。其中 fail_stage / fail_system 由 _classify_failure() 从 verdict 纯读派生
  (observe-only:只写进 RunRecord 供退路树分流,绝不回头改 verdict 的 accept/reject 裁决——
  judge 纯代码判的门结果是唯一裁决源,本模块只观测、不裁判,防 Goodhart)。

CLI 用法:
  # mini-desktop(真跑;先 M3 证路一格):
  python batch_run.py --out results/spike-runs.jsonl --models MiniMax-M3 --variants 面包店 --n 1
  # mini-desktop(便宜档比成本,5 变体 × 6 次 = n30):
  python batch_run.py --out results/spike-runs.jsonl \
      --models deepseek-v4-flash deepseek-v4-pro --n 6
  # 6c6g(只验编排骨架,走桩、不碰网络):
  python batch_run.py --out /tmp/dry.jsonl --models MiniMax-M3 deepseek-v4-flash --n 2 --dry-run

可 import 用法:
  from batch_run import run_matrix, BatchConfig
  records = await run_matrix(BatchConfig(out_path="...", models=[...], variants=[...], n=6))
"""

from __future__ import annotations

import argparse
import asyncio
import sys
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Awaitable, Callable, Optional

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

from worker import run_record  # noqa: E402  —— A3 采集记录(纯数据,无 agentscope 依赖,6c6g 可 import)
from worker.run_record import RunRecord, append_record  # noqa: E402

# 结果落点(承袭 run_engine.py 的 RESULTS_DIR 惯例)。
RESULTS_DIR = Path(__file__).resolve().parent / "results"
# 批跑台账默认文件名(矩阵级底料;analyze 侧逐行读)。
DEFAULT_JSONL = RESULTS_DIR / "spike-runs.jsonl"


# ──────────────────────────────────────────────────────────────────────────
# 题面变体(G3 图说钉死的 5 个一句话变体;键 = 变体名,值 = 喂 run 的一句话 brief)
#   每个 brief 都落在 business-sim(经营/合成富游戏 = mini-肥鹅)品类,只换主题皮肤;
#   主题只影响 item id 的 slug 命名与表现层风格,不改三系统耦合结构(那是考题本身,见 G2)。
# ──────────────────────────────────────────────────────────────────────────
BRIEF_VARIANTS: dict[str, str] = {
    "美食": "做一个美食大排档经营游戏,合成食材做成招牌菜交给食客订单赚金币",
    "水果": "做一个水果铺经营合成游戏,合成水果拼成果盘满足顾客订单赚金币",
    "咖啡": "做一个咖啡馆经营游戏,合成咖啡豆与配料调出饮品交订单赚金币",
    "面包": "做一个面包店经营合成游戏,合成原料做成成品交订单赚金币",
    "糖水店": "做一个糖水店经营合成游戏,合成食材熬成糖水甜品交订单赚金币",
}

# 默认题面变体跑序(G3 的 5 变体;CLI --variants 可覆盖)。
DEFAULT_VARIANTS: list[str] = list(BRIEF_VARIANTS.keys())

# ──────────────────────────────────────────────────────────────────────────
# 跑序权威(G3 图说 / 2026-06-21 创始人定):先 M3 证路、再便宜档比成本、强基线只验上限。
#   本模块据此给传入的 models[] 排序,使「先证路」的档先跑——这样路若不通,在更便宜的档上空耗前就能止损。
#   档名对齐 fallback_tree.py 的角色档名常量(同一套档名,真跑改档名两处一起改)。
# ──────────────────────────────────────────────────────────────────────────
_RUN_ORDER_RANK: dict[str, int] = {
    # 0 = 最先(中等 agentic 档,先证路)
    "MiniMax-M3": 0, "M3": 0,
    # 1 = 便宜档比成本(强便宜 → 主力便宜)
    "deepseek-v4-pro": 1, "v4-pro": 1,
    "deepseek-v4-flash": 2, "v4-flash": 2,
    # 3 = 强基线只验上限(最后跑,不进成本评估)
    "Opus": 3, "Fable": 3, "opus": 3, "fable": 3,
}


def _run_order_key(model: str) -> int:
    """按 G3 跑序给模型档定序号;未登记的档排在便宜档与强基线之间(rank=2.5 → 用 25)。"""
    return _RUN_ORDER_RANK.get(model, 25)


# ──────────────────────────────────────────────────────────────────────────
# 失败分类(observe-only):从 verdict 纯读派生 fail_stage / fail_system,喂退路树分流。
#   只读 verdict、不改它;decision=accept(过门)时两者都 None。
#   词表严格对齐 worker/run_record.py 的 FailStage / FailSystem Literal(集成接缝:产单条 → 聚合 → 喂判定器)。
# ──────────────────────────────────────────────────────────────────────────

# 致命门名 → fail_stage 的映射(九门名按 D 族三层校验的段序归类)。
#   九门里:结构/构建段(structure_*)→ build;可运行段(A_boot…I_control)→ seven_gate。
_GATE_TO_STAGE: dict[str, str] = {
    # 结构/构建相关门(bundle 没打出来 / 结构非法)→ build 段
    "structure": "build", "structureOk": "build", "build": "build",
    # 其余九门(boot/render/live/progress/control/wiring…)= 可运行层,统归 seven_gate
}


def _classify_failure(verdict: dict | None,
                      circuit_break: dict | None) -> tuple[Optional[str], Optional[str]]:
    """从 verdict(+ 熔断)纯读派生 (fail_stage, fail_system),observe-only。

    判定顺序(从早段到晚段,取最早命中的失败段——失败定位要落在「最先崩的那道关」):
      1) verdict 为 None / 熔断:抽取或装载即崩、或熔断停 → fail_stage='extract'(产出取不出可用结构);
      2) decision == 'accept' 且 L1.passed → 过门,(None, None);
      3) 九门致命门未过:据门名归 build(结构/构建)或 seven_gate(可运行层);
      4) 富游戏门 tripleLink 未过 → triple_link 段;其子检查 requiresReachable/dagAcyclic 失败
         多由订单或合成数据表错引起 → fail_system 落 order(requiresReachable)/ merge(dagAcyclic);
      5) 富游戏门 economy 未过 → economy 段,fail_system 落 resource(经济枢纽);
      6) 否则按 verdict.reasons / findings 兜底归类,取不到则 (None, None)。

    fail_system 经营品类特有(退路树分流关键依据,G4/G5):
      - presentation(表现层 56%):F 门 wiring calls=0、E_live 没渲染等多在表现层没接输入/没画;
      - merge(合成):dagAcyclic 失败(合成链有环);
      - order(订单):requiresReachable 失败(订单要的物品合不出来);
      - resource(资源/经济枢纽):economy 门失败(金币不动 / 经济不闭环)。

    注:这是 best-effort 启发式归类,不是精确根因——它给退路树一个可分流的方向键,
    真根因仍以 verdict.reasons / harness log 为准。归类绝不影响门裁决(observe-only)。
    """
    # 1) 无 verdict(抽取/装载即崩)或熔断 → extract 段。
    if not verdict:
        # 熔断 stuck(卡死)多发生在表现层没接输入导致门一直不绿 → presentation;其余熔断系统不明。
        sys_hint = "presentation" if (circuit_break or {}).get("kind") == "stuck" else None
        return ("extract", sys_hint)

    l1 = ((verdict.get("layerResults") or {}).get("L1") or {})
    decision = verdict.get("decision")
    # 2) 过门:accept + L1 全绿 → 不失败。
    if decision == "accept" and l1.get("passed"):
        return (None, None)

    # 3) 九门致命门未过:取第一道未过的致命门,据门名归 build / seven_gate。
    for g in (l1.get("gateResults") or []):
        if g.get("passed") is False and g.get("fatal"):
            gate_name = (g.get("gate") or "")
            # 门名可能带前缀(如 A_boot / F_wiring);取下划线前的字母段或整名查映射。
            stage = _GATE_TO_STAGE.get(gate_name)
            if stage is None:
                # 默认归可运行层 seven_gate;F_wiring(接线)失败多在表现层没接输入 → presentation。
                stage = "seven_gate"
            sys_hint = None
            low = gate_name.lower()
            if "wiring" in low or "control" in low or "render" in low or "live" in low:
                sys_hint = "presentation"  # 接线/渲染/活动门多挂在表现层 56%
            return (stage, sys_hint)

    # 4)~5) 富游戏三门(tripleLink / economy / latch)未过:据子检查名定位系统。
    rich = l1.get("richGameGates") or {}
    tl = rich.get("tripleLink")
    if isinstance(tl, dict) and tl.get("passed") is False:
        bad = {c.get("name") for c in (tl.get("checks") or []) if c.get("ok") is False}
        if "requiresReachable" in bad:
            return ("triple_link", "order")   # 订单要的物品合不出来 = 订单/可达性问题
        if "dagAcyclic" in bad:
            return ("triple_link", "merge")    # 合成链有环 = 合成系统问题
        return ("triple_link", "presentation")  # 其余三联动子检查多在表现层没把系统接起来
    econ = rich.get("economy")
    if isinstance(econ, dict) and econ.get("passed") is False:
        return ("economy", "resource")          # 经济门挂 = 资源/经济枢纽不闭环
    latch = rich.get("latch")
    if isinstance(latch, dict) and latch.get("passed") is False:
        # latch 终态没 latch 多因 tick 没接(表现层删了 tick)→ presentation。
        return ("seven_gate", "presentation")

    # 6) 兜底:有 findings 但未落到上面任何门 → 段/系统不明,只标段为 None(让退路树按整体分布兜底)。
    return (None, None)


# ──────────────────────────────────────────────────────────────────────────
# result → RunRecord(G4 字段表落地;observe-only 接线)
# ──────────────────────────────────────────────────────────────────────────
def result_to_record(result: dict, *, run_id: str, brief_variant: str,
                     timestamp: float) -> RunRecord:
    """把 run_studio 返回的 result(见 agent_loop/studio.py 末尾)抽成一条 RunRecord。

    字段抽取口径严格对齐 worker/run_record.py 各字段 docstring 标注的【来源】:
      - model            ← result['model']
      - pass_gate        ← verdict.decision=='accept' 且 L1.passed(judge 纯代码判,零自评)
      - repairs          ← breaker_counters.model_calls 兜底为 0(自治循环代价;真值由 studio attempt 计数,
                           本期 result 未单列 attempt,用 model_calls 作收敛代价的近似观测,注明)
      - fail_stage/system← _classify_failure(verdict, circuit_break)(observe-only,见上)
      - cost_yuan        ← cost.cost_rmb;tokens_*  ← result['tokens'];tokens_by_model ← cost.tokens_by_model
      - wall_seconds     ← result['wall_s']
      - file_count       ← len(result['file_tree']);total_loc ← 对 source_project 各文件 content 计行
      - decision/finished← verdict.decision / result['finished']

    stage 字段:本批跑每格都贯穿设计→build→play 三段,成功落 play;
      未真 finish 的失败款,stage 仍标其最终落点(用 fail_stage 区分中途终止的关序)。
    """
    verdict = result.get("last_verdict") or {}
    l1 = ((verdict.get("layerResults") or {}).get("L1") or {})
    cb = result.get("circuit_break")
    cost = result.get("cost") or {}
    tokens = result.get("tokens") or {}

    decision = verdict.get("decision")
    pass_gate = bool(decision == "accept" and l1.get("passed"))
    fail_stage, fail_system = _classify_failure(verdict if verdict else None, cb)

    # stage:成功款落 play;失败款若有 fail_stage 落在 build 段则标 build,否则也算走到 play(真玩段)。
    stage: str = "play"
    if not pass_gate and fail_stage == "build":
        stage = "build"
    elif not pass_gate and fail_stage == "extract":
        stage = "build"  # 抽取/装载即崩=没走到真玩,落 build 段(extract 在 fail_stage 里另记)

    # total_loc:对 finish 交付的源工程各文件 content 计行(未 finish 则 0)。
    total_loc = 0
    sp = result.get("source_project") or {}
    for f in (sp.get("files") or sp.get("fileTree") or []):
        content = f.get("content") if isinstance(f, dict) else None
        if isinstance(content, str):
            total_loc += content.count("\n") + (1 if content and not content.endswith("\n") else 0)

    return RunRecord(
        run_id=run_id,
        model=result.get("model") or "unknown",
        stage=stage,  # type: ignore[arg-type]  —— Literal,运行期不强校验
        brief_variant=brief_variant,
        pass_gate=pass_gate,
        # repairs:近似取模型调用数作收敛代价观测(真值=studio attempt;result 未单列,注明近似)。
        repairs=int((result.get("breaker_counters") or {}).get("model_calls") or 0),
        fail_stage=fail_stage,  # type: ignore[arg-type]
        fail_system=fail_system,  # type: ignore[arg-type]
        cost_yuan=float(cost.get("cost_rmb") or 0.0),
        tokens_in=int(tokens.get("prompt") or 0),
        tokens_out=int(tokens.get("completion") or 0),
        tokens_cached=int(tokens.get("cached") or 0),
        tokens_by_model=cost.get("tokens_by_model") or {},
        wall_seconds=float(result.get("wall_s") or 0.0),
        file_count=len(result.get("file_tree") or []),
        total_loc=total_loc,
        # 长程一致性三字段:本期 result 未单列(ctx_compressed/checkpoint_recovered 待控制面 trace 落库后接);
        #   先取保守默认(无压缩 / 未恢复 / 幂等无脏),真值由后续观测面补。
        ctx_compressed=False,
        checkpoint_recovered=False,
        idempotency_clean=True,
        decision=decision,  # type: ignore[arg-type]
        finished=bool(result.get("finished")),
        timestamp=timestamp,
        # 编排层 infra 归因(observe-only):从 studio result 抽 infra_flags(缺则空列表,向后兼容旧桩)。
        #   防御式 list(...) 复制,避免与 result 内引用共享;取值 ∈ run_record.INFRA_FLAGS 受控值集。
        infra_flags=list(result.get("infra_flags") or []),
    )


# ──────────────────────────────────────────────────────────────────────────
# 批跑配置
# ──────────────────────────────────────────────────────────────────────────
@dataclass
class BatchConfig:
    """一次矩阵批跑的配置(model × brief_variant × n)。"""

    out_path: str                              # JSONL 台账落点(断点续跑也读它判已跑)
    models: list[str]                          # 模型档列表(会按 G3 跑序重排:先 M3 后便宜档)
    variants: list[str] = field(default_factory=lambda: list(DEFAULT_VARIANTS))  # 题面变体(默认 5 个)
    n: int = 6                                 # 每 (model, variant) 格的重复跑次数(G3:5 变体 ×6 = n30)
    game_id_prefix: str = "feie"               # 工程标识前缀(每格 game_id = <prefix>-<model>-<variant>-<i>)
    # 单次 run 的模型参数(透传给 run_engine._run;默认对齐 run_engine.py 的 CLI 默认)。
    max_tokens: int = 16000
    thinking_budget: int = 8000
    # max_iters:None=交 run_studio 运行时读 generation.yaml iteration.writer_max_iters(单源)。
    #   2026-07-04 实证:旧 default=40 显式透传会把 genconfig 短路——阈值批把 writer 提到 60,r3 实跑仍打
    #   「max iteration numbers 40」;middleware 系旋钮(step_cap/墙钟)生效而 writer 不生效,就是这颗双源默认值。
    max_iters: int | None = None
    do_design: bool = True
    template: str = "business-sim"
    dry_run: bool = False                      # True=走桩 run_one(不碰网络/chrome,本机校验编排骨架)
    # ── 批级(单次调用=单轮)护栏(切片二 U2;默认 None=不限,保持旧行为)──
    #   作用域 = 本次 run_matrix 调用累计;跨轮(n≤20)由 assert_under_round_cap 另管。达限 fail-closed 抛、已跑款不丢。
    batch_rmb_cap: float | None = None         # 单轮累计 ¥ 上限(收敛环 directional v1=20.0)
    batch_wall_cap_s: float | None = None      # 单轮累计墙钟秒上限(directional v1=7200)
    batch_tokens_cap: int | None = None        # 单轮累计 tokens_out 上限(directional v1=600_000)


def _slug(s: str) -> str:
    """把模型名/变体名压成 game_id 安全的 slug(去掉路径不安全字符,中文保留为占位 hash)。"""
    safe = "".join(ch if (ch.isalnum() or ch in "-_") else "" for ch in s)
    if safe:
        return safe
    # 全是中文/特殊字符(如变体名「面包」)→ 用短 hash 占位,保证 game_id 唯一且文件系统安全。
    import hashlib
    return "v" + hashlib.md5(s.encode("utf-8")).hexdigest()[:6]


def _cell_run_id(prefix: str, model: str, variant: str, i: int) -> str:
    """一格内第 i 次跑的 run_id(行级主键 + game_id 复用同一串;断点续跑据它判已跑)。"""
    return f"{prefix}-{_slug(model)}-{_slug(variant)}-{i}"


def _load_done_run_ids(out_path: str) -> set[str]:
    """读已有 JSONL 台账,取已完成的 run_id 集合(断点续跑:已有 run_id 跳过)。

    台账不存在 → 空集(首次跑)。坏行(JSON 解析失败)跳过不计、不中断(best-effort 续跑)。
    """
    done: set[str] = set()
    p = Path(out_path)
    if not p.exists():
        return done
    for line in p.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            rec = RunRecord.from_jsonl_line(line)
            done.add(rec.run_id)
        except Exception:  # noqa: BLE001 —— 坏行不阻断续跑;但要让人看见(打印告警)
            print(f"[batch_run] ⚠ 台账坏行已跳过(续跑不计):{line[:80]}", file=sys.stderr)
    return done


# ── 默认单格生成函数:复用 run_engine._run(不 re-implement 生成逻辑)──
async def _default_run_one(*, game_id: str, brief: str, model: str,
                          cfg: BatchConfig) -> dict:
    """调 run_engine._run 真跑一格(真发 LLM + 真跑 chrome,只在 mini-desktop)。

    这里【绝不复制生成链】,只把矩阵格的参数翻译成 run_engine._run 的入参。
    缺 NEWAPI key / 无 chrome 时,run_engine._run 内部会诚实失败(返回结构化失败 result),
    本批跑据 result 落一条 RunRecord(失败也是数据,计入分母),不静默吞、不拖垮全批。
    """
    # 延迟 import:run_engine 顶层会 import agentscope(经 client 装代理旁路),
    #   6c6g 无 agentscope wheel → import 即失败。故只在真跑路径(非 dry-run)才 import,
    #   保证本机 --dry-run 校验骨架时不触发 agentscope import。
    import run_engine  # noqa: E402  —— 仅真跑路径加载
    return await run_engine._run(
        game_id, brief,
        model_name=model, max_tokens=cfg.max_tokens, thinking_budget=cfg.thinking_budget,
        max_iters=cfg.max_iters, do_design=cfg.do_design, template=cfg.template,
    )


# ── 桩单格生成函数(--dry-run 用:不碰网络/chrome,产一个形状合法的假 result)──
async def _stub_run_one(*, game_id: str, brief: str, model: str,
                       cfg: BatchConfig) -> dict:
    """dry-run 桩:产一个形状对齐 run_studio result 的假结果,供本机校验编排骨架(矩阵/跑序/续跑/隔离/落盘)。

    桩按 game_id 末位数字制造可预测的多样性:偶数=过门(accept),奇数=失败(轮流挂不同门),
    使聚合与退路树在 dry-run 下也能跑出非平凡的分组结果(便于人核对编排正确性)。
    绝不发任何网络/子进程——纯内存造数。
    """
    # 用 game_id 末位与 model 名制造确定性的桩结果(可复现、可核对)。
    last = game_id[-1]
    seed = (sum(ord(c) for c in game_id)) % 4
    await asyncio.sleep(0)  # 让出事件循环(模拟 await 点,使并发语义与真跑一致)

    if seed == 0:
        # 过门款(accept + L1 全绿)。
        verdict = {"decision": "accept",
                   "layerResults": {"L1": {"passed": True, "gateResults": [], "richGameGates": {}}}}
        finished, src = True, {"files": [{"path": "src/main.js", "content": "x\n" * 40}]}
    elif seed == 1:
        # 失败款:三联动门 requiresReachable 挂 → order 系统。
        verdict = {"decision": "fix",
                   "layerResults": {"L1": {"passed": False, "gateResults": [],
                       "richGameGates": {"tripleLink": {"passed": False,
                           "checks": [{"name": "requiresReachable", "ok": False}]}}}}}
        finished, src = False, {}
    elif seed == 2:
        # 失败款:F_wiring 致命门挂 → 表现层 presentation。
        verdict = {"decision": "fix",
                   "layerResults": {"L1": {"passed": False,
                       "gateResults": [{"gate": "F_wiring", "passed": False, "fatal": True,
                                        "detail": "calls=0 输入没接进 core"}],
                       "richGameGates": {}}}}
        finished, src = False, {}
    else:
        # 失败款:经济门挂 → resource 枢纽。
        verdict = {"decision": "fix",
                   "layerResults": {"L1": {"passed": False, "gateResults": [],
                       "richGameGates": {"economy": {"passed": False,
                           "checks": [{"name": "bankruptPathLose", "ok": False}]}}}}}
        finished, src = False, {}

    return {
        "game_id": game_id, "model": model,
        "design_text": "[dry-run 桩]", "writer_final_text": "[dry-run 桩]",
        "source_project": src, "finished": finished,
        "last_verdict": verdict, "circuit_break": None,
        "breaker_counters": {"tool_calls": 3 + seed, "model_calls": 4 + seed},
        "tokens": {"prompt": 100000 + seed * 1000, "completion": 30000, "cached": 5000},
        "cost": {"cost_rmb": round(1.0 + seed * 0.7, 5),
                 "tokens_by_model": {model: {"in": 100000, "out": 30000, "cached": 5000}},
                 "pricingSource": "dry-run-stub"},
        "trace": {"traceId": game_id, "steps": 0, "dropped": 0},
        "wall_s": round(60.0 + seed * 15, 1),
        "file_tree": [f["path"] for f in (src.get("files") or [])],
    }


# 单格生成函数的类型(可注入:真跑=_default_run_one,dry-run=_stub_run_one,测试可传自定义桩)。
RunOneFn = Callable[..., Awaitable[dict]]


# ──────────────────────────────────────────────────────────────────────────
# 护栏异常 + 跨轮机器闸(切片二 U2;fail-closed,§6.10 规则编译成机器门)
# ──────────────────────────────────────────────────────────────────────────
class BatchBudgetExceeded(Exception):
    """单轮(本次 run_matrix 调用)累计 ¥/墙钟/token 超 BatchConfig 上限 → 硬停。已跑款已落盘、不丢。"""


class RoundCapExceeded(Exception):
    """跨轮累计已达 n≤20 上限(N5 收敛环最多 4 轮 × 5)→ 拒再开跑,应停环转退路树。"""


def assert_under_round_cap(out_path: str, *, max_total: int = 20) -> None:
    """跨轮机器闸:开跑前读台账总记录数,已达/超 max_total 即抛 RoundCapExceeded、拒跑。

    把"收敛环不滑向无限追加(n≤20)"从人工纪律编译成机器门(Opus M4 / §6.10)。
    台账不存在 = 0 条 = 放行(首轮)。坏行不计(best-effort,与 _load_done_run_ids 同口径)。
    """
    p = Path(out_path)
    if not p.exists():
        return
    total = sum(1 for line in p.read_text(encoding="utf-8").splitlines() if line.strip())
    if total >= max_total:
        raise RoundCapExceeded(
            f"台账 {out_path} 已 {total} 条 ≥ 上限 {max_total}(n≤20):应早已停环转退路树,拒再开跑。")


# ──────────────────────────────────────────────────────────────────────────
# 主编排:run_matrix —— 展开矩阵、按跑序串跑、落 RunRecord、断点续跑、失败隔离
# ──────────────────────────────────────────────────────────────────────────
async def run_matrix(cfg: BatchConfig, *, run_one: Optional[RunOneFn] = None) -> list[RunRecord]:
    """跑一张 model × brief_variant × n 矩阵,每格落一条 RunRecord 进 JSONL,返回本次新落的记录列表。

    编排语义(对齐 G3 图说 + task 要求):
      - 跑序:models[] 先按 G3 跑序重排(先 M3 证路、再便宜档、强基线最后),外层 model、中层 variant、内层 i。
        串行跑(不并发):真跑要烧 token + 占 chrome/端口,且「先证路再比成本」要求 M3 结果先出来人看;
        失败隔离也更清晰(一格异常只 except 这一格)。
      - 断点续跑:已落台账的 run_id 直接跳过(读 out_path 的已有记录;首次跑=空集)。
      - 失败隔离:单格 run_one 抛异常 → except 住、落一条「异常款」RunRecord(fail_stage=extract)、继续下一格,
        绝不让一格挂拖垮整批(spike 一跑几十格,中途一格崩不能白跑前面的)。
      - 落盘:每格跑完立即 append_record 落盘(append-only),即使中途整批被 kill,已跑的也不丢、可续。

    :param run_one: 单格生成函数(默认按 cfg.dry_run 选 _default_run_one / _stub_run_one;测试可注入桩)。
    :return: 本次实际跑(非跳过)的 RunRecord 列表。
    """
    if run_one is None:
        run_one = _stub_run_one if cfg.dry_run else _default_run_one

    # 校验变体名都在 BRIEF_VARIANTS 里(防把不存在的变体名传进来导致 brief 缺失)。
    unknown = [v for v in cfg.variants if v not in BRIEF_VARIANTS]
    if unknown:
        raise ValueError(
            f"未知题面变体 {unknown};合法变体 = {list(BRIEF_VARIANTS.keys())}(见 G3 图说 5 变体)。")

    # 跑序:模型档按 G3 跑序排(先证路后比成本);变体与 i 保持给定顺序。
    models_ordered = sorted(cfg.models, key=_run_order_key)
    done = _load_done_run_ids(cfg.out_path)
    print(f"[batch_run] 矩阵 = {len(models_ordered)} 档 × {len(cfg.variants)} 变体 × n{cfg.n} "
          f"= {len(models_ordered) * len(cfg.variants) * cfg.n} 格;"
          f"已跑 {len(done)} 格(断点续跑跳过);dry_run={cfg.dry_run};跑序={models_ordered}",
          file=sys.stderr)

    new_records: list[RunRecord] = []
    cell_idx = 0
    total_cells = len(models_ordered) * len(cfg.variants) * cfg.n
    # 批级护栏累计器(本次调用=单轮;切片二 U2):每格落盘后累加,超任一非 None 上限 fail-closed。
    cum_rmb = 0.0
    cum_wall = 0.0
    cum_tokens = 0
    for model in models_ordered:                 # 外层:模型档(先 M3 证路)
        for variant in cfg.variants:             # 中层:题面变体
            brief = BRIEF_VARIANTS[variant]
            for i in range(cfg.n):               # 内层:重复 n 次
                cell_idx += 1
                run_id = _cell_run_id(cfg.game_id_prefix, model, variant, i)
                if run_id in done:
                    print(f"[batch_run] ({cell_idx}/{total_cells}) 跳过已跑 {run_id}", file=sys.stderr)
                    continue
                ts = time.time()
                print(f"[batch_run] ({cell_idx}/{total_cells}) 跑 {run_id} "
                      f"model={model} variant={variant} …", file=sys.stderr)
                try:
                    result = await run_one(game_id=run_id, brief=brief, model=model, cfg=cfg)
                    rec = result_to_record(result, run_id=run_id, brief_variant=variant, timestamp=ts)
                except Exception as e:  # noqa: BLE001 —— 失败隔离:一格挂不拖垮全批,落异常款继续
                    # 异常款:落一条标记 extract 段失败的 RunRecord(不静默吞;异常也是 spike 数据)。
                    print(f"[batch_run] ✗ {run_id} 异常已隔离(落异常款,继续):"
                          f"{type(e).__name__}: {e}", file=sys.stderr)
                    rec = RunRecord(
                        run_id=run_id, model=model, stage="build", brief_variant=variant,
                        pass_gate=False, repairs=0,
                        fail_stage="extract", fail_system=None,
                        decision=None, finished=False, timestamp=ts,
                    )
                # 每格跑完立即落盘(append-only;中途被 kill 也不丢已跑的)。
                append_record(cfg.out_path, rec)
                new_records.append(rec)
                done.add(run_id)
                _kind = "PASS" if rec.pass_gate else f"FAIL({rec.fail_stage}/{rec.fail_system})"
                print(f"[batch_run] ({cell_idx}/{total_cells}) ✓ {run_id} → {_kind} "
                      f"cost=¥{rec.cost_yuan} wall={rec.wall_seconds}s", file=sys.stderr)
                # 批级护栏(切片二 U2):累加本格 → 超任一非 None 上限即 fail-closed 抛(已跑款已落盘、不丢)。
                cum_rmb += rec.cost_yuan
                cum_wall += rec.wall_seconds
                cum_tokens += rec.tokens_out
                _over: list[str] = []
                if cfg.batch_rmb_cap is not None and cum_rmb >= cfg.batch_rmb_cap:
                    _over.append(f"¥{cum_rmb:.2f}≥{cfg.batch_rmb_cap}")
                if cfg.batch_wall_cap_s is not None and cum_wall >= cfg.batch_wall_cap_s:
                    _over.append(f"墙钟{cum_wall:.0f}s≥{cfg.batch_wall_cap_s}")
                if cfg.batch_tokens_cap is not None and cum_tokens >= cfg.batch_tokens_cap:
                    _over.append(f"tokens_out{cum_tokens}≥{cfg.batch_tokens_cap}")
                if _over:
                    raise BatchBudgetExceeded(
                        f"单轮批级上限触发({'; '.join(_over)});已落 {len(new_records)} 款不丢,fail-closed 停批。")

    print(f"[batch_run] 本次新跑 {len(new_records)} 格,落 → {cfg.out_path}", file=sys.stderr)
    return new_records


# ──────────────────────────────────────────────────────────────────────────
# CLI
# ──────────────────────────────────────────────────────────────────────────
def main() -> None:
    ap = argparse.ArgumentParser(
        description="tier2 0号 spike 批跑编排(model × brief_variant 矩阵 → JSONL run-records)")
    ap.add_argument("--out", default=str(DEFAULT_JSONL),
                    help=f"JSONL 台账落点(默认 {DEFAULT_JSONL};断点续跑也读它判已跑)")
    ap.add_argument("--models", nargs="+", required=True,
                    help="模型档列表(会按 G3 跑序重排:先 M3 后便宜档);如 MiniMax-M3 deepseek-v4-flash")
    ap.add_argument("--variants", nargs="+", default=DEFAULT_VARIANTS,
                    help=f"题面变体(默认 5 个:{DEFAULT_VARIANTS});合法值见 G3 图说")
    ap.add_argument("--n", type=int, default=6, help="每 (model,variant) 格的重复跑次数(G3:5×6=n30)")
    ap.add_argument("--prefix", default="feie", help="game_id 前缀(默认 feie)")
    ap.add_argument("--max-tokens", type=int, default=16000)
    ap.add_argument("--thinking-budget", type=int, default=8000)
    ap.add_argument("--max-iters", type=int, default=None,
                    help="单写 ReAct 最大轮数;缺省 None=读 generation.yaml iteration.writer_max_iters(单源,勿再写死)")
    ap.add_argument("--no-design", action="store_true", help="跳过阶段 1 设计")
    ap.add_argument("--template", default="business-sim", help="品类模板(默认 business-sim=mini-肥鹅)")
    ap.add_argument("--dry-run", action="store_true",
                    help="走桩(不碰网络/chrome/agentscope,本机校验编排骨架);6c6g 用它")
    # ── 护栏(切片二 U2;默认 None=不限)──
    ap.add_argument("--batch-rmb-cap", type=float, default=None, help="单轮累计 ¥ 上限(收敛环 20.0)")
    ap.add_argument("--batch-wall-cap-s", type=float, default=None, help="单轮累计墙钟秒上限(收敛环 7200)")
    ap.add_argument("--batch-tokens-cap", type=int, default=None, help="单轮累计 tokens_out 上限(收敛环 600000)")
    ap.add_argument("--round-cap", type=int, default=None,
                    help="跨轮闸:台账总记录数达此值即拒跑(收敛环 n≤20 传 20)")
    args = ap.parse_args()

    cfg = BatchConfig(
        out_path=args.out, models=args.models, variants=args.variants, n=args.n,
        game_id_prefix=args.prefix, max_tokens=args.max_tokens,
        thinking_budget=args.thinking_budget, max_iters=args.max_iters,
        do_design=not args.no_design, template=args.template, dry_run=args.dry_run,
        batch_rmb_cap=args.batch_rmb_cap, batch_wall_cap_s=args.batch_wall_cap_s,
        batch_tokens_cap=args.batch_tokens_cap,
    )
    # 跨轮闸(切片二 U2;给了 --round-cap 才查):台账已达上限即拒跑、非 0 退出,应停环转退路树。
    if args.round_cap is not None:
        try:
            assert_under_round_cap(cfg.out_path, max_total=args.round_cap)
        except RoundCapExceeded as e:
            print(f"\n[batch_run] ✗ 跨轮闸拒跑:{e}", file=sys.stderr)
            sys.exit(3)
    try:
        recs = asyncio.run(run_matrix(cfg))
    except BatchBudgetExceeded as e:
        # fail-closed:已跑款已落盘(断点可续);非 0 退出,让 `&&` 链跳过后续 aggregate。
        print(f"\n[batch_run] ✗ 批级护栏触发,已硬停:{e}\n  已落台账 → {cfg.out_path}(断点续跑可读)", file=sys.stderr)
        sys.exit(4)
    n_pass = sum(1 for r in recs if r.pass_gate)
    print(f"\n=== batch_run 完成:本次新跑 {len(recs)} 格,过门 {n_pass} 格 ===")
    print(f"  台账 → {cfg.out_path}")
    print(f"  下一步:python aggregate.py --in {cfg.out_path}(聚合三图 + 退路树判定)")
    # 退出码:0=编排正常跑完(无论过门率高低;过门率判定交 aggregate + fallback_tree)。
    sys.exit(0)


if __name__ == "__main__":
    main()