games-development-ai/docs/plans/2026-07-02-配置控制面阶段一②-cheap归并Service-plan.md

date, topic, status, 关联设计, 上级, 前置, 范围(创始人 2026-07-02 已拍三决策)

date	topic	status	关联设计	上级	前置	范围(创始人 2026-07-02 已拍三决策)
2026-07-02	配置控制面-阶段一②	§6.8 双评审已过(Opus+Codex,4 Critical+4 Important 已文内修)+ 创始人 3 决策已拍(终止闸 max_repairs 兜成本/硬切/自检 follow-up)· 待创始人终评审 → SDD 执行	docs/agent-specs/2026-06-30-配置控制面一次性按序实现-设计.md(§3.4 护城河 middleware / §4 阶段一)	docs/agent-specs/2026-06-30-配置控制面一次性按序实现-设计.md	docs/plans/2026-07-02-配置控制面阶段一-护城河middleware-plan.md(阶段一①,HEAD 10aeed3c:gate_judge.GateJudgment / RepairMiddleware / CircuitBreaker.soft_budget / app._tier2_middlewares_factory / control_plane.single_post 已落,本 plan 复用)	① cheap 软预算切 soft(与 tier2 一致),成本上界由 max_repairs=6 优雅终止兜(6 轮续修成本本就可控 ~¥2.7),三道次数/轮数闸(max_tool_calls/max_model_calls/max_iters)一起抬到 150 当纯失控兜底;② cheap 与 tier2 各起独立 Service(build_cheap_app,不合用),默认 run_fn 硬切驱动 Service、不加金丝雀旗;③ 并发端口按 session 派生(cheap play 不再固定 4320/9222,避撞)。双评审收口:C2 产物目录/回调统一后端 gameId(Service 工厂经 sidecar 维护 session_id→gameId 映射);九门自检工具留 cutover 后 follow-up(先量真实空转率)。有默认项见 Global Constraints。

配置控制面 · 阶段一② · cheap-worker 归并 Service /chat · 实施计划

For agentic workers: REQUIRED SUB-SKILL:用 superpowers:subagent-driven-development(荐)或 superpowers:executing-plans 逐任务执行。步骤用 - [ ] 复选框跟踪。每个改代码的步骤都给了完整代码,不留占位。

Goal: 把便宜档从「裸 HTTP /generate + 进程内 for-resume 循环」归并到 AgentScope Service /chat,复用阶段一① 已落的护城河 middleware(续修 RepairMiddleware / 软预算 CircuitBreaker.soft_budget / 门判 GateJudgment)。便宜档现在靠 cheap_studio.run_studio 里一圈 for attempt in range(max_resumes+1) 续修——每轮重开一次 writer.reply();归并后这圈由 on_reasoning middleware 在 finish 点接管,整局收进一个 POST,续修不再每轮重开连接,收敛判据也从「check+build 绿」升为与 tier2 同构的「九门绿」。

Architecture: 三个任务、单链依赖。T1 判据线在共用文件 worker/gate_judge.py 加 judge_cheap_verdict(把便宜档 {pass, guards} 归一成 GateJudgment、feedback 从各门 detail 拼中文;放行判据要求顶层 pass 且 guards 非空,防空 guards 假绿),并在 cheap 侧新建 cheap_gates.py 的 run_cheap_gates(stage→smoke→ensure_play_spec→play 同步包装,供续修 check 经 to_thread 调)。T2 cheap Service 壳新建 cheap_service_app.py(镜像 tier2 service/app.py 但独立进程):_cheap_tools_factory 复用 tier2 的 _extract_function_tools 抠六工具,_cheap_middlewares_factory 返 [collector, tracer, RepairMiddleware, CircuitBreakerMiddleware(soft_budget=True, 三闸=150 失控兜底)],cheap M3 走 OpenAI 兼容凭据(base+/v1),并发端口按 session 从进程内池派生;两工厂经 sidecar 把 session_id 解析回后端 gameId(C2:产物目录/评门/回调统一后端 gameId)。T3 消费方归并新建 cheap_service_driver.py 驱动 Service /chat(注册凭据→scaffold(后端 gameId)→建 agent/session→写 session→gameId 映射 sidecar→SSE 等单回合真结束→读九门 verdict + Service 收口采集 sidecar 组 run-summary),worker_service.py 的生成 run_fn 从进程内 run_studio 硬切成驱动 driver(默认 run_fn 直接换、不加金丝雀旗),result_out HMAC 回调链一字不动。T1 本地单测全绿;T2/T3 本地测结构与判据、真门端到端(要 chrome/esbuild + Redis)排 mini-desktop 窗口(cutover 前尽力验通真后端 e2e)。

Tech Stack: AgentScope 2.0.2(create_app 服务壳、extra_agent_tools/extra_agent_middlewares 工厂注入、MiddlewareBase 洋葱链、OpenAICredential(type="openai_credential"))· Python 3.12(cheap-worker/.venv,已装 agentscope 2.0.2)· pytest · 现有件复用:阶段一① 的 worker/gate_judge.py/worker/middleware.py(RepairMiddleware/CircuitBreakerMiddleware.soft_budget)/service/app.py(_extract_function_tools)/service/control_plane.py(_wait_for_turn_end)/service/infra_config.py(Redis 参数);cheap 现有 cheap_run.py(六工具底座 + scaffold/stage/smoke/ensure_play_spec/play)/cheap_toolkit.py(build_toolkit)/cheap_roles.py(build_system_prompt)/cheap_studio.py(build_trace_source/_verdict_brief)/cheap_verify.py(verify_richness)/result_out.py(七值回调契约)/worker_service.py(裸 HTTP + 队列 + HMAC)/_bootstrap.py(key + M3 装配)。

Global Constraints(全局约束 · 每个任务隐含继承)

• AgentScope 版本钉死 == 2.0.2:读的=跑的;API 疑点以已装 cheap-worker/.venv 的 2.0.2 源码为准。已核实:OpenAICredential.type == "openai_credential"(credential/_openai.py:21);工厂签名 (user_id, agent_id, session_id) -> list,不传 record;框架 middleware 前置 [InboxMiddleware, StateChange, ToolOffload] 后再 append extras(_chat.py:260-280),故 on_reasoning 链 = [InboxMiddleware, tracer, repair]、repair 仍是最内层能拦原始 finish;storage.get_agent(user_id, agent_id) 存在但 AgentData/SessionConfig 是严格 pydantic 模型、无自由字段。
• 创始人决策① cheap 软预算切 soft + 成本上界靠 max_repairs、次数/轮数闸只做失控兜底:_cheap_middlewares_factory 的 breaker 传 soft_budget=True(与 tier2 一致),超预算不 fail-closed 断链、软停交尽力产物、复用① 的 budget_exhausted(实测已花 ¥ + repairs>0)放行路。真正兜成本上界的是 max_repairs=6:续修达 6 轮即优雅终止、交尽力产物,6 轮续修成本本就可控(~¥2.7 典型、最坏 <¥10)。双评审坐实(middleware.py:315-318/303-309):on_reply 里 max_model_calls/max_tool_calls 越限的 _trip 与超时 _trip 都是无条件硬熔断(soft_budget 只护 on_model_call 的 ¥ 闸、管不到这几道),故绝不能把它们当贴脸成本尺——一旦 k 估偏一点就抢在 max_repairs 前把「优雅交尽力产物」翻成「硬中断无产物」。三道次数/轮数闸(max_tool_calls/max_model_calls/max_iters)一起抬到 150(远高于 6 轮续修包络 ~84-91,给 ~1.6× 头寸)当纯失控兜底(续修循环 bug 时才触发),不承担成本约束;外置 genconfig budget.cheap_max_tool_calls/cheap_max_model_calls/cheap_max_iters(default 150),标定后改 YAML。Codex C4 已修:CircuitBreakerMiddleware 默认 max_tool_calls=60(middleware.py:223),7 轮 ReAct 工具步会先撞——故必须显式设 150,不能漏传。
• 创始人决策② cheap 与 tier2 各起独立 Service:新建 cheap_service_app.build_cheap_app(镜像 service/app.py 但独立进程、独立端口 8300、独立 workspace 根),不与 tier2 合用;两档 credential 协议(cheap=OpenAI 兼容 base+/v1;tier2=Anthropic 原生 host 根)/ 预算档位 / 工具面各管各。
• 创始人决策③ 并发端口按 session 派生:cheap play 固定 4320/9222 在并发(≤15)下撞端口;改由续修 check 闭包从进程内端口池派生唯一端口对(base 4400/9400、池大小 16 > 上限 15;与 tier2 4330/9322、旧 cheap 4320/9222 错开)。已核实 serve-and-play.sh 与 cheap_run.play/smoke 早已收 [port] [cdpPort] 参并全程使用(shell 无需改)——只需调用方派生并传入。
• 有默认项 A(回调层保留):worker_service.py 演进成「驱动 Service /chat + reply 外收口 + result_out HMAC 回调」,不把后端 HMAC 塞进通用 Service 壳(回调契约绑死 game-cloud DifyCallbackReqVO/七值枚举,是 cheap 专属)。
• 有默认项 B(session→gameId 映射 + write_whitelist/scaffold_template 载体;C2 收口):两个 Service 工厂签名固定 (user_id, agent_id, session_id),拿不到后端 gameId 也拿不到 storage/session 记录(源码核实 _service/_chat.py:249/275-278 只传三参、不透 session_record);而后端 gameId 才是产物目录 / 评门 / 回调 trace 的统一键(C2)。因 AgentData/SessionConfig 也无自由字段可携,改用 worker↔Service 同机共享 FS sidecar——driver 在 /chat 前按 session_id 写一份会话注册表 game-runtime/games/_cheap-sessions/<session_id>.json(cheap_run.session_cfg_path,_ 前缀同 _template/_wg1-gen 惯例、非游戏目录,避免在 games/ 堆 amgen-<uuid> 空壳),内含 {external_game_id: <后端 gameId>, write_whitelist, scaffold_template, restricted};两工厂 best-effort 读它把 session_id 解析回后端 gameId,CheapSession/评门/collector 全绑 external_game_id。I1 fail-closed(Codex):sidecar 缺 external_game_id → 工厂回落 session_id(此时产物目录与 scaffold 不一致、生成会尽早失败,是响亮失败非静默;driver 恒在 /chat 前写、shared FS 同步,正常路必到);write_whitelist 受限模式(reskin/modify)由 restricted:true 标记,restricted 为真但白名单缺失 → 收窄成空集(禁写 L3、绝不回落 create 的不收窄),create 路 restricted:false+write_whitelist=None 才允许全 L3。写/读一次(session 维度唯一键)、cross-process 安全、不 abuse 类型化字段。
• 有默认项 C(判据升级走)+ 慢门超时按最坏放宽(C3 收口):cheap 收敛判据从「check+build 绿」升为「九门绿」——续修 check 闭包 = stage→smoke→ensure_play_spec→play(经 to_thread)→ judge_cheap_verdict,与 tier2 同构(九门成放行权威)。行为变化:当前 cheap 允许 check+build 绿但九门不过就收尾(收口只记录),归并后九门不过会续修。单次九门收口最坏时延 ~390s(cheap_run 的 subprocess 超时上界:stage≤90 + smoke≤120 + play≤180);门跑期间 to_thread 阻塞、reply 不产任何 AgentEvent、SSE 只剩 30s 心跳。故所有超时按最坏放宽、绝不收紧:cheap_gate_timeout_s=420(≥390 覆盖最坏)、breaker step_timeout_s 与 driver SSE idle_timeout_s 都取 cheap_repair_step_timeout_s=600(> 单次门跑 + 推理余量,治慢门误熔断/误弃单;driver idle 从 genconfig 派生、不再硬编码 300)、总墙钟 cheap_repair_wall_timeout_s=(max_repairs+1)×(cheap_gate_timeout_s+180)+300。窗口 TODO 的方向是「量真实门时延后放宽」(慢局别被自己的门跑撑爆 breaker/SSE),不是收紧。成本/空转风险见风险段。
• 有默认项 D(cheap 不加九门自检工具 · 创始人决策③ 拍板留 follow-up):保六工具(read/list/write/check/build/finish),最小改动;agent 纯文本终止后 check 闭包才第一次真跑九门,续修靠 judge_cheap_verdict 的 feedback 引导盲修。创始人 2026-07-02 决策③:本 plan 不加九门自检工具,cutover 后在 mini-desktop 窗口量真实「续修空转率」(反复 finish→九门没绿→盲修→仍没绿),据实测数据再决定是否把「给 cheap 加一个九门自检工具让 agent 先自检再终止」从 follow-up 提前——这条明确进风险段 + 窗口 TODO。cheap 的 finish 工具保留作 agent 自评内门(check+build),与 RepairMiddleware 拦的「自然终止纯文本 Msg」不冲突(同 tier2 保留 finish 工具 + RepairMiddleware 外层拦)。
• 有默认项 E(T2-b check try 兜底放工厂闭包):cheap check 的异常兜底放在工厂的 _cheap_check 闭包里(镜像① service/app.py:210-216 的 try→返 GateJudgment(passed=False)),不下沉进 RepairMiddleware 共用类(免动 tier2 回归);budget_exhausted 是纯属性比较、不包 try。
• 九门不做 MCP:设计 §4 的「九门 MCP」是远期项,①② 都缓做;cheap 门镜像 tier2 直调 cheap_run.play(经 to_thread)。
• 代理旁路必保(内网 fake-ip 红线)+ resolve 必须先于重依赖 import(I2 收口):install_proxy_bypass/resolve_base_url 把 host 并入 NO_PROXY,必须在 import agentscope/httpx(trust_env=True 默认吃 HTTP_PROXY)之前调用(client.py:51 明写此红线),否则本机 clash fake-ip(198.18.x)把请求转走 → 502。故 build_cheap_app 先 client.resolve_base_url() 再 from agentscope.app import create_app(不能先 import 后装旁路);driver 先 client.resolve_base_url() + client.install_proxy_bypass(base_url)(把本地 Service host 127.0.0.1 也并入 NO_PROXY)再 import httpx——resolve_base_url 只并 new-api host、不含 127.0.0.1,而 driver→Service 打的是 127.0.0.1:8300,不显式旁路则可能被 fake-ip 拦(Opus/Codex I2;cheap proxy 坑史更硬)。注册的 OpenAI 凭据 base 末尾带 /v1。回调侧 worker_service._NO_PROXY_OPENER 已禁代理(不动)。
• 惰性 import 红线:cheap 侧新模块顶层绝不 import agentscope/redis/fastapi/httpx;所有重依赖只在工厂体 / build_cheap_app / driver 函数体内 import,保 6c6g py_compile + 裸 import 过(cheap-worker/.venv 装了 agentscope,故 cheap 侧「import 不炸」在 .venv 上真验;6c6g 惰性只靠 py_compile + 不含顶层重 import 静态保证)。
• 中文注释 + 可追溯日志:所有代码带完整简体中文注释;续修放行/续修注入/软预算软停/九门判定/凭据注册/回调这些外部行为与错误路径必须有可追溯日志。
• 最小改动 + 复用优先:复用① 的 RepairMiddleware/CircuitBreaker.soft_budget/GateJudgment/_extract_function_tools/_wait_for_turn_end;复用 cheap 的 cheap_run/cheap_toolkit/cheap_roles/build_trace_source/verify_richness/result_out;不重写任何生成/九门/回调逻辑。cheap-worker 无独立 worker/ 包——经 _bootstrap.py 把 tier2/gen-worker 加进 sys.path 复用 tier2 框架层(worker.* / service.* / observability.* 均可 import)。
• 测试约定:tier2 侧 PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/<t>.py -v;cheap 侧测试文件顶部 sys.path.insert(0, cheap-worker/) + import _bootstrap(同现有 cheap 测试),跑 cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/<t>.py -v。def test_* + assert,mock/monkeypatch、零网络/LLM/chrome。
• 精确 git add(禁 -A):每 task 只 add 该 task 明列文件。
• 提交风格:feat(cheap): <中文> (切片一 阶段一②/TN);结尾 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>。仅创始人要求时才 commit/push。
• 落档散文标准:本 plan 散文段(Goal/Architecture/rationale/风险)按资深工程师散文写——无元叙述、无 AI 造词黑话、无套话空强调。

cheap 成本上界 = max_repairs=6;三道次数/轮数闸 = 150(纯失控兜底)

双评审共识 + 创始人决策① 的口径:兜成本上界的是 max_repairs,不是次数闸。理由(逐条读代码坐实):

1. soft 档下真正的成本约束是续修轮数:切 soft 后 ¥10 不再 fail-closed,但整局最多续修 max_repairs=6 轮即优雅终止(RepairMiddleware 在 repairs>=max_repairs 放行 finish、交尽力产物,middleware.py:570)。6 轮续修的成本本就可控——便宜档 M3 单次调用 ~¥0.015–0.05(晚局大 context 最坏 ~¥0.1),6 轮 × 单周期 ~12 次调用 × ¥0.03 ≈ ¥2.7 典型、最坏 <¥10。优雅终止即成本上界,不需要一道贴脸的次数闸当紧尺。
2. 次数/轮数闸若贴脸包络会适得其反:CircuitBreakerMiddleware.on_reply 里 max_model_calls/max_tool_calls 越限的 _trip 是无条件硬熔断(middleware.py:315-324),max_iters 撞顶产 EXCEED_MAX_ITERS;三者都不受 soft_budget 约束(soft 只护 on_model_call 的 ¥ 闸)。若把它们压到与 6 轮续修的调用/轮数包络(~84 调用 / ~91 轮)几乎重合,只要单周期实际比估计高一点,就会抢在 max_repairs 之前把「优雅交尽力产物」翻成「硬中断无产物 / EXCEED_MAX_ITERS failed」——直接废掉决策① 的 soft 优雅收尾。
3. 故三闸一起抬到 150 当纯失控兜底:max_tool_calls/max_model_calls/max_iters 全设 150(远高于 ~84-91 包络、~1.6× 头寸),只在续修循环真失控(bug 死圈烧穿 max_repairs 都没停)时才触发,平时轮不到它们当家。Codex C4:CircuitBreakerMiddleware 默认 max_tool_calls=60(middleware.py:223),7 轮 ReAct 的工具步(读手册/读起点/写/check/build/finish 各轮)会先撞 60——故必须显式设 150、不能漏传(漏了就等于 60 的贴脸闸)。
4. 外置可标定:genconfig budget.cheap_max_tool_calls / cheap_max_model_calls / cheap_max_iters(default 均 150);max_repairs 沿用 genconfig iteration.max_resumes(现值 6)。mini-desktop 量真实单局成本/轮数分布后,若 6 轮续修的实际成本远低于 ¥10 可再放宽 max_repairs、或据失控观测收紧 150,改 YAML 即可、不必改码。

---

执行顺序与依赖

graph LR
  T1[T1 · 判据线<br/>gate_judge.judge_cheap_verdict + cheap_gates.run_cheap_gates<br/>纯函数/端口透传·先行] --> T2[T2 · cheap Service 壳<br/>build_cheap_app + 工厂 + soft + 三闸兜底 + 端口池 + session→gameId sidecar]
  T2 --> T3[T3 · 消费方归并<br/>cheap_service_driver 驱动 /chat + worker_service run_fn 换驱动 + result_out 收口]

• T1 先行:judge_cheap_verdict 是 T2 续修 check 的判据、run_cheap_gates 是 check 闭包的同步门跑;纯函数 / 可 monkeypatch,零 LLM/chrome 可独立测。
• T2 依赖 T1:_cheap_check 闭包调 cheap_gates.run_cheap_gates + judge_cheap_verdict。
• T3 依赖 T2:driver 驱动 T2 起的 Service;worker_service run_fn 换成 driver。
• SDD 串行(T1→T2→T3);真门端到端(chrome/esbuild/Redis)排 mini-desktop 窗口。

---

Task 1:判据线 —— judge_cheap_verdict(gate_judge.py)+ run_cheap_gates(cheap_gates.py)

把便宜档九门 verdict({pass, guards{门:{pass,...}}})归一成与 tier2 同一个 GateJudgment 契约,让续修 middleware 的 check 两档共用。放行判据 = 便宜档顶层 pass(它已是九门 AND 后的总放行,等价 gen.mjs done 门,不放松),failed_gates 复用 _verdict_brief 口径,feedback 因 tier2 的 verdict_feedback 吃 layerResults(cheap 无)而新写——从 guards 各门的具体字段(B_uncaught.uncaught / F_wiring.expected / H_progress.checks[].why 等)拼人读中文修复方向。另在 cheap 侧建 run_cheap_gates:把收口 pipeline stage→smoke→ensure_play_spec→play 收成一个同步函数并透传 session 派生端口,供 T2 的续修 check 经 to_thread 调。

Files:

• Modify:tier2/gen-worker/worker/gate_judge.py(在 judge_tier2_verdict 之后加 _CHEAP_GATE_HINTS / _cheap_gate_detail / _cheap_verdict_feedback / judge_cheap_verdict)
• Create:cheap-worker/cheap_gates.py(run_cheap_gates)
• Test:tier2/gen-worker/tests/test_gate_judge_cheap.py(judge_cheap_verdict:绿 / 红列门 + feedback / verdict 缺失三态)
• Test:cheap-worker/tests/test_cheap_gates.py(run_cheap_gates:monkeypatch cheap_run 桩,透传端口 + stage 失败返 {})

Interfaces:

• Consumes:GateJudgment(NamedTuple)(阶段一① 现有,gate_judge.py:13-23);cheap_run.stage/smoke/ensure_play_spec/play(现有)。

• Produces:

  • judge_cheap_verdict(play_result: dict) -> GateJudgment(play_result = cheap_run.play 的 verdict {pass, guards},或 None/空)。
  • run_cheap_gates(game_id: str, port: int, cdp_port: int) -> dict(返九门 verdict {pass, guards} 或 {};端口由调用方派生)。

• Step 1:写失败测试(judge_cheap_verdict 三态)

Create tier2/gen-worker/tests/test_gate_judge_cheap.py:

"""gate_judge.judge_cheap_verdict 单测:便宜档九门 verdict 归一成 GateJudgment(绿 / 红列门+feedback / 缺失 三态)。
跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge_cheap.py -v
"""
from worker.gate_judge import GateJudgment, judge_cheap_verdict


def test_cheap_green_passes():
    # 顶层 pass=True 且各门 pass=True → passed=True、无失败门。
    v = {"pass": True, "guards": {"A_boot": {"pass": True}, "H_progress": {"pass": True}}}
    j = judge_cheap_verdict(v)
    assert isinstance(j, GateJudgment)
    assert j.passed is True
    assert j.failed_gates == []


def test_cheap_pass_but_empty_guards_not_passed():
    # I3 防假绿:顶层 pass=True 但 guards 空(play 没真跑出逐门,或被伪造 pass)→ passed=False。
    #   放行判据 = 顶层 pass **且** guards 非空;单看 pass 会把「空 verdict 强标 pass」当绿。
    for v in ({"pass": True, "guards": {}}, {"pass": True}):
        j = judge_cheap_verdict(v)
        assert j.passed is False, "pass=True 但无逐门 guards 不放行"
        assert j.failed_gates == []
        assert j.feedback and "verdict" in j.feedback  # 落 harness 说明(未跑出逐门)


def test_cheap_red_lists_failed_gates_and_feedback():
    # 顶层 pass=False + 有未过门 → passed=False、失败门入列、feedback 非空且含门中文提示与 detail。
    v = {"pass": False, "guards": {
        "A_boot": {"pass": True},
        "F_wiring": {"pass": False, "callCount": 0, "expected": ["sessionScore."], "sample": []},
        "H_progress": {"pass": False, "checks": [
            {"path": "score", "op": "increased", "before": 0, "after": 0, "pass": False, "why": "真玩应加分"}],
            "latch": {"pass": True}},
    }}
    j = judge_cheap_verdict(v)
    assert j.passed is False
    assert "F_wiring" in j.failed_gates and "H_progress" in j.failed_gates
    assert "A_boot" not in j.failed_gates
    assert j.feedback and isinstance(j.feedback, str)
    assert "F_wiring" in j.feedback and "H_progress" in j.feedback
    assert "score" in j.feedback  # H_progress detail 抽到未过 check 的 path


def test_cheap_missing_verdict_is_not_passed():
    # play 没跑出 verdict(None / 无 guards)→ passed=False、失败门空、feedback 落 harness 说明。
    for v in (None, {}, {"pass": False}):
        j = judge_cheap_verdict(v)
        assert j.passed is False
        assert j.failed_gates == []
        assert "verdict" in j.feedback  # 「九门真玩没有跑出 verdict…」

• Step 2:跑测试,确认失败

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge_cheap.py -v Expected:FAIL(ImportError: cannot import name 'judge_cheap_verdict' from 'worker.gate_judge')

• Step 3:在 gate_judge.py 加 cheap 适配器

在 tier2/gen-worker/worker/gate_judge.py 末尾(judge_tier2_verdict 之后)追加:

# ── 便宜档九门中文提示(门名 → 未过时给 agent 的修复方向;各门 detail 由 _cheap_gate_detail 补)──
#   便宜档九门与 tier2 是同一套引擎无关门(serve-and-play.sh),门名散见 verdict.json guards:
#   A_boot / B_uncaught / C_frame / D_render / E_live / F_wiring / G_input / H_progress / I_control。
_CHEAP_GATE_HINTS = {
    "A_boot": "游戏没能正常启动(A_boot):检查 index.html / bundle.iife.js 能否加载、__GameBundle 是否正确导出。",
    "B_uncaught": "运行时抛了未捕获异常(B_uncaught):按下方 uncaught 信息定位并修掉抛错的代码。",
    "C_frame": "引擎主循环没有推进(C_frame,帧未变化):确认每帧 update/render 被调用、有真实状态变更。",
    "D_render": "画面没有有效渲染(D_render,过暗或单一色):确认每帧真的画了东西(精灵/文字/形状)。",
    "E_live": "运行态不鲜活(E_live,状态无区分度):画面/状态要随时间与输入真实变化,不能是静止定帧。",
    "F_wiring": "引擎能力没接线(F_wiring,期望的引擎调用没发生):按 play-spec 的 expectedEngineCallPrefixes,"
                "用 plugins.<能力> 真正调用得分/音频/juice 等能力。",
    "G_input": "输入没有驱动游戏(G_input,控制无响应):确认按键/点击真正改变了游戏状态。",
    "H_progress": "没有真实进展(H_progress):真玩后关键指标未按预期变化,或没到达终局 latch;"
                  "核对得分/进度随真玩上升、终局态可达。",
    "I_control": "控制门未通过(I_control)。",
}


def _cheap_gate_detail(name: str, g: dict) -> str:
    """据门名从 guard detail 抽一句人读补充(各门字段不同,见 play.cdp.cjs 产出;缺字段返空,绝不抛)。"""
    try:
        if name == "B_uncaught":
            u = g.get("uncaught")
            return f" 未捕获:{str(u)[:200]}" if u else ""
        if name == "F_wiring":
            return (f" 期望前缀={g.get('expected')} 实际 callCount={g.get('callCount')}"
                    f" 样例={str(g.get('sample'))[:120]}")
        if name == "E_live":
            return f" 去重态数={g.get('distinctStates')}"
        if name == "C_frame":
            return f" f0={g.get('f0')} f1={g.get('f1')} delta={g.get('delta')}"
        if name == "D_render":
            return f" 亮度={g.get('bright')} 最大通道={g.get('maxCh')}"
        if name == "H_progress":
            parts = []
            for c in (g.get("checks") or []):
                if isinstance(c, dict) and c.get("pass") is False:
                    parts.append(f"{c.get('path')} {c.get('op')}(before={c.get('before')} "
                                 f"after={c.get('after')};{(c.get('why') or '')[:80]})")
            latch = g.get("latch") or {}
            if isinstance(latch, dict) and latch.get("pass") is False:
                parts.append(f"终局 latch 未达(after={latch.get('after')} {(latch.get('reason') or '')[:80]})")
            return (" " + "; ".join(parts)) if parts else ""
    except Exception:  # noqa: BLE001 —— 抽 detail 失败不连累 feedback 主体
        return ""
    return ""


def _cheap_verdict_feedback(v: dict) -> str:
    """把便宜档九门 verdict 的未过门拼成给 agent 的中文续修回喂。

    tier2 的 verdict_feedback 吃 layerResults 形状,cheap verdict 无 layerResults(只有 guards map),故新写。
    verdict 缺失 / 无 guards(play 没跑出 verdict)→ 落一句 harness 说明(不伪造门绿)。
    """
    v = v or {}
    guards = v.get("guards") or {}
    if not guards:
        return ("九门真玩没有跑出 verdict(可能 stage/build/smoke 失败或工程跑不起来):"
                "请确认 check 与 build 都绿、游戏能被引擎启动后再交付。")
    lines = []
    for name, g in guards.items():
        if isinstance(g, dict) and g.get("pass") is False:
            hint = _CHEAP_GATE_HINTS.get(name, f"{name} 门未通过。")
            lines.append("· " + hint + _cheap_gate_detail(name, g))
    if not lines:
        return "九门整体未放行,但未定位到具体未过门;请检查 check/build 与真玩可运行性。"
    return "以下真玩验收门未通过,请据每条修复后再交付,不要直接结束:\n" + "\n".join(lines)


def judge_cheap_verdict(play_result: dict) -> GateJudgment:
    """把便宜档 cheap_run.play 的九门 verdict({pass, guards{门:{pass,...}}})归一成 GateJudgment(与 judge_tier2_verdict 并列)。

    放行判据 = 顶层 pass **且 guards 非空**(便宜档 verdict.pass 已是九门 AND 后的总放行,等价 gen.mjs done 门;判据不放松)。
    **I3 防假绿**:只看 `bool(v.get("pass"))` 会把「play 没跑出逐门却被强标 pass」当绿——故要求 `guards` 非空
    (九门真跑必产逐门 map);空 guards + pass=True → passed=False、feedback 落 harness 说明。
    failed_gates 复用 _verdict_brief 口径(guards 里 pass is False 的门名;H_progress 嵌套已在 g.pass 层判)。
    feedback 由 _cheap_verdict_feedback 从各门 detail 拼中文(tier2 的 verdict_feedback 吃 layerResults,cheap 无,不复用)。
    play_result 为 None / 空 / 无 guards(play 没产出 verdict)→ passed=False、失败门空、feedback 落 harness 说明。
    """
    v = play_result or {}
    guards = v.get("guards") or {}
    # 放行必须「顶层 pass 且有逐门 guards」:空 guards 无从证明九门真跑过,不放行(防空 verdict 假绿)。
    passed = bool(v.get("pass")) and bool(guards)
    failed_gates = [k for k, g in guards.items() if isinstance(g, dict) and g.get("pass") is False]
    feedback = _cheap_verdict_feedback(v)
    return GateJudgment(passed=passed, failed_gates=failed_gates, feedback=feedback)

• Step 4:跑测试,确认通过

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge_cheap.py -v Expected:PASS(4 passed —— 绿 / 空 guards 假绿防护 / 红列门+feedback / 缺失)

• Step 5:写失败测试(run_cheap_gates 端口透传 + stage 失败)

Create cheap-worker/tests/test_cheap_gates.py:

"""cheap_gates.run_cheap_gates 单测:同步跑九门收口的端口透传 + 逐段桩(monkeypatch cheap_run,零真 chrome/esbuild)。
跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_gates.py -v
"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import _bootstrap  # noqa: E402,F401  仅加 sys.path(import 时不取 key、不触网)
import cheap_gates  # noqa: E402
import cheap_run  # noqa: E402


def test_run_cheap_gates_threads_ports_and_returns_verdict(monkeypatch):
    seen = {}
    monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": True, "output": ""})
    monkeypatch.setattr(cheap_run, "smoke",
                        lambda gid, port, cdp_port: seen.update(smoke=(port, cdp_port)) or {"ok": True, "state": {"targets": []}})
    monkeypatch.setattr(cheap_run, "ensure_play_spec", lambda gid, state: {"ok": True, "wrote": True, "driverType": "tap-targets"})
    monkeypatch.setattr(cheap_run, "play",
                        lambda gid, port, cdp_port: seen.update(play=(port, cdp_port)) or {"ok": True, "verdict": {"pass": True, "guards": {}}})
    out = cheap_gates.run_cheap_gates("g1", 4407, 9407)
    assert out == {"pass": True, "guards": {}}
    assert seen["smoke"] == (4407, 9407), "smoke 应收到派生端口"
    assert seen["play"] == (4407, 9407), "play 应收到派生端口"


def test_run_cheap_gates_stage_fail_returns_empty(monkeypatch):
    monkeypatch.setattr(cheap_run, "stage", lambda gid: {"ok": False, "output": "esbuild bundle 缺失"})
    called = {"smoke": 0}
    monkeypatch.setattr(cheap_run, "smoke", lambda *a, **k: called.update(smoke=called["smoke"] + 1) or {"ok": True})
    out = cheap_gates.run_cheap_gates("g1", 4407, 9407)
    assert out == {}, "stage 失败应短路返 {}(judge 判未过)"
    assert called["smoke"] == 0, "stage 失败不应再跑 smoke"

• Step 6:跑测试,确认失败

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_gates.py -v Expected:FAIL(ModuleNotFoundError: No module named 'cheap_gates')

• Step 7:写实现

Create cheap-worker/cheap_gates.py:

"""cheap_gates.py — 便宜档九门收口的同步包装(供 Service 续修 check 闭包经 to_thread 调用)。

归并后便宜档收敛判据从「check+build 绿」升为「九门绿」:RepairMiddleware 的 check 闭包在 finish 点
把工程收口跑一遍九门(stage→smoke→ensure_play_spec→play),经 worker.gate_judge.judge_cheap_verdict 归一。
本模块只做「同步 pipeline + 端口透传」,不做判据归一(那在 gate_judge,与 tier2 并列)。
端口由调用方按 session 派生传入(决策③ 并发避撞);shell-out/subprocess 阻塞,故封在同步函数里给 to_thread 跑。
"""

import cheap_run


def run_cheap_gates(game_id: str, port: int, cdp_port: int) -> dict:
    """同步跑便宜档九门收口 stage→smoke→ensure_play_spec→play,返回九门 verdict({pass, guards})或 {}(未跑出)。

    任一段失败(stage 打包不出 / play 没产 verdict)→ 返 {}(judge_cheap_verdict 判未过、feedback 落 harness 说明,
    不伪造门绿)。端口按 session 派生传入,避免并发多局撞固定 4320/9222(决策③)。
    """
    st = cheap_run.stage(game_id)
    if not st["ok"]:
        print(f"[cheap-gates] game={game_id} stage 失败(判未过):{st['output'][:300]}", flush=True)
        return {}
    sm = cheap_run.smoke(game_id, port=port, cdp_port=cdp_port)
    # ensure_play_spec 据 smoke 抓的 state 产 driver(已存在不覆盖),让九门 driven=true、E_live/H_progress 由 advisory 升致命,
    #   根治「裸 harness 无 driver → 假绿」。smoke 失败(state=None)时按保守 key-cycle 薄 spec,play 仍照跑。
    cheap_run.ensure_play_spec(game_id, sm.get("state"))
    pr = cheap_run.play(game_id, port=port, cdp_port=cdp_port)
    return pr.get("verdict") or {}

• Step 8:跑测试,确认通过

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_gates.py -v Expected:PASS(2 passed)

• Step 9:提交

git add tier2/gen-worker/worker/gate_judge.py tier2/gen-worker/tests/test_gate_judge_cheap.py cheap-worker/cheap_gates.py cheap-worker/tests/test_cheap_gates.py
git commit -m "feat(cheap): 便宜档九门判据 judge_cheap_verdict + run_cheap_gates 同步门跑(端口透传) (切片一 阶段一②/T1)"

---

Task 2:cheap Service 壳 —— build_cheap_app + 工厂 + soft 预算 + 三闸失控兜底 + 端口池 + session→gameId sidecar(cheap_service_app.py)

新建便宜档独立 Service 壳(决策②),镜像 tier2 service/app.py 的装配范式但各起进程。_cheap_tools_factory 复用 tier2 的 _extract_function_tools 从 build_toolkit 抠六工具,经会话注册表 sidecar 把 session_id 解析回后端 gameId 再绑六工具(C2)、write_whitelist 走 I1 fail-closed helper(create 路 None);_cheap_middlewares_factory 返 [collector, tracer, RepairMiddleware, CircuitBreakerMiddleware],续修 check 在 finish 点跑九门(端口从进程内池派生、决策③),breaker soft_budget=True(决策①,成本上界靠 max_repairs=6 优雅终止)+ 三闸(tool/model/iters)=150 纯失控兜底(Codex C4)+ 超时按最坏门跑 ~390s 放宽(C3)。因续修/软预算的内存态(breaker 已花 ¥、repair 续修数、tracer 摘要)都在 Service 进程里、worker 侧 driver 拿不到,加一个只读的 collector middleware 在 REPLY_END 同步 flush(C1)+ 单 POST reply 收尾把这些写进 evidence sidecar,供 T3 的 driver 组 result-out。

Files:

• Modify:cheap-worker/cheap_run.py(加 session_cfg_path(session_id) 会话注册表路径 helper —— C2 sidecar 的单一事实源,driver 写 / 工厂读同用)
• Create:cheap-worker/cheap_service_app.py(端口池 + sidecar 读(session→gameId 解析 + I1 fail-closed)+ 两工厂 + collector(C1 REPLY_END 同步 flush)+ _cheap_check + build_cheap_app(I2 旁路先于 import)+ main)
• Test:cheap-worker/tests/test_cheap_service_app.py(import 不炸 + sidecar 读兜底/external_game_id 解析/I1 fail-closed + 端口池 + 工厂结构/soft/三闸=150 + collector REPLY_END flush)

Interfaces:

• Consumes:阶段一① 的 worker.middleware.{RepairMiddleware, CircuitBreakerMiddleware, Tier2TraceMiddleware} / worker.gate_judge.{GateJudgment, judge_cheap_verdict}(T1);tier2 service.app._extract_function_tools / service.infra_config.redis_params;worker.client.resolve_base_url / worker.genconfig.get;cheap cheap_toolkit.build_toolkit / cheap_run.{game_dir, session_cfg_path} / cheap_gates.run_cheap_gates;agentscope.app.create_app(惰性)。

• Produces:

  • cheap_run.session_cfg_path(session_id) -> Path(会话注册表 game-runtime/games/_cheap-sessions/<session_id>.json;C2 sidecar 单一事实源)。
  • build_cheap_app(*, title=SERVICE_TITLE) -> FastAPI(独立 cheap Service,默认端口 8300)。
  • _cheap_tools_factory / _cheap_middlewares_factory(签名 (user_id, agent_id, session_id) -> list;经 sidecar 把 session_id 解析回后端 gameId)。
  • acquire_ports() -> (port, cdp) / release_ports(pair)(进程内端口池,base 4400/9400、size 16)。

• Step 0:给 cheap_run.py 加会话注册表路径 helper(C2 sidecar 单一事实源)

在 cheap-worker/cheap_run.py 的 wg1_game_dir 之后追加(与 _GAMES_DIR 同一路径基准,_ 前缀非游戏目录):

# 便宜档 Service 会话注册表目录(C2:driver 按 session_id 写 session→后端 gameId 映射 + 本 session 配置,
#   Service 两工厂读它把框架分配的 session_id 解析回后端 gameId)。_ 前缀 = 基建目录(同 _template/_wg1-gen),
#   非游戏目录,不会被 scaffold/list 当游戏,避免在 games/ 堆 amgen-<session uuid> 空壳。
_CHEAP_SESSIONS_DIR = _GAMES_DIR / "_cheap-sessions"


def session_cfg_path(session_id: str) -> Path:
    """本 session 的 cheap 配置 sidecar 路径(按 session_id 键;driver 在 /chat 前写、Service 工厂读)。

    内容 {external_game_id, write_whitelist, scaffold_template, restricted}:因 AgentScope 工厂签名固定
    (user_id, agent_id, session_id)、拿不到后端 gameId 也拿不到 session 记录,故用同机共享 FS sidecar 把
    「session_id → 后端 gameId + 本 session 配置」跨进程传给工厂(C2 收口)。
    """
    return _CHEAP_SESSIONS_DIR / f"{session_id}.json"

• Step 1:写失败测试(import + sidecar + 端口池 + 工厂)

Create cheap-worker/tests/test_cheap_service_app.py:

"""cheap_service_app 单测:import 惰性红线 + 会话注册表读兜底/映射解析/I1 fail-closed + 端口池派生 + 工厂结构/soft/三闸=150 + collector REPLY_END flush。
真起 Service(Redis)、真跑门(chrome)排 mini-desktop 窗口;本测只验装配零件与惰性,零网络/LLM/chrome。
跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_app.py -v
"""
import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import _bootstrap  # noqa: E402,F401
import cheap_run  # noqa: E402
import cheap_service_app as A  # noqa: E402


def test_read_session_cfg_missing_returns_empty(tmp_path, monkeypatch):
    # 注册表不存在 → {}(工厂据此回落 game_id=session_id;正常路 driver 恒在 /chat 前写、不会缺)。
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    assert A._read_session_cfg("s1") == {}


def test_read_session_cfg_reads_mapping_and_whitelist(tmp_path, monkeypatch):
    # 读会话注册表:external_game_id(后端 gameId)+ write_whitelist + scaffold_template + restricted。
    import json
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    p = tmp_path / "_cheap-sessions" / "s2.json"
    p.parent.mkdir(parents=True)
    p.write_text(json.dumps({"external_game_id": "70012", "write_whitelist": ["game-logic.js"],
                             "scaffold_template": "_template-shop", "restricted": True}), encoding="utf-8")
    cfg = A._read_session_cfg("s2")
    assert cfg["external_game_id"] == "70012"
    assert cfg["write_whitelist"] == ["game-logic.js"]
    assert cfg["scaffold_template"] == "_template-shop"
    assert cfg["restricted"] is True


def test_resolve_external_game_id_falls_back_to_session(tmp_path, monkeypatch):
    # sidecar 缺 external_game_id → 工厂回落 session_id(响亮失败,不静默用错目录)。
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    assert A._resolve_external_game_id("sX", {}) == "sX"
    assert A._resolve_external_game_id("sX", {"external_game_id": "70099"}) == "70099"


def test_port_pool_derives_distinct_pairs():
    # 端口池派生不同端口对、释放后可再取(并发避撞的机制核)。
    p1 = A.acquire_ports()
    p2 = A.acquire_ports()
    assert p1 != p2
    assert p1[0] >= A._CHEAP_PORT_BASE and p1[1] >= A._CHEAP_CDP_BASE
    A.release_ports(p1)
    A.release_ports(p2)
    p3 = A.acquire_ports()  # 释放后能再取到(不泄漏)
    assert p3 in (p1, p2)
    A.release_ports(p3)


def test_tools_factory_returns_six_tools_bound_to_external_game_id(tmp_path, monkeypatch):
    import json
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    p = tmp_path / "_cheap-sessions" / "s1.json"
    p.parent.mkdir(parents=True)
    p.write_text(json.dumps({"external_game_id": "70012"}), encoding="utf-8")  # create 路:仅映射
    tools = asyncio.run(A._cheap_tools_factory("u1", "a1", "s1"))
    assert len(tools) == 6, "便宜档六工具(read/list/write/check/build/finish)"


def test_resolve_write_whitelist_i1_fail_closed():
    # I1 纯函数:create 路(非 restricted、无白名单)→ None(不收窄,全 L3 自由写);
    #   reskin 路(restricted:true + 白名单)→ 收窄成 set;restricted 但白名单缺失 → 空集(fail-closed 禁写、不回落 None)。
    assert A._resolve_write_whitelist({}) is None
    assert A._resolve_write_whitelist({"restricted": False}) is None
    assert A._resolve_write_whitelist({"restricted": True, "write_whitelist": ["game-logic.js"]}) == {"game-logic.js"}
    assert A._resolve_write_whitelist({"restricted": True}) == set(), "restricted 但缺白名单 → 空集(禁写)"
    assert A._resolve_write_whitelist({"restricted": True, "write_whitelist": []}) == set()


def test_middlewares_factory_soft_budget_and_failsafe_caps(tmp_path, monkeypatch):
    import json
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
    p = tmp_path / "_cheap-sessions" / "s2.json"
    p.parent.mkdir(parents=True)
    p.write_text(json.dumps({"external_game_id": "70020"}), encoding="utf-8")
    mws = asyncio.run(A._cheap_middlewares_factory("u1", "a1", "s2"))
    names = [type(m).__name__ for m in mws]
    assert names == ["_CheapRunCollector", "Tier2TraceMiddleware", "RepairMiddleware", "CircuitBreakerMiddleware"]
    breaker = mws[3]
    assert breaker.soft_budget is True, "决策①:cheap 软预算切 soft"
    assert breaker.rmb_hard_limit == 10.0, "cheap ¥10(soft 档软目标,不再是硬地板)"
    # 决策① + Codex C4:成本上界靠 max_repairs=6;三道次数/轮数闸抬到 150 当纯失控兜底(非成本尺)。
    assert breaker.max_model_calls == 150, "失控兜底 150(非成本尺)"
    assert breaker.max_tool_calls == 150, "C4:必须显式设 150,否则默认 60 会先撞"
    assert breaker.step_timeout_s >= 390, "C3:单步静默阈值须 > 单次门跑最坏(~390s)"
    assert breaker.enable_rmb_gate is True
    collector = mws[0]
    assert collector._game_id == "70020", "C2:collector 绑后端 gameId(经 sidecar 解析),非 session_id"
    repair = mws[2]
    assert repair._max_repairs == 6  # genconfig iteration.max_resumes(成本上界=优雅终止)
    # budget_exhausted 读实测已花 ¥ 超上限(非预估):未超 False、超 True。
    breaker._rmb_gate_active = True
    breaker.spent_rmb = 0.0
    assert repair._budget_exhausted() is False
    breaker.spent_rmb = 10.5
    assert repair._budget_exhausted() is True


def test_collector_flushes_on_reply_end(tmp_path, monkeypatch):
    # C1:collector 在 REPLY_END 流经时**同步 flush**(不等 finally),关闭「driver 读 REPLY_END 时 sidecar 还空」竞态。
    import json
    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")

    class _FakeBreaker:
        spent_rmb, _rmb_gate_active, budget_soft_tripped = 0.42, True, False

    class _FakeRepair:
        repairs = 2

    class _FakeTracer:
        def summary(self):
            return {"steps": 3}

    class ReplyEndEvent:  # 类名即判据:collector 认 type(evt).__name__ == "ReplyEndEvent"(镜像框架事件类名)
        pass

    Collector = A._get_collector_cls()
    col = Collector(game_id="70021", breaker=_FakeBreaker(), repair=_FakeRepair(), tracer=_FakeTracer())

    async def _fake_next(**_):
        yield ReplyEndEvent()   # REPLY_END 先于 finish Msg 流经(_agent.py:615)

    async def _drive():
        seen_after = None
        async for evt in col.on_reply(None, {}, _fake_next):
            # yield REPLY_END 之前 flush 必已发生(同步、无 await)→ 此刻 sidecar 已在盘上
            seen_after = (tmp_path / "amgen-70021" / "evidence" / "service-run-summary.json").exists()
        return seen_after

    assert asyncio.run(_drive()) is True, "REPLY_END 流经时 sidecar 应已落盘(同步 flush)"
    summary = json.loads((tmp_path / "amgen-70021" / "evidence" / "service-run-summary.json").read_text(encoding="utf-8"))
    assert summary["costRmb"] == 0.42 and summary["repairs"] == 2

• Step 2:跑测试,确认失败

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_app.py -v Expected:FAIL(ModuleNotFoundError: No module named 'cheap_service_app')

• Step 3:写实现

Create cheap-worker/cheap_service_app.py:

"""cheap_service_app.py — 便宜档 · 独立 Agent Service 服务壳(镜像 tier2 service/app.py,cheap 各起进程;决策②)。

把便宜档从「裸 HTTP /generate + 进程内 for-resume」归并到 AgentScope Service /chat:六工具/续修/软预算/trace
经工厂 per-turn 注入,续修由阶段一① 的 RepairMiddleware 在 finish 点拦九门续跑(取代进程内 resume 循环)。
与 tier2 Service 各起独立进程:cheap credential 走 OpenAI 兼容路(base+/v1)、预算 soft+¥10 软目标(成本上界靠
max_repairs 优雅终止、三闸=150 失控兜底)、六工具面、并发按 session 从进程内端口池派生(避撞固定 4320/9222)。
两工厂经会话注册表 sidecar 把框架分配的 session_id 解析回后端 gameId(C2:产物目录/评门/回调统一后端 gameId)。

【惰性 import 红线】顶层绝不 import agentscope/redis/fastapi;重依赖只在 build_cheap_app / 工厂体内 import。
【代理旁路必保】build_cheap_app 启动时 client.resolve_base_url() 装 NO_PROXY,否则 M3 请求被 clash fake-ip 转走 → 502。
"""

from __future__ import annotations

import os
import queue as _queue
import sys
import threading
from pathlib import Path
from typing import TYPE_CHECKING

# cheap-worker/ 加进 sys.path(_bootstrap 再把 tier2/gen-worker 加进去,worker.*/service.*/observability.* 可 import)。
sys.path.insert(0, str(Path(__file__).resolve().parent))
import _bootstrap  # noqa: E402,F401  仅加 sys.path + key(import 时不触网)

if TYPE_CHECKING:  # pragma: no cover
    from fastapi import FastAPI

# 服务标题(OpenAPI docs 显示)。
SERVICE_TITLE = "cheap-littlejs-agent-service"

# ── 并发端口池(决策③):便宜档并发≤15,play 用固定 4320/9222 会撞;按 session 从池派生唯一端口对 ──
#   与 tier2(4330/9322)、旧 cheap 串行默认(4320/9222)错开,避免同机多 Service 撞端口。池大小 16 > 上限 15。
#   续修 check 每次 acquire 一对、finally release;单局各 check 串行(一 reply 内),并发多局各持一对,池 16 足够。
_CHEAP_PORT_BASE = int(os.environ.get("CHEAP_GATE_PORT_BASE", "4400"))
_CHEAP_CDP_BASE = int(os.environ.get("CHEAP_GATE_CDP_BASE", "9400"))
_CHEAP_PORT_POOL_SIZE = int(os.environ.get("CHEAP_GATE_POOL_SIZE", "16"))
_cheap_port_pool: "_queue.Queue | None" = None
_cheap_port_pool_lock = threading.Lock()


def _port_pool() -> "_queue.Queue":
    """惰性建端口池(线程安全):16 对 (port, cdp) 从 base 递增派生。"""
    global _cheap_port_pool
    with _cheap_port_pool_lock:
        if _cheap_port_pool is None:
            pool: "_queue.Queue" = _queue.Queue()
            for i in range(_CHEAP_PORT_POOL_SIZE):
                pool.put((_CHEAP_PORT_BASE + i, _CHEAP_CDP_BASE + i))
            _cheap_port_pool = pool
    return _cheap_port_pool


def acquire_ports(timeout: float = 900.0) -> tuple:
    """取一对空闲端口(并发满时阻塞至有空,timeout 兜底;上游并发≤15 < 池 16,正常不阻塞)。"""
    return _port_pool().get(timeout=timeout)


def release_ports(pair) -> None:
    """归还端口对(best-effort,绝不抛)。"""
    try:
        _port_pool().put_nowait(pair)
    except Exception:  # noqa: BLE001 —— 归还失败只丢一对端口,不连累生成
        pass


def _read_session_cfg(session_id: str) -> dict:
    """读本 session 的会话注册表 sidecar(driver 在 /chat 前按 session_id 写;C2 收口)。

    因 AgentScope 工厂签名固定 (user_id, agent_id, session_id)、拿不到后端 gameId 也拿不到 session 记录,
    且 AgentData/SessionConfig 无自由字段,改用 worker↔Service 同机共享 FS sidecar
    game-runtime/games/_cheap-sessions/<session_id>.json 把「session_id → 后端 gameId + 本 session 配置」传给工厂。
    缺失 / 坏 JSON → {}(工厂据此回落 game_id=session_id;正常路 driver 恒在 /chat 前写、不会缺;best-effort,绝不抛)。
    """
    import json  # noqa: PLC0415

    import cheap_run  # noqa: PLC0415

    p = cheap_run.session_cfg_path(session_id)
    if not p.exists():
        return {}
    try:
        obj = json.loads(p.read_text(encoding="utf-8"))
        return obj if isinstance(obj, dict) else {}
    except Exception as e:  # noqa: BLE001 —— 坏 sidecar 回落 session_id,不阻断
        print(f"[cheap-service] session-cfg 读失败(回落 game_id=session_id):{type(e).__name__}: {e}", flush=True)
        return {}


def _resolve_external_game_id(session_id: str, cfg: dict) -> str:
    """把框架分配的 session_id 解析回后端 gameId(C2:产物目录/评门/回调统一后端 gameId)。

    cfg.external_game_id 存在即用它;缺失(sidecar 未写/坏)→ 回落 session_id —— 此时产物目录与 driver scaffold
    的后端 gameId 目录不一致,生成会尽早失败(agent read 不到起点 / write 越界),是响亮失败、不静默用错目录。
    """
    ext = cfg.get("external_game_id")
    if ext:
        return str(ext)
    print(f"[cheap-service] ⚠ 会话注册表缺 external_game_id(session={session_id}),回落 game_id=session_id;"
          "产物目录可能与 scaffold 不一致、生成将失败——正常路 driver 恒在 /chat 前写映射。", flush=True)
    return session_id


def _resolve_write_whitelist(cfg: dict):
    """据会话注册表算写边界白名单(I1 fail-closed)。

    create 路(非 restricted)→ None(不收窄,全 L3 自由写);reskin/modify 路(restricted:true)→ 收窄成
    write_whitelist 的 set;**restricted 为真但白名单缺失/空 → 空集**(禁写 L3,绝不回落 create 的不收窄)——
    受限会话的 sidecar 若被写坏/漏字段,宁可禁写也不放开写边界(Codex I1:受限模式缺失 fail-closed)。
    """
    if not cfg.get("restricted"):
        return None  # create 路:不收窄
    wl = cfg.get("write_whitelist")
    return set(wl) if isinstance(wl, list) and wl else set()  # restricted 缺白名单 → 空集(禁写)


async def _cheap_tools_factory(user_id: str, agent_id: str, session_id: str) -> list:
    """extra_agent_tools 工厂:每回合产便宜档六工具(game_id=后端 gameId,经 sidecar 解析;C2)。

    C2:六工具的写/评门目录必须与 driver scaffold 的目录、system prompt 里的 ⟦G⟧ 路径一致——都用**后端 gameId**
    (driver scaffold 在 amgen-<后端 gameId>、prompt 写 amgen-<后端 gameId>)。工厂只拿到 session_id,故经会话注册表
    sidecar 把 session_id 解析回 external_game_id(后端 gameId)再绑 CheapSession,避免「prompt 指 A、工具锁 B」的写越界。
    write_whitelist 走 I1 fail-closed helper。复用 tier2 service.app._extract_function_tools 从 build_toolkit 的 Toolkit
    抠 FunctionTool(丢 group.mcps,九门不做 MCP)。
    """
    from service.app import _extract_function_tools  # noqa: PLC0415 —— 复用 tier2 抠取(纯函数,6c6g 安全)

    from cheap_toolkit import CheapSession, build_toolkit  # noqa: PLC0415

    cfg = _read_session_cfg(session_id)
    game_id = _resolve_external_game_id(session_id, cfg)   # C2:session_id → 后端 gameId
    write_whitelist = _resolve_write_whitelist(cfg)        # I1:create None / restricted 缺白名单则空集(禁写)
    session = CheapSession(game_id=game_id)  # 六工具据后端 gameId 管 amgen-<后端 gameId> 目录(与 scaffold/prompt 一致)
    toolkit = build_toolkit(session, write_whitelist=write_whitelist)
    return _extract_function_tools(toolkit)


def _get_collector_cls():
    """惰性定义 collector middleware 类(subclass MiddlewareBase,缓存一次;顶层不 import agentscope 保 6c6g)。"""
    global _COLLECTOR_CLS
    if _COLLECTOR_CLS is not None:
        return _COLLECTOR_CLS
    from agentscope.middleware import MiddlewareBase  # noqa: PLC0415

    class _CheapRunCollector(MiddlewareBase):
        """便宜档收口采集(cheap-only,不碰 tier2 共用类):把成本/trace/续修数写 evidence sidecar 供 driver 跨进程读。

        归并后 breaker/tracer/repair 都在 Service 进程内(per-turn),worker 侧 driver 拿不到它们的内存态;
        本 middleware 把 costRmb/rmbGate/repairs/budgetSoftTripped/traceSummary 写 game_dir(后端 gameId)/evidence/
        service-run-summary.json,driver 读它 + 九门 verdict 组 result-out(result_out 的 trace 需 costRmb/repairs 承重,
        D11 就绪分据此算)。**C1 竞态修复**:在 ReplyEndEvent 流经时**同步 flush**(在 yield 它给下游 SSE publisher 前),
        关闭「driver 读到 REPLY_END 时 Service 侧 finally 还没 flush → costRmb=0」竞态;finally 仍兜底(降级三路见 on_reply)。
        纯旁路 best-effort:采集/写盘任何异常只告警、不影响生成;放注入序最外层,finally 在内层全跑完后执行。
        """

        def __init__(self, *, game_id, breaker, repair, tracer) -> None:
            self._game_id = game_id  # C2:后端 gameId(经 sidecar 解析),与 driver 读 evidence 的目录一致
            self._breaker = breaker
            self._repair = repair
            self._tracer = tracer

        async def on_reply(self, agent, input_kwargs, next_handler):
            # C1:框架 _agent.py:615 先 yield ReplyEndEvent 再 yield finish Msg 再收尾;collector 是最外层 on_reply,
            #   在把 ReplyEndEvent yield 给 SSE publisher 前先落盘(flush 同步无 await)→ SSE 侧收到 REPLY_END 时盘上必有,
            #   关闭 driver 的 read-before-flush 竞态。此时最后一次模型调用已 accumulate、breaker.spent_rmb 为终值。
            try:
                async for evt in next_handler(**input_kwargs):
                    if type(evt).__name__ == "ReplyEndEvent":
                        self._flush()  # 正常路:REPLY_END 前落盘
                    yield evt  # 纯旁路透传,不拦截、不修改
            finally:
                # 降级三路:① 正常 REPLY_END 已 flush(此处幂等覆盖、不怕重复);② breaker 硬熔断(execute_chain 异常
                #   沿 async-for 上抛)→ 没走到 REPLY_END,finally 用当前 breaker 态兜底 flush(有部分成本);③ 进程中途崩
                #   (未走 finally)→ 盘上无 sidecar,driver 有界轮询后诚实降级 costRmb=0/degraded(不伪造)。
                self._flush()

        def _flush(self) -> None:
            try:
                import json  # noqa: PLC0415

                import cheap_run  # noqa: PLC0415

                b = self._breaker
                summary = {
                    "costRmb": round(float(getattr(b, "spent_rmb", 0.0) or 0.0), 4),
                    "rmbGate": "active" if getattr(b, "_rmb_gate_active", False) else "degraded",
                    "repairs": getattr(self._repair, "repairs", 0),
                    "budgetSoftTripped": bool(getattr(b, "budget_soft_tripped", False)),
                    "traceSummary": self._tracer.summary(),
                }
                ev = cheap_run.game_dir(self._game_id) / "evidence"
                ev.mkdir(parents=True, exist_ok=True)
                (ev / "service-run-summary.json").write_text(
                    json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
                print(f"[cheap-service] 收口采集落盘 game={self._game_id} costRmb={summary['costRmb']} "
                      f"repairs={summary['repairs']} softTripped={summary['budgetSoftTripped']}", flush=True)
            except Exception as e:  # noqa: BLE001 —— 采集 best-effort,失败绝不影响生成
                print(f"[cheap-service] 收口采集失败(忽略,不影响生成):{type(e).__name__}: {e}", flush=True)

    _COLLECTOR_CLS = _CheapRunCollector
    return _COLLECTOR_CLS


_COLLECTOR_CLS = None


async def _cheap_middlewares_factory(user_id: str, agent_id: str, session_id: str) -> list:
    """extra_agent_middlewares 工厂:每回合产 [collector, trace, 续修, 软预算熔断] 四件(镜像 tier2,cheap 独立参数)。

    续修 check = finish 点独立跑便宜档九门(cheap_gates.run_cheap_gates 经 to_thread,端口按 session 从池派生)→
    judge_cheap_verdict;异常兜底放工厂闭包 try(T2-b:镜像 tier2 app.py:210-216,不下沉 RepairMiddleware 共用类)。
    breaker soft_budget=True(决策①:¥10 转软目标、超预算软停交尽力产物);成本上界靠 max_repairs=6 优雅终止,
    三道次数/轮数闸(max_tool_calls/max_model_calls/max_iters)抬到 150 当纯失控兜底(Codex C4:不设 max_tool_calls
    会吃默认 60 先撞);超时按最坏九门 ~390s 放宽(C3)。评门 game_id = 后端 gameId(经 sidecar 解析,C2)。
    注入序(外→内):collector / tracer / repair / breaker;框架在其外另前置 InboxMiddleware(on_reasoning 每轮
    drain inbox 后透传全部 evt、不吞 finish),故 on_reasoning 链 = [Inbox, tracer, repair],repair 仍最内层能拦原始 finish。
    """
    import asyncio  # noqa: PLC0415

    import cheap_gates  # noqa: PLC0415
    import cheap_run  # noqa: PLC0415
    from worker import genconfig  # noqa: PLC0415
    from worker.gate_judge import GateJudgment, judge_cheap_verdict  # noqa: PLC0415
    from worker.middleware import (  # noqa: PLC0415
        CircuitBreakerMiddleware,
        RepairMiddleware,
        Tier2TraceMiddleware,
    )

    # C2:评门/collector/trace 全绑后端 gameId(经会话注册表 sidecar 把 session_id 解析回后端 gameId;与 driver
    #   scaffold、system prompt 的 ⟦G⟧、六工具写目录一致)。缺映射则回落 session_id(响亮失败,见 _resolve_external_game_id)。
    game_id = _resolve_external_game_id(session_id, _read_session_cfg(session_id))

    # trace sink:落 game_dir/trace.jsonl(与旧 cheap_studio 同路径);import/构造失败回落无 sink(非阻塞)。
    trace_sink = None
    try:
        from observability.trace import make_jsonl_sink  # noqa: PLC0415

        trace_sink = make_jsonl_sink(cheap_run.game_dir(game_id) / "trace.jsonl")
    except Exception as e:  # noqa: BLE001 —— trace 非阻塞:接线失败只告警、回落无 sink
        print(f"[cheap-service] trace sink 接线失败(回落无 sink,不阻断):{type(e).__name__}: {e}", flush=True)
    tracer = Tier2TraceMiddleware(trace_id=game_id, sink=trace_sink)  # C2:trace_id 用后端 gameId(与 trace 文件目录/trace.gameId 一致)

    max_repairs = genconfig.get("iteration", "max_resumes", 6)  # 成本上界=优雅终止(决策①)
    # C3:单次九门收口最坏 ~390s(cheap_run 超时上界 stage≤90 + smoke≤120 + play≤180);门跑期间 to_thread 阻塞、
    #   reply 不产 AgentEvent、SSE 只剩心跳。故 gate 基准 ≥390(取 420),单步静默/墙钟按它放宽,绝不用旧 200(< 390 会误熔断)。
    gate_timeout_s = genconfig.get("budget", "cheap_gate_timeout_s", 420)  # ≥ 最坏单次门跑,别再取 200
    repair_step_s = genconfig.get("budget", "cheap_repair_step_timeout_s", 600)  # > 单次门跑 + 推理余量(治慢门误熔断);driver idle 同源派生
    repair_wall_s = genconfig.get("budget", "cheap_repair_wall_timeout_s",
                                  (max_repairs + 1) * (gate_timeout_s + 180) + 300)  # (轮数)×(门+推理)+余量,容 N 次串行真门
    # 决策① + Codex C4:soft 档成本上界靠 max_repairs=6 优雅终止;三道次数/轮数闸抬到 150 当纯失控兜底(非成本尺)。
    #   max_tool_calls 不显式设会吃默认 60(middleware.py:223)→ 7 轮 ReAct 工具步先撞,故必须显式给 150。
    cheap_max_tool_calls = genconfig.get("budget", "cheap_max_tool_calls", 150)
    cheap_max_model_calls = genconfig.get("budget", "cheap_max_model_calls", 150)
    breaker = CircuitBreakerMiddleware(
        enable_rmb_gate=True,
        rmb_hard_limit=genconfig.get("budget", "cheap_rmb_hard_limit", 10.0),  # ¥10 现为软目标(soft 档、非硬地板)
        soft_budget=True,                        # 决策①:超预算软停交尽力产物、不 fail-closed 断链
        max_tool_calls=cheap_max_tool_calls,     # C4 失控兜底:显式 150(不设吃默认 60、会先撞)
        max_model_calls=cheap_max_model_calls,   # 失控兜底 150(非成本尺;成本上界=max_repairs)
        wall_timeout_s=repair_wall_s,            # 墙钟容纳 (max_repairs+1) 次串行真门(C3)
        step_timeout_s=repair_step_s,            # 单步静默阈值 > 单次门跑最坏(门跑期间无 evt 不误杀,C3)
    )

    async def _cheap_check(agent):
        # 独立跑九门:端口按 session 从池派生(决策③避撞),subprocess 经 to_thread 防阻塞事件循环;
        #   异常兜底(T2-b 工厂闭包 try,镜像 tier2 app.py:210-216):任何异常判未过续修,绝不穿透 on_reasoning。
        pair = acquire_ports()
        try:
            try:
                verdict = await asyncio.to_thread(cheap_gates.run_cheap_gates, game_id, pair[0], pair[1])
            except Exception as e:  # noqa: BLE001 —— 门执行任何异常都判未过续修
                print(f"[cheap-repair] 九门执行异常(判未过、续修):{type(e).__name__}: {e}", flush=True)
                return GateJudgment(passed=False, failed_gates=["cheap_gates_error"],
                                    feedback=f"九门执行异常,请确认工程可 check/build/运行后再交付:{type(e).__name__}: {e}")
        finally:
            release_ports(pair)
        return judge_cheap_verdict(verdict)

    repair = RepairMiddleware(
        check=_cheap_check,
        max_repairs=max_repairs,
        # 实测已花 ¥ 超上限(非预估软停标记):配① 的 repairs>0 保护,首个未绿 finish 必先修一次再因预算放行。
        budget_exhausted=lambda: (
            breaker._rmb_gate_active and breaker.spent_rmb >= breaker.rmb_hard_limit),
    )

    # 收口采集(cost/trace/repairs 跨进程回收):reply 收尾写 evidence/service-run-summary.json,供 T3 driver 组 result-out。
    collector = _get_collector_cls()(game_id=game_id, breaker=breaker, repair=repair, tracer=tracer)
    return [collector, tracer, repair, breaker]


def build_cheap_app(*, title: str = SERVICE_TITLE) -> "FastAPI":
    """组装便宜档独立 Agent Service(create_app + Redis storage/message_bus + 本地 workspace;决策②:与 tier2 各起进程)。

    部署入口:`from cheap_service_app import build_cheap_app; app = build_cheap_app()` 再 uvicorn(见 main)。
    重依赖(agentscope.app / redis)在此惰性 import;代理旁路必保(I2):**先 client.resolve_base_url() 装 NO_PROXY,
    再 import agentscope.app**(agentscope 牵出 httpx,trust_env=True 默认吃 HTTP_PROXY,装旁路必须先于 import,client.py:51)。
    """
    from worker import client  # noqa: PLC0415 —— 先导 client(不牵 httpx),装旁路
    # I2 代理旁路必保:装 NO_PROXY(内网 M3 请求经 clash fake-ip 会 502),**必须在 import agentscope/httpx 前**(红线)。
    client.resolve_base_url()

    from agentscope.app import create_app  # noqa: PLC0415 —— 旁路已装,此后 import 才安全
    from agentscope.app.message_bus import RedisMessageBus  # noqa: PLC0415
    from agentscope.app.storage import RedisStorage  # noqa: PLC0415
    from agentscope.app.workspace_manager import LocalWorkspaceManager  # noqa: PLC0415
    from service import infra_config  # noqa: PLC0415 —— 复用 tier2 Redis 参数(mini-infra)

    storage_params = infra_config.redis_params(for_message_bus=False)
    bus_params = infra_config.redis_params(for_message_bus=True)
    # cheap 独立 workspace 根(与 tier2 的 _service-workspaces 分开;本线生成产物另落 game-runtime/games/amgen-<id>)。
    ws_basedir = os.environ.get(
        "CHEAP_SERVICE_WORKSPACES",
        str(Path(__file__).resolve().parent / "_cheap-service-workspaces"),
    )
    print(f"[cheap-service] build_cheap_app: redis storage db{storage_params['db']} bus db{bus_params['db']} "
          f"auth={'on' if storage_params['password'] else 'off'} ws_basedir={ws_basedir}", flush=True)

    return create_app(
        storage=RedisStorage(**storage_params),
        message_bus=RedisMessageBus(**bus_params),
        workspace_manager=LocalWorkspaceManager(basedir=ws_basedir),
        extra_agent_tools=_cheap_tools_factory,
        extra_agent_middlewares=_cheap_middlewares_factory,
        custom_subagent_templates=[],
        custom_agent_cls=None,
        title=title,
    )


def main() -> None:
    """部署入口:起 uvicorn 跑便宜档 Agent Service(默认 0.0.0.0:8300,与 tier2 8200 错开)。只在 mini-desktop 跑。"""
    import uvicorn  # noqa: PLC0415

    host = os.environ.get("CHEAP_SERVICE_HOST", "0.0.0.0")
    port = int(os.environ.get("CHEAP_SERVICE_PORT", "8300"))
    print(f"[cheap-service] 启动 Agent Service:http://{host}:{port}", flush=True)
    uvicorn.run(build_cheap_app(), host=host, port=port)


if __name__ == "__main__":
    main()

• Step 4:跑测试,确认通过

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_app.py -v Expected:PASS(8 passed —— sidecar 兜底/读映射+白名单/external 回落 · 端口池派生 · 六工具 · I1 白名单 fail-closed · 工厂 soft+三闸=150+collector 绑后端 gameId+budget_exhausted · collector REPLY_END 同步 flush)

• Step 5:惰性红线自检(6c6g 静态保证:顶层无 agentscope/redis/fastapi import)

Run:cheap-worker/.venv/bin/python -m py_compile cheap-worker/cheap_service_app.py && grep -nE "^(import|from) (agentscope|redis|fastapi|httpx)" cheap-worker/cheap_service_app.py Expected:py_compile 无输出(通过);grep 无匹配(顶层无重依赖 import,惰性红线守住)。

• Step 6:提交

git add cheap-worker/cheap_run.py cheap-worker/cheap_service_app.py cheap-worker/tests/test_cheap_service_app.py
git commit -m "feat(cheap): 便宜档独立 Service 壳 build_cheap_app + 工厂(soft 预算/三闸失控兜底/端口池/session→gameId sidecar/收口采集) (切片一 阶段一②/T2)"

---

Task 3:消费方归并 —— cheap_service_driver 驱动 /chat + worker_service run_fn 换驱动 + result_out 收口

把 worker_service.py 生成路的进程内 run_studio 换成驱动 T2 起的 Service。新建 cheap_service_driver.py:注册 OpenAI 兼容凭据 → 建 agent(system prompt 的 ⟦G⟧=后端 gameId)/session → scaffold(后端 gameId,起点落 amgen-<后端 gameId>)+ 写 session→gameId 映射注册表 sidecar → 设 BYPASS → 发 kick → 复用 tier2 _wait_for_turn_end 等这一次(内部续修多轮)回合真结束 → 读九门 verdict + Service 收口采集 sidecar 组一份与 run_studio 同形的 run-summary → reply 外做非阻塞丰富度评分。C2 收口:产物目录/评门/回调 trace 全用后端 gameId(不用框架分配的 session_id),Service 两工厂经会话注册表把 session_id 解析回后端 gameId——否则 prompt 指 amgen-<后端 gameId>、六工具却锁 amgen-<session_id>,首轮 read 起点失败 + write 越界(Codex C2 实据)。worker_service.py 只改一处:生成路默认 run_fn 从进程内 run_studio 硬切成 _service_run_fn(驱动 driver),不加 env 金丝雀旗(创始人决策②:硬换全量切;_default_run_fn 保留作对照/回滚,改回默认一行即退);result_out HMAC 回调链、队列、幂等、modify 路一字不动。这样便宜档整局收进一个 POST,续修由 Service 端 middleware 接管,回调契约保留在 worker 包装层。真端到端 cutover 就绪判定要严(mini-desktop 真后端 e2e 尽力验通,见窗口 TODO)。

Files:

• Create:cheap-worker/cheap_service_driver.py(_cheap_credential_payload / _write_session_cfg / _read_last_cheap_verdict / _read_service_run_summary / _read_driver_type / _build_summary / _failed_summary / drive_cheap_generation)
• Modify:cheap-worker/worker_service.py(加 _service_run_fn;WorkerState.__init__ 默认 run_fn = run_fn or _service_run_fn,替原 _default_run_fn)
• Test:cheap-worker/tests/test_cheap_service_driver.py(凭据体 base+/v1+type · _build_summary succeeded/failed · trace 端到端契约(I4)· session→后端 gameId 注册表往返(C2)· service-run-summary 有界轮询(C1)· fake-SSE driver 契约 · run_fn 默认硬切)

Interfaces:

• Consumes:tier2 service.control_plane._wait_for_turn_end(现有 SSE 等回合,-> {ended, reason, endEvent})/ worker.config.build_context_config / worker.genconfig.get / worker.client.{resolve_base_url, install_proxy_bypass, get_api_key}(I2 旁路);cheap cheap_run.{scaffold, game_dir, wg1_game_dir, session_cfg_path} / cheap_roles.build_system_prompt / cheap_studio.{build_trace_source, _verdict_brief} / cheap_verify.verify_richness / worker.gate_judge.judge_cheap_verdict(T1);result_out.build_result_out(现有,消费 summary,不改)。

• Produces:

  • drive_cheap_generation(job: dict, *, base_url: str | None = None, user_id: str = "cheap") -> tuple[dict, Path](返 (run-summary, game_dir),与旧 _default_run_fn 同签名契约,process_job 零改动消费)。
  • worker_service._service_run_fn(job) -> tuple[dict, Path](同步壳:asyncio.run(drive_cheap_generation(job)))。

• Step 1:写失败测试(凭据体 + _build_summary + sidecar + run_fn 默认)

Create cheap-worker/tests/test_cheap_service_driver.py:

"""cheap_service_driver 单测:OpenAI 兼容凭据体(base+/v1) + _build_summary 组 result_out 可吃形 + sidecar 往返 +
worker_service 默认 run_fn 已切到驱动 Service(不再进程内 run_studio)。真起 Service/真门排 mini-desktop 窗口。
跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_driver.py -v
"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import _bootstrap  # noqa: E402,F401
import cheap_run  # noqa: E402
import cheap_service_driver as D  # noqa: E402
import result_out  # noqa: E402
import worker_service as W  # noqa: E402


def test_credential_payload_openai_compatible_with_v1(monkeypatch):
    # 便宜档 M3 走 OpenAI 兼容路:type=openai_credential、base_url 带 /v1、api_key 非空。
    monkeypatch.setattr(D, "_resolve_base", lambda: "http://100.64.0.8:3000")
    monkeypatch.setattr(D, "_resolve_key", lambda: "sk-test")
    payload = D._cheap_credential_payload()
    assert payload["data"]["type"] == "openai_credential"
    assert payload["data"]["base_url"].endswith("/v1")
    assert payload["data"]["api_key"] == "sk-test"


def test_build_summary_shape_feeds_result_out(tmp_path, monkeypatch):
    # _build_summary 组的 summary 能被 result_out.build_result_out 吃出 succeeded + trace(costRmb/repairs)。
    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
    gd = tmp_path / "amgen-g9"
    gd.mkdir(parents=True)
    (gd / "bundle.iife.js").write_text("var x=1; window.__GameBundle={};", encoding="utf-8")
    monkeypatch.setattr(D, "_read_driver_type", lambda gid: "tap-targets")
    verdict = {"pass": True, "guards": {"A_boot": {"pass": True}, "H_progress": {"pass": True}}}
    svc = {"costRmb": 0.42, "rmbGate": "active", "repairs": 2, "budgetSoftTripped": False}
    turn = {"ended": True, "reason": "REPLY_END"}
    summary = D._build_summary("g9", "点击得分小游戏", verdict, svc, turn, t0=0.0)
    assert summary["ok"] is True and summary["finished"] is True
    assert summary["verdict"]["pass"] is True
    assert summary["attempts"] == 3  # repairs=2 → attempts=repairs+1(result_out repairs=attempts-1 反推回 2)
    assert summary["costRmb"] == 0.42
    assert summary["verdictFull"] == verdict and summary["driverType"] == "tap-targets"
    payload = result_out.build_result_out({"traceId": "t", "brief": "点击"}, summary, cheap_run.game_dir("g9"))
    assert payload["status"] == "succeeded"
    assert payload["trace"]["repairs"] == 2 and payload["trace"]["cost"]["totalRmb"] == 0.42
    assert payload["trace"]["gatespec"]["driver"] == "tap-targets"


def test_build_summary_no_verdict_is_failed(tmp_path, monkeypatch):
    # 回合没跑出 verdict(续修耗尽/软停无绿)→ ok=False,result_out 走 failed(七值枚举)。
    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
    (tmp_path / "amgen-g8").mkdir(parents=True)  # 无 bundle
    monkeypatch.setattr(D, "_read_driver_type", lambda gid: None)
    summary = D._build_summary("g8", "brief", None, {"costRmb": 0.1, "rmbGate": "active", "repairs": 6},
                               {"ended": True, "reason": "REPLY_END"}, t0=0.0)
    assert summary["ok"] is False
    payload = result_out.build_result_out({"traceId": "t"}, summary, cheap_run.game_dir("g8"))
    assert payload["status"] == "failed"
    assert payload["failureReason"] in result_out._FAILURE_REASONS


def test_session_cfg_roundtrip_maps_session_to_backend_gameid(tmp_path, monkeypatch):
    # C2:driver 按 session_id 写会话注册表(含 external_game_id=后端 gameId),Service 工厂读 session_id → 解析回后端 gameId。
    import cheap_service_app as A
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    # create 路:driver 写映射、白名单 None → restricted False。
    D._write_session_cfg("sess-abc", external_game_id="70012", write_whitelist=None, scaffold_template=None)
    cfg = A._read_session_cfg("sess-abc")
    assert A._resolve_external_game_id("sess-abc", cfg) == "70012", "session_id → 后端 gameId"
    assert A._resolve_write_whitelist(cfg) is None, "create 路不收窄"
    # reskin 路:driver 传白名单 → restricted True + 工厂收窄成 set。
    D._write_session_cfg("sess-def", external_game_id="70013", write_whitelist={"game-logic.js"})
    cfg2 = A._read_session_cfg("sess-def")
    assert cfg2["restricted"] is True
    assert A._resolve_write_whitelist(cfg2) == {"game-logic.js"}


def test_build_summary_trace_contract_end_to_end(tmp_path, monkeypatch):
    # I4:_build_summary 组的 summary 喂 build_result_out 后,trace 七项 + sevenGateVerdict + gatespec.driver +
    #   cost.totalRmb 齐备,engineBundle 承重;不空口说「result_out 零改动」,而是端到端断言其当前消费字段。
    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
    gd = tmp_path / "amgen-70012"
    gd.mkdir(parents=True)
    (gd / "bundle.iife.js").write_text("var x=1; window.__GameBundle={};", encoding="utf-8")
    monkeypatch.setattr(D, "_read_driver_type", lambda gid: "tap-targets")
    verdict = {"pass": True, "guards": {"A_boot": {"pass": True}, "H_progress": {"pass": True}}}
    svc = {"costRmb": 0.42, "rmbGate": "active", "repairs": 2, "budgetSoftTripped": False}
    summary = D._build_summary("70012", "点击得分小游戏", verdict, svc,
                               {"ended": True, "reason": "REPLY_END"}, t0=0.0)
    payload = result_out.build_result_out({"traceId": "t-1", "brief": "点击"}, summary, cheap_run.game_dir("70012"))
    assert payload["status"] == "succeeded"
    assert payload["traceId"] == "t-1"                       # 回调关联键 = job.traceId(非 gameId)
    assert "__GameBundle" in payload["engineBundle"]         # engineBundle 承重入 feed
    tr = payload["trace"]
    for k in ("pass", "repairs", "wallS", "models", "attempts", "gameId", "stage"):
        assert k in tr, f"trace 必填七项缺 {k}"
    assert tr["gameId"] == "70012"                            # C2:trace.gameId = 后端 gameId(诊断可关联)
    assert tr["repairs"] == 2 and tr["cost"]["totalRmb"] == 0.42
    assert tr["sevenGateVerdict"]["pass"] is True and "H_progress" in tr["sevenGateVerdict"]["guards"]
    assert tr["gatespec"]["driver"] == "tap-targets"         # 键名 driver(非 driverType,后端 hasDriver 据此)


def test_read_service_run_summary_bounded_poll(tmp_path, monkeypatch):
    # C1:sidecar 在 wait 窗口内出现 → 读到;窗口内始终缺失 → 诚实降级 {}(不阻塞、不伪造)。
    import json
    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
    assert D._read_service_run_summary("miss", wait_s=0.05) == {}, "缺失且过期 → 降级 {}"
    ev = tmp_path / "amgen-hit" / "evidence"
    ev.mkdir(parents=True)
    (ev / "service-run-summary.json").write_text(json.dumps({"costRmb": 0.3, "repairs": 1}), encoding="utf-8")
    got = D._read_service_run_summary("hit", wait_s=0.05)
    assert got["costRmb"] == 0.3 and got["repairs"] == 1


def test_drive_cheap_generation_fake_sse(tmp_path, monkeypatch):
    # 契约(fake-SSE):mock httpx + _wait_for_turn_end,驱动真 drive_cheap_generation —— 断言 C2(scaffold 用后端
    #   gameId、注册表写 external_game_id=后端 gameId)+ 回合真结束读 verdict/sidecar 组 succeeded + 回合未结束诚实降级。
    import asyncio
    import json
    import service.control_plane as CP

    monkeypatch.setattr(cheap_run, "game_dir", lambda gid: tmp_path / f"amgen-{gid}")
    monkeypatch.setattr(cheap_run, "wg1_game_dir", lambda gid: tmp_path / "_wg1-gen" / gid)
    monkeypatch.setattr(cheap_run, "session_cfg_path", lambda sid: tmp_path / "_cheap-sessions" / f"{sid}.json")
    scaffolded = {}
    monkeypatch.setattr(cheap_run, "scaffold", lambda gid, tpl=None: scaffolded.update(gid=gid) or {"ok": True, "output": ""})
    # richness 非阻塞:stub 成同步返回(避免真 LLM)。
    async def _fake_richness(gid, *, brief=""): return {"score": 5, "max": 8, "degraded": False}
    monkeypatch.setattr("cheap_verify.verify_richness", _fake_richness)
    # 无网络/无 key 也能跑:stub 凭据体 + key env(真凭据/旁路真跑在 mini-desktop 窗口验)。
    import _bootstrap
    monkeypatch.setattr(_bootstrap, "ensure_api_key_env", lambda: None)
    monkeypatch.setattr(D, "_cheap_credential_payload",
                        lambda: {"data": {"type": "openai_credential", "api_key": "sk", "base_url": "http://x/v1"}})

    class _Resp:
        def __init__(self, d): self._d = d
        def json(self): return self._d

    class _FakeHttp:
        def __init__(self, *a, **k): pass
        async def __aenter__(self): return self
        async def __aexit__(self, *a): return False
        async def post(self, url, **k):
            return _Resp({"credential_id": "c1"} if url.endswith("/credential/") else
                         {"agent_id": "a1"} if url.endswith("/agent/") else
                         {"session_id": "sess-9"} if url.endswith("/sessions/") else {})
        async def patch(self, url, **k): return _Resp({})

    import httpx
    monkeypatch.setattr(httpx, "AsyncClient", _FakeHttp)

    # 造后端 gameId=70012 的 on-disk 产物(scaffold 后 Service 侧应写在这里,fake 直接铺好)。
    (tmp_path / "amgen-70012").mkdir(parents=True)
    (tmp_path / "amgen-70012" / "bundle.iife.js").write_text("window.__GameBundle={};", encoding="utf-8")
    vev = tmp_path / "_wg1-gen" / "70012" / "evidence"; vev.mkdir(parents=True)
    (vev / "verdict.json").write_text(json.dumps({"pass": True, "guards": {"A_boot": {"pass": True}}}), encoding="utf-8")
    sev = tmp_path / "amgen-70012" / "evidence"; sev.mkdir(parents=True)
    (sev / "service-run-summary.json").write_text(json.dumps({"costRmb": 0.5, "rmbGate": "active", "repairs": 1}), encoding="utf-8")

    # ① 回合真结束(REPLY_END)→ 读 verdict/sidecar 组 succeeded 形。
    async def _ended(*a, **k): return {"ended": True, "reason": "REPLY_END", "endEvent": {}}
    monkeypatch.setattr(CP, "_wait_for_turn_end", _ended)
    summary, gdir = asyncio.run(D.drive_cheap_generation({"gameId": 70012, "traceId": "t9", "brief": "点球"}))
    assert scaffolded["gid"] == "70012", "C2:scaffold 用后端 gameId(str),非 session_id"
    cfg = json.loads((tmp_path / "_cheap-sessions" / "sess-9.json").read_text(encoding="utf-8"))
    assert cfg["external_game_id"] == "70012", "C2:注册表写 session→后端 gameId 映射"
    assert summary["ok"] is True and summary["gameId"] == "70012" and summary["costRmb"] == 0.5
    assert gdir == cheap_run.game_dir("70012")

    # ② 回合未真结束(SSE 超时)→ 诚实降级(不卡死),summary 仍组出、reason 反映未结束。
    async def _not_ended(*a, **k): return {"ended": False, "reason": "total_timeout", "endEvent": None}
    monkeypatch.setattr(CP, "_wait_for_turn_end", _not_ended)
    s2, _ = asyncio.run(D.drive_cheap_generation({"gameId": 70012, "traceId": "t9b", "brief": "点球"}))
    assert s2["stoppedReason"] == "total_timeout"


def test_worker_default_run_fn_is_service_driver():
    # worker_service 默认 run_fn 已切到驱动 Service(不再进程内 run_studio)。
    state = W.WorkerState()
    assert state.run_fn is W._service_run_fn

• Step 2:跑测试,确认失败

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_driver.py -v Expected:FAIL(ModuleNotFoundError: No module named 'cheap_service_driver';及 worker_service 无 _service_run_fn)

• Step 3:写 driver 实现

Create cheap-worker/cheap_service_driver.py:

"""cheap_service_driver.py — 便宜档 · 单 POST 消费方(镜像 tier2 bootstrap + control_plane.single_post,cheap 收口)。

worker_service.py 收 §6.1 job 后,不再进程内 run_studio,而是驱动 cheap Service /chat:
注册 OpenAI 兼容凭据 → 建 agent(system prompt 的 ⟦G⟧=后端 gameId)/session → scaffold(后端 gameId,起点落
amgen-<后端 gameId>)+ 写 session→gameId 映射注册表 sidecar → 设 BYPASS → 发 kick → SSE 等这一次(内部续修多轮)
回合真结束 → 读九门 verdict + Service 收口采集 sidecar 组 run-summary → reply 外非阻塞丰富度评分。续修/门判/软预算
全在 Service 端 middleware(阶段一① 复用),消费方只发一次 POST。**C2**:产物目录/评门/回调 trace 全用后端 gameId,
Service 两工厂经会话注册表把框架分配的 session_id 解析回后端 gameId(工厂拿不到后端 gameId,故靠 sidecar 桥接)。

【惰性 import 红线】顶层零重依赖;httpx/agentscope 牵出的件在函数体内 import。
【代理旁路必保(I2)】**先 client.resolve_base_url()+install_proxy_bypass(base_url) 再 import httpx**(装 NO_PROXY 含
  new-api host 与本地 Service 127.0.0.1);回调侧旁路由 worker_service._NO_PROXY_OPENER 兜。
"""

from __future__ import annotations

import os
import time
from pathlib import Path


def _resolve_base() -> str:
    """new-api host 根(已装代理旁路);cheap 凭据在其后补 /v1(OpenAI 兼容路)。"""
    from worker import client  # noqa: PLC0415

    return client.resolve_base_url()


def _resolve_key() -> str:
    """NEWAPI_KEY(_bootstrap 已从凭据档注入 env;client.get_api_key 从 env 读)。"""
    from worker import client  # noqa: PLC0415

    return client.get_api_key()


def _cheap_credential_payload() -> dict:
    """组注册 cheap M3 凭据(POST /credential/)的请求体 —— OpenAI 兼容路,base 末尾补 /v1(决策②)。

    与 tier2(anthropic_credential,host 根)不同:cheap 走 OpenAI 兼容 /v1/chat/completions,故
    type=openai_credential、base_url 带 /v1(与 config.build_model_openai 的 /v1 补全同口径)。
    """
    base = _resolve_base().rstrip("/")
    if not base.endswith("/v1"):
        base = base + "/v1"
    return {"data": {"type": "openai_credential", "api_key": _resolve_key(), "base_url": base}}


def _write_session_cfg(session_id: str, *, external_game_id: str,
                       write_whitelist=None, scaffold_template=None) -> None:
    """写本 session 的会话注册表 sidecar(C2:Service 两工厂读它把 session_id 解析回后端 gameId + 取 write_whitelist)。

    按 session_id 键写 game-runtime/games/_cheap-sessions/<session_id>.json(worker↔Service 同机共享 FS);
    **external_game_id = 后端 gameId 恒写**(工厂据它绑六工具/评门/collector 目录,与 driver scaffold 一致)。
    restricted = 是否受限写(create 路 False + write_whitelist=None;reskin/modify 路 True + 白名单;I1 fail-closed
    依赖 restricted 标记)。best-effort:写失败只告警(Service 侧读缺失 → 回落 session_id、生成响亮失败)。
    """
    import json  # noqa: PLC0415

    import cheap_run  # noqa: PLC0415

    try:
        restricted = bool(write_whitelist)
        wl = sorted(write_whitelist) if write_whitelist else None
        p = cheap_run.session_cfg_path(session_id)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps({
            "external_game_id": str(external_game_id),
            "write_whitelist": wl,
            "scaffold_template": scaffold_template,
            "restricted": restricted,
        }, ensure_ascii=False), encoding="utf-8")
    except Exception as e:  # noqa: BLE001 —— sidecar best-effort,写失败 Service 侧回落 session_id
        print(f"[cheap-driver] session-cfg 写失败(Service 侧将回落 game_id=session_id、生成响亮失败):"
              f"{type(e).__name__}: {e}", flush=True)


def _read_last_cheap_verdict(game_id: str):
    """读服务端最后一次九门 play 已落的 verdict(_wg1-gen/<id>/evidence/verdict.json)。缺失/坏 → None(判未过,不伪造)。"""
    import json  # noqa: PLC0415

    import cheap_run  # noqa: PLC0415

    vp = cheap_run.wg1_game_dir(game_id) / "evidence" / "verdict.json"
    if not vp.exists():
        return None
    try:
        return json.loads(vp.read_text(encoding="utf-8"))
    except Exception as e:  # noqa: BLE001
        print(f"[cheap-driver] verdict 读失败(按未产出):{type(e).__name__}: {e}", flush=True)
        return None


def _read_service_run_summary(game_id: str, *, wait_s: float = 5.0) -> dict:
    """读 Service 收口采集 sidecar(game_dir/evidence/service-run-summary.json:costRmb/rmbGate/repairs/…)。

    C1 竞态兜底:Service 侧 collector 已在 yield REPLY_END 前**同步 flush**,正常 driver 读 REPLY_END 时盘上必有;
    但为防极端时序(SSE 到达早于文件系统可见 / collector 走 finally 兜底路),这里再加**有界轮询**——最多等 wait_s
    (每 100ms 探一次「存在且可解析」),仍读不到才诚实降级 {}(driver 落 costRmb=0/degraded,不伪造)。竞态窗口是
    ms 级,短轮询即闭合;这是 M3b「result_out 只产 trace.cost → D11 三维恒中性」坑的归并版防线。
    """
    import json  # noqa: PLC0415
    import time as _time  # noqa: PLC0415

    import cheap_run  # noqa: PLC0415

    p = cheap_run.game_dir(game_id) / "evidence" / "service-run-summary.json"
    deadline = _time.time() + max(0.0, wait_s)
    while True:
        if p.exists():
            try:
                obj = json.loads(p.read_text(encoding="utf-8"))
                if isinstance(obj, dict):
                    return obj
            except Exception:  # noqa: BLE001 —— 半成品/坏 JSON:轮询期内再等,过期后按空降级
                pass
        if _time.time() >= deadline:
            print(f"[cheap-driver] service-run-summary 有界轮询({wait_s:.1f}s)未读到,诚实降级 costRmb=0/degraded。",
                  flush=True)
            return {}
        _time.sleep(0.1)


def _read_driver_type(game_id: str):
    """读 ensure_play_spec 产的 driver 类型(_wg1-gen/<id>/play-spec.json.driver.type);缺失 → None(trace.gatespec 省略)。"""
    import json  # noqa: PLC0415

    import cheap_run  # noqa: PLC0415

    p = cheap_run.wg1_game_dir(game_id) / "play-spec.json"
    if not p.exists():
        return None
    try:
        return ((json.loads(p.read_text(encoding="utf-8")) or {}).get("driver") or {}).get("type")
    except Exception:  # noqa: BLE001
        return None


def _build_summary(game_id: str, brief: str, verdict, svc: dict, turn: dict, t0: float) -> dict:
    """据九门 verdict + Service 收口采集 sidecar 组一份与 run_studio 同形的 run-summary(供 result_out.build_result_out)。

    复用 cheap_studio 的 _verdict_brief(verdict 简报)+ build_trace_source(9d-trace 源维度 verdictFull/driverType/models/stage);
    costRmb/rmbGate/repairs 来自 Service 收口采集(worker 进程拿不到 Service 内存态,靠 sidecar 跨进程)。
    attempts = repairs+1(result_out 的 trace.repairs = max(0, attempts-1) 会反推回 repairs)。
    """
    import cheap_studio  # noqa: PLC0415
    from worker.gate_judge import judge_cheap_verdict  # noqa: PLC0415

    j = judge_cheap_verdict(verdict or {})
    vb = cheap_studio._verdict_brief(verdict) if verdict else None
    svc = svc or {}
    repairs = svc.get("repairs")
    attempts = (repairs + 1) if isinstance(repairs, int) else 1
    driver_type = _read_driver_type(game_id)
    stage = "play" if verdict is not None else "code"  # 简化:有 verdict=跑到 play;无=最远到写码(细分阶段列 follow-up)
    model = cheap_studio._bootstrap.SPIKE_MODEL
    summary = {
        "ok": j.passed,
        "gameId": game_id,
        "brief": brief,
        "model": model,
        "finished": j.passed,
        "attempts": attempts,
        "verdict": vb,
        "costRmb": round(float(svc.get("costRmb", 0.0) or 0.0), 4),
        "rmbGate": svc.get("rmbGate", "degraded"),
        "budgetSoftTripped": bool(svc.get("budgetSoftTripped", False)),
        "wallSec": round(time.time() - t0, 1),
        "stoppedReason": (turn.get("reason") if not turn.get("ended") else "turn_ended"),
    }
    # 9d-trace 源维度(verdictFull/driverType/models/stage);对照 None 时各键自然降级(result_out 省略)。
    summary.update(cheap_studio.build_trace_source(verdict, driver_type, attempts, stage, model))
    return summary


def _failed_summary(game_id: str, reason: str) -> dict:
    """早失败(scaffold 失败等)时的最小 failed summary(verdict None → result_out 走 failed 路)。"""
    import cheap_studio  # noqa: PLC0415

    print(f"[cheap-driver] game={game_id} 早失败:{reason}", flush=True)
    return {
        "ok": False, "gameId": game_id, "model": cheap_studio._bootstrap.SPIKE_MODEL,
        "finished": False, "attempts": 1, "verdict": None,
        "costRmb": 0.0, "rmbGate": "degraded", "stage": "scaffold", "stoppedReason": reason,
    }


async def drive_cheap_generation(job: dict, *, base_url: str | None = None, user_id: str = "cheap"):
    """驱动 cheap Service /chat 跑一局便宜档生成(单 POST + 洋葱内续修),返回 (run-summary, game_dir)。

    与旧 worker_service._default_run_fn 同契约(process_job 零改动消费)。create 路默认:通用 scaffold + 通用系统提示 +
    write_whitelist=None(对齐现 _default_run_fn 的 run_studio(game_id, brief) 无 scaffold/whitelist)。
    """
    import asyncio  # noqa: PLC0415

    from worker import client, config, genconfig  # noqa: PLC0415 —— client 先导(不牵 httpx),装旁路

    base_url = base_url or os.environ.get("CHEAP_SERVICE_URL", "http://127.0.0.1:8300")
    # I2 代理旁路必保(**必须先于 import httpx**):resolve_base_url 把 new-api host 并入 NO_PROXY(装凭据/M3 侧);
    #   driver→Service 打的是 127.0.0.1:8300、resolve_base_url 不含它,故再 install_proxy_bypass(base_url) 把本地
    #   Service host 也并入 NO_PROXY,避免 trust_env=True 的 httpx 被 clash fake-ip 拦本地环回(Opus/Codex I2)。
    client.resolve_base_url()
    client.install_proxy_bypass(base_url)

    import httpx  # noqa: PLC0415 —— 旁路已装,此后 import 才安全

    import _bootstrap  # noqa: PLC0415
    import cheap_run  # noqa: PLC0415
    import cheap_verify  # noqa: PLC0415
    from cheap_roles import build_system_prompt  # noqa: PLC0415
    from service.control_plane import _wait_for_turn_end  # noqa: PLC0415 —— 复用 tier2 SSE 等回合

    game_id = job.get("gameId") or job.get("job_id")
    # 真后端 gameId 是 Java Long → JSON 数字 → int;cheap_run 的 subprocess/路径要求 str(同 worker_service 边界归一)。
    game_id = str(game_id) if game_id is not None else None
    brief = job.get("brief") or ""
    t0 = time.time()

    # create 路默认参(对齐现 _default_run_fn):通用 scaffold(无 per-genre 模板)、通用系统提示、write_whitelist=None。
    scaffold_template = None
    scaffold_desc = None
    write_whitelist = None
    # 失控兜底轮数(与三闸 150 对齐;成本上界=max_repairs=6 优雅终止,而非轮数闸;放开防单 POST 多续修被 EXCEED_MAX_ITERS 提前切断)。
    max_iters = genconfig.get("budget", "cheap_max_iters", 150)
    max_tokens = genconfig.get("model", "max_tokens", 16000)

    _bootstrap.ensure_api_key_env()  # 保证 NEWAPI_KEY(装凭据要用)

    headers = {"X-User-Id": user_id}
    async with httpx.AsyncClient() as http:
        # ① 注册 OpenAI 兼容凭据(cheap M3 走 base+/v1)。
        cred = (await http.post(f"{base_url}/credential/", json=_cheap_credential_payload(),
                                headers=headers, timeout=30.0)).json()
        credential_id = cred.get("credential_id") or cred.get("id")
        # ② 建 agent(system_prompt=cheap build_system_prompt;context_config 历史压缩;react_config 放开轮数)。
        ctx_cfg = config.build_context_config()
        agent_body = {
            "name": "cheap-writer",
            "system_prompt": build_system_prompt(game_id, scaffold_desc),
            "context_config": ctx_cfg.model_dump(mode="json"),
            "react_config": {"max_iters": max_iters},
        }
        agent = (await http.post(f"{base_url}/agent/", json=agent_body, headers=headers, timeout=30.0)).json()
        agent_id = agent.get("agent_id") or agent.get("id")
        # ③ 建 session(chat_model_config = openai_credential + MiniMax-M3 + max_tokens,无 thinking 分离)。
        session_body = {
            "agent_id": agent_id,
            "chat_model_config": {
                "type": "openai_credential",
                "credential_id": credential_id,
                "model": _bootstrap.SPIKE_MODEL,
                "parameters": {"max_tokens": max_tokens},
            },
        }
        session = (await http.post(f"{base_url}/sessions/", json=session_body, headers=headers, timeout=30.0)).json()
        session_id = session.get("session_id") or session.get("id")

        # ④ scaffold(clone 模板 + SAA 信封)——必须在 /chat 前,**game_id=后端 gameId**(与 system prompt 的 ⟦G⟧ 一致),
        #    agent 起点就位在 amgen-<后端 gameId>(subprocess 经 to_thread)。C2:产物目录统一后端 gameId、不用 session_id。
        sc = await asyncio.to_thread(cheap_run.scaffold, game_id, scaffold_template)
        if not sc["ok"]:
            return _failed_summary(game_id, f"scaffold 失败:{sc['output'][:300]}"), cheap_run.game_dir(game_id)
        # ⑤ 写会话注册表 sidecar(C2:Service 两工厂据 session_id 读它解析回后端 gameId、绑六工具/评门/collector;
        #    create 路 write_whitelist=None → restricted=False)。external_game_id 恒写、正常路工厂必读到。
        _write_session_cfg(session_id, external_game_id=game_id,
                           write_whitelist=write_whitelist, scaffold_template=scaffold_template)
        # ⑥ 设 BYPASS 权限(六工具默认 ASK,服务态无人确认,不设首个工具调用即卡死;PATCH 需 query agent_id,同 tier2)。
        await http.patch(f"{base_url}/sessions/{session_id}", json={"permission_mode": "bypass"},
                         headers=headers, params={"agent_id": agent_id}, timeout=30.0)
        # ⑦ 发 kick(引导 read skill → 写 game-logic.js → check/build → finish;文本对齐旧 run_studio 的 create kick)。
        kick = (f"请按这个 brief 造一款游戏:「{brief}」。先 read_file 读手册(.agents/skills/littlejs-game-dev.md)"
                "和你的起点 game-logic.js 再动手;核心玩法实现完、check 与 build 都绿了就立即 finish。")
        await http.post(f"{base_url}/chat/", json={
            "agent_id": agent_id, "session_id": session_id,
            "input": {"name": "user", "role": "user", "content": [{"type": "text", "text": kick}]},
        }, headers=headers, timeout=30.0)

    # ⑧ SSE 等这一次回合真结束(内部续修多轮)。C3:idle 从 genconfig 派生(不硬编码 300;慢门 ~390s 静默不误弃单),
    #    总超时按 (max_repairs+1)×(门+推理)放宽、且 > breaker 墙钟(让 breaker 先优雅收、driver 别抢先弃单)。
    sse_idle_s = genconfig.get("budget", "cheap_repair_step_timeout_s", 600)  # = breaker step_timeout,> 单次门跑最坏
    sse_timeout = genconfig.get("budget", "cheap_sse_turn_timeout_s",
                                (genconfig.get("iteration", "max_resumes", 6) + 1)
                                * (genconfig.get("budget", "cheap_gate_timeout_s", 420) + 180) + 600)
    turn = await _wait_for_turn_end(base_url, agent_id, session_id, user_id=user_id,
                                    timeout_s=sse_timeout, idle_timeout_s=sse_idle_s)
    if not turn.get("ended"):
        # 回合未真结束(SSE 总超时/断流):Service 可能仍在续修写盘,读中间态会误判 → 据现有产物 best-effort 组 summary、诚实标 reason。
        print(f"[cheap-driver] game={game_id} 回合未真结束(reason={turn.get('reason')}),据现有产物组 summary。", flush=True)

    # ⑨ 读九门 verdict + Service 收口采集 sidecar → 组 run-summary(供 result_out;C2:全用后端 gameId)。
    #    _read_service_run_summary 带有界轮询(C1:关闭 collector flush 与 driver 读 REPLY_END 的竞态)。
    verdict = _read_last_cheap_verdict(game_id)
    svc = _read_service_run_summary(game_id)
    summary = _build_summary(game_id, brief, verdict, svc, turn, t0)

    # ⑩ reply 外收口:非阻塞丰富度 LLM 评分(additive;不进 verdict、不改 ok;token 不污染生成成本台账)。
    try:
        summary["richness"] = await cheap_verify.verify_richness(game_id, brief=brief)
        rv = summary["richness"]
        print(f"[cheap-driver] game={game_id} 丰富度 score={rv.get('score')}/{rv.get('max')} "
              f"degraded={rv.get('degraded')}", flush=True)
    except Exception as e:  # noqa: BLE001 —— richness 非阻塞铁律:绝不影响回调
        summary["richness"] = {"score": None, "degraded": True, "reason": f"richness 接线异常:{type(e).__name__}: {e}"}

    print(f"[cheap-driver] game={game_id} 单 POST 结束: ok={summary['ok']} attempts={summary['attempts']} "
          f"costRmb={summary['costRmb']} wallSec={summary['wallSec']}", flush=True)
    return summary, cheap_run.game_dir(game_id)

• Step 4:改 worker_service.py —— 生成路默认 run_fn 换成驱动 Service

在 cheap-worker/worker_service.py 的 _default_run_fn(现进程内 run_studio)之后加 _service_run_fn,并把 WorkerState.__init__ 的默认从 _default_run_fn 改成 _service_run_fn。

先在 _default_run_fn 定义之后新增(保留 _default_run_fn 作 fallback/对照,不删):

def _service_run_fn(job: dict):
    """默认 run_fn(阶段一②):驱动 cheap Service /chat 跑一局(单 POST + 洋葱内续修),返 (run-summary, game_dir)。

    取代旧 _default_run_fn 的进程内 run_studio:续修/门判/软预算全在 Service 端 middleware(阶段一① 复用),
    消费方只发一次 POST。懒导入 cheap_service_driver(牵 httpx/tier2),单测注入 run_fn 不触发。
    """
    import asyncio

    import cheap_service_driver

    base_url = os.environ.get("CHEAP_SERVICE_URL", "http://127.0.0.1:8300")
    return asyncio.run(cheap_service_driver.drive_cheap_generation(job, base_url=base_url))

worker_service.py 顶部 import 段补 os(现无;_service_run_fn 读 env 用):

import os

WorkerState.__init__ 里默认 run_fn 一行改:

        self.run_fn = run_fn or _service_run_fn        # job -> (summary, game_dir)（阶段一②:默认驱动 Service /chat）

（原为 self.run_fn = run_fn or _default_run_fn;_default_run_fn 保留但不再是默认。process_job / try_enqueue / result_out 回调链 / modify 路一字不动。）

• Step 5:跑测试,确认通过

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_service_driver.py -v Expected:PASS(8 passed —— 凭据 base+/v1 · _build_summary succeeded/failed · trace 端到端契约(I4)· session→后端 gameId 注册表往返(C2)· service-run-summary 有界轮询(C1)· fake-SSE driver 契约(REPLY_END/未结束)· 默认 run_fn 已硬切驱动)

• Step 6:回归 worker_service 现有单测(确认 run_fn 默认切换未破坏队列/回调/modify)

Run:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_worker_service.py -v Expected:全 PASS(现有测试都注入 run_fn/send_fn 桩,不走真驱动;默认切换只影响未注入路径)

• Step 7:惰性红线 + py_compile

Run:cheap-worker/.venv/bin/python -m py_compile cheap-worker/cheap_service_driver.py cheap-worker/worker_service.py && grep -nE "^(import|from) (agentscope|httpx)" cheap-worker/cheap_service_driver.py Expected:py_compile 无输出;grep 无匹配(driver 顶层无 agentscope/httpx import,惰性守住)。

• Step 8:提交

git add cheap-worker/cheap_service_driver.py cheap-worker/worker_service.py cheap-worker/tests/test_cheap_service_driver.py
git commit -m "feat(cheap): worker 归并驱动 Service /chat(单 POST + 洋葱内续修)+ reply 外 result_out 收口 (切片一 阶段一②/T3)"

---

随 mini-desktop 后端窗口验证 TODO(真门要 chrome/esbuild + Redis + cheap Service)

本地单测覆盖判据 / 端口池 / 工厂结构 / 凭据体 / summary 组形 / run_fn 切换;下列真门端到端须在 mini-desktop(有 chrome+esbuild+Redis)按创始人窗口验:

• 先量真实门时延(定超时的前提;C3 方向=对慢局放宽,非收紧):量单次便宜档九门收口(run_cheap_gates:stage+smoke+play)真实时延(最坏 ~390s = stage≤90+smoke≤120+play≤180;典型可能更短)。标定方向是放宽不是收紧:cheap_gate_timeout_s(默认 420)须 ≥ 实测最坏门时延,cheap_repair_step_timeout_s(默认 600,同时是 breaker step 与 driver SSE idle)须 > 单次门跑 + 推理余量、cheap_repair_wall_timeout_s / driver cheap_sse_turn_timeout_s 按 (max_repairs+1)×(门+推理)+余量 且 driver 总超时 > breaker 墙钟。别再用旧 200/300——门跑期间 reply 静默 ~390s,idle=300 会误弃单、step=320 会误熔断(Codex C3)。driver idle 已从 genconfig 派生、不硬编码。
• cheap 单 POST 收敛 + 续修真拦:一局便宜档真生成 —— agent 首轮 finish(check+build 绿)但九门没绿 → 续修 middleware 拦住自然终止、注入九门失败反馈、agent 续跑修 → 直到九门绿 finish 或续修/预算耗尽放行。观察 [cheap-repair]/[repair] 日志的续修轮次与放行原因;attempts==repairs+1、回合真结束(ended=True)后 driver 读到九门 verdict。
• 放行权威 = 独立重跑九门:agent 谎报「做完了」也不放行 —— 放行只认续修 middleware 独立跑的 run_cheap_gates。
• C2 目录/id 一致(首轮不越界):一局真生成确认 —— scaffold 落 amgen-<后端 gameId>、system prompt 的 ⟦G⟧ 指同一目录、六工具(经会话注册表解析回后端 gameId)写同一目录;首轮 agent read 起点 game-logic.js 成功、write 不报越界;_cheap-sessions/<session_id>.json 里 external_game_id = 后端 gameId;result_out 的 trace.gameId = 后端 gameId(非 session uuid)。这是 C2 修复的直接兑现点(修前 prompt 指后端 gameId、工具锁 session_id → 首轮必越界)。
• 软预算优雅收尾 + max_repairs 兜成本(决策①):把 budget.cheap_rmb_hard_limit 压到极小,一局超预算 → agent 收「预算已达上限」提醒后尽快 finish、续修 middleware 见实测 budget_exhausted 放行(至少续修一次),不 fail-closed 断链(对比改造前 hard 档抛熔断丢尽力产物)。成本上界验的是 max_repairs=6 优雅终止(6 轮续修用满仍未绿即交尽力产物、成本 ~¥2.7 典型 <¥10),不是三道次数/轮数闸——三闸设 150 只做失控兜底,正常局不该撞到它们(若撞到 150 说明续修循环失控、是 bug 信号,重点排查而非当成本尺)。量一局真实 costRmb / 续修轮数分布后,回填 cheap_rmb_hard_limit、必要时调 max_resumes;150 一般不必动。
• 并发不撞端口(决策③):mini-desktop 起数局并发(逼近 15)经 cheap Service 生成,确认各局九门 play 用不同派生端口(4400+/9400+)、无 Chrome/serve 端口竞态、无 4330/9322(tier2)串扰。若上游 worker 仍串行(现状单 worker 线程),先验端口池机制正确(串行下每局取同一 slot、释放不泄漏);真并发(多 worker / 放开 worker 并行)是容量 follow-up。
• 收口采集跨进程回收 cost/trace(C1 竞态复验):确认 Service 端 _CheapRunCollector 在 REPLY_END 流经时同步 flush(不只 finally)写出 evidence/service-run-summary.json(costRmb/repairs/rmbGate),driver 读到(有界轮询兜底)并组进 result_out 的 trace.cost.totalRmb/trace.repairs。重点验成功局 costRmb 非 0(Codex C1:REPLY_END 先于 finally flush 会让成功局也上报 costRmb=0 → D11 efficiency 维恒中性,是 M3b 坑的归并版静默数据劣化);跑若干成功局确认 costRmb 都 >0、rmbGate=active,无「status 对但 cost=0」。
• result_out 七值回调 + engineBundle 承重:门绿 → status=succeeded + engineBundle(bundle.iife.js 含 __GameBundle)入 feed;门没绿 → status=failed + failureReason 七值枚举(不造非枚举值,回调事务不回滚)。真后端 e2e 抓 proxy 502 / costRmb / failureReason 七值(M3a 三 bug 复验位)。
• 续修空转率(创始人决策③:cutover 后量真实空转率):cheap 无九门自检工具、agent 盲修(它没跑过门就据 feedback 改);cutover 后在 mini-desktop 窗口量真实「续修空转率」——反复 finish→九门没绿→盲修→仍没绿,烧满 6 续修(最坏 ~7×390s + 成本)才撞 max_repairs。据实测空转率再决定是否把「给 cheap 加一个九门自检工具(让 agent 先自检再终止,减盲修)」从 follow-up 提前。这是 cutover 后的头号观测项(盲修空转会显著抬每局时延、未必提良率)。
• 代理旁路(内网 fake-ip)+ 凭据 base 路由(I2/I4):确认 cheap Service 进程在 import agentscope 前装了 NO_PROXY(build_cheap_app 先 client.resolve_base_url() 再 import),M3 请求不被 clash fake-ip 转走 502;driver→Service(127.0.0.1:8300)也验一遍(driver 先 resolve_base_url()+install_proxy_bypass(base_url) 再 import httpx,防本地环回被 fake-ip 拦);driver 回调经 worker_service._NO_PROXY_OPENER 禁代理。I4:同批验 openai_credential 的 base_url(末尾 /v1)真到 M3 —— 确认 2.0.2 get_model 不把 base 二次拼成 /v1/v1(标准 openai SDK base 含 /v1、低风险,但真门必验),抓 502/凭据路由三处一起看。
• modify 路不回归:A11 调整路(deterministic / regenerate-module)仍走 worker_service._process_modify_job/_process_regenerate_job(未归并、进程内 cheap_modify),确认默认 run_fn 切换未影响 modify 分流(modify 在 process_job 最前分流、不经 run_fn)。
• cutover 硬切就绪判定(创始人决策②:全量切、无金丝雀旗):默认 run_fn 已硬换 _service_run_fn,cutover 即全量。故 cutover 就绪判定要严:上述真门端到端(单 POST 收敛 + 续修真拦 + C2 不越界 + C1 cost 非 0 + 并发不撞端口 + 七值回调 + 代理旁路)在 mini-desktop 真后端 e2e 尽力全验通再算就绪;真跑抓 M3a 三 bug 复验位(proxy 502 / costRmb / failureReason 七值)。回滚 = WorkerState.__init__ 默认改回 _default_run_fn(一行),_default_run_fn 完整保留。

风险与回滚

• 续修判据升级为九门(有默认项 C 的代价)+ 慢门超时(C3):归并后每续修周期跑一次完整九门收口(stage+smoke+play,最坏 ~390s = stage≤90+smoke≤120+play≤180),比旧「check+build 绿即收尾」重。最坏 (max_repairs+1)=7 轮串行九门 ≈ 数十分钟一局。Codex C3 已修:门跑期间 reply 静默(to_thread 阻塞、无 AgentEvent、SSE 只剩心跳),故 breaker step_timeout_s、driver SSE idle_timeout_s 都取 cheap_repair_step_timeout_s=600(> 最坏门跑 + 推理余量,driver idle 从 genconfig 派生、不再硬编码 300)、gate 基准 cheap_gate_timeout_s=420、总墙钟按 (max_repairs+1)×(门+推理)+余量——旧 200/300 会在慢门上误熔断/误弃单。真值 directional 待标定,方向是放宽(窗口 TODO 首条);性能优化(门结果缓存 / 增量门)列 follow-up(接受慢、先重标定跑通,同① 创始人口径)。空转风险见下 + 窗口 TODO「续修空转率」。
• soft 预算丢硬地板、靠 max_repairs 兜成本(决策①;C3/C4 收口):切 soft 后 ¥10 不再 fail-closed,成本上界改由 max_repairs=6 优雅终止兜(6 轮续修用满即交尽力产物,成本 ~¥2.7 典型 <¥10)。双评审纠正的口径:on_reply 里 max_model_calls/max_tool_calls 越限与超时的 _trip 都是无条件硬熔断、不受 soft_budget 约束(middleware.py:315-324/303-309),故绝不能把它们当贴脸成本尺——一旦贴脸就会抢在 max_repairs 前把优雅收尾翻成硬中断无产物(还得空等一个 idle 周期),部分抵消决策①。三闸(max_tool_calls/max_model_calls/max_iters)全设 150 当纯失控兜底(Codex C4:max_tool_calls 不显式设吃默认 60 会先撞)。回滚 = 把 soft_budget 关(改工厂一处),退回 hard 档 ¥10 fail-closed。
• cost/trace 跨进程靠 collector sidecar + 正常路竞态(C1 已修):breaker/repair/tracer 在 Service 进程、driver 在 worker 进程,cost 靠 _CheapRunCollector 写 sidecar 传递。Codex C1 的关键洞察:框架 _agent.py:615 先 yield ReplyEndEvent 再 yield finish Msg,生成器结束后 finally 才 flush——若 collector 只在 finally 写盘,driver 一读到 REPLY_END 就返回读文件时 sidecar 还空 → 成功局也上报 costRmb=0(不只是崩溃降级,是 ms 级正常路竞态,重开 M3b 的 D11 恒中性坑)。双侧修:① collector 在 ReplyEndEvent 流经时同步 flush(在 yield 它给 SSE publisher 前),关闭竞态源头;② driver _read_service_run_summary 加有界轮询(≤5s)兜底。降级三路:正常 REPLY_END 已 flush / breaker 硬熔断 finally 兜底 flush(有部分成本)/ 进程中途崩 → sidecar 缺失 → driver 诚实降级 costRmb=0(不伪造)。mini-desktop 重点验成功局 costRmb 都 >0(窗口 TODO「收口采集 C1 竞态复验」)。
• session→gameId 映射 + write_whitelist 载体用 FS sidecar(有默认项 B;C2/I1 收口):因 AgentScope 工厂签名固定 (user_id, agent_id, session_id)、拿不到后端 gameId 与 session 记录,且记录无自由字段,external_game_id(后端 gameId)+ write_whitelist/scaffold_template/restricted 走会话注册表 _cheap-sessions/<session_id>.json(cheap_run.session_cfg_path;_ 前缀非游戏目录、不在 games/ 堆 amgen- 空壳)。依赖 worker↔Service 同机共享 FS(与 game-runtime/games/ 同目录,tier2 亦然)。I1 fail-closed(Codex):sidecar 缺 external_game_id → 工厂回落 session_id(产物目录与 scaffold 不一致 → 生成响亮失败,非静默错;正常路 driver 恒在 /chat 前写、必到);受限模式(reskin/modify,restricted:true)白名单缺失 → 收窄成空集(禁写、绝不回落 create 的不收窄)。create 路 write_whitelist=None、restricted:false、sidecar 只承 session→gameId 映射;reskin 路(未在 ② 生产链接线)才用白名单。若未来 cheap Service 与 worker 跨机部署,此载体需改(Redis 注册表或凭 storage 读 workspace_id)——列 follow-up。
• cheap 与 tier2 共享 Redis 逻辑库:build_cheap_app 复用 infra_config.redis_params(与 tier2 同 DB)。session/agent 记录按 uuid 全局唯一、按 user_id 命名空间(cheap user_id="cheap"),不冲突;run-lock 按 session_id,不串。若要完全键空间隔离,给 cheap 独立 Redis DB(需扩 redis_params 加 offset)——列 follow-up、不阻塞本 plan。
• worker 仍串行、真并发是容量 follow-up:② 保留 worker_service 单 worker 线程串行消费(现状),故实际并发=1;端口池(决策③)是为并发设计的机制,真放开并发(多 worker / worker 并行驱动)是容量 follow-up([便宜档生成容量+并发上限] memory,conc≤15)。端口池串行下也正确(每局取同一 slot、释放不泄漏)。
• 盲修空转抬时延(决策③ follow-up,cutover 后量):cheap 无九门自检工具,agent 每次纯文本终止后 check 才第一次真跑九门、不过就据 feedback 盲修(它没跑过门)。盲修低效时会烧满 6 续修(最坏 ~7×390s + 成本)才撞 max_repairs,显著抬每局时延且未必提良率。这是已上线路径(M3a/M3b)的真行为变化。创始人决策③:本 plan 不加自检工具,cutover 后先量真实空转率再定是否把「加九门自检工具」从 follow-up 提前(窗口 TODO「续修空转率」)。
• 硬切全量、无金丝雀(创始人决策②):worker_service 默认 run_fn 硬换 _service_run_fn,cutover 即全量切、不加 env 旗。故 cutover 就绪判定要严(真后端 e2e 尽力全验通,窗口 TODO「cutover 硬切就绪判定」)。回滚:默认 run_fn 改回 _default_run_fn(进程内 run_studio,一行)即退回归并前的裸 HTTP + 进程内 resume 路;cheap Service 不起、driver 不调即无副作用。_default_run_fn 完整保留作对照/回滚。

Self-Review(controller 重点复核这几处)

• spec 覆盖:创始人 3 决策(soft+max_repairs 兜成本/三闸=150 失控兜底 / 各起 Service + 默认 run_fn 硬切 / 派生端口)、5 有默认项(回调保留 / session→gameId+whitelist sidecar 载体 / 判据升九门+慢门放宽 / 不加自检工具留 follow-up / T2-b 工厂 try)、九门不做 MCP、代理旁路必保 —— 均落进 Global Constraints + 对应 task 代码。双评审 4C+4I 已文内修:C1 collector REPLY_END 同步 flush + driver 有界轮询、C2 产物目录/回调统一后端 gameId(sidecar 映射)、C3 超时按最坏门跑放宽 + driver idle 派生、C4 三闸显式设 150(治默认 60);I1 whitelist fail-closed、I2 旁路先于 import + 127.0.0.1、I3 判据要求 guards 非空、I4 trace 端到端契约测试。与 recon 不符的一处:recon 决策③ 写「改 serve-and-play.sh 收端口参」,实测早已收 [port] [cdpPort] 参,shell 零改(端口池在 cheap_service_app)。
• 占位扫描:全文无 TBD / 「类似 TN」/ 无代码的「实现 X」;每个改代码步骤给完整代码。引用的类型/函数均在某 task 定义(judge_cheap_verdict/run_cheap_gates=T1;session_cfg_path/工厂/端口池/collector/_resolve_external_game_id/_resolve_write_whitelist/build_cheap_app=T2;driver 各 helper=T3)或现状存在(GateJudgment/RepairMiddleware/CircuitBreaker.{soft_budget,max_tool_calls}/_extract_function_tools/_wait_for_turn_end/build_toolkit/build_system_prompt/build_trace_source/_verdict_brief/verify_richness/build_result_out/scaffold/game_dir/wg1_game_dir/build_context_config/resolve_base_url/install_proxy_bypass/get_api_key/redis_params/make_jsonl_sink,均已逐一读代码核实签名)。
• 类型一致:judge_cheap_verdict 返 GateJudgment(与 tier2 同,续修 check 契约一致);_cheap_middlewares_factory 返 [collector, tracer, repair, breaker](collector 最外层,on_reply finally 在 breaker 计费后写盘);drive_cheap_generation 返 (summary, game_dir)(与旧 _default_run_fn 同,process_job 零改动);summary 键与 result_out.build_result_out/_build_trace 读取口径逐字对齐(verdict/costRmb/attempts/gameId/stage/verdictFull/driverType/models/wallSec)。
• 双评审已解三处早期不确定项:① collector cost 可靠性 → C1 修(REPLY_END 同步 flush + driver 有界轮询,关闭正常路竞态,不只崩溃降级);② sidecar 载体 → C2 收口(升为 session→后端 gameId 映射 + I1 fail-closed,relocate 到 _cheap-sessions/);③ 判据升九门性能 → C3 修(超时按最坏门跑放宽、driver idle 派生)。留给创始人终评审的判断项:(a) 决策② 硬切全量、cutover 就绪只靠真后端 e2e 尽力验通(无金丝雀),风险敞口是否接受;(b) 决策③ 盲修空转不加自检工具、先量真实空转率,是否接受首版可能的时延代价;(c) 三闸 150 / gate 420 / step 600 / max_repairs 6 是 directional 值,mini-desktop 标定前是否接受宁大勿小。