feat(tier2): 设计阶段接进服务路径(修服务态不收敛根因)

006 真跑实证:服务态 writer 拿空 design_text → 富游戏三门(tripleLink/economy/latch)全挂、
4 次 resume 不收敛(产物结构合法能渲染,但三系统联动没接对)。CLI 线 run_studio 先跑工作室设计
阶段再写、故更易过门;服务路径此前整段跳过设计(app.py 文件头标的 followup)。

修:bootstrap 加 _run_design_phase——在控制面进程内(有 agentscope+模型)复用 studio._design_stage
(工作室星形多 agent 团队 + 失败 degrade 回单 agent)产出 design_text,喂进 AgentRecord.system_prompt
(roles.writer_system 第二参)。每局建 agent 时跑一次,不随 resume 重跑。best-effort:设计失败返 ""
不阻断生成。do_design 经 drive_generation 透传(control_plane --no-design 可关,对照用)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tier2/gen-worker/service/bootstrap.py

@@ -137,6 +137,36 @@ async def _set_permission_bypass(client: Any, base: str, session_id: str, agent_
     print(f"[tier2-bootstrap] 会话 {session_id} 权限已设 BYPASS(跳过工具 ASK,无人值守自治)。", flush=True)
 
 
+async def _run_design_phase(brief: str, model_name: str, *, use_team: bool = True) -> str:
+    """阶段 1 设计:在控制面进程内跑「工作室设计团队」产出富游戏 design_text,喂给服务态 writer 系统提示词。
+
+    为什么必须有(2026-06-24 cp-smoke-006 真跑实证):服务态此前 writer 拿【空 design_text】
+    (start_new_game 传 ""),M3 单写缺富游戏设计指引 → 富游戏三门(tripleLink/economy/latch)全挂、
+    4 次有界 resume 不收敛(产物结构合法、能渲染,但三系统联动没接对)。CLI 线 run_studio 先跑设计阶段
+    再写、故更易过门。本函数把【同一道设计阶段】接进服务路径:复用 studio._design_stage(工作室星形多 agent
+    团队 + 失败 degrade 回单 agent),在控制面进程内(有 agentscope + 模型)直接跑,产出 design_text 写进
+    AgentRecord.system_prompt(roles.writer_system 的第二参)。设计阶段每局只跑一次(建 agent 时),不随 resume 重跑。
+
+    best-effort:任何失败都返回 ""(绝不阻断生成——拿不到设计也让 writer 试,只是少指引),并记日志。
+    """
+    import functools  # noqa: PLC0415
+    from worker import config  # noqa: PLC0415
+    from worker.agent_loop import studio as _studio  # noqa: PLC0415 —— 牵出 agentscope,控制面进程内可用
+
+    try:
+        # model_factory:零参偏函数,每次返回新 M3 客户端(team 给 leader/各专家各取一个;degrade 单 agent 也取一个)。
+        model_factory = functools.partial(config.build_model, model_name)
+        models: list = []  # 设计阶段造的模型客户端(成本可在此汇总;服务态暂不并入 run 台账,列 followup)
+        design_text = await _studio._design_stage(brief, model_factory, model_sink=models, use_team=use_team)
+        print(f"[tier2-bootstrap] 设计阶段产出 design_text {len(design_text or '')} 字(喂服务态 writer)。",
+              flush=True)
+        return design_text or ""
+    except Exception as e:  # noqa: BLE001 —— 设计阶段绝不阻断主链(拿不到设计也让 writer 试)
+        print(f"[tier2-bootstrap] ⚠ 设计阶段失败(writer 将拿空 design,降级继续):"
+              f"{type(e).__name__}: {e}", flush=True)
+        return ""
+
+
 async def start_new_game(
     base_url: str,
     brief: str,
@@ -145,6 +175,7 @@ async def start_new_game(
     model_name: str = "MiniMax-M3",
     writer_max_iters: int = 40,
     credential_id: str | None = None,
+    do_design: bool = True,
 ) -> dict:
     """【路①】新建:注册凭据(若需)→ 建 agent → 建 session → 发 kick,启动一款新富游戏的生成。
 
@@ -173,13 +204,17 @@ async def start_new_game(
                                user_header=user_id)
             credential_id = cred.get("credential_id") or cred.get("id")
 
-        # ② 建单写 agent:system_prompt=roles.writer_system(题面 + 空 design_text;设计阶段服务态落位见 followup),
+        # ②a 设计阶段(2026-06-24 接进服务路径修收敛):先在控制面进程内跑工作室设计团队产出 design_text,
+        #     喂给 writer 系统提示词——否则 writer 拿空 design、富游戏三门全挂不收敛(006 实证)。best-effort。
+        design_text = await _run_design_phase(brief, model_name) if do_design else ""
+
+        # ②b 建单写 agent:system_prompt=roles.writer_system(题面 + 富游戏 design_text),
         #    react_config 放开轮数,context_config=tier2 富游戏历史压缩(防长程多文件丢约定)。
         #    context_config / react_config 经 model_dump 转 JSON 传(端点 schema 收 ContextConfig/ReActConfig)。
         ctx_cfg = config.build_context_config()
         agent_body = {
             "name": "tier2-writer",
-            "system_prompt": roles.writer_system(brief, "", fixture_hint=""),
+            "system_prompt": roles.writer_system(brief, design_text, fixture_hint=""),
             "context_config": ctx_cfg.model_dump(mode="json"),
             "react_config": {"max_iters": writer_max_iters},
         }

tier2/gen-worker/service/control_plane.py

@@ -244,6 +244,7 @@ async def drive_generation(
     model_name: str = "MiniMax-M3",
     writer_max_iters: int | None = None,
     user_id: str = "tier2",
+    do_design: bool = True,
     sse_turn_timeout_s: float = 1200.0,
     sse_idle_timeout_s: float = 300.0,
 ) -> dict:
@@ -291,7 +292,7 @@ async def drive_generation(
     try:
         started = await bootstrap.start_new_game(
             base_url, brief, user_id=user_id, model_name=model_name,
-            writer_max_iters=writer_max_iters)
+            writer_max_iters=writer_max_iters, do_design=do_design)
     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)
@@ -407,12 +408,14 @@ def main(argv: list[str] | None = None) -> int:
                         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)。")
+    parser.add_argument("--no-design", action="store_true",
+                        help="跳过阶段 1 工作室设计(writer 拿空 design_text;调试/对照用)。默认跑设计。")
     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))
+        model_name=args.model_name, user_id=args.user_id, do_design=not args.no_design))
     # 结构化结果打到 stdout(可被上层脚本捕获)。
     print(json.dumps(res, ensure_ascii=False, indent=2), flush=True)
     # 退出码:门绿落库成功 = 0,否则 1(供编排脚本判成败)。