diff --git a/tier2/ci/check-prompt-registry.sh b/tier2/ci/check-prompt-registry.sh
new file mode 100755
index 00000000..e9d72e9a
--- /dev/null
+++ b/tier2/ci/check-prompt-registry.sh
@@ -0,0 +1,178 @@
+#!/usr/bin/env bash
+# tier2 prompt 一致性守门 —— 把「prompts.py 该加载的每个 prompt == registry/外置文件、无漂移」编译成机器门
+#   (对齐项目 AGENTS.md §6 条款10:文档治理规则必须编译成机器门,违规红线拦截,而非只作散文)。
+#
+# 背景:tier2 的 8 条 system prompt 已外置到 contracts/prompts/09-tier2-richgame/*.md + registry.yaml,
+#   worker/prompts.py 经契约 #8 Prompt Registry 运行时按 id 读正文,读不到/脏 → 回落 roles.py 内置原文。
+#   外置后最容易出的漂移:① 改了 registry 加了新 id,roles.py 没接(孤儿条目);② roles.py 加载某 id,
+#   registry 没登记(回落内置原文、外置形同虚设);③ registry 的 file 指向的 .md 不存在(运行时静默回落)。
+#   本门把这三类漂移在 wave-close / pre-commit 拦下来。
+#
+# 校验内容(纯文件级三向一致性比对;不需 runtime/网络/重依赖,6c6g 本机与 CMI(mini-desktop)均可跑):
+#   A) roles.py 里 prompts.load("tier2.*") 声明加载的 id 集合
+#      == registry.yaml 里 file 指向 09-tier2-richgame/ 的 tier2.* 条目 id 集合(双向无差集);
+#   B) registry 每个 tier2.* 条目的 file 字段所指 .md 在 contracts/prompts/ 下真实存在;
+#   C) contracts/prompts/09-tier2-richgame/ 下每个 .md 都被某 registry 条目引用(无未登记的孤儿 .md)。
+#   任一项漂移 → 非零退出 + 指出哪个 prompt / 哪个文件漂了;全一致 → 退出 0。
+#
+# ⚠️ import 级校验(bare-import worker/prompts.py + 逐 id load() 断言「命中 registry 非回落内置原文、
+#   且正文与 .md 逐字节一致」)是更强的一层:它能抓出「frontmatter 剥离口径变了 / registry 解析器口径变了」
+#   这类纯文件比对看不出的语义漂移。worker/prompts.py 顶层不牵 agentscope(只 os/re/threading/pathlib),
+#   故 6c6g 本机可裸 import;本门【尽力】跑这层(python3 可用且 prompts.py 可裸 import 时),跑不了就
+#   降级为纯文件级比对并打印「import 级校验需在能 import worker 包的环境(mini-desktop)补跑」的提示——
+#   纯文件级比对仍是硬门(漂移即拦),import 级只是增强、缺它不放过文件级漂移。
+set -euo pipefail
+
+# ROOT = tier2/(本脚本在 tier2/ci/);与 check-forbidden-import.sh 同款定位。
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+# 仓库根 = tier2/ 的上一级(contracts/ 在仓根下)。
+REPO_ROOT="$(cd "$ROOT/.." && pwd)"
+
+ROLES_PY="$ROOT/gen-worker/worker/roles.py"
+PROMPTS_PY="$ROOT/gen-worker/worker/prompts.py"
+REGISTRY="$REPO_ROOT/contracts/prompts/registry.yaml"
+PROMPT_DIR="$REPO_ROOT/contracts/prompts/09-tier2-richgame"
+
+fail=0
+
+# 前置:三个关键路径必须存在(缺任何一个都是配置外置被破坏,直接红线)。
+for p in "$ROLES_PY" "$REGISTRY" "$PROMPT_DIR"; do
+  if [ ! -e "$p" ]; then
+    echo "❌ prompt-registry: 关键路径缺失:$p" >&2
+    fail=1
+  fi
+done
+if [ "$fail" -ne 0 ]; then
+  exit 1
+fi
+
+# ── 提取三个 id/文件集合 ────────────────────────────────────────────────────────────
+# (1) roles.py 声明加载的 tier2.* id(prompts.load("tier2.xxx", ...) 的第一个串参)。
+roles_ids=$(grep -oE 'prompts\.load\("tier2\.[A-Za-z0-9._-]+"' "$ROLES_PY" \
+  | sed -E 's/.*prompts\.load\("//; s/"$//' | sort -u)
+
+# (2) registry.yaml 里 file 指向 09-tier2-richgame/ 的条目:抽出 (id, file) 对。
+#     registry 是「- id: xxx」起头、随后若干「  key: val」的列表块;我们只关心 tier2.* 且 file 落在
+#     09-tier2-richgame/ 的条目(其余阶段 01~07 的条目与本门无关,不碰)。
+#     awk:遇 `- id:` 记当前 id;遇 `file:` 记当前 file;到下一个 `- id:` 或文件尾时,若 file 在
+#     09-tier2-richgame/ 下就输出「id<TAB>file」。剥行内注释(# 之后)与首尾空白。
+reg_pairs=$(awk '
+  function trim(s){ sub(/#.*/,"",s); gsub(/^[ \t]+|[ \t]+$/,"",s); return s }
+  /^[[:space:]]*-[[:space:]]+id[[:space:]]*:/ {
+    if (cur_id != "" && cur_file ~ /^09-tier2-richgame\//) print cur_id "\t" cur_file;
+    line=$0; sub(/^[^:]*:[[:space:]]*/,"",line); cur_id=trim(line); cur_file=""; next
+  }
+  /^[[:space:]]+file[[:space:]]*:/ {
+    line=$0; sub(/^[^:]*:[[:space:]]*/,"",line); cur_file=trim(line); next
+  }
+  END { if (cur_id != "" && cur_file ~ /^09-tier2-richgame\//) print cur_id "\t" cur_file }
+' "$REGISTRY")
+
+reg_ids=$(printf '%s\n' "$reg_pairs" | awk -F'\t' 'NF>=1 && $1!=""{print $1}' | sort -u)
+
+# (3) 09-tier2-richgame/ 下真实存在的 .md 文件名(相对 contracts/prompts/ 的路径,与 registry file 字段同口径)。
+fs_files=$(cd "$REPO_ROOT/contracts/prompts" && ls 09-tier2-richgame/*.md 2>/dev/null | sort -u || true)
+
+# ── 校验 A:roles.py 声明集 == registry tier2 条目集(双向无差集)───────────────────────
+only_roles=$(comm -23 <(printf '%s\n' "$roles_ids") <(printf '%s\n' "$reg_ids") || true)
+only_reg=$(comm -13 <(printf '%s\n' "$roles_ids") <(printf '%s\n' "$reg_ids") || true)
+if [ -n "$only_roles" ]; then
+  echo "❌ prompt-registry[A]: roles.py 加载了这些 id,registry 却未登记(外置失效→运行时静默回落内置原文):" >&2
+  printf '%s\n' "$only_roles" | sed 's/^/   /' >&2
+  fail=1
+fi
+if [ -n "$only_reg" ]; then
+  echo "❌ prompt-registry[A]: registry 登记了这些 tier2 条目,roles.py 却没加载(孤儿条目):" >&2
+  printf '%s\n' "$only_reg" | sed 's/^/   /' >&2
+  fail=1
+fi
+
+# ── 校验 B:registry 每个 tier2 条目的 file 指向的 .md 真实存在 ───────────────────────────
+while IFS=$'\t' read -r id file; do
+  [ -z "$id" ] && continue
+  if [ ! -f "$REPO_ROOT/contracts/prompts/$file" ]; then
+    echo "❌ prompt-registry[B]: registry 条目 '$id' 的 file 指向不存在的文件:$file" >&2
+    fail=1
+  fi
+done <<< "$reg_pairs"
+
+# ── 校验 C:09-tier2-richgame/ 下每个 .md 都被某 registry 条目引用(无孤儿 .md)──────────────
+reg_files=$(printf '%s\n' "$reg_pairs" | awk -F'\t' 'NF>=2 && $2!=""{print $2}' | sort -u)
+orphan_md=$(comm -23 <(printf '%s\n' "$fs_files") <(printf '%s\n' "$reg_files") || true)
+if [ -n "$orphan_md" ]; then
+  echo "❌ prompt-registry[C]: 这些 .md 在 09-tier2-richgame/ 下却没被任何 registry 条目引用(孤儿文件):" >&2
+  printf '%s\n' "$orphan_md" | sed 's/^/   /' >&2
+  fail=1
+fi
+
+# ── import 级增强校验(尽力跑;跑不了优雅降级,不影响上面文件级硬门的结论)──────────────────
+#   bare-import worker/prompts.py(顶层不牵 agentscope),对每个 roles.py 声明的 id 调 load(id, SENTINEL):
+#   返回值 != SENTINEL 即「命中 registry(非回落内置原文)」。这能抓出「文件存在但解析/剥 frontmatter
+#   口径漂移导致 load 实际回落」这类纯文件比对看不出的问题。需 python3 且 prompts.py 可裸 import。
+import_checked=0
+if command -v python3 >/dev/null 2>&1; then
+  ids_csv=$(printf '%s\n' "$roles_ids" | paste -sd, -)
+  # PROMPTS_DIR 显式指向仓内 09 的父目录,避免受调用环境 env TIER2_PROMPTS_DIR 干扰。
+  import_out=$(WORKER_DIR="$ROOT/gen-worker/worker" REPO_PROMPTS="$REPO_ROOT/contracts/prompts" IDS="$ids_csv" \
+    python3 - <<'PY' 2>&1 || true
+import os, sys, importlib.util
+worker_dir = os.environ["WORKER_DIR"]
+ids = [x for x in os.environ.get("IDS", "").split(",") if x]
+# 直接按文件路径加载 prompts.py 模块(避免触发 worker 包的相对 import / agentscope)。
+spec = importlib.util.spec_from_file_location("_tier2_prompts_probe", os.path.join(worker_dir, "prompts.py"))
+mod = importlib.util.module_from_spec(spec)
+try:
+    spec.loader.exec_module(mod)
+except Exception as e:  # 任何 import 期异常 → 标记不可跑,降级(由 shell 侧打印降级提示)
+    print("PROBE_SKIP:" + type(e).__name__ + ":" + str(e))
+    sys.exit(0)
+# 强制用仓内 prompts 目录(不受外部 env 干扰)。
+os.environ["TIER2_PROMPTS_DIR"] = os.environ["REPO_PROMPTS"]
+mod.reload()
+SENTINEL = "<<__FALLBACK_SENTINEL__>>"
+bad = []
+for pid in ids:
+    body = mod.load(pid, SENTINEL)
+    if body == SENTINEL or not body:
+        bad.append(pid)
+if bad:
+    print("PROBE_FALLBACK:" + ",".join(bad))
+else:
+    print("PROBE_OK:" + str(len(ids)))
+PY
+)
+  # prompts.py 的 best-effort 告警会打到 stdout 与探针标记行混在一起;只取最后一行 PROBE_* 标记判定。
+  import_marker=$(printf '%s\n' "$import_out" | grep -E '^PROBE_(OK|FALLBACK|SKIP):' | tail -n1)
+  [ -z "$import_marker" ] && import_marker="$import_out"
+  case "$import_marker" in
+    PROBE_OK:*)
+      import_checked=1
+      echo "✓ prompt-registry[import]: ${import_out#PROBE_OK:} 条 prompt 经 prompts.load 命中 registry(非回落内置原文)"
+      ;;
+    PROBE_FALLBACK:*)
+      echo "❌ prompt-registry[import]: 这些 id 经 prompts.load 仍回落内置原文(registry 命中失败=漂移):" >&2
+      echo "   ${import_out#PROBE_FALLBACK:}" >&2
+      fail=1
+      ;;
+    PROBE_SKIP:*)
+      echo "⚠ prompt-registry[import]: prompts.py 裸 import 失败(${import_out#PROBE_SKIP:})→ 仅做文件级比对;import 级校验请在能 import worker 包的环境(mini-desktop)补跑。"
+      ;;
+    *)
+      echo "⚠ prompt-registry[import]: 探针输出异常,降级为文件级比对(请在 mini-desktop 补跑 import 级校验)。输出:$import_out"
+      ;;
+  esac
+else
+  echo "⚠ prompt-registry[import]: 本机无 python3 → 仅做文件级比对;import 级校验请在 mini-desktop 补跑。"
+fi
+
+if [ "$fail" -ne 0 ]; then
+  echo "❌ prompt-registry 守门未通过:tier2 prompt 与 registry/外置文件存在漂移(见上)。" >&2
+  exit 1
+fi
+
+n_ids=$(printf '%s\n' "$roles_ids" | grep -c . || true)
+if [ "$import_checked" -eq 1 ]; then
+  echo "✓ prompt-registry 守门通过(tier2 ${n_ids} 条 prompt:文件级三向一致 + import 级命中 registry)"
+else
+  echo "✓ prompt-registry 守门通过(tier2 ${n_ids} 条 prompt:文件级三向一致;import 级增强校验本机未跑,见上提示)"
+fi
diff --git a/tier2/config/generation.yaml b/tier2/config/generation.yaml
index ce704b50..a0f1f59c 100644
--- a/tier2/config/generation.yaml
+++ b/tier2/config/generation.yaml
@@ -25,6 +25,8 @@ model:
   default_model_name: MiniMax-M3      # 单写 agent 默认模型(M3 路;model_name_from_env 默认值)
   deepseek_flash: deepseek-v4-flash   # 便宜主力档名(spike 主问「便宜到什么程度还守得住」)
   deepseek_pro: deepseek-v4-pro       # 强便宜救场档名(退路树 Q1 读它的过门率)
+  baseline_opus: claude-opus-4-8      # 强基线 Opus(Anthropic 原生经 new-api;A 门「M3 证路(软)」主用,跑同一条 studio 线证「路本身通」)
+  baseline_fable: claude-fable-5      # 强基线 Fable(列出但不假设网关一定可用;build_baseline_model 取不到即诚实报错、不静默)
   max_tokens: 16000                   # 输出上限(非上下文窗;build_model 硬约束 max_tokens > thinking_budget)
   thinking_enable: true               # M3 是否开 thinking 分离(便宜档无此约束,只 M3 路用)
   thinking_budget: 8000               # M3 thinking 预算(必须 < max_tokens,否则 build_model 抛)
diff --git a/tier2/gen-worker/observability/__init__.py b/tier2/gen-worker/observability/__init__.py
index 93a9811e..c862aca3 100644
--- a/tier2/gen-worker/observability/__init__.py
+++ b/tier2/gen-worker/observability/__init__.py
@@ -34,6 +34,14 @@ from .trace import (
     TraceAdapter,
     TraceStep,
     to_trace_step,
+    with_studio,
+)
+# Studio 观测接线(C1):把统一 trace 事件推到 AgentScope Studio 的旁路 sink 工厂。
+#   注意:studio_sink 顶层零重依赖(otel SDK 在函数内惰性 import),6c6g 仍可裸 import 本包。
+from .studio_sink import (
+    make_studio_sink,
+    resolve_studio_url,
+    StudioSink,
 )
 
 __all__ = [
@@ -54,4 +62,9 @@ __all__ = [
     "TraceAdapter",
     "TraceStep",
     "to_trace_step",
+    "with_studio",
+    # ── Studio 观测接线(C1)──
+    "make_studio_sink",
+    "resolve_studio_url",
+    "StudioSink",
 ]
diff --git a/tier2/gen-worker/observability/studio_sink.py b/tier2/gen-worker/observability/studio_sink.py
new file mode 100644
index 00000000..57861460
--- /dev/null
+++ b/tier2/gen-worker/observability/studio_sink.py
@@ -0,0 +1,337 @@
+"""observability/studio_sink.py —— tier2 富游戏自治线 · 把统一 trace 事件推到 AgentScope Studio(C1 观测接线)。
+
+【这份解决什么问题】
+  tier2 一次生成 run 经 Tier2TraceMiddleware / TraceAdapter.ingest 把每步事件映射成统一 trace 形状
+  (observability/trace.py 的 TraceStep:traceId/step/cost/verdict/timestamp + 推理/动作/观察扩展段),
+  现状只落内存 adapter.steps、收口由编排器读 summary。本模块把这同一条事件流再接一个旁路 sink —— 把每步
+  推成 OTLP span 发给 AgentScope Studio,让 tier2 生成 run 在 Studio 的 Trace 页可视化(服务于创始人
+  「迭代过程拿证据/日志做优化」)。
+
+【Studio 真实接入方式(源码/reference 核实,绝不臆造)】
+  AgentScope Studio 的可观测是 **OpenTelemetry / OTLP 协议** 的(不是某私有 REST 形状):
+    - studio README:「Tracing: OpenTelemetry-based trace visualization for LLM calls, token usage,
+      and agent invocations」(/root/oss/agentscope-studio/README.md)。
+    - studio tracing 教程(/root/oss/agentscope-studio/docs/tutorial/en/develop/tracing.md):Studio 默认暴露
+        · OTLP/Trace/gRPC:localhost:4317(env OTEL_GRPC_PORT 调)
+        · OTLP/Trace/HTTP:localhost:3000(env PORT 调)
+      「Studio currently only supports receiving Trace type data.」
+    - 该教程「Advanced Integration: Import Custom Trace Data」段给了任意来源接 Studio 的官方配方 ——
+      用 OpenTelemetry Python SDK 的 OTLPSpanExporter(endpoint=Studio HTTP)+ BatchSpanProcessor +
+      TracerProvider,自己 start_as_current_span 组 span(原文示例 endpoint="http://localhost:3000")。
+    - 接收侧实现在 studio server(/root/oss/agentscope-studio/packages/server/src/index.ts:136 提到
+      「traces will be received via HTTP endpoint /v1/traces」;OTLP HTTP exporter 自动在 base 末尾拼 /v1/traces)。
+  AgentScope 框架原生那条(agentscope.init(studio_url=...) + middleware.TracingMiddleware)同样落到这个
+  OTLP 端点(middleware/_tracing/_trace.py:用 opentelemetry 全局 tracer 发 span;setup 由 init 装 SDK provider)。
+
+【本模块接的是哪条 / 为什么】
+  tier2 已经有自己一套结构化 trace 事件(TraceStep),且 C1 的任务是「把 TraceAdapter ingest 的事件接到
+  Studio」—— 所以走 **Advanced Integration 自组 span 这条**(把每条 TraceStep 映射成一个 gen-ai 语义约定 span),
+  而不是去替换/抢装 agentscope.init 的全局 TracerProvider(那是编排器/服务启动层的事,会与框架原生 middleware
+  争 set_tracer_provider 全局单例)。本 sink 自带一个 **独立私有 TracerProvider**(不碰 otel 全局),
+  span 经独立 OTLPSpanExporter 直发 Studio HTTP,与框架原生 trace 互不干扰、可单独开关。
+
+【observe-only 铁律(C1 红线)】
+  trace / Studio 是纯旁路:绝不改 verdict / decision / L1 / 任何门判定,绝不中断生成主链。
+  - 取不到 Studio 地址 → sink 直接降级为 no-op(__call__ 立即 return),只留内存 adapter,绝不报错;
+  - 推送失败(Studio 没起 / 网络错 / 缺 otel 依赖)→ best-effort 记一次告警日志后继续,绝不抛、绝不阻塞;
+  - 本模块只「读」TraceStep 往外发,从不回写任何决策字段。
+
+【惰性 import(6c6g 红线)】
+  opentelemetry SDK / exporter 在函数体内 import(6c6g 静态校验环境无这些 wheel),本模块顶层零重依赖 ——
+  保证 6c6g 能 py_compile 且裸 import。真正建 provider / 发 span 只在装了依赖的 mini-desktop / Mac 上发生。
+
+【开关 / 地址来源】
+  Studio HTTP 地址按以下顺序取(默认关闭 = 取不到即 no-op):
+    ① env TIER2_STUDIO_URL(单点覆盖,临时开一次实验最省事);
+    ② service/infra_config.py 的 [studio].http_url(若该分区/项存在;infra.yaml 内网入仓口径);
+    ③ 显式传给 make_studio_sink(studio_url=...) 的参数。
+  三者都没有 → resolve_studio_url() 返回 None → make_studio_sink 返回 no-op sink(现有 CLI 真跑不受影响)。
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Callable, Optional
+
+
+# ── env 开关键(与 genconfig/infra_config 的 env 风格一致:全大写、TIER2_ 前缀)──
+#   取到非空即视为「开 Studio 观测」并用作 Studio HTTP base(如 http://100.64.0.7:3000)。
+ENV_STUDIO_URL = "TIER2_STUDIO_URL"
+
+# ── tier2 这条 trace 在 Studio 里的服务名(OTLP resource service.name;Studio 按它聚合归类)──
+SERVICE_NAME = "tier2-gen-worker"
+
+# ── 一次性告警去重(同一 tag 只 print 一次,避免长 run 刷屏;对齐 genconfig._warn_once 风格)──
+_WARNED: set[str] = set()
+
+
+def _warn_once(tag: str, msg: str) -> None:
+    """同一 tag 的告警只落一次(best-effort 告警去重,绝不抛)。"""
+    if tag in _WARNED:
+        return
+    _WARNED.add(tag)
+    print(f"[tier2-studio-sink] {msg}", flush=True)
+
+
+def resolve_studio_url(explicit: Optional[str] = None) -> Optional[str]:
+    """解析 Studio HTTP base 地址:env > infra_config[studio].http_url > 显式参数;都没有 → None(=关闭)。
+
+    :param explicit: 调用方显式传入的 Studio HTTP 地址(最低优先级,作兜底)。
+    :return: 归一化后的 Studio HTTP base(去尾斜杠);取不到任何来源 → None(make_studio_sink 据此降级 no-op)。
+    """
+    # ① env 单点覆盖(最高优先级)。
+    env_val = os.environ.get(ENV_STUDIO_URL)
+    if env_val and env_val.strip():
+        return env_val.strip().rstrip("/")
+
+    # ② infra_config.py 的 [studio].http_url(惰性 import:infra_config 顶层无重依赖,但仍 best-effort 兜)。
+    try:
+        # 包内/直跑兼容导入(直跑 observability 下脚本时 service 包仍可解析)。
+        try:
+            from service import infra_config  # type: ignore
+        except Exception:  # pragma: no cover —— 直跑兜底:把 gen-worker/ 加进 sys.path 再取
+            import sys
+            from pathlib import Path
+
+            sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+            from service import infra_config  # type: ignore
+
+        cfg_val = infra_config.get("studio", "http_url", None)
+        if cfg_val and str(cfg_val).strip():
+            return str(cfg_val).strip().rstrip("/")
+    except Exception as exc:  # best-effort:读配置任何异常都不阻断,落一次告警后继续
+        _warn_once("infra_config", f"读 infra_config[studio].http_url 失败(忽略):{exc}")
+
+    # ③ 显式参数兜底。
+    if explicit and explicit.strip():
+        return explicit.strip().rstrip("/")
+
+    return None
+
+
+# ── TraceStep 工具结果状态 → OTLP span 状态码的映射(observe 段 verdict 决定 span 是否标 ERROR)──
+#   注意:这只是「把工具/门结果状态如实反映到 span 颜色」,不改任何生成侧判定(observe-only)。
+_ERROR_VERDICTS = {"error", "interrupted", "denied"}
+
+
+class StudioSink:
+    """把统一 trace 事件(TraceStep dict)推成 OTLP span 发给 AgentScope Studio 的旁路 sink。
+
+    用法(插进现有 TraceAdapter 的 sink 槽,零改 ingest / 事件结构):
+        sink = make_studio_sink(trace_id="<gen-task-trace>")   # 取不到 Studio 地址 → 返回 no-op
+        adapter = TraceAdapter(trace_id="<gen-task-trace>", sink=sink)
+        async for event in agent.reply_stream(user_msg):
+            adapter.ingest(event)        # 映射 + 落内存 + 经本 sink 推 Studio,全程 best-effort
+        sink.shutdown()                  # 收口:flush 残留 span(best-effort,不抛)
+
+    sink 契约 = Callable[[dict], None](与 TraceAdapter 的 sink 一致):吃一条 TraceStep dict → 发一个 span。
+    任何环节失败(缺 otel 依赖 / 建 provider 失败 / 发送失败)一律 best-effort:落一次告警、自动降级 no-op、绝不抛。
+    """
+
+    def __init__(self, trace_id: str, studio_url: str) -> None:
+        """
+        Args:
+            trace_id: 本次生成的 traceId(贯穿;作为 span 的 gen_ai.conversation.id,Studio 据它把同一 run 的步归一条 trace)。
+            studio_url: Studio HTTP base(已归一化;OTLP HTTP exporter 会自动在末尾拼 /v1/traces)。
+        """
+        self.trace_id = trace_id
+        self.studio_url = studio_url
+        self._enabled = True          # best-effort 自降级开关:首次建 provider 失败即永久置 False
+        self._provider: Any = None    # 私有 TracerProvider(惰性建;不碰 otel 全局单例)
+        self._tracer: Any = None      # 从私有 provider 取的 tracer
+        self._initialized = False     # 是否已尝试过初始化(成功或失败都置 True,避免每条都重试)
+
+    def _ensure_tracer(self) -> bool:
+        """惰性建私有 TracerProvider + OTLPSpanExporter(HTTP→Studio);成功返回 True,失败 best-effort 降级返回 False。
+
+        为什么用私有 provider 而非 otel 全局 set_tracer_provider:本 sink 是「把 tier2 自己的 TraceStep 另发一份给
+        Studio」的旁路,不应抢占 otel 全局单例(那是 agentscope.init / 框架原生 TracingMiddleware 的地盘);
+        私有 provider 经自带 exporter 直发,互不干扰、可独立开关。
+        """
+        if self._initialized:
+            return self._enabled
+        self._initialized = True
+        try:
+            # 惰性 import:6c6g 无 opentelemetry SDK / exporter wheel,本模块仍可被裸 import(此处才触发依赖)。
+            from opentelemetry.sdk.trace import TracerProvider  # noqa: PLC0415
+            from opentelemetry.sdk.trace.export import (  # noqa: PLC0415
+                BatchSpanProcessor,
+            )
+            from opentelemetry.exporter.otlp.proto.http.trace_exporter import (  # noqa: PLC0415
+                OTLPSpanExporter,
+            )
+            from opentelemetry.sdk.resources import Resource  # noqa: PLC0415
+
+            # resource.service.name = tier2-gen-worker:Studio 按服务名聚合归类这条线的 trace。
+            resource = Resource.create({"service.name": SERVICE_NAME})
+            provider = TracerProvider(resource=resource)
+            # OTLP/HTTP exporter:endpoint 传 Studio HTTP base,SDK 默认在末尾拼 /v1/traces(对齐 studio /v1/traces 接收)。
+            #   显式给 traces 全路径,避免不同 SDK 版本对 base/自动拼路径行为不一致(reference 示例用 base,
+            #   这里给全路径更稳;两者 studio 都收)。
+            exporter = OTLPSpanExporter(endpoint=self.studio_url + "/v1/traces")
+            # 批处理 processor:异步攒批后台发,推送不卡 ingest 主路(observe-only 不阻塞生成)。
+            provider.add_span_processor(BatchSpanProcessor(exporter))
+            # 从私有 provider 取 tracer(不经 otel.trace.get_tracer 全局,避免取到框架/默认 provider)。
+            self._provider = provider
+            self._tracer = provider.get_tracer("tier2-gen-worker", "1.0.0")
+            _warn_once(
+                "init-ok",
+                f"Studio 观测已开(traceId={self.trace_id} → {self.studio_url}/v1/traces)",
+            )
+            return True
+        except Exception as exc:  # best-effort:缺依赖 / 建 provider 失败 → 永久降级 no-op,不抛、不阻塞
+            self._enabled = False
+            _warn_once(
+                "init-fail",
+                f"建 OTLP→Studio 通道失败,本 run 降级为不推 Studio(只留内存 trace):{exc}",
+            )
+            return False
+
+    def __call__(self, trace_step: dict) -> None:
+        """吃一条 TraceStep dict → 发一个 OTLP span 给 Studio(全程 best-effort,绝不抛、绝不阻塞主链)。
+
+        映射(把 TraceStep 公共核心子集 + 扩展段铺成 gen-ai 语义约定 + agentscope 扩展约定的 span 属性):
+          - span.name        = 取 ext.action.tool / 相位 kind / raw.event,组成可读步名(如 "step 3 · write_source")。
+          - gen_ai.conversation.id = traceId(Studio 据它把同一 run 的步归一条 trace —— 与框架 TracingMiddleware 同键)。
+          - agentscope.function.* / gen_ai.usage.* = 把 ext 三段(推理/动作/观察)+ cost.tokens 铺成属性,JSON 序列化。
+          - span 状态:observe 段 verdict ∈ {error/interrupted/denied} → 标 ERROR(如实反映工具/门结果,不改判定)。
+        """
+        if not self._enabled:
+            return  # 已降级 no-op(取不到地址 / 初始化失败):直接返回,observe-only 不影响主链
+        if not self._ensure_tracer():
+            return  # 首次初始化失败已落告警并永久降级
+        try:
+            self._emit_span(trace_step)
+        except Exception as exc:  # best-effort:单条 span 发送失败只告警一次,不连累后续、不抛
+            _warn_once(
+                "emit-fail",
+                f"推 Studio span 失败(best-effort 计告警,traceId={self.trace_id}):{exc}",
+            )
+
+    def _emit_span(self, trace_step: dict) -> None:
+        """把单条 TraceStep 组成一个 span 并即时收尾(同步 start→set→end,交 BatchSpanProcessor 异步发出)。"""
+        import json  # noqa: PLC0415 —— 仅序列化属性用,函数内 import 无妨
+        from opentelemetry.trace import StatusCode  # noqa: PLC0415
+
+        step = trace_step.get("step")
+        ext = trace_step.get("ext") or {}
+        raw = ext.get("raw") or {}
+
+        # ── span 名:优先工具名 / 相位 kind / 原始事件类型,拼上步序,便于在 Studio 列表里一眼认出这一步在干嘛 ──
+        label = (
+            (ext.get("action") or {}).get("tool")
+            or (ext.get("action") or {}).get("kind")
+            or (ext.get("observation") or {}).get("kind")
+            or (ext.get("reasoning") or {}).get("kind")
+            or raw.get("event")
+            or "step"
+        )
+        span_name = f"step {step} · {label}"
+
+        # 用私有 tracer 起 span(end_on_exit 默认 True:with 退出即 end;本步是「一条已成事实的 trace 记录」,
+        #   无需跨步嵌套,即起即收最简单、也不会因异常漏 end)。
+        with self._tracer.start_as_current_span(span_name) as span:
+            # ── 公共核心子集 → 标准/约定属性 ──
+            # traceId 作 gen_ai.conversation.id(与框架原生 TracingMiddleware 同键:Studio 据此把同一 run 的步聚一条 trace)。
+            span.set_attribute("gen_ai.conversation.id", str(self.trace_id))
+            if step is not None:
+                span.set_attribute("agentscope.tier2.step", int(step))
+            ts = trace_step.get("timestamp")
+            if ts:
+                span.set_attribute("agentscope.tier2.timestamp", str(ts))
+
+            # ── cost.tokens → gen_ai.usage.*(Studio 的 token 用量视图按这两个标准键展示)──
+            cost = trace_step.get("cost") or {}
+            tokens = (cost or {}).get("tokens") or {}
+            if "in" in tokens:
+                span.set_attribute("gen_ai.usage.input_tokens", int(tokens.get("in") or 0))
+            if "out" in tokens:
+                span.set_attribute("gen_ai.usage.output_tokens", int(tokens.get("out") or 0))
+
+            # ── tier2 扩展三段 → agentscope.function.* + tier2 自有属性(JSON 序列化,Studio 在 Metadata 区展示)──
+            #   用 agentscope.function.input/output 这两个 reference 明确列出的「Opt-In」扩展键承载推理/动作/观察,
+            #   让 Studio 能按 AgentScope 扩展约定展示;原始三段也各存一份 tier2 私有键便于反查。
+            for seg_key in ("reasoning", "action", "observation"):
+                seg = ext.get(seg_key)
+                if seg:
+                    span.set_attribute(
+                        f"agentscope.tier2.{seg_key}",
+                        json.dumps(seg, ensure_ascii=False),
+                    )
+            # 动作段的工具名也铺到标准键(gen_ai.tool.name),Studio 工具调用视图按它高亮。
+            tool = (ext.get("action") or {}).get("tool")
+            if tool:
+                span.set_attribute("gen_ai.tool.name", str(tool))
+            # 原始事件类型(反查用)。
+            if raw.get("event"):
+                span.set_attribute("agentscope.tier2.raw_event", str(raw.get("event")))
+
+            # ── verdict → span 状态:如实反映工具/门结果(observe-only,不改任何生成判定)──
+            verdict = trace_step.get("verdict")
+            if verdict is not None:
+                span.set_attribute("agentscope.tier2.verdict", str(verdict))
+                if str(verdict).lower() in _ERROR_VERDICTS:
+                    # 工具失败 / 被拒 / 中断 → span 标 ERROR(只是把已发生的结果反映到 trace 颜色,不影响主链)。
+                    span.set_status(StatusCode.ERROR, str(verdict))
+                else:
+                    span.set_status(StatusCode.OK)
+
+    def shutdown(self) -> None:
+        """收口:flush + 关私有 provider,把 BatchSpanProcessor 攒批里残留的 span 发出(best-effort,绝不抛)。
+
+        编排器在一次 run 收口处调一次(对齐 TraceAdapter.summary 那一步);不调也不会崩(进程退出时 batch 可能丢尾批,
+        但 observe-only 容忍 —— trace 丢尾批不影响生成产物)。
+        """
+        if self._provider is None:
+            return
+        try:
+            # force_flush:把攒批里没发的 span 立刻发出(给个上限超时,避免收口卡住)。
+            self._provider.force_flush(timeout_millis=5000)
+        except Exception as exc:  # best-effort:flush 失败只告警
+            _warn_once("flush-fail", f"force_flush 残留 span 失败(忽略):{exc}")
+        try:
+            self._provider.shutdown()
+        except Exception as exc:
+            _warn_once("shutdown-fail", f"关 Studio provider 失败(忽略):{exc}")
+
+
+# ── no-op sink:取不到 Studio 地址时返回它,插进 TraceAdapter.sink 槽等于「没接 Studio」(只留内存 adapter)──
+class _NoopStudioSink:
+    """关闭态占位 sink:__call__ 什么都不做、shutdown 什么都不做。让调用方无需判 None、统一拿 sink + 调 shutdown。"""
+
+    enabled = False
+
+    def __call__(self, trace_step: dict) -> None:  # noqa: D401 —— 故意 no-op
+        return
+
+    def shutdown(self) -> None:  # noqa: D401 —— 故意 no-op
+        return
+
+
+def make_studio_sink(
+    trace_id: str,
+    *,
+    studio_url: Optional[str] = None,
+) -> Callable[[dict], None]:
+    """构造一个 Studio 旁路 sink:解析到 Studio 地址 → StudioSink;取不到 → no-op sink(默认关闭)。
+
+    这是给编排器/run 的唯一入口:无论开没开 Studio,都拿到一个可直接塞进 TraceAdapter(sink=...) 的可调用对象,
+    且都带 .shutdown()(no-op 时也安全可调)—— 调用方零分支判断,observe-only 不影响主链。
+
+    :param trace_id: 本次生成 traceId(贯穿)。
+    :param studio_url: 显式 Studio HTTP base(最低优先级;一般留空,走 env TIER2_STUDIO_URL / infra_config)。
+    :return: 一个 Callable[[dict], None] 且带 .shutdown() 的 sink 对象(StudioSink 或 _NoopStudioSink)。
+    """
+    url = resolve_studio_url(studio_url)
+    if not url:
+        # 取不到地址 = 默认关闭:返回 no-op(现有 CLI 真跑不受任何影响)。
+        return _NoopStudioSink()
+    return StudioSink(trace_id=trace_id, studio_url=url)
+
+
+if __name__ == "__main__":  # pragma: no cover —— 本地自检:无 Studio 地址时应拿到 no-op,且 __call__/shutdown 不炸
+    s = make_studio_sink(trace_id="demo-trace-001")
+    print("sink type:", type(s).__name__)  # 期望 _NoopStudioSink(未设 TIER2_STUDIO_URL)
+    s({"traceId": "demo-trace-001", "step": 0, "ext": {"raw": {"event": "ReplyStartEvent"}}})
+    s.shutdown()
+    print("noop sink ok（observe-only 关闭态:吃事件不报错、shutdown 不报错）")
diff --git a/tier2/gen-worker/observability/trace.py b/tier2/gen-worker/observability/trace.py
index a03f8a91..9d538801 100644
--- a/tier2/gen-worker/observability/trace.py
+++ b/tier2/gen-worker/observability/trace.py
@@ -377,6 +377,59 @@ class TraceAdapter:
         }
 
 
+def with_studio(
+    trace_id: str,
+    *,
+    sink: Optional[Callable[[dict], None]] = None,
+    keep_in_memory: bool = True,
+    studio_url: Optional[str] = None,
+) -> "TraceAdapter":
+    """便利件:建一个把每步既落原 sink、又旁路推 AgentScope Studio 的 TraceAdapter(C1 观测接线)。
+
+    为什么是「便利件」而不改 TraceAdapter 构造:TraceAdapter 的 sink 槽是单 sink 契约(sink(dict)->None),
+    本件不破坏它 —— 而是把「原 sink」与「Studio sink」用一个 _fanout 串成一个 sink 再塞进去(对调用方透明)。
+    Studio sink 默认关闭(取不到 Studio 地址即 no-op,见 studio_sink.make_studio_sink),所以:
+      - 没开 Studio(无 TIER2_STUDIO_URL / infra_config[studio]):行为 == 直接 TraceAdapter(sink=sink),零影响;
+      - 开了 Studio:每步在原 sink 之外多推一份 OTLP span 给 Studio(observe-only,best-effort,不阻塞主链)。
+
+    studio_sink 在函数内**惰性 import**:6c6g 无 opentelemetry 依赖时本模块顶层仍可裸 import(只有真用到本件、
+    且 Studio 地址可达时才触发 otel 依赖)。Studio sink 的 .shutdown() 由编排器在 run 收口处调
+    (经返回 adapter 的 _studio_sink 属性拿到,或调用方自己持有 sink 调 shutdown)。
+
+    :param trace_id: 本次生成 traceId(贯穿)。
+    :param sink: 原有落库 sink(JSONL/入库/message_bus 等);None → 不另落,仅(可能)推 Studio + 内存。
+    :param keep_in_memory: 是否在 adapter.steps 留内存副本(透传 TraceAdapter)。
+    :param studio_url: 显式 Studio HTTP base(一般留空,走 env / infra_config)。
+    :return: 配好 fanout sink 的 TraceAdapter;其 ._studio_sink 属性 = Studio sink(供收口调 shutdown)。
+    """
+    # 惰性 import:把 otel 依赖关进本函数,保证 trace.py 顶层 6c6g 仍可裸 import。
+    from .studio_sink import make_studio_sink  # noqa: PLC0415
+
+    studio_sink = make_studio_sink(trace_id, studio_url=studio_url)
+
+    def _fanout(trace_step: dict) -> None:
+        """把一条 trace_step 同时喂给原 sink 与 Studio sink;**两者各自 best-effort,互不连累**。
+
+        注意:本 fanout 仍被 TraceAdapter._emit_to_sink 的外层 try 兜一次;但为「原 sink 失败不连累推 Studio」、
+        「推 Studio 失败不连累原 sink」,这里再各包一层 —— 任一侧异常只告警、不抛、不阻断另一侧与主链(observe-only)。
+        """
+        if sink is not None:
+            try:
+                sink(trace_step)
+            except Exception as exc:  # 原 sink 失败:不连累 Studio 推送,也不抛(交外层计 dropped)
+                print(f"[tier2-trace] 原 sink 写失败(fanout 兜,traceId={trace_id}):{exc}", flush=True)
+        # Studio sink 自身已 best-effort(取不到地址 = no-op;发送失败只告警),这里再兜一层防御性 try。
+        try:
+            studio_sink(trace_step)
+        except Exception as exc:  # 理论上 studio_sink 自己已兜;防御性再兜,绝不让观测旁路波及主链
+            print(f"[tier2-trace] Studio sink 旁路异常(忽略,traceId={trace_id}):{exc}", flush=True)
+
+    adapter = TraceAdapter(trace_id, sink=_fanout, keep_in_memory=keep_in_memory)
+    # 把 Studio sink 挂到 adapter 上,供编排器在 run 收口处 adapter._studio_sink.shutdown()(flush 残留 span)。
+    adapter._studio_sink = studio_sink  # type: ignore[attr-defined]
+    return adapter
+
+
 if __name__ == "__main__":  # pragma: no cover —— 本地自检:用桩事件(dict 形态)跑映射链路
     import json
 
diff --git a/tier2/gen-worker/service/control_plane.py b/tier2/gen-worker/service/control_plane.py
new file mode 100644
index 00000000..f9499ef8
--- /dev/null
+++ b/tier2/gen-worker/service/control_plane.py
@@ -0,0 +1,425 @@
+"""service/control_plane.py —— tier2 富游戏自治线 · 服务态【有界 resume 控制面】(本批皇冠;B1)。
+
+【它补的是哪个 followup】
+  CLI 主链 worker.agent_loop.studio.run_studio(:404-458)在 Agent 的 ReAct 循环【之外】套了一圈
+  有界 resume:AgentScope 2.0.2 原生 ReAct 在「模型产出无 tool_call 的纯文本回合」即退出
+  (agentscope/agent/_agent.py:612)——实测 M3 调一次 run_gates 看到 decision=fix 就产空文本收尾、
+  循环退出(看一次 verdict 就放弃、不自纠)。run_studio 用「for attempt in range(max_resumes+1):
+  reply→checkpoint→门绿则踹 finish / 否则带 verdict 反馈 resume 续修」兜住了它。
+  但**服务态(service/app.py 的 create_app :8200)没有这道循环**:它的 chat 是每回合一次、
+  fire-and-forget(_service/_chat.py),由消费方(控制面)决定要不要再 POST /chat。service/app.py
+  文件头(:34-42)与 service/bootstrap.py(:25-32)都把它列为 followup。本模块就是来补这个 followup——
+  在服务消费方实现等价 CLI max_resumes 的有界 resume 控制面,**绝不重写任何 REST 调用 / 生成逻辑**,
+  只 import 复用 bootstrap(REST 原语)+ run(纯代码门 / 落库)。
+
+【与 CLI 态的关键差异 —— 为什么不靠内存 session 判收敛】
+  服务态九工具(scaffold_init/write_source/build/run_gates/finish)由 _tier2_tools_factory 在【每个
+  chat 回合】工厂新建一个 Tier2Session(app.py:_tier2_tools_factory:129-154,把 session_id 当 game_id),
+  其内存态 session.finished / last_verdict **跨回合不保留**(每回合一个新 Tier2Session 实例)。
+  所以本控制面【不靠内存 session】判收敛,而是【每回合 chat run 结束后独立跑 run.run_gates 判门】——
+  门是机器判的(run_gates 纯代码、零 LLM、判 on-disk 源工程),这正是项目「门机器判、控制面绝不
+  自评翻绿」的哲学。落库同理:据 on-disk 源工程重建 source_project + run.persist_source_project,
+  不依赖内存 session.finished。
+
+【game_id ↔ run_gates 评门目标的绑定(诚实写清)】
+  服务态九工具实际写文件的工程目录 = run._workdir(<game_id>),而九工具的 game_id 绑的是 AgentScope
+  的 session_id(app.py:_tier2_tools_factory:150「game_id 绑 session_id」)。故本控制面评门时,
+  **run_gates 的 game_id 必须用 start_new_game 返回的 session_id**——这样评的就是九工具刚写的那个工程目录,
+  二者指向同一物理目录(game-runtime/games/_tier2-gen/<session_id>),不会评错对象。
+
+【收敛 / 停机判据(对齐 run_studio:436-449)】
+  外层 for attempt in range(max_resumes+1):
+    a. 等本回合 chat run 结束:消费 SSE GET /sessions/{session_id}/stream,读到本轮 reply 的
+       REPLY_END 事件(agentscope/event/_event.py:EventType.REPLY_END;也认 EXCEED_MAX_ITERS 为结束)
+       即视为「这轮 agent 跑完了」。带总超时兜底;SSE 断流也据当前 on-disk 状态继续评门,绝不卡死。
+    b. 独立评门:res = run.run_gates(session_id, play_spec);v = res['verdict']。
+    c. 门绿(v.decision=='accept' 且 layerResults.L1.passed)→ 据 on-disk 源工程落库,break 成功。
+    d. attempt>=max_resumes → 预算耗尽,break(停机原因 budget_exhausted)。
+    e. 否则:fb = run.verdict_feedback(v);bootstrap.resume_session(..., feedback=续跑指令),回到 a。
+  熔断/异常(resume 抛 / SSE 长时间无事件 / run_gates 异常)→ best-effort 记日志、落部分结果返回,绝不静默吞。
+
+【惰性 import 红线】本模块顶层【只 import 标准库】。bootstrap/run/store/genconfig 这些包内模块顶层会
+  牵出 agentscope / httpx(6c6g 未装),故对它们一律在函数体内 import,保证 6c6g 能
+  `python -m py_compile service/control_plane.py` 且能裸 `import service.control_plane` 不炸。
+  真跑门(run_gates / chrome / esbuild)只在 mini-desktop;6c6g 上 run_gates 返回结构化失败(诚实失败)。
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import Any
+
+
+# ── SSE 事件常量(源码核验 agentscope/event/_event.py:EventType)──
+#   EventBase.model_config use_enum_values=True ⇒ model_dump(mode="json") 的 "type" 字段是字符串值。
+#   ReplyEndEvent.type = EventType.REPLY_END → 序列化后 type=="REPLY_END",标志本回合 agent.reply_stream 收尾。
+#   ExceedMaxItersEvent(type=="EXCEED_MAX_ITERS")是另一种「本回合结束」收尾(内层 ReAct 撞 max_iters)。
+_EVENT_REPLY_END = "REPLY_END"
+_EVENT_EXCEED_MAX_ITERS = "EXCEED_MAX_ITERS"
+# 本回合「结束」事件集合(读到其一即视为这轮 chat run 跑完,可评门)。
+_TURN_END_EVENTS = frozenset({_EVENT_REPLY_END, _EVENT_EXCEED_MAX_ITERS})
+
+
+def _infer_role(rel: str) -> str:
+    """据工程内相对路径推断 A3 fileTree[].role(与 run.load_fixture_scaffold 的 _role 同口径)。
+
+    服务态控制面据 on-disk 文件重建 source_project,需要给每个文件标 role(交付契约 fileTree[].role)。
+    与 fixture 脚手架同一套推断规则,保证重建出的 fileTree role 与生成期一致(无 split-brain)。
+    """
+    if rel == "src/main.js":
+        return "entry"
+    if "/scenes/" in rel:
+        return "scene"
+    if "/systems/" in rel or rel == "src/game-core.js":
+        return "system"
+    if "/data/" in rel:
+        return "config"
+    if "/util/" in rel or rel.endswith("seeded-random.js"):
+        return "lib"
+    if "/assets/" in rel:
+        return "asset-manifest"
+    return "other"
+
+
+def _collect_on_disk_project(game_id: str) -> dict | None:
+    """据 on-disk 工程目录(run._workdir/<game_id>)重建源工程交付形状(七要素)+ 文件全文清单。
+
+    【为什么由控制面据 on-disk 重建,而不取内存 session.finished】服务态九工具每回合工厂新建 Tier2Session,
+    内存态跨回合不保留(见文件头)。门绿后要落库的源工程,只能从九工具刚写盘的工程目录读回重建。
+    重建的形状对齐 worker.toolkit.finish 组装的 A3 源项目契约七要素(schemaVersion/projectType/fileTree/
+    entry/buildProfile/depLock/contentHash/addressing),保证落库口径与 CLI finish 一致(F3 无漂移)。
+
+    Returns:
+        {"source_project": {七要素}, "file_list": [{path, content}]} 或 None(目录不存在/无源文件)。
+    """
+    # 惰性 import:run / toolkit 顶层牵出 agentscope(6c6g 未装),只在真要落库时才 import。
+    from worker import run as _run  # noqa: PLC0415
+    from worker import toolkit as _toolkit  # noqa: PLC0415
+
+    wd = _run._workdir(game_id)
+    if not wd.exists():
+        return None
+
+    # 收集 on-disk 源文件(排除构建产物 bundle / evidence / play-spec / index.html 这些非源文件)。
+    #   只收 src/ 与 data/ 下的源(对齐 toolkit 累积的工程文件树语义),其余是 build/run_gates 副产物。
+    file_list: list[dict] = []
+    for p in sorted(wd.rglob("*")):
+        if not p.is_file():
+            continue
+        rel = str(p.relative_to(wd)).replace("\\", "/")
+        # 跳过构建 / 真玩副产物(非交付源文件)。
+        if rel in ("bundle.iife.js", "play-spec.json", "index.html"):
+            continue
+        if rel.startswith("evidence/") or "/node_modules/" in ("/" + rel):
+            continue
+        try:
+            file_list.append({"path": rel, "content": p.read_text(encoding="utf-8")})
+        except Exception as e:  # noqa: BLE001 —— 单文件读失败记日志、不中断重建其余文件
+            print(f"[tier2-control] ⚠ 重建源工程读文件失败 {rel}: {type(e).__name__}: {e}", flush=True)
+
+    if not file_list:
+        return None
+
+    # 组装 A3 源项目契约七要素(与 toolkit.finish:344-357 同口径;contentHash 复用 toolkit._content_hash)。
+    file_tree = [{"path": f["path"], "role": _infer_role(f["path"])} for f in file_list]
+    chash = _toolkit._content_hash(file_list)
+    source_project = {
+        "schemaVersion": "tier2-1.0",
+        "projectType": "tier2-phaser",
+        "fileTree": file_tree,
+        "entry": "src/main.js",
+        "buildProfile": {
+            "bundler": "esbuild", "format": "iife",
+            "globalName": _run.DEFAULT_GLOBAL_NAME, "minify": True, "target": "es2019",
+        },
+        # depLock.phaser:据 on-disk 无法可靠读出精确版本,用 toolkit 默认值(与 finish 同源)。
+        "depLock": {"phaser": "3.80.1"},
+        "contentHash": chash,
+        "addressing": {"store": "mysql+oss", "fetchById": "contentHash"},
+    }
+    return {"source_project": source_project, "file_list": file_list}
+
+
+async def _wait_for_turn_end(base_url: str, agent_id: str, session_id: str, *,
+                             user_id: str, timeout_s: float, idle_timeout_s: float) -> dict:
+    """消费 SSE GET /sessions/{session_id}/stream,等本回合 chat run 结束(读到 REPLY_END / EXCEED_MAX_ITERS)。
+
+    源码核验(agentscope/app/_router/_session.py:425-534):
+      - 端点 = GET /sessions/{session_id}/stream,**需带 query agent_id**(:432);
+      - 返回 text/event-stream,帧形如 `data: {json}\\n\\n`,空闲每 30s 发心跳注释 `:\\n\\n`(:446/522);
+      - 流先 replay 当前 run 的缓冲事件,再 live 订阅;订阅是长连(不随单回合结束而关,只随 bus 关闭结束),
+        所以本函数靠「读到本回合的 REPLY_END/EXCEED_MAX_ITERS」自行收尾,不等服务端关流。
+
+    兜底纪律(绝不卡死):
+      - 总超时 timeout_s:超时即返回 {ended: False, reason: 'total_timeout'},由上层据 on-disk 继续评门;
+      - 空闲超时 idle_timeout_s:连续这么久没收到任何 data 帧(只有心跳/无字节)→ 视为断流,返回
+        {ended: False, reason: 'idle_timeout'},同样据 on-disk 继续评门(SSE 断流不致命)。
+
+    Returns:
+        {ended: bool, reason: str, endEvent: dict|None}。ended=True 表示读到本回合结束事件。
+    """
+    import httpx  # noqa: PLC0415 —— 惰性 import 红线(6c6g 未必装 httpx)
+
+    headers = {"X-User-Id": user_id} if user_id else {}
+    url = f"{base_url}/sessions/{session_id}/stream"
+    params = {"agent_id": agent_id}
+    t_start = time.perf_counter()
+    last_event_at = t_start
+    print(f"[tier2-control] 订阅 SSE 等本回合结束: session={session_id} agent={agent_id} "
+          f"(总超时 {timeout_s:.0f}s / 空闲超时 {idle_timeout_s:.0f}s)", flush=True)
+    try:
+        # read=None:逐行读流不受单次读超时打断;总时长 / 空闲由本函数显式判(下方 elapsed/idle)。
+        timeout_cfg = httpx.Timeout(connect=15.0, read=None, write=15.0, pool=15.0)
+        async with httpx.AsyncClient(timeout=timeout_cfg) as http:
+            async with http.stream("GET", url, params=params, headers=headers) as resp:
+                if resp.status_code >= 300:
+                    body = (await resp.aread())[:300]
+                    print(f"[tier2-control] ⚠ SSE 订阅非 2xx: {resp.status_code}: {body!r}", flush=True)
+                    return {"ended": False, "reason": f"sse_http_{resp.status_code}", "endEvent": None}
+                async for line in resp.aiter_lines():
+                    now = time.perf_counter()
+                    # 总超时:本回合等太久,据 on-disk 继续(不卡死)。
+                    if now - t_start > timeout_s:
+                        print(f"[tier2-control] ⚠ SSE 等本回合结束总超时(>{timeout_s:.0f}s),"
+                              "据 on-disk 继续评门。", flush=True)
+                        return {"ended": False, "reason": "total_timeout", "endEvent": None}
+                    if not line:
+                        # 空行(SSE 帧分隔)——不更新空闲计时(只有真 data 帧才算「有进展」)。
+                        if now - last_event_at > idle_timeout_s:
+                            print(f"[tier2-control] ⚠ SSE 空闲超时(>{idle_timeout_s:.0f}s 无 data 帧),"
+                                  "疑断流,据 on-disk 继续评门。", flush=True)
+                            return {"ended": False, "reason": "idle_timeout", "endEvent": None}
+                        continue
+                    if line.startswith(":"):
+                        # 心跳注释帧(:\n\n);连接活着但本回合还没结束,据空闲超时判断是否断流。
+                        if now - last_event_at > idle_timeout_s:
+                            print(f"[tier2-control] ⚠ SSE 只收到心跳、空闲超时(>{idle_timeout_s:.0f}s),"
+                                  "据 on-disk 继续评门。", flush=True)
+                            return {"ended": False, "reason": "idle_timeout", "endEvent": None}
+                        continue
+                    if not line.startswith("data:"):
+                        continue
+                    # 收到一个 data 帧:解析事件,刷新空闲计时。
+                    last_event_at = now
+                    payload = line[len("data:"):].strip()
+                    if not payload:
+                        continue
+                    try:
+                        event = json.loads(payload)
+                    except Exception as e:  # noqa: BLE001 —— 单帧解析失败记日志、不中断订阅
+                        print(f"[tier2-control] ⚠ SSE 帧解析失败(已跳过): {type(e).__name__}: {e}", flush=True)
+                        continue
+                    etype = event.get("type")
+                    # 只认本回合的结束事件(REPLY_END / EXCEED_MAX_ITERS);其余事件(模型流 / 工具调用)略过。
+                    if etype in _TURN_END_EVENTS:
+                        print(f"[tier2-control] SSE 读到本回合结束事件 type={etype}"
+                              f"(reply_id={event.get('reply_id')});评门。", flush=True)
+                        return {"ended": True, "reason": etype, "endEvent": event}
+                # 流自然结束(bus 关闭等)——本回合可能已收尾,据 on-disk 评门。
+                print("[tier2-control] ⚠ SSE 流结束但未读到本回合结束事件,据 on-disk 继续评门。", flush=True)
+                return {"ended": False, "reason": "stream_closed", "endEvent": None}
+    except Exception as e:  # noqa: BLE001 —— SSE 任何异常都 best-effort:不卡死,据 on-disk 继续评门
+        print(f"[tier2-control] ⚠ SSE 订阅异常(据 on-disk 继续评门): {type(e).__name__}: {e}", flush=True)
+        return {"ended": False, "reason": f"sse_error:{type(e).__name__}", "endEvent": None}
+
+
+# ── 续跑指令(等价 CLI run_studio:454-458 的「停了但门没绿,按失败门继续修、门绿再 finish」)──
+def _resume_feedback_text(verdict_feedback: str) -> str:
+    """把失败门反馈包成续跑指令(口径对齐 studio.py:454-458)。"""
+    return (
+        "你刚才停下了,但验收门还没全绿——不要放弃。这是上次 run_gates 的失败门:\n"
+        f"{verdict_feedback}\n"
+        "请在循环里:据失败门 write_source 针对性修(数据表 schema 错就先 validate_datatable 看平台要的 key),"
+        "build→run_gates→read_verdict,直到门绿再 finish。一步步来,先修最关键的致命门。")
+
+
+async def drive_generation(
+    base_url: str,
+    game_id: str,
+    brief: str,
+    play_spec: dict | None = None,
+    *,
+    max_resumes: int | None = None,
+    model_name: str = "MiniMax-M3",
+    writer_max_iters: int | None = None,
+    user_id: str = "tier2",
+    sse_turn_timeout_s: float = 1200.0,
+    sse_idle_timeout_s: float = 300.0,
+) -> dict:
+    """服务态有界 resume 控制面:驱动一款富游戏从启动到门绿落库(等价 CLI run_studio 的外层 resume)。
+
+    经 bootstrap 的现成 REST 原语 + run 的纯代码门驱动 service/app.py 的 create_app;**绝不重写 REST/生成逻辑**。
+    收敛判据全靠 run.run_gates 机器判门(零自评、绝不放松门),落库据 on-disk 源工程(不靠内存 session)。
+
+    Args:
+        base_url: 运行中的 Agent Service 根地址(如 http://100.64.0.7:8200)。
+        game_id: 调用方语义上的游戏标识(仅用于日志/返回;真实评门 game_id = 框架分配的 session_id,见下)。
+        brief: 一句话题面。
+        play_spec: 真玩驱动规格(透传 run.run_gates;None 走默认 business-sim driver)。
+        max_resumes: 外层 resume 上限;None → genconfig.get('iteration','max_resumes',6)(对齐 studio)。
+        model_name: M3 模型名(默认 MiniMax-M3,经 new-api 走 Anthropic 原生)。
+        writer_max_iters: 单写 ReAct 放开轮数;None → genconfig.get('iteration','writer_max_iters',40)。
+        user_id: 多租户用户标识(经 X-User-Id 头)。
+        sse_turn_timeout_s: 单回合 SSE 等结束的总超时(兜底,绝不卡死)。
+        sse_idle_timeout_s: 单回合 SSE 空闲(无 data 帧)断流判定超时。
+
+    Returns:
+        {game_id, agent_id, session_id, finished(bool=门绿+落库), attempts, last_verdict,
+         store_addressing, wall_s, stopped_reason}。
+    """
+    # 惰性 import:bootstrap(REST 原语,牵出 httpx/worker)/ run(门 + 落库,牵出 agentscope)/ genconfig。
+    from worker import genconfig, run as _run  # noqa: PLC0415
+    from . import bootstrap  # noqa: PLC0415
+
+    t0 = time.perf_counter()
+    # 旋钮口径对齐 studio:max_resumes / writer_max_iters 缺省从 generation.yaml 读。
+    if max_resumes is None:
+        max_resumes = genconfig.get("iteration", "max_resumes", 6)
+    if writer_max_iters is None:
+        writer_max_iters = genconfig.get("iteration", "writer_max_iters", 40)
+
+    result: dict[str, Any] = {
+        "game_id": game_id, "agent_id": None, "session_id": None,
+        "finished": False, "attempts": 0, "last_verdict": None,
+        "store_addressing": None, "wall_s": 0.0, "stopped_reason": None,
+    }
+
+    # ── ① 启动:bootstrap.start_new_game(注册凭据→建 agent→建 session→发 kick)──
+    #   关键集成点:评门用的 game_id 必须 = 框架分配的 session_id(九工具据 session_id 当 game_id 管目录,
+    #   见 app.py:_tier2_tools_factory:150),否则 run_gates 评的不是九工具刚写的那个工程目录。
+    try:
+        started = await bootstrap.start_new_game(
+            base_url, brief, user_id=user_id, model_name=model_name,
+            writer_max_iters=writer_max_iters)
+    except Exception as e:  # noqa: BLE001 —— 启动失败 best-effort 落结果、记日志,不静默吞
+        result["stopped_reason"] = f"start_failed:{type(e).__name__}: {e}"
+        result["wall_s"] = round(time.perf_counter() - t0, 1)
+        print(f"[tier2-control] game={game_id} 启动失败:{type(e).__name__}: {e}", flush=True)
+        return result
+
+    agent_id = started.get("agent_id")
+    session_id = started.get("session_id")
+    result["agent_id"] = agent_id
+    result["session_id"] = session_id
+    # 评门 game_id = 框架 session_id(与九工具工程目录一致)。
+    gate_game_id = session_id
+    print(f"[tier2-control] game={game_id} 启动: agent={agent_id} session={session_id} "
+          f"(评门 game_id=session_id;max_resumes={max_resumes})", flush=True)
+    if not session_id or not agent_id:
+        result["stopped_reason"] = "start_missing_ids(start_new_game 未回 agent_id/session_id)"
+        result["wall_s"] = round(time.perf_counter() - t0, 1)
+        return result
+
+    last_verdict: dict | None = None
+    # ── ② 外层有界 resume 循环(对齐 run_studio:424-458)──
+    for attempt in range(max_resumes + 1):
+        result["attempts"] = attempt + 1
+        # a. 等本回合 chat run 结束(SSE;超时/断流据 on-disk 继续,绝不卡死)。
+        await _wait_for_turn_end(
+            base_url, agent_id, session_id, user_id=user_id,
+            timeout_s=sse_turn_timeout_s, idle_timeout_s=sse_idle_timeout_s)
+
+        # b. 独立评门:run.run_gates 纯代码判 on-disk 源工程(零 LLM、确定性、绝不自评翻绿)。
+        try:
+            gate = _run.run_gates(gate_game_id, play_spec)
+        except Exception as e:  # noqa: BLE001 —— 评门异常 best-effort:记日志,带空 verdict 进续修分支
+            print(f"[tier2-control] game={game_id} attempt={attempt} run_gates 异常:"
+                  f"{type(e).__name__}: {e}", flush=True)
+            gate = {"rc": 1, "verdict": None, "log": f"run_gates 异常:{type(e).__name__}: {e}"}
+        v = gate.get("verdict") or {}
+        last_verdict = v or last_verdict
+        result["last_verdict"] = last_verdict
+
+        # c. 门绿判定(decision==accept 且 L1.passed)。
+        l1 = ((v.get("layerResults") or {}).get("L1") or {})
+        if v.get("decision") == "accept" and l1.get("passed"):
+            print(f"[tier2-control] game={game_id} attempt={attempt} 验收门全绿,据 on-disk 落库。", flush=True)
+            # 据 on-disk 源工程落库(不依赖内存 session.finished)。
+            built = _collect_on_disk_project(gate_game_id)
+            if built:
+                addr = _run.persist_source_project(
+                    gate_game_id, built["source_project"], built["file_list"], now_ts=time.time())
+                result["store_addressing"] = addr
+                result["finished"] = bool(addr)  # 落库成功才算真 finished(门绿 + 落库)
+                if not addr:
+                    print(f"[tier2-control] game={game_id} ⚠ 门绿但落库失败(产物仍在 GEN_DIR workdir);"
+                          "stopped_reason=persist_failed。", flush=True)
+                    result["stopped_reason"] = "gates_green_but_persist_failed"
+                else:
+                    result["stopped_reason"] = "gates_green_persisted"
+            else:
+                # 门绿却读不到 on-disk 源工程(异常态)——诚实标注,不伪造落库。
+                print(f"[tier2-control] game={game_id} ⚠ 门绿但 on-disk 工程目录读不到源文件,无法落库。",
+                      flush=True)
+                result["stopped_reason"] = "gates_green_but_no_on_disk_source"
+            break
+
+        # d. 预算耗尽:已是最后一次 attempt 仍未门绿 → 停。
+        if attempt >= max_resumes:
+            print(f"[tier2-control] game={game_id} resume 预算耗尽({max_resumes} 次)仍未门绿,停。",
+                  flush=True)
+            result["stopped_reason"] = "budget_exhausted"
+            break
+
+        # e. 门没绿 → 带 verdict 失败反馈 resume 续修(头号自纠机制;等价 studio.py:451-458)。
+        fb = _run.verdict_feedback(v, gate.get("log") or "") if v else \
+            ("尚无 verdict(疑真玩未产出 / harness 未就位)。" + (gate.get("log", "")[:400]))
+        print(f"[tier2-control] game={game_id} attempt={attempt} 门未绿"
+              f"(decision={v.get('decision')}),带反馈 resume 续修。", flush=True)
+        try:
+            await bootstrap.resume_session(
+                base_url, agent_id, session_id, user_id=user_id,
+                feedback=_resume_feedback_text(fb))
+        except Exception as e:  # noqa: BLE001 —— resume 抛 = 熔断:best-effort 落部分结果返回,绝不静默吞
+            print(f"[tier2-control] game={game_id} attempt={attempt} resume 异常(熔断):"
+                  f"{type(e).__name__}: {e}", flush=True)
+            result["stopped_reason"] = f"resume_failed:{type(e).__name__}: {e}"
+            break
+
+    if result["stopped_reason"] is None:
+        # 循环正常走完未命中任何分支(理论不达;兜底标注)。
+        result["stopped_reason"] = "loop_ended"
+    result["wall_s"] = round(time.perf_counter() - t0, 1)
+    print(f"[tier2-control] game={game_id} 结束: finished={result['finished']} "
+          f"attempts={result['attempts']} reason={result['stopped_reason']} "
+          f"wall={result['wall_s']}s", flush=True)
+    return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI 入口:python -m service.control_plane <game_id> --base-url ... --brief ...。
+
+    惰性 import asyncio / argparse(顶层零重依赖红线;此处也只用标准库)。真跑在 mini-desktop。
+    """
+    import argparse  # noqa: PLC0415
+    import asyncio  # noqa: PLC0415
+
+    parser = argparse.ArgumentParser(
+        description="tier2 服务态有界 resume 控制面:驱动一款富游戏到门绿落库。")
+    parser.add_argument("game_id", help="游戏标识(用于日志/返回;真实评门用框架分配的 session_id)。")
+    parser.add_argument("--base-url", required=True,
+                        help="运行中的 Agent Service 根地址(如 http://100.64.0.7:8200)。")
+    parser.add_argument("--brief", required=True, help="一句话题面。")
+    parser.add_argument("--max-resumes", type=int, default=None,
+                        help="外层 resume 上限(默认从 generation.yaml iteration.max_resumes 读)。")
+    parser.add_argument("--writer-max-iters", type=int, default=None,
+                        help="单写 ReAct 放开轮数(默认从 generation.yaml iteration.writer_max_iters 读)。")
+    parser.add_argument("--model-name", default="MiniMax-M3", help="模型名(默认 MiniMax-M3)。")
+    parser.add_argument("--user-id", default="tier2", help="多租户用户标识(默认 tier2)。")
+    args = parser.parse_args(argv)
+
+    res = asyncio.run(drive_generation(
+        args.base_url, args.game_id, args.brief,
+        max_resumes=args.max_resumes, writer_max_iters=args.writer_max_iters,
+        model_name=args.model_name, user_id=args.user_id))
+    # 结构化结果打到 stdout(可被上层脚本捕获)。
+    print(json.dumps(res, ensure_ascii=False, indent=2), flush=True)
+    # 退出码:门绿落库成功 = 0,否则 1(供编排脚本判成败)。
+    return 0 if res.get("finished") else 1
+
+
+if __name__ == "__main__":
+    import sys  # noqa: PLC0415
+
+    sys.exit(main())
diff --git a/tier2/gen-worker/worker/agent_loop/design_team.py b/tier2/gen-worker/worker/agent_loop/design_team.py
index 7d7152a0..41a8a934 100644
--- a/tier2/gen-worker/worker/agent_loop/design_team.py
+++ b/tier2/gen-worker/worker/agent_loop/design_team.py
@@ -179,9 +179,9 @@ async def run_design_team(
     model_factory,
     *,
     model_sink: list | None = None,
-    leader_max_iters: int = 12,
-    per_expert_cap: int = 2,
-    timeout_s: float = 240.0,
+    leader_max_iters: int | None = None,
+    per_expert_cap: int | None = None,
+    timeout_s: float | None = None,
 ) -> str:
     """跑工作室星形设计团队,返回一份连贯的富游戏设计稿(str;形状同原 _design_stage)。
 
@@ -189,9 +189,12 @@ async def run_design_team(
     汇总。成功 = leader 跑完后产出非空最终设计稿;否则抛 DesignTeamError 由调用方 degrade。
 
     预算约束(多 agent 烧 token,必须有上限):
-      - leader_max_iters:leader ReAct 总轮数上限(= 调专家 + 汇总的天花板;默认 12,够调四专家各 ≤2 次 + 汇总);
-      - per_expert_cap:单专家被调上限(默认 2;防 leader 反复刷同一专家);
-      - timeout_s:整团队墙钟硬超时(默认 240s),兜住模型挂起/leader 不收敛。
+      - leader_max_iters:leader ReAct 总轮数上限(= 调专家 + 汇总的天花板;够调四专家各 ≤2 次 + 汇总);
+      - per_expert_cap:单专家被调上限(防 leader 反复刷同一专家);
+      - timeout_s:整团队墙钟硬超时(秒),兜住模型挂起/leader 不收敛。
+    三者均为 None → 运行时读 generation.yaml 的 design_team 段(default 12 / 2 / 240.0;口径同 studio.py
+    读 iteration.* 旋钮)。**这是配置外置的兑现点**:改 generation.yaml 的 design_team.timeout_s 即生效,
+    不必改代码(此前硬编码默认 240 导致 YAML 调 timeout 不生效,已修)。
     任一预算触发都不让设计阶段挂死:超时/异常 → 抛 DesignTeamError → studio degrade 回单 agent。
 
     Args:
@@ -211,6 +214,17 @@ async def run_design_team(
     if model_sink is None:
         model_sink = []  # 容错:调用方没传则内部建(此时成本汇总会漏算 team token,仅兜底用)
 
+    # ── 旋钮外置(运行时读 generation.yaml 的 design_team 段;口径同 studio.py 读 iteration.*)──
+    #   显式传入(非 None)优先;未传 → 从 genconfig 读,default 对齐历史硬编码值(12 / 2 / 240.0)。
+    #   修复点:此前这三个是写死的函数默认值,studio 调用时不传 → generation.yaml 的 design_team.* 形同虚设
+    #   (改 YAML 调 timeout 不生效)。现在 None→genconfig.get,配置外置真正兑现。
+    if leader_max_iters is None:
+        leader_max_iters = genconfig.get("design_team", "leader_max_iters", 12)
+    if per_expert_cap is None:
+        per_expert_cap = genconfig.get("design_team", "per_expert_cap", 2)
+    if timeout_s is None:
+        timeout_s = genconfig.get("design_team", "timeout_s", 240.0)
+
     # 每专家调用计数(跨同名工具多次调用累计;预算闸据它判超限)。
     counter: dict[str, int] = {}
 
diff --git a/tier2/gen-worker/worker/config.py b/tier2/gen-worker/worker/config.py
index 19239fd8..be3b28a2 100644
--- a/tier2/gen-worker/worker/config.py
+++ b/tier2/gen-worker/worker/config.py
@@ -51,6 +51,22 @@ from agentscope.agent import ContextConfig  # noqa: E402  —— 历史压缩配
 DEEPSEEK_FLASH = genconfig.get("model", "deepseek_flash", "deepseek-v4-flash")  # 主力便宜档
 DEEPSEEK_PRO = genconfig.get("model", "deepseek_pro", "deepseek-v4-pro")  # 强便宜档(退路树 Q1 读它的过门率)
 
+# ── 强基线档默认模型名(C3;A 门「M3 证路(软)」)──
+#   用强模型(Opus/Fable)跑同一条 studio/控制面线,把「路本身不通」和「便宜模型太弱」分开:
+#   若强基线都过不了门,说明是路/harness 的问题;若强基线能过、便宜档不过,才是便宜模型天花板(退路树 R3)。
+#   两者都是 Anthropic 原生经 new-api(与 M3 同协议);Fable 列出但不假设网关一定可用(取不到诚实报错,见 build_baseline_model)。
+#   退路树 fallback_tree.STRONG_BASELINE_KEYS 用 "Opus"/"Fable" 当聚合 key;这里的常量是它们在 new-api 上的真实模型名。
+BASELINE_OPUS = genconfig.get("model", "baseline_opus", "claude-opus-4-8")  # 强基线 Opus(A 门主用)
+BASELINE_FABLE = genconfig.get("model", "baseline_fable", "claude-fable-5")  # 强基线 Fable(列出,不假设可用)
+
+# 强基线聚合 key(fallback_tree 用)→ new-api 真实模型名 的映射。
+#   build_baseline_model 接受「聚合 key(Opus/Fable,大小写不敏感)」或「真实模型名」两种入参:
+#   传 key → 查本表换成真名;传真名 → 原样用(便于调用方直接点名网关上的具体模型)。
+_BASELINE_NAME_BY_KEY = {
+    "opus": BASELINE_OPUS,
+    "fable": BASELINE_FABLE,
+}
+
 
 class RecordingChatModel(AnthropicChatModel):
     """记录每次模型调用 usage 的 AnthropicChatModel 子类(per-call token 取证,供成本台账)。
@@ -315,6 +331,72 @@ def build_model_openai(
     )
 
 
+def build_baseline_model(
+    model_name: str,
+    *,
+    base_url: str | None = None,
+    api_key: str | None = None,
+    max_tokens: int | None = None,
+    thinking_budget: int | None = None,
+    stream: bool = False,
+    max_retries: int | None = None,
+    record: bool = True,
+) -> AnthropicChatModel:
+    """构建「强基线」模型客户端(C3;A 门「M3 证路(软)」用)——让同一条 studio/控制面线换 Opus/Fable 等强模型跑。
+
+    为什么要它:A 门要把「路本身不通」和「便宜模型太弱」分开。把强模型(claude-opus-4-8 等)接到与 M3
+    完全同一条生成线上——同样经 build_model 的 Anthropic 原生路、同样的 studio.py 装配——若强基线都过不了门,
+    问题在路/harness;若强基线能过、便宜档不过,才坐实「便宜模型天花板」(退路树 fallback_tree R3 判定即读此)。
+
+    与 build_model 的关系(加性,不动 build_model 的签名与行为):
+      - 本函数复用 build_model 的全部机制(凭据解析 client.resolve_base_url/get_api_key、代理旁路、
+        RecordingChatModel usage 取证、Anthropic 原生 /v1/messages 协议),只是把模型名换成强基线;
+      - **thinking 默认关**:基线只为「软证路通」,默认 thinking_enable=False,故 thinking_budget 默认 0、
+        max_tokens 不受「必须 > thinking_budget」约束(传 0 时 build_model 的 0<max_tokens 恒成立);
+        调用方若想让 Opus 也开 thinking,显式传 thinking_budget>0 即可(此时 build_model 会强制 max_tokens>budget)。
+
+    Args:
+        model_name: 强基线档——可传聚合 key 'Opus'/'Fable'(大小写不敏感,经 _BASELINE_NAME_BY_KEY 换成
+            new-api 真名),或直接传真名(如 'claude-opus-4-8',原样用)。
+        base_url / api_key: 同 build_model;None → 走 client 默认解析(已装代理旁路)。
+        max_tokens: 输出上限;None → 读 generation.yaml 的 model.max_tokens(默认 16000)。
+        thinking_budget: None → 默认 0(关 thinking,纯证路);>0 → 开 thinking 且 build_model 强制 max_tokens>budget。
+        stream / max_retries / record: 同 build_model。
+
+    Returns:
+        AnthropicChatModel(record=True 时为 RecordingChatModel,基线跑的成本也进台账)。
+
+    诚实报错纪律(Fable):本函数不预判某基线在网关上一定可用——它把模型名透传给 build_model,真正的
+      "网关有没有这个模型 / key 有没有权限"由 new-api 在首次真调时裁决。一旦取不到(404/403/连不上),
+      build_model 内部的 anthropic SDK 会抛、经 RecordingChatModel 透传上来——**绝不静默吞**,
+      调用方据真异常判该基线在本网关不可用(而非误以为跑出了 0 过门率)。本机(6c6g)无网关、无 anthropic,
+      故 import/构造可成(惰性),真调一定在 mini-desktop。
+    """
+    # 聚合 key(Opus/Fable)→ new-api 真名;传的若已是真名(不在表里)则原样用。
+    resolved_name = _BASELINE_NAME_BY_KEY.get(model_name.strip().lower(), model_name)
+    # thinking 默认关(基线纯证路):budget=None → 0;此时 build_model 的「max_tokens>budget」恒成立。
+    #   显式传 budget>0 → 开 thinking,build_model 会强制校验 max_tokens>budget。
+    budget = 0 if thinking_budget is None else thinking_budget
+    thinking_on = budget > 0
+    print(
+        f"[tier2-config] build_baseline_model: 入参='{model_name}' → 真名='{resolved_name}' "
+        f"thinking={'on' if thinking_on else 'off'}(budget={budget}) record={record}",
+        flush=True,
+    )
+    # 复用 build_model 全部机制(Anthropic 原生路 + 凭据 + 代理旁路 + 成本取证),只换模型名与 thinking 开关。
+    return build_model(
+        resolved_name,
+        base_url=base_url,
+        api_key=api_key,
+        max_tokens=max_tokens,
+        thinking_enable=thinking_on,
+        thinking_budget=budget,
+        stream=stream,
+        max_retries=max_retries,
+        record=record,
+    )
+
+
 def model_name_from_env(default: str | None = None) -> str:
     """单写 agent 的模型名:env TIER2_MODEL 覆盖 > 显式 default > generation.yaml 的 model.default_model_name(现值 MiniMax-M3)。
 
diff --git a/tier2/gen-worker/worker/genconfig.py b/tier2/gen-worker/worker/genconfig.py
index 4f3d879f..3c78290c 100644
--- a/tier2/gen-worker/worker/genconfig.py
+++ b/tier2/gen-worker/worker/genconfig.py
@@ -73,6 +73,12 @@ _BUILTIN_DEFAULTS: dict[str, dict[str, Any]] = {
         "default_model_name": "MiniMax-M3",     # model_name_from_env 默认(M3 路)
         "deepseek_flash": "deepseek-v4-flash",   # DEEPSEEK_FLASH 便宜主力档
         "deepseek_pro": "deepseek-v4-pro",       # DEEPSEEK_PRO 强便宜救场档
+        # ── 强基线档(C3;A 门「M3 证路(软)」用强模型跑同一条 studio/控制面线,把「路不通」与「便宜模型太弱」分开)──
+        #   Opus = Anthropic 原生经 new-api(与 M3 同路);Fable 同协议但 **不假设网关一定可用**
+        #   (build_baseline_model 取不到即诚实报错,不静默)。退路树 fallback_tree.STRONG_BASELINE_KEYS 读 "Opus"/"Fable" 当 key,
+        #   这里两键存的是 new-api 上的真实模型名(消费方:worker/config.py build_baseline_model 的 baseline_name → 真名映射)。
+        "baseline_opus": "claude-opus-4-8",      # 强基线 Opus(Anthropic 原生路;A 门软证路主用)
+        "baseline_fable": "claude-fable-5",      # 强基线 Fable(列出但不假设网关可用;取不到诚实报错)
         "max_tokens": 16000,                     # build_model / build_model_openai 默认输出上限
         "thinking_enable": True,                 # build_model 默认开 thinking 分离
         "thinking_budget": 8000,                 # build_model 默认 thinking 预算(须 < max_tokens)
diff --git a/tier2/gen-worker/worker/mcp_tools.py b/tier2/gen-worker/worker/mcp_tools.py
new file mode 100644
index 00000000..d4989261
--- /dev/null
+++ b/tier2/gen-worker/worker/mcp_tools.py
@@ -0,0 +1,337 @@
+"""mcp_tools.py —— tier2 单写 toolkit 的【可选】MCP 工具接入层(图说 E 族 · skills/工具面可扩展性落地)。
+
+【这份解决什么问题】
+tier2 单写 agent 现在用 build_toolkit 装九个本地工具(scaffold_init/write_source/build/run_gates/finish…),
+那是 spike feie-005 已验证、过九门的主链,绝不能动。图说 E 族另规划了 skills/工具面的【可扩展性】——
+经 MCP(Model Context Protocol)接入外部工具(如资产检索 / 素材库 / 联网查询类),让生成 agent 在九工具
+之外按需多一组外部能力。本模块就是那条接入线:把「按配置注册若干 MCP server 的工具进 toolkit」收敛成一处,
+默认关闭——不开 MCP 时,toolkit 与现在【字节级一致】(九工具零改),只有显式开关打开才会多挂 MCP 工具。
+
+【AgentScope 2.0.2 真实 MCP API(逐一核对源码,绝不臆造)】
+来源:/root/oss/agentscope/src/agentscope/mcp/{_mcp_client.py,_config.py,__init__.py}
+      /root/oss/agentscope/src/agentscope/tool/{_toolkit.py,_tool_group.py}
+  - MCP 客户端类 = agentscope.mcp.MCPClient(BaseModel),构造参数(全部核实):
+      name(str,须匹配 ^[a-zA-Z0-9_-]+$,会拼成模型可见的工具名 mcp__{name}__{tool})、
+      is_stateful(bool,STDIO 必须 True;HTTP 可 True/False)、
+      mcp_config(StdioMCPConfig | HttpMCPConfig,按 type 字段判别)、
+      可选 enable_tools / disable_tools(list[str])、execution_timeout(float)。
+  - 传输配置两类(agentscope.mcp.{StdioMCPConfig,HttpMCPConfig}):
+      StdioMCPConfig(command, args?, env?, cwd?)        —— 本地进程,stdio 传输;
+      HttpMCPConfig(url, headers?, timeout?=30.0)        —— 远端 HTTP,SSE / streamable-http 由 url 后缀自动判
+                                                           (url 以 /sse 或 /messages/ 结尾 → SSE,否则 streamable-http)。
+  - 注册进 toolkit 的唯一姿势(2.0.2 无 Toolkit.register_tool / register_mcp_client 这类后置方法):
+      构造器 Toolkit(tools=[...], mcps=[MCPClient, ...]) —— mcps 落进保留的 "basic" 组(_toolkit.py:127-134)。
+      toolkit 在运行时【惰性】对每个 client 调 list_tools() 枚举其工具(_toolkit.py:_get_available_tools L519-521),
+      故 stateless 的 HTTP client 不必预先 connect();只有 is_stateful=True 的 client 必须在构造 toolkit 前已 connect
+      (构造器会校验:stateful 且未连接直接 ValueError,_toolkit.py:145-151)。
+  - "basic" 组对象暴露在 toolkit.tool_groups[0].mcps(list[MCPClient],_tool_group.py:38),可后置 append
+    —— 这是 toolkit 构造完成后唯一能再加 MCP 的合法路径(供 async 场景连 stateful 客户端后挂入)。
+
+【两个入口(对应两种调用时机)】
+  ① build_mcp_clients(configs) -> list[MCPClient](同步):
+     只【构造】MCPClient 对象(不 connect、不 list_tools),供 build_toolkit 这种【同步】调用点把结果塞进
+     Toolkit(mcps=...)。默认只接 stateless 客户端(HTTP 且 is_stateful=False,无需 await connect);遇到声明
+     stateful 的配置——同步路径下无法 connect——跳过并告警(让它走 ② 的 async 路径)。
+  ② async def register_mcp_tools(toolkit, *, configs)(异步):
+     供【异步】调用点(已在事件循环里)用:构造 client,对 stateful 的先 await connect(),再 append 进
+     toolkit 的 "basic" 组 mcps。这是 spec 要求的 register_mcp_tools 入口,覆盖 STDIO / stateful HTTP。
+
+【门控:默认关闭(本相红线)】
+配置从 tier2/config/generation.yaml 的【新增 mcp 段】读(genconfig 同源 config_path(),env TIER2_GENCONFIG 可改指);
+也支持 env 单点开关 TIER2_GEN__MCP__ENABLED。判定 enabled 的三要素任一不满足即【整体 no-op】(返回空列表):
+  - mcp.enabled 必须显式为真(默认 false);
+  - mcp.servers 必须是非空列表。
+mcp 段缺失 / 脏 / 解析失败 → no-op + 一次性告警,【绝不抛、绝不中断生成主链】(best-effort,对齐 genconfig 纪律)。
+
+【惰性 import(6c6g 红线)】
+6c6g 没装 agentscope[full] / mcp 客户端依赖;agentscope.mcp 一律在【函数体内】import,模块顶层零重依赖,
+保证 6c6g 能 py_compile + 裸 import 本模块。import 失败 → 当作「MCP 不可用」no-op + 告警,不阻断主链。
+
+【最小示例配置(接一个 MCP server;默认不依赖任何外部 MCP 进程)】
+在 tier2/config/generation.yaml 末尾加(整段默认不存在 = 默认关闭):
+
+    mcp:
+      enabled: false            # 总开关:默认 false。改 true 才会挂 MCP 工具(九工具行为永远不受影响)
+      servers:
+        # 例 A:远端 HTTP MCP(stateless,无需 connect;最适合同步 build_toolkit 路径)
+        - name: asset-search           # 仅 [a-zA-Z0-9_-];会拼成工具名 mcp__asset-search__<tool>
+          transport: http              # http | stdio
+          url: http://100.64.0.x:9xxx/mcp   # streamable-http;以 /sse 结尾则自动走 SSE
+          stateful: false              # HTTP 默认 stateless,同步路径可用
+          # headers: { Authorization: "Bearer xxx" }   # 可选;凭据走 docs/内网凭据与端点.md,勿硬编码
+          # timeout: 30.0
+          # enable_tools: [search_asset]               # 可选白名单
+        # 例 B:本地 STDIO MCP(必 stateful,只能走 async register_mcp_tools 路径)
+        - name: local-fs
+          transport: stdio
+          command: mcp-server-filesystem
+          args: ["--root", "/some/dir"]
+          # stateful: true            # STDIO 强制 stateful(填不填都按 true)
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# 顶层只引标准库 + genconfig(后者顶层也零重依赖);agentscope.mcp 一律在函数体内惰性 import(6c6g 无)。
+try:
+    from . import genconfig as _genconfig  # 包内相对导入(worker.mcp_tools)
+except Exception:  # pragma: no cover —— 兜底:被当顶层模块跑时退化为绝对导入
+    import importlib
+
+    _genconfig = importlib.import_module("worker.genconfig")
+
+
+# ── env 单点总开关名(与 genconfig 的 TIER2_GEN__<AREA>__<KEY> 同前缀口径)──
+_ENV_ENABLED = "TIER2_GEN__MCP__ENABLED"
+
+
+def _warn_once(tag: str, msg: str) -> None:
+    """复用 genconfig 的一次性告警(同 [tier2-genconfig] 渠道)防刷屏;失败兜底用 print。"""
+    try:
+        _genconfig._warn_once(tag, "[mcp] " + msg)  # 复用同一去重集合,统一可追溯前缀
+    except Exception:  # pragma: no cover —— genconfig 内部结构变动时的兜底
+        print(f"[tier2-genconfig] [mcp] {msg}", flush=True)
+
+
+def _read_mcp_section() -> dict[str, Any]:
+    """运行时读 generation.yaml 的【结构化】mcp 段(genconfig.get 只取标量,故此处直接读原始段)。
+
+    复用 genconfig._load_yaml()(带 mtime 缓存 + best-effort,任何异常都返回空 dict);本函数再做一层
+    防御:取不到 / 非 dict → 返回空 dict。绝不抛。
+    """
+    try:
+        data = _genconfig._load_yaml()  # 已缓存 + best-effort;读不到/脏 → {}
+        section = data.get("mcp") if isinstance(data, dict) else None
+        return section if isinstance(section, dict) else {}
+    except Exception as exc:  # noqa: BLE001 —— 读配置绝不中断主链
+        _warn_once("mcp-read", f"读 mcp 配置段失败({type(exc).__name__}: {exc})→ 视为未配置(no-op)。")
+        return {}
+
+
+def _enabled(section: dict[str, Any]) -> bool:
+    """MCP 接入是否开启:env 单点开关 > yaml mcp.enabled;默认 false。
+
+    env TIER2_GEN__MCP__ENABLED 优先(便于一次性实验);其次 yaml 段里的 enabled。两者都按宽松真值判定,
+    缺省一律 false(默认关闭,本相红线)。
+    """
+    import os
+
+    raw = os.environ.get(_ENV_ENABLED)
+    if raw is None:
+        raw = section.get("enabled")
+    if isinstance(raw, bool):
+        return raw
+    if raw is None:
+        return False
+    # 字符串/数值字面量按常见真值判定(避免 bool('false')==True 的坑)。
+    return str(raw).strip().lower() in ("1", "true", "yes", "on")
+
+
+def _server_list(section: dict[str, Any]) -> list[dict]:
+    """取 mcp.servers 列表(每项一个 server 配置 dict);非列表 / 空 → 返回 []。"""
+    servers = section.get("servers")
+    if not isinstance(servers, list):
+        return []
+    # 只保留 dict 项(过滤写错的标量项,best-effort 不抛)。
+    return [s for s in servers if isinstance(s, dict)]
+
+
+def _build_one_client(spec: dict, *, allow_stateful: bool):
+    """据单个 server 配置 dict 构造一个 MCPClient(惰性 import agentscope.mcp;失败/不合法返回 None)。
+
+    Args:
+        spec: 单 server 配置(name / transport / 传输专属字段 / 可选 stateful·enable_tools·disable_tools·timeout)。
+        allow_stateful: 当前调用路径能否承载 stateful 客户端
+            —— 同步路径(build_mcp_clients)传 False:stateful 需 await connect,同步下做不到 → 跳过该 client;
+               异步路径(register_mcp_tools)传 True:由调用方负责后续 connect。
+
+    Returns:
+        构造好的 MCPClient(尚未 connect);任何不合法 / 无法在本路径承载 → 返回 None(并告警)。绝不抛。
+    """
+    # 惰性 import:6c6g 无 agentscope.mcp;import 失败 → 当作 MCP 不可用,返回 None。
+    try:
+        from agentscope.mcp import MCPClient, StdioMCPConfig, HttpMCPConfig
+    except Exception as exc:  # noqa: BLE001 —— 6c6g 等无重依赖环境:不可用即跳过,绝不阻断主链
+        _warn_once("mcp-import", f"agentscope.mcp 不可用({type(exc).__name__})→ 跳过 MCP 接入(no-op)。")
+        return None
+
+    name = (spec.get("name") or "").strip()
+    if not name:
+        _warn_once(f"mcp-noname:{id(spec)}", "某 MCP server 配置缺 name → 跳过该项。")
+        return None
+
+    transport = str(spec.get("transport") or "").strip().lower()
+    try:
+        if transport == "stdio":
+            # STDIO 传输:必 stateful(MCPClient.model_post_init 会强制校验)。同步路径无法 connect → 跳过。
+            if not allow_stateful:
+                _warn_once(
+                    f"mcp-stdio-sync:{name}",
+                    f"server '{name}' 是 stdio(必 stateful,需 await connect),同步 build_toolkit 路径无法承载"
+                    " → 跳过;如需用它,请走 async register_mcp_tools 路径(在事件循环内连接)。",
+                )
+                return None
+            command = (spec.get("command") or "").strip()
+            if not command:
+                _warn_once(f"mcp-stdio-nocmd:{name}", f"stdio server '{name}' 缺 command → 跳过。")
+                return None
+            args = spec.get("args")
+            env = spec.get("env")
+            cwd = spec.get("cwd")
+            cfg = StdioMCPConfig(
+                command=command,
+                args=args if isinstance(args, list) else None,
+                env=env if isinstance(env, dict) else None,
+                cwd=cwd if isinstance(cwd, str) else None,
+            )
+            is_stateful = True  # stdio 强制 stateful
+
+        elif transport == "http":
+            url = (spec.get("url") or "").strip()
+            if not url:
+                _warn_once(f"mcp-http-nourl:{name}", f"http server '{name}' 缺 url → 跳过。")
+                return None
+            headers = spec.get("headers")
+            timeout = spec.get("timeout")
+            cfg = HttpMCPConfig(
+                url=url,
+                headers=headers if isinstance(headers, dict) else None,
+                timeout=float(timeout) if isinstance(timeout, (int, float)) else 30.0,
+            )
+            # HTTP 默认 stateless(同步路径可用);仅当显式 stateful=true 才标 stateful。
+            is_stateful = bool(spec.get("stateful")) is True
+            if is_stateful and not allow_stateful:
+                _warn_once(
+                    f"mcp-http-stateful-sync:{name}",
+                    f"http server '{name}' 声明 stateful=true(需 await connect),同步 build_toolkit 路径无法承载"
+                    " → 跳过;改 stateful=false 或走 async register_mcp_tools 路径。",
+                )
+                return None
+        else:
+            _warn_once(
+                f"mcp-badtransport:{name}",
+                f"server '{name}' transport='{transport}' 非法(应为 http | stdio)→ 跳过。",
+            )
+            return None
+
+        # 可选白名单 / 黑名单 / 执行超时(类型不对就忽略,best-effort)。
+        enable_tools = spec.get("enable_tools")
+        disable_tools = spec.get("disable_tools")
+        exec_timeout = spec.get("execution_timeout")
+        client = MCPClient(
+            name=name,
+            is_stateful=is_stateful,
+            mcp_config=cfg,
+            enable_tools=enable_tools if isinstance(enable_tools, list) else None,
+            disable_tools=disable_tools if isinstance(disable_tools, list) else None,
+            execution_timeout=float(exec_timeout) if isinstance(exec_timeout, (int, float)) else None,
+        )
+        return client
+    except Exception as exc:  # noqa: BLE001 —— 单个 client 构造失败不连累其它 / 不阻断主链
+        _warn_once(
+            f"mcp-build:{name}",
+            f"构造 MCPClient '{name}' 失败({type(exc).__name__}: {exc})→ 跳过该项。",
+        )
+        return None
+
+
+def build_mcp_clients(configs: dict[str, Any] | None = None) -> list:
+    """【同步】据配置构造可在同步路径(build_toolkit→Toolkit(mcps=...))承载的 MCPClient 列表。
+
+    供 build_toolkit 的门控挂点调用:只构造、不 connect、不 list_tools(stateless HTTP 客户端无需 connect,
+    toolkit 在运行时惰性枚举其工具)。stateful / stdio 客户端在此被跳过(需 async connect,见 register_mcp_tools)。
+
+    Args:
+        configs: 直接传入的 mcp 配置段(测试 / 外部注入用);为 None 则运行时从 generation.yaml 的 mcp 段读。
+
+    Returns:
+        list[MCPClient]:可直接喂给 Toolkit(mcps=...) 的客户端列表。
+        【默认关闭 / 未配置 / 不可用 / 任意异常 → 返回 []】(此时 toolkit 行为与现在字节级一致)。绝不抛。
+    """
+    try:
+        section = configs if isinstance(configs, dict) else _read_mcp_section()
+        if not _enabled(section):
+            return []  # 默认关闭:no-op(九工具行为零影响)
+        servers = _server_list(section)
+        if not servers:
+            _warn_once("mcp-noservers", "mcp.enabled=true 但 servers 为空/非列表 → 无 MCP 工具可挂(no-op)。")
+            return []
+        clients = []
+        for spec in servers:
+            c = _build_one_client(spec, allow_stateful=False)  # 同步路径:只收 stateless
+            if c is not None:
+                clients.append(c)
+        if clients:
+            print(f"[tier2-genconfig] [mcp] 已构造 {len(clients)} 个 stateless MCP 客户端,挂入 toolkit basic 组。",
+                  flush=True)
+        return clients
+    except Exception as exc:  # noqa: BLE001 —— 整体 best-effort:任何意外都退化为 no-op
+        _warn_once("mcp-buildall", f"构造 MCP 客户端整体失败({type(exc).__name__}: {exc})→ no-op(九工具不受影响)。")
+        return []
+
+
+async def register_mcp_tools(toolkit, *, configs: dict[str, Any] | None = None) -> int:
+    """【异步】把配置里的 MCP server 工具注册进【已构造好的】toolkit(覆盖 stateful / stdio 客户端)。
+
+    用于已在事件循环内的调用点:构造 client → 对 stateful 的先 await connect() → append 进 toolkit 的
+    "basic" 组 mcps(2.0.2 唯一合法的后置注册路径:toolkit.tool_groups[0].mcps,见 _tool_group.py:38)。
+    toolkit 会在运行时惰性 list_tools() 枚举它们的工具。stateless HTTP 客户端也一并挂入(无需 connect)。
+
+    与 build_mcp_clients 的分工:build_mcp_clients 走同步 build_toolkit 路径(只能接 stateless);本函数走
+    async 路径,额外能连 stateful / stdio。两者都【默认关闭 + best-effort no-op】,九工具行为永不受影响。
+
+    Args:
+        toolkit: 已构造的 agentscope.tool.Toolkit 实例。
+        configs: 直接传入的 mcp 配置段;为 None 则从 generation.yaml 的 mcp 段读。
+
+    Returns:
+        int:成功挂入的 MCP 客户端数(默认关闭 / 未配置 / 不可用 → 0)。绝不抛。
+    """
+    try:
+        section = configs if isinstance(configs, dict) else _read_mcp_section()
+        if not _enabled(section):
+            return 0  # 默认关闭
+        servers = _server_list(section)
+        if not servers:
+            _warn_once("mcp-noservers", "mcp.enabled=true 但 servers 为空/非列表 → 无 MCP 工具可挂(no-op)。")
+            return 0
+
+        # 定位 toolkit 的 "basic" 组(后置注册的唯一落点);拿不到则放弃(不破坏 toolkit)。
+        basic_group = None
+        try:
+            for g in getattr(toolkit, "tool_groups", []) or []:
+                if getattr(g, "name", None) == "basic":
+                    basic_group = g
+                    break
+        except Exception as exc:  # noqa: BLE001
+            _warn_once("mcp-nobasic", f"定位 toolkit basic 组失败({type(exc).__name__})→ 放弃 MCP 注册(no-op)。")
+            return 0
+        if basic_group is None or not isinstance(getattr(basic_group, "mcps", None), list):
+            _warn_once("mcp-nobasic2", "toolkit 无可写的 basic.mcps → 放弃 MCP 注册(no-op,九工具不受影响)。")
+            return 0
+
+        registered = 0
+        for spec in servers:
+            client = _build_one_client(spec, allow_stateful=True)  # async 路径:可承载 stateful
+            if client is None:
+                continue
+            # stateful 客户端必须先连接(toolkit 惰性 list_tools 时要求已 connect)。连接失败 → 跳过该 client。
+            try:
+                if getattr(client, "is_stateful", False):
+                    await client.connect()
+            except Exception as exc:  # noqa: BLE001 —— 单个连接失败不连累其它 / 不阻断主链
+                _warn_once(
+                    f"mcp-connect:{getattr(client, 'name', '?')}",
+                    f"连接 MCP '{getattr(client, 'name', '?')}' 失败({type(exc).__name__}: {exc})→ 跳过该项。",
+                )
+                continue
+            basic_group.mcps.append(client)
+            registered += 1
+
+        if registered:
+            print(f"[tier2-genconfig] [mcp] async 注册了 {registered} 个 MCP 客户端进 toolkit basic 组。", flush=True)
+        return registered
+    except Exception as exc:  # noqa: BLE001 —— 整体 best-effort
+        _warn_once("mcp-regall", f"register_mcp_tools 整体失败({type(exc).__name__}: {exc})→ no-op(九工具不受影响)。")
+        return 0
diff --git a/tier2/gen-worker/worker/toolkit.py b/tier2/gen-worker/worker/toolkit.py
index 5cb2048b..25e207ec 100644
--- a/tier2/gen-worker/worker/toolkit.py
+++ b/tier2/gen-worker/worker/toolkit.py
@@ -376,4 +376,21 @@ def build_toolkit(session: Tier2Session) -> Toolkit:
         FunctionTool(query_asset),
         FunctionTool(finish),
     ]
-    return Toolkit(tools=tools)
+
+    # ── 加性挂点(图说 E 族 · 可选 MCP 工具接入)── 默认关闭,九工具行为零改 ──
+    #   仅当 generation.yaml 的 mcp 段 enabled=true(或 env TIER2_GEN__MCP__ENABLED)且 servers 非空时,
+    #   build_mcp_clients 才返回若干【stateless】MCP 客户端,经 Toolkit(mcps=...) 落进保留的 "basic" 组,
+    #   让 agent 在九工具之外多一组外部工具(2.0.2 MCP 接入的唯一构造姿势,见 mcp_tools.py 头部 API 核对)。
+    #   默认 / 未配置 / 任何异常 → build_mcp_clients 返回 [] → Toolkit(tools=tools, mcps=[]) 与原
+    #   Toolkit(tools=tools) 字节级一致(构造器 mcps or [] → 同为空 list),九工具(spike 已验证主链)绝不受影响。
+    #   stateful / stdio 客户端需 await connect(),不能在此同步路径承载——走 mcp_tools.register_mcp_tools(async)。
+    #   惰性 import:6c6g 无 agentscope.mcp,故在挂点处函数内 import(import 失败即 no-op,不阻断)。
+    mcp_clients: list = []
+    try:
+        from .mcp_tools import build_mcp_clients
+        mcp_clients = build_mcp_clients()  # 默认关闭 → []
+    except Exception as exc:  # noqa: BLE001 —— MCP 接入是加性增强,任何失败都退化为「不挂 MCP」,不连累九工具
+        print(f"[tier2-toolkit] MCP 接入挂点异常({type(exc).__name__}: {exc})→ 仅装九工具(无 MCP)。", flush=True)
+        mcp_clients = []
+
+    return Toolkit(tools=tools, mcps=mcp_clients)