From 8d9bf1e57775a2298135f86b2e3276c3ed7b252e Mon Sep 17 00:00:00 2001
From: lili <lili@lili-mac.local>
Date: Thu, 2 Jul 2026 12:36:38 -0700
Subject: [PATCH] =?UTF-8?q?feat(cheap):=20worker=20=E5=BD=92=E5=B9=B6?=
 =?UTF-8?q?=E9=A9=B1=E5=8A=A8=20Service=20/chat(=E5=8D=95=20POST=20+=20?=
 =?UTF-8?q?=E6=B4=8B=E8=91=B1=E5=86=85=E7=BB=AD=E4=BF=AE)+=20reply=20?=
 =?UTF-8?q?=E5=A4=96=20result=5Fout=20=E6=94=B6=E5=8F=A3=20(=E5=88=87?=
 =?UTF-8?q?=E7=89=87=E4=B8=80=20=E9=98=B6=E6=AE=B5=E4=B8=80=E2=91=A1/T3)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 cheap-worker/cheap_service_driver.py          | 306 ++++++++++++++++++
 .../tests/test_cheap_service_driver.py        | 180 +++++++++++
 cheap-worker/worker_service.py                |  17 +-
 3 files changed, 502 insertions(+), 1 deletion(-)
 create mode 100644 cheap-worker/cheap_service_driver.py
 create mode 100644 cheap-worker/tests/test_cheap_service_driver.py

diff --git a/cheap-worker/cheap_service_driver.py b/cheap-worker/cheap_service_driver.py
new file mode 100644
index 00000000..2e774042
--- /dev/null
+++ b/cheap-worker/cheap_service_driver.py
@@ -0,0 +1,306 @@
+"""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)
diff --git a/cheap-worker/tests/test_cheap_service_driver.py b/cheap-worker/tests/test_cheap_service_driver.py
new file mode 100644
index 00000000..1d1b2b94
--- /dev/null
+++ b/cheap-worker/tests/test_cheap_service_driver.py
@@ -0,0 +1,180 @@
+"""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
diff --git a/cheap-worker/worker_service.py b/cheap-worker/worker_service.py
index 5b627842..bf335447 100644
--- a/cheap-worker/worker_service.py
+++ b/cheap-worker/worker_service.py
@@ -17,6 +17,7 @@ import argparse
 import hashlib
 import hmac
 import json
+import os
 import queue
 import threading
 import urllib.error
@@ -146,6 +147,20 @@ def _default_run_fn(job: dict):
     return summary, cheap_run.game_dir(game_id)
 
 
+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 状态 + 有界队列 + 串行消费 ----------
 
 class WorkerState:
@@ -155,7 +170,7 @@ class WorkerState:
                  run_fn=None, send_fn=None, profile_fn=None, classify_fn=None, modify_fn=None, regen_fn=None):
         self.callback_secret = callback_secret
         self.queue: queue.Queue = queue.Queue(maxsize=queue_maxsize)
-        self.run_fn = run_fn or _default_run_fn         # job -> (summary, game_dir)
+        self.run_fn = run_fn or _service_run_fn        # job -> (summary, game_dir)（阶段一②:默认驱动 Service /chat）
         self.send_fn = send_fn or post_callback          # (url, payload, secret) -> (status, body)
         self.profile_fn = profile_fn or derive_profile   # (job, game_dir) -> profile | None
         self.classify_fn = classify_fn or cheap_classify.classify  # (rawText, sourceProject) -> 建议改动(A11 M2 判意图)