2c4e2e2248
修「gate 通过≠人能玩」盲区(feie-005 实证:过九门+富三门却棋盘渲染空 + 8s 耐心人玩不了)。 补齐 D 族三层校验,全 advisory(写 verdict 不碰 decision;自带 Goodhart 守护:advisory 全挂也不翻 decision): - 人可玩门(play-phaser.cdp.cjs):render-reflects-state(状态有 item→该格 canvas 像素须非空) + render-sanity(无 [object Object]/undefined/NaN 漏底文本)+ human-timing(订单耐心 ≥ HUMAN_MIN_PATIENCE_MS=12000·★directional) - L2 设计符合 full(run.py compute_l2_signals):systemsPresent/mergeDag/ordersReachable/economyConsistent 四硬判据, 从落盘源工程确定性算(替桩) - L3 升门判据(run.py derive_l3_egregious + studio.py):消费 humanPlayability 的恶性渲染失败→suggestFatal 信号 - schema additive:humanPlayability 段 + L2.mismatches + L3.egregiousRenderFailure 离线验证:py_compile / node --check / JSON 合法 / 24 断言单测(含 Goodhart 守护:advisory 全挂 decision 字节不变)/ 金标 L2 离线 passed=True + 反例精确报错。 **fatal 提升 + 金标冒烟集成验证待 mini-desktop**:确认金标仍 ACCEPT(新检查不抛、全 pass)+ feie-005 被挂后, 逐条提 fatal(品类限定有 board/orders 档)。校准数据:金标 patience=16-20s / feie-005=8s / 阈值候选 12s。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
565 lines
33 KiB
Python
565 lines
33 KiB
Python
"""agent_loop/studio.py —— tier2 富游戏单写 ReAct 编排(on AgentScope v2.0.2)。
|
||
|
||
闭环(对照 wg1 studio.py 的「单轮 + 外层 Python for repair」,tier2 改成「Agent 内多轮 ReAct 自治」):
|
||
阶段 1:设计 agent(单轮)把题面 → 富游戏设计稿(承袭 wg1 design 范式,换富游戏多系统语义)。
|
||
阶段 2:单写 agent —— 一个 Agent 配九工具 Toolkit + 放开 max_iters,在 ReAct 循环内自调
|
||
scaffold_init/write_source/build/run_gates/finish 自治收敛,**不再外层 Python for repair**。
|
||
|
||
框架接缝(2.0.2,从 wg1 种子平移 + 升级):
|
||
- L32/33 平移:`from agentscope.agent import Agent, ReActConfig` / `from agentscope.message import UserMsg`。
|
||
- 关键升级:`ReActConfig(max_iters=1)`(wg1 单轮)→ 放开(单写 ≤ writer_max_iters);配 `toolkit=`(九工具面);
|
||
model `OpenAIChatModel` → `AnthropicChatModel`(M3 路,config.build_model)。
|
||
|
||
自治化两个必备(源码核验):
|
||
① 权限:FunctionTool 默认 check_permissions 返回 ASK(_adapters.py:99),无人值守会卡死 →
|
||
用 `PermissionMode.BYPASS`(沙箱无人值守的官方档位,permission/_types.py)经 AgentState 注入,跳过 ASK。
|
||
② 熔断:四道熔断做成 CircuitBreakerMiddleware(on_reply 洋葱钩子),挂中间件洋葱;触发抛
|
||
Tier2CircuitBreak,本编排 catch → 落 verdict.breakerKind。
|
||
|
||
设计依据:docs/architecture/架构/生成引擎/tier2细节图说-{C,D,E,F}.md;契约 tier2/contracts/toolkit-signatures.md。
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
# 包内/直跑兼容导入(直跑时把 tier2/ 加进 sys.path,使 `gen_worker.*` 可解析)。
|
||
try:
|
||
from .. import config, roles, run
|
||
from ..toolkit import Tier2Session, build_toolkit
|
||
from ..middleware import CircuitBreakerMiddleware, Tier2CircuitBreak, Tier2TraceMiddleware
|
||
except ImportError: # pragma: no cover —— 直接 python studio.py 兜底
|
||
# 本文件在 tier2/gen-worker/worker/agent_loop/studio.py;把 gen-worker/ 加进 sys.path,
|
||
# 使顶层包 `worker` 可解析(包名 worker 合法;目录名 gen-worker 含连字符不可直接 import)。
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from worker import config, roles, run # type: ignore
|
||
from worker.toolkit import Tier2Session, build_toolkit # type: ignore
|
||
from worker.middleware import ( # type: ignore
|
||
CircuitBreakerMiddleware, Tier2CircuitBreak, Tier2TraceMiddleware)
|
||
|
||
# 成本接线(H3):RecordingChatModel.records → cost_for_run 折¥;new-api 计费参数活读取(取不到回落显式参数)。
|
||
# observability 是 gen-worker 顶层包(非 worker 子包);直跑兜底里已把 gen-worker/ 加进 sys.path。
|
||
try:
|
||
from observability.cost import tokens_by_model_from_records, cost_for_run
|
||
from observability.newapi_pricing import fetch_pricing_params, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE
|
||
except Exception: # pragma: no cover —— 直跑/路径未就位兜底
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from observability.cost import tokens_by_model_from_records, cost_for_run # type: ignore
|
||
from observability.newapi_pricing import ( # type: ignore
|
||
fetch_pricing_params, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE)
|
||
|
||
# 框架接缝(2.0.2):Agent + ReActConfig + UserMsg + AgentState + BYPASS 权限。
|
||
from agentscope.agent import Agent, ReActConfig
|
||
from agentscope.message import UserMsg
|
||
# L3 视觉软检多模态消息块(2.0.2 源码核验:UserMsg.content 允许 text/data 两类块;
|
||
# 图走 DataBlock(source=Base64Source(media_type='image/png')),AnthropicChatFormatter 会转成
|
||
# Anthropic image base64 入参,见 formatter/_anthropic_formatter.py:227 _format_data_block)。
|
||
from agentscope.message import TextBlock, DataBlock, Base64Source
|
||
from agentscope.state import AgentState
|
||
from agentscope.permission import PermissionContext, PermissionMode
|
||
|
||
|
||
def user_msg(text: str) -> UserMsg:
|
||
"""2.0.1+:UserMsg 必须带 name;字符串 content 会被包成 TextBlock。"""
|
||
return UserMsg(name="user", content=text)
|
||
|
||
|
||
def text_of(resp) -> str:
|
||
"""从 AgentScope Msg 抽纯文本(content 是 block 列表,兼容 dict/对象)。"""
|
||
blocks = getattr(resp, "content", None)
|
||
if blocks is None and hasattr(resp, "get"):
|
||
blocks = resp.get("content")
|
||
out = []
|
||
for b in (blocks or []):
|
||
if isinstance(b, dict):
|
||
if b.get("type") == "text":
|
||
out.append(b.get("text", "") or "")
|
||
elif getattr(b, "type", None) == "text":
|
||
out.append(getattr(b, "text", "") or "")
|
||
return "".join(out)
|
||
|
||
|
||
async def _design_stage(brief: str, model) -> str:
|
||
"""阶段 1:设计 agent(单轮,无工具)把题面 → 富游戏设计稿。"""
|
||
agent = Agent(
|
||
name="design",
|
||
system_prompt=roles.DESIGN_SYSTEM,
|
||
model=model,
|
||
react_config=ReActConfig(max_iters=1), # 设计只发散一轮,不自调工具。
|
||
)
|
||
resp = await agent.reply(user_msg(brief))
|
||
return text_of(resp)
|
||
|
||
|
||
def _bypass_state() -> AgentState:
|
||
"""造一个权限=BYPASS 的 AgentState,使单写 agent 无人值守自调工具不卡在 ASK。
|
||
|
||
BYPASS = 沙箱/无人值守档位(permission/_types.py):跳过工具 ASK,只认用户显式 deny/ask 规则。
|
||
本线运行在受控生成环境(工具只写工程 workdir / 调本地子进程),适用 BYPASS。
|
||
"""
|
||
return AgentState(permission_context=PermissionContext(mode=PermissionMode.BYPASS))
|
||
|
||
|
||
# ── L3 视觉软检(observe-only · 绝不当门 · 防 Goodhart)──────────────────────────
|
||
# 设计依据:tier2细节图说-D-三层校验与九门.md 图 D1 三层校验全景。
|
||
# L3 = 效果层:特效/美观/好不好玩,处置「只评分、绝不解决、绝不阻塞拒发」。工具 = M3 多模态视觉软检
|
||
# (看截图打分),产出只进质量趋势/告警/给人工终审减负。**永不参与 decision**——decision 只由 L1 硬门裁。
|
||
# 为何死活不当门:扩确定性门判好玩→可被刷的代理指标;纯 LLM 当裁判→Goodhart 红线(优化代理指标偏离真目标)。
|
||
# verdict schema 把 L3.scoreOnly 焊成 const true,本实现只写 score/notes,绝不回写 decision/L1。
|
||
|
||
# L3 玩家人格(单一视觉软检位;wg1 player panel 同款「急性子休闲玩家」口径,凭第一眼观感+反馈下判断)。
|
||
L3_PERSONA = "急性子休闲玩家,凭第一眼观感和系统联动反馈下判断,只信看到的证据、不脑补"
|
||
|
||
|
||
def _b64_data_block(png_path: Path) -> DataBlock | None:
|
||
"""把一张本地 PNG 读成 L3 多模态消息块(DataBlock + Base64Source);读不到/空文件返 None。
|
||
|
||
截图只取证、绝不进硬门(防 Goodhart);本函数只为 L3 软检喂图。任何 IO 异常都吞成 None
|
||
(L3 是 observe-only,缺图绝不能中断主链),由调用方据 None 走「无图」兜底。
|
||
"""
|
||
try:
|
||
if not png_path.exists() or png_path.stat().st_size == 0:
|
||
return None
|
||
data = base64.b64encode(png_path.read_bytes()).decode("ascii")
|
||
return DataBlock(source=Base64Source(media_type="image/png", data=data))
|
||
except Exception as e: # noqa: BLE001 —— 读图失败绝不抛(L3 observe-only),记日志后走无图兜底
|
||
print(f"[tier2-studio][L3] 读截图失败(走无图兜底):{png_path} {type(e).__name__}: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
def _l3_play_summary(verdict: dict | None) -> str:
|
||
"""把 verdict 的真玩取证(playReport / 富游戏门)摘成给 L3 玩家 agent 的运行数据文字。
|
||
|
||
只喂确定性 harness 的客观观测(是否真渲染/真玩到终态/终局语义 state),让玩家 agent 据证据判,
|
||
而非脑补。注意:L3 看的是「真玩取证 + 截图」,judge 的硬门结论(pass/fail)不喂给它当答案
|
||
(出题的和被考的分离)。
|
||
"""
|
||
if not verdict:
|
||
return "(无 verdict:疑似装载即崩,无真玩取证)"
|
||
pr = verdict.get("playReport") or {}
|
||
rich = (((verdict.get("layerResults") or {}).get("L1") or {}).get("richGameGates")) or {}
|
||
snap = pr.get("stateSnapshot") or {}
|
||
lines = [
|
||
f"- 是否捕获 boot 就绪(loaded):{pr.get('loaded')}",
|
||
f"- harness 真输入次数(drivenInputs):{pr.get('drivenInputs')}",
|
||
f"- 是否驱动到终态(reachedTerminal):{pr.get('reachedTerminal')} 终态 phase={pr.get('terminalPhase')}",
|
||
f"- 终局语义 state 快照:coins={snap.get('coins')} ingredients={snap.get('ingredients')} "
|
||
f"orders={snap.get('orders')} phase={snap.get('phase')}",
|
||
]
|
||
# 富游戏三门的子检查名(只给「测了哪些联动」的客观项名,不给过/不过当答案)。
|
||
for name in ("tripleLink", "economy", "latch"):
|
||
g = rich.get(name)
|
||
if isinstance(g, dict):
|
||
sub = [c.get("name") for c in (g.get("checks") or [])]
|
||
lines.append(f"- 富游戏门 {name} 观测子项:{sub}")
|
||
return "确定性 harness 真玩取证:\n" + "\n".join(lines)
|
||
|
||
|
||
def _parse_l3_judge(raw: str) -> dict:
|
||
"""容错解析 L3 玩家 agent 的 JSON 评判({completeness,fun,problems,verdict,note})。
|
||
|
||
便宜模型/M3 偶尔吐带前后缀的脏 JSON,故先抠出第一个 {...} 再 json.loads;全失败 → 给一个
|
||
score=null 的诚实空判(observe-only,绝不因解析失败中断或翻 GREEN)。
|
||
"""
|
||
obj = None
|
||
try:
|
||
obj = json.loads(raw)
|
||
except Exception:
|
||
# 抠出首个花括号块再试(去掉模型可能加的 ```json 包裹 / 解说前后缀)。
|
||
try:
|
||
s = raw.find("{")
|
||
e = raw.rfind("}")
|
||
if s >= 0 and e > s:
|
||
obj = json.loads(raw[s:e + 1])
|
||
except Exception:
|
||
obj = None
|
||
if not isinstance(obj, dict):
|
||
return {"completeness": None, "fun": None, "problems": [], "verdict": "?",
|
||
"note": (raw or "")[:160]}
|
||
probs = obj.get("problems")
|
||
if not isinstance(probs, list):
|
||
obj["problems"] = [str(probs)] if probs else []
|
||
return obj
|
||
|
||
|
||
def _fun_to_score(fun) -> float | None:
|
||
"""把玩家 agent 的 fun(1-5 锚)线性映射到 verdict.L3.score(0-100;schema 允许 number|null)。
|
||
|
||
映射:fun=1→0,fun=5→100(score=(fun-1)/4*100)。非法/缺失 → None(诚实留空,不编分)。
|
||
"""
|
||
try:
|
||
f = float(fun)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if f < 1 or f > 5:
|
||
return None
|
||
return round((f - 1) / 4 * 100, 1)
|
||
|
||
|
||
async def _l3_player_panel(game_id: str, brief: str, verdict: dict | None, model) -> dict:
|
||
"""L3 视觉软检:player_system 提示词 + 真截图 + 真玩取证,调一次 M3 produce score/note。
|
||
|
||
**observe-only 铁律**:本函数只产 {score, notes, raw}(写进 verdict.L3),绝不回写 decision/L1/accept;
|
||
任何失败(截不到图/模型超时/解析失败)→ score=None + notes 记错,绝不中断、绝不翻 GREEN。
|
||
|
||
Args:
|
||
game_id: 工程标识(定位 evidence 目录里的真截图)。
|
||
brief: 题面(给玩家 agent 上下文)。
|
||
verdict: 最近一次 run_gates 的 verdict(摘真玩取证;无则只凭截图判)。
|
||
model: 复用单写/设计同档 M3 模型客户端(AnthropicChatModel;含 thinking,多模态由 formatter 适配)。
|
||
|
||
Returns:
|
||
{score: float|None, notes: [str], raw: {...}}。score 即 verdict.L3.score 的取值。
|
||
"""
|
||
notes: list[str] = []
|
||
# ① 取真截图:优先玩后帧(after-play.png,更能看出系统联动/进展),回落首帧(first-paint.png)。
|
||
# 真图由 CDP harness 在 run_gates 真玩时落盘(mini-desktop);6c6g 无 chrome → 无图,走无图软检。
|
||
ev_dir = run._workdir(game_id) / "evidence"
|
||
img_block = None
|
||
used_shot = None
|
||
for fn in ("after-play.png", "first-paint.png"):
|
||
blk = _b64_data_block(ev_dir / fn)
|
||
if blk is not None:
|
||
img_block, used_shot = blk, fn
|
||
break
|
||
if used_shot:
|
||
notes.append(f"L3 看图:{used_shot}")
|
||
else:
|
||
notes.append("L3 无截图(6c6g 无 chrome 或 harness 未产图),仅凭真玩取证软评")
|
||
|
||
# ② 组装多模态 UserMsg:文本(题面 + 真玩取证)+ 截图块(若有)。
|
||
summary = _l3_play_summary(verdict)
|
||
user_text = (f"游戏题面:\n{(brief or '').strip()}\n\n{summary}\n\n"
|
||
"请基于以上确定性真玩取证(及截图,若附)给出你的玩家评判,严格只输出约定 JSON。")
|
||
content: list = [TextBlock(text=user_text)]
|
||
if img_block is not None:
|
||
content.append(TextBlock(text="【玩后截图 after-play.png(如缺则为首帧)】"))
|
||
content.append(img_block)
|
||
user = UserMsg(name="user", content=content)
|
||
|
||
# ③ 调一次 M3 玩家 agent(单轮、无工具;system = player_system 提示词)。任何异常吞成 score=None。
|
||
try:
|
||
agent = Agent(
|
||
name="l3-player",
|
||
system_prompt=roles.player_system(L3_PERSONA),
|
||
model=model,
|
||
react_config=ReActConfig(max_iters=1), # 软检只发散一轮、不调工具(它也没 toolkit)
|
||
)
|
||
resp = await agent.reply(user)
|
||
raw = text_of(resp)
|
||
except Exception as e: # noqa: BLE001 —— 模型超时/网关 502 等绝不中断主链(L3 observe-only)
|
||
print(f"[tier2-studio][L3] 玩家 agent 调用失败(score=null 兜底):{type(e).__name__}: {e}", flush=True)
|
||
notes.append(f"L3 模型调用失败:{type(e).__name__}: {e}")
|
||
return {"score": None, "notes": notes, "raw": None}
|
||
|
||
# ④ 解析评判 → fun 映射 score;problems/note 进 notes(给人工终审减负,非判定)。
|
||
judge = _parse_l3_judge(raw)
|
||
score = _fun_to_score(judge.get("fun"))
|
||
if score is None:
|
||
notes.append(f"L3 fun 缺失/非法(fun={judge.get('fun')}),score 留 null")
|
||
if judge.get("note"):
|
||
notes.append(f"L3 总评:{str(judge.get('note'))[:160]}")
|
||
for p in (judge.get("problems") or [])[:5]:
|
||
notes.append(f"L3 劣化信号:{str(p)[:120]}")
|
||
print(f"[tier2-studio][L3] game={game_id} fun={judge.get('fun')} completeness={judge.get('completeness')} "
|
||
f"→ score={score}(observe-only,不参与 decision)", flush=True)
|
||
return {"score": score, "notes": notes,
|
||
"raw": {"completeness": judge.get("completeness"), "fun": judge.get("fun"),
|
||
"verdict": judge.get("verdict")}}
|
||
|
||
|
||
async def run_studio(
|
||
game_id: str,
|
||
brief: str,
|
||
play_spec: dict | None = None,
|
||
*,
|
||
model_name: str | None = None,
|
||
max_tokens: int = 16000,
|
||
thinking_budget: int = 8000,
|
||
writer_max_iters: int = 40,
|
||
fixture_hint: str = "",
|
||
do_design: bool = True,
|
||
) -> dict:
|
||
"""tier2 富游戏单写主编排。返回结果 dict(含 design / 源工程交付 / verdict / 熔断)。
|
||
|
||
Args:
|
||
game_id: 工程标识(决定 workdir = game-runtime/games/_tier2-gen/<game_id>)。
|
||
brief: 一句话/题面。
|
||
play_spec: 真玩驱动规格(driver/assertAfterPlay/expectLatch;run_gates 用)。
|
||
model_name: M3 模型名(默认 env TIER2_MODEL 或 MiniMax-M3)。
|
||
max_tokens / thinking_budget: M3 输出上限 / thinking 预算(硬约束 max_tokens > thinking_budget)。
|
||
writer_max_iters: 单写 ReAct 放开的最大轮数(C3 建议整局 ≤40)。
|
||
fixture_hint: 品类/靶子提示(如 mini-肥鹅 三系统约定)。
|
||
do_design: 是否先跑阶段 1 设计(默认 True)。
|
||
"""
|
||
t0 = time.perf_counter()
|
||
mname = model_name or config.model_name_from_env()
|
||
|
||
# ── 阶段 1:设计(单轮)──
|
||
design_text = ""
|
||
if do_design:
|
||
design_model = config.build_model(
|
||
mname, max_tokens=max_tokens, thinking_budget=thinking_budget)
|
||
design_text = await _design_stage(brief, design_model)
|
||
|
||
# ── 阶段 2:单写 ReAct(放开 max_iters + 九工具 + 四熔断 + trace 接线 + BYPASS 权限)──
|
||
session = Tier2Session(game_id, play_spec=play_spec)
|
||
toolkit = build_toolkit(session)
|
||
breaker = CircuitBreakerMiddleware() # 四道熔断 + 软刹(on_reply / on_system_prompt)
|
||
# trace 接线(H1/H2):traceId 用 game_id(贯穿本次生成,对接 verdict.evidence.traceId / 成本关联键)。
|
||
# sink=None → 只留内存 steps(spike 期收口由编排器读 adapter.summary;真落库 sink 随控制面 phase-1 接)。
|
||
# 单实例对应单 agent;它与 breaker 都挂 on_reply,由框架按 middlewares 列表序串成洋葱链(trace 在外、纯旁路)。
|
||
tracer = Tier2TraceMiddleware(trace_id=game_id)
|
||
writer_model = config.build_model(
|
||
mname, max_tokens=max_tokens, thinking_budget=thinking_budget)
|
||
|
||
writer = Agent(
|
||
name="writer",
|
||
system_prompt=roles.writer_system(brief, design_text, fixture_hint=fixture_hint),
|
||
model=writer_model,
|
||
toolkit=toolkit,
|
||
# trace middleware 列在 breaker 前 → 它是更外层洋葱:先 ingest 事件(纯旁路),再进熔断巡检。
|
||
# 熔断抛 Tier2CircuitBreak 时,已 ingest 的轨迹仍在 adapter 内(可收口反查),不丢。
|
||
middlewares=[tracer, breaker],
|
||
state=_bypass_state(),
|
||
react_config=ReActConfig(max_iters=writer_max_iters),
|
||
)
|
||
|
||
breaker_tripped = None
|
||
final_text = ""
|
||
# ── 外层有界自纠循环(spike feie-001 根因 1)──
|
||
# AgentScope 2.0.2 原生 ReAct 在「模型产出无 tool_call 的纯文本回合」即退出(_agent.py:612)——
|
||
# 实测 M3 调一次 run_gates 看到 decision=fix 就产空文本收尾、循环退出(没到 max_iters、没熔断),
|
||
# = 看一次 verdict 就放弃、不自纠。纯内层 ReAct 给不了「停了再踹回去继续修」的保证。
|
||
# 故在编排层补一道有界 resume:agent 停下后,若【没真 finish】且【门没绿】且【还有预算】,
|
||
# 就带着 verdict 失败反馈再 reply 一次(AgentScope agent 跨 reply 保留 memory,等于原地续修),
|
||
# 直到 finish / 门绿 / resume 预算耗尽 / 熔断。这是 wg1「外层 Python repair」范式的回归——
|
||
# 实证表明纯自治 ReAct 会过早放弃,有界外层踹更稳,且不放松任何验收门(门仍是 judge 纯代码判)。
|
||
max_resumes = 6 # 外层 resume 上限(每次 resume 内层 ReAct 仍可多轮;总轮数受 middleware 硬熔断兜底)
|
||
kick_text = (
|
||
"开始实现这款富游戏。先 scaffold_init 起手,然后在循环里 write_source→(validate_datatable)→"
|
||
"build→run_gates→read_verdict→针对失败门 write_source 修→再 run_gates……门全绿后调 finish 交付。"
|
||
"切记:run_gates 的 verdict 是机器判的,看到 decision=fix 不要停下来收尾,要按失败门继续修。")
|
||
try:
|
||
for attempt in range(max_resumes + 1):
|
||
resp = await writer.reply(user_msg(kick_text))
|
||
final_text = text_of(resp)
|
||
# 真 finish(finish 工具组装了源工程)→ 收敛,退出外层。
|
||
if session.finished is not None:
|
||
break
|
||
# 读最近 verdict:门绿则不必再踹(理论上应已 finish,但容错处理)。
|
||
v = session.last_verdict or {}
|
||
l1 = ((v.get("layerResults") or {}).get("L1") or {})
|
||
if v.get("decision") == "accept" and l1.get("passed"):
|
||
print(f"[tier2-studio] game={game_id} 门已绿但未 finish(attempt={attempt});"
|
||
"踹一脚让它 finish。", flush=True)
|
||
kick_text = "验收门已全绿。现在直接调 finish 交付源工程,不要再改。"
|
||
continue
|
||
if attempt >= max_resumes:
|
||
print(f"[tier2-studio] game={game_id} resume 预算耗尽({max_resumes} 次)仍未收敛,停。",
|
||
flush=True)
|
||
break
|
||
# 门没绿 + agent 自行停下 → 带 verdict 失败反馈踹回去继续修(头号自纠机制)。
|
||
fb = run.verdict_feedback(v) if v else "尚无 verdict;先 run_gates 真玩取证。"
|
||
print(f"[tier2-studio] game={game_id} attempt={attempt} agent 过早停下(门未绿),"
|
||
f"带反馈 resume 续修。decision={v.get('decision')}", flush=True)
|
||
kick_text = (
|
||
"你刚才停下了,但验收门还没全绿——不要放弃。这是上次 run_gates 的失败门:\n"
|
||
f"{fb}\n"
|
||
"请在循环里:据失败门 write_source 针对性修(数据表 schema 错就先 validate_datatable 看平台要的 key),"
|
||
"build→run_gates→read_verdict,直到门绿再 finish。一步步来,先修最关键的致命门。")
|
||
except Tier2CircuitBreak as e:
|
||
# 熔断触发:据 kind 落 breakerKind(对接 tier2-verdict)。
|
||
breaker_tripped = {"kind": e.kind, "reason": e.reason}
|
||
print(f"[tier2-studio] game={game_id} 熔断 kind={e.kind}: {e.reason}", flush=True)
|
||
except Exception as e: # noqa: BLE001 —— 其余异常落结果、不静默吞(可追溯)
|
||
breaker_tripped = {"kind": None, "reason": f"未捕获异常:{type(e).__name__}: {e}"}
|
||
print(f"[tier2-studio] game={game_id} 异常:{type(e).__name__}: {e}", flush=True)
|
||
|
||
# ── L2 设计符合层 full(observe-only · 收口后算一次 · 零 LLM · 绝不参与 accept/reject)──────
|
||
# 时机:ReAct 收敛、跑过门后(无论 finish/kill/熔断)算一次。它从已落盘源工程(数据表 + src/systems/)
|
||
# 确定性算「设计声明的系统在不在 / 合成链 DAG / 订单可达 / 经济胜负自洽」四项硬结构信号 + 设计稿弱对账,
|
||
# 写进 verdict.layerResults.L2.{passed(advisory),signals,mismatches}。本阶段纪律:enforced 恒 false、
|
||
# **绝不改 decision**——只为 observe→enforce 积累信号。金标 fixture 先天满足四项判据,故金标 L2 必 passed=true,
|
||
# 这是将来提门时「金标必须过」的校准基线。失败兜底:数据表不可读 → passed=False + mismatch,绝不中断主链。
|
||
l2_passed = None
|
||
l2_signals: list[str] = []
|
||
l2_mismatches: list[str] = []
|
||
try:
|
||
l2 = run.compute_l2_signals(game_id, design_text)
|
||
l2_passed = l2.get("passed")
|
||
l2_signals = l2.get("signals") or []
|
||
l2_mismatches = l2.get("mismatches") or []
|
||
v = session.last_verdict
|
||
if isinstance(v, dict):
|
||
lr = v.setdefault("layerResults", {})
|
||
l2seg = lr.setdefault("L2", {})
|
||
l2seg["enforced"] = False # spike 期恒 observe-only(不计入 decision)
|
||
l2seg["passed"] = l2_passed # boolean|null;advisory,不影响 decision
|
||
# signals 累加(保留 harness 已写的 l2Signals,再并入 full 计算信号 + 不符项)。
|
||
prev = l2seg.get("signals") or []
|
||
merged = list(prev)
|
||
for s in (l2_signals + [f"[不符] {m}" for m in l2_mismatches]):
|
||
if s not in merged:
|
||
merged.append(s)
|
||
l2seg["signals"] = merged
|
||
print(f"[tier2-studio][L2] game={game_id} passed(advisory)={l2_passed} "
|
||
f"signals={len(l2_signals)} mismatches={len(l2_mismatches)}"
|
||
"（observe-only,不参与 decision）", flush=True)
|
||
except Exception as e: # noqa: BLE001 —— L2 整体兜底:任何异常都不中断主链、不翻 decision
|
||
print(f"[tier2-studio][L2] 设计符合层计算异常(不影响 decision):"
|
||
f"{type(e).__name__}: {e}", flush=True)
|
||
l2_mismatches = [f"L2 计算异常:{type(e).__name__}: {e}"]
|
||
|
||
# ── L3 视觉软检(observe-only · 收口后调一次 · 绝不参与 accept/reject)───────────────
|
||
# 时机:过完 L1 九门/富游戏门、ReAct 收敛后(无论 finish/kill/熔断)调一次。
|
||
# 它用 player_system 提示词 + 真截图 + 真玩取证产 score/简评,只写进 verdict.L3.score/notes。
|
||
# **不改 decision、不改 L1/runnableOk/accept**(裁决仍只看 L1 硬门)——保金标 GREEN 不被 L3 翻动(防 Goodhart)。
|
||
# 失败兜底:截不到图/模型超时/解析失败 → score=None + log,绝不中断、绝不翻 GREEN。
|
||
# 复用 writer_model(同档 M3 客户端);其 token 计入本 run 成本(L3 也是这次生成的真实开销)。
|
||
l3_score = None
|
||
l3_notes: list[str] = []
|
||
try:
|
||
l3 = await _l3_player_panel(game_id, brief, session.last_verdict, writer_model)
|
||
l3_score = l3.get("score")
|
||
l3_notes = l3.get("notes") or []
|
||
# observe-only:把 L3 结果写回最近 verdict 的 layerResults.L3(scoreOnly 恒 true,只填 score/notes)。
|
||
# 只在已有 verdict(真跑过门)时回写;verdict 缺失则 L3 结果只留在 result.l3(下方),不伪造 verdict。
|
||
# L3 升门准备(确定性恶性渲染判据):吃单元 A 的 render-sanity / render-reflects-state
|
||
# (harness 真玩落进 verdict.humanPlayability),提「漏底文本 / 空棋盘」这类**确定性**恶性失败,
|
||
# 整理进 L3.notes 并标 egregious=true(信号:建议主会话校准后提 fatal)。**仍 advisory、不改 decision**。
|
||
egr = run.derive_l3_egregious(session.last_verdict)
|
||
l3_egregious_notes = egr.get("notes") or []
|
||
if l3_egregious_notes:
|
||
l3_notes = list(l3_notes) + l3_egregious_notes # 并入软检 notes(下方写回 verdict 与 result.l3)
|
||
v = session.last_verdict
|
||
if isinstance(v, dict):
|
||
lr = v.setdefault("layerResults", {})
|
||
l3seg = lr.setdefault("L3", {})
|
||
l3seg["scoreOnly"] = True # schema const true 不变量(防 L3 退化成阻塞门)
|
||
l3seg["score"] = l3_score # number|null;observe-only,不影响 decision
|
||
# notes 累加(保留 harness 已写的 l3Notes,再并入软检备注 + 恶性渲染确定性条目)。
|
||
prev = l3seg.get("notes") or []
|
||
l3seg["notes"] = list(prev) + [n for n in l3_notes if n not in prev]
|
||
# 恶性渲染确定性判据写成结构化 advisory 段(machine-readable 升门信号;仍不改 decision/scoreOnly)。
|
||
l3seg["egregiousRenderFailure"] = {
|
||
"egregious": bool(egr.get("egregious")),
|
||
"suggestFatal": bool(egr.get("suggestFatal")),
|
||
"reasons": egr.get("reasons") or [],
|
||
}
|
||
if egr.get("egregious"):
|
||
# 仅日志告警(给主会话升门校准信号);不写任何字段进 decision/L1/runnableOk,严守 advisory。
|
||
print(f"[tier2-studio][L3] game={game_id} 检出确定性恶性渲染失败 egregious=true "
|
||
f"reasons={egr.get('reasons')} —— 建议主会话校准后提 fatal(当前 advisory,不改 decision)。",
|
||
flush=True)
|
||
except Exception as e: # noqa: BLE001 —— L3 整体兜底:任何异常都不中断主链、不翻 GREEN
|
||
print(f"[tier2-studio][L3] 软检整体异常(score=null 兜底,不影响 decision):"
|
||
f"{type(e).__name__}: {e}", flush=True)
|
||
l3_notes = [f"L3 软检整体异常:{type(e).__name__}: {e}"]
|
||
|
||
# ── 汇总结果 ──
|
||
wall_s = round(time.perf_counter() - t0, 1)
|
||
# 成本取证:RecordingChatModel.records → (in, out, cached) 汇总(M3 路 + 设计阶段;按模型分列供折¥)。
|
||
# 一次 run 可能跨多个模型客户端(设计 model + 单写 model;本期同档 mname,但各自一份 records),
|
||
# 按模型名聚合成 records_by_model 再汇总,与 cost.tokens_by_model_from_records 的入参形状对齐。
|
||
tin = tout = tcached = 0
|
||
records_by_model: dict[str, list] = {}
|
||
for m in (locals().get("design_model"), writer_model):
|
||
recs = getattr(m, "records", None) or []
|
||
for r in recs:
|
||
tin += r[0]; tout += r[1]; tcached += (r[2] if len(r) >= 3 else 0)
|
||
if recs:
|
||
# 同档多客户端(设计 + 单写)的 records 合并到同一模型名下(本期 mname 单档)。
|
||
records_by_model.setdefault(mname, []).extend(recs)
|
||
|
||
# ── 成本接线(H3):tokens_by_model → cost_for_run 折¥,写进 result['cost'] ──
|
||
# 计费参数活读取:优先 new-api /api/pricing + /api/status 权威活值;取不到则回落显式默认参数 + 告警
|
||
# (best-effort 铁律:取价失败绝不中断主链;回落口径见下)。折算本身是纯函数,绝不抛(cost.compute 内部容错)。
|
||
tokens_by_model = tokens_by_model_from_records(records_by_model)
|
||
params = fetch_pricing_params() # None = 活读取失败(已在内部 best-effort 告警)
|
||
if params is not None:
|
||
# 活读取成功:用 new-api 权威倍率折¥。
|
||
pricing, qpu, usd_rate = params["pricing"], params["qpu"], params["usd_rate"]
|
||
pricing_source = "newapi-live"
|
||
else:
|
||
# 回落:无活倍率则 pricing 空表(cost.compute 对缺 model 取保守默认 model_ratio=0 → ¥=0),
|
||
# qpu/usd_rate 用 new-api 编译默认(500000 / 7.3)。¥ 折不出真值但不中断、不抛;
|
||
# 接线方(RunRecord B2)据 pricing_source=fallback 知道这笔 ¥ 不可信(待补真倍率重算)。
|
||
pricing, qpu, usd_rate = {}, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE
|
||
pricing_source = "fallback"
|
||
print(f"[tier2-studio] game={game_id} 成本折算回落显式默认参数(pricing 空表,¥ 不可信);"
|
||
"真值待编排器据 new-api pricing 重算。", flush=True)
|
||
cost = cost_for_run(tokens_by_model, pricing, qpu, usd_rate)
|
||
# cost = {cost_rmb, tokens_by_model, by_model};附带本次折算用的计费口径(审计追溯 + 标注 ¥ 是否可信)。
|
||
cost["pricingSource"] = pricing_source
|
||
cost["qpu"] = qpu
|
||
cost["usdRate"] = usd_rate
|
||
|
||
# ── trace 收口(H1/H2):读 tracer 摘要(traceId / 步数 / 丢弃数),供编排器对账反查 ──
|
||
trace_summary = tracer.summary()
|
||
|
||
result = {
|
||
"game_id": game_id,
|
||
"model": mname,
|
||
"design_text": design_text,
|
||
"writer_final_text": final_text,
|
||
# 单写 agent 交付的源工程(finish 工具组装的 A3 形状;未 finish 则 None)。
|
||
"source_project": session.finished,
|
||
"finished": session.finished is not None,
|
||
# 最近一次 run_gates 的 verdict(tier2-verdict 形状;judge 纯代码产出,零自评;若真跑过门,
|
||
# 其 layerResults.L3.score/notes 已由上面 L3 软检 observe-only 回填)。
|
||
"last_verdict": session.last_verdict,
|
||
# L2 设计符合层结果(observe-only,绝不参与 decision;enforced 恒 false,passed 是 advisory 判定)。
|
||
# verdict 缺失时仍在此可见(不伪造 verdict);金标 fixture 应 passed=true(将来提门校准基线)。
|
||
"l2": {"passed": l2_passed, "signals": l2_signals, "mismatches": l2_mismatches},
|
||
# L3 视觉软检结果(observe-only,绝不参与 decision;verdict 缺失时仍在此可见,不伪造 verdict)。
|
||
"l3": {"score": l3_score, "notes": l3_notes},
|
||
# 四熔断触发记录(对接 verdict.breakerKind:step_cap/budget/stuck/timeout)。
|
||
"circuit_break": breaker_tripped,
|
||
"breaker_counters": {"tool_calls": breaker.tool_calls, "model_calls": breaker.model_calls},
|
||
# 成本取证(M3 路 token;扁平汇总,向后兼容既有消费方)。
|
||
"tokens": {"prompt": tin, "completion": tout, "cached": tcached},
|
||
# 成本接线(H3,新增):run 级折¥ + per-model token 明细 + 计费口径
|
||
# (cost_rmb=run 级总¥;by_model=逐档折算明细;pricingSource=newapi-live/fallback)。
|
||
# RunRecord 接线(B2)从此抽 cost_yuan / tokens_by_model 填 G4 采集字段。
|
||
"cost": cost,
|
||
# trace 收口摘要(H1/H2,新增):{traceId, steps, dropped, middlewareDropped}。
|
||
# spike 期 trace step 留在 tracer.adapter.steps 内存(sink=None);真落库 sink 随控制面 phase-1 接。
|
||
"trace": trace_summary,
|
||
"wall_s": wall_s,
|
||
"file_tree": session.file_tree(),
|
||
}
|
||
return result
|
||
|
||
|
||
def main() -> None:
|
||
"""CLI:tier2 单写 studio(真跑需 mini-desktop:chrome + esbuild;6c6g 仅静态校验)。"""
|
||
ap = argparse.ArgumentParser(description="tier2 富游戏单写 ReAct studio")
|
||
ap.add_argument("game_id")
|
||
ap.add_argument("--brief", required=True)
|
||
ap.add_argument("--model", default=None, help="M3 模型名(默认 env TIER2_MODEL / MiniMax-M3)")
|
||
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=40, help="单写 ReAct 放开的最大轮数")
|
||
ap.add_argument("--no-design", action="store_true", help="跳过阶段 1 设计")
|
||
args = ap.parse_args()
|
||
|
||
res = asyncio.run(run_studio(
|
||
args.game_id, args.brief, play_spec=None,
|
||
model_name=args.model, max_tokens=args.max_tokens,
|
||
thinking_budget=args.thinking_budget, writer_max_iters=args.max_iters,
|
||
do_design=not args.no_design))
|
||
cb = res.get("circuit_break")
|
||
print(f"\n[tier2-studio] game={res['game_id']} finished={res['finished']} "
|
||
f"wall={res['wall_s']}s tokens(in/out)={res['tokens']['prompt']}/{res['tokens']['completion']} "
|
||
f"{'熔断=' + cb['kind'] if cb else ''}")
|
||
sys.exit(0 if res["finished"] else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|