diff --git a/tests/test_capability_discovery.py b/tests/test_capability_discovery.py
new file mode 100644
index 0000000..400ab25
--- /dev/null
+++ b/tests/test_capability_discovery.py
@@ -0,0 +1,381 @@
+"""P1-1: 单测 — capability_discovery 核心函数(双 Claw 共享层)。
+
+为什么这个文件存在:
+    历史上 #866 部署因为端侧重写了发现逻辑,把 #764/#765 的注入改动覆盖回退。
+    这次把 discovery 层的关键纯函数 + 三级降级路径锁死为单测,
+    防止「第二个实现」再悄悄出现。
+
+验证范围:
+    1. rank_by_query: 池子 ≤ top_k 原样返回, 池子 > top_k 按词元重叠排序
+    2. rank_by_query: 中文友好(整词 + 2-gram)
+    3. _env_int / _env_flag: 缺失 / 异常 / 正常 路径
+    4. _safe_name: 命名安全(消除 mcp__tool__ 里被注入特殊字符)
+    5. discover_skills: 三级降级(installed / published / empty) + 异常降级
+    6. discover_agents: explicit 优先 + 三级降级
+    7. discover_mcp_servers: explicit 走 resolve_active, auto 走 list_enabled
+    8. 双 Claw 标签注入(log 里有 claw=web/im)
+"""
+from __future__ import annotations
+
+import os
+import sys
+import asyncio
+import logging
+import importlib
+
+import pytest
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(_HERE, ".."))
+
+from app.core import capability_discovery as cd  # noqa: E402
+from app.core.capability_discovery import (  # noqa: E402
+    rank_by_query,
+    _env_int,
+    _env_flag,
+    _safe_name,
+    _tokens,
+    discover_skills,
+    discover_agents,
+    discover_mcp_servers,
+    build_mcp_tool_defs,
+)
+
+
+# =====================================================================
+# 1. rank_by_query: 纯函数
+# =====================================================================
+
+def test_rank_by_query_empty():
+    assert rank_by_query([], "query", 5) == []
+
+
+def test_rank_by_query_below_limit_returns_intact():
+    items = [
+        {"name": "search", "description": "网页搜索"},
+        {"name": "calc", "description": "数学计算"},
+    ]
+    out = rank_by_query(items, "anything", 5)
+    assert out == items  # top_k 大于池子,原样返回
+
+
+def test_rank_by_query_above_limit_sorts_by_overlap():
+    items = [
+        {"name": "weather", "description": "查天气"},
+        {"name": "translate", "description": "翻译文档"},
+        {"name": "weather_alert", "description": "天气预警"},
+        {"name": "ocr", "description": "图片文字识别"},
+    ]
+    out = rank_by_query(items, "天气", top_k=2)
+    # "天气" 在 weather/weather_alert 描述里都出现,应排前
+    assert len(out) == 2
+    assert all("weather" in (x.get("name") or "") or "天气" in (x.get("description") or "") for x in out)
+
+
+def test_rank_by_query_zh_2gram_friendly():
+    items = [
+        {"name": "code_review", "description": "代码审查"},
+        {"name": "music", "description": "音乐生成"},
+    ]
+    out = rank_by_query(items, "代码审查", top_k=1)
+    assert out[0]["name"] == "code_review"
+
+
+def test_rank_by_query_top_k_zero_returns_empty():
+    assert rank_by_query([{"a": 1}], "x", 0) == []
+
+
+def test_tokens_chinese_split():
+    toks = _tokens("代码审查 Code Review")
+    assert "code" in toks
+    assert "review" in toks
+    assert "代码" in toks
+    assert "码审" in toks  # 2-gram
+
+
+# =====================================================================
+# 2. 环境旋钮
+# =====================================================================
+
+def test_env_int_default_when_missing(monkeypatch):
+    monkeypatch.delenv("CAP_SKILL_LIMIT", raising=False)
+    assert _env_int("CAP_SKILL_LIMIT", 12) == 12
+
+
+def test_env_int_uses_value(monkeypatch):
+    monkeypatch.setenv("CAP_SKILL_LIMIT", "20")
+    assert _env_int("CAP_SKILL_LIMIT", 12) == 20
+
+
+def test_env_int_handles_garbage(monkeypatch):
+    monkeypatch.setenv("CAP_SKILL_LIMIT", "not-a-number")
+    assert _env_int("CAP_SKILL_LIMIT", 12) == 12
+
+
+def test_env_flag_default_true():
+    assert _env_flag("__NONEXIST__") is True
+
+
+def test_env_flag_default_false():
+    assert _env_flag("__NONEXIST__", default=False) is False
+
+
+def test_env_flag_truthy(monkeypatch):
+    for v in ("1", "true", "yes", "on", "TRUE"):
+        monkeypatch.setenv("__TEST_FLAG__", v)
+        assert _env_flag("__TEST_FLAG__") is True
+
+
+def test_env_flag_falsy(monkeypatch):
+    for v in ("0", "false", "no", "off", "garbage"):
+        monkeypatch.setenv("__TEST_FLAG__", v)
+        assert _env_flag("__TEST_FLAG__") is False
+
+
+# =====================================================================
+# 3. _safe_name — MCP 工具命名安全
+# =====================================================================
+
+def test_safe_name_basic():
+    assert _safe_name("hello world", 30) == "hello_world"
+
+
+def test_safe_name_maxlen():
+    assert _safe_name("abcdefghij" * 5, 10) == "abcdefghij"
+
+
+def test_safe_name_special_chars():
+    # 工具名只允许 [A-Za-z0-9_-]
+    assert _safe_name("tool/name:v1", 30) == "tool_name_v1"
+
+
+def test_safe_name_empty():
+    assert _safe_name("", 30) == ""
+
+
+# =====================================================================
+# 4. discover_skills: 三级降级(用 mock db 模拟 SQLAlchemy session)
+# =====================================================================
+
+class FakeQuery:
+    def __init__(self, rows):
+        self._rows = rows
+    def filter(self, *a, **kw):
+        return self
+    def all(self):
+        return self._rows
+    def order_by(self, *a, **kw):
+        return self
+
+
+class FakeDb:
+    """极简 db mock: 支持 .query(Class).filter(...).all() 链式调用。"""
+    def __init__(self, app_install_rows=None, market_app_rows=None):
+        self._app_install = app_install_rows or []
+        self._market_app = market_app_rows or []
+    def query(self, cls):
+        name = getattr(cls, "__name__", str(cls))
+        if name == "AppInstallation":
+            return FakeQuery(self._app_install)
+        if name == "MarketApp":
+            return FakeQuery(self._market_app)
+        return FakeQuery([])
+
+
+class FakeAppInstall:
+    def __init__(self, app_id):
+        self.app_id = app_id
+
+
+class FakeMarketApp:
+    def __init__(self, id, type, status):
+        self.id = id
+        self.type = type
+        self.status = status
+
+
+def test_discover_skills_installed_path(monkeypatch):
+    """user_id 提供 + app_installations 非空 → 走 installed 路径"""
+    db = FakeDb(
+        app_install_rows=[FakeAppInstall("skill-1")],
+        market_app_rows=[FakeMarketApp("skill-1", "skill", "published")],
+    )
+    # stub _resolve_market_skills 返回一个标定 list
+    monkeypatch.setattr(
+        "app.api.chat_skill._resolve_market_skills",
+        lambda d, ids, uid: [{"id": "skill-1", "name": "Search"}],
+        raising=False,
+    )
+    out = discover_skills(db, user_id="u-1", query="", limit=5, claw="web")
+    assert out and out[0]["id"] == "skill-1"
+
+
+def test_discover_skills_fallback_published(monkeypatch):
+    """app_installations 为空 → 降级 published 全集"""
+    db = FakeDb(
+        app_install_rows=[],
+        market_app_rows=[
+            FakeMarketApp("skill-1", "skill", "published"),
+            FakeMarketApp("skill-2", "skill", "published"),
+        ],
+    )
+    monkeypatch.setattr(
+        "app.api.chat_skill._resolve_market_skills",
+        lambda d, ids, uid: [{"id": i} for i in ids],
+        raising=False,
+    )
+    out = discover_skills(db, user_id="u-1", query="", limit=5, claw="web")
+    assert len(out) == 2
+    ids = {x["id"] for x in out}
+    assert ids == {"skill-1", "skill-2"}
+
+
+def test_discover_skills_fallback_disabled(monkeypatch):
+    """CAP_SKILL_FALLBACK_PUBLISHED=0 → 不降级,空池"""
+    monkeypatch.setenv("CAP_SKILL_FALLBACK_PUBLISHED", "0")
+    db = FakeDb(
+        app_install_rows=[],
+        market_app_rows=[FakeMarketApp("skill-1", "skill", "published")],
+    )
+    out = discover_skills(db, user_id="u-1", query="", limit=5, claw="web")
+    assert out == []
+
+
+def test_discover_skills_handles_exception(monkeypatch, caplog):
+    """db.query 抛异常 → 降级继续,不冒泡"""
+    class BoomDb:
+        def query(self, *a, **kw):
+            raise RuntimeError("DB down")
+
+    with caplog.at_level(logging.WARNING):
+        out = discover_skills(BoomDb(), user_id="u-1", query="", limit=5, claw="web")
+    assert out == []
+    assert any("安装技能发现失败" in r.message for r in caplog.records)
+
+
+def test_discover_skills_limit_zero():
+    db = FakeDb()
+    out = discover_skills(db, user_id=None, query="", limit=0, claw="im")
+    assert out == []
+
+
+# =====================================================================
+# 5. discover_agents: explicit > installed > published
+# =====================================================================
+
+def test_discover_agents_explicit_priority(monkeypatch):
+    """explicit_ids 存在 → 跳过 installed/published"""
+    db = FakeDb()  # 即便有 installed/published,也不该用
+    monkeypatch.setattr(
+        "app.api.chat_skill._resolve_market_agents",
+        lambda d, ids, uid: [{"id": "a-explicit"}],
+        raising=False,
+    )
+    out = discover_agents(db, user_id="u-1", query="", explicit_ids=["a-explicit"], limit=5, claw="im")
+    assert out and out[0]["id"] == "a-explicit"
+
+
+def test_discover_agents_installed_path(monkeypatch):
+    db = FakeDb(
+        app_install_rows=[FakeAppInstall("a-1")],
+        market_app_rows=[FakeMarketApp("a-1", "agent", "published")],
+    )
+    monkeypatch.setattr(
+        "app.api.chat_skill._resolve_market_agents",
+        lambda d, ids, uid: [{"id": i} for i in ids],
+        raising=False,
+    )
+    out = discover_agents(db, user_id="u-1", query="", explicit_ids=None, limit=5, claw="web")
+    assert out and out[0]["id"] == "a-1"
+
+
+# =====================================================================
+# 6. discover_mcp_servers: explicit vs auto
+# =====================================================================
+
+def test_discover_mcp_explicit(monkeypatch):
+    db = FakeDb()
+    monkeypatch.setattr(
+        "app.api.chat_mcp._resolve_active_mcp_servers",
+        lambda d, ids: [{"id": i, "name": i} for i in ids],
+        raising=False,
+    )
+    out = discover_mcp_servers(db, explicit_ids=["m1", "m2"], limit=5, claw="web")
+    assert len(out) == 2
+    assert {x["id"] for x in out} == {"m1", "m2"}
+
+
+def test_discover_mcp_auto(monkeypatch):
+    db = FakeDb()
+    monkeypatch.setattr(
+        "app.api.chat_mcp._list_enabled_mcp_servers",
+        lambda d: [{"id": "m-auto", "name": "auto"}],
+        raising=False,
+    )
+    out = discover_mcp_servers(db, explicit_ids=None, limit=5, claw="im")
+    assert out and out[0]["id"] == "m-auto"
+
+
+def test_discover_mcp_auto_disabled(monkeypatch):
+    monkeypatch.setenv("CAP_MCP_AUTO", "0")
+    db = FakeDb()
+    out = discover_mcp_servers(db, explicit_ids=None, limit=5, claw="im")
+    assert out == []
+
+
+def test_discover_mcp_limit_clips(monkeypatch):
+    db = FakeDb()
+    monkeypatch.setattr(
+        "app.api.chat_mcp._resolve_active_mcp_servers",
+        lambda d, ids: [{"id": i, "name": i} for i in ids],
+        raising=False,
+    )
+    out = discover_mcp_servers(db, explicit_ids=["a", "b", "c", "d"], limit=2, claw="web")
+    assert len(out) == 2
+
+
+# =====================================================================
+# 7. build_mcp_tool_defs: 命名 + 异常降级
+# =====================================================================
+
+def test_build_mcp_tool_defs_naming():
+    """验证命名规则 mcp__{server}__{tool}__{idx} 严格一致"""
+    # 旁路 list_mcp_tools 直接构造 servers
+    async def fake_list(srv):
+        return [{"name": "list_things", "description": "List things", "inputSchema": {"type": "object"}}]
+    import app.mcp_client
+    orig = app.mcp_client.list_mcp_tools
+    app.mcp_client.list_mcp_tools = fake_list
+    try:
+        defs, tool_map = asyncio.run(build_mcp_tool_defs(
+            [{"id": "s1", "name": "github"}, {"id": "s2", "name": "fs"}],
+            idx_start=0,
+        ))
+    finally:
+        app.mcp_client.list_mcp_tools = orig
+    assert len(defs) == 2
+    assert defs[0]["name"] == "mcp__github__list_things__0"
+    assert defs[1]["name"] == "mcp__fs__list_things__1"
+    assert defs[0]["name"] in tool_map
+    assert tool_map[defs[0]["name"]]["server_name"] == "github"
+
+
+def test_build_mcp_tool_defs_continues_on_error(caplog):
+    async def fake_list(srv):
+        if srv["id"] == "bad":
+            raise RuntimeError("mcp boom")
+        return [{"name": "ok", "description": "ok", "inputSchema": {}}]
+    import app.mcp_client
+    orig = app.mcp_client.list_mcp_tools
+    app.mcp_client.list_mcp_tools = fake_list
+    try:
+        with caplog.at_level(logging.WARNING):
+            defs, tool_map = asyncio.run(build_mcp_tool_defs(
+                [{"id": "bad", "name": "broken"}, {"id": "ok", "name": "good"}],
+            ))
+    finally:
+        app.mcp_client.list_mcp_tools = orig
+    assert len(defs) == 1
+    assert defs[0]["name"] == "mcp__good__ok__1"
+    assert any("MCP 拉取工具失败" in r.message for r in caplog.records)
+
diff --git a/tests/test_llm_router.py b/tests/test_llm_router.py
new file mode 100644
index 0000000..e8ca5de
--- /dev/null
+++ b/tests/test_llm_router.py
@@ -0,0 +1,267 @@
+"""P1-1: 单测 — llm_router 路由/熔断/上下文(双 Claw 共享 #875)。
+
+为什么这个文件存在:
+    LLM Router 是 #875 加速带的关键能力,但 #937 副密钥 failover 设计依托
+    同一熔断 / 决策路径。锁死关键纯函数 + mock 路径,防止回归。
+
+验证范围:
+    1. 上下文注入: set_route_ctx → get_route_ctx 含 tenant/claw/trace
+    2. OTel 桥接: set_route_ctx 同步注入 OTel 双 Claw 标签
+    3. _is_text_only: 排除 media-only 模型(关键坑:failover 误用图像模型给文本)
+    4. _tenant_authorized: 全局 / 私有模型归属校验
+    5. _fallbacks: 排除 exclude_ids + 仅纯文本 + is_default desc
+    6. 熔断: report_failure ×3 → _fail_count = 3(内存降级)
+    7. 熔断: report_success 清零
+    8. 熔断: TTL 过期自动清零
+    9. RouteDecision 数据类可序列化 / 字段完整
+"""
+from __future__ import annotations
+
+import os
+import sys
+import time
+
+import pytest
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(_HERE, ".."))
+
+from app.core import llm_router as lr  # noqa: E402
+from app.core.llm_router import (  # noqa: E402
+    SHADOW,
+    _is_text_only,
+    _tenant_authorized,
+    _fallbacks,
+    _fail_count,
+    _BREAKER_LOCAL,
+    report_failure,
+    report_success,
+    set_route_ctx,
+    get_route_ctx,
+    RouteDecision,
+)
+
+
+# =====================================================================
+# 1. 上下文注入
+# =====================================================================
+
+def test_set_route_ctx_default():
+    set_route_ctx()
+    ctx = get_route_ctx()
+    assert ctx["claw_type"] == "web"
+    assert ctx["tenant_id"] == ""
+    assert "trace_id" in ctx
+
+
+def test_set_route_ctx_im():
+    set_route_ctx(tenant_id="t-1", claw_type="im")
+    ctx = get_route_ctx()
+    assert ctx["tenant_id"] == "t-1"
+    assert ctx["claw_type"] == "im"
+
+
+def test_set_route_ctx_otel_bridge(monkeypatch):
+    """set_route_ctx 应同步注入 OTel 双 Claw 标签"""
+    captured = {}
+
+    def fake_set_otel_ctx(*, tenant_id, claw_type):
+        captured["tenant_id"] = tenant_id
+        captured["claw_type"] = claw_type
+
+    # patch the lazy import target
+    import app.core.otel as otel_mod
+    monkeypatch.setattr(otel_mod, "set_otel_ctx", fake_set_otel_ctx)
+    set_route_ctx(tenant_id="t-9", claw_type="im")
+    assert captured == {"tenant_id": "t-9", "claw_type": "im"}
+
+
+def test_set_route_ctx_otel_silently_swallows_failure(monkeypatch):
+    """otel 不可用时静默跳过,不冒泡"""
+    import app.core.otel as otel_mod
+    def boom(**kw):
+        raise RuntimeError("otel down")
+    monkeypatch.setattr(otel_mod, "set_otel_ctx", boom)
+    # 应不抛
+    set_route_ctx(tenant_id="t-x", claw_type="web")
+
+
+# =====================================================================
+# 2. _is_text_only — failover 备选关键判断
+# =====================================================================
+
+class FakeModel:
+    def __init__(self, capabilities):
+        self.capabilities = capabilities
+
+
+def test_is_text_only_pure_text():
+    assert _is_text_only(FakeModel(["text"])) is True
+
+
+def test_is_text_only_excludes_image():
+    """纯图像模型不是文本 failover 备选"""
+    assert _is_text_only(FakeModel(["image"])) is False
+
+
+def test_is_text_only_excludes_mixed_with_media():
+    """含 text 但同时有 image → 视为媒体"""
+    assert _is_text_only(FakeModel(["text", "image"])) is False
+
+
+def test_is_text_only_handles_empty():
+    assert _is_text_only(FakeModel([])) is False
+    assert _is_text_only(FakeModel(None)) is False
+
+
+# =====================================================================
+# 3. _tenant_authorized
+# =====================================================================
+
+class FakeModelWithTenant:
+    def __init__(self, tenant_id):
+        self.tenant_id = tenant_id
+
+
+def test_tenant_authorized_global_open():
+    """全局模型(无 tenant_id)任何人可用"""
+    m = FakeModelWithTenant(None)
+    assert _tenant_authorized(m, "t-1") is True
+    assert _tenant_authorized(m, "") is True
+
+
+def test_tenant_authorized_private_match():
+    m = FakeModelWithTenant("t-1")
+    assert _tenant_authorized(m, "t-1") is True
+
+
+def test_tenant_authorized_private_mismatch():
+    m = FakeModelWithTenant("t-1")
+    assert _tenant_authorized(m, "t-2") is False
+
+
+def test_tenant_authorized_no_tenant_blocked():
+    """私有模型 + caller 没传 tenant → 拒绝"""
+    m = FakeModelWithTenant("t-1")
+    assert _tenant_authorized(m, "") is False
+
+
+# =====================================================================
+# 4. _fallbacks: 排除 exclude_ids + 纯文本 + is_default desc
+# =====================================================================
+
+class FakeAimModel:
+    def __init__(self, id, capabilities, is_default=False, is_enabled=True):
+        self.id = id
+        self.capabilities = capabilities
+        self.is_default = is_default
+        self.is_enabled = is_enabled
+
+
+class FakeDb:
+    def __init__(self, rows):
+        self._rows = rows
+    def query(self, cls):
+        rows = self._rows
+        class Q:
+            def filter(self_inner, *a, **kw):
+                return self_inner
+            def order_by(self_inner, *a, **kw):
+                return self_inner
+            def all(self_inner):
+                return rows
+        return Q()
+
+
+def test_fallbacks_excludes_exclude_ids():
+    db = FakeDb([
+        FakeAimModel("a", ["text"], is_default=True),
+        FakeAimModel("b", ["text"], is_default=False),
+    ])
+    out = _fallbacks(db, exclude_ids={"a"}, limit=4)
+    assert [m.id for m in out] == ["b"]
+
+
+def test_fallbacks_excludes_media():
+    db = FakeDb([
+        FakeAimModel("a", ["text"], is_default=True),
+        FakeAimModel("b", ["image"], is_default=True),  # 媒体不算
+    ])
+    out = _fallbacks(db, exclude_ids=set(), limit=4)
+    assert [m.id for m in out] == ["a"]
+
+
+def test_fallbacks_limit_caps():
+    db = FakeDb([
+        FakeAimModel(f"m{i}", ["text"], is_default=False) for i in range(10)
+    ])
+    out = _fallbacks(db, exclude_ids=set(), limit=3)
+    assert len(out) == 3
+
+
+# =====================================================================
+# 5. 熔断: 内存降级(无 Redis)
+# =====================================================================
+
+def setup_function(_):
+    """每个测试前清空内存 breaker 状态"""
+    _BREAKER_LOCAL.clear()
+
+
+def test_breaker_increments_and_reads():
+    report_failure("m-1")
+    report_failure("m-1")
+    report_failure("m-1")
+    assert _fail_count("m-1") == 3
+
+
+def test_breaker_success_clears():
+    report_failure("m-2")
+    report_failure("m-2")
+    assert _fail_count("m-2") == 2
+    report_success("m-2")
+    assert _fail_count("m-2") == 0
+
+
+def test_breaker_ttl_expires(monkeypatch):
+    """TTL 过期后 _fail_count 返回 0,本地条目清理"""
+    report_failure("m-3")
+    # 假装时间穿越到未来(after TTL)
+    fake_now = time.time() + 1000
+    monkeypatch.setattr(lr.time, "time", lambda: fake_now)
+    assert _fail_count("m-3") == 0
+    assert "m-3" not in _BREAKER_LOCAL
+
+
+def test_breaker_empty_model_id():
+    report_failure("")
+    assert _fail_count("") == 0
+
+
+# =====================================================================
+# 6. RouteDecision 数据类
+# =====================================================================
+
+def test_route_decision_defaults():
+    rd = RouteDecision(capability="text", claw_type="web", tenant_id="t-1")
+    assert rd.capability == "text"
+    assert rd.claw_type == "web"
+    assert rd.tenant_id == "t-1"
+    assert rd.models == []
+    assert rd.primary is None
+    assert rd.shadow is True
+    assert rd.matched_legacy is None
+    assert rd.reason == ""
+
+
+def test_route_decision_fields_assignable():
+    rd = RouteDecision(capability="text", claw_type="im", tenant_id="t-1")
+    rd.models = ["a", "b"]
+    rd.primary = "a"
+    rd.reason = "explicit"
+    rd.matched_legacy = True
+    assert rd.models == ["a", "b"]
+    assert rd.primary == "a"
+    assert rd.reason == "explicit"
+    assert rd.matched_legacy is True
+