diff --git a/router_system/pipeline.py b/router_system/pipeline.py
index 68080b9..2abdb46 100644
--- a/router_system/pipeline.py
+++ b/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,
diff --git a/tests/test_pipeline_tier_fn.py b/tests/test_pipeline_tier_fn.py
new file mode 100644
index 0000000..fd84e79
--- /dev/null
+++ b/tests/test_pipeline_tier_fn.py
@@ -0,0 +1,161 @@
+"""tier_fn 钩子测试（T-G6）：None 零回归 / T1 / T2 单次直答 / T3 完整管线。"""
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from router_system.pipeline import CollaborativePipeline  # noqa: E402
+from router_system.architect import ArchitectClient  # noqa: E402
+from router_system.worker import WorkerLoop  # noqa: E402
+
+
+def asyncio_run(coro):
+    import asyncio
+    return asyncio.run(coro)
+
+
+class FakeArchitect(ArchitectClient):
+    """可脚本化的假架构师：记录 brief/直答调用。"""
+
+    def __init__(self, calls):
+        super().__init__(model="fake", api_key="k")
+        self.calls = calls
+
+    async def brief(self, query, ws):
+        self.calls.append("brief")
+        return {"goal": "g", "constraints": [], "tags": ["code"],
+                "acceptance": [], "plan": [{"id": "s1", "task": "实现加法",
+                                            "deps": [], "done_criteria": "c"}]}
+
+    async def decide(self, ws):
+        self.calls.append("decide")
+        return {"reply": "hint", "patch_plan": []}
+
+    async def final_review(self, ws):
+        self.calls.append("review")
+        return {"verdict": "done", "notes": "", "fix_issues": []}
+
+    async def _chat_once(self, ws, messages):
+        self.calls.append("chat_once")
+        return "直接答案"
+
+
+class ScriptedWorker(WorkerLoop):
+    """脚本化小模型：先给坏答案（触发自修/升级），再给好答案。"""
+
+    def __init__(self, calls):
+        import asyncio
+        calls_worker = calls
+
+        async def gen(prompt):
+            calls_worker.append("gen")
+            if len([c for c in calls_worker if c == "gen"]) == 1:
+                return "不够长的占位"
+            return ("```python\ndef add(a, b):\n"
+                    "    if not isinstance(a, int) or not isinstance(b, int):\n"
+                    "        raise ValueError\n"
+                    "    return a + b\n\n"
+                    "def test_add():\n"
+                    "    assert add(2, 3) == 5\n```")
+
+        super().__init__(generate=gen, max_fix_attempts=2, model_used="fake-worker")
+
+
+def _pipeline(calls, tier_fn=None, fast_path=True):
+    return CollaborativePipeline(architect=FakeArchitect(calls),
+                                 worker=ScriptedWorker(calls),
+                                 fast_path=fast_path, tier_fn=tier_fn)
+
+
+def test_tier_fn_none_unchanged(tmp_path):
+    """D-G5 零回归：tier_fn=None 现行为（快路径 miss -> 完整管线）。"""
+    calls = []
+    pipe = _pipeline(calls)
+    r = asyncio_run(pipe.run("讲讲排序", request_id="tf-none"))
+    assert "brief" in r.route and "reviewing" in r.route   # 完整管线
+    assert r.status == "done"
+
+
+def test_tier_fn_t1_fast_path(tmp_path):
+    """T1 -> 入口为 fast path（本地直答命中）。"""
+    calls = []
+
+    class GoodWorker(WorkerLoop):
+        def __init__(self):
+            async def gen(prompt):
+                return "排序算法是把一组数据按照特定顺序重新排列的过程，广泛用于数据处理。"
+            super().__init__(generate=gen, max_fix_attempts=2, model_used="fake-worker")
+
+    async def tier_fn(query):
+        return "T1"
+
+    pipe = CollaborativePipeline(architect=FakeArchitect(calls),
+                                 worker=GoodWorker(), fast_path=True, tier_fn=tier_fn)
+    r = asyncio_run(pipe.run("讲讲排序", request_id="tf-t1"))
+    assert r.status == "fast_path"
+    assert any("fast_path@domain" in x for x in r.route)
+
+
+def test_tier_fn_t2_single_cloud_direct(tmp_path):
+    """T2 新档：单次云端直答（chat_once），跳过 brief/循环/review。"""
+    calls = []
+
+    class DirectArch(FakeArchitect):
+        async def _chat_once(self, ws, messages):
+            self.calls.append("chat_once")
+            return "T2 直答结果"
+
+    async def tier_fn(query):
+        return "T2"
+
+    pipe = CollaborativePipeline(architect=DirectArch(calls),
+                                 worker=ScriptedWorker(calls),
+                                 fast_path=True, tier_fn=tier_fn)
+    r = asyncio_run(pipe.run("实现一个函数", request_id="tf-t2"))
+    assert r.status == "fast_path"                       # T2 复用 fast_path 状态位
+    assert "t2:direct" in r.route
+    assert "brief" not in r.route                        # 跳过 brief
+    assert "loop" not in r.route                         # 跳过协作循环
+    assert r.model_used == "fake"                        # 云端模型
+
+
+def test_tier_fn_t3_full_pipeline(tmp_path):
+    """T3 -> 强制完整管线（跳过 fast path，走 brief/循环/review）。"""
+    calls = []
+
+    class DirectArch(FakeArchitect):
+        async def _chat_once(self, ws, messages):
+            self.calls.append("chat_once")
+            return "直接答案"
+
+    async def tier_fn(query):
+        return "T3"
+
+    pipe = CollaborativePipeline(architect=DirectArch(calls),
+                                 worker=ScriptedWorker(calls),
+                                 fast_path=True, tier_fn=tier_fn)
+    r = asyncio_run(pipe.run("实现加法函数", request_id="tf-t3"))
+    assert r.status in ("done", "escalated")
+    assert "brief" in r.route and "reviewing" in r.route
+    assert "fast_path@" not in " ".join(r.route)         # 不走快路径
+
+
+def test_tier_fn_t2_direct_miss_falls_back(tmp_path):
+    """T2 直答失败（上游异常）-> 落回完整管线（D-G1 阶梯）。"""
+    calls = []
+
+    class FailArch(FakeArchitect):
+        async def _chat_once(self, ws, messages):
+            raise RuntimeError("云端不可用")
+
+    async def tier_fn(query):
+        return "T2"
+
+    pipe = CollaborativePipeline(architect=FailArch(calls),
+                                 worker=ScriptedWorker(calls),
+                                 fast_path=False, tier_fn=tier_fn)
+    r = asyncio_run(pipe.run("实现加法函数", request_id="tf-fb"))
+    # 直答失败 -> 落回完整管线（brief/loop/review）
+    assert "brief" in r.route or r.status == "failed"
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index f6f741b..66925b1 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -161,6 +161,6 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T-G3b | KnnHead（架构变体 B）：kNN 投票 + conformal-kNN + 按桶分区/封顶/压缩；hybrid fusion 预留（§14） | ⬜ 待办 | |
 | T-G4 | 线性头：离线训练脚本 + LinearHead 纯 Python 推理 + 登记 | ✅ 完成 | T-G4 |
 | T-G5 | Grader：决策组合（特征门×概率×conformal）+ /v1/route + 三态 mode | ✅ 完成 | T-G5 |
-| T-G6 | live 分流：pipeline tier_fn 三档钩子 + proxy 档位映射 + T2 档升级阶梯 | ⬜ 待办 | |
+| T-G6 | live 分流：pipeline tier_fn 三档钩子 + proxy 档位映射 + T2 档升级阶梯 | ✅ 完成 | T-G6 |
 | T-G7 | 审计+前端：ReviewQueue 抽样 + tier 指标卡 + 客户端来源显示/一键升级 | ⬜ 待办 | |
 | T-G8 | 实验：E-G1/E-G3 报告；（可选）LoraRemote + E-G2 线性 vs LoRA | ⬜ 待办 | |