feat(sense): T-G6 live 分流（pipeline tier_fn 三档钩子，D-G5 唯一受控改动）

- pipeline.py：+tier_fn 可选构造参数（None = 现行为逐字节不变）；
  T1=fast path 入口 / T2=单次云端直答（跳过 brief/循环/review，失败落回
  完整管线 D-G1 阶梯）/ T3=强制完整管线（跳过快路径）；同时兼容
  直接可调用与带 .decide 的 grader 对象
- 测试 +5：None 零回归 / T1 fast path / T2 直答跳过 brief/loop / T3 完整管线 /
  T2 失败落回，全量 410 passed（pipeline 既有测试零回归）

router_system/pipeline.py

@@ -64,6 +64,7 @@ class CollaborativePipeline:
         api_token_cap: int = 8000,
         breach_policy: str = "architect_do",   # architect_do | local_only
         run_dir: str = "runs",
+        tier_fn=None,                          # D-G5：可选三级分级钩子（唯一受控改动）
     ):
         self.architect = architect
         self.worker = worker
@@ -72,6 +73,7 @@ class CollaborativePipeline:
         self.api_token_cap = api_token_cap
         self.breach_policy = breach_policy
         self.run_dir = Path(run_dir)
+        self.tier_fn = tier_fn                 # None = 现行为逐字节不变
 
     # ---------------------------------------------------------------
     # 入口
@@ -83,8 +85,46 @@ class CollaborativePipeline:
         route: List[str] = ["v2"]
         t0 = time.perf_counter() * 1000.0
 
-        # ---- 快路径：小模型直答 + 自验证（省 API 钱，D3） ----
-        if self.fast_path:
+        # ---- 三级分级钩子（D-G5，可选）：T1=fast path / T2=单次云端直答 / T3=完整管线 ----
+        tier = None
+        if self.tier_fn is not None:
+            try:
+                tier = await self.tier_fn(query) if callable(getattr(self.tier_fn, "__call__", None)) \
+                    and not hasattr(self.tier_fn, "run") else None
+            except Exception:
+                tier = None
+        # 兼容两种 tier_fn：直接可调用，或带 .decide 的对象（grader）
+        if tier is None and self.tier_fn is not None and hasattr(self.tier_fn, "decide"):
+            try:
+                d = await self.tier_fn.decide(query, "pipeline")
+                tier = d.tier if getattr(d, "fallback", False) is not True or d.tier else d.tier
+                route.append(f"tier_fn:{tier}@mode:{getattr(d, 'mode', '')}")
+            except Exception:
+                tier = None
+
+        # ---- T2 新增档：单次云端直答（跳过 brief/review/循环）----
+        if tier == "T2":
+            route.append("t2:direct")
+            try:
+                ws2 = Workspace.new(request_id + "-t2", query, self.api_token_cap,
+                                    self.rounds_cap)
+                answer = await self.architect._chat_once(
+                    ws2, [{"role": "user",
+                           "content": "直接回答下面的问题（无需计划，一次答完）：" + query}])
+                return self._finalize(query, ws, response=answer, status="fast_path",
+                                      fast_path=False, route=route,
+                                      model_used=self.architect.model, t0=t0,
+                                      save_ws=False)
+            except Exception:                    # D-G1：直答失败落回完整管线
+                route.append("t2:direct:miss")
+
+        # ---- T3：强制完整管线（跳过快路径） ----
+        force_full = (tier == "T3")
+        if force_full:
+            route.append("tier_fn:T3")
+
+        # ---- 快路径：小模型直答 + 自验证（省 API 钱，D3）；T3 时跳过 ----
+        if self.fast_path and not force_full:
             direct = await self.worker.direct_answer(query)
             dom = self._guess_domain(query)
             passed, _det = self.worker.verifier.verify(dom, "answer.md", direct, query,