feat(tier2): 统一 trace 落库 sink + SAA schema + 生成控制面只读后端(切片一 C1/C3a)

JsonlFileSink 把五字段 trace 落 workdir/trace.jsonl(非阻塞·写失败不阻断生成);studio.py sink 接线;SAA 扩展段 schema(接口对称五核心+内容不对称 ext);管理面 3 只读端点 roles/traces/cost(惰性 import 保住「仅 import service.app 不牵 fastapi」红线)。测试 test_jsonl_sink 3 / test_admin_routes 6 全绿。

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

contracts/trace/README.md

@@ -18,6 +18,19 @@ SAA 那一轨的扩展段由 Java 线另一个 adapter 产,字段与本契约的
 
 消费链路按 H 族图说 H2 是「Event System → TracingMiddleware → OpenTelemetry span → AgentScope Studio」。trace 记录最终落进已部署的 MySQL 加对象存储,不上重型可观测中间件;观测要早建,是为了 spike 调试和成本对账当下就用得上,而不是要先铺一套独立基建。`traceId` 贯穿整条生成任务链路,既是反查键,也是成本关联键——它对接 `tier2-verdict.schema.json` 的 `evidence.traceId`,让一次验收终判能反查到它对应的全过程轨迹。成本侧由 `cost.py` 读 new-api 的 `logs.quota` 权威口径,按 `traceId` 关联后把人民币金额 best-effort 回填进 `cost.cost_rmb`(trace 阶段只记 token,不携金额估算)。
 
+## 两份 schema 的关系
+
+本目录维护两份 trace schema,对应两条异构生成线:
+
+| schema 文件 | 对应生成线 | 特有扩展段 |
+|---|---|---|
+| `tier2-trace-event.schema.json` | tier2 富游戏自治线(Python / AgentScope ReAct) | `ext`:reasoning / action / observation 三段(ReAct 推理→动作→观察) |
+| `saa-trace-event.schema.json` | SAA 16 节点廉价生成线(Java / Spring AI Alibaba) | `ext`:nodeIndex / nodeName / nodeStatus(节点位置与执行状态) |
+
+核心五字段(`traceId` / `step` / `cost` / `verdict` / `timestamp`)在两份 schema 里**完全同名同义**——这是 H1「接口对称」子集,让两条线写进同一张 trace 表时无需 join 也能按 `traceId` 对账。`ext` 扩展段各写各的、字段不要求对齐(H1「内容不对称」):tier2 的推理/动作/观察三段 SAA 轨不存在,SAA 的节点编号/名称/状态 tier2 轨也不存在,**没有的字段绝不编造**。
+
+谁消费哪份:目前 tier2 侧 adapter(`tier2/gen-worker/observability/trace.py` 的 `TraceAdapter`)已产出并落盘(`JsonlFileSink` 写进 `_tier2-gen/<game_id>/trace.jsonl`);SAA 侧 adapter 在 Java 后端待产线接线(控制面 phase-1)。
+
 ## 与 H 族图说的对应
 
 设计权威是 `docs/architecture/架构/生成引擎/tier2细节图说-H-观测与成本.md`。

contracts/trace/saa-trace-event.schema.json

@@ -0,0 +1,88 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://wanxiang.ai/contracts/trace/saa-trace-event.schema.json",
+  "title": "SaaTraceEvent",
+  "description": "SAA(Spring AI Alibaba)16 节点廉价生成线的统一 trace 事件契约(H 族 · A2)。【这是什么】一条 trace 记录 = 一次生成轨迹里的一步,由 SAA Java adapter 产出。核心五字段(traceId / step / cost / verdict / timestamp)与 tier2-trace-event.schema.json 完全同名同义(H1 接口对称子集);ext 扩展段是 SAA 这一轨独有的,记录当前节点在 16 节点图上的位置与执行状态(H1 内容不对称:各轨各写各的,没有的字段绝不编造)。【为什么独立成约】两条异构生成线写进同一张 trace 表,靠的是「接口对称、内容不对称」。本 schema 定义 SAA 轨写进统一表的记录形状;tier2 那一轨的扩展段在 tier2-trace-event.schema.json 里,字段与本 schema 的 ext 不同名、也不要求对齐。【谁产】SAA Java 线的 trace adapter(game-cloud/game-module-aigc 或 studio 模块,待接线);本 schema 是字段口径约束,不是采集机制。【状态】spec 已立,SAA Java adapter 待产线接线(控制面 phase-1)。详见 contracts/trace/README.md 与 docs/architecture/架构/生成引擎/tier2细节图说-H-观测与成本.md 图 H1。",
+  "type": "object",
+  "required": ["traceId", "step", "timestamp", "ext"],
+  "additionalProperties": false,
+  "properties": {
+    "traceId": {
+      "description": "公共核心子集字段(两条生成线同名同义必填)。一次生成的轨迹主键,贯穿整条生成任务链路;是对账 / 成本关联键,对接 tier2-verdict.schema.json 的 evidence.traceId。SAA 侧由 Java adapter 在构造时钉定,本次轨迹所有 step 共用同一个 traceId。",
+      "type": "string",
+      "minLength": 1
+    },
+    "step": {
+      "description": "公共核心子集字段。第几步(SAA 轨 = 第几个执行节点,0 起;对应 ext.nodeIndex。tier2 轨语义为 ReAct 第几轮 reply / 工具调用步,两者语义不同但字段名对称)。",
+      "type": "integer",
+      "minimum": 0
+    },
+    "cost": {
+      "description": "公共核心子集字段。这一步折成人民币的成本载体;仅在 SAA 节点真正发起模型调用时非空,其余节点步为 null。cost_rmb 由 cost.py 按 new-api quota 权威口径事后 best-effort 回填(成本权威源是 new-api 计费行,trace 不携金额估算)。",
+      "type": ["object", "null"],
+      "$ref": "#/$defs/cost"
+    },
+    "verdict": {
+      "description": "公共核心子集字段。这一步 / 这道门的裁决(可空)。SAA 轨当前在节点执行终态时填入(success / error / skipped),与 ext.nodeStatus 语义一致;门级裁决由裁决引擎 judge 另行回填,不在 adapter 映射阶段产生。",
+      "type": ["string", "null"]
+    },
+    "timestamp": {
+      "description": "公共核心子集字段。本步节点执行的发生时刻(ISO8601 字符串)。节点缺时间戳时为兜底空串。",
+      "type": "string"
+    },
+    "ext": {
+      "description": "SAA 扩展段(不对称段,各轨各写各的;tier2 轨的扩展段字段与此不同名也不要求对齐)。记录当前节点在 SAA 16 节点图上的位置(nodeIndex / nodeName)与执行状态(nodeStatus)。additionalProperties:false 锁死段名,防止 adapter 之外注入未约定字段。",
+      "$ref": "#/$defs/ext"
+    }
+  },
+  "$defs": {
+    "cost": {
+      "type": ["object", "null"],
+      "description": "成本载体(节点发起模型调用时非空)。tokens 由 adapter 实填(从模型调用 response 抓 input_tokens / output_tokens);cost_rmb 由 cost.py 按 new-api quota 事后回填(权威口径,trace 不携金额估算)。",
+      "additionalProperties": false,
+      "required": ["tokens"],
+      "properties": {
+        "tokens": {
+          "description": "本次模型调用的 token 用量。in / out 由 adapter 实填;cached 预留缓存命中 token(当前 adapter 未填)。",
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["in", "out"],
+          "properties": {
+            "in": { "description": "输入 token 数。", "type": "integer", "minimum": 0 },
+            "out": { "description": "输出 token 数。", "type": "integer", "minimum": 0 },
+            "cached": { "description": "缓存命中 token 数(可选,预留;当前 adapter 未实填)。", "type": "integer", "minimum": 0 }
+          }
+        },
+        "cost_rmb": {
+          "description": "本步折算人民币(可选)。adapter 不填,由 cost.py 读 new-api logs.quota 按 traceId 关联后 best-effort 回填。",
+          "type": "number",
+          "minimum": 0
+        }
+      }
+    },
+    "ext": {
+      "type": "object",
+      "description": "SAA 扩展段对象。三个必填字段描述当前节点在 16 节点图上的位置与执行状态。additionalProperties:false 锁死字段集。",
+      "additionalProperties": false,
+      "required": ["nodeIndex", "nodeName", "nodeStatus"],
+      "properties": {
+        "nodeIndex": {
+          "description": "当前节点在 SAA 16 节点执行图中的位置(0 起整数;与顶层 step 同义但独立落盘,便于按节点序号切片分析)。",
+          "type": "integer",
+          "minimum": 0,
+          "maximum": 15
+        },
+        "nodeName": {
+          "description": "SAA 节点的可读名称(如 game_concept_generator / code_writer / asset_patcher 等;由 Java adapter 按 SAANode.name 填入,便于在 trace 里直接定位是哪个节点步)。",
+          "type": "string",
+          "minLength": 1
+        },
+        "nodeStatus": {
+          "description": "节点执行状态(终态才填;running 表示节点仍在执行中——仅在 adapter 需要记录中间态时使用)。该值同时被填进顶层 verdict 作这一步裁决。",
+          "type": "string",
+          "enum": ["pending", "running", "success", "error", "skipped"]
+        }
+      }
+    }
+  }
+}

tier2/gen-worker/observability/trace.py

@@ -28,6 +28,8 @@ contracts/trace 现行边界:统一 trace 契约的正式 schema 落处 = contra
 
 from __future__ import annotations
 
+import json
+from pathlib import Path
 from typing import Any, Callable, Optional
 
 
@@ -377,6 +379,74 @@ class TraceAdapter:
         }
 
 
+class JsonlFileSink:
+    """把统一 trace 事件(TraceStep dict)追加写到 JSONL 文件的落盘 sink。
+
+    用法:
+        sink = JsonlFileSink(workdir / "trace.jsonl")
+        adapter = TraceAdapter(trace_id="...", sink=sink)
+
+    为什么要它:
+        - 五字段 schema(traceId / step / cost / verdict / timestamp)原样落盘,供成本对账、replay 用;
+        - 每条事件是独立 JSON 行(JSONL),追加写、无需整文件加锁,适合长跑生成;
+        - 落盘路径 = 生成产物 workdir / "trace.jsonl",与 game_id 一一对应,随产物一起留存。
+
+    best-effort 铁律(与 TraceAdapter 一致):
+        - 目录不存在则自动创建;
+        - 写失败只 print 告警,**绝不抛异常**、绝不阻断主生成流程(trace 是旁路观测)。
+    """
+
+    def __init__(self, output_path: "Path | str") -> None:
+        """
+        Args:
+            output_path: trace.jsonl 落盘路径(全路径);父目录不存在时在首次写前自动创建。
+        """
+        self._path = Path(output_path)
+        # 预建父目录(best-effort;目录已存在不报错)。
+        try:
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+        except Exception as exc:
+            # 预建失败(如路径不可写)只告警,每次 __call__ 时再尝试一次。
+            print(
+                f"[tier2-trace] 预建 trace.jsonl 目录失败(best-effort,写时再试):"
+                f"{self._path.parent} — {exc}",
+                flush=True,
+            )
+
+    def __call__(self, trace_step: dict) -> None:
+        """把一条 TraceStep dict 序列化为 JSON 追加写一行到 trace.jsonl。
+
+        五字段(traceId / step / cost / verdict / timestamp)由 TraceAdapter.ingest 已填进
+        trace_step(TraceStep 是 dict 子类),此处原样序列化落盘,保留 ext 扩展段。
+
+        写失败(磁盘满 / 权限错等)只 print 告警,绝不抛 —— 一次落库失败不得废掉已成功的生成产物。
+        """
+        try:
+            # 防御性再建一次目录(首次 __init__ 预建可能因时序失败)。
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+            line = json.dumps(trace_step, ensure_ascii=False)
+            with self._path.open("a", encoding="utf-8") as f:
+                f.write(line + "\n")
+        except Exception as exc:
+            print(
+                f"[tier2-trace] trace.jsonl 写失败(best-effort,不阻断生成,path={self._path}):{exc}",
+                flush=True,
+            )
+
+
+def make_jsonl_sink(output_path: "Path | str") -> "JsonlFileSink":
+    """构造 JSONL 落盘 sink,目标路径 output_path;可直接塞进 TraceAdapter(sink=...)。
+
+    用法示例(在编排器 run_studio 里):
+        sink = make_jsonl_sink(run._workdir(game_id) / "trace.jsonl")
+        tracer = Tier2TraceMiddleware(trace_id=game_id, sink=sink)
+
+    :param output_path: trace.jsonl 落盘路径(全路径)。
+    :return: JsonlFileSink 实例(Callable[[dict], None])。
+    """
+    return JsonlFileSink(output_path)
+
+
 def with_studio(
     trace_id: str,
     *,

tier2/gen-worker/service/admin_routes.py

@@ -0,0 +1,272 @@
+"""service/admin_routes.py —— tier2 管理面只读 API(phase-1 · U-C3a)。
+
+三个只读 GET 端点,供运营回看生成配置与轨迹:
+  GET /control/roles        —— 各生成角色当前用的模型(从 worker/config.py 常量读)。
+  GET /control/traces/{id}  —— 读 run._workdir(traceId)/trace.jsonl 返回全部步骤。
+  GET /control/cost/{id}    —— 对 trace.jsonl 做 new-api quota 口径成本对账。
+
+设计决策(附中文注释说明):
+  ① 全部 GET 只读,不触发任何生成、不写文件、不起真服务;
+  ② worker/config.py 顶层 import agentscope(mini-desktop 上有;6c6g/测试环境无),
+     本模块对 config 一律惰性 import + try/except,确保 6c6g py_compile / import 不炸;
+  ③ /control/traces/{id} 文件不存在返 404(REST 语义:资源不存在即 404),响应体带 found=false
+     方便客户端分支一致处理,优于返回 200+found=false(后者歧义:200 暗示"有了");
+  ④ /control/cost/{id} 取价不可达 → pricingSource="degraded" + 200(不抛 500),
+     对齐 best-effort 铁律(成本观测不得中断主链);
+  ⑤ _workdir_for_trace 与 run._workdir 同口径独立实现(无需 import run),便于轻量测试打桩。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+# 路由前缀 /control;标签 admin 供 OpenAPI docs 分组展示。
+router = APIRouter(prefix="/control", tags=["admin"])
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 辅助:角色 → 模型映射
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _get_roles() -> list[dict]:
+    """读各生成角色当前配的模型名。
+
+    优先路:import worker.config 取真实编译常量(只在 mini-desktop 上成功,那里装了 agentscope)。
+    回落路:worker.config 因缺 agentscope 而 import 失败时,直接读 worker.genconfig(无 agentscope 依赖)。
+    两条路产出相同形状:[{"role": ..., "model": ...}, ...]。
+    角色说明:
+      writer       —— tier2 单写 ReAct agent(默认 MiniMax-M3;env TIER2_MODEL 可临时压)。
+      cheap_flash  —— 便宜档主力(deepseek-v4-flash,经 new-api OpenAI 兼容路)。
+      cheap_pro    —— 便宜档强档(deepseek-v4-pro,退路树 Q1 用)。
+      baseline     —— 跨族对照(当前 deepseek-v4-pro;网关开通 Opus/Fable 后改 generation.yaml 切换)。
+    """
+    try:
+        # 优先路:worker.config 顶层 import agentscope(mini-desktop 真跑时走这条)。
+        from worker import config as _cfg  # noqa: PLC0415 — 惰性 import
+        return [
+            {"role": "writer",      "model": _cfg.model_name_from_env()},
+            {"role": "cheap_flash", "model": _cfg.DEEPSEEK_FLASH},
+            {"role": "cheap_pro",   "model": _cfg.DEEPSEEK_PRO},
+            {"role": "baseline",    "model": _cfg.BASELINE_CROSSCHECK},
+        ]
+    except Exception:
+        # 回落路:agentscope 未装(6c6g / 测试环境)→ genconfig 直读,无 agentscope 依赖。
+        from worker import genconfig as _gc  # noqa: PLC0415
+        return [
+            {"role": "writer",
+             "model": os.environ.get("TIER2_MODEL",
+                                     _gc.get("model", "default_model_name", "MiniMax-M3"))},
+            {"role": "cheap_flash",
+             "model": _gc.get("model", "deepseek_flash", "deepseek-v4-flash")},
+            {"role": "cheap_pro",
+             "model": _gc.get("model", "deepseek_pro", "deepseek-v4-pro")},
+            {"role": "baseline",
+             "model": _gc.get("model", "baseline_crosscheck", "deepseek-v4-pro")},
+        ]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 辅助:workdir 路径计算(与 run._workdir 同口径,无需 import run)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _workdir_for_trace(trace_id: str) -> Path:
+    """计算 trace_id 对应的 tier2 工程目录,口径与 worker/run._workdir 完全一致。
+
+    为何独立实现而不直接 import worker.run:
+      run.py 顶层 import archetypes(agentscope 链),保持本模块尽量轻量(6c6g / 测试环境友好);
+      路径计算是纯 pathlib 操作,独立复制最小逻辑更清晰。
+    路径:
+      本文件在 tier2/gen-worker/service/admin_routes.py → 上溯 4 级到仓根:
+        service/ → gen-worker/ → tier2/ → 仓根(games-development-ai)
+      GEN_DIR = 仓根/game-runtime/games/_tier2-gen(tier2 专用生成目录)。
+    此函数是测试桩点:测试用 unittest.mock.patch 替换它指向临时目录即可,无需真实工程目录存在。
+    """
+    repo_root = Path(__file__).resolve().parents[3]
+    gen_dir = repo_root / "game-runtime" / "games" / "_tier2-gen"
+    return gen_dir / trace_id
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 辅助:从 trace 步骤提取 tokens_by_model
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _extract_tokens_by_model(steps: list[dict]) -> dict[str, dict[str, int]]:
+    """从 trace 步骤列表提取 tokens_by_model 聚合(按模型分 in/out/cached token 累加)。
+
+    关联规则(与 to_trace_step 映射规则一致):
+      - ext.raw.model_name 非空 → 更新「当前模型名」(对应 ModelCallStartEvent)。
+      - cost.tokens.in / out 非零 → 记入当前模型的 bucket(对应 ModelCallEndEvent)。
+    跨多轮模型调用:同一模型的 token 累加;未见模型名时用 "unknown" 兜底。
+
+    :param steps: trace.jsonl 解析后的步骤列表。
+    :return: {model_name: {"in": Σin, "out": Σout, "cached": Σcached}}。
+    """
+    tokens_by_model: dict[str, dict[str, int]] = {}
+    current_model = "unknown"  # 每遇 ModelCallStart 类事件更新
+    for step in steps:
+        # 更新当前模型名:从 ext.raw.model_name 读(ModelCallStartEvent 标记)。
+        ext = step.get("ext") or {}
+        raw = ext.get("raw") or {}
+        model_name_in_step = raw.get("model_name")
+        if model_name_in_step:
+            current_model = str(model_name_in_step)
+        # 读 token 计数:cost.tokens.in / out(ModelCallEndEvent / Tier2ModelCallPhaseEvent 填)。
+        cost = step.get("cost") or {}
+        tokens = cost.get("tokens") if isinstance(cost, dict) else None
+        if not isinstance(tokens, dict):
+            continue
+        in_tok = int(tokens.get("in") or 0)
+        out_tok = int(tokens.get("out") or 0)
+        # cached 字段在 trace 当前格式里一般不出现(由 RecordingChatModel 记录,非 trace 字段),
+        # 预留:若将来 trace 扩展落 tokens.cached 可直接读到。
+        cached_tok = int(tokens.get("cached") or 0)
+        if in_tok == 0 and out_tok == 0:
+            continue  # 零 token 步骤跳过(非 ModelCallEnd)
+        bucket = tokens_by_model.setdefault(current_model, {"in": 0, "out": 0, "cached": 0})
+        bucket["in"] += in_tok
+        bucket["out"] += out_tok
+        bucket["cached"] += cached_tok
+    return tokens_by_model
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 端点一:GET /control/roles
+# ─────────────────────────────────────────────────────────────────────────────
+
+@router.get("/roles")
+def get_roles() -> dict:
+    """GET /control/roles —— 各生成角色当前配的模型(只读,无副作用)。
+
+    数据源:worker/config.py 编译常量(role→model 静态映射);agentscope 不可达时回落 genconfig。
+    返回:{"roles": [{"role": "writer", "model": "MiniMax-M3"}, ...]}。
+    """
+    return {"roles": _get_roles()}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 端点二:GET /control/traces/{trace_id}
+# ─────────────────────────────────────────────────────────────────────────────
+
+@router.get("/traces/{trace_id}")
+def get_traces(trace_id: str):
+    """GET /control/traces/{traceId} —— 读 trace.jsonl 返回全部步骤。
+
+    数据源:run._workdir(traceId)/trace.jsonl(JsonlFileSink 写的 JSONL,五字段:
+      traceId / step / cost / verdict / timestamp + ext 扩展段)。
+
+    返回(200):{"traceId":..., "steps":[...], "count":n, "found":true}。
+    文件不存在(404):{"traceId":..., "steps":[], "count":0, "found":false}。
+
+    REST 选型理由(404 而非 200+found=false):
+      资源不存在 = 404 是 HTTP 语义约定,让客户端无需解析 found 字段即可判断;
+      响应体仍带 found=false + 空 steps 以方便调用方分支一致解析。
+
+    健壮性:坏行跳过不崩(best-effort,对齐 JsonlFileSink 的 best-effort 铁律)。
+    """
+    workdir = _workdir_for_trace(trace_id)
+    trace_file = workdir / "trace.jsonl"
+    if not trace_file.exists():
+        # 资源不存在 → 404;带 found=false 供调用方分支处理。
+        return JSONResponse(
+            status_code=404,
+            content={"traceId": trace_id, "steps": [], "count": 0, "found": False},
+        )
+    steps: list[dict] = []
+    for raw_line in trace_file.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        try:
+            steps.append(json.loads(line))
+        except Exception:
+            # 坏行跳过:不崩、不计入 count(best-effort;trace 写失败不得影响读)。
+            pass
+    return {"traceId": trace_id, "steps": steps, "count": len(steps), "found": True}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 端点三:GET /control/cost/{trace_id}
+# ─────────────────────────────────────────────────────────────────────────────
+
+@router.get("/cost/{trace_id}")
+def get_cost(trace_id: str):
+    """GET /control/cost/{traceId} —— 对 trace.jsonl 做 new-api quota 口径成本对账。
+
+    数据源:
+      ① 读 trace.jsonl → 提取 tokens_by_model(按模型分 in/out token 累加)。
+      ② 调 newapi_pricing.fetch_pricing_params() 活取计费三件套(pricing/qpu/usd_rate)。
+      ③ cost_for_run(tokens, pricing, qpu, usd_rate) 按 new-api quota 公式折¥。
+
+    返回(200):
+      {"traceId":..., "costRmb":float, "byModel":{model:{quota,usd,rmb,...}}, "pricingSource":"live"}
+
+    降级处理(best-effort 铁律,成本观测不得阻断):
+      - trace 不存在         → 404(与 /traces 同语义)。
+      - 无 token 记录        → pricingSource="degraded", costRmb=0.0, byModel={} 返回 200。
+      - 取价不可达(网络/6c6g) → pricingSource="degraded", costRmb=0.0, byModel={} 返回 200。
+      - cost 模块 import 失败 → 同上 degraded 路径(防御)。
+    """
+    workdir = _workdir_for_trace(trace_id)
+    trace_file = workdir / "trace.jsonl"
+    if not trace_file.exists():
+        # trace 文件不存在 → 404(与 /traces 端点语义一致)。
+        return JSONResponse(
+            status_code=404,
+            content={"traceId": trace_id, "costRmb": 0.0, "byModel": {},
+                     "pricingSource": "notfound"},
+        )
+
+    # ① 读 trace.jsonl 并解析步骤(坏行跳过,best-effort)。
+    steps: list[dict] = []
+    for raw_line in trace_file.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        try:
+            steps.append(json.loads(line))
+        except Exception:
+            pass  # 坏行跳过
+
+    # ② 提取 tokens_by_model。
+    tokens_by_model = _extract_tokens_by_model(steps)
+    if not tokens_by_model:
+        # 无 token 记录(trace 全是非 ModelCallEnd 步骤,或还没生成过)→ degraded。
+        print(f"[tier2-admin] GET /cost/{trace_id}: trace 无 token 记录,返回 degraded。",
+              flush=True)
+        return {"traceId": trace_id, "costRmb": 0.0, "byModel": {}, "pricingSource": "degraded"}
+
+    # ③ 惰性 import cost 模块(observability 不依赖 agentscope,但防御性兜住任何 import 错误)。
+    try:
+        from observability.newapi_pricing import fetch_pricing_params  # noqa: PLC0415
+        from observability.cost import cost_for_run  # noqa: PLC0415
+    except Exception as exc:
+        # 极端环境下 import 失败(如依赖缺失)→ degraded,不抛 500。
+        print(f"[tier2-admin] GET /cost/{trace_id}: 导入 cost/pricing 模块失败(degraded):{exc}",
+              flush=True)
+        return {"traceId": trace_id, "costRmb": 0.0, "byModel": {}, "pricingSource": "degraded"}
+
+    # ④ 活取 new-api 计费参数(best-effort:取不到 → degraded,不抛)。
+    pricing_params = fetch_pricing_params()  # 内部 best-effort,失败返 None
+    if pricing_params is None:
+        # 取价不可达(6c6g / 网关下线 / httpx 超时)→ degraded 标识,返 200 不抛 500。
+        print(f"[tier2-admin] GET /cost/{trace_id}: new-api 取价不可达,返回 degraded。", flush=True)
+        return {"traceId": trace_id, "costRmb": 0.0, "byModel": {}, "pricingSource": "degraded"}
+
+    # ⑤ 按 new-api quota 公式折¥(cost_for_run 纯函数,全显式传参,不读全局)。
+    cost_result = cost_for_run(
+        tokens_by_model,
+        pricing_params["pricing"],
+        pricing_params["qpu"],
+        pricing_params["usd_rate"],
+    )
+    return {
+        "traceId": trace_id,
+        "costRmb": cost_result["cost_rmb"],
+        "byModel": cost_result.get("by_model", {}),
+        "pricingSource": "live",
+    }

tier2/gen-worker/service/app.py

@@ -238,7 +238,7 @@ def build_app(*, title: str = SERVICE_TITLE) -> "FastAPI":
         flush=True,
     )
 
-    return create_app(
+    app = create_app(
         storage=RedisStorage(**storage_params),
         message_bus=RedisMessageBus(**bus_params),
         workspace_manager=LocalWorkspaceManager(basedir=ws_basedir),
@@ -252,6 +252,14 @@ def build_app(*, title: str = SERVICE_TITLE) -> "FastAPI":
         title=title,
     )
 
+    # ── 挂管理面只读路由(phase-1 U-C3a;GET /control/roles · /control/traces/{id} · /control/cost/{id})──
+    #   惰性 import 红线:admin_routes 顶层 import fastapi。create_app 既已建出 FastAPI app,
+    #   fastapi 在能调 build_app 的环境(mini-desktop)必然在场;在 build_app 体内 import,保证 6c6g
+    #   仅 `import service.app`(不调 build_app)时不牵 fastapi、不破坏顶层惰性红线。
+    from .admin_routes import router as admin_router  # noqa: PLC0415 —— 惰性 import 红线
+    app.include_router(admin_router)
+    return app
+
 
 def main() -> None:
     """部署入口:起 uvicorn 跑 tier2 Agent Service(只在 mini-desktop 装好 agentscope[full]+redis 后跑)。

tier2/gen-worker/tests/test_admin_routes.py

@@ -0,0 +1,179 @@
+"""test_admin_routes.py —— U-C3a 管理面只读 API 单测(TDD:先红后绿)。
+
+用 FastAPI TestClient(进程内,**不起真 uvicorn 服务**、不连 Redis/agentscope、不跑 gen)验三个只读端点:
+  GET /control/roles        —— 200 + 各角色模型(至少含 writer)。
+  GET /control/traces/{id}  —— 临时样本 trace.jsonl 打桩 _workdir → 200 + steps 数对 + 五字段;
+                               坏行 best-effort 跳过;不存在 id → 404 + found:false。
+  GET /control/cost/{id}    —— 样本 trace + 打桩取价:live 含 costRmb;取价不可达 → degraded 标识,不崩。
+
+跑法(借 cheap-worker venv 解释器,绝不 import/读写 cheap-worker 源码):
+  PYTHONPATH=tier2/gen-worker <venv>/bin/python -m pytest tier2/gen-worker/tests/test_admin_routes.py -v
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from unittest import mock
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+# 包内/直跑兼容:把 gen-worker/ 加进 sys.path,使 service / observability / worker 可 import
+#   (PYTHONPATH 已指 gen-worker 时这步是冗余保险;直接 python 跑本文件也能解析包)。
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_GW = os.path.dirname(_HERE)  # gen-worker/
+if _GW not in sys.path:
+    sys.path.insert(0, _GW)
+
+from service.admin_routes import router  # noqa: E402 —— 被测路由
+
+
+# ── 合法样本 trace.jsonl 内容(五字段 + ext)──────────────────────────────────────
+#   覆盖:ReplyStart(reply 边界)/ ModelCallStart(带 model_name,供成本按模型归集)/
+#   ModelCallEnd(带 token,供成本折算)/ ToolResultEnd(verdict=success)。
+_SAMPLE_TRACE_LINES = [
+    {"traceId": "t-sample", "step": 0, "cost": None, "verdict": None,
+     "timestamp": "2026-06-27T10:00:00",
+     "ext": {"raw": {"event": "ReplyStartEvent", "agent_name": "writer"}}},
+    {"traceId": "t-sample", "step": 1, "cost": None, "verdict": None,
+     "timestamp": "2026-06-27T10:00:01",
+     "ext": {"raw": {"event": "ModelCallStartEvent", "model_name": "MiniMax-M3"}}},
+    {"traceId": "t-sample", "step": 2, "cost": {"tokens": {"in": 1200, "out": 800}}, "verdict": None,
+     "timestamp": "2026-06-27T10:00:02",
+     "ext": {"raw": {"event": "ModelCallEndEvent", "model_call_end": True}}},
+    {"traceId": "t-sample", "step": 3, "cost": None, "verdict": "success",
+     "timestamp": "2026-06-27T10:00:03",
+     "ext": {"observation": {"phase": "observation", "state": "success"}}},
+]
+
+
+@pytest.fixture
+def client() -> TestClient:
+    """最小 FastAPI app,只挂被测 admin router(不调 build_app、不连 Redis/agentscope)。"""
+    app = FastAPI()
+    app.include_router(router)
+    return TestClient(app)
+
+
+def _write_sample_trace(tmp_path: Path) -> Path:
+    """在临时目录写样本工程 workdir + trace.jsonl(末尾附一坏行,验 best-effort 跳过)。
+
+    返回样本 workdir(tmp_path/'t-sample'),其下 trace.jsonl = 五字段合法行 + 1 坏行。
+    """
+    workdir = tmp_path / "t-sample"
+    workdir.mkdir(parents=True, exist_ok=True)
+    lines = [json.dumps(o, ensure_ascii=False) for o in _SAMPLE_TRACE_LINES]
+    lines.append("{ 这是一行坏 JSON —— best-effort 应跳过、不计入 count ")  # 坏行
+    (workdir / "trace.jsonl").write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return workdir
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# 端点一:GET /control/roles
+# ════════════════════════════════════════════════════════════════════════════
+def test_roles_returns_writer(client):
+    """GET /control/roles → 200 + roles 列表含 writer 角色且其 model 非空。"""
+    resp = client.get("/control/roles")
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert "roles" in data, f"响应缺 roles 字段:{data}"
+    roles = {r["role"]: r["model"] for r in data["roles"]}
+    assert "writer" in roles, f"roles 应含 writer:{roles}"
+    assert roles["writer"], f"writer 的 model 不应为空:{roles}"
+    print(f"  [ok] roles={roles}")
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# 端点二:GET /control/traces/{id}
+# ════════════════════════════════════════════════════════════════════════════
+def test_traces_found_with_sample(client, tmp_path):
+    """样本 trace.jsonl 打桩 _workdir → 200 + steps 数对(坏行跳过)+ 五字段在。"""
+    _write_sample_trace(tmp_path)
+    # 打桩:_workdir_for_trace(tid) → tmp_path/tid(端点据此读 <workdir>/trace.jsonl)。
+    with mock.patch("service.admin_routes._workdir_for_trace",
+                    side_effect=lambda tid: tmp_path / tid):
+        resp = client.get("/control/traces/t-sample")
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert data["found"] is True, f"found 应为 True:{data}"
+    # 坏行被跳过 → count == 合法样本行数。
+    assert data["count"] == len(_SAMPLE_TRACE_LINES), \
+        f"steps 数应为 {len(_SAMPLE_TRACE_LINES)}(坏行跳过),实际 {data['count']}"
+    assert len(data["steps"]) == data["count"]
+    # 五字段齐(traceId/step/cost/verdict/timestamp)。
+    core = {"traceId", "step", "cost", "verdict", "timestamp"}
+    for s in data["steps"]:
+        assert core <= set(s.keys()), f"步骤缺五字段,缺 {core - set(s.keys())}:{s}"
+    # 抽验内容:第 2 步 token、第 3 步 verdict。
+    assert data["steps"][2]["cost"]["tokens"]["in"] == 1200
+    assert data["steps"][3]["verdict"] == "success"
+    print(f"  [ok] count={data['count']}(坏行已跳过),五字段齐,token/verdict 对")
+
+
+def test_traces_not_found_returns_404(client, tmp_path):
+    """不存在的 traceId → 404 + found:false + 空 steps。"""
+    with mock.patch("service.admin_routes._workdir_for_trace",
+                    side_effect=lambda tid: tmp_path / tid):
+        resp = client.get("/control/traces/does-not-exist")
+    assert resp.status_code == 404, resp.text
+    data = resp.json()
+    assert data["found"] is False, f"found 应为 False:{data}"
+    assert data["steps"] == [], f"steps 应为空:{data}"
+    assert data["count"] == 0
+    print(f"  [ok] 不存在 traceId → 404 found=false")
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# 端点三:GET /control/cost/{id}
+# ════════════════════════════════════════════════════════════════════════════
+def test_cost_live_with_stub_pricing(client, tmp_path):
+    """样本 trace + 打桩取价(live)→ 200 + costRmb>0 + pricingSource=live + byModel 含模型。"""
+    _write_sample_trace(tmp_path)
+    # 桩 new-api 计费三件套(模拟网关返回,避免真网络;cost_for_run 是纯函数,折算确定)。
+    stub_params = {
+        "pricing": {"MiniMax-M3": {"model_ratio": 2.0, "completion_ratio": 4.0, "cache_ratio": 0.1}},
+        "qpu": 500000.0,
+        "usd_rate": 7.3,
+    }
+    with mock.patch("service.admin_routes._workdir_for_trace",
+                    side_effect=lambda tid: tmp_path / tid), \
+         mock.patch("observability.newapi_pricing.fetch_pricing_params",
+                    return_value=stub_params):
+        resp = client.get("/control/cost/t-sample")
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert "costRmb" in data, f"响应应含 costRmb:{data}"
+    assert data["pricingSource"] == "live", f"pricingSource 应为 live:{data}"
+    assert data["costRmb"] > 0, f"样本有 token,costRmb 应 > 0:{data}"
+    assert "MiniMax-M3" in data["byModel"], f"byModel 应含 MiniMax-M3:{data}"
+    print(f"  [ok] live costRmb={data['costRmb']} byModel={list(data['byModel'])}")
+
+
+def test_cost_degraded_when_pricing_unreachable(client, tmp_path):
+    """取价不可达(fetch_pricing_params→None)→ 200 + pricingSource=degraded + 含 costRmb,不崩。"""
+    _write_sample_trace(tmp_path)
+    with mock.patch("service.admin_routes._workdir_for_trace",
+                    side_effect=lambda tid: tmp_path / tid), \
+         mock.patch("observability.newapi_pricing.fetch_pricing_params",
+                    return_value=None):
+        resp = client.get("/control/cost/t-sample")
+    assert resp.status_code == 200, resp.text  # 取价失败不抛 500
+    data = resp.json()
+    assert data["pricingSource"] == "degraded", f"pricingSource 应为 degraded:{data}"
+    assert "costRmb" in data, f"degraded 也应带 costRmb 字段:{data}"
+    print(f"  [ok] degraded(取价不可达)costRmb={data['costRmb']} 不崩")
+
+
+def test_cost_trace_not_found_returns_404(client, tmp_path):
+    """cost 对不存在的 traceId → 404(与 /traces 同语义,不崩)。"""
+    with mock.patch("service.admin_routes._workdir_for_trace",
+                    side_effect=lambda tid: tmp_path / tid):
+        resp = client.get("/control/cost/no-such-trace")
+    assert resp.status_code == 404, resp.text
+    data = resp.json()
+    assert data["pricingSource"] == "notfound", f"应标 notfound:{data}"
+    print(f"  [ok] cost 不存在 traceId → 404 notfound")

tier2/gen-worker/tests/test_jsonl_sink.py

@@ -0,0 +1,130 @@
+"""test_jsonl_sink.py —— U-C1a JSONL sink 落盘单测(TDD:先红后绿)。
+
+跑:python3 tests/test_jsonl_sink.py(无 pytest,__main__ 跑全部断言)。
+验 JsonlFileSink:
+  1. emit 三条 trace step → trace.jsonl 落盘;
+  2. 每行是合法 JSON;
+  3. 五字段(traceId / step / cost / verdict / timestamp)齐全;
+  4. 目录不存在 → 自动创建;
+  5. 写失败(不可写路径) → 只告警、不抛,不中断调用方。
+"""
+import sys
+import os
+import json
+import tempfile
+from pathlib import Path
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_GW = os.path.dirname(_HERE)  # gen-worker/
+sys.path.insert(0, _GW)
+
+from observability.trace import JsonlFileSink, make_jsonl_sink, TraceAdapter  # noqa: E402
+
+
+# ── 测试一:核心落盘 + 五字段完整性 ──────────────────────────────────────────────
+def test_jsonl_sink_creates_file_and_lines():
+    """emit 三条 trace step → trace.jsonl 落盘、每行合法 JSON、五字段齐、内容对。"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        out = Path(tmpdir) / "trace.jsonl"
+        sink = JsonlFileSink(out)
+
+        # 用 TraceAdapter + dict 桩事件(无需 AgentScope 真环境)驱动 sink。
+        adapter = TraceAdapter(trace_id="test-trace-001", sink=sink)
+        fake_events = [
+            # ReplyStartEvent → 落 raw.reply_boundary / agent_name
+            {"type": "ReplyStartEvent", "created_at": "2026-06-29T10:00:00",
+             "reply_id": "r1", "name": "writer"},
+            # ModelCallEndEvent → cost.tokens 非空
+            {"type": "ModelCallEndEvent", "created_at": "2026-06-29T10:00:01",
+             "input_tokens": 100, "output_tokens": 50},
+            # ToolResultEndEvent → verdict=success
+            {"type": "ToolResultEndEvent", "created_at": "2026-06-29T10:00:02",
+             "state": "success"},
+        ]
+        for e in fake_events:
+            adapter.ingest(e)
+
+        # 断言①:文件已落盘
+        assert out.exists(), f"trace.jsonl 应存在,但未落盘:{out}"
+
+        lines = out.read_text(encoding="utf-8").strip().split("\n")
+        # 断言②:三条事件 → 三行
+        assert len(lines) == 3, f"应有 3 行,实际 {len(lines)} 行:\n" + "\n".join(lines)
+
+        # 断言③:每行合法 JSON + 五字段齐
+        CORE_FIELDS = {"traceId", "step", "cost", "verdict", "timestamp"}
+        for i, line in enumerate(lines):
+            obj = json.loads(line)  # 若非合法 JSON 此处抛
+            missing = CORE_FIELDS - set(obj.keys())
+            assert not missing, f"第 {i} 行缺五字段 {missing}:{line}"
+            assert obj["traceId"] == "test-trace-001", (
+                f"第 {i} 行 traceId 错(期望 test-trace-001,实际 {obj['traceId']})")
+            assert obj["step"] == i, f"第 {i} 行 step 错(期望 {i},实际 {obj['step']})"
+
+        # 断言④:ModelCallEnd 步 cost 非空且 token 字段正确
+        second = json.loads(lines[1])
+        assert second["cost"] is not None, "ModelCallEnd 步 cost 应非空"
+        assert second["cost"]["tokens"]["in"] == 100, "input_tokens 应为 100"
+        assert second["cost"]["tokens"]["out"] == 50, "output_tokens 应为 50"
+
+        # 断言⑤:ToolResultEnd 步 verdict=success
+        third = json.loads(lines[2])
+        assert third["verdict"] == "success", (
+            f"ToolResultEnd 步 verdict 应为 success,实际 {third['verdict']}")
+
+        print(f"  [ok] lines={len(lines)}, traceId={json.loads(lines[0])['traceId']}, "
+              f"cost.tokens.in={second['cost']['tokens']['in']}, verdict={third['verdict']}")
+
+
+# ── 测试二:嵌套目录自动创建 ─────────────────────────────────────────────────────
+def test_jsonl_sink_auto_creates_dir():
+    """目录不存在 → sink 自动创建嵌套目录;emit 成功落盘。"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        out = Path(tmpdir) / "subdir" / "nested" / "trace.jsonl"
+        sink = make_jsonl_sink(out)
+        step = {"traceId": "t2", "step": 0, "cost": None, "verdict": None,
+                "timestamp": "2026-06-29T00:00:00",
+                "ext": {"raw": {"event": "ReplyStartEvent"}}}
+        sink(step)
+
+        assert out.exists(), f"嵌套目录下 trace.jsonl 应自动创建:{out}"
+        obj = json.loads(out.read_text(encoding="utf-8").strip())
+        assert obj["traceId"] == "t2", f"traceId 错:{obj['traceId']}"
+        print(f"  [ok] auto-mkdir:{out}")
+
+
+# ── 测试三:写失败只告警、不抛 ────────────────────────────────────────────────────
+def test_jsonl_sink_write_failure_no_raise():
+    """写失败(不可写路径) → 只 print 告警,绝不抛,不中断调用方。"""
+    # /proc/nonexistent/ 在 macOS 下不存在且不可写,触发写失败场景。
+    bad_path = Path("/proc/nonexistent_for_test_sink/trace.jsonl")
+    sink = JsonlFileSink(bad_path)
+    try:
+        sink({"traceId": "t3", "step": 0, "cost": None, "verdict": None,
+              "timestamp": "", "ext": {"raw": {"event": "X"}}})
+        print("  [ok] write-fail → 只告警,不抛")
+    except Exception as e:
+        raise AssertionError(f"sink 写失败时不应抛异常,但抛了:{e}") from e
+
+
+# ── 入口 ────────────────────────────────────────────────────────────────────────
+if __name__ == "__main__":
+    tests = [
+        ("test_jsonl_sink_creates_file_and_lines", test_jsonl_sink_creates_file_and_lines),
+        ("test_jsonl_sink_auto_creates_dir", test_jsonl_sink_auto_creates_dir),
+        ("test_jsonl_sink_write_failure_no_raise", test_jsonl_sink_write_failure_no_raise),
+    ]
+    passed = failed = 0
+    for name, fn in tests:
+        try:
+            print(f"[RUN] {name}")
+            fn()
+            print(f"[PASS] {name}")
+            passed += 1
+        except Exception as exc:
+            import traceback
+            print(f"[FAIL] {name}: {exc}")
+            traceback.print_exc()
+            failed += 1
+    print(f"\n结果:{passed} 通过 / {failed} 失败")
+    sys.exit(0 if failed == 0 else 1)

tier2/gen-worker/worker/agent_loop/studio.py

@@ -51,11 +51,13 @@ except ImportError:  # pragma: no cover —— 直接 python studio.py 兜底
 try:
     from observability.cost import tokens_by_model_from_records, cost_for_run
     from observability.newapi_pricing import fetch_pricing_params, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE
+    from observability.trace import make_jsonl_sink
 except Exception:  # pragma: no cover —— 直跑/路径未就位兜底
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from observability.cost import tokens_by_model_from_records, cost_for_run  # type: ignore
     from observability.newapi_pricing import (  # type: ignore
         fetch_pricing_params, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE)
+    from observability.trace import make_jsonl_sink  # type: ignore
 
 # 框架接缝(2.0.2):Agent + ReActConfig + UserMsg + AgentState + BYPASS 权限。
 from agentscope.agent import Agent, ReActConfig
@@ -365,9 +367,11 @@ async def run_studio(
     toolkit = build_toolkit(session)
     breaker = CircuitBreakerMiddleware()  # 四道熔断 + 软刹(on_reply / on_system_prompt)
     # trace 接线(H1/H2):traceId 用 game_id(贯穿本次生成,对接 verdict.evidence.traceId / 成本关联键)。
-    #   sink=None → 只留内存 steps(spike 期收口由编排器读 adapter.summary;真落库 sink 随控制面 phase-1 接)。
+    #   JSONL sink → 把每条 trace step(五字段 schema:traceId/step/cost/verdict/timestamp + ext)
+    #   追加写到产物 workdir/trace.jsonl,供成本对账 / replay 用;sink 失败 best-effort 只告警不阻断主链。
     #   单实例对应单 agent;它与 breaker 都挂 on_reply,由框架按 middlewares 列表序串成洋葱链(trace 在外、纯旁路)。
-    tracer = Tier2TraceMiddleware(trace_id=game_id)
+    _trace_jsonl_path = run._workdir(game_id) / "trace.jsonl"
+    tracer = Tier2TraceMiddleware(trace_id=game_id, sink=make_jsonl_sink(_trace_jsonl_path))
     writer_model = config.build_model(
         mname, max_tokens=max_tokens, thinking_budget=thinking_budget)
 
@@ -629,7 +633,8 @@ async def run_studio(
         #   RunRecord 接线(B2)从此抽 cost_yuan / tokens_by_model 填 G4 采集字段。
         "cost": cost,
         # trace 收口摘要(H1/H2,新增):{traceId, steps, dropped, middlewareDropped}。
-        #   spike 期 trace step 留在 tracer.adapter.steps 内存(sink=None);真落库 sink 随控制面 phase-1 接。
+        #   trace step 既留 tracer.adapter.steps 内存(供本摘要 / replay),又经 JsonlFileSink 落盘
+        #   到 workdir/trace.jsonl(控制面 phase-1 已接,见上文 make_jsonl_sink 接线点)。
         "trace": trace_summary,
         "wall_s": wall_s,
         "file_tree": session.file_tree(),