diff --git a/gateway/api.py b/gateway/api.py
index 502f0ef..d5b5151 100644
--- a/gateway/api.py
+++ b/gateway/api.py
@@ -214,6 +214,16 @@ try:
     except Exception as _pe:  # pragma: no cover - 代理层装配失败不拖垮主应用
         print(f"[gateway] 代理层未启用（{_pe}）")
 
+    # 语义分析器（T-G0；D-G7：sense.enabled=False 时不注册任何路由）
+    try:
+        from gateway.sense import build_sense_router
+        from gateway.sense.config import build_sense_config
+        _sense_cfg = build_sense_config(settings_store().to_dict())
+        if _sense_cfg.enabled:
+            app.include_router(build_sense_router(_sense_cfg))
+    except Exception as _se:  # pragma: no cover
+        print(f"[gateway] 语义分析器未启用（{_se}）")
+
     @app.get("/", response_class=HTMLResponse, tags=["ui"])
     async def index():
         """Vue SPA 的 index.html（FastAPI API 路由优先，此处仅作 fallback）。"""
diff --git a/gateway/sense/__init__.py b/gateway/sense/__init__.py
new file mode 100644
index 0000000..c3185e2
--- /dev/null
+++ b/gateway/sense/__init__.py
@@ -0,0 +1,17 @@
+"""语义分析器与三级任务分级（T-G0…T-G8）：/v1/route + /sense/admin/*。
+
+组装唯一入口：build_sense_router(cfg) -> APIRouter；
+由 gateway.api 在 sense.enabled 时 include（D-G7 关闭时不注册任何路由）。
+灰度：mode=collect（只记不决策）-> shadow（决策只比对）-> live（真分流）；
+collect 攒满 min_labels 前不得开 live（晋升门 §6）。
+"""
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+from gateway.sense.config import SenseConfig, build_sense_config  # noqa: F401
+
+
+def build_sense_router(cfg: SenseConfig) -> APIRouter:
+    from gateway.sense.routes import build_sense_router as _build
+    return _build(cfg)
diff --git a/gateway/sense/config.py b/gateway/sense/config.py
new file mode 100644
index 0000000..7630995
--- /dev/null
+++ b/gateway/sense/config.py
@@ -0,0 +1,90 @@
+"""SenseConfig：从 settings 的 sense 段构建语义分析器配置（T-G0）。
+
+灰度纪律（D-G7）：
+- enabled 默认 False（总开关，关闭时不注册任何 /sense 或 /v1/route 路由）；
+- mode 默认 "collect"（只记不决策）；collect 攒满 min_labels 前禁止 live
+  （晋升校验在 grader/admin，配置层不阻止写值——但 admin promote 会复核）。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict
+
+MODES = ("collect", "shadow", "live")
+
+
+@dataclass
+class EmbedderCfg:
+    base_url: str = "http://127.0.0.1:8902/v1"
+    model: str = "bge-m3-Q4_K_M"
+    timeout_s: float = 5.0
+    dim: int = 1024
+
+
+@dataclass
+class SenseConfig:
+    enabled: bool = False
+    mode: str = "collect"                    # collect | shadow | live
+    db_path: str = "data/sense.sqlite3"
+    admin_key: str = ""
+    embedder: EmbedderCfg = field(default_factory=EmbedderCfg)
+    t1_max_tokens: int = 512
+    t1_max_turns: int = 2
+    intent_blacklist: tuple = ("重构", "脚手架", "迁移", "实现", "多文件", "项目")
+    code_t1_kinds: tuple = ("解释", "补全")
+    alpha: float = 0.05
+    min_labels: int = 500
+    t2_prefer_local_when_idle: bool = True
+    consumers: Dict[str, Dict[str, str]] = field(default_factory=lambda: {
+        "proxy": {"t1": "local-small", "t2": "budget", "t3": "premium"}})
+    models_dir: str = "data/sense_models"
+    retention_days: int = 180
+
+    def tier_pool_hint(self, consumer: str, tier: str) -> str:
+        """消费方×档位 -> model_pool 档位名（§5 consumers；缺省兜底 local/budget/premium）。"""
+        m = self.consumers.get(consumer) or {}
+        return m.get(tier) or {"t1": "local", "t2": "budget",
+                               "t3": "premium"}.get(tier, "budget")
+
+
+def build_sense_config(settings_dict: Dict[str, Any]) -> SenseConfig:
+    """从 settings.to_dict() 的 sense 段构建（缺省兜底；mode 非法回落 collect）。"""
+    raw = settings_dict.get("sense") or {}
+    if not isinstance(raw, dict):
+        raw = {}
+    emb = raw.get("embedder") or {}
+    if not isinstance(emb, dict):
+        emb = {}
+    feats = raw.get("features") or {}
+    if not isinstance(feats, dict):
+        feats = {}
+    policy = raw.get("policy") or {}
+    if not isinstance(policy, dict):
+        policy = {}
+    consumers = raw.get("consumers") or {}
+    if not isinstance(consumers, dict):
+        consumers = {}
+    mode = str(raw.get("mode", "collect") or "collect")
+    if mode not in MODES:
+        mode = "collect"
+    return SenseConfig(
+        enabled=bool(raw.get("enabled", False)),
+        mode=mode,
+        db_path=str(raw.get("db_path") or "data/sense.sqlite3"),
+        admin_key=str(raw.get("admin_key") or ""),
+        embedder=EmbedderCfg(
+            base_url=str(emb.get("base_url") or "http://127.0.0.1:8902/v1"),
+            model=str(emb.get("model") or "bge-m3-Q4_K_M"),
+            timeout_s=float(emb.get("timeout_s", 5) or 5),
+            dim=int(emb.get("dim", 1024) or 1024),
+        ),
+        t1_max_tokens=int(feats.get("t1_max_tokens", 512) or 512),
+        t1_max_turns=int(feats.get("t1_max_turns", 2) or 2),
+        intent_blacklist=tuple(feats.get("intent_blacklist")
+                               or ("重构", "脚手架", "迁移", "实现", "多文件", "项目")),
+        code_t1_kinds=tuple(feats.get("code_t1_kinds") or ("解释", "补全")),
+        alpha=float(policy.get("alpha", 0.05) or 0.05),
+        min_labels=int(policy.get("min_labels", 500) or 500),
+        t2_prefer_local_when_idle=bool(policy.get("t2_prefer_local_when_idle", True)),
+        consumers=consumers or SenseConfig().consumers,
+    )
diff --git a/gateway/sense/errors.py b/gateway/sense/errors.py
new file mode 100644
index 0000000..b2fcf18
--- /dev/null
+++ b/gateway/sense/errors.py
@@ -0,0 +1,19 @@
+"""Sense 错误类型。"""
+from __future__ import annotations
+
+
+class SenseError(Exception):
+    status_code = 500
+    code = "sense_error"
+
+
+class EmbedderDown(SenseError):
+    """Embedder 不可用（D-G4：决策退化为规则门 + 默认 T2）。"""
+    status_code = 503
+    code = "embedder_down"
+
+
+class ArtifactMissing(SenseError):
+    """分类器工件缺失（同上降级）。"""
+    status_code = 404
+    code = "artifact_missing"
diff --git a/gateway/sense/routes.py b/gateway/sense/routes.py
new file mode 100644
index 0000000..d6da685
--- /dev/null
+++ b/gateway/sense/routes.py
@@ -0,0 +1,21 @@
+"""Sense 路由（T-G0 骨架：健康检查；/v1/route 与 /v1/embeddings 在 T-G1/G5 落地）。"""
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+from gateway.sense.config import SenseConfig
+
+
+def build_sense_router(cfg: SenseConfig) -> APIRouter:
+    """组装 sense 面路由（唯一组装点）。"""
+    router = APIRouter(prefix="/sense")
+
+    @router.get("/health", tags=["sense"])
+    async def health():
+        """sense 面健康检查（含灰度状态，供看板/运维）。"""
+        return {"enabled": cfg.enabled, "mode": cfg.mode,
+                "embedder": cfg.embedder.base_url}
+
+    # /v1/route（T-G5 grader）、/v1/embeddings（T-G1）、/sense/admin/*（T-G3+）后续追加。
+    # 注意 /v1/route 不带 /sense 前缀——由 api.py 单独 include 一个无前缀子路由（T-G5）。
+    return router
diff --git a/gateway/sense/store.py b/gateway/sense/store.py
new file mode 100644
index 0000000..107a163
--- /dev/null
+++ b/gateway/sense/store.py
@@ -0,0 +1,174 @@
+"""Sense 存储（T-G0：DDL + 基础查询；labeler/grader 的查询在后续任务落地）。
+
+§4 数据模型：tier_observations（观察-标签闭环主表）+ sense_artifacts（工件登记）。
+WAL；全部访问经全局锁 + 每操作新连接（ReviewQueue 模式），异步调用方 to_thread。
+隐私（D-G6）：不落 query 原文（只存哈希+特征+int8 embedding BLOB）；留存 180 天。
+"""
+from __future__ import annotations
+
+import sqlite3
+import threading
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS tier_observations(
+  id INTEGER PRIMARY KEY, ts INTEGER NOT NULL,
+  request_id TEXT NOT NULL, consumer TEXT NOT NULL,
+  bucket TEXT NOT NULL DEFAULT 'default', domain TEXT DEFAULT '',
+  decided_tier TEXT NOT NULL, executed_tier TEXT NOT NULL,
+  probs TEXT NOT NULL,
+  policy_version TEXT NOT NULL, features TEXT NOT NULL,
+  embedding BLOB,
+  outcome TEXT DEFAULT '', true_tier TEXT DEFAULT '', human_override TEXT DEFAULT '');
+CREATE INDEX IF NOT EXISTS idx_obs_ts ON tier_observations(ts);
+CREATE INDEX IF NOT EXISTS idx_obs_policy ON tier_observations(policy_version);
+
+CREATE TABLE IF NOT EXISTS sense_artifacts(
+  version TEXT PRIMARY KEY, kind TEXT NOT NULL,
+  path TEXT NOT NULL, metrics TEXT NOT NULL, created_ts INTEGER NOT NULL,
+  active INTEGER DEFAULT 0);
+"""
+
+
+class SenseStore:
+    """sense.sqlite3 访问（同步实现；异步调用方 to_thread）。"""
+
+    def __init__(self, db_path: str | Path):
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+        self._init_db()
+
+    @classmethod
+    def init_db(cls, db_path: str | Path) -> "SenseStore":
+        return cls(db_path)
+
+    def _connect(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _init_db(self) -> None:
+        with self._lock, self._connect() as conn:
+            conn.executescript(_SCHEMA)
+
+    # ---------- 观察 ----------
+    def insert_observation(self, obs: Dict[str, Any]) -> None:
+        """写入一条观察（T-G2 批量缓冲的落盘终点）。"""
+        with self._lock, self._connect() as conn:
+            conn.execute(
+                """INSERT OR IGNORE INTO tier_observations
+                (ts, request_id, consumer, bucket, domain, decided_tier,
+                 executed_tier, probs, policy_version, features, embedding,
+                 outcome, true_tier, human_override)
+                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+                (int(obs.get("ts", time.time())), obs["request_id"],
+                 obs["consumer"], obs.get("bucket", "default"),
+                 obs.get("domain", ""), obs["decided_tier"], obs["executed_tier"],
+                 obs.get("probs", "{}"), obs.get("policy_version", ""),
+                 obs.get("features", "{}"), obs.get("embedding"),
+                 obs.get("outcome", ""), obs.get("true_tier", ""),
+                 obs.get("human_override", "")))
+
+    def update_outcome(self, request_id: str, outcome: str,
+                       executed_tier: Optional[str] = None) -> bool:
+        """升级阶梯回写（outcome=escalated 等）。"""
+        with self._lock, self._connect() as conn:
+            if executed_tier:
+                cur = conn.execute(
+                    "UPDATE tier_observations SET outcome = ?, executed_tier = ?"
+                    " WHERE request_id = ?", (outcome, executed_tier, request_id))
+            else:
+                cur = conn.execute(
+                    "UPDATE tier_observations SET outcome = ? WHERE request_id = ?",
+                    (outcome, request_id))
+            return cur.rowcount > 0
+
+    def set_true_tier(self, request_id: str, true_tier: str) -> bool:
+        """labeler 回填（T-G3）。"""
+        with self._lock, self._connect() as conn:
+            cur = conn.execute(
+                "UPDATE tier_observations SET true_tier = ? WHERE request_id = ?",
+                (true_tier, request_id))
+            return cur.rowcount > 0
+
+    def labeled_rows(self, policy_version: str = "", limit: int = 100000) -> List[Dict[str, Any]]:
+        """true_tier 非空的行（训练/校准输入）。"""
+        with self._lock, self._connect() as conn:
+            if policy_version:
+                rows = conn.execute(
+                    "SELECT * FROM tier_observations WHERE true_tier != ''"
+                    " AND policy_version = ? ORDER BY ts LIMIT ?",
+                    (policy_version, limit)).fetchall()
+            else:
+                rows = conn.execute(
+                    "SELECT * FROM tier_observations WHERE true_tier != ''"
+                    " ORDER BY ts LIMIT ?", (limit,)).fetchall()
+        return [dict(r) for r in rows]
+
+    def count_labeled(self) -> int:
+        """已标签条数（min_labels 晋升门检查）。"""
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                "SELECT COUNT(*) AS c FROM tier_observations"
+                " WHERE true_tier != ''").fetchone()
+        return int(row["c"])
+
+    def purge_older_than(self, ts: float) -> int:
+        """留存清理（D-G6：默认 180 天，夜间任务顺带）。"""
+        with self._lock, self._connect() as conn:
+            cur = conn.execute(
+                "DELETE FROM tier_observations WHERE ts < ?", (int(ts),))
+            return cur.rowcount
+
+    # ---------- 工件 ----------
+    def register_artifact(self, version: str, kind: str, path: str,
+                          metrics: Dict[str, Any], active: bool = False,
+                          created_ts: Optional[int] = None) -> None:
+        """登记模型/阈值工件（默认 active=0，人工 promote 切换）。"""
+        with self._lock, self._connect() as conn:
+            conn.execute(
+                "INSERT OR REPLACE INTO sense_artifacts"
+                "(version, kind, path, metrics, created_ts, active)"
+                " VALUES (?,?,?,?,?,?)",
+                (version, kind, path, json_dumps(metrics),
+                 int(created_ts if created_ts is not None else time.time()),
+                 1 if active else 0))
+
+    def activate_artifact(self, version: str, kind: str) -> bool:
+        """切换 active（同 kind 互斥）。"""
+        with self._lock, self._connect() as conn:
+            conn.execute("UPDATE sense_artifacts SET active = 0 WHERE kind = ?",
+                         (kind,))
+            cur = conn.execute(
+                "UPDATE sense_artifacts SET active = 1 WHERE version = ? AND kind = ?",
+                (version, kind))
+            return cur.rowcount > 0
+
+    def active_artifact(self, kind: str) -> Optional[Dict[str, Any]]:
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM sense_artifacts WHERE kind = ? AND active = 1"
+                " ORDER BY created_ts DESC LIMIT 1", (kind,)).fetchone()
+        return dict(row) if row else None
+
+    # ---------- 自省 ----------
+    def table_names(self) -> List[str]:
+        with self._lock, self._connect() as conn:
+            rows = conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'").fetchall()
+        return [r["name"] for r in rows]
+
+    def index_names(self) -> List[str]:
+        with self._lock, self._connect() as conn:
+            rows = conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='index'"
+                " AND name LIKE 'idx_%'").fetchall()
+        return [r["name"] for r in rows]
+
+
+def json_dumps(obj: Any) -> str:
+    import json
+    return json.dumps(obj, ensure_ascii=False)
diff --git a/gateway/settings.py b/gateway/settings.py
index cd177e9..3baed67 100644
--- a/gateway/settings.py
+++ b/gateway/settings.py
@@ -68,6 +68,21 @@ DEFAULTS: Dict[str, Any] = {
         "semcache": {"enabled": True, "sim_threshold": 0.92,
                      "max_entries": 300000, "promote_frequency": 5},
     },
+    # 语义分析器与三级分级（T-G0；结构见《实施方案_语义分析器与三级分级.md》§5）
+    "sense": {
+        "enabled": False,                    # D-G7 总开关：默认关
+        "mode": "collect",                   # collect | shadow | live（collect 攒满标签前禁 live）
+        "db_path": "data/sense.sqlite3",
+        "admin_key": "",
+        "embedder": {"base_url": "http://127.0.0.1:8902/v1",
+                     "model": "bge-m3-Q4_K_M", "timeout_s": 5, "dim": 1024},
+        "features": {"t1_max_tokens": 512, "t1_max_turns": 2,
+                     "intent_blacklist": ["重构", "脚手架", "迁移", "实现", "多文件", "项目"],
+                     "code_t1_kinds": ["解释", "补全"]},
+        "policy": {"alpha": 0.05, "min_labels": 500,
+                   "t2_prefer_local_when_idle": True},
+        "consumers": {"proxy": {"t1": "local-small", "t2": "budget", "t3": "premium"}},
+    },
 }
 
 
diff --git a/tests/test_sense_routes.py b/tests/test_sense_routes.py
new file mode 100644
index 0000000..299c8a5
--- /dev/null
+++ b/tests/test_sense_routes.py
@@ -0,0 +1,79 @@
+"""语义分析器骨架测试（T-G0）：enabled 门控 / DDL / 独立挂载 / 配置构建。"""
+import pytest
+
+pytest.importorskip("fastapi")
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import gateway.api as ga
+from gateway.sense.config import build_sense_config
+from gateway.sense.store import SenseStore
+
+
+def test_sense_disabled_by_default_404():
+    """D-G7：默认 enabled=False -> 全局 app 不注册任何 /sense 路由。"""
+    client = TestClient(ga.app)
+    assert client.get("/sense/health").status_code == 404
+    assert client.get("/sense/admin/agreement").status_code == 404
+
+
+def test_ddl_creates_tables_and_indexes(tmp_path):
+    """两表 + 两索引幂等创建。"""
+    store = SenseStore.init_db(tmp_path / "sense.sqlite3")
+    tables = set(store.table_names())
+    assert {"tier_observations", "sense_artifacts"} <= tables
+    idx = set(store.index_names())
+    assert {"idx_obs_ts", "idx_obs_policy"} <= idx
+    SenseStore.init_db(tmp_path / "sense.sqlite3")     # 幂等
+
+
+def test_build_sense_config_defaults():
+    """缺省值 + mode 非法回落 collect（D-G7）。"""
+    cfg = build_sense_config({})
+    assert cfg.enabled is False
+    assert cfg.mode == "collect"
+    assert cfg.embedder.dim == 1024
+    assert cfg.min_labels == 500
+    assert cfg.tier_pool_hint("proxy", "t1") == "local-small"
+    assert cfg.tier_pool_hint("proxy", "t3") == "premium"
+    assert cfg.tier_pool_hint("client", "t2") == "budget"   # 无配置兜底
+    bad = build_sense_config({"sense": {"mode": "turbo", "enabled": True}})
+    assert bad.mode == "collect"
+
+
+def test_enabled_router_health(tmp_path):
+    """enabled=True 独立挂载：/sense/health 透出灰度状态。"""
+    cfg = build_sense_config({"sense": {"enabled": True, "mode": "collect",
+                                        "db_path": str(tmp_path / "s.sqlite3")}})
+    app = FastAPI()
+    app.include_router(
+        __import__("gateway.sense", fromlist=["x"]).build_sense_router(cfg))
+    client = TestClient(app)
+    r = client.get("/sense/health")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["enabled"] is True and data["mode"] == "collect"
+
+
+def test_observation_roundtrip(tmp_path):
+    """观察写入/标签回填/计数/留存清理（store 基础链路）。"""
+    store = SenseStore.init_db(tmp_path / "s.sqlite3")
+    store.insert_observation({
+        "request_id": "r1", "consumer": "pipeline", "decided_tier": "T2",
+        "executed_tier": "T2", "probs": '{"t1":0.2,"t2":0.6,"t3":0.2}',
+        "policy_version": "v0-rule", "features": '{"turns":1}'})
+    store.insert_observation({
+        "request_id": "r2", "consumer": "proxy", "decided_tier": "T1",
+        "executed_tier": "T1", "probs": "{}", "policy_version": "v0-rule",
+        "features": "{}", "outcome": "ok"})
+    assert store.count_labeled() == 0
+    store.set_true_tier("r1", "T2")
+    assert store.count_labeled() == 1
+    rows = store.labeled_rows()
+    assert rows[0]["request_id"] == "r1" and rows[0]["true_tier"] == "T2"
+    assert store.update_outcome("r2", "escalated", executed_tier="T2") is True
+    # 180d 清理
+    import time
+    assert store.purge_older_than(time.time() + 10) >= 2
+    assert store.count_labeled() == 0
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index 5a1ba60..3ac4842 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -158,6 +158,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T-G1 | Embedder：/v1/embeddings 客户端 + int8 量化 + 降级阶梯 | ⬜ 待办 | |
 | T-G2 | 观察埋点：observer + 三消费方埋点（pipeline/proxy/client） | ⬜ 待办 | |
 | T-G3 | 标签+校准：夜间 true_tier 推导 + split-conformal + 工件表 | ⬜ 待办 | |
+| T-G3b | KnnHead（架构变体 B）：kNN 投票 + conformal-kNN + 按桶分区/封顶/压缩；hybrid fusion 预留（§14） | ⬜ 待办 | |
 | T-G4 | 线性头：离线训练脚本 + LinearHead 纯 Python 推理 + 登记 | ⬜ 待办 | |
 | T-G5 | Grader：决策组合（特征门×概率×conformal）+ /v1/route + 三态 mode | ⬜ 待办 | |
 | T-G6 | live 分流：pipeline tier_fn 三档钩子 + proxy 档位映射 + T2 档升级阶梯 | ⬜ 待办 | |