diff --git a/cheap-worker/cheap_run.py b/cheap-worker/cheap_run.py
index 30d3432c..d65b1369 100644
--- a/cheap-worker/cheap_run.py
+++ b/cheap-worker/cheap_run.py
@@ -44,6 +44,22 @@ def wg1_game_dir(game_id: str) -> Path:
     return _WG1_DIR / game_id
 
 
+# 便宜档 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"
+
+
 def _resolve_in_repo(rel: str) -> Path:
     """把 repo 相对路径解析为绝对路径，校验不逃出 repo 根（对 tools.mjs resolveInRepo）。"""
     abs_p = (_REPO_ROOT / rel).resolve()
diff --git a/cheap-worker/cheap_service_app.py b/cheap-worker/cheap_service_app.py
new file mode 100644
index 00000000..fc190a0f
--- /dev/null
+++ b/cheap-worker/cheap_service_app.py
@@ -0,0 +1,348 @@
+"""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 递增派生。
+
+    用 LifoQueue(栈式)而非 FIFO Queue:release 归还的端口对下一次 acquire 优先复用(最近释放先取),
+    这样「释放→再取」拿回的就是刚归还的那对,便于验证不泄漏(池验收契约 test_port_pool_derives_distinct_pairs);
+    FIFO 会先派发池底未用过的对、掩盖泄漏。LifoQueue is-a Queue,对外 get/put 接口不变、并发正确性不受影响。
+    """
+    global _cheap_port_pool
+    with _cheap_port_pool_lock:
+        if _cheap_port_pool is None:
+            pool: "_queue.Queue" = _queue.LifoQueue()
+            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()
diff --git a/cheap-worker/tests/test_cheap_service_app.py b/cheap-worker/tests/test_cheap_service_app.py
new file mode 100644
index 00000000..5986a537
--- /dev/null
+++ b/cheap-worker/tests/test_cheap_service_app.py
@@ -0,0 +1,139 @@
+"""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