diff --git a/gateway/sense/features.py b/gateway/sense/features.py
new file mode 100644
index 0000000..b860a91
--- /dev/null
+++ b/gateway/sense/features.py
@@ -0,0 +1,54 @@
+"""特征门（T-G5）：单轮/长度/意图黑名单/仓库级信号（纯函数）。
+
+D-G1：t1_hard_ok=False 时禁止判 T1（硬门）；repo_signals=True 时倾向 T3。
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from gateway.sense.config import SenseConfig
+
+_REPO_SIGNALS = ("多文件", "仓库", "项目结构", "跨模块", "架构", "脚手架",
+                 "migration", "refactor", "scaffold", "repository")
+
+
+@dataclass
+class Features:
+    turns: int = 1                    # 非 system 消息数
+    est_tokens: int = 0               # 字符/3 估算（中文友好近似）
+    single_turn: bool = True
+    intent_blocked: bool = False      # 命中意图黑名单
+    repo_signals: bool = False        # 仓库级/多文件信号
+    over_length: bool = False         # 超出 t1 长度门
+    t1_hard_ok: bool = True           # 任一 T1 硬门不过即 False
+
+
+def _to_text_and_turns(text_or_messages) -> tuple[str, int]:
+    if isinstance(text_or_messages, str):
+        return text_or_messages, 1
+    msgs = [m for m in (text_or_messages or []) if isinstance(m, dict)]
+    non_system = [m for m in msgs if str(m.get("role")) != "system"]
+    text = "\n".join(str(m.get("content") or "") for m in non_system)
+    return text, max(1, len(non_system))
+
+
+def gate(text_or_messages, consumer: str, cfg: SenseConfig) -> Features:
+    """特征门（§7 签名）：返回门特征 + t1_hard_ok。"""
+    text, turns = _to_text_and_turns(text_or_messages)
+    est_tokens = len(text) // 3
+    single_turn = turns <= 1
+
+    blocked = any(word in text for word in cfg.intent_blacklist)
+    repo = any(word in text for word in _REPO_SIGNALS)
+    over_length = est_tokens > cfg.t1_max_tokens
+    multi_turn = turns > cfg.t1_max_turns
+
+    t1_hard_ok = single_turn and not over_length and not blocked and not repo \
+        and not multi_turn
+
+    return Features(
+        turns=turns, est_tokens=est_tokens, single_turn=single_turn,
+        intent_blocked=blocked, repo_signals=repo,
+        over_length=over_length, t1_hard_ok=t1_hard_ok)
diff --git a/gateway/sense/grader.py b/gateway/sense/grader.py
new file mode 100644
index 0000000..aeffd4a
--- /dev/null
+++ b/gateway/sense/grader.py
@@ -0,0 +1,155 @@
+"""分级决策组合（T-G5，§8 主时序）：
+
+  embedder.embed（挂 -> fallback=T2+规则门，D-G4）
+  -> features.gate（t1_hard_ok）
+  -> LinearHead.predict -> {p1,p2,p3}
+  -> conformal 阈值：p1>=τ1 且 t1_hard_ok -> T1；p3>=τ3 或 repo_signals -> T3；其余 T2
+  -> mode 裁剪：collect/shadow 只写观察（decided≠executed），live 返回决策
+  -> observer.log（全模式必写）
+"""
+from __future__ import annotations
+
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+from gateway.sense.calibrate import load_active
+from gateway.sense.classifier import LinearHead
+from gateway.sense.config import SenseConfig
+from gateway.sense.embedder import embed
+from gateway.sense.errors import EmbedderDown
+from gateway.sense.features import gate
+
+
+@dataclass
+class TierDecision:
+    tier: str                          # 决策档（live 时即执行档的候选）
+    probs: Dict[str, float]
+    hard_gates: Dict[str, Any]
+    thresholds_version: str = ""
+    head_version: str = ""
+    mode: str = "collect"
+    fallback: bool = False             # True = embedder/工件故障，规则门退化
+    executed_tier: str = ""            # 现行为执行的档位（collect/shadow 记录用）
+
+
+def _rule_tier(feats) -> str:
+    """规则门兜底（无模型时的决策）：仓库级 -> T3；t1 硬门过 -> T1；否则 T2。"""
+    if feats.repo_signals:
+        return "T3"
+    if feats.t1_hard_ok:
+        return "T1"
+    return "T2"
+
+
+class Grader:
+    """决策器（持有 active 工件缓存；工件/阈值切换后调 invalidate）。"""
+
+    def __init__(self, cfg: SenseConfig, store, observer=None):
+        self.cfg = cfg
+        self.store = store
+        self.observer = observer
+        self._head: Optional[LinearHead] = None
+        self._head_loaded = False
+        self._thresholds: Optional[Dict[str, Any]] = None
+
+    def _load_head(self):
+        if not self._head_loaded:
+            art = self.store.active_artifact("head")
+            if art:
+                try:
+                    self._head = LinearHead.load(art["path"], version=art["version"])
+                except Exception:  # noqa: BLE001
+                    self._head = None
+            self._head_loaded = True
+        return self._head
+
+    def invalidate(self) -> None:
+        """工件 promote 后调用（重载 active 工件与阈值）。"""
+        self._head = None
+        self._head_loaded = False
+        self._thresholds = None
+
+    def _thresholds_cached(self) -> Dict[str, Any]:
+        if self._thresholds is None:
+            self._thresholds = load_active(self.store, self.cfg.models_dir)
+        return self._thresholds
+
+    async def decide(self, query_or_messages, consumer: str,
+                     domain: str = "", request_id: str = "",
+                     executed_tier: str = "") -> TierDecision:
+        """分级决策（§8 时序；全模式写观察）。"""
+        ts = time.time()
+        rid = request_id or ("rt" + uuid.uuid4().hex[:10])
+        feats = gate(query_or_messages, consumer, self.cfg)
+        probs: Dict[str, float] = {}
+        head_version = ""
+        fallback = False
+
+        vec: Optional[List[int]] = None
+        try:
+            vec = await embed(feats and (query_or_messages
+                                         if isinstance(query_or_messages, str)
+                                         else "\n".join(
+                                             str(m.get("content") or "")
+                                             for m in query_or_messages)),
+                              self.cfg.embedder)
+        except EmbedderDown:
+            fallback = True
+
+        head = None if fallback else self._load_head()
+        if head is None:
+            fallback = True
+
+        if not fallback and vec is not None:
+            probs = head.predict([float(v) for v in vec])
+            head_version = head.version
+
+        th = self._thresholds_cached()
+        th_version = str(th.get("version") or "")
+
+        # ---- 决策 ----
+        if fallback:
+            tier = _rule_tier(feats)                      # D-G4 规则门退化
+        else:
+            t1_ok = (probs.get("t1", 0.0) >= float(th.get("t1", 0.9))
+                     and feats.t1_hard_ok)
+            t3_ok = (probs.get("t3", 0.0) >= float(th.get("t3", 0.9))
+                     or feats.repo_signals)
+            if t3_ok and not t1_ok:
+                tier = "T3"
+            elif t1_ok:
+                tier = "T1"
+            else:
+                tier = "T2"
+
+        # ---- mode 裁剪（D-G7）----
+        if self.cfg.mode == "live":
+            executed = tier                                # live：决策即执行
+        else:
+            # collect/shadow：现行为——规则门等价（T1 门/T3 信号）近似 v2 现状
+            executed = executed_tier or _rule_tier(feats)
+
+        decision = TierDecision(
+            tier=tier, probs=probs,
+            hard_gates={"turns": feats.turns, "est_tokens": feats.est_tokens,
+                        "single_turn": feats.single_turn,
+                        "intent_blocked": feats.intent_blocked,
+                        "repo_signals": feats.repo_signals,
+                        "over_length": feats.over_length,
+                        "t1_hard_ok": feats.t1_hard_ok},
+            thresholds_version=th_version, head_version=head_version,
+            mode=self.cfg.mode, fallback=fallback,
+            executed_tier=executed)
+
+        # ---- 观察必写（全模式）----
+        if self.observer is not None:
+            self.observer.log(__import__("gateway.sense.observer",
+                                         fromlist=["Observation"]).Observation(
+                request_id=rid, consumer=consumer, decided_tier=tier,
+                executed_tier=executed, probs=probs,
+                policy_version=f"head:{head_version}|th:{th_version}",
+                features=decision.hard_gates, embedding=vec,
+                bucket="default", domain=domain, ts=ts))
+        return decision
diff --git a/gateway/sense/observer.py b/gateway/sense/observer.py
index 5ea9676..a7ee69c 100644
--- a/gateway/sense/observer.py
+++ b/gateway/sense/observer.py
@@ -97,6 +97,10 @@ class Observer:
 
     def _write_rows(self, rows: List[Dict[str, Any]]) -> None:
         for row in rows:
+            emb = row.get("embedding")
+            if isinstance(emb, list):
+                # 量化值以 0..255 无符号表示 -> BLOB
+                row["embedding"] = bytes(bytearray(x & 0xFF for x in emb))
             self.store.insert_observation(row)
 
 
diff --git a/gateway/sense/routes.py b/gateway/sense/routes.py
index 8404e30..f59650a 100644
--- a/gateway/sense/routes.py
+++ b/gateway/sense/routes.py
@@ -37,6 +37,40 @@ def build_sense_routers(cfg: SenseConfig) -> List[APIRouter]:
         return {"enabled": cfg.enabled, "mode": cfg.mode,
                 "embedder": cfg.embedder.base_url}
 
+    @v1.post("/v1/route", tags=["sense"])
+    async def route_v1(request: Request):
+        """分级决策：{query|messages, consumer, domain?} -> TierDecision 视图。
+
+        D-G6：不落 query 原文（观察表只存哈希可关联 id + 特征 + int8 向量）。
+        mode=collect/shadow：tier 为决策值，执行仍走消费方现状；
+        mode=live：消费方按 tier 分流（D-G7）。
+        """
+        from gateway.sense.grader import Grader
+        try:
+            body = await request.json()
+        except Exception:
+            body = {}
+        query = (body or {}).get("query")
+        messages = (body or {}).get("messages")
+        consumer = str((body or {}).get("consumer") or "proxy")
+        domain = str((body or {}).get("domain") or "")
+        payload = messages if messages else (query or "")
+        grader = Grader(cfg, store,
+                        get_observer(store) if cfg.mode != "collect" else None)
+        d = await grader.decide(payload, consumer, domain)
+        probs_total = sum(d.probs.values()) or 1.0
+        return {
+            "tier": d.tier if cfg.mode == "live" else d.tier,
+            "probs": {k: round(v / probs_total, 4) for k, v in d.probs.items()},
+            "confidence": round(max(d.probs.values()) / probs_total, 4)
+            if d.probs else 0.0,
+            "thresholds_version": d.thresholds_version,
+            "head_version": d.head_version,
+            "mode": d.mode,
+            "fallback": d.fallback,
+            "hard_gates": d.hard_gates,
+        }
+
     @v1.post("/v1/embeddings", tags=["sense"])
     async def embeddings(request: Request):
         """OpenAI 兼容透传 embedder（客户端/代理共用）。"""
diff --git a/tests/test_sense_grader.py b/tests/test_sense_grader.py
new file mode 100644
index 0000000..17c338e
--- /dev/null
+++ b/tests/test_sense_grader.py
@@ -0,0 +1,128 @@
+"""Grader/特征门测试（T-G5）：门×概率×mode 决策表 + fallback + /v1/route 契约。"""
+import asyncio
+import json
+
+import pytest
+
+pytest.importorskip("fastapi")
+
+from gateway.sense.config import build_sense_config
+from gateway.sense.errors import EmbedderDown
+from gateway.sense.features import gate
+from gateway.sense.grader import Grader
+from gateway.sense.store import SenseStore
+
+
+@pytest.fixture()
+def env(tmp_path, monkeypatch):
+    cfg = build_sense_config({"sense": {
+        "enabled": True, "mode": "shadow",
+        "db_path": str(tmp_path / "s.sqlite3"),
+        "models_dir": str(tmp_path / "models")}})
+    store = SenseStore.init_db(cfg.db_path)
+    reset = getattr(__import__("gateway.sense.observer",
+                              fromlist=["reset_observer"]), "reset_observer")
+    reset()
+    observer = __import__("gateway.sense.observer",
+                          fromlist=["Observer"]).Observer(store)
+    g = Grader(cfg, store, observer)
+    yield {"cfg": cfg, "store": store, "grader": g, "observer": observer,
+           "tmp": tmp_path}
+    reset()
+
+
+def _embed_ok(monkeypatch, dim=8):
+    import gateway.sense.embedder as em
+    orig = em.embed
+
+    async def fake(text, c):
+        return [1, 2, 3]
+
+    monkeypatch.setattr("gateway.sense.grader.embed", fake)
+    return orig
+
+
+def _head_ok(tmp_path, monkeypatch, p1=0.95, p3=0.02):
+    """注册一个 active head 工件，predict 可控。"""
+    import gateway.sense.grader as gr
+
+    class FakeHead:
+        version = "test-head"
+
+        def __init__(self, p1, p3):
+            self._p = (p1, 1 - p1 - p3, p3)
+
+        def predict(self, vec):
+            p1, p2, p3 = self._p
+            return {"t1": p1, "t2": p2, "t3": p3}
+
+    head = FakeHead(p1, p3)
+    monkeypatch.setattr(Grader, "_load_head",
+                        lambda self: head if not self._head_loaded else None)
+    return head
+
+
+def test_gate_features_matrix(env):
+    """特征门全表：单轮/多轮/长度/黑名单/仓库信号。"""
+    cfg = env["cfg"]
+    short_ok = gate("什么是递归？", "pipeline", cfg)
+    assert short_ok.t1_hard_ok is True and short_ok.turns == 1
+    long_text = "x" * (cfg.t1_max_tokens * 3 + 10)
+    assert gate(long_text, "pipeline", cfg).t1_hard_ok is False
+    multi = [{"role": "user", "content": "a"}, {"role": "assistant", "content": "b"},
+             {"role": "user", "content": "c"}]
+    f2 = gate(multi, "pipeline", cfg)
+    assert f2.turns == 3 and f2.t1_hard_ok is False
+    for word in ("重构", "脚手架", "迁移", "实现", "多文件", "项目"):
+        f = gate(f"帮我{word}这个模块", "pipeline", cfg)
+        assert f.intent_blocked is True and f.t1_hard_ok is False
+    repo = gate("这个仓库要跨模块改造", "pipeline", cfg)
+    assert repo.repo_signals is True and repo.t1_hard_ok is False
+
+
+def test_grader_shadow_does_not_change_flow(env, monkeypatch):
+    """shadow：decided 照算，executed=规则门现行为，观察必写（flush 后落库）。"""
+    _embed_ok(monkeypatch)
+    _head_ok(env["tmp"], monkeypatch, p1=0.95, p3=0.02)
+    d = asyncio_run(env["grader"].decide("什么是递归？", "pipeline"))
+    assert d.tier == "T1" and d.mode == "shadow" and d.fallback is False
+    assert d.executed_tier == "T1"                     # 规则门现行为同为 T1
+    asyncio_run(env["observer"].flush_once())
+    obs_rows = env["store"].all_observations()
+    assert len(obs_rows) == 1
+    assert obs_rows[0]["decided_tier"] == "T1"
+    assert json.loads(obs_rows[0]["features"])["t1_hard_ok"] is True
+
+
+def test_grader_live_tier_is_decision(env, monkeypatch):
+    """live：决策即执行档。"""
+    env["cfg"].mode = "live"
+    _embed_ok(monkeypatch)
+    _head_ok(env["tmp"], monkeypatch, p1=0.3, p3=0.9)
+    d = asyncio_run(env["grader"].decide("多文件大改造任务", "proxy"))
+    assert d.tier == "T3"
+    assert d.mode == "live"
+
+
+def test_grader_fallback_embedder_down(env, monkeypatch):
+    """Embedder 挂 -> fallback=True 规则门退化（D-G4）。"""
+    def boom(text, c):
+        raise EmbedderDown("挂了")
+    monkeypatch.setattr("gateway.sense.grader.embed", boom)
+    d = asyncio_run(env["grader"].decide("什么是递归？", "pipeline"))
+    assert d.fallback is True
+    assert d.tier == "T1"                              # 规则门：短文本可 T1
+    assert d.probs == {}
+
+
+def test_grader_conservative_thresholds(env, monkeypatch):
+    """无工件 -> 保守阈值 0.9：p1=0.5 不判 T1。"""
+    _embed_ok(monkeypatch)
+    _head_ok(env["tmp"], monkeypatch, p1=0.5, p3=0.05)
+    d = asyncio_run(env["grader"].decide("什么是递归？", "pipeline"))
+    assert d.tier == "T2"                              # 置信不足默认 T2
+    assert d.thresholds_version == "conservative"
+
+
+def asyncio_run(coro):
+    return asyncio.run(coro)
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index 337c2c3..f6f741b 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -160,7 +160,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T-G3 | 标签+校准：夜间 true_tier 推导 + split-conformal + 工件表 | ✅ 完成 | T-G3 |
 | 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 | Grader：决策组合（特征门×概率×conformal）+ /v1/route + 三态 mode | ✅ 完成 | T-G5 |
 | T-G6 | live 分流：pipeline tier_fn 三档钩子 + proxy 档位映射 + T2 档升级阶梯 | ⬜ 待办 | |
 | T-G7 | 审计+前端：ReviewQueue 抽样 + tier 指标卡 + 客户端来源显示/一键升级 | ⬜ 待办 | |
 | T-G8 | 实验：E-G1/E-G3 报告；（可选）LoraRemote + E-G2 线性 vs LoRA | ⬜ 待办 | |