""" 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 →循环外九门 play。 熔断（CircuitBreakerMiddleware）/ trace（Tier2TraceMiddleware）/ 历史压缩（ContextConfig）经 import 复用 tier2，零重写。 为什么要外层 resume：AgentScope 2.0.3 内置 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 hashlib import json import sys import time import uuid 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, SCAFFOLD_DESC_BY_TEMPLATE # 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, classify_cheap_failure_layer # noqa: E402 CLI 回喂:九门判据与生产 RepairMiddleware 同源;失败三层归因(W-AXIS 波1) 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.3,纪律同 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) # fix③(死圈修复 2026-07-08):brief 落 evidence/brief.json 供根因取证。 # 为什么单独落而不只靠 run-summary.brief:① 早落(scaffold 后即写)——run 硬失败/挂死时 run-summary 收不了口、 # brief.json 已在盘上,取证不断链;② 带 sha256 完整性戳(核 worker 收到的 brief 与后端下发字节一致,查"是不是 brief # 被截断/篡改害的生成翻车");③ 限长脱敏(超 PROMPT_MAX 截断 + truncated 标记),防超长 brief 撑爆落盘。 # 落 evidence/(forensics 目录,不随 src/ 进素材市场交易),绝不落工程根——源工程会被交易,brief 是用户私有输入、不得泄进可交易源。 _BRIEF_PROMPT_MAX = 1000 # brief 落盘限长(字符);超出截断 + truncated=True,sha256 仍按【完整原文】算(截断不掩盖完整性核对) def _persist_brief_forensics(game_id: str, brief: str, source: str = "job.brief") -> None: """把本局 brief 落 evidence/brief.json(取证用,见上方注释)。非阻塞:任何异常只告警、绝不咬生成主链。""" try: text = brief or "" raw = text.encode("utf-8") rec = { "brief": text[:_BRIEF_PROMPT_MAX], # 限长:超长只留前 N 字符(脱敏 + 防撑爆) "bytes": len(raw), # 完整 brief 的 UTF-8 字节数 "sha256": hashlib.sha256(raw).hexdigest(), # 完整 brief 摘要:核 worker↔后端字节一致 "truncated": len(text) > _BRIEF_PROMPT_MAX, # 是否因限长被截 "promptMax": _BRIEF_PROMPT_MAX, "source": source, # 来源溯源(默认 job.brief) } ev_dir = cheap_run.game_dir(game_id) / "evidence" ev_dir.mkdir(parents=True, exist_ok=True) (ev_dir / "brief.json").write_text(json.dumps(rec, ensure_ascii=False, indent=2), encoding="utf-8") _rec(f"brief 落盘取证 → {ev_dir / 'brief.json'} bytes={rec['bytes']} truncated={rec['truncated']}") except Exception as e: # noqa: BLE001 非阻塞铁律:取证落盘绝不影响生成主链 _rec(f"brief 落盘异常(已忽略,不影响生成):{type(e).__name__}: {e}") # fix①-B(重实):missing-param 类 stuck 续修时给 agent 的贴脸补参反馈(比默认 schema 校验错更明确,并催改用 edit_file 短载荷)。 _MALFORMED_RESUME_FEEDBACK = ( "【重要·纠偏】你刚才反复漏了工具调用的必填参数(通常是 path),调用被拒并触发了熔断。" "重发工具调用时【必须一次带全参数】:write_file 要 {path, content}、edit_file 要 {path, old, new}——path 绝不能省。" "小改强烈建议用 edit_file(只发变化的几行、参数短、最不易漏 path),别对整个 game-logic.js 用 write_file" "(内容越长越容易把 path 漏掉)。现在:先 read_file 看清你要改的文件路径,再带【完整参数】重发修改,然后 check → build → finish。" ) 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, interaction_profile_id=None) -> dict: """CLI 收口门流水线 stage → smoke →九门 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, interaction_profile_id=interaction_profile_id, ) out["smoke_ok"] = sm["ok"] _rec(f"smoke {'PASS' if sm['ok'] else 'FAIL'}(抓 state 供 play-spec)") if not sm["ok"]: # smoke 失败真因必须可见(2026-07-04 bake_off 实证:并发争用致 100% FAIL→全场 key-cycle 回退考卷, # 而失败原因被吞在 raw 里,烧掉一轮诊断)——打印 raw 尾供归因。 _rec(f"smoke raw 尾:{(sm.get('raw') or '')[-300:]}") # W-AXIS-V2 波2:ensure_play_spec 已摘——play 以 undriven 硬集(A/B/C/D + F/G/I)跑,floor 投影只取 A/B/C/D; # driver_type 随 tap-targets 驱动器退役恒 None(out 初始化即 None;trace.driverType 观测维不再承 D11 首局信号,已迁 trace.playtest)。 (cheap_run.wg1_game_dir(game_id) / "evidence" / "verdict.json").unlink(missing_ok=True) # 红线③:清陈旧 verdict,防读回上轮绿 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, } def acceptance_v3_mode() -> str: """返回当前验收模式；只有 v3/v3_shadow 才进入可信证据闭环。""" return str(cheap_verify._acceptance_v3_cfg().get("mode") or "v3_shadow") def _new_local_acceptance_trace_id(game_id: str) -> str: """给没有后端 traceId 的直接 CLI 调用生成本次唯一任务标识。 同一 ``run_studio`` 内的唯一 repair 复用已经冻结的 identity，不会再次调用本函数；下一次 CLI 即使复用 gameId 和完全相同的产物，也会得到不同 taskBindingHash，不能命中旧任务证据。 """ return f"studio-{game_id}-{uuid.uuid4().hex}" def _resolve_acceptance_task_trace_id(game_id: str, supplied_trace_id, *, identity_supplied: bool) -> str: """解析当前调用的可信任务标识；外部 identity 没有原始 traceId 时拒绝继续。""" if supplied_trace_id is not None: # 复用 canonical 非空字符串校验，避免编排层和验收层出现两套 traceId 形状规则。 cheap_verify.task_binding_hash_v3(supplied_trace_id) return supplied_trace_id if identity_supplied: raise ValueError("调用方提供 acceptanceIdentity 时必须同时提供当前任务 traceId") return _new_local_acceptance_trace_id(str(game_id)) def build_acceptance_v3_request(game_id: str, brief: str, verdict, *, acceptance_identity, idempotency_key=None, repair_count=None, parent_run_id=None, writer_cost_rmb=0.0, reference_asset_policy_id=None, reference_asset_generation_receipts=None) -> dict: """组 v3 唯一入口请求；只提交本轮 writer 增量，历史成本由验收边界读取封存父决策。""" identity = dict(acceptance_identity or {}) ordinal = int(identity.get("repairOrdinal") or 0) if repair_count is not None and int(repair_count) != ordinal: raise ValueError("repair_count 必须与 acceptanceIdentity.repairOrdinal 一致") policy_present = reference_asset_policy_id is not None receipts_present = reference_asset_generation_receipts is not None if policy_present != receipts_present: raise ValueError("referenceAssetPolicyId/referenceAssetGenerationReceipts 必须成对出现") reference_fields = {} if policy_present: if not isinstance(reference_asset_policy_id, str) or not reference_asset_policy_id: raise ValueError("referenceAssetPolicyId 必须是非空字符串") if (not isinstance(reference_asset_generation_receipts, list) or not reference_asset_generation_receipts or any(not isinstance(receipt, dict) for receipt in reference_asset_generation_receipts)): raise ValueError("referenceAssetGenerationReceipts 必须是非空 JSON 对象数组") for index, receipt in enumerate(reference_asset_generation_receipts): errors = cheap_verify._validate_v3_schema_only( cheap_verify._REFERENCE_ASSET_RECEIPT_SCHEMA, receipt) if errors: raise ValueError( f"referenceAssetGenerationReceipts[{index}] 形状非法:" + ";".join(errors[:4])) # canonical 往返得到与 Writer 前冻结值等价的 plain JSON 副本，避免后续调用方原地篡改。 import reference_asset_gate # noqa: PLC0415 frozen_receipts = json.loads(reference_asset_gate.canonical_json_bytes( reference_asset_generation_receipts).decode("utf-8")) reference_fields = { "referenceAssetPolicyId": reference_asset_policy_id, "referenceAssetGenerationReceipts": frozen_receipts, } return { "gameId": str(game_id), "brief": brief or "", "acceptanceIdentity": identity, "verdict": verdict or {}, "acceptanceMode": acceptance_v3_mode(), "evidenceMode": "native", "idempotencyKey": idempotency_key or f"{game_id}:acceptance-v3", "repairCountAcrossParentChain": ordinal, "parentRunId": parent_run_id, "writerCostRmb": cheap_verify._finite_nonnegative_rmb_v3(writer_cost_rmb, "writerCostRmb"), **reference_fields, } def apply_acceptance_v3(summary: dict, acceptance: dict, *, first_pass: dict | None = None) -> dict: """把 v3 终态单向投影进 run-summary，并保留修复前 factual decision 供批账审计。""" out = dict(summary or {}) acceptance = acceptance if isinstance(acceptance, dict) else {} compatibility = acceptance.get("compatibility") if isinstance(acceptance.get("compatibility"), dict) else {} for key, value in compatibility.items(): if key != "trace": out[key] = value trace = dict(out.get("trace") or {}) trace.update(compatibility.get("trace") or {}) out["trace"] = trace if isinstance(acceptance.get("floor"), dict): out["floor"] = acceptance["floor"] decision = acceptance.get("decision") if isinstance(acceptance.get("decision"), dict) else {} out["acceptanceV3"] = acceptance # result-out 防重放边界只认本次 run-summary 与 sealed v3 的逐字段镜像；旧 accept 不能授权新 job/bundle。 out["acceptanceRunId"] = acceptance.get("runId") out["acceptanceRequestHash"] = acceptance.get("acceptanceRequestHash") out["acceptanceTaskBindingHash"] = acceptance.get("taskBindingHash") out["acceptanceArtifactHash"] = acceptance.get("artifactHash") out["acceptanceBriefHash"] = acceptance.get("briefHash") out["publishFrozen"] = bool(decision.get("publishFrozen", True)) first = first_pass if isinstance(first_pass, dict) else acceptance first_decision = first.get("decision") if isinstance(first.get("decision"), dict) else {} repair_attempted = isinstance(first_pass, dict) if repair_attempted: # 只在真实发生 repair 时保留首轮完整对象；终态权威仍只有 acceptanceV3。 out["acceptanceV3FirstPass"] = first_pass out["firstPassAccepted"] = first_decision.get("outcome") == "accept" and first_decision.get("accepted") is True out["repairAttempted"] = repair_attempted out["acceptedAfterRepair"] = bool(repair_attempted and decision.get("outcome") == "accept" and decision.get("accepted") is True) return out def apply_v3_entry_failure(summary: dict, reason: str, *, mode: str) -> dict: """入口缺 canonical genre/完整 payload 时显式冻结；声明 v3 后禁止回落旧 accepted。""" out = dict(summary or {}) out.update({"acceptanceVersion": mode, "accepted": False, "ok": False, "publishFrozen": True, "failureReason": reason, "repairAttempted": False, "acceptedAfterRepair": False}) out["failureLayer"] = {"layer": "tester_degraded", "reason": reason, "failedGates": []} return out def build_v3_repair_prompt(feedback: str) -> str: """把 v3 硬证反馈包装成一次性 writer 指令；禁止顺手重做，并强制重过 check/build/finish。""" return ((feedback or "v3 已验证拒绝，请按硬证修复玩法闭环") + "\n只修上述硬证指向的问题，不要扩需求；完成后必须重新 check → build → finish。") def next_v3_writer_cost(writer_cost_now: float, writer_cost_accounted: float) -> float: """只返回尚未计入验收的 writer 增量；父链历史成本不再由调用方搬运。""" current = cheap_verify._finite_nonnegative_rmb_v3(writer_cost_now, "writerCostRmb.current") accounted = cheap_verify._finite_nonnegative_rmb_v3( writer_cost_accounted, "writerCostRmb.accounted") return max(0.0, current - accounted) async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens=16000, port=4320, cdp_port=9222, run_gates=True, interaction_profile_id=None, system_prompt=None, initial_kick=None, write_whitelist=None, prepare=None, scaffold_template=None, scaffold_desc=None, verify_richness_enabled=True, genre=None, acceptance_parent_run_id=None, acceptance_repair_count=0, acceptance_brief=None, acceptance_identity=None, acceptance_template_route=None, acceptance_task_trace_id=None, reference_asset_policy_id=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→九门 play（undriven；ensure_play_spec 波2 已摘）。 run_gates=False（generation-only，给 compare_node 对照用）：收口只到 stage+smoke 为止，**不**自动产 play-spec、 **不**跑内部九门 play —— 由调用方在生成与 play 之间注入金标 spec 再单独 play，把驱动器从对照变量里摘掉 （Codex C1：run_studio 内部已 play，对照需把生成与 play 拆开）。 interaction_profile_id（缺省 None=原行为）：只供 v3_shadow 校准路径显式选择。可信编排在 scaffold 前 交叉核对 profile/template/proof，随后固化进 Writer 不可写的 entry-bundle、验收 identity 与 smoke runner。 普通 puzzle 不传该字段，interactionBinding 保持 null，禁止按 brief 或模板自动猜测。 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}。 · acceptance_parent_run_id / acceptance_repair_count：只给 A11 的 v3 修复复验传 parentRun 血缘。 · acceptance_brief：writer 指令与验收题面不同时显式传原 brief，防修复反馈污染 briefHash。 · reference_asset_policy_id：可信调用方显式选择冻结参照策略；None 时不触发 gate、不改变 session/kick。 """ t0 = time.time() _rec(f"model={_bootstrap.SPIKE_MODEL} id={game_id} brief=「{brief}」max_iters={max_iters} max_resumes={max_resumes}") acceptance_mode = acceptance_v3_mode() frozen_reference_assets = None frozen_reference_constraint_block = None reference_asset_generation_receipts = None if reference_asset_policy_id is not None: # 可信消费必须先于 scaffold/Writer；helper 内固定仓根、release 与 hash，调用方不能自报信任材料。 frozen_reference_assets = cheap_verify.preflight_reference_asset_policy( reference_asset_policy_id, acceptance_mode) frozen_reference_constraint_block = cheap_verify.build_frozen_reference_asset_constraint_block( frozen_reference_assets) import reference_asset_gate # noqa: PLC0415 reference_asset_generation_receipts = reference_asset_gate.to_json_value( frozen_reference_assets.receipts) v3_enabled = run_gates and acceptance_mode in ("v3", "v3_shadow") v3_brief = brief if acceptance_brief is None else acceptance_brief # create 的可信路由必须发生在 scaffold/Writer 前；路由一旦选定就同时决定实际模板与 proof profile。 # 无法命中五个可信模板时 fail-closed，不允许先用通用模板生成、验收时再按 genre 猜 profile。 if v3_enabled and prepare is None and scaffold_template is None: from cheap_genre_route import route_genre # noqa: PLC0415 scaffold_template, routed_genre = route_genre(v3_brief) genre = genre or routed_genre if scaffold_template and scaffold_desc is None: scaffold_desc = SCAFFOLD_DESC_BY_TEMPLATE.get(scaffold_template) if v3_enabled and acceptance_identity is None and not (acceptance_template_route or scaffold_template): return apply_v3_entry_failure( {"ok": False, "gameId": game_id, "brief": brief, "finished": False}, "Writer 前无法选择可信 templateRoute；v3 禁止使用通用模板或仅凭 genre 反推 profile", mode=acceptance_mode, ) supplied_binding = ((acceptance_identity or {}).get("interactionBinding") if isinstance(acceptance_identity, dict) else None) supplied_profile_id = (supplied_binding.get("interactionProfileId") if isinstance(supplied_binding, dict) else None) if (interaction_profile_id is not None and supplied_profile_id is not None and interaction_profile_id != supplied_profile_id): return apply_v3_entry_failure( {"ok": False, "gameId": game_id, "brief": brief, "finished": False}, "显式 interaction_profile_id 与 acceptanceIdentity.interactionBinding 不一致", mode=acceptance_mode, ) requested_profile_id = interaction_profile_id or supplied_profile_id trusted_route = (acceptance_template_route or scaffold_template or ((acceptance_identity or {}).get("templateRoute") if isinstance(acceptance_identity, dict) else None)) trusted_genre = (genre or ((acceptance_identity or {}).get("genre") if isinstance(acceptance_identity, dict) else None) or (cheap_verify.GENRE_BY_TEMPLATE.get(trusted_route) if trusted_route else None)) if requested_profile_id is not None: try: # 显式 profile 属于冻结校准身份，必须在 archive/scaffold/Writer 任一写操作前验明。 cheap_verify.preflight_interaction_profile_v3( requested_profile_id, acceptance_mode, genre=str(trusted_genre or ""), template_route=str(trusted_route or ""), proof_profile_id=((acceptance_identity or {}).get("proofProfileId") if isinstance(acceptance_identity, dict) else None)) except Exception as exc: # noqa: BLE001 交互身份不可信时不得先改盘 return apply_v3_entry_failure( {"ok": False, "gameId": game_id, "brief": brief, "finished": False}, f"Writer 前 interaction profile 预检失败:{type(exc).__name__}: {exc}", mode=acceptance_mode, ) # 外部 canonical identity 是显式可信输入；其 binding 通过预检后，scaffold/smoke/repair 共用同一 profile。 interaction_profile_id = requested_profile_id # W-AXIS 波1 F0-b per-run 归档:create 路 scaffold 会 rmSync 整个 amgen-/、重跑同 gid 抹掉上一 run 的 # trace.jsonl/turns.jsonl/证据。故 scaffold 前先把上一 run 整体移进带时间戳归档位(永不覆盖)。modify 路 # (prepare 给定)base 源须保留、不归档。best-effort:关/失败退回旧覆盖行为(archived_to=None)。 archived_to = cheap_run.archive_prior_run(game_id) if prepare is None else None # create=scaffold clone 模板(扩模板:scaffold_template 传 per-genre 黄金骨架名,如经营=_template-shop)；modify=上游已预备的 noop prep = prepare or (lambda gid: cheap_run.scaffold( gid, scaffold_template, interaction_profile_id=interaction_profile_id)) 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 源已预备'}）") _persist_brief_forensics(game_id, brief) # fix③:scaffold 后即落 brief 取证(硬失败/挂死也留得下,不进可交易工程根) # W-AXIS 波1 F0-c evidence 清理清单化(create 路;取代旧红线③只清 verdict.json 一个文件)——清 _wg1-gen//evidence/ # 残留(verdict/续修反馈/日志/截图全列),封「读回上一 run 反馈/绿 verdict → 误归因/假绿」。归档已移走则为 no-op。 if prepare is None: cheap_run.clean_stale_evidence(game_id) # v3 身份必须在 Writer 创建前由可信编排层冻结。templateRoute 是 profile 选择轴；genre 只能交叉核对， # 绝不允许在验收阶段仅凭 genre 重新猜 profile。modify 路由调用方通过 acceptance_template_route # 或完整 acceptance_identity 传入源项目可信路由，身份仅存本函数局部变量。 v3_identity = None if v3_enabled: try: # 生产 Service 透传后端 traceId；直接 CLI 没有后端任务时，在 Writer 前生成本次唯一标识。 # 外部 identity 若缺原始 traceId 无法证明属于当前任务，必须 fail-closed。 local_task_trace_id = _resolve_acceptance_task_trace_id( str(game_id), acceptance_task_trace_id, identity_supplied=acceptance_identity is not None) expected_task_binding = cheap_verify.task_binding_hash_v3(local_task_trace_id) if acceptance_identity is not None: supplied = dict(acceptance_identity) if expected_task_binding is not None and supplied.get("taskBindingHash") != expected_task_binding: raise ValueError("acceptanceIdentity.taskBindingHash 与当前任务 traceId 不一致") expected = cheap_verify.build_acceptance_v3_identity( game_id, v3_brief, genre=str(supplied.get("genre") or ""), template_route=str(supplied.get("templateRoute") or ""), source_artifact_hash=supplied.get("sourceArtifactHash"), parent_acceptance_request_hash=supplied.get("parentAcceptanceRequestHash"), repair_ordinal=int(supplied.get("repairOrdinal") or 0), proof_profile_id=supplied.get("proofProfileId"), proof_registry_version=supplied.get("proofRegistryVersion"), task_binding_hash=supplied.get("taskBindingHash"), interaction_binding=supplied.get("interactionBinding"), # W-GOLD-LIVE 检查点 3a：外部身份的参照资产消费声明三字段原样透传重建， # 供下方全等比对与消费对账闸消费；缺省 None/[]/null ≡ 未声明（旧行为不变）。 design_ref=supplied.get("designRef"), reference_asset_record_ids=supplied.get("referenceAssetRecordIds"), consumer_ref=supplied.get("consumerRef"), ) if supplied != expected: raise ValueError("调用方 acceptanceIdentity 与 canonical registry 不一致") v3_identity = supplied else: if int(acceptance_repair_count or 0) != 0: raise ValueError("修复入口必须显式传入锁定 parent/profile/registry 的 acceptanceIdentity") v3_identity = cheap_verify.build_acceptance_v3_identity( game_id, v3_brief, genre=str(trusted_genre or ""), template_route=str(trusted_route or ""), repair_ordinal=0, task_binding_hash=expected_task_binding, interaction_profile_id=interaction_profile_id, reference_asset_record_ids=( [receipt["recordId"] for receipt in reference_asset_generation_receipts] if reference_asset_generation_receipts is not None else None), consumer_ref=(reference_asset_generation_receipts[0]["consumerRef"] if reference_asset_generation_receipts is not None else None)) except Exception as exc: # noqa: BLE001 —— 无可信身份时禁止 Writer 改盘 return apply_v3_entry_failure( {"ok": False, "gameId": game_id, "brief": brief, "finished": False}, f"Writer 前无法冻结 v3 acceptance identity:{type(exc).__name__}: {exc}", mode=acceptance_mode, ) if frozen_reference_assets is not None and v3_identity is not None: # 外部 acceptanceIdentity 即使自身 canonical，也必须与本次 /2 冻结回执逐项一致； # 否则会让 Writer 在验收侧最终拒绝之前先消费一套未进入 identity 的参照字节。 frozen_record_ids = [receipt["recordId"] for receipt in reference_asset_generation_receipts] frozen_consumers = {receipt["consumerRef"] for receipt in reference_asset_generation_receipts} if (v3_identity.get("referenceAssetRecordIds") != frozen_record_ids or len(frozen_consumers) != 1 or v3_identity.get("consumerRef") not in frozen_consumers): return apply_v3_entry_failure( {"ok": False, "gameId": game_id, "brief": brief, "finished": False}, "Writer 前冻结 policy 与 acceptance identity 不一致", mode=acceptance_mode, ) # Registry/1 identity 消费闸仅作旧 acceptance 回放兼容；/2 生产消费已在 scaffold 前完成冻结预检， # 其 consumerRef 与 /1 历史值不同，已有冻结快照时不得再进入 /1 对账。 v3_reference_constraint_block = None if v3_enabled and v3_identity is not None and frozen_reference_assets is None: try: reference_gate = cheap_verify.build_v3_reference_asset_generation_constraints(v3_identity) except ValueError as exc: # noqa: BLE001 —— 声明消费而对账失败：fail-closed 拒绝生成入口 return apply_v3_entry_failure( {"ok": False, "gameId": game_id, "brief": brief, "finished": False}, f"参照资产消费对账失败，生成入口拒绝:{exc}", mode=acceptance_mode, ) if reference_gate is not None: v3_reference_constraint_block = reference_gate["constraint_block"] session = CheapSession( game_id=game_id, reference_files=(frozen_reference_assets.reference_files if frozen_reference_assets else None), reference_roots=(frozen_reference_assets.reference_roots if frozen_reference_assets else None), ) 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-/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}") # 阶段四观测 波③ T3-1:CLI 路同款并接 OTLP span sink(发 Collector,service.name=cheap-gen-worker)。 # 默认关(env CHEAP_OTLP_ENDPOINT 未设)→ 原样返回上面的 jsonl sink,CLI 现有行为字节不变; # 开则包 fan-out(jsonl 落盘照旧、OTLP 额外发一份)。best-effort:观测挂掉绝不咬生成(设计 §8)。 try: import cheap_otlp_sink # noqa: PLC0415 函数体内惰性 import,顶层不牵 otel _trace_sink = cheap_otlp_sink.build_trace_sink(_trace_sink, trace_id=game_id) except Exception as _otlp_err: _rec(f"OTLP sink 并接异常（回落原 jsonl sink，不阻断生成）：{_otlp_err}") tracer = Tier2TraceMiddleware(trace_id=game_id, sink=_trace_sink) # W-AXIS 波1 真相层:CLI 路同款逐 turn 全文落 amgen-/turns.jsonl(模型文本/工具全参/工具返回/门反馈)。 # CLI 路无 RepairMiddleware,门反馈是外层 resume 循环重发的 kick(name=user)——turns 中间件在 on_reply 起点 # 读 input_kwargs['inputs'] 即收得到。observe-only、best-effort,默认开;失败/关→None 不加入 middlewares。 _turns_mw = None try: import cheap_turns_sink # noqa: PLC0415 _turns_mw = cheap_turns_sink.build_turns_middleware(game_id, trace_id=game_id) except Exception as _te: # noqa: BLE001 —— 真相层接线失败绝不阻断生成 _rec(f"turns 真相层中间件接线异常(降级不落 turns,不阻断生成):{type(_te).__name__}: {_te}") writer = Agent( name="cheap-writer", # create 路统一用通用 create prompt（scaffold_desc 引导起点，不写锁、模型自由改全部游戏文件）； # modify 路由调用方显式传 system_prompt=build_modify_system_prompt（write_whitelist 只用于 toolkit 写边界收窄， # 与 prompt 选择解耦——W-AXIS 波2 拆除旧 create 路换皮写锁 build_reskin_system_prompt 后此处不再分叉）。 system_prompt=system_prompt or build_system_prompt(game_id, scaffold_desc), model=model, toolkit=toolkit, # trace 外层、熔断内层（列表序=洋葱序）;turns 真相层 observe-only 置中（只读事件、不拦截,顺序无碍）。 middlewares=([tracer, _turns_mw, breaker] if _turns_mw is not None else [tracer, breaker]), 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。" # 旧 Registry/1 identity 约束与 v2 冻结策略约束均只在各自显式选择后追加；None 路径文本不变。 + (f"\n\n{v3_reference_constraint_block}" if v3_reference_constraint_block else "") + (f"\n\n{frozen_reference_constraint_block}" if frozen_reference_constraint_block else "")) breaker_tripped = None attempts = 0 closeout = None # CLI 回喂:最近一次九门收口结果(回喂循环产出,收口段复用、不重复跑门) acceptance_v3 = None acceptance_v3_first_pass = None v3_entry_error = None v3_repair_count = int((v3_identity or {}).get("repairOrdinal") or acceptance_repair_count or 0) v3_parent_run_id = acceptance_parent_run_id # 已计入父链的 writer 成本基线；每轮验收只加自上次验收后的新增 writer delta。 v3_writer_cost_accounted = 0.0 # v3 的唯一一次 writer 修复不占旧 resume 配额；旧配额继续只服务 check/build 与畸形工具调用恢复。 total_attempts = max_resumes + 1 + (1 if v3_enabled and v3_repair_count == 0 else 0) try: for attempt in range(total_attempts): attempts = attempt + 1 _rec(f"resume attempt {attempts}/{total_attempts} → writer.reply …") try: await writer.reply(_user_msg(kick)) except Tier2CircuitBreak as _cb: # fix①-B(重实·resume 层,2026-07-08 n=5 收敛环坐实 heritage 死圈复现)—— # 根因:MiniMax 长参数下漏 write_file/edit_file 必填 path,schema 校验拒 → 连撞 stuck 阈值 → 终态杀、九门没跑。 # 先前 on_acting 版无效(_agent.py:1376 校验先于 Step3 工具执行/on_acting、畸形调用直接 return 不进钩子); # 改挂此处:missing-param 类 stuck 不当终态,还有 resume 预算就带【贴脸补参反馈】续修——与 gate-fail resume # 同机制、跨 reply memory 保留=原地续修、能真正触达 agent。只清 stuck 连击计数(不动 spent_rmb/tool_calls,防误清预算); # 非 missing-param 的 stuck(反复 check 同红线 / build 同编译错=真死圈)/ budget / 其它熔断照旧向上抛(终态、不放行)。 _missing_param = _cb.kind == "stuck" and "is a required property" in (_cb.reason or "") if _missing_param and attempt < max_resumes and not getattr(breaker, "budget_soft_tripped", False): _rec(f"熔断 stuck(漏必填参 path)→ fix①-B 带补参反馈续修(attempt {attempts}):{(_cb.reason or '')[:70]}") breaker._fail_repeat = 0 # 只清 stuck 连击计数(下一 attempt 干净起,不动 spent_rmb/tool_calls) breaker._last_fail_sig = None kick = _MALFORMED_RESUME_FEEDBACK continue raise # 其它熔断 / 预算耗尽 / 非漏参 stuck → 终态,交外层 except 记 breaker_tripped # ① 真 finish：先跑机械收口。v3 只听 final decision，九门失败本身不再触发玩法续修； # 显式历史 v1/v2 回放才保留旧 C6 九门反馈 resume。 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, interaction_profile_id=interaction_profile_id, ) vb = _verdict_brief(closeout["verdict"]) _rec(f"九门 play pass={vb['pass']} failedGates={vb['failedGates']}") if v3_enabled: writer_cost_now = float(getattr(breaker, "spent_rmb", 0.0) or 0.0) request_writer_cost = next_v3_writer_cost( writer_cost_now, v3_writer_cost_accounted) acceptance_v3 = await cheap_verify.run_acceptance_v3(build_acceptance_v3_request( game_id, v3_brief, closeout["verdict"], acceptance_identity=v3_identity, idempotency_key=f"{game_id}:studio-v3", repair_count=v3_repair_count, parent_run_id=v3_parent_run_id, writer_cost_rmb=request_writer_cost, reference_asset_policy_id=reference_asset_policy_id, reference_asset_generation_receipts=reference_asset_generation_receipts)) decision = acceptance_v3.get("decision") or {} _rec(f"验收 v3 mode={acceptance_mode} outcome={decision.get('outcome')} " f"accepted={decision.get('accepted')} publishFrozen={decision.get('publishFrozen')} " f"repairEligible={decision.get('repairEligible')}") if cheap_verify.is_v3_repair_authorized(acceptance_v3) and v3_repair_count == 0: # 只有 final postguard 产出的 verified reject 才能到这里；清掉旧终态后让同一 writer 修一次。 acceptance_v3_first_pass = acceptance_v3 # 同一 Writer 继续前先冻结修复身份：profile/registry/templateRoute 原样锁定， # 只把 parent request 与首轮最终 artifact 作为 ordinal=1 血缘写入。 v3_identity = cheap_verify.build_acceptance_v3_identity( game_id, v3_brief, genre=v3_identity["genre"], template_route=v3_identity["templateRoute"], source_artifact_hash=acceptance_v3["artifactHash"], parent_acceptance_request_hash=v3_identity["acceptanceRequestHash"], repair_ordinal=1, proof_profile_id=v3_identity["proofProfileId"], proof_registry_version=v3_identity["proofRegistryVersion"], task_binding_hash=v3_identity["taskBindingHash"], interaction_binding=v3_identity.get("interactionBinding"), design_ref=v3_identity.get("designRef"), reference_asset_record_ids=v3_identity.get("referenceAssetRecordIds"), consumer_ref=v3_identity.get("consumerRef"), ) v3_repair_count = v3_identity["repairOrdinal"] v3_parent_run_id = acceptance_v3.get("runId") v3_writer_cost_accounted = writer_cost_now kick = build_v3_repair_prompt(decision.get("repairFeedback")) session.finished = None acceptance_v3 = None # 产物即将变化，旧 artifactHash 的决策不得冒充终态。 _rec("v3 verified reject → 同一 writer 仅一次修复；完成后重跑机械门与 v3") continue break j = judge_cheap_verdict(closeout["verdict"], game_id=game_id, staged_dir=cheap_run.wg1_game_dir(game_id)) 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 →循环外九门 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, interaction_profile_id=interaction_profile_id, ) # 防御:正常路循环内已跑 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, interaction_profile_id=interaction_profile_id, ) smoke_ok = sm["ok"] _rec(f"smoke {'PASS' if smoke_ok else 'FAIL'}(抓 state 供 play-spec)") if not smoke_ok: _rec(f"smoke raw 尾:{(sm.get('raw') or '')[-300:]}") # 真因可见(同 _closeout_gates 注) _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)) # ── W-AXIS 波1:失败三层归因 + per-run 归档指针随 summary 透传(additive;供批次账每 run 记一笔)── # verdict=None(没跑到 play)时 classify 归 layer=none(不误标坏死);archived_to 首跑/关/失败为 None、不带该键。 summary["failureLayer"] = classify_cheap_failure_layer(verdict, staged_dir=cheap_run.wg1_game_dir(game_id)) if archived_to: summary["archivedPriorRunTo"] = archived_to # ── 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}") # ── 统一验收入口：v3 可信证据闭环；仅历史 v1/v2 模式保留旧编排器 ── # 拆着杀:契约无关四门(A/B/C/D 投影)= 预筛权威(取代旧 verdict.pass 九门口径,那随契约退役会坍缩), # 过筛者交测试 agent 视觉引导真玩裁 broken/hollow/off-brief;mode=v2 阻断、shadow 灰度对照、v1 旧口径。 # run_acceptance 内建 fail-closed 与顶层兜底、绝不抛,additive 写 summary['floor']/['playtest']/['judge'] # /['acceptanceVersion'] 并更新 ['ok']/['accepted']。只在 run_gates(有真 verdict)时接入;对照路零改动。 if run_gates and finished: if v3_enabled: if v3_entry_error: summary = apply_v3_entry_failure(summary, v3_entry_error, mode=acceptance_mode) else: # 正常路径已在 finish 分支完成 v3；防御性补跑只覆盖没有进入该分支的异常控制流。 if acceptance_v3 is None: writer_cost_now = float(getattr(breaker, "spent_rmb", 0.0) or 0.0) acceptance_v3 = await cheap_verify.run_acceptance_v3(build_acceptance_v3_request( game_id, v3_brief, verdict, acceptance_identity=v3_identity, idempotency_key=f"{game_id}:studio-v3", repair_count=v3_repair_count, parent_run_id=v3_parent_run_id, writer_cost_rmb=next_v3_writer_cost( writer_cost_now, v3_writer_cost_accounted), reference_asset_policy_id=reference_asset_policy_id, reference_asset_generation_receipts=reference_asset_generation_receipts)) summary = apply_acceptance_v3(summary, acceptance_v3, first_pass=acceptance_v3_first_pass) else: summary = await cheap_verify.run_acceptance(summary, game_id=game_id, brief=brief, verdict=verdict) _js = summary.get("judge") or {} _pt = summary.get("playtest") or {} _rec(f"历史验收 mode={summary.get('acceptanceVersion')} floor={(summary.get('floor') or {}).get('pass')} " f"playtest.accepted={_pt.get('accepted')} rolls={_pt.get('rollCount')} degraded={_pt.get('degraded')} " f"costRmb={_pt.get('costRmb')} judge.verdict={_js.get('verdict')} " f"→ accepted={summary.get('accepted')} ok={summary.get('ok')}") 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']}") # ── 面二观测:发 llm_tokens{kind} 计数 + llm_cache_hit_rate(默认关 / best-effort;绝不咬生成)── # 本路(CLI/批跑)model 是 RecordingOpenAIChatModel,records 每条 (in,out,cached) 三元组,cached 由 # usage.cache_input_tokens 归一 → 三项全可得(生产 Service 路框架默认 model 无 records,cached 拿不到、 # 见 cheap_service_app collector)。取不到 CHEAP_OTLP_ENDPOINT 即 no-op,与 trace sink 默认关同源; # 面五(game-cloud GenMetrics)不发 token 计数,故不双计。发送异常只报告、绝不阻断生成。 try: import cheap_otlp_sink # noqa: PLC0415 惰性 import(顶层不牵 otel) _cached_tok = sum(r[2] for r in model.records if len(r) > 2) cheap_otlp_sink.record_llm_token_metrics(tok_in, tok_out, _cached_tok) except Exception as _me: # noqa: BLE001 面二 metric best-effort,绝不影响生成主链 _rec(f"面二 metric 发送异常(已忽略,不影响生成):{type(_me).__name__}: {_me}") # 注：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)