diff --git a/gateway/sense/decision_cache.py b/gateway/sense/decision_cache.py
new file mode 100644
index 0000000..7739f47
--- /dev/null
+++ b/gateway/sense/decision_cache.py
@@ -0,0 +1,74 @@
+"""路由决策缓存（T-X3，采纳 cortiq router_client 的决策哈希缓存思路）。
+
+- key = sha256(scope + 0x00 + 归一文本)；LRU + TTL（默认 60s / 4096 条，
+  对齐 auth._AuthCache 的进程内模式，D-P9 单进程前提）。
+- 只缓存**纯决策负载**（如 probs/head_version/vec）；调用方的观察与审计
+  落盘职责不因缓存命中而跳过。
+- 时钟可注入（now(monotonic)），供测试确定性。
+"""
+from __future__ import annotations
+
+import hashlib
+import threading
+import time
+from collections import OrderedDict
+from typing import Any, Callable, Optional
+
+_DEFAULT_TTL_S = 60.0
+_DEFAULT_MAX = 4096
+
+
+class DecisionCache:
+    """进程内决策缓存：LRU + TTL，键为 (scope, 归一文本) 的 sha256。"""
+
+    def __init__(self, maxsize: int = _DEFAULT_MAX, ttl_s: float = _DEFAULT_TTL_S,
+                 now: Optional[Callable[[], float]] = None):
+        self._max = max(1, int(maxsize))
+        self._ttl = float(ttl_s)
+        self._now = now or time.monotonic
+        self._data: "OrderedDict[str, tuple[float, Any]]" = OrderedDict()
+        self._lock = threading.Lock()
+        self.hits = 0
+        self.misses = 0
+
+    @staticmethod
+    def key(scope: str, text: str) -> str:
+        """缓存键：scope 隔离消费方/模式，文本归一由调用方完成。"""
+        return hashlib.sha256(f"{scope}\x00{text}".encode("utf-8")).hexdigest()
+
+    def get(self, scope: str, text: str) -> Optional[Any]:
+        now = self._now()
+        k = self.key(scope, text)
+        with self._lock:
+            item = self._data.get(k)
+            if item is None:
+                self.misses += 1
+                return None
+            ts, payload = item
+            if now - ts > self._ttl:
+                self._data.pop(k, None)
+                self.misses += 1
+                return None
+            self._data.move_to_end(k)
+            self.hits += 1
+            return payload
+
+    def put(self, scope: str, text: str, payload: Any) -> None:
+        k = self.key(scope, text)
+        with self._lock:
+            self._data[k] = (self._now(), payload)
+            self._data.move_to_end(k)
+            while len(self._data) > self._max:
+                self._data.popitem(last=False)
+
+    def clear(self) -> None:
+        """清空条目（阈值/工件版本切换时由 Grader.invalidate 调用）。"""
+        with self._lock:
+            self._data.clear()
+
+    def stats(self) -> dict:
+        with self._lock:
+            return {"entries": len(self._data), "hits": self.hits,
+                    "misses": self.misses,
+                    "hit_rate": round(self.hits / (self.hits + self.misses), 4)
+                    if (self.hits + self.misses) else 0.0}
diff --git a/gateway/sense/grader.py b/gateway/sense/grader.py
index aeffd4a..f635c3d 100644
--- a/gateway/sense/grader.py
+++ b/gateway/sense/grader.py
@@ -17,6 +17,7 @@ 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.decision_cache import DecisionCache
 from gateway.sense.embedder import embed
 from gateway.sense.errors import EmbedderDown
 from gateway.sense.features import gate
@@ -53,6 +54,7 @@ class Grader:
         self._head: Optional[LinearHead] = None
         self._head_loaded = False
         self._thresholds: Optional[Dict[str, Any]] = None
+        self._dcache = DecisionCache()      # T-X3：live 模式决策缓存
 
     def _load_head(self):
         if not self._head_loaded:
@@ -66,10 +68,11 @@ class Grader:
         return self._head
 
     def invalidate(self) -> None:
-        """工件 promote 后调用（重载 active 工件与阈值）。"""
+        """工件 promote 后调用（重载 active 工件与阈值；同步清空决策缓存）。"""
         self._head = None
         self._head_loaded = False
         self._thresholds = None
+        self._dcache.clear()
 
     def _thresholds_cached(self) -> Dict[str, Any]:
         if self._thresholds is None:
@@ -79,32 +82,47 @@ class Grader:
     async def decide(self, query_or_messages, consumer: str,
                      domain: str = "", request_id: str = "",
                      executed_tier: str = "") -> TierDecision:
-        """分级决策（§8 时序；全模式写观察）。"""
+        """分级决策（§8 时序；全模式写观察）。
+
+        T-X3：live 模式下对 (consumer, 文本) 的纯决策负载（probs/head_version/vec）
+        做 60s LRU 缓存，命中时跳过 embed + 线性头；观察/审计照常落盘。
+        collect/shadow 模式不走缓存（校准数据必须全量产出）。
+        """
         ts = time.time()
         rid = request_id or ("rt" + uuid.uuid4().hex[:10])
+        text = (query_or_messages if isinstance(query_or_messages, str)
+                else "\n".join(str(m.get("content") or "")
+                               for m in query_or_messages))
         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
+        cached = self._dcache.get(consumer, text) \
+            if self.cfg.mode == "live" else None
+        if cached is not None:
+            probs = dict(cached["probs"])
+            head_version = str(cached["head_version"])
+            vec = cached["vec"]
+        else:
+            try:
+                vec = await embed(text, self.cfg.embedder)
+            except EmbedderDown:
+                fallback = True
 
-        head = None if fallback else self._load_head()
-        if head is None:
-            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
+            if not fallback and vec is not None:
+                probs = head.predict([float(v) for v in vec])
+                head_version = head.version
+                if self.cfg.mode == "live":
+                    self._dcache.put(consumer, text,
+                                     {"probs": dict(probs),
+                                      "head_version": head_version,
+                                      "vec": vec})
 
         th = self._thresholds_cached()
         th_version = str(th.get("version") or "")
diff --git a/tests/test_sense_decision_cache.py b/tests/test_sense_decision_cache.py
new file mode 100644
index 0000000..56025fb
--- /dev/null
+++ b/tests/test_sense_decision_cache.py
@@ -0,0 +1,157 @@
+"""决策缓存测试（T-X3）：LRU+TTL 单元契约 + Grader live 模式接线（embed 去重）。"""
+import asyncio
+
+import pytest
+
+from gateway.sense.config import build_sense_config
+from gateway.sense.decision_cache import DecisionCache
+from gateway.sense.grader import Grader
+from gateway.sense.store import SenseStore
+
+
+# ---------------- DecisionCache 单元 ----------------
+
+class _Clock:
+    def __init__(self):
+        self.t = 100.0
+
+    def __call__(self):
+        return self.t
+
+
+def test_key_scope_and_text_sensitive():
+    assert DecisionCache.key("proxy", "abc") == DecisionCache.key("proxy", "abc")
+    assert DecisionCache.key("proxy", "abc") != DecisionCache.key("pipeline", "abc")
+    assert DecisionCache.key("proxy", "abc") != DecisionCache.key("proxy", "abd")
+
+
+def test_put_get_hit_miss_counters():
+    c = DecisionCache()
+    assert c.get("s", "q") is None
+    c.put("s", "q", {"tier": "T1"})
+    assert c.get("s", "q") == {"tier": "T1"}
+    st = c.stats()
+    assert st["hits"] == 1 and st["misses"] == 1 and st["entries"] == 1
+
+
+def test_ttl_expiry_with_fake_clock():
+    clock = _Clock()
+    c = DecisionCache(ttl_s=60.0, now=clock)
+    c.put("s", "q", {"v": 1})
+    assert c.get("s", "q") == {"v": 1}
+    clock.t += 61.0                    # 超时
+    assert c.get("s", "q") is None
+    assert c.stats()["entries"] == 0   # 过期条目被清除
+
+
+def test_lru_eviction_at_maxsize():
+    c = DecisionCache(maxsize=2)
+    c.put("s", "a", 1)
+    c.put("s", "b", 2)
+    c.get("s", "a")                    # a 变为最近使用
+    c.put("s", "c", 3)                 # 淘汰 b
+    assert c.get("s", "a") == 1
+    assert c.get("s", "b") is None
+    assert c.get("s", "c") == 3
+
+
+def test_clear_keeps_counters_but_drops_entries():
+    c = DecisionCache()
+    c.put("s", "q", 1)
+    c.get("s", "q")
+    c.clear()
+    assert c.get("s", "q") is None
+    assert c.stats()["hits"] == 1      # 计数保留（审计口径不回退）
+
+
+# ---------------- Grader live 模式接线 ----------------
+
+class _CountingObserver:
+    def __init__(self):
+        self.logs = []
+
+    def log(self, obs):
+        self.logs.append(obs)
+
+
+class _FakeHead:
+    version = "test-head"
+
+    def predict(self, vec):
+        return {"t1": 0.95, "t2": 0.03, "t3": 0.02}
+
+
+def _make_grader(tmp_path, mode, counter, observer=None):
+    """构造测试 Grader：embed 打点计数 + FakeHead 注入。"""
+    import gateway.sense.grader as gr
+
+    cfg = build_sense_config({"sense": {
+        "enabled": True, "mode": mode,
+        "db_path": str(tmp_path / "s.sqlite3"),
+        "models_dir": str(tmp_path / "models")}})
+    store = SenseStore.init_db(cfg.db_path)
+    g = Grader(cfg, store, observer or _CountingObserver())
+    g._load_head = lambda: _FakeHead()
+
+    async def fake_embed(text, c):
+        counter["n"] += 1
+        return [1, 2, 3]
+
+    orig = gr.embed
+    gr.embed = fake_embed
+    return g, orig
+
+
+def test_grader_live_mode_dedupes_embed(tmp_path):
+    """live 模式：同文本第二次 decide 命中缓存，embed 只跑一次；观察仍写两条。"""
+    import gateway.sense.grader as gr
+    counter = {"n": 0}
+    observer = _CountingObserver()
+    g, orig = _make_grader(tmp_path, "live", counter, observer)
+    try:
+        async def run():
+            d1 = await g.decide("什么是递归", "proxy")
+            d2 = await g.decide("什么是递归", "proxy")
+            d3 = await g.decide("换一个问题", "proxy")
+            return d1, d2, d3
+
+        d1, d2, d3 = asyncio.run(run())
+    finally:
+        gr.embed = orig
+    assert d1.tier == "T1" and d2.tier == "T1" and d3.tier == "T1"
+    assert counter["n"] == 2                       # 同文本去重：2 个不同文本各 1 次
+    assert d2.head_version == d1.head_version
+    assert len(observer.logs) == 3                 # 观察不因缓存命中而跳过
+
+
+def test_grader_collect_mode_no_cache(tmp_path):
+    """collect/shadow：校准数据必须全量产出，不走缓存。"""
+    import gateway.sense.grader as gr
+    counter = {"n": 0}
+    g, orig = _make_grader(tmp_path, "collect", counter)
+    try:
+        async def run():
+            await g.decide("同一个问题", "proxy")
+            await g.decide("同一个问题", "proxy")
+
+        asyncio.run(run())
+    finally:
+        gr.embed = orig
+    assert counter["n"] == 2
+
+
+def test_grader_invalidate_clears_cache(tmp_path):
+    """invalidate（工件切换）后缓存清空：同文本重新 embed。"""
+    import gateway.sense.grader as gr
+    counter = {"n": 0}
+    g, orig = _make_grader(tmp_path, "live", counter)
+    try:
+        async def run():
+            await g.decide("什么是递归", "proxy")
+            g.invalidate()
+            await g.decide("什么是递归", "proxy")
+
+        asyncio.run(run())
+    finally:
+        gr.embed = orig
+    assert counter["n"] == 2
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index ef666a6..8418616 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -167,3 +167,4 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | OPT-1 | 分支推进：语义缓存 L2 查找 3.39x（免并集计分+预筛）+ 安全加固（15 高危清零：SSRF/路径穿越/假凭据） | ✅ 完成 | ad3bf41 |
 | T-X1 | 预算四档渐进干预（外部采纳 ai-model-router）：budget_mode 整数基点判定（80/95/100%）+ optimize/cheap 自动降档 + X-Budget-Mode 上报；黄金用例锁边界 | ✅ 完成 | T-X1 |
 | T-X2 | 上游有序降级链（外部采纳 cortiq tier 链）：池内候选按档位/单价排序、首 token 前 failover、X-Upstream-Fallback 三元组响应头 + admin/stats failover 聚合 | ✅ 完成 | T-X2 |
+| T-X3 | 路由决策缓存（外部采纳 cortiq 决策哈希缓存）：DecisionCache（sha256+LRU+TTL60s，4096 条）前置 grader live 模式，embed+线性头去重；观察/审计不跳过、invalidate 同步清空、collect/shadow 不缓存 | ✅ 完成 | T-X3 |