docs: 补齐 PR #11 评审要求的文档同步

- 技术栈说明实施状态：RAG Benchmark 标记为已完成、Agent Benchmark 暂缓
- 新增 Benchmark 开发说明，并登记到文档索引
- README 回归基线更新为后端 157 / 前端 29

Co-Authored-By: Claude Code <noreply@anthropic.com>

README.md

@@ -118,7 +118,7 @@ cd frontend
 pnpm test
 ```
 
-当前回归基线为后端 81 项测试、前端 27 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。
+当前回归基线为后端 157 项测试、前端 29 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。
 
 构建产物位于 `frontend/dist`，该目录不提交到 Git。
 
@@ -132,6 +132,7 @@ pnpm test
 | [后端接口契约](docs/contracts/后端接口契约-开发版.md) | HTTP/SSE 接口、错误和当前实现状态 |
 | [第二阶段接口契约](docs/contracts/第二阶段接口契约-开发版.md) | 第二阶段公共 DTO、计划接口、SSE、错误码与联调顺序 |
 | [AI Core 与 Agent Core](docs/development/AI-Core与Agent-Core开发说明.md) | Provider、Agent、Tool、Permission 与 Extension Core |
+| [MCP Bridge 与 Plugin Host](docs/development/MCP-Bridge与Plugin-Host开发说明.md) | stdio MCP、隔离进程、Tool 映射、状态与错误边界 |
 | [Git 使用细则](docs/guides/Git使用细则-团队开发版.md) | 分支、提交、PR、Review 与合并流程 |
 | [CI/CD 细则](docs/guides/CI-CD细则-团队开发版.md) | Gitea 流水线、质量门禁、产物、发布与回滚规则 |
 | [Agent Trace 复盘](docs/retrospectives/Agent-Core第二阶段问题与修复复盘.md) | Agent 持久化、SSE 恢复、事件契约与脱敏问题复盘 |

backend/README.md

@@ -23,7 +23,7 @@ uv run uvicorn app.main:app --reload --host 127.0.0.1 --port 8000
 uv run pytest
 ```
 
-当前基线为 81 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入；不要把真实密钥写入仓库。
+当前基线为 92 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入；不要把真实密钥写入仓库。
 
 团队接口清单见 `../docs/contracts/后端接口契约-开发版.md`，机器可读契约以运行时的 `/openapi.json` 为准。
 

backend/app/agent/runtime.py

@@ -491,7 +491,13 @@ class AgentRuntime:
     async def _invoke_tool(self, record: RunRecord, call: ToolCall) -> ToolResult:
         try:
             return await asyncio.wait_for(
-                self.tools.execute(call, ToolExecutionContext(run_id=record.run.run_id)),
+                self.tools.execute(
+                    call,
+                    ToolExecutionContext(
+                        run_id=record.run.run_id,
+                        tool_call_id=call.tool_call_id,
+                    ),
+                ),
                 timeout=record.request.tool_timeout_seconds,
             )
         except TimeoutError:

backend/app/agent/tools.py

@@ -1,6 +1,7 @@
 """Agent 工具注册与执行边界。"""
 
 import inspect
+import threading
 from dataclasses import dataclass
 from time import perf_counter
 from typing import Any, Awaitable, Callable
@@ -17,6 +18,7 @@ ToolExecutor = Callable[[BaseModel, "ToolExecutionContext"], Any | Awaitable[Any
 @dataclass(frozen=True, slots=True)
 class ToolExecutionContext:
     run_id: str
+    tool_call_id: str | None = None
 
 
 @dataclass(slots=True)
@@ -30,11 +32,21 @@ class ToolNotFoundError(LookupError):
     pass
 
 
+class ToolExecutionError(RuntimeError):
+    """Executor 可预期失败，保留领域错误码而不是折叠成通用异常。"""
+
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
 class ToolRegistry:
     """统一校验工具入参并隔离执行异常，避免单个工具击穿 Agent 主循环。"""
 
     def __init__(self) -> None:
         self._tools: dict[str, RegisteredTool] = {}
+        self._lock = threading.RLock()
 
     def register(
         self,
@@ -42,33 +54,38 @@ class ToolRegistry:
         arguments_model: type[BaseModel],
         executor: ToolExecutor,
     ) -> None:
-        if definition.name in self._tools:
-            raise ValueError(f"Tool already registered: {definition.name}")
-        self._tools[definition.name] = RegisteredTool(
-            definition=definition,
-            arguments_model=arguments_model,
-            executor=executor,
-        )
+        with self._lock:
+            if definition.name in self._tools:
+                raise ValueError(f"Tool already registered: {definition.name}")
+            self._tools[definition.name] = RegisteredTool(
+                definition=definition,
+                arguments_model=arguments_model,
+                executor=executor,
+            )
 
     def unregister(self, name: str) -> None:
-        self._tools.pop(name, None)
+        with self._lock:
+            self._tools.pop(name, None)
 
     def contains(self, name: str) -> bool:
-        return name in self._tools
+        with self._lock:
+            return name in self._tools
 
     def get(self, name: str) -> RegisteredTool:
-        try:
-            return self._tools[name]
-        except KeyError as exc:
-            raise ToolNotFoundError(name) from exc
+        with self._lock:
+            try:
+                return self._tools[name]
+            except KeyError as exc:
+                raise ToolNotFoundError(name) from exc
 
     def definitions(self, allowed: list[str] | None = None) -> list[ToolDefinition]:
         names = set(allowed) if allowed is not None else None
-        return [
-            item.definition.model_copy(deep=True)
-            for name, item in self._tools.items()
-            if names is None or name in names
-        ]
+        with self._lock:
+            return [
+                item.definition.model_copy(deep=True)
+                for name, item in self._tools.items()
+                if names is None or name in names
+            ]
 
     async def execute(self, call: ToolCall, context: ToolExecutionContext) -> ToolResult:
         started = perf_counter()
@@ -108,6 +125,15 @@ class ToolRegistry:
                 output=output,
                 duration_ms=round((perf_counter() - started) * 1000),
             )
+        except ToolExecutionError as exc:
+            return ToolResult(
+                tool_call_id=call.tool_call_id,
+                name=call.name,
+                success=False,
+                error_code=exc.code,
+                error_message=exc.message,
+                duration_ms=round((perf_counter() - started) * 1000),
+            )
         except Exception as exc:  # 工具失败转换成结构化结果，由模型决定是否降级或重试。
             return ToolResult(
                 tool_call_id=call.tool_call_id,

backend/app/benchmarks/__init__.py

@@ -0,0 +1,8 @@
+"""Benchmark 服务：RAG / Agent 数据集注册、指标计算与运行管理。
+
+模块划分：
+- metrics.py  纯函数指标（Hit@K / Recall@K / MRR / CitationHit / 分位数）
+- datasets.py 受控目录的 Dataset 注册与校验
+- rag.py      RAG Benchmark Runner（调用 retrieval.engine.search）
+- service.py  运行注册表、配置快照与报告组装
+"""

backend/app/benchmarks/datasets.py

@@ -0,0 +1,198 @@
+"""Benchmark Dataset 注册：从受控目录加载 JSON 数据集并校验。
+
+Dataset 只能来自配置目录（settings.benchmark_datasets_path），API 不接受调用方提交
+任意文件路径。目录不存在或为空时按「无数据集」处理，不报错。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from pydantic import BaseModel, Field, ValidationError
+
+from app.config import get_settings
+from app.contracts import (
+    BenchmarkDatasetInfo,
+    BenchmarkKind,
+    RAGDatasetCase,
+)
+from app.errors import ApiError
+
+
+@dataclass
+class RAGDataset:
+    """内存中的 RAG 数据集：元信息 + 已校验的 Case 列表 + 内容哈希。"""
+
+    dataset_id: str
+    kind: BenchmarkKind
+    version: str
+    description: str
+    cases: list[RAGDatasetCase] = field(default_factory=list)
+    content_hash: str = ""
+
+
+class _DatasetMeta(BaseModel):
+    """Dataset 元数据的最小校验模型。
+
+    list_datasets 用它逐文件校验元信息字段结构，把「合法 JSON 但字段类型错误」
+    （如 cases: 42）这类损坏文件隔离掉，而不是让 len() 抛 TypeError 拖垮整个列表。
+    """
+
+    dataset_id: str = Field(min_length=1)
+    kind: str = ""
+    version: str = ""
+    description: str = ""
+    cases: list = Field(default_factory=list)
+
+
+def _datasets_dir() -> Path:
+    return get_settings().benchmark_datasets_path
+
+
+def _dataset_files() -> list[Path]:
+    directory = _datasets_dir()
+    if not directory.is_dir():
+        return []
+    return sorted(directory.glob("*.json"))
+
+
+def _content_hash(raw: bytes) -> str:
+    return "sha256:" + hashlib.sha256(raw).hexdigest()
+
+
+def _read_json(path: Path) -> tuple[dict, bytes]:
+    """读取并解析 JSON 文件，返回 (dict, 原始字节)；非法 JSON 抛 BENCHMARK_DATASET_INVALID。"""
+    try:
+        raw_bytes = path.read_bytes()
+        return json.loads(raw_bytes.decode("utf-8")), raw_bytes
+    except (json.JSONDecodeError, OSError, UnicodeDecodeError) as exc:
+        raise ApiError(
+            422,
+            "BENCHMARK_DATASET_INVALID",
+            f"Dataset file is not valid JSON: {path.name}",
+            {"path": str(path)},
+        ) from exc
+
+
+def _dataset_from_raw(raw: dict, raw_bytes: bytes, kind: BenchmarkKind) -> RAGDataset:
+    """把单个数据集 JSON 解析为 RAGDataset，非法结构抛 BENCHMARK_DATASET_INVALID。"""
+    dataset_id = raw.get("dataset_id")
+    if not isinstance(dataset_id, str) or not dataset_id:
+        raise ApiError(
+            422,
+            "BENCHMARK_DATASET_INVALID",
+            "Dataset must declare a non-empty string 'dataset_id'.",
+            {},
+        )
+    file_kind = raw.get("kind", kind.value)
+    if file_kind != kind.value:
+        raise ApiError(
+            422,
+            "BENCHMARK_DATASET_INVALID",
+            f"Dataset kind mismatch: expected '{kind.value}', got '{file_kind}'.",
+            {"dataset_id": dataset_id},
+        )
+    raw_cases = raw.get("cases")
+    if not isinstance(raw_cases, list) or not raw_cases:
+        raise ApiError(
+            422,
+            "BENCHMARK_DATASET_INVALID",
+            "Dataset 'cases' must be a non-empty list.",
+            {"dataset_id": dataset_id},
+        )
+
+    cases: list[RAGDatasetCase] = []
+    for index, case in enumerate(raw_cases):
+        try:
+            parsed = RAGDatasetCase.model_validate(case)
+        except ValidationError as exc:
+            raise ApiError(
+                422,
+                "BENCHMARK_DATASET_INVALID",
+                f"Dataset case #{index} is invalid.",
+                {"dataset_id": dataset_id, "case_index": index, "errors": exc.errors()},
+            ) from exc
+        # 每个 Case 至少要声明一个期望 id，否则无法计算命中/召回
+        if not parsed.expected_note_ids and not parsed.expected_block_ids:
+            raise ApiError(
+                422,
+                "BENCHMARK_DATASET_INVALID",
+                f"Dataset case '{parsed.case_id}' must declare expected_note_ids or expected_block_ids.",
+                {"dataset_id": dataset_id, "case_id": parsed.case_id},
+            )
+        # citation_required=true 时必须声明 expected_block_ids，否则无法计算 Citation Hit Rate
+        if parsed.citation_required and not parsed.expected_block_ids:
+            raise ApiError(
+                422,
+                "BENCHMARK_DATASET_INVALID",
+                f"Dataset case '{parsed.case_id}' requires expected_block_ids when citation_required is true.",
+                {"dataset_id": dataset_id, "case_id": parsed.case_id},
+            )
+        cases.append(parsed)
+
+    return RAGDataset(
+        dataset_id=dataset_id,
+        kind=kind,
+        version=str(raw.get("version", "")),
+        description=str(raw.get("description", "")),
+        cases=cases,
+        content_hash=_content_hash(raw_bytes),
+    )
+
+
+def list_datasets(kind: BenchmarkKind) -> list[BenchmarkDatasetInfo]:
+    """枚举受控目录下指定 kind 的数据集元信息（不含 Case 内容）。
+
+    逐文件用 _DatasetMeta 校验元信息字段结构，单个损坏文件隔离跳过而非整体失败，
+    保证列表接口健壮；损坏细节由 load_dataset 抛出。
+    """
+    infos: list[BenchmarkDatasetInfo] = []
+    for path in _dataset_files():
+        try:
+            raw, raw_bytes = _read_json(path)
+            meta = _DatasetMeta.model_validate(raw)
+        except (ApiError, ValidationError):
+            continue
+        if meta.kind not in ("", kind.value):
+            continue
+        infos.append(
+            BenchmarkDatasetInfo(
+                dataset_id=meta.dataset_id,
+                kind=kind,
+                version=meta.version,
+                description=meta.description,
+                case_count=len(meta.cases),
+                content_hash=_content_hash(raw_bytes),
+            )
+        )
+    return infos
+
+
+def load_dataset(dataset_id: str, kind: BenchmarkKind) -> RAGDataset:
+    """按文件名加载并校验数据集；找不到抛 BENCHMARK_DATASET_NOT_FOUND。
+
+    只读取与请求 dataset_id 同名的文件（{dataset_id}.json），无关文件的损坏（JSON 语法
+    错误、UTF-8 解码错误、顶层非对象）不会阻断目标数据集加载；只有目标文件本身损坏
+    才抛 BENCHMARK_DATASET_INVALID。按现有文件 stem 精确匹配，不拼接调用方传入的路径。
+    """
+    for path in _dataset_files():
+        if path.stem != dataset_id:
+            continue
+        raw, raw_bytes = _read_json(path)
+        if not isinstance(raw, dict):
+            raise ApiError(
+                422,
+                "BENCHMARK_DATASET_INVALID",
+                "Dataset top-level must be a JSON object.",
+                {"dataset_id": dataset_id, "path": path.name},
+            )
+        return _dataset_from_raw(raw, raw_bytes, kind)
+    raise ApiError(
+        404,
+        "BENCHMARK_DATASET_NOT_FOUND",
+        f"Benchmark dataset does not exist: {dataset_id}",
+        {"dataset_id": dataset_id, "kind": kind.value},
+    )

backend/app/benchmarks/metrics.py

@@ -0,0 +1,58 @@
+"""Benchmark 指标纯函数。
+
+所有指标只依赖「按相关性降序的 retrieved id 列表」和「期望 id 集合」，不接触任何
+外部状态，便于单元测试与未来 Agent Benchmark 复用。retrieved 顺序越靠前越相关。
+"""
+
+from __future__ import annotations
+
+
+def hit_at_k(retrieved: list[str], expected: set[str], k: int) -> bool:
+    """前 k 个结果里是否命中任意期望 id（用于 Hit@1 / Hit@5）。"""
+    return any(item in expected for item in retrieved[:k])
+
+
+def recall_at_k(retrieved: list[str], expected: set[str], k: int) -> float:
+    """前 k 个结果召回的期望 id 占比；期望为空时视为 0。
+
+    结果先去重：检索结果是 Block 级，同一 Note 可能经多个 Block 重复出现，
+    直接逐项计数会把同一 Note 算多次、导致 Recall 超过 1。
+    """
+    if not expected:
+        return 0.0
+    return len(set(retrieved[:k]) & expected) / len(expected)
+
+
+def reciprocal_rank(retrieved: list[str], expected: set[str]) -> float:
+    """首个命中的倒数排名；未命中返回 0。rank 从 1 开始。"""
+    for rank, item in enumerate(retrieved, start=1):
+        if item in expected:
+            return 1.0 / rank
+    return 0.0
+
+
+def citation_hit(retrieved_block_ids: list[str], expected: set[str]) -> bool:
+    """首条结果的 block_id 是否为期望引用块（Citation Hit Rate 的逐 Case 判据）。"""
+    if not retrieved_block_ids or not expected:
+        return False
+    return retrieved_block_ids[0] in expected
+
+
+def mean(values: list[float]) -> float:
+    return sum(values) / len(values) if values else 0.0
+
+
+def percentile(values: list[float], p: float) -> float:
+    """线性插值分位数（p ∈ [0, 100]），用于 P50 / P95 延迟。空列表返回 0。"""
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    if len(ordered) == 1:
+        return ordered[0]
+    rank = (len(ordered) - 1) * (p / 100.0)
+    lo = int(rank)
+    hi = lo + 1
+    if hi >= len(ordered):
+        return ordered[-1]
+    frac = rank - lo
+    return ordered[lo] + (ordered[hi] - ordered[lo]) * frac

backend/app/benchmarks/rag.py

@@ -0,0 +1,143 @@
+"""RAG Benchmark Runner：调用检索引擎对数据集逐 Case 求值并聚合指标。
+
+只读操作，直接复用 app.retrieval.engine 的 search()，不旁路检索链路。指标按
+(mode, case, repeat) 逐样本计算，再按 mode 聚合；失败样本按零分计入质量指标分母，
+避免把执行失败误判为检索质量（同时保留 total/successful/failed/failure_rate）。
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Callable
+
+from app.benchmarks import metrics as m
+from app.benchmarks.datasets import RAGDataset
+from app.contracts import (
+    RAGCaseResult,
+    RAGDatasetCase,
+    RAGMetrics,
+    RAGRunRequest,
+    SearchMode,
+    SearchRequest,
+)
+from app.retrieval.engine import engine
+
+logger = logging.getLogger(__name__)
+
+
+class BenchmarkCancelled(Exception):
+    """运行在 Case 之间被取消时抛出，用于中断后台执行并标记 cancelled。"""
+
+
+async def run_rag(
+    dataset: RAGDataset,
+    request: RAGRunRequest,
+    on_case: Callable[[RAGCaseResult, int, int], None] | None = None,
+    should_cancel: Callable[[], bool] | None = None,
+) -> tuple[dict[str, RAGMetrics], list[RAGCaseResult]]:
+    """执行 RAG Benchmark，返回 (按 mode 聚合的指标, 全部逐样本结果)。
+
+    on_case 在每个样本求值完成后回调 (result, done, total)，供上层更新进度与事件。
+    should_cancel 在每个样本开始前被检查；返回 True 时抛出 BenchmarkCancelled 中断运行。
+    """
+    total = len(request.modes) * len(dataset.cases) * request.repeat
+    done = 0
+    results: list[RAGCaseResult] = []
+
+    for mode in request.modes:
+        for case in dataset.cases:
+            for repeat in range(request.repeat):
+                if should_cancel is not None and should_cancel():
+                    raise BenchmarkCancelled()
+                result = await _evaluate_one(case, mode, request, repeat)
+                results.append(result)
+                done += 1
+                if on_case is not None:
+                    on_case(result, done, total)
+
+    metrics_by_mode = {mode.value: _aggregate(results, mode) for mode in request.modes}
+    return metrics_by_mode, results
+
+
+async def _evaluate_one(
+    case: RAGDatasetCase, mode: SearchMode, request: RAGRunRequest, repeat: int
+) -> RAGCaseResult:
+    search_request = SearchRequest(
+        query=case.query,
+        mode=mode,
+        limit=request.retrieval.top_k,
+        include_snippet=False,
+        rrf_k=request.retrieval.rrf_k,
+        rerank=request.retrieval.rerank,
+        rerank_candidates=request.retrieval.rerank_candidates,
+        score_threshold=request.retrieval.score_threshold,
+    )
+    start = time.perf_counter()
+    try:
+        response = await engine.search(search_request)
+        latency_ms = (time.perf_counter() - start) * 1000.0
+    except Exception as exc:  # 单个样本失败不中断整个 Benchmark
+        # 详细异常只进日志，公开响应只带项目错误码与安全消息，避免泄露路径/SQL 等敏感信息
+        logger.warning(
+            "RAG case evaluation failed: case=%s mode=%s", case.case_id, mode.value,
+            exc_info=exc,
+        )
+        return RAGCaseResult(
+            case_id=case.case_id,
+            mode=mode,
+            repeat=repeat,
+            latency_ms=(time.perf_counter() - start) * 1000.0,
+            citation_applicable=case.citation_required,
+            error="RAG case evaluation failed.",
+            error_code="BENCHMARK_CASE_EVALUATION_FAILED",
+        )
+
+    retrieved_note_ids = [item.note_id for item in response.items]
+    retrieved_block_ids = [item.block_id for item in response.items]
+    expected_notes = set(case.expected_note_ids)
+    expected_blocks = set(case.expected_block_ids)
+    k = request.retrieval.top_k
+
+    return RAGCaseResult(
+        case_id=case.case_id,
+        mode=mode,
+        repeat=repeat,
+        latency_ms=latency_ms,
+        retrieved_note_ids=retrieved_note_ids,
+        retrieved_block_ids=retrieved_block_ids,
+        hit_at_1=m.hit_at_k(retrieved_note_ids, expected_notes, 1),
+        hit_at_5=m.hit_at_k(retrieved_note_ids, expected_notes, 5),
+        recall=m.recall_at_k(retrieved_note_ids, expected_notes, k),
+        reciprocal_rank=m.reciprocal_rank(retrieved_note_ids, expected_notes),
+        citation_hit=m.citation_hit(retrieved_block_ids, expected_blocks),
+        citation_applicable=case.citation_required,
+    )
+
+
+def _aggregate(cases: list[RAGCaseResult], mode: SearchMode) -> RAGMetrics:
+    samples = [c for c in cases if c.mode == mode]
+    total = len(samples)
+    failed = sum(1 for c in samples if c.error is not None)
+    successful = total - failed
+    if total == 0:
+        return RAGMetrics()
+
+    # 延迟只统计成功样本；失败样本按零分计入质量指标分母，避免汇总虚高
+    latencies = [c.latency_ms for c in samples if c.error is None]
+    citation_samples = [c for c in samples if c.citation_applicable]
+    return RAGMetrics(
+        hit_at_1=m.mean([1.0 if (c.error is None and c.hit_at_1) else 0.0 for c in samples]),
+        hit_at_5=m.mean([1.0 if (c.error is None and c.hit_at_5) else 0.0 for c in samples]),
+        recall_at_k=m.mean([c.recall if c.error is None else 0.0 for c in samples]),
+        mrr=m.mean([c.reciprocal_rank if c.error is None else 0.0 for c in samples]),
+        citation_hit_rate=m.mean(
+            [1.0 if (c.error is None and c.citation_hit) else 0.0 for c in citation_samples]
+        ),
+        p50_latency_ms=m.percentile(latencies, 50.0),
+        p95_latency_ms=m.percentile(latencies, 95.0),
+        total_cases=total,
+        successful_cases=successful,
+        failed_cases=failed,
+        failure_rate=failed / total,
+    )

backend/app/benchmarks/service.py

@@ -0,0 +1,348 @@
+"""Benchmark 服务：运行注册表、配置快照与报告组装。
+
+RAG Benchmark 采用「创建即返回 queued、后台 Task 异步执行」的模式（与 index_service
+的 rebuild 一致）：POST 创建后立即返回 202 queued 的 BenchmarkRun，由受管 asyncio.Task
+在后台逐 Case 求值，进度与事件实时写入内存注册表，供 SSE 订阅。运行记录、事件与报告
+暂存内存（_runs/_events/_reports），不持久化到 SQLite；后续接入异步任务队列时再落库。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import sys
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from app import repository
+from app.benchmarks import datasets
+from app.benchmarks.datasets import RAGDataset
+from app.benchmarks.rag import BenchmarkCancelled, run_rag
+from app.config import get_settings
+from app.contracts import (
+    BenchmarkEvent,
+    BenchmarkEventType,
+    BenchmarkKind,
+    BenchmarkReport,
+    BenchmarkRun,
+    BenchmarkStatus,
+    RAGCaseResult,
+    RAGMetrics,
+    RAGRunRequest,
+    SearchMode,
+)
+from app.errors import ApiError
+from app.retrieval.engine import engine
+
+logger = logging.getLogger(__name__)
+
+_runs: dict[str, BenchmarkRun] = {}
+_events: dict[str, list[BenchmarkEvent]] = {}
+_reports: dict[str, BenchmarkReport] = {}
+_tasks: dict[str, asyncio.Task] = {}
+_subscribers: dict[str, list[asyncio.Queue[BenchmarkEvent]]] = {}
+_cancel_flags: dict[str, asyncio.Event] = {}
+MAX_RUNS = 100
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _forget(run_id: str) -> None:
+    """移除一条 run 的全部内存态；仅在 run 处于终态时调用，避免打断活动任务。"""
+    _runs.pop(run_id, None)
+    _events.pop(run_id, None)
+    _reports.pop(run_id, None)
+    _tasks.pop(run_id, None)
+    _subscribers.pop(run_id, None)
+    _cancel_flags.pop(run_id, None)
+
+
+def _evict_terminal() -> bool:
+    """超过容量时淘汰最旧的终态 run；全部为活动 run 无法淘汰时返回 False。
+
+    绝不能删除仍在运行（queued/running）的 run：那会连带移除其 _cancel_flags 与
+    _subscribers，使后台 Task 访问时抛出 KeyError。
+    """
+    terminal = (BenchmarkStatus.completed, BenchmarkStatus.failed, BenchmarkStatus.cancelled)
+    while len(_runs) >= MAX_RUNS:
+        victim = next(
+            (rid for rid, run in _runs.items() if run.status in terminal), None
+        )
+        if victim is None:
+            return False
+        _forget(victim)
+    return True
+
+
+def _config_snapshot(request: RAGRunRequest, dataset: RAGDataset) -> dict:
+    """记录运行时的模型 / 索引 / 环境信息，保证报告可解释、可复现。"""
+    settings = get_settings()
+    return {
+        "dataset_id": dataset.dataset_id,
+        "dataset_hash": dataset.content_hash,
+        "dataset_version": dataset.version,
+        "modes": [m.value for m in request.modes],
+        "retrieval": request.retrieval.model_dump(),
+        "repeat": request.repeat,
+        "embedding": {
+            "model_id": engine.embedding.model_id,
+            "version": engine.embedding.version,
+            "dim": engine.embedding.dim,
+        },
+        "reranker": {
+            "model_id": engine.reranker.model_id,
+            "version": engine.reranker.version,
+        },
+        "index_meta": repository.get_index_meta(),
+        "app": {"version": settings.version, "environment": settings.environment},
+        "python": sys.version.split()[0],
+        "metadata": request.metadata,
+    }
+
+
+async def _validate_index_compatibility(request: RAGRunRequest) -> None:
+    """创建 RAG Run 前校验索引已建立且与当前 Embedding 模型/维度兼容。
+
+    空索引或不兼容索引会让所有模式得到全 0 指标，把环境/索引错误误判为检索质量差，
+    故在创建时即拒绝，返回 BENCHMARK_INDEX_INCOMPATIBLE。
+    """
+    stats = repository.stats()
+    meta = repository.get_index_meta()
+    needs_vector = any(m in (SearchMode.vector, SearchMode.hybrid) for m in request.modes)
+
+    reasons: list[str] = []
+    if stats["blocks"] == 0:
+        reasons.append("index is empty (no indexed blocks; run /api/index/rebuild first)")
+    if needs_vector:
+        if meta.get("embedding_model") != engine.embedding.model_id:
+            reasons.append(
+                f"embedding model mismatch: index={meta.get('embedding_model')!r}, "
+                f"engine={engine.embedding.model_id!r}"
+            )
+        if meta.get("embedding_dim") != str(engine.embedding.dim):
+            reasons.append(
+                f"embedding dimension mismatch: index={meta.get('embedding_dim')!r}, "
+                f"engine={engine.embedding.dim}"
+            )
+        if await engine.vector_store.count() == 0:
+            reasons.append("vector index is empty")
+    if reasons:
+        raise ApiError(
+            409,
+            "BENCHMARK_INDEX_INCOMPATIBLE",
+            "Benchmark index is not built or is incompatible with the current retrieval engine.",
+            {"reasons": reasons},
+        )
+
+
+async def create_rag_run(request: RAGRunRequest) -> BenchmarkRun:
+    """创建一次 RAG Benchmark，立即返回 queued 的 BenchmarkRun，由后台 Task 执行。"""
+    dataset = datasets.load_dataset(request.dataset_id, BenchmarkKind.rag)
+    await _validate_index_compatibility(request)
+
+    # 容量检查：先淘汰终态 run 腾空间；满容量且全为活动 run 时拒绝创建
+    if not _evict_terminal():
+        raise ApiError(
+            429,
+            "BENCHMARK_CAPACITY_EXCEEDED",
+            "Benchmark run capacity exceeded; wait for active runs to finish.",
+            {},
+        )
+
+    run_id = "benchmark_" + uuid4().hex[:12]
+    snapshot = _config_snapshot(request, dataset)
+    run = BenchmarkRun(
+        run_id=run_id,
+        kind=BenchmarkKind.rag,
+        dataset_id=dataset.dataset_id,
+        dataset_hash=dataset.content_hash,
+        status=BenchmarkStatus.queued,
+        progress=0.0,
+        config_snapshot=snapshot,
+        created_at=_now(),
+    )
+    _runs[run_id] = run
+    _events[run_id] = []
+    _subscribers[run_id] = []
+    _cancel_flags[run_id] = asyncio.Event()
+    _tasks[run_id] = asyncio.create_task(_execute_rag(run_id, request, dataset, snapshot))
+    return run
+
+
+async def _execute_rag(
+    run_id: str, request: RAGRunRequest, dataset: RAGDataset, snapshot: dict
+) -> None:
+    """后台执行 RAG Benchmark，实时更新进度/事件，结束后写入报告并关闭订阅。"""
+    cancel_event = _cancel_flags[run_id]
+
+    def emit(event_type: BenchmarkEventType, data: dict) -> None:
+        sequence = len(_events[run_id])
+        event = BenchmarkEvent(
+            event=event_type, run_id=run_id, sequence=sequence, data=data, timestamp=_now()
+        )
+        _events[run_id].append(event)
+        for queue in _subscribers.get(run_id, []):
+            queue.put_nowait(event)
+
+    def finish() -> None:
+        _subscribers.pop(run_id, None)
+        _cancel_flags.pop(run_id, None)
+
+    _runs[run_id] = _runs[run_id].model_copy(
+        update={"status": BenchmarkStatus.running, "started_at": _now()}
+    )
+    emit(
+        BenchmarkEventType.run_started,
+        {"dataset_id": dataset.dataset_id, "modes": [m.value for m in request.modes]},
+    )
+    total = len(request.modes) * len(dataset.cases) * request.repeat
+
+    def on_case(result: RAGCaseResult, done: int, _total: int) -> None:
+        progress = done / total if total else 1.0
+        _runs[run_id] = _runs[run_id].model_copy(update={"progress": progress})
+        emit(BenchmarkEventType.case_completed, result.model_dump(mode="json"))
+
+    try:
+        metrics_by_mode, results = await run_rag(
+            dataset,
+            request,
+            on_case=on_case,
+            should_cancel=cancel_event.is_set,
+        )
+    except BenchmarkCancelled:
+        _runs[run_id] = _runs[run_id].model_copy(
+            update={
+                "status": BenchmarkStatus.cancelled,
+                "progress": 1.0,
+                "completed_at": _now(),
+            }
+        )
+        emit(BenchmarkEventType.run_cancelled, {"status": BenchmarkStatus.cancelled.value})
+        _reports[run_id] = BenchmarkReport(
+            run_id=run_id,
+            kind=BenchmarkKind.rag,
+            dataset_id=dataset.dataset_id,
+            dataset_hash=dataset.content_hash,
+            status=BenchmarkStatus.cancelled,
+            config_snapshot=snapshot,
+        )
+        finish()
+        return
+    except Exception as exc:  # 单次运行失败不拖垮服务，记录错误后结束
+        # 详细异常只进日志，公开响应仅带项目错误码与安全消息，避免泄露路径/SQL 等敏感信息
+        logger.exception("Benchmark run failed: run_id=%s", run_id)
+        _runs[run_id] = _runs[run_id].model_copy(
+            update={
+                "status": BenchmarkStatus.failed,
+                "progress": 1.0,
+                "error": "Benchmark run failed.",
+                "error_code": "BENCHMARK_RUN_FAILED",
+                "completed_at": _now(),
+            }
+        )
+        emit(
+            BenchmarkEventType.run_failed,
+            {"error": "Benchmark run failed.", "error_code": "BENCHMARK_RUN_FAILED"},
+        )
+        _reports[run_id] = BenchmarkReport(
+            run_id=run_id,
+            kind=BenchmarkKind.rag,
+            dataset_id=dataset.dataset_id,
+            dataset_hash=dataset.content_hash,
+            status=BenchmarkStatus.failed,
+            config_snapshot=snapshot,
+            error="Benchmark run failed.",
+            error_code="BENCHMARK_RUN_FAILED",
+        )
+        finish()
+        return
+
+    metrics = {mode: m.model_dump() for mode, m in metrics_by_mode.items()}
+    _runs[run_id] = _runs[run_id].model_copy(
+        update={
+            "status": BenchmarkStatus.completed,
+            "progress": 1.0,
+            "metrics": metrics,
+            "completed_at": _now(),
+        }
+    )
+    emit(BenchmarkEventType.run_completed, {"metrics": metrics})
+    _reports[run_id] = BenchmarkReport(
+        run_id=run_id,
+        kind=BenchmarkKind.rag,
+        dataset_id=dataset.dataset_id,
+        dataset_hash=dataset.content_hash,
+        status=BenchmarkStatus.completed,
+        config_snapshot=snapshot,
+        metrics=metrics,
+        cases=results,
+    )
+    finish()
+
+
+def list_runs(
+    kind: BenchmarkKind | None = None,
+    status: BenchmarkStatus | None = None,
+    limit: int = 50,
+    offset: int = 0,
+) -> tuple[list[BenchmarkRun], int]:
+    runs = list(_runs.values())
+    if kind is not None:
+        runs = [r for r in runs if r.kind == kind]
+    if status is not None:
+        runs = [r for r in runs if r.status == status]
+    runs.sort(key=lambda r: r.created_at, reverse=True)
+    total = len(runs)
+    return runs[offset : offset + limit], total
+
+
+def get_run(run_id: str) -> BenchmarkRun | None:
+    return _runs.get(run_id)
+
+
+def get_report(run_id: str) -> BenchmarkReport | None:
+    return _reports.get(run_id)
+
+
+def get_events(run_id: str) -> list[BenchmarkEvent]:
+    return _events.get(run_id, [])
+
+
+def cancel_run(run_id: str) -> BenchmarkRun | None:
+    """取消运行：对 queued/running 设置取消标志，后台 Task 在 Case 边界检查后置为 cancelled。"""
+    run = _runs.get(run_id)
+    if run is None:
+        return None
+    if run.status in (BenchmarkStatus.queued, BenchmarkStatus.running):
+        _cancel_flags[run_id].set()
+    return run
+
+
+def subscribe(run_id: str) -> asyncio.Queue[BenchmarkEvent] | None:
+    """订阅运行事件流；运行已结束（completed/failed/cancelled）时返回 None。"""
+    run = _runs.get(run_id)
+    if run is None or run.status in (
+        BenchmarkStatus.completed,
+        BenchmarkStatus.failed,
+        BenchmarkStatus.cancelled,
+    ):
+        return None
+    queue: asyncio.Queue[BenchmarkEvent] = asyncio.Queue()
+    _subscribers.setdefault(run_id, []).append(queue)
+    return queue
+
+
+def unsubscribe(run_id: str, queue: asyncio.Queue[BenchmarkEvent]) -> None:
+    subscribers = _subscribers.get(run_id)
+    if subscribers and queue in subscribers:
+        subscribers.remove(queue)
+
+
+async def wait_for_run(run_id: str) -> BenchmarkRun:
+    """等待后台任务结束（测试/轮询用）；无任务时直接返回当前状态。"""
+    task = _tasks.get(run_id)
+    if task is not None:
+        await task
+    return _runs.get(run_id)

backend/app/config.py

@@ -24,6 +24,7 @@ class Settings:
     db_path: Path
     vault_path: Path
     attachments_path: Path
+    benchmark_datasets_path: Path
 
 
 @lru_cache
@@ -41,4 +42,7 @@ def get_settings() -> Settings:
         attachments_path=Path(
             os.getenv("APP_ATTACHMENTS_PATH", str(data_dir / "attachments"))
         ),
+        benchmark_datasets_path=Path(
+            os.getenv("APP_BENCHMARK_DATASETS_PATH", str(data_dir / "benchmarks"))
+        ),
     )

backend/app/container.py

@@ -3,7 +3,7 @@ from dataclasses import dataclass
 from app.agent import AgentRuntime, PermissionManager, PermissionPolicy, ToolRegistry
 from app.agent.builtin_tools import register_builtin_tools
 from app.contracts import ModelCapability, ProviderConfig, ProviderType
-from app.config import BACKEND_DIR
+from app.config import BACKEND_DIR, get_settings
 from app.extensions import PluginRuntime, SkillRuntime
 from app.providers import MockProvider, ProviderFactory, ProviderRegistry
 from app.providers.credentials import (
@@ -26,6 +26,7 @@ class ApplicationContainer:
 
 
 def build_container() -> ApplicationContainer:
+    settings = get_settings()
     credentials = EncryptedCredentialStore()
     provider_factory = ProviderFactory(
         ChainedCredentialResolver(credentials, EnvironmentCredentialResolver())
@@ -50,7 +51,12 @@ def build_container() -> ApplicationContainer:
     tools = ToolRegistry()
     register_builtin_tools(tools)
 
-    plugins = PluginRuntime(tools)
+    plugins = PluginRuntime(
+        tools,
+        # 当前 Python Host 尚无 OS 沙箱。生产构建必须保持关闭，直到
+        # Tauri/Rust Host 能签发绑定命令摘要的可信启动许可。
+        allow_unsandboxed_mcp=settings.environment == "development",
+    )
     plugins.install(BACKEND_DIR / "extensions" / "plugins" / "text-tools")
     plugins.enable("text-tools")
 

backend/app/contracts.py

@@ -2,7 +2,7 @@ from datetime import datetime
 from enum import Enum
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, SecretStr
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator
 
 
 class Contract(BaseModel):
@@ -144,6 +144,12 @@ class SearchRequest(Contract):
     limit: int = Field(default=20, ge=1, le=100)
     offset: int = Field(default=0, ge=0)
     include_snippet: bool = True
+    # 检索调优参数（Benchmark 与 Skill 共用）：控制 RRF / 精排 / 候选池 / 分数阈值。
+    # rerank_candidates=None 表示对全部候选精排（保留原有行为），Benchmark 传显式值。
+    rrf_k: int = Field(default=60, ge=1)
+    rerank: bool = True
+    rerank_candidates: int | None = Field(default=None, ge=1)
+    score_threshold: float = Field(default=0.0, ge=0.0)
 
 
 class Citation(Contract):
@@ -417,6 +423,10 @@ class ExtensionInstallRequest(Contract):
 class PluginBackend(Contract):
     type: Literal["mcp", "internal_rpc", "none"] = "none"
     transport: Literal["stdio", "http", "none"] = "none"
+    command: str | None = None
+    args: list[str] = Field(default_factory=list)
+    startup_timeout_seconds: int = Field(default=10, ge=1, le=60)
+    tool_timeout_seconds: int = Field(default=30, ge=1, le=600)
 
 
 class PluginContribution(Contract):
@@ -460,6 +470,28 @@ class PluginListResponse(Contract):
     items: list[Plugin] = Field(default_factory=list)
 
 
+class PluginHostState(str, Enum):
+    stopped = "stopped"
+    starting = "starting"
+    ready = "ready"
+    unhealthy = "unhealthy"
+    error = "error"
+
+
+class PluginHostStatus(Contract):
+    plugin_id: str
+    backend_type: Literal["mcp", "internal_rpc", "none"]
+    transport: Literal["stdio", "http", "none"]
+    status: PluginHostState
+    tools_count: int = 0
+    started_at: datetime | None = None
+    last_seen_at: datetime | None = None
+    protocol_version: str | None = None
+    server_name: str | None = None
+    server_version: str | None = None
+    error: str | None = None
+
+
 class PluginPermissionGrantRequest(Contract):
     permissions: list[str] = Field(default_factory=list)
 
@@ -626,3 +658,151 @@ class IndexJob(Contract):
     status: Literal["queued", "running", "completed", "failed"]
     scope: Literal["all", "notes", "vectors"]
     created_at: datetime
+
+
+# Benchmark
+class BenchmarkKind(str, Enum):
+    rag = "rag"
+    agent = "agent"
+
+
+class BenchmarkStatus(str, Enum):
+    queued = "queued"
+    running = "running"
+    completed = "completed"
+    failed = "failed"
+    cancelled = "cancelled"
+
+
+class RAGDatasetCase(Contract):
+    case_id: str
+    query: str = Field(min_length=1)
+    expected_note_ids: list[str] = Field(default_factory=list)
+    expected_block_ids: list[str] = Field(default_factory=list)
+    citation_required: bool = False
+    tags: list[str] = Field(default_factory=list)
+
+
+class RAGRetrievalConfig(Contract):
+    """RAG Benchmark 的检索参数。top_k 映射到 SearchRequest.limit，
+    其余参数透传到 SearchRequest，由检索引擎实际执行。"""
+
+    top_k: int = Field(default=10, ge=1, le=100)
+    rrf_k: int = Field(default=60, ge=1)
+    rerank: bool = True
+    rerank_candidates: int = Field(default=20, ge=1)
+    score_threshold: float = Field(default=0.0, ge=0.0)
+
+
+class RAGRunRequest(Contract):
+    dataset_id: str = Field(min_length=1)
+    modes: list[SearchMode] = Field(
+        default_factory=lambda: [SearchMode.fts, SearchMode.vector, SearchMode.hybrid],
+        min_length=1,
+    )
+    retrieval: RAGRetrievalConfig = Field(default_factory=RAGRetrievalConfig)
+    repeat: int = Field(default=1, ge=1, le=10)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("modes")
+    @classmethod
+    def _no_duplicate_modes(cls, value: list[SearchMode]) -> list[SearchMode]:
+        if len(value) != len(set(value)):
+            raise ValueError("modes must not contain duplicates")
+        return value
+
+
+class RAGMetrics(Contract):
+    hit_at_1: float = 0.0
+    hit_at_5: float = 0.0
+    recall_at_k: float = 0.0
+    mrr: float = 0.0
+    citation_hit_rate: float = 0.0
+    p50_latency_ms: float = 0.0
+    p95_latency_ms: float = 0.0
+    # 样本构成：失败样本按零分计入质量指标，汇总不虚高；报告据此可知实际分母
+    total_cases: int = 0
+    successful_cases: int = 0
+    failed_cases: int = 0
+    failure_rate: float = 0.0
+
+
+class BenchmarkDatasetInfo(Contract):
+    dataset_id: str
+    kind: BenchmarkKind
+    version: str
+    description: str = ""
+    case_count: int
+    content_hash: str
+
+
+class BenchmarkDatasetListResponse(Contract):
+    items: list[BenchmarkDatasetInfo] = Field(default_factory=list)
+
+
+class BenchmarkRun(Contract):
+    run_id: str
+    kind: BenchmarkKind
+    dataset_id: str
+    dataset_hash: str
+    status: BenchmarkStatus
+    progress: float | None = None
+    metrics: dict[str, Any] | None = None
+    config_snapshot: dict[str, Any] = Field(default_factory=dict)
+    error: str | None = None
+    error_code: str | None = None
+    created_at: datetime
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+
+
+class BenchmarkRunListResponse(Contract):
+    items: list[BenchmarkRun] = Field(default_factory=list)
+    page: PageMeta = Field(default_factory=PageMeta)
+
+
+class BenchmarkEventType(str, Enum):
+    run_started = "RunStarted"
+    case_completed = "CaseCompleted"
+    run_completed = "RunCompleted"
+    run_failed = "RunFailed"
+    run_cancelled = "RunCancelled"
+
+
+class BenchmarkEvent(Contract):
+    event: BenchmarkEventType
+    run_id: str
+    sequence: int
+    data: dict[str, Any] = Field(default_factory=dict)
+    timestamp: datetime
+
+
+class RAGCaseResult(Contract):
+    case_id: str
+    mode: SearchMode
+    repeat: int
+    latency_ms: float
+    retrieved_note_ids: list[str] = Field(default_factory=list)
+    retrieved_block_ids: list[str] = Field(default_factory=list)
+    hit_at_1: bool = False
+    hit_at_5: bool = False
+    recall: float = 0.0
+    reciprocal_rank: float = 0.0
+    citation_hit: bool = False
+    # 该 Case 是否声明了 expected_block_ids（决定是否计入 citation_hit_rate 分母）
+    citation_applicable: bool = False
+    error: str | None = None
+    error_code: str | None = None
+
+
+class BenchmarkReport(Contract):
+    run_id: str
+    kind: BenchmarkKind
+    dataset_id: str
+    dataset_hash: str
+    status: BenchmarkStatus
+    config_snapshot: dict[str, Any] = Field(default_factory=dict)
+    metrics: dict[str, Any] = Field(default_factory=dict)
+    cases: list[RAGCaseResult] = Field(default_factory=list)
+    error: str | None = None
+    error_code: str | None = None

backend/app/extensions/__init__.py

@@ -4,5 +4,13 @@ from app.extensions.runtime import (
     PluginRuntime,
     SkillRuntime,
 )
+from app.extensions.mcp import McpBridge, McpBridgeError
 
-__all__ = ["AgentConfiguration", "ExtensionError", "PluginRuntime", "SkillRuntime"]
+__all__ = [
+    "AgentConfiguration",
+    "ExtensionError",
+    "McpBridge",
+    "McpBridgeError",
+    "PluginRuntime",
+    "SkillRuntime",
+]

backend/app/extensions/mcp.py

@@ -0,0 +1,785 @@
+"""本地 stdio MCP Bridge。
+
+第三方 Server 始终运行在子进程中。Bridge 只把通过校验的 MCP Tool 转换为项目内部
+ToolDefinition/ToolResult，不把 MCP 原始协议泄露给 Agent Runtime 或前端。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import queue
+import subprocess
+import threading
+from collections import deque
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+
+from jsonschema import Draft202012Validator
+from jsonschema.exceptions import SchemaError
+
+from app.agent.permissions import KNOWN_PERMISSIONS
+from app.agent.tools import ToolExecutionError
+from app.contracts import (
+    PluginBackend,
+    PluginHostState,
+    PluginHostStatus,
+    ToolDefinition,
+)
+
+MCP_PROTOCOL_VERSION = "2025-11-25"
+SUPPORTED_PROTOCOL_VERSIONS = {
+    MCP_PROTOCOL_VERSION,
+    "2025-06-18",
+    "2025-03-26",
+    "2024-11-05",
+}
+MAX_MCP_MESSAGE_BYTES = 2 * 1024 * 1024
+MAX_MCP_TOOL_RESULT_BYTES = 256 * 1024
+MAX_MCP_TOOLS = 500
+MAX_MCP_LIST_PAGES = 100
+
+
+class McpBridgeError(RuntimeError):
+    def __init__(self, code: str, message: str, *, status_code: int = 502) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.status_code = status_code
+
+
+@dataclass(frozen=True, slots=True)
+class McpDiscoveredTool:
+    remote_name: str
+    definition: ToolDefinition
+
+
+@dataclass(slots=True)
+class _PendingRequest:
+    response: queue.Queue[dict[str, Any] | BaseException]
+
+
+class McpStdioClient:
+    """线程驱动的换行分隔 JSON-RPC 客户端，避免阻塞 FastAPI 事件循环。"""
+
+    def __init__(
+        self,
+        command: list[str],
+        *,
+        cwd: Path,
+        on_seen: Callable[[], None],
+        on_broken: Callable[[str], None],
+        on_tools_changed: Callable[[], None],
+    ) -> None:
+        self.command = command
+        self.cwd = cwd
+        self.on_seen = on_seen
+        self.on_broken = on_broken
+        self.on_tools_changed = on_tools_changed
+        self.process: subprocess.Popen[str] | None = None
+        self._write_lock = threading.Lock()
+        self._pending_lock = threading.Lock()
+        self._pending: dict[int, _PendingRequest] = {}
+        self._next_id = 1
+        self._stopping = False
+        # stderr 只在 Host 内部保留有限尾部，不进入 API、Trace 或普通日志。
+        self._stderr_tail: deque[str] = deque(maxlen=50)
+
+    def start(self) -> None:
+        if self.process is not None and self.process.poll() is None:
+            return
+        # TODO(extension-security): 社区 Plugin 开放前迁移到 Tauri/Rust Host 的
+        # 平台级沙箱启动器；uvx 只隔离 Python 依赖，不能替代系统权限限制。
+        creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0
+        environment = _subprocess_environment()
+        environment.setdefault("PYTHONUNBUFFERED", "1")
+        try:
+            self.process = subprocess.Popen(
+                self.command,
+                cwd=self.cwd,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                bufsize=1,
+                shell=False,
+                env=environment,
+                creationflags=creation_flags,
+            )
+        except OSError as exc:
+            raise McpBridgeError(
+                "PLUGIN_HOST_START_FAILED",
+                f"Cannot start MCP server process: {exc}",
+                status_code=503,
+            ) from exc
+        threading.Thread(target=self._stdout_loop, daemon=True).start()
+        threading.Thread(target=self._stderr_loop, daemon=True).start()
+
+    def request(
+        self,
+        method: str,
+        params: dict[str, Any],
+        *,
+        timeout: float,
+        timeout_code: str,
+        response_error_code: str = "MCP_TOOL_CALL_FAILED",
+    ) -> dict[str, Any]:
+        request_id, pending = self.begin_request(method, params)
+        return self.wait_response(
+            request_id,
+            pending,
+            timeout=timeout,
+            timeout_code=timeout_code,
+            response_error_code=response_error_code,
+        )
+
+    def begin_request(
+        self, method: str, params: dict[str, Any]
+    ) -> tuple[int, _PendingRequest]:
+        self._ensure_running()
+        with self._pending_lock:
+            request_id = self._next_id
+            self._next_id += 1
+            pending = _PendingRequest(response=queue.Queue(maxsize=1))
+            self._pending[request_id] = pending
+        try:
+            self._send(
+                {
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "method": method,
+                    "params": params,
+                }
+            )
+        except BaseException:
+            with self._pending_lock:
+                self._pending.pop(request_id, None)
+            raise
+        return request_id, pending
+
+    def wait_response(
+        self,
+        request_id: int,
+        pending: _PendingRequest,
+        *,
+        timeout: float,
+        timeout_code: str,
+        response_error_code: str = "MCP_TOOL_CALL_FAILED",
+    ) -> dict[str, Any]:
+        try:
+            response = pending.response.get(timeout=timeout)
+        except queue.Empty as exc:
+            self.cancel(request_id, "Request timed out.")
+            self.abandon(request_id)
+            raise McpBridgeError(timeout_code, "MCP request timed out.", status_code=504) from exc
+        if isinstance(response, BaseException):
+            raise response
+        if "error" in response:
+            error = response.get("error")
+            message = (
+                str(error.get("message", "MCP JSON-RPC error."))
+                if isinstance(error, dict)
+                else "MCP JSON-RPC error."
+            )
+            raise McpBridgeError(response_error_code, message)
+        result = response.get("result")
+        if not isinstance(result, dict):
+            raise McpBridgeError(
+                response_error_code, "MCP response result must be an object."
+            )
+        return result
+
+    def notify(self, method: str, params: dict[str, Any] | None = None) -> None:
+        payload: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
+        if params is not None:
+            payload["params"] = params
+        self._send(payload)
+
+    def cancel(self, request_id: int, reason: str = "Cancelled by host.") -> None:
+        try:
+            self.notify(
+                "notifications/cancelled",
+                {"requestId": request_id, "reason": reason},
+            )
+        except McpBridgeError:
+            pass
+
+    def abandon(
+        self, request_id: int, wake_error: BaseException | None = None
+    ) -> None:
+        with self._pending_lock:
+            pending = self._pending.pop(request_id, None)
+        # asyncio.to_thread 被取消时不会停止底层线程；主动唤醒 Queue，避免线程
+        # 一直占用默认线程池直至远端超时。
+        if pending is not None and wake_error is not None:
+            try:
+                pending.response.put_nowait(wake_error)
+            except queue.Full:
+                pass
+
+    def stop(self) -> None:
+        process = self.process
+        if process is None:
+            return
+        self._stopping = True
+        try:
+            if process.stdin:
+                try:
+                    process.stdin.close()
+                except (BrokenPipeError, OSError, ValueError):
+                    pass
+            try:
+                process.wait(timeout=2)
+            except subprocess.TimeoutExpired:
+                process.terminate()
+                try:
+                    process.wait(timeout=2)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait(timeout=2)
+        finally:
+            self._fail_pending(
+                McpBridgeError("PLUGIN_HOST_UNAVAILABLE", "MCP host stopped.", status_code=503)
+            )
+            self.process = None
+
+    def _send(self, message: dict[str, Any]) -> None:
+        self._ensure_running()
+        encoded = json.dumps(message, ensure_ascii=False, separators=(",", ":"))
+        if len(encoded.encode("utf-8")) > MAX_MCP_MESSAGE_BYTES:
+            raise McpBridgeError("MCP_TOOL_CALL_FAILED", "MCP request is too large.")
+        process = self.process
+        assert process is not None and process.stdin is not None
+        try:
+            with self._write_lock:
+                process.stdin.write(encoded + "\n")
+                process.stdin.flush()
+        except (BrokenPipeError, OSError, ValueError) as exc:
+            raise McpBridgeError(
+                "PLUGIN_HOST_UNAVAILABLE", "MCP host input is closed.", status_code=503
+            ) from exc
+
+    def _stdout_loop(self) -> None:
+        process = self.process
+        assert process is not None and process.stdout is not None
+        failure: str | None = None
+        try:
+            while True:
+                # readline(size) 在换行缺失时仍有硬上限，不能先把任意大的
+                # 第三方 stdout 行完整读入宿主内存再检查。
+                raw_line = process.stdout.readline(MAX_MCP_MESSAGE_BYTES + 1)
+                if raw_line == "":
+                    break
+                if not raw_line.endswith("\n"):
+                    failure = "MCP server emitted an oversized or unterminated message."
+                    break
+                if len(raw_line.encode("utf-8")) > MAX_MCP_MESSAGE_BYTES:
+                    failure = "MCP server emitted an oversized protocol message."
+                    break
+                try:
+                    message = json.loads(raw_line)
+                except json.JSONDecodeError:
+                    failure = "MCP server emitted invalid JSON on stdout."
+                    break
+                if not isinstance(message, dict) or message.get("jsonrpc") != "2.0":
+                    failure = "MCP server emitted an invalid JSON-RPC message."
+                    break
+                self.on_seen()
+                if "id" in message and ("result" in message or "error" in message):
+                    request_id = message.get("id")
+                    if isinstance(request_id, int):
+                        with self._pending_lock:
+                            pending = self._pending.pop(request_id, None)
+                        if pending:
+                            pending.response.put(message)
+                    continue
+                method = message.get("method")
+                if method == "notifications/tools/list_changed":
+                    self.on_tools_changed()
+                elif isinstance(method, str) and "id" in message:
+                    self._send(
+                        {
+                            "jsonrpc": "2.0",
+                            "id": message["id"],
+                            "error": {"code": -32601, "message": "Method not supported."},
+                        }
+                    )
+        except (McpBridgeError, OSError, ValueError) as exc:
+            failure = f"MCP stdout closed unexpectedly: {type(exc).__name__}."
+        finally:
+            if failure and process.poll() is None:
+                process.terminate()
+            exit_code = process.poll()
+            if exit_code is None:
+                try:
+                    exit_code = process.wait(timeout=1)
+                except subprocess.TimeoutExpired:
+                    exit_code = None
+            if not self._stopping:
+                message = failure or f"MCP host exited unexpectedly with code {exit_code}."
+                error = McpBridgeError(
+                    "PLUGIN_HOST_UNAVAILABLE", message, status_code=503
+                )
+                self._fail_pending(error)
+                self.on_broken(message)
+
+    def _stderr_loop(self) -> None:
+        process = self.process
+        assert process is not None and process.stderr is not None
+        try:
+            while True:
+                # stderr 不是协议通道，但同样按块读取，避免无换行日志造成
+                # 宿主侧的无界字符串分配。
+                line = process.stderr.readline(1025)
+                if line == "":
+                    break
+                self._stderr_tail.append(line.rstrip()[:1024])
+        except (OSError, ValueError):
+            return
+
+    def _ensure_running(self) -> None:
+        if self.process is None or self.process.poll() is not None:
+            raise McpBridgeError(
+                "PLUGIN_HOST_UNAVAILABLE", "MCP host is not running.", status_code=503
+            )
+
+    def _fail_pending(self, error: BaseException) -> None:
+        with self._pending_lock:
+            pending = list(self._pending.values())
+            self._pending.clear()
+        for item in pending:
+            item.response.put(error)
+
+
+@dataclass(slots=True)
+class _McpHost:
+    backend: PluginBackend
+    client: McpStdioClient
+    status: PluginHostStatus
+
+
+class McpBridge:
+    """管理每个 Plugin 的独立 MCP Client，并执行 Contract 转换。"""
+
+    def __init__(self) -> None:
+        self._hosts: dict[str, _McpHost] = {}
+        self._statuses: dict[str, PluginHostStatus] = {}
+        self._calls: dict[tuple[str, str], int] = {}
+        self._lock = threading.RLock()
+
+    def start(
+        self,
+        plugin_id: str,
+        backend: PluginBackend,
+        package_path: Path,
+        declared_permissions: list[str],
+        on_unavailable: Callable[[str, str], None],
+    ) -> list[McpDiscoveredTool]:
+        if backend.transport != "stdio":
+            raise McpBridgeError(
+                "MCP_CAPABILITY_UNSUPPORTED",
+                "Phase C only supports the MCP stdio transport.",
+                status_code=501,
+            )
+        command = self._resolve_command(package_path, backend)
+        now = datetime.now(timezone.utc)
+        status = PluginHostStatus(
+            plugin_id=plugin_id,
+            backend_type="mcp",
+            transport="stdio",
+            status=PluginHostState.starting,
+            started_at=now,
+            last_seen_at=now,
+        )
+        host_ref: dict[str, _McpHost] = {}
+
+        def seen() -> None:
+            host = host_ref.get("host")
+            if host:
+                host.status.last_seen_at = datetime.now(timezone.utc)
+
+        def broken(message: str) -> None:
+            host = host_ref.get("host")
+            if host:
+                host.status.status = PluginHostState.unhealthy
+                host.status.error = message
+            on_unavailable(plugin_id, message)
+
+        def tools_changed() -> None:
+            broken("MCP tool list changed; restart the Plugin Host to revalidate tools.")
+
+        client = McpStdioClient(
+            command,
+            cwd=package_path,
+            on_seen=seen,
+            on_broken=broken,
+            on_tools_changed=tools_changed,
+        )
+        host = _McpHost(backend=backend, client=client, status=status)
+        host_ref["host"] = host
+        with self._lock:
+            if plugin_id in self._hosts:
+                raise McpBridgeError(
+                    "PLUGIN_HOST_START_FAILED",
+                    f"MCP host is already running: {plugin_id}",
+                    status_code=409,
+                )
+            self._hosts[plugin_id] = host
+            self._statuses[plugin_id] = status
+        try:
+            client.start()
+            initialize = client.request(
+                "initialize",
+                {
+                    "protocolVersion": MCP_PROTOCOL_VERSION,
+                    "capabilities": {},
+                    "clientInfo": {"name": "NotesAgent", "version": "0.1.0"},
+                },
+                timeout=backend.startup_timeout_seconds,
+                timeout_code="MCP_INITIALIZE_FAILED",
+                response_error_code="MCP_INITIALIZE_FAILED",
+            )
+            version = initialize.get("protocolVersion")
+            if version not in SUPPORTED_PROTOCOL_VERSIONS:
+                raise McpBridgeError(
+                    "MCP_INITIALIZE_FAILED",
+                    f"Unsupported MCP protocol version: {version}",
+                )
+            capabilities = initialize.get("capabilities")
+            if not isinstance(capabilities, dict) or not isinstance(
+                capabilities.get("tools"), dict
+            ):
+                raise McpBridgeError(
+                    "MCP_CAPABILITY_UNSUPPORTED",
+                    "MCP server does not declare the tools capability.",
+                )
+            server_info = initialize.get("serverInfo")
+            if not isinstance(server_info, dict):
+                server_info = {}
+            status.protocol_version = str(version)
+            status.server_name = _optional_string(server_info.get("name"))
+            status.server_version = _optional_string(server_info.get("version"))
+            client.notify("notifications/initialized")
+            discovered = self._discover_tools(
+                plugin_id, client, backend, declared_permissions
+            )
+            status.status = PluginHostState.ready
+            status.tools_count = len(discovered)
+            status.last_seen_at = datetime.now(timezone.utc)
+            status.error = None
+            return discovered
+        except McpBridgeError as exc:
+            status.status = PluginHostState.error
+            status.error = exc.message
+            client.stop()
+            with self._lock:
+                self._hosts.pop(plugin_id, None)
+            raise
+        except Exception as exc:
+            status.status = PluginHostState.error
+            status.error = f"MCP initialization failed: {type(exc).__name__}."
+            client.stop()
+            with self._lock:
+                self._hosts.pop(plugin_id, None)
+            raise McpBridgeError("MCP_INITIALIZE_FAILED", status.error) from exc
+
+    async def call_tool(
+        self,
+        plugin_id: str,
+        remote_name: str,
+        arguments: dict[str, Any],
+        *,
+        request_id: str,
+    ) -> Any:
+        host = self._host(plugin_id)
+        rpc_id, pending = host.client.begin_request(
+            "tools/call", {"name": remote_name, "arguments": arguments}
+        )
+        call_key = (plugin_id, request_id)
+        with self._lock:
+            self._calls[call_key] = rpc_id
+        try:
+            result = await asyncio.to_thread(
+                host.client.wait_response,
+                rpc_id,
+                pending,
+                timeout=host.backend.tool_timeout_seconds,
+                timeout_code="MCP_TOOL_CALL_FAILED",
+            )
+        except asyncio.CancelledError:
+            host.client.cancel(rpc_id)
+            host.client.abandon(
+                rpc_id,
+                McpBridgeError(
+                    "MCP_TOOL_CALL_FAILED", "MCP request was cancelled."
+                ),
+            )
+            raise
+        except McpBridgeError as exc:
+            raise ToolExecutionError(exc.code, exc.message) from exc
+        finally:
+            with self._lock:
+                self._calls.pop(call_key, None)
+
+        encoded_size = len(
+            json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+        )
+        if encoded_size > MAX_MCP_TOOL_RESULT_BYTES:
+            raise ToolExecutionError(
+                "MCP_TOOL_RESULT_TOO_LARGE",
+                "MCP tool result exceeds the configured size limit.",
+            )
+        if result.get("isError") is True:
+            raise ToolExecutionError(
+                "MCP_TOOL_CALL_FAILED", _mcp_error_message(result.get("content"))
+            )
+        structured = result.get("structuredContent")
+        if structured is not None:
+            if not isinstance(structured, dict):
+                raise ToolExecutionError(
+                    "MCP_TOOL_CALL_FAILED",
+                    "MCP structuredContent must be an object.",
+                )
+            return structured
+        content = result.get("content", [])
+        if not isinstance(content, list):
+            raise ToolExecutionError(
+                "MCP_TOOL_CALL_FAILED", "MCP tool content must be an array."
+            )
+        return {"content": content}
+
+    def cancel(self, plugin_id: str, request_id: str) -> None:
+        with self._lock:
+            rpc_id = self._calls.get((plugin_id, request_id))
+            host = self._hosts.get(plugin_id)
+        if rpc_id is not None and host is not None:
+            host.client.cancel(rpc_id)
+
+    def stop(self, plugin_id: str) -> None:
+        with self._lock:
+            host = self._hosts.pop(plugin_id, None)
+        if host:
+            host.client.stop()
+            host.status.status = PluginHostState.stopped
+            host.status.tools_count = 0
+            host.status.error = None
+
+    def remove(self, plugin_id: str) -> None:
+        """停止 Host，并清除卸载后不应跨安装保留的状态与调用索引。"""
+
+        self.stop(plugin_id)
+        with self._lock:
+            self._statuses.pop(plugin_id, None)
+            stale_calls = [key for key in self._calls if key[0] == plugin_id]
+            for key in stale_calls:
+                self._calls.pop(key, None)
+
+    def status(self, plugin_id: str, backend: PluginBackend) -> PluginHostStatus:
+        with self._lock:
+            status = self._statuses.get(plugin_id)
+            if status:
+                return status.model_copy(deep=True)
+        return PluginHostStatus(
+            plugin_id=plugin_id,
+            backend_type=backend.type,
+            transport=backend.transport,
+            status=PluginHostState.stopped,
+        )
+
+    def _discover_tools(
+        self,
+        plugin_id: str,
+        client: McpStdioClient,
+        backend: PluginBackend,
+        declared_permissions: list[str],
+    ) -> list[McpDiscoveredTool]:
+        discovered: list[McpDiscoveredTool] = []
+        cursor: str | None = None
+        for _ in range(MAX_MCP_LIST_PAGES):
+            params = {"cursor": cursor} if cursor else {}
+            result = client.request(
+                "tools/list",
+                params,
+                timeout=backend.startup_timeout_seconds,
+                timeout_code="MCP_INITIALIZE_FAILED",
+                response_error_code="MCP_INITIALIZE_FAILED",
+            )
+            raw_tools = result.get("tools")
+            if not isinstance(raw_tools, list):
+                raise McpBridgeError(
+                    "MCP_TOOL_SCHEMA_INVALID", "MCP tools/list must return a tools array."
+                )
+            for raw in raw_tools:
+                discovered.append(
+                    self._map_tool(plugin_id, raw, declared_permissions)
+                )
+                if len(discovered) > MAX_MCP_TOOLS:
+                    raise McpBridgeError(
+                        "MCP_TOOL_SCHEMA_INVALID",
+                        f"MCP server exposes more than {MAX_MCP_TOOLS} tools.",
+                    )
+            next_cursor = result.get("nextCursor")
+            if next_cursor is None:
+                break
+            if not isinstance(next_cursor, str) or not next_cursor:
+                raise McpBridgeError(
+                    "MCP_TOOL_SCHEMA_INVALID", "MCP nextCursor must be a non-empty string."
+                )
+            cursor = next_cursor
+        else:
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID", "MCP tools/list exceeded the page limit."
+            )
+        names = [item.definition.name for item in discovered]
+        if len(names) != len(set(names)):
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID", "MCP server returned duplicate tool names."
+            )
+        return discovered
+
+    @staticmethod
+    def _map_tool(
+        plugin_id: str, raw: Any, declared_permissions: list[str]
+    ) -> McpDiscoveredTool:
+        if not isinstance(raw, dict):
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID", "MCP tool definition must be an object."
+            )
+        remote_name = raw.get("name")
+        if not isinstance(remote_name, str) or not remote_name:
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID", "MCP tool name must be a non-empty string."
+            )
+        if (
+            len(remote_name) > 128
+            or not remote_name[0].isalnum()
+            or not all(
+                character.islower()
+                or character.isdigit()
+                or character in "._-"
+                for character in remote_name
+            )
+        ):
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID",
+                f"MCP tool name is not a valid NotesAgent id: {remote_name}",
+            )
+        schema = raw.get("inputSchema", {"type": "object", "properties": {}})
+        if not isinstance(schema, dict) or schema.get("type", "object") != "object":
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID",
+                f"MCP tool inputSchema must be an object schema: {remote_name}",
+            )
+        try:
+            Draft202012Validator.check_schema(schema)
+        except SchemaError as exc:
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID",
+                f"Invalid MCP tool schema for {remote_name}: {exc.message}",
+            ) from exc
+        metadata = raw.get("_meta")
+        permission = (
+            metadata.get("notesagent/permission") if isinstance(metadata, dict) else None
+        )
+        if permission is not None and (
+            not isinstance(permission, str) or permission not in KNOWN_PERMISSIONS
+        ):
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID",
+                f"MCP tool declares an unknown permission: {remote_name}",
+            )
+        if permission and permission not in declared_permissions:
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID",
+                f"MCP tool permission is missing from Plugin manifest: {permission}",
+            )
+        description = raw.get("description")
+        return McpDiscoveredTool(
+            remote_name=remote_name,
+            definition=ToolDefinition(
+                name=f"{plugin_id}.{remote_name}",
+                description=description if isinstance(description, str) else remote_name,
+                parameters=schema,
+                permission=permission,
+                source="plugin",
+            ),
+        )
+
+    def _host(self, plugin_id: str) -> _McpHost:
+        with self._lock:
+            host = self._hosts.get(plugin_id)
+        if host is None or host.status.status != PluginHostState.ready:
+            raise ToolExecutionError(
+                "PLUGIN_HOST_UNAVAILABLE", f"MCP Plugin Host is not ready: {plugin_id}"
+            )
+        return host
+
+    @staticmethod
+    def _resolve_command(root: Path, backend: PluginBackend) -> list[str]:
+        if not backend.command or not backend.command.strip():
+            raise McpBridgeError(
+                "PLUGIN_HOST_START_FAILED", "MCP stdio backend requires a command."
+            )
+        command = backend.command.strip()
+        if Path(command).is_absolute() or "/" in command or "\\" in command:
+            executable = (
+                (root / command).resolve()
+                if not Path(command).is_absolute()
+                else Path(command).resolve()
+            )
+            try:
+                executable.relative_to(root)
+            except ValueError as exc:
+                raise McpBridgeError(
+                    "PLUGIN_HOST_START_FAILED",
+                    "MCP executable path must stay inside the Plugin package.",
+                ) from exc
+            command = str(executable)
+        return [command, *backend.args]
+
+
+def _mcp_error_message(content: Any) -> str:
+    if isinstance(content, list):
+        texts = [
+            item.get("text")
+            for item in content
+            if isinstance(item, dict)
+            and item.get("type") == "text"
+            and isinstance(item.get("text"), str)
+        ]
+        if texts:
+            return "\n".join(texts)[:4096]
+    return "MCP tool returned an error result."
+
+
+def _optional_string(value: Any) -> str | None:
+    return value if isinstance(value, str) else None
+
+
+def _subprocess_environment() -> dict[str, str]:
+    """只传递启动进程所需的系统变量，隔离 Provider Key、Vault 路径等宿主状态。"""
+
+    allowed = {
+        "PATH",
+        "PATHEXT",
+        "SYSTEMROOT",
+        "WINDIR",
+        "COMSPEC",
+        "TEMP",
+        "TMP",
+        "TMPDIR",
+        "LANG",
+        "LC_ALL",
+        "VIRTUAL_ENV",
+    }
+    environment = {
+        key: value for key, value in os.environ.items() if key.upper() in allowed
+    }
+    environment["PYTHONUNBUFFERED"] = "1"
+    environment["PYTHONIOENCODING"] = "utf-8"
+    return environment

backend/app/extensions/runtime.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+import threading
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Literal
@@ -16,6 +17,7 @@ from app.contracts import (
     ModelCapability,
     Plugin,
     PluginManifest,
+    PluginHostStatus,
     PluginStatus,
     RetrievalConfig,
     Skill,
@@ -23,6 +25,7 @@ from app.contracts import (
     SkillStatus,
     ToolDefinition,
 )
+from app.extensions.mcp import McpBridge, McpBridgeError, McpDiscoveredTool
 
 _EXTENSION_ID = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
 
@@ -242,18 +245,29 @@ class _PluginRecord:
     tools: list[DeclarativeToolSpec]
     package_path: Path
     registered_tools: list[str]
+    mcp_remote_names: dict[str, str]
 
 
 class PluginRuntime:
     """Plugin Manifest、生命周期及 Tool Contribution 注册。"""
 
-    def __init__(self, tools: ToolRegistry, host: DeclarativePluginHost | None = None) -> None:
+    def __init__(
+        self,
+        tools: ToolRegistry,
+        host: DeclarativePluginHost | None = None,
+        mcp_bridge: McpBridge | None = None,
+        *,
+        allow_unsandboxed_mcp: bool = False,
+    ) -> None:
         self.registry = tools
         self.host = host or DeclarativePluginHost()
+        self.mcp = mcp_bridge or McpBridge()
+        self.allow_unsandboxed_mcp = allow_unsandboxed_mcp
         self._records: dict[str, _PluginRecord] = {}
+        self._lock = threading.RLock()
 
     def install(self, package_path: str | Path) -> Plugin:
-        # 当前只加载声明式清单，不导入或执行插件包中的任意 Python 代码。
+        # 安装阶段只读取清单；MCP 子进程必须在权限授予后的 enable 阶段启动。
         root = _package_dir(package_path)
         raw = _read_yaml(root / "plugin.yaml")
         if "id" in raw and "plugin_id" not in raw:
@@ -271,15 +285,17 @@ class PluginRuntime:
                 status_code=409,
             )
 
-        specs = self._load_tools(root)
-        declared = set(manifest.contributes.tools)
-        actual = {spec.name for spec in specs}
-        if declared != actual:
-            raise ExtensionError(
-                "PLUGIN_CONTRIBUTION_INVALID",
-                "plugin.yaml tool contributions must exactly match tools.yaml",
-                details={"declared": sorted(declared), "actual": sorted(actual)},
-            )
+        _validate_backend(manifest)
+        specs = [] if manifest.backend.type == "mcp" else self._load_tools(root)
+        if manifest.backend.type != "mcp":
+            declared = set(manifest.contributes.tools)
+            actual = {spec.name for spec in specs}
+            if declared != actual:
+                raise ExtensionError(
+                    "PLUGIN_CONTRIBUTION_INVALID",
+                    "plugin.yaml tool contributions must exactly match tools.yaml",
+                    details={"declared": sorted(declared), "actual": sorted(actual)},
+                )
         for spec in specs:
             _validate_id("tool", spec.name)
             _validate_tool_schema(spec)
@@ -302,6 +318,7 @@ class PluginRuntime:
             tools=specs,
             package_path=root,
             registered_tools=[],
+            mcp_remote_names={},
         )
         self._records[manifest.plugin_id] = record
         return record.plugin.model_copy(deep=True)
@@ -313,18 +330,14 @@ class PluginRuntime:
         return self._record(plugin_id).plugin.model_copy(deep=True)
 
     def enable(self, plugin_id: str) -> Plugin:
+        # Host 启动和 Tool 批量注册必须串行，避免并发 enable 产生重复进程或半注册状态。
+        with self._lock:
+            return self._enable(plugin_id)
+
+    def _enable(self, plugin_id: str) -> Plugin:
         record = self._record(plugin_id)
         if record.plugin.enabled:
             return record.plugin.model_copy(deep=True)
-        if record.plugin.manifest.backend.type == "mcp":
-            # TODO(extension): 第二阶段以隔离进程实现 MCP Host，并补充签名与来源校验。
-            record.plugin.status = PluginStatus.dependency_missing
-            raise ExtensionError(
-                "PLUGIN_HOST_UNAVAILABLE",
-                "MCP Plugin Host is reserved for the second development phase.",
-                status_code=501,
-                details={"plugin_id": plugin_id, "backend": "mcp"},
-            )
         missing_grants = sorted(
             set(record.plugin.manifest.permissions) - set(record.plugin.granted_permissions)
         )
@@ -336,7 +349,18 @@ class PluginRuntime:
                 status_code=409,
                 details={"plugin_id": plugin_id, "permissions": missing_grants},
             )
-        conflicts = [spec.name for spec in record.tools if self.registry.contains(spec.name)]
+        if (
+            record.plugin.manifest.backend.type == "mcp"
+            and not self.allow_unsandboxed_mcp
+        ):
+            raise ExtensionError(
+                "MCP_TRUST_APPROVAL_REQUIRED",
+                "Unsandboxed MCP Hosts are disabled outside development mode.",
+                status_code=403,
+                details={"plugin_id": plugin_id},
+            )
+        declared_tools = list(record.plugin.manifest.contributes.tools)
+        conflicts = [name for name in declared_tools if self.registry.contains(name)]
         if conflicts:
             raise ExtensionError(
                 "PLUGIN_TOOL_CONFLICT",
@@ -346,42 +370,75 @@ class PluginRuntime:
             )
         record.plugin.status = PluginStatus.starting
         try:
-            for spec in record.tools:
-                arguments_model = _arguments_model(spec)
+            if record.plugin.manifest.backend.type == "mcp":
+                discovered = self._start_mcp(record)
+                actual = {item.definition.name for item in discovered}
+                declared = set(declared_tools)
+                if actual != declared:
+                    raise ExtensionError(
+                        "PLUGIN_CONTRIBUTION_INVALID",
+                        "Discovered MCP tools must exactly match Plugin contributions.",
+                        details={"declared": sorted(declared), "actual": sorted(actual)},
+                    )
+                for item in discovered:
+                    self._register_mcp_tool(record, item)
+            else:
+                for spec in record.tools:
+                    arguments_model = _arguments_model(spec)
 
-                async def executor(
-                    arguments: BaseModel,
-                    context: ToolExecutionContext,
-                    _handler: str = spec.handler,
-                ) -> Any:
-                    return await self.host.execute(_handler, arguments, context)
+                    async def executor(
+                        arguments: BaseModel,
+                        context: ToolExecutionContext,
+                        _handler: str = spec.handler,
+                    ) -> Any:
+                        return await self.host.execute(_handler, arguments, context)
 
-                self.registry.register(
-                    ToolDefinition(
-                        name=spec.name,
-                        description=spec.description,
-                        parameters=spec.parameters,
-                        permission=spec.permission,
-                        source="plugin",
-                    ),
-                    arguments_model,
-                    executor,
-                )
-                record.registered_tools.append(spec.name)
+                    self.registry.register(
+                        ToolDefinition(
+                            name=spec.name,
+                            description=spec.description,
+                            parameters=spec.parameters,
+                            permission=spec.permission,
+                            source="plugin",
+                        ),
+                        arguments_model,
+                        executor,
+                    )
+                    record.registered_tools.append(spec.name)
         except Exception as exc:
             # 注册过程必须具备回滚语义，防止半启用插件污染全局工具表。
             for name in record.registered_tools:
                 self.registry.unregister(name)
             record.registered_tools.clear()
+            record.mcp_remote_names.clear()
+            self.mcp.stop(plugin_id)
             record.plugin.status = PluginStatus.error
-            record.plugin.error_message = str(exc)
-            raise
+            record.plugin.error_message = _safe_extension_message(exc)
+            if isinstance(exc, ExtensionError):
+                raise
+            if isinstance(exc, McpBridgeError):
+                raise ExtensionError(
+                    exc.code,
+                    exc.message,
+                    status_code=exc.status_code,
+                    details={"plugin_id": plugin_id},
+                ) from exc
+            raise ExtensionError(
+                "PLUGIN_HOST_START_FAILED",
+                record.plugin.error_message,
+                status_code=503,
+                details={"plugin_id": plugin_id},
+            ) from exc
         record.plugin.enabled = True
         record.plugin.status = PluginStatus.ready
         record.plugin.error_message = None
         return record.plugin.model_copy(deep=True)
 
     def set_permissions(self, plugin_id: str, permissions: list[str]) -> Plugin:
+        with self._lock:
+            return self._set_permissions(plugin_id, permissions)
+
+    def _set_permissions(self, plugin_id: str, permissions: list[str]) -> Plugin:
         record = self._record(plugin_id)
         requested = set(permissions)
         declared = set(record.plugin.manifest.permissions)
@@ -403,15 +460,125 @@ class PluginRuntime:
         return record.plugin.model_copy(deep=True)
 
     def disable(self, plugin_id: str) -> Plugin:
+        with self._lock:
+            return self._disable(plugin_id)
+
+    def _disable(self, plugin_id: str) -> Plugin:
         record = self._record(plugin_id)
         for name in record.registered_tools:
             self.registry.unregister(name)
         record.registered_tools.clear()
+        record.mcp_remote_names.clear()
+        if record.plugin.manifest.backend.type == "mcp":
+            self.mcp.stop(plugin_id)
         record.plugin.enabled = False
         record.plugin.status = PluginStatus.disabled
         return record.plugin.model_copy(deep=True)
 
+    def get_host_status(self, plugin_id: str) -> PluginHostStatus:
+        record = self._record(plugin_id)
+        return self.mcp.status(plugin_id, record.plugin.manifest.backend)
+
+    def restart_host(self, plugin_id: str) -> PluginHostStatus:
+        with self._lock:
+            return self._restart_host(plugin_id)
+
+    def _restart_host(self, plugin_id: str) -> PluginHostStatus:
+        record = self._record(plugin_id)
+        if record.plugin.manifest.backend.type != "mcp":
+            raise ExtensionError(
+                "PLUGIN_HOST_UNAVAILABLE",
+                "Plugin does not use an MCP Host.",
+                status_code=409,
+                details={"plugin_id": plugin_id},
+            )
+        if record.plugin.status in {
+            PluginStatus.installed,
+            PluginStatus.disabled,
+            PluginStatus.permission_required,
+        }:
+            raise ExtensionError(
+                "PLUGIN_HOST_UNAVAILABLE",
+                "Disabled or inactive MCP Plugins must be started with Enable.",
+                status_code=409,
+                details={"plugin_id": plugin_id, "status": record.plugin.status.value},
+            )
+        for name in record.registered_tools:
+            self.registry.unregister(name)
+        record.registered_tools.clear()
+        record.mcp_remote_names.clear()
+        self.mcp.stop(plugin_id)
+        record.plugin.enabled = False
+        record.plugin.status = PluginStatus.installed
+        record.plugin.error_message = None
+        self.enable(plugin_id)
+        return self.get_host_status(plugin_id)
+
+    def shutdown(self) -> None:
+        """关闭所有隔离 Host；用于 FastAPI lifespan 和测试清理。"""
+
+        with self._lock:
+            for plugin_id, record in list(self._records.items()):
+                if record.plugin.manifest.backend.type == "mcp":
+                    self.mcp.stop(plugin_id)
+
+    def _start_mcp(self, record: _PluginRecord) -> list[McpDiscoveredTool]:
+        manifest = record.plugin.manifest
+        return self.mcp.start(
+            manifest.plugin_id,
+            manifest.backend,
+            record.package_path,
+            manifest.permissions,
+            self._handle_mcp_unavailable,
+        )
+
+    def _register_mcp_tool(
+        self, record: _PluginRecord, discovered: McpDiscoveredTool
+    ) -> None:
+        definition = discovered.definition
+        arguments_model = _arguments_model_from_schema(
+            definition.name, definition.parameters
+        )
+        plugin_id = record.plugin.manifest.plugin_id
+        remote_name = discovered.remote_name
+
+        async def executor(
+            arguments: BaseModel,
+            context: ToolExecutionContext,
+        ) -> Any:
+            return await self.mcp.call_tool(
+                plugin_id,
+                remote_name,
+                # 省略的可选字段不能被补成 null；显式传入的 null 仍由
+                # model_fields_set 保留并交给 MCP Server。
+                arguments.model_dump(exclude_unset=True),
+                request_id=context.tool_call_id or f"{context.run_id}:{definition.name}",
+            )
+
+        self.registry.register(definition, arguments_model, executor)
+        record.registered_tools.append(definition.name)
+        record.mcp_remote_names[definition.name] = remote_name
+
+    def _handle_mcp_unavailable(self, plugin_id: str, message: str) -> None:
+        with self._lock:
+            record = self._records.get(plugin_id)
+            if record is None:
+                return
+            for name in record.registered_tools:
+                self.registry.unregister(name)
+            record.registered_tools.clear()
+            record.mcp_remote_names.clear()
+            record.plugin.enabled = False
+            record.plugin.status = PluginStatus.error
+            record.plugin.error_message = message
+
     def uninstall(self, plugin_id: str, dependent_skills: list[str] | None = None) -> None:
+        with self._lock:
+            self._uninstall(plugin_id, dependent_skills)
+
+    def _uninstall(
+        self, plugin_id: str, dependent_skills: list[str] | None = None
+    ) -> None:
         record = self._record(plugin_id)
         if dependent_skills:
             raise ExtensionError(
@@ -420,8 +587,13 @@ class PluginRuntime:
                 status_code=409,
                 details={"plugin_id": plugin_id, "skills": dependent_skills},
             )
+        is_mcp = record.plugin.manifest.backend.type == "mcp"
         if record.plugin.enabled:
             self.disable(plugin_id)
+        if is_mcp:
+            # stop 只结束本次进程并保留状态供故障诊断；真正卸载时必须连同
+            # 历史状态一起遗忘，避免同 ID 重装继承旧协商信息。
+            self.mcp.remove(plugin_id)
         del self._records[plugin_id]
 
     def _record(self, plugin_id: str) -> _PluginRecord:
@@ -498,24 +670,18 @@ def _manifest_error(kind: str, exc: ValidationError) -> ExtensionError:
 
 def _arguments_model(spec: DeclarativeToolSpec) -> type[BaseModel]:
     schema = spec.parameters or {"type": "object", "properties": {}}
+    return _arguments_model_from_schema(spec.name, schema)
+
+
+def _arguments_model_from_schema(
+    tool_name: str, schema: dict[str, Any]
+) -> type[BaseModel]:
     if schema.get("type", "object") != "object":
         raise ExtensionError("PLUGIN_TOOL_SCHEMA_INVALID", "Tool parameters must be an object schema.")
-    properties = schema.get("properties", {})
-    required = set(schema.get("required", []))
-    fields: dict[str, tuple[Any, Any]] = {}
-    types = {
-        "string": str,
-        "number": float,
-        "integer": int,
-        "boolean": bool,
-        "array": list[Any],
-        "object": dict[str, Any],
-    }
-    for name, field_schema in properties.items():
-        annotation = types.get(field_schema.get("type"), Any)
-        fields[name] = (annotation, ... if name in required else None)
-    model_name = "PluginArgs_" + re.sub(r"\W+", "_", spec.name)
-    return create_model(model_name, __config__=ConfigDict(extra="forbid"), **fields)
+    model_name = "PluginArgs_" + re.sub(r"\W+", "_", tool_name)
+    # 完整 JSON Schema 已在 ToolRegistry 中先行校验。参数载体不重复声明字段，
+    # 从而完整保留 model_dump、连字符键、联合类型和动态属性等合法 JSON 键值。
+    return create_model(model_name, __config__=ConfigDict(extra="allow"))
 
 
 def _validate_tool_schema(spec: DeclarativeToolSpec) -> None:
@@ -536,3 +702,30 @@ def _validate_tool_schema(spec: DeclarativeToolSpec) -> None:
             "Tool parameters must be an object schema with object properties.",
             details={"tool": spec.name},
         )
+
+
+def _validate_backend(manifest: PluginManifest) -> None:
+    backend = manifest.backend
+    if backend.type == "mcp":
+        if backend.transport != "stdio":
+            raise ExtensionError(
+                "MCP_CAPABILITY_UNSUPPORTED",
+                "Phase C MCP Plugins must use stdio transport.",
+                status_code=501,
+            )
+        if not backend.command or not backend.command.strip():
+            raise ExtensionError(
+                "EXTENSION_MANIFEST_INVALID",
+                "MCP stdio backend requires a command.",
+            )
+    elif backend.command is not None or backend.args:
+        raise ExtensionError(
+            "EXTENSION_MANIFEST_INVALID",
+            "Only MCP stdio backends may declare command or args.",
+        )
+
+
+def _safe_extension_message(exc: Exception) -> str:
+    if isinstance(exc, (ExtensionError, McpBridgeError)):
+        return exc.message
+    return f"Plugin Host operation failed: {type(exc).__name__}."

backend/app/main.py

@@ -1,19 +1,31 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
 from starlette.exceptions import HTTPException as StarletteHttpException
 
 from app.config import get_settings
+from app.container import container
 from app.errors import ApiError, api_error_handler, http_error_handler, validation_error_handler
 from app.routes import router as api_router
 from app.schemas import HealthResponse, ServiceStatusResponse
 
 settings = get_settings()
 
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    yield
+    # 第三方 MCP Server 必须跟随 AI Core 退出，不能遗留孤儿进程。
+    container.plugins.shutdown()
+
+
 app = FastAPI(
     title=settings.name,
     version=settings.version,
     description="AI 笔记软件的本地 AI Core 与 Agent Core 服务。",
+    lifespan=lifespan,
 )
 
 app.add_middleware(

backend/app/retrieval/embedding.py

@@ -19,6 +19,7 @@ class EmbeddingProvider(Protocol):
     """统一 Embedding 接口（与文档一致）。"""
 
     model_id: str
+    version: str
     dim: int
 
     async def embed_documents(self, texts: list[str]) -> list[list[float]]: ...
@@ -33,6 +34,7 @@ class HashEmbeddingProvider:
     """
 
     model_id = "hash-v1"
+    version = "1"
     dim = EMBEDDING_DIM
 
     async def embed_documents(self, texts: list[str]) -> list[list[float]]:

backend/app/retrieval/engine.py

@@ -29,6 +29,8 @@ from app.textutils import make_snippet, match_query
 CANDIDATE_POOL = 50
 # 分页窗口上限：候选池至少覆盖 offset+limit，但设上限防止超大 offset 撑爆内存
 MAX_CANDIDATE_POOL = 200
+# FTS 全量取回上限：统一归一化 + 阈值过滤后再分页，保证阈值语义跨页一致
+FTS_FETCH_LIMIT = 5000
 # 带 metadata 过滤时放大召回倍数，缓解「先截断候选池再过滤」造成的漏召回
 OVERSCAN_FACTOR = 4
 
@@ -84,7 +86,7 @@ class RetrievalEngine:
         elif request.mode == SearchMode.vector:
             candidate_scores = vec_scores
         else:  # hybrid：RRF 融合
-            candidate_scores = rrf_fuse([fts_ranked, vec_ranked])
+            candidate_scores = rrf_fuse([fts_ranked, vec_ranked], k=request.rrf_k)
 
         if not candidate_scores:
             return self._empty(request)
@@ -97,14 +99,23 @@ class RetrievalEngine:
         if not filtered:
             return self._empty(request)
 
-        # 4. 排序 / 精排
+        # 4. 排序 / 精排：hybrid 先按融合分预排序，再对前 rerank_candidates 个候选做精排，
+        #    剩余候选按融合分排在精排结果之后；rerank=False 时跳过精排直接按融合分排序。
         if request.mode == SearchMode.hybrid:
-            candidates = [
-                RankedCandidate(block_id=h.block_id, score=candidate_scores[h.block_id], text=h.content)
-                for h in filtered
-            ]
-            ranked = await self.reranker.rerank(request.query, candidates)
-            ordered = [(c.block_id, c.score) for c in ranked]
+            pre_sorted = sorted(filtered, key=lambda h: -candidate_scores[h.block_id])
+            if request.rerank:
+                limit = request.rerank_candidates
+                pool = pre_sorted if limit is None else pre_sorted[:limit]
+                rest = [] if limit is None else pre_sorted[limit:]
+                candidates = [
+                    RankedCandidate(block_id=h.block_id, score=candidate_scores[h.block_id], text=h.content)
+                    for h in pool
+                ]
+                ranked = await self.reranker.rerank(request.query, candidates)
+                ordered = [(c.block_id, c.score) for c in ranked]
+                ordered += [(h.block_id, candidate_scores[h.block_id]) for h in rest]
+            else:
+                ordered = [(h.block_id, candidate_scores[h.block_id]) for h in pre_sorted]
         else:
             ordered = sorted(
                 ((h.block_id, candidate_scores[h.block_id]) for h in filtered),
@@ -112,6 +123,8 @@ class RetrievalEngine:
             )
 
         ordered = normalize_scores(ordered)
+        # score_threshold：归一化后过滤低分结果（默认 0 不过滤）
+        ordered = [(bid, score) for bid, score in ordered if score >= request.score_threshold]
 
         # 5. 分页：total = 过滤后候选集大小。fts 已取全量（≤FTS_FETCH_LIMIT）故为真实命中数；
         #    vector/hybrid 为 KNN 候选集，无全局 total。
@@ -126,15 +139,18 @@ class RetrievalEngine:
         )
 
     def _search_fts(self, request: SearchRequest) -> SearchResponse:
-        """FTS 专用路径：过滤、COUNT 与分页全部在 SQLite 中完成。"""
+        """FTS 专用路径：先取全量命中（≤FTS_FETCH_LIMIT），统一归一化 + 阈值过滤后再分页。
+
+        阈值过滤必须在计数与分页之前完成，否则 score_threshold 只作用于当前页，
+        且返回的 total 与 items 数量不一致（如 items 为空但 total 非零）。"""
         match = match_query(request.query)
         if not match:
             return self._empty(request)
 
-        fts_hits, total = repository.fts_search_page(
+        fts_hits, _ = repository.fts_search_page(
             match=match,
-            limit=request.limit,
-            offset=request.offset,
+            limit=FTS_FETCH_LIMIT,
+            offset=0,
             folders=request.folders,
             note_ids=request.note_ids,
             tags=request.tags,
@@ -144,17 +160,18 @@ class RetrievalEngine:
             updated_to=request.updated_to,
         )
         if not fts_hits:
-            return SearchResponse(
-                query=request.query,
-                mode=request.mode,
-                page=PageMeta(total=total, limit=request.limit, offset=request.offset),
-            )
+            return self._empty(request)
 
-        hits = {h.block_id: h for h in repository.get_block_hits([hit.block_id for hit in fts_hits])}
-        ordered = normalize_scores(
-            [(hit.block_id, -hit.bm25) for hit in fts_hits if hit.block_id in hits]
-        )
-        items = [self._build_result(hits[block_id], request, score) for block_id, score in ordered]
+        ordered = normalize_scores([(hit.block_id, -hit.bm25) for hit in fts_hits])
+        ordered = [(bid, score) for bid, score in ordered if score >= request.score_threshold]
+        total = len(ordered)
+        page = ordered[request.offset : request.offset + request.limit]
+        hits = {h.block_id: h for h in repository.get_block_hits([bid for bid, _ in page])}
+        items = [
+            self._build_result(hits[block_id], request, score)
+            for block_id, score in page
+            if block_id in hits
+        ]
         return SearchResponse(
             query=request.query,
             mode=request.mode,

backend/app/retrieval/reranker.py

@@ -24,6 +24,7 @@ class RerankerProvider(Protocol):
     """统一 Reranker 接口：输入候选块，输出按相关性重排后的候选块。"""
 
     model_id: str
+    version: str
 
     async def rerank(self, query: str, candidates: list[RankedCandidate]) -> list[RankedCandidate]: ...
 
@@ -32,6 +33,7 @@ class LexicalReranker:
     """轻量精排：query 与块正文的词面重叠度，与归一化后的原始分数加权求和。"""
 
     model_id = "lexical-v1"
+    version = "1"
 
     def __init__(self, lexical_weight: float = 0.5) -> None:
         self.lexical_weight = lexical_weight

backend/app/retrieval/vectorstore.py

@@ -35,6 +35,7 @@ class VectorStore(Protocol):
     async def upsert(self, records: list[VectorRecord]) -> None: ...
     async def delete(self, ids: list[str]) -> None: ...
     async def search(self, vector: list[float], *, top_k: int) -> list[VectorHit]: ...
+    async def count(self) -> int: ...
 
 
 class SqliteVecStore:
@@ -92,3 +93,10 @@ class SqliteVecStore:
                 conn.execute("DELETE FROM vec_blocks")
         finally:
             conn.close()
+
+    async def count(self) -> int:
+        conn = connect()
+        try:
+            return conn.execute("SELECT COUNT(*) FROM vec_blocks").fetchone()[0]
+        finally:
+            conn.close()

backend/app/routes.py

@@ -1,3 +1,4 @@
+import asyncio
 from collections.abc import AsyncIterator
 from datetime import datetime, timezone
 from uuid import uuid4
@@ -11,6 +12,14 @@ from app.contracts import (
     AgentRunListResponse,
     AgentTraceResponse,
     ChatRequest,
+    BenchmarkDatasetListResponse,
+    BenchmarkEventType,
+    BenchmarkKind,
+    BenchmarkReport,
+    BenchmarkRun,
+    BenchmarkRunListResponse,
+    BenchmarkStatus,
+    RAGRunRequest,
     CredentialStatus,
     CredentialWriteRequest,
     ExtensionInstallRequest,
@@ -32,6 +41,7 @@ from app.contracts import (
     PageMeta,
     PermissionDecisionRequest,
     Plugin,
+    PluginHostStatus,
     PluginListResponse,
     PluginPermissionGrantRequest,
     ProviderConfig,
@@ -59,6 +69,8 @@ from app.contracts import (
     WorkspaceSnapshot,
 )
 from app.agent import AgentCapacityError, AgentRunNotFoundError
+from app.benchmarks import datasets as benchmark_datasets
+from app.benchmarks import service as benchmark_service
 from app.container import container
 from app.errors import ApiError
 from app.extensions import ExtensionError
@@ -130,6 +142,15 @@ def extension_call(operation):
         raise ApiError(exc.status_code, exc.code, exc.message, exc.details) from exc
 
 
+async def extension_call_async(operation):
+    """进程启动/关闭可能等待 stdio Host，移出 FastAPI 事件循环。"""
+
+    try:
+        return await asyncio.to_thread(operation)
+    except ExtensionError as exc:
+        raise ApiError(exc.status_code, exc.code, exc.message, exc.details) from exc
+
+
 # Workspace (single configured Vault in Web development mode)
 @router.get("/workspace", response_model=WorkspaceInfo, tags=["Workspace"])
 async def get_workspace() -> WorkspaceInfo:
@@ -483,7 +504,7 @@ async def install_plugin(request: ExtensionInstallRequest) -> Plugin:
     tags=["Plugins"],
 )
 async def enable_plugin(plugin_id: str) -> Plugin:
-    return extension_call(lambda: container.plugins.enable(plugin_id))
+    return await extension_call_async(lambda: container.plugins.enable(plugin_id))
 
 
 @router.post(
@@ -492,7 +513,7 @@ async def enable_plugin(plugin_id: str) -> Plugin:
     tags=["Plugins"],
 )
 async def disable_plugin(plugin_id: str) -> Plugin:
-    return extension_call(lambda: container.plugins.disable(plugin_id))
+    return await extension_call_async(lambda: container.plugins.disable(plugin_id))
 
 
 @router.put(
@@ -503,11 +524,37 @@ async def disable_plugin(plugin_id: str) -> Plugin:
 async def set_plugin_permissions(
     plugin_id: str, request: PluginPermissionGrantRequest
 ) -> Plugin:
-    return extension_call(
+    return await extension_call_async(
         lambda: container.plugins.set_permissions(plugin_id, request.permissions)
     )
 
 
+@router.get(
+    "/plugins/{plugin_id}/host",
+    response_model=PluginHostStatus,
+    tags=["Plugins"],
+)
+async def get_plugin_host_status(plugin_id: str) -> PluginHostStatus:
+    return extension_call(lambda: container.plugins.get_host_status(plugin_id))
+
+
+@router.post(
+    "/plugins/{plugin_id}/host/restart",
+    response_model=OperationResponse,
+    status_code=202,
+    tags=["Plugins"],
+)
+async def restart_plugin_host(plugin_id: str) -> OperationResponse:
+    status = await extension_call_async(
+        lambda: container.plugins.restart_host(plugin_id)
+    )
+    return OperationResponse(
+        status="accepted",
+        resource_id=plugin_id,
+        message=f"Plugin Host status: {status.status.value}",
+    )
+
+
 @router.delete(
     "/plugins/{plugin_id}",
     response_model=OperationResponse,
@@ -516,7 +563,9 @@ async def set_plugin_permissions(
 async def uninstall_plugin(plugin_id: str) -> OperationResponse:
     plugin = extension_call(lambda: container.plugins.get(plugin_id))
     dependent_skills = container.skills.depending_on_tools(plugin.manifest.contributes.tools)
-    extension_call(lambda: container.plugins.uninstall(plugin_id, dependent_skills))
+    await extension_call_async(
+        lambda: container.plugins.uninstall(plugin_id, dependent_skills)
+    )
     return OperationResponse(status="completed", resource_id=plugin_id, message="uninstalled")
 
 
@@ -795,3 +844,160 @@ async def get_index_job(job_id: str) -> IndexJob:
     if job is None:
         raise ApiError(404, "RESOURCE_NOT_FOUND", "index job not found", {"job_id": job_id})
     return job
+
+
+# Benchmark
+@router.get(
+    "/benchmarks/datasets",
+    response_model=BenchmarkDatasetListResponse,
+    tags=["Benchmark"],
+)
+async def list_benchmark_datasets(
+    kind: BenchmarkKind = Query(default=BenchmarkKind.rag),
+) -> BenchmarkDatasetListResponse:
+    return BenchmarkDatasetListResponse(items=benchmark_datasets.list_datasets(kind))
+
+
+@router.post(
+    "/benchmarks/rag/runs",
+    response_model=BenchmarkRun,
+    status_code=202,
+    tags=["Benchmark"],
+)
+async def create_rag_benchmark(request: RAGRunRequest) -> BenchmarkRun:
+    return await benchmark_service.create_rag_run(request)
+
+
+@router.get(
+    "/benchmarks/runs",
+    response_model=BenchmarkRunListResponse,
+    tags=["Benchmark"],
+)
+async def list_benchmark_runs(
+    kind: BenchmarkKind | None = Query(default=None),
+    status: BenchmarkStatus | None = Query(default=None),
+    limit: int = Query(default=50, ge=1, le=100),
+    offset: int = Query(default=0, ge=0),
+) -> BenchmarkRunListResponse:
+    items, total = benchmark_service.list_runs(
+        kind=kind, status=status, limit=limit, offset=offset
+    )
+    return BenchmarkRunListResponse(
+        items=items, page=PageMeta(total=total, limit=limit, offset=offset)
+    )
+
+
+@router.get(
+    "/benchmarks/runs/{run_id}",
+    response_model=BenchmarkRun,
+    tags=["Benchmark"],
+)
+async def get_benchmark_run(run_id: str) -> BenchmarkRun:
+    run = benchmark_service.get_run(run_id)
+    if run is None:
+        raise ApiError(
+            404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id}
+        )
+    return run
+
+
+@router.post(
+    "/benchmarks/runs/{run_id}/cancel",
+    response_model=OperationResponse,
+    tags=["Benchmark"],
+)
+async def cancel_benchmark_run(run_id: str) -> OperationResponse:
+    run = benchmark_service.cancel_run(run_id)
+    if run is None:
+        raise ApiError(
+            404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id}
+        )
+    return OperationResponse(
+        status="accepted",
+        resource_id=run_id,
+        message=f"Benchmark run status: {run.status.value}",
+    )
+
+
+@router.get(
+    "/benchmarks/runs/{run_id}/events",
+    response_class=StreamingResponse,
+    responses={
+        200: {
+            "description": "BenchmarkEvent Server-Sent Events stream",
+            "content": {"text/event-stream": {}},
+        }
+    },
+    tags=["Benchmark"],
+)
+async def benchmark_events(
+    run_id: str,
+    after_sequence: int = Query(default=-1, ge=-1),
+    last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
+) -> StreamingResponse:
+    if benchmark_service.get_run(run_id) is None:
+        raise ApiError(
+            404, "BENCHMARK_RUN_NOT_FOUND", "benchmark run not found", {"run_id": run_id}
+        )
+
+    # SSE 断线重连：Last-Event-ID 优先于 after_sequence，用于从上次收到的事件继续
+    cursor = after_sequence
+    if last_event_id is not None:
+        try:
+            cursor = int(last_event_id)
+        except ValueError as exc:
+            raise ApiError(
+                400,
+                "BENCHMARK_EVENT_CURSOR_INVALID",
+                "Last-Event-ID must be an integer sequence.",
+                {"last_event_id": last_event_id},
+            ) from exc
+        if cursor < -1:
+            raise ApiError(
+                400,
+                "BENCHMARK_EVENT_CURSOR_INVALID",
+                "Last-Event-ID must be greater than or equal to -1.",
+            )
+
+    async def stream() -> AsyncIterator[str]:
+        # 先订阅（保证订阅之后产生的事件也能收到），再回放历史事件，最后实时输出新事件
+        queue = benchmark_service.subscribe(run_id)
+        last_sequence = cursor
+        for event in benchmark_service.get_events(run_id):
+            if event.sequence <= cursor:
+                continue
+            yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence)
+            last_sequence = event.sequence
+        if queue is None:
+            return
+        try:
+            while True:
+                event = await queue.get()
+                if event.sequence <= last_sequence:
+                    continue
+                yield as_sse(event.event.value, event.model_dump_json(), event_id=event.sequence)
+                last_sequence = event.sequence
+                if event.event in (
+                    BenchmarkEventType.run_completed,
+                    BenchmarkEventType.run_failed,
+                    BenchmarkEventType.run_cancelled,
+                ):
+                    break
+        finally:
+            benchmark_service.unsubscribe(run_id, queue)
+
+    return StreamingResponse(stream(), media_type="text/event-stream")
+
+
+@router.get(
+    "/benchmarks/runs/{run_id}/report",
+    response_model=BenchmarkReport,
+    tags=["Benchmark"],
+)
+async def get_benchmark_report(run_id: str) -> BenchmarkReport:
+    report = benchmark_service.get_report(run_id)
+    if report is None:
+        raise ApiError(
+            404, "BENCHMARK_RUN_NOT_FOUND", "benchmark report not found", {"run_id": run_id}
+        )
+    return report

backend/data/benchmarks/rag-core-v1.json

@@ -0,0 +1,48 @@
+{
+  "dataset_id": "rag-core-v1",
+  "kind": "rag",
+  "version": "1.0.0",
+  "description": "基础中文笔记检索集（对应 backend/data/vault 内置语料，重建索引后即可复现）",
+  "cases": [
+    {
+      "case_id": "rag-vector-sim",
+      "query": "向量数据库如何进行相似度检索",
+      "expected_note_ids": ["note_c1454740a0e55ef5"],
+      "expected_block_ids": ["blk_07c4c6bce0ec4d12", "blk_605fb3593809f224"],
+      "citation_required": true,
+      "tags": ["向量数据库", "检索"]
+    },
+    {
+      "case_id": "rag-python-func",
+      "query": "Python 如何定义函数",
+      "expected_note_ids": ["note_424c3742c6f0e555"],
+      "expected_block_ids": ["blk_45d48cae2fed40fe", "blk_0768d9c25c2ecf07"],
+      "citation_required": true,
+      "tags": ["python"]
+    },
+    {
+      "case_id": "rag-citation",
+      "query": "搜索结果如何定位到原文位置",
+      "expected_note_ids": ["note_0c619caa30b1614c"],
+      "expected_block_ids": ["blk_3f6fcead71c25fc6", "blk_9af7b12e9ce909fc"],
+      "citation_required": true,
+      "tags": ["RAG"]
+    },
+    {
+      "case_id": "rag-hybrid",
+      "query": "混合检索怎么融合全文和向量",
+      "expected_note_ids": ["note_c1454740a0e55ef5"],
+      "expected_block_ids": ["blk_82b45418dba9f720"],
+      "citation_required": true,
+      "tags": ["检索"]
+    },
+    {
+      "case_id": "rag-tech-stack",
+      "query": "这个项目用什么后端和检索技术",
+      "expected_note_ids": ["note_3327e6cf18f3701f"],
+      "expected_block_ids": ["blk_feb2a9c42e7d31ad"],
+      "citation_required": false,
+      "tags": ["项目"]
+    }
+  ]
+}

backend/extensions/fixtures/mcp-echo/plugin.yaml

@@ -0,0 +1,21 @@
+id: mcp-fixture
+name: MCP Fixture
+version: 1.0.0
+description: 阶段 C 离线联调 Fixture，覆盖 MCP Tool 生命周期与错误边界。
+permissions:
+  - notes.read
+contributes:
+  tools:
+    - mcp-fixture.echo
+    - mcp-fixture.fail
+    - mcp-fixture.sleep
+    - mcp-fixture.large
+    - mcp-fixture.environment
+    - mcp-fixture.exit
+backend:
+  type: mcp
+  transport: stdio
+  command: python
+  args: [server.py]
+  startup_timeout_seconds: 5
+  tool_timeout_seconds: 1

backend/extensions/fixtures/mcp-echo/server.py

@@ -0,0 +1,204 @@
+"""确定性的 MCP stdio 测试 Server；仅使用标准库，不依赖产品代码。"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import threading
+import time
+from typing import Any
+
+WRITE_LOCK = threading.Lock()
+CANCELLED: dict[int, threading.Event] = {}
+MODE = sys.argv[1] if len(sys.argv) > 1 else "normal"
+
+
+def send(message: dict[str, Any]) -> None:
+    with WRITE_LOCK:
+        sys.stdout.write(json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n")
+        sys.stdout.flush()
+
+
+def respond(request_id: int, result: dict[str, Any]) -> None:
+    send({"jsonrpc": "2.0", "id": request_id, "result": result})
+
+
+def tool(name: str, description: str, properties: dict[str, Any] | None = None) -> dict[str, Any]:
+    return {
+        "name": name,
+        "description": description,
+        "inputSchema": {
+            "type": "object",
+            "properties": properties or {},
+            "required": list(properties or {}),
+            "additionalProperties": False,
+        },
+    }
+
+
+TOOLS = {
+    "echo": {
+        **tool(
+            "echo",
+            "Return the provided text.",
+            {
+                "text": {"type": "string"},
+                "suffix": {"type": ["string", "null"]},
+            },
+        ),
+        "_meta": {"notesagent/permission": "notes.read"},
+    },
+    "fail": tool("fail", "Return an MCP business error."),
+    "sleep": tool("sleep", "Wait until completed or cancelled.", {"seconds": {"type": "number"}}),
+    "large": tool("large", "Return a result larger than the host limit."),
+    "environment": tool("environment", "Report whether host secrets leaked into the process."),
+    "exit": tool("exit", "Terminate the fixture process."),
+}
+# suffix 是可选字段，用于验证 Host 不会把缺省值擅自补成 null。
+TOOLS["echo"]["inputSchema"]["required"] = ["text"]
+
+
+def call_tool(request_id: int, params: dict[str, Any]) -> None:
+    name = params.get("name")
+    arguments = params.get("arguments") or {}
+    if name == "echo":
+        text = str(arguments.get("text", ""))
+        structured_content = {"echo": text}
+        if "suffix" in arguments:
+            structured_content["suffix"] = arguments["suffix"]
+        respond(
+            request_id,
+            {
+                "content": [{"type": "text", "text": text}],
+                "structuredContent": structured_content,
+                "isError": False,
+            },
+        )
+        return
+    if name == "fail":
+        respond(
+            request_id,
+            {
+                "content": [{"type": "text", "text": "fixture failure"}],
+                "isError": True,
+            },
+        )
+        return
+    if name == "large":
+        respond(
+            request_id,
+            {
+                "content": [{"type": "text", "text": "x" * 300_000}],
+                "isError": False,
+            },
+        )
+        return
+    if name == "environment":
+        respond(
+            request_id,
+            {
+                "content": [{"type": "text", "text": "environment checked"}],
+                "structuredContent": {
+                    "has_openai_key": "OPENAI_API_KEY" in os.environ,
+                    "has_app_db_path": "APP_DB_PATH" in os.environ,
+                },
+                "isError": False,
+            },
+        )
+        return
+    if name == "exit":
+        os._exit(17)
+    if name == "sleep":
+        cancelled = CANCELLED.setdefault(request_id, threading.Event())
+        seconds = max(0.0, min(float(arguments.get("seconds", 0)), 30.0))
+        if cancelled.wait(seconds):
+            respond(
+                request_id,
+                {
+                    "content": [{"type": "text", "text": "cancelled"}],
+                    "isError": True,
+                },
+            )
+        else:
+            respond(
+                request_id,
+                {
+                    "content": [{"type": "text", "text": "completed"}],
+                    "structuredContent": {"slept": seconds},
+                    "isError": False,
+                },
+            )
+        CANCELLED.pop(request_id, None)
+        return
+    send(
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "error": {"code": -32602, "message": f"Unknown tool: {name}"},
+        }
+    )
+
+
+def main() -> None:
+    for line in sys.stdin:
+        message = json.loads(line)
+        method = message.get("method")
+        request_id = message.get("id")
+        params = message.get("params") or {}
+        if method == "initialize" and isinstance(request_id, int):
+            if MODE == "invalid-result":
+                send({"jsonrpc": "2.0", "id": request_id, "result": None})
+                continue
+            if MODE == "oversized-stdout":
+                # 不带换行，验证 Host 在读取完整内容前执行硬上限。
+                sys.stdout.write("x" * (2 * 1024 * 1024 + 1))
+                sys.stdout.flush()
+                time.sleep(10)
+                return
+            respond(
+                request_id,
+                {
+                    "protocolVersion": params.get("protocolVersion"),
+                    "capabilities": (
+                        {} if MODE == "no-tools" else {"tools": {"listChanged": False}}
+                    ),
+                    "serverInfo": {"name": "notesagent-mcp-fixture", "version": "1.0.0"},
+                },
+            )
+        elif method == "tools/list" and isinstance(request_id, int):
+            if MODE == "invalid-schema":
+                respond(
+                    request_id,
+                    {
+                        "tools": [
+                            {
+                                "name": "broken",
+                                "description": "invalid schema",
+                                "inputSchema": {"type": "string"},
+                            }
+                        ]
+                    },
+                )
+            elif params.get("cursor") == "page-2":
+                respond(
+                    request_id,
+                    {"tools": [TOOLS["large"], TOOLS["environment"], TOOLS["exit"]]},
+                )
+            else:
+                respond(
+                    request_id,
+                    {"tools": [TOOLS["echo"], TOOLS["fail"], TOOLS["sleep"]], "nextCursor": "page-2"},
+                )
+        elif method == "tools/call" and isinstance(request_id, int):
+            threading.Thread(target=call_tool, args=(request_id, params), daemon=True).start()
+        elif method == "notifications/cancelled":
+            cancelled_id = params.get("requestId")
+            if isinstance(cancelled_id, int):
+                CANCELLED.setdefault(cancelled_id, threading.Event()).set()
+        elif method == "ping" and isinstance(request_id, int):
+            respond(request_id, {})
+
+
+if __name__ == "__main__":
+    main()

backend/tests/test_api.py

@@ -93,6 +93,8 @@ def test_openapi_contains_documented_frontend_interfaces() -> None:
         "/api/skills",
         "/api/plugins",
         "/api/plugins/install",
+        "/api/plugins/{plugin_id}/host",
+        "/api/plugins/{plugin_id}/host/restart",
         "/api/plugins/{plugin_id}/enable",
         "/api/plugins/{plugin_id}/disable",
         "/api/providers/test",

backend/tests/test_benchmark.py

@@ -0,0 +1,441 @@
+"""Benchmark 服务的单元与端到端测试。
+
+沿用 conftest 的隔离机制：APP_DATA_DIR / DB / Vault 都指向临时目录，benchmark
+数据集也落在临时目录（settings.benchmark_datasets_path），不读写真实数据。
+
+运行采用「创建即 queued + 后台 Task 执行」的异步模型，测试通过 _run 在同一事件循环内
+创建并等待后台任务结束，得到终态 BenchmarkRun 后再断言。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+from pydantic import ValidationError
+
+from app.benchmarks import datasets, metrics as m, service
+from app.config import get_settings
+from app.contracts import (
+    BenchmarkKind,
+    BenchmarkRun,
+    BenchmarkStatus,
+    RAGRunRequest,
+    SearchMode,
+)
+from app.errors import ApiError
+
+
+def _write_dataset(dataset_id: str, cases: list[dict], *, kind: str = "rag") -> None:
+    directory = get_settings().benchmark_datasets_path
+    directory.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "dataset_id": dataset_id,
+        "kind": kind,
+        "version": "1.0.0",
+        "description": "test dataset",
+        "cases": cases,
+    }
+    (directory / f"{dataset_id}.json").write_text(
+        json.dumps(payload, ensure_ascii=False), encoding="utf-8"
+    )
+
+
+def _write_raw(dataset_id: str, raw: dict) -> None:
+    directory = get_settings().benchmark_datasets_path
+    directory.mkdir(parents=True, exist_ok=True)
+    (directory / f"{dataset_id}.json").write_text(
+        json.dumps(raw, ensure_ascii=False), encoding="utf-8"
+    )
+
+
+def _run(request: RAGRunRequest):
+    """创建运行并在同一事件循环内等待后台任务结束，返回终态 BenchmarkRun。"""
+    from app.contracts import BenchmarkRun
+
+    async def _execute() -> BenchmarkRun:
+        run = await service.create_rag_run(request)
+        return await service.wait_for_run(run.run_id)
+
+    return asyncio.run(_execute())
+
+
+# --------------------------------------------------------------------------- #
+# 指标纯函数
+# --------------------------------------------------------------------------- #
+def test_hit_at_k_and_recall() -> None:
+    retrieved = ["a", "b", "c"]
+    expected = {"b", "z"}
+
+    assert m.hit_at_k(retrieved, expected, 1) is False
+    assert m.hit_at_k(retrieved, expected, 2) is True
+    assert m.recall_at_k(retrieved, expected, 5) == 0.5  # 只召回 b
+
+
+def test_recall_at_k_dedups_duplicate_notes() -> None:
+    # 同一 Note 经多个 Block 重复出现，去重后 Recall 不应超过 1
+    assert m.recall_at_k(["note-a", "note-a"], {"note-a"}, 2) == 1.0
+    assert m.recall_at_k(["note-a", "note-a", "note-b"], {"note-a"}, 3) == 1.0
+
+
+def test_reciprocal_rank_and_citation_hit() -> None:
+    assert m.reciprocal_rank(["x", "a", "b"], {"b"}) == 1 / 3
+    assert m.reciprocal_rank(["x"], {"b"}) == 0.0
+    assert m.citation_hit(["blk_1"], {"blk_1"}) is True
+    assert m.citation_hit(["blk_2"], {"blk_1"}) is False
+    assert m.citation_hit([], {"blk_1"}) is False
+
+
+def test_percentile() -> None:
+    assert m.percentile([1.0, 2.0, 3.0, 4.0], 50.0) == 2.5
+    assert m.percentile([], 50.0) == 0.0
+    assert m.percentile([7.0], 95.0) == 7.0
+
+
+# --------------------------------------------------------------------------- #
+# Dataset 注册与校验
+# --------------------------------------------------------------------------- #
+def test_list_datasets_empty_by_default() -> None:
+    assert datasets.list_datasets(BenchmarkKind.rag) == []
+
+
+def test_load_missing_dataset_raises() -> None:
+    with pytest.raises(ApiError) as exc:
+        datasets.load_dataset("does-not-exist", BenchmarkKind.rag)
+    assert exc.value.status_code == 404
+    assert exc.value.code == "BENCHMARK_DATASET_NOT_FOUND"
+
+
+def test_dataset_without_expected_ids_is_invalid() -> None:
+    _write_dataset("bad-v1", [{"case_id": "x", "query": "q", "citation_required": False}])
+    with pytest.raises(ApiError) as exc:
+        datasets.load_dataset("bad-v1", BenchmarkKind.rag)
+    assert exc.value.code == "BENCHMARK_DATASET_INVALID"
+
+
+def test_dataset_kind_mismatch_is_invalid() -> None:
+    _write_dataset("agent-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}], kind="agent")
+    with pytest.raises(ApiError) as exc:
+        datasets.load_dataset("agent-v1", BenchmarkKind.rag)
+    assert exc.value.code == "BENCHMARK_DATASET_INVALID"
+
+
+def test_citation_required_requires_expected_block_ids() -> None:
+    # citation_required=true 却没有 expected_block_ids，无法计算 Citation Hit Rate，应拒绝
+    _write_dataset(
+        "cit-req-v1",
+        [{"case_id": "x", "query": "q", "expected_note_ids": ["n"], "citation_required": True}],
+    )
+    with pytest.raises(ApiError) as exc:
+        datasets.load_dataset("cit-req-v1", BenchmarkKind.rag)
+    assert exc.value.code == "BENCHMARK_DATASET_INVALID"
+
+
+def test_list_datasets_skips_corrupted_structure() -> None:
+    # 合法 JSON 但字段结构错误（cases: 42），列表接口应隔离该文件而非整体 500
+    _write_raw("bad-structure", {"dataset_id": "bad-structure", "kind": "rag", "cases": 42})
+    _write_dataset("good-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}])
+
+    infos = datasets.list_datasets(BenchmarkKind.rag)
+    ids = {info.dataset_id for info in infos}
+    assert "good-v1" in ids
+    assert "bad-structure" not in ids
+
+
+# --------------------------------------------------------------------------- #
+# 请求校验（空 / 重复 modes）
+# --------------------------------------------------------------------------- #
+def test_empty_modes_rejected() -> None:
+    with pytest.raises(ValidationError):
+        RAGRunRequest(dataset_id="x", modes=[])
+
+
+def test_duplicate_modes_rejected() -> None:
+    with pytest.raises(ValidationError):
+        RAGRunRequest(dataset_id="x", modes=[SearchMode.fts, SearchMode.fts])
+
+
+# --------------------------------------------------------------------------- #
+# RAG Benchmark 端到端
+# --------------------------------------------------------------------------- #
+def _single_note_case() -> tuple[str, str, dict]:
+    from app.services import note_service
+
+    note = asyncio.run(
+        note_service.create_note(
+            title="向量库",
+            markdown="向量数据库用于存储高维向量并支持近似最近邻检索。",
+            folder="",
+            tags=["向量"],
+        )
+    )
+    case = {
+        "case_id": "c1",
+        "query": "向量数据库相似度检索",
+        "expected_note_ids": [note.note_id],
+        "expected_block_ids": [note.blocks[0].block_id],
+        "citation_required": True,
+        "tags": ["向量"],
+    }
+    return note.note_id, note.blocks[0].block_id, case
+
+
+def test_rag_benchmark_end_to_end() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("e2e-v1", [case])
+
+    run = _run(RAGRunRequest(dataset_id="e2e-v1", modes=[SearchMode.fts]))
+
+    assert run.status.value == "completed"
+    assert run.dataset_hash.startswith("sha256:")
+    assert run.metrics is not None
+
+    fts = run.metrics["fts"]
+    assert fts["hit_at_1"] == 1.0
+    assert fts["recall_at_k"] == 1.0
+    assert fts["mrr"] == 1.0
+    assert fts["citation_hit_rate"] == 1.0
+    assert fts["p50_latency_ms"] >= 0.0
+    assert fts["p95_latency_ms"] >= fts["p50_latency_ms"]
+
+
+def test_rag_benchmark_all_modes_produce_metrics() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("e2e-modes-v1", [case])
+
+    run = _run(RAGRunRequest(dataset_id="e2e-modes-v1"))
+    assert run.status.value == "completed"
+
+    for mode in ("fts", "vector", "hybrid"):
+        assert mode in run.metrics
+        for key in ("hit_at_1", "hit_at_5", "recall_at_k", "mrr", "citation_hit_rate"):
+            assert 0.0 <= run.metrics[mode][key] <= 1.0
+
+
+def test_config_snapshot_records_index_and_models() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("snapshot-v1", [case])
+
+    run = _run(RAGRunRequest(dataset_id="snapshot-v1", modes=[SearchMode.fts]))
+
+    snapshot = run.config_snapshot
+    assert snapshot["index_meta"] is not None
+    assert snapshot["embedding"]["version"]
+    assert snapshot["embedding"]["dim"]
+    assert snapshot["reranker"]["version"]
+    assert snapshot["retrieval"]["rrf_k"] == 60
+
+
+def test_benchmark_report_and_events() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("report-v1", [case])
+
+    run = _run(RAGRunRequest(dataset_id="report-v1", modes=[SearchMode.fts]))
+    report = service.get_report(run.run_id)
+    events = service.get_events(run.run_id)
+
+    assert report is not None
+    assert report.run_id == run.run_id
+    assert len(report.cases) == 1
+    assert report.cases[0].case_id == "c1"
+    assert report.cases[0].hit_at_1 is True
+
+    assert events, "运行应产生事件"
+    assert events[0].event.value == "RunStarted"
+    assert events[-1].event.value == "RunCompleted"
+
+
+def test_cancel_completed_run_keeps_status() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("cancel-v1", [case])
+
+    run = _run(RAGRunRequest(dataset_id="cancel-v1", modes=[SearchMode.fts]))
+    assert run.status.value == "completed"
+
+    cancelled = service.cancel_run(run.run_id)
+    assert cancelled.status.value == "completed"  # 已结束，不再变 cancelled
+
+
+def test_cancel_queued_run_marks_cancelled() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("cancel-queued-v1", [case])
+
+    async def _scenario():
+        run = await service.create_rag_run(
+            RAGRunRequest(dataset_id="cancel-queued-v1", modes=[SearchMode.fts])
+        )
+        service.cancel_run(run.run_id)
+        return await service.wait_for_run(run.run_id)
+
+    run = asyncio.run(_scenario())
+    assert run.status.value == "cancelled"
+
+
+# --------------------------------------------------------------------------- #
+# 指标聚合：Citation Hit Rate 只统计 citation_required 样本
+# --------------------------------------------------------------------------- #
+def test_citation_hit_rate_only_counts_citation_required() -> None:
+    from app.benchmarks import rag as rag_module
+    from app.contracts import RAGCaseResult
+
+    cases = [
+        RAGCaseResult(
+            case_id="a", mode=SearchMode.fts, repeat=0, latency_ms=1.0,
+            citation_hit=True, citation_applicable=True,
+        ),
+        RAGCaseResult(
+            case_id="b", mode=SearchMode.fts, repeat=0, latency_ms=1.0,
+            citation_hit=False, citation_applicable=False,
+        ),
+    ]
+    metrics = rag_module._aggregate(cases, SearchMode.fts)
+    # 只有 citation_applicable（citation_required=true）的样本计入分母
+    assert metrics.citation_hit_rate == 1.0
+
+
+# --------------------------------------------------------------------------- #
+# 路由接入
+# --------------------------------------------------------------------------- #
+def test_benchmark_routes_wired() -> None:
+    from app import routes
+
+    _, _, case = _single_note_case()
+    _write_dataset("route-v1", [case])
+
+    async def _scenario():
+        listed = await routes.list_benchmark_datasets(BenchmarkKind.rag)
+        assert any(item.dataset_id == "route-v1" for item in listed.items)
+
+        run = await routes.create_rag_benchmark(
+            RAGRunRequest(dataset_id="route-v1", modes=[SearchMode.fts])
+        )
+        assert run.status.value == "queued"
+        return await service.wait_for_run(run.run_id)
+
+    run = asyncio.run(_scenario())
+    assert run.status.value == "completed"
+
+    got = asyncio.run(routes.get_benchmark_run(run.run_id))
+    assert got.run_id == run.run_id
+
+    report = asyncio.run(routes.get_benchmark_report(run.run_id))
+    assert report.cases[0].case_id == "c1"
+
+
+def test_benchmark_run_not_found_raises() -> None:
+    from app import routes
+
+    with pytest.raises(ApiError) as exc:
+        asyncio.run(routes.get_benchmark_run("benchmark_missing"))
+    assert exc.value.code == "BENCHMARK_RUN_NOT_FOUND"
+
+
+# --------------------------------------------------------------------------- #
+# 审阅回归：索引兼容 / 容量 / 失败样本 / 取消事件 / 数据集隔离
+# --------------------------------------------------------------------------- #
+def test_create_rag_run_requires_built_index() -> None:
+    # 空索引（无已索引 block）会让所有模式得到全 0 指标，应在创建时拒绝而非跑出误导结果
+    _write_dataset("empty-index-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}])
+    with pytest.raises(ApiError) as exc:
+        asyncio.run(
+            service.create_rag_run(
+                RAGRunRequest(dataset_id="empty-index-v1", modes=[SearchMode.fts])
+            )
+        )
+    assert exc.value.status_code == 409
+    assert exc.value.code == "BENCHMARK_INDEX_INCOMPATIBLE"
+
+
+def test_capacity_exceeded_when_all_runs_active(monkeypatch) -> None:
+    # 满容量且全为活动（非终态）run 时，无法淘汰，应拒绝创建而非删掉正在运行的 run
+    _, _, case = _single_note_case()
+    _write_dataset("capacity-v1", [case])
+
+    monkeypatch.setattr(service, "MAX_RUNS", 1)
+    fake_id = "benchmark_fake_active"
+    service._runs[fake_id] = BenchmarkRun(
+        run_id=fake_id,
+        kind=BenchmarkKind.rag,
+        dataset_id="capacity-v1",
+        dataset_hash="sha256:fake",
+        status=BenchmarkStatus.queued,
+        created_at=service._now(),
+    )
+    try:
+        with pytest.raises(ApiError) as exc:
+            asyncio.run(
+                service.create_rag_run(
+                    RAGRunRequest(dataset_id="capacity-v1", modes=[SearchMode.fts])
+                )
+            )
+        assert exc.value.status_code == 429
+        assert exc.value.code == "BENCHMARK_CAPACITY_EXCEEDED"
+    finally:
+        service._runs.pop(fake_id, None)
+
+
+def test_failed_samples_counted_as_zero_in_aggregate() -> None:
+    from app.benchmarks import rag as rag_module
+    from app.contracts import RAGCaseResult
+
+    cases = [
+        RAGCaseResult(
+            case_id="ok", mode=SearchMode.fts, repeat=0, latency_ms=10.0,
+            hit_at_1=True, recall=1.0, reciprocal_rank=1.0,
+            citation_hit=True, citation_applicable=True,
+        ),
+        RAGCaseResult(
+            case_id="boom", mode=SearchMode.fts, repeat=0, latency_ms=0.0,
+            error="RAG case evaluation failed.",
+            error_code="BENCHMARK_CASE_EVALUATION_FAILED",
+        ),
+    ]
+    metrics = rag_module._aggregate(cases, SearchMode.fts)
+
+    assert metrics.total_cases == 2
+    assert metrics.successful_cases == 1
+    assert metrics.failed_cases == 1
+    assert metrics.failure_rate == 0.5
+    # 失败样本按零分计入质量指标分母，汇总不虚高
+    assert metrics.hit_at_1 == 0.5
+    assert metrics.recall_at_k == 0.5
+    # 延迟只统计成功样本
+    assert metrics.p50_latency_ms == 10.0
+
+
+def test_cancel_emits_run_cancelled_event() -> None:
+    _, _, case = _single_note_case()
+    _write_dataset("cancel-event-v1", [case])
+
+    async def _scenario():
+        run = await service.create_rag_run(
+            RAGRunRequest(dataset_id="cancel-event-v1", modes=[SearchMode.fts])
+        )
+        service.cancel_run(run.run_id)
+        return await service.wait_for_run(run.run_id)
+
+    run = asyncio.run(_scenario())
+    assert run.status.value == "cancelled"
+    events = service.get_events(run.run_id)
+    assert events[-1].event.value == "RunCancelled"
+
+
+def test_load_dataset_ignores_corrupted_unrelated_files() -> None:
+    # 无关文件损坏（非法 JSON / 顶层非对象）不应阻断目标数据集加载
+    directory = get_settings().benchmark_datasets_path
+    directory.mkdir(parents=True, exist_ok=True)
+    (directory / "broken.json").write_text("{ not valid json", encoding="utf-8")
+    (directory / "array.json").write_text('["a", "b"]', encoding="utf-8")
+    _write_dataset("ok-v1", [{"case_id": "x", "query": "q", "expected_note_ids": ["n"]}])
+
+    dataset = datasets.load_dataset("ok-v1", BenchmarkKind.rag)
+    assert dataset.dataset_id == "ok-v1"
+    assert len(dataset.cases) == 1
+
+
+def test_load_dataset_top_level_must_be_object() -> None:
+    _write_raw("array-top", ["a", "b"])
+    with pytest.raises(ApiError) as exc:
+        datasets.load_dataset("array-top", BenchmarkKind.rag)
+    assert exc.value.code == "BENCHMARK_DATASET_INVALID"

backend/tests/test_extension_core.py

@@ -1,4 +1,7 @@
 import asyncio
+import shutil
+import threading
+import time
 
 import pytest
 
@@ -12,14 +15,31 @@ from app.contracts import (
     ToolCall,
 )
 from app.extensions import ExtensionError
+from app.extensions.mcp import McpStdioClient
+from app.extensions.runtime import _arguments_model_from_schema
 from app.services import note_service
-from app.config import get_settings
+from app.config import BACKEND_DIR, get_settings
+
+
+MCP_FIXTURE = BACKEND_DIR / "extensions" / "fixtures" / "mcp-echo"
 
 
 def run(coroutine):
     return asyncio.run(coroutine)
 
 
+@pytest.fixture
+def mcp_container():
+    container = build_container()
+    installed = container.plugins.install(MCP_FIXTURE)
+    assert installed.status == "permission_required"
+    container.plugins.set_permissions("mcp-fixture", ["notes.read"])
+    try:
+        yield container
+    finally:
+        container.plugins.shutdown()
+
+
 def test_bundled_plugin_registers_tool_and_skill_is_ready() -> None:
     async def scenario() -> None:
         container = build_container()
@@ -297,3 +317,300 @@ def test_attachment_and_transcription_tools_use_host_storage() -> None:
         assert transcription.output["text"] == "会议转写内容"
 
     run(scenario())
+
+
+def test_mcp_stdio_host_discovers_namespaced_tools_and_maps_results(
+    mcp_container, monkeypatch
+) -> None:
+    async def scenario() -> None:
+        monkeypatch.setenv("OPENAI_API_KEY", "must-not-enter-plugin-host")
+        enabled = mcp_container.plugins.enable("mcp-fixture")
+        status = mcp_container.plugins.get_host_status("mcp-fixture")
+        definition = mcp_container.tools.get("mcp-fixture.echo").definition
+        result = await mcp_container.tools.execute(
+            ToolCall(
+                tool_call_id="call_mcp_echo",
+                name="mcp-fixture.echo",
+                arguments={"text": "hello mcp"},
+            ),
+            ToolExecutionContext(
+                run_id="run_mcp_fixture", tool_call_id="call_mcp_echo"
+            ),
+        )
+
+        assert enabled.status == "ready" and enabled.enabled is True
+        assert status.status == "ready"
+        environment = await mcp_container.tools.execute(
+            ToolCall(
+                tool_call_id="call_mcp_environment",
+                name="mcp-fixture.environment",
+                arguments={},
+            ),
+            ToolExecutionContext(run_id="run_mcp_fixture"),
+        )
+
+        assert status.tools_count == 6
+        assert status.protocol_version == "2025-11-25"
+        assert status.server_name == "notesagent-mcp-fixture"
+        assert definition.permission == "notes.read"
+        assert result.success is True
+        assert result.output == {"echo": "hello mcp"}
+        explicit_null = await mcp_container.tools.execute(
+            ToolCall(
+                tool_call_id="call_mcp_explicit_null",
+                name="mcp-fixture.echo",
+                arguments={"text": "null stays explicit", "suffix": None},
+            ),
+            ToolExecutionContext(run_id="run_mcp_fixture"),
+        )
+        assert explicit_null.success is True
+        assert explicit_null.output == {
+            "echo": "null stays explicit",
+            "suffix": None,
+        }
+        assert environment.success is True
+        assert environment.output == {
+            "has_openai_key": False,
+            "has_app_db_path": False,
+        }
+
+        disabled = mcp_container.plugins.disable("mcp-fixture")
+        assert disabled.status == "disabled"
+        assert mcp_container.plugins.get_host_status("mcp-fixture").status == "stopped"
+        assert not mcp_container.tools.contains("mcp-fixture.echo")
+        with pytest.raises(ExtensionError) as exc:
+            mcp_container.plugins.restart_host("mcp-fixture")
+        assert exc.value.code == "PLUGIN_HOST_UNAVAILABLE"
+        assert mcp_container.plugins.get("mcp-fixture").status == "disabled"
+        assert not mcp_container.tools.contains("mcp-fixture.echo")
+
+        mcp_container.plugins.uninstall("mcp-fixture")
+        reinstalled = mcp_container.plugins.install(MCP_FIXTURE)
+        fresh_status = mcp_container.plugins.get_host_status("mcp-fixture")
+        assert reinstalled.status == "permission_required"
+        assert fresh_status.status == "stopped"
+        assert fresh_status.started_at is None
+        assert fresh_status.protocol_version is None
+        assert fresh_status.server_name is None
+
+    run(scenario())
+
+
+def test_agent_calls_mcp_tool_through_registry_and_writes_trace(mcp_container) -> None:
+    async def scenario() -> None:
+        mcp_container.plugins.enable("mcp-fixture")
+        created = await mcp_container.agent.create_run(
+            AgentRunCreateRequest(
+                input='/tool mcp-fixture.echo {"text":"agent mcp"}',
+                provider_id="mock",
+                model="mock-1",
+                allowed_tools=["mcp-fixture.echo"],
+            )
+        )
+        completed = await mcp_container.agent.wait(created.run_id)
+        trace = mcp_container.agent.get_trace(
+            created.run_id, after_sequence=-1, limit=100
+        )
+
+        assert completed.status == AgentRunStatus.completed
+        assert completed.tool_results[0].success is True
+        assert completed.tool_results[0].output == {"echo": "agent mcp"}
+        assert any(
+            item.event == "ToolCall" and item.data.get("name") == "mcp-fixture.echo"
+            for item in trace.items
+        )
+
+    run(scenario())
+
+
+def test_mcp_business_error_size_limit_and_timeout_are_structured(mcp_container) -> None:
+    async def scenario() -> None:
+        mcp_container.plugins.enable("mcp-fixture")
+        context = ToolExecutionContext(run_id="run_mcp_errors")
+
+        failed = await mcp_container.tools.execute(
+            ToolCall(tool_call_id="call_fail", name="mcp-fixture.fail", arguments={}),
+            context,
+        )
+        oversized = await mcp_container.tools.execute(
+            ToolCall(tool_call_id="call_large", name="mcp-fixture.large", arguments={}),
+            context,
+        )
+        timed_out = await mcp_container.tools.execute(
+            ToolCall(
+                tool_call_id="call_sleep",
+                name="mcp-fixture.sleep",
+                arguments={"seconds": 5},
+            ),
+            ToolExecutionContext(
+                run_id="run_mcp_errors", tool_call_id="call_sleep"
+            ),
+        )
+        recovered = await mcp_container.tools.execute(
+            ToolCall(
+                tool_call_id="call_after_timeout",
+                name="mcp-fixture.echo",
+                arguments={"text": "still ready"},
+            ),
+            context,
+        )
+
+        assert failed.success is False
+        assert failed.error_code == "MCP_TOOL_CALL_FAILED"
+        assert failed.error_message == "fixture failure"
+        assert oversized.success is False
+        assert oversized.error_code == "MCP_TOOL_RESULT_TOO_LARGE"
+        assert timed_out.success is False
+        assert timed_out.error_code == "MCP_TOOL_CALL_FAILED"
+        assert recovered.success is True
+        assert mcp_container.plugins.get_host_status("mcp-fixture").status == "ready"
+
+    run(scenario())
+
+
+def test_mcp_cancel_releases_blocking_response_thread(
+    mcp_container, monkeypatch
+) -> None:
+    async def scenario() -> None:
+        mcp_container.plugins.enable("mcp-fixture")
+        released = threading.Event()
+        original_wait = McpStdioClient.wait_response
+
+        def tracked_wait(self, *args, **kwargs):
+            try:
+                return original_wait(self, *args, **kwargs)
+            finally:
+                released.set()
+
+        monkeypatch.setattr(McpStdioClient, "wait_response", tracked_wait)
+        task = asyncio.create_task(
+            mcp_container.plugins.mcp.call_tool(
+                "mcp-fixture",
+                "sleep",
+                {"seconds": 5},
+                request_id="call_cancel_release",
+            )
+        )
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+        deadline = time.monotonic() + 0.5
+        while not released.is_set() and time.monotonic() < deadline:
+            await asyncio.sleep(0.01)
+        assert released.is_set(), "cancelled MCP wait must not occupy a worker until timeout"
+
+    run(scenario())
+
+
+def test_mcp_argument_model_preserves_json_schema_additional_properties() -> None:
+    arguments_model = _arguments_model_from_schema(
+        "mcp-fixture.dynamic",
+        {
+            "type": "object",
+            "properties": {"model_dump": {"type": "string"}},
+            "required": ["model_dump"],
+            "additionalProperties": {"type": "string"},
+        },
+    )
+
+    arguments = arguments_model.model_validate(
+        {"model_dump": "method name remains data", "dynamic-key": "value"}
+    )
+
+    assert arguments.model_dump() == {
+        "model_dump": "method name remains data",
+        "dynamic-key": "value",
+    }
+
+
+def test_production_rejects_unsandboxed_mcp_host(monkeypatch) -> None:
+    monkeypatch.setenv("APP_ENVIRONMENT", "production")
+    get_settings.cache_clear()
+    container = build_container()
+    installed = container.plugins.install(MCP_FIXTURE)
+    assert installed.status == "permission_required"
+    container.plugins.set_permissions("mcp-fixture", ["notes.read"])
+    try:
+        with pytest.raises(ExtensionError) as exc:
+            container.plugins.enable("mcp-fixture")
+        assert exc.value.code == "MCP_TRUST_APPROVAL_REQUIRED"
+        assert container.plugins.get_host_status("mcp-fixture").status == "stopped"
+        assert not container.tools.contains("mcp-fixture.echo")
+    finally:
+        container.plugins.shutdown()
+        get_settings.cache_clear()
+
+
+def test_mcp_abnormal_exit_unregisters_tools_and_restart_recovers(mcp_container) -> None:
+    async def scenario() -> None:
+        mcp_container.plugins.enable("mcp-fixture")
+        crashed = await mcp_container.tools.execute(
+            ToolCall(tool_call_id="call_exit", name="mcp-fixture.exit", arguments={}),
+            ToolExecutionContext(run_id="run_mcp_exit", tool_call_id="call_exit"),
+        )
+
+        deadline = time.monotonic() + 2
+        while mcp_container.tools.contains("mcp-fixture.echo") and time.monotonic() < deadline:
+            await asyncio.sleep(0.02)
+
+        plugin = mcp_container.plugins.get("mcp-fixture")
+        status = mcp_container.plugins.get_host_status("mcp-fixture")
+        assert crashed.success is False
+        assert crashed.error_code == "PLUGIN_HOST_UNAVAILABLE"
+        assert plugin.status == "error" and plugin.enabled is False
+        assert status.status == "unhealthy"
+        assert not mcp_container.tools.contains("mcp-fixture.echo")
+
+        restarted = mcp_container.plugins.restart_host("mcp-fixture")
+        assert restarted.status == "ready"
+        assert restarted.tools_count == 6
+        assert mcp_container.tools.contains("mcp-fixture.echo")
+
+    run(scenario())
+
+
+@pytest.mark.parametrize(
+    ("mode", "contributions", "expected_code"),
+    [
+        ("no-tools", "[]", "MCP_CAPABILITY_UNSUPPORTED"),
+        ("invalid-schema", "[mcp-invalid.broken]", "MCP_TOOL_SCHEMA_INVALID"),
+        ("invalid-result", "[]", "MCP_INITIALIZE_FAILED"),
+        ("oversized-stdout", "[]", "PLUGIN_HOST_UNAVAILABLE"),
+    ],
+)
+def test_mcp_rejects_invalid_initialization_and_discovery(
+    tmp_path, mode, contributions, expected_code
+) -> None:
+    package = tmp_path / f"mcp-{mode}"
+    package.mkdir()
+    shutil.copyfile(MCP_FIXTURE / "server.py", package / "server.py")
+    (package / "plugin.yaml").write_text(
+        f"""
+id: mcp-invalid
+name: Invalid MCP Fixture
+version: 1.0.0
+contributes:
+  tools: {contributions}
+backend:
+  type: mcp
+  transport: stdio
+  command: python
+  args: [server.py, {mode}]
+  startup_timeout_seconds: 5
+  tool_timeout_seconds: 1
+""".strip(),
+        encoding="utf-8",
+    )
+    container = build_container()
+    container.plugins.install(package)
+    try:
+        with pytest.raises(ExtensionError) as exc:
+            container.plugins.enable("mcp-invalid")
+        assert exc.value.code == expected_code
+        assert container.plugins.get("mcp-invalid").status == "error"
+        assert container.plugins.get_host_status("mcp-invalid").status == "error"
+        assert not container.tools.contains("mcp-invalid.broken")
+    finally:
+        container.plugins.shutdown()

backend/tests/test_retrieval.py

@@ -436,6 +436,44 @@ def test_fts_pagination_is_not_truncated_at_one_thousand(vault) -> None:
     assert len(response.items) == 10
 
 
+def test_fts_score_threshold_filters_before_total(vault) -> None:
+    """score_threshold 先于计数与分页生效：total 反映过滤后数量，与 items 一致。
+
+    高阈值过滤掉全部结果时 total==0 且 items 为空，杜绝「空页但 total>0」的
+    不一致（审阅 P2-7）。
+    """
+    from app.retrieval.engine import engine
+    from app.services import note_service
+
+    # 10 个 block，含「目标」次数递增，bm25 分数各异，min-max 归一化后分数落在 [0,1]
+    markdown = "\n\n".join(f"{'目标' * i} 分隔内容" for i in range(1, 11))
+    asyncio.run(
+        note_service.create_note(title="阈值过滤", markdown=markdown, folder="", tags=[])
+    )
+
+    all_hits = asyncio.run(
+        engine.search(
+            SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=0.0)
+        )
+    )
+    filtered = asyncio.run(
+        engine.search(
+            SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=0.5)
+        )
+    )
+    none = asyncio.run(
+        engine.search(
+            SearchRequest(query="目标", mode=SearchMode.fts, limit=20, score_threshold=2.0)
+        )
+    )
+
+    assert all_hits.page.total >= 10
+    assert 0 < filtered.page.total < all_hits.page.total  # 阈值过滤掉部分而非全部
+    assert filtered.page.total == len(filtered.items)
+    assert none.page.total == 0
+    assert none.items == []
+
+
 # --------------------------------------------------------------------------- #
 # 审阅回归：PATCH tags 语义 / 向量-块一致性 / 过滤漏召回 / rebuild 语义与回滚
 # --------------------------------------------------------------------------- #

docs/README.md

@@ -30,7 +30,9 @@
 
 - [AI Core 与 Agent Core 开发说明](development/AI-Core与Agent-Core开发说明.md)
 - [Knowledge 与 Retrieval Core 开发说明](development/Knowledge与Retrieval-Core开发说明.md)
+- [Benchmark 开发说明](development/Benchmark开发说明.md)
 - [模型提供商与模型发现开发说明](development/模型提供商与模型发现开发说明.md)
+- [MCP Bridge 与 Plugin Host 开发说明](development/MCP-Bridge与Plugin-Host开发说明.md)
 - [前端壳子与接口层开发说明](development/前端壳子与接口层开发说明.md)
 - [前端写作体验优化开发说明](development/前端写作体验优化开发说明.md)
 - [前端视觉与轻量动效优化开发说明](development/前端视觉与轻量动效优化开发说明.md)

docs/architecture/AI笔记软件技术栈说明-团队版-v2.3.md

@@ -5,7 +5,7 @@
 > 适用范围：桌面客户端、本地知识库、RAG、Agent、Skill、多模型接入、多模态处理与可选云同步  
 > 目标读者：前端、Rust 桌面端、Python AI Core、算法、测试与后续接手项目的开发成员
 
-> 实施状态更新：2026-09-01。本文同时包含目标架构、当前实现和第二阶段接口基线。第一阶段已完成 Vue Web 联调前端、FastAPI、Knowledge/Retrieval、Agent/Tool/Permission、Skill/Plugin 声明式运行时、Mock/OpenAI-Compatible/Ollama Provider、DeepSeek/OpenAI 预设、模型发现及开发阶段 Fernet 凭据存储。Web Workspace 已通过 FastAPI 接入后端配置的真实单 Vault，第二阶段 Agent Trace 持久化、分页快照和可恢复 SSE 已完成。后续继续接入真实音频处理、MCP、Plugin Command/Settings、Provider 协议增强、Benchmark、文档导出、主题包、Trace 可视化、Mermaid 和函数图像。Tauri/Rust Host、Stronghold、原生多 Vault 文件系统和 Sync Server 仍未实现。
+> 实施状态更新：2026-09-03。本文同时包含目标架构、当前实现和第二阶段接口基线。第一阶段已完成 Vue Web 联调前端、FastAPI、Knowledge/Retrieval、Agent/Tool/Permission、Skill/Plugin 声明式运行时、Mock/OpenAI-Compatible/Ollama Provider、DeepSeek/OpenAI 预设、模型发现及开发阶段 Fernet 凭据存储。Web Workspace 已通过 FastAPI 接入后端配置的真实单 Vault；第二阶段 Agent Trace 持久化、分页快照、可恢复 SSE、stdio MCP Bridge 与隔离 Plugin Host 已完成；RAG Benchmark 检索评测（Dataset 加载、异步运行、SSE 进度、指标聚合与报告）已完成，Agent Benchmark 暂缓。后续继续接入真实音频处理、Plugin Command/Settings、Provider 协议增强、文档导出、主题包、Trace 可视化、Mermaid 和函数图像。Tauri/Rust Host、Stronghold、原生多 Vault 文件系统和 Sync Server 仍未实现。
 
 ---
 
@@ -1040,10 +1040,18 @@ Plugin Host 负责：
 
 内置 Plugin 可以使用相同的 Plugin Interface 注册能力，减少内置功能和社区扩展之间的接口差异。
 
+Python 包形式的 MCP Server 推荐使用固定版本的 `uvx --isolated --from <package>==<version> <command>` 启动，以隔离依赖并避免污染 AI Core 环境；包内脚本和非 Python Server 仍可使用受控 `command + args`。`uvx` 的虚拟环境不是安全沙箱，不能限制文件、网络、子进程或系统调用。
+
+面向社区或不可信 Plugin 开放前，Tauri/Rust Host 必须增加平台级沙箱、完整进程树回收、包来源/签名校验，并在首次安装或命令变化时向用户完整展示 executable 和参数、要求明确同意。当前 Python Host 的独立进程、环境裁剪和 Permission 只用于可信开发联调，不能替代这些生产安全门槛。
+
+在该门槛完成前，后端仅允许 `APP_ENVIRONMENT=development` 启动未沙箱化 MCP Host；生产环境统一返回 `MCP_TRUST_APPROVAL_REQUIRED`。Python `uvx` Server 在开发模式首次运行可能联网解析依赖，生产版本必须在安装/更新阶段预取并验证固定版本，正常运行阶段只使用已经准备好的环境。
+
 ### 12.5 MCP Bridge
 
 MCP Bridge 用于接入具有 MCP Server 接口的插件或外部工具服务。
 
+当前已实现本地 stdio 首版：Plugin Runtime 在授权后的启用阶段启动独立 Server 进程，完成 `initialize`、capability negotiation、分页 `tools/list`、`tools/call`、取消、超时、异常退出和 Host Restart。实现接受 `2025-11-25`、`2025-06-18`、`2025-03-26` 与 `2024-11-05` 协议版本；Streamable HTTP、Resource、Prompt、Sampling 与操作系统级沙箱仍属于后续范围。
+
 MCP Tool 进入系统后的调用路径为：
 
 ```text
@@ -1062,7 +1070,7 @@ Tool Registry 仍使用项目自己的 `ToolDefinition` 和 `ToolResult`。MCP B
 
 MCP 能力首先用于 Tool 和 Resource 类扩展。需要复杂 UI 的插件通过 Frontend Extension Slot 单独处理。
 
-第二阶段 MCP Bridge 至少覆盖以下协议边界：
+当前 stdio MCP Bridge 已覆盖以下协议边界：
 
 ```text
 Server Process / Connection Lifecycle
@@ -1074,7 +1082,7 @@ tools/call 与 ToolResult 映射
 健康检查与 Tool 注销
 ```
 
-首个宿主实现优先支持本地 `stdio` 传输；其他传输在兼容性测试后增加。外部 Server 的 Tool 名称进入项目注册表前添加 Plugin 命名空间，并校验 JSON Schema、权限和重复 ID。MCP 内容不得绕过项目自己的 Permission、超时、日志净化和结果大小限制。
+首个宿主实现支持本地 `stdio` 传输；其他传输在兼容性测试后增加。外部 Server 的 Tool 名称进入项目注册表前添加 Plugin 命名空间，并校验 JSON Schema、权限、重复 ID 及其与 Manifest Contribution 的一致性。MCP 调用复用项目自己的 Permission、超时、Agent Trace、日志净化和结果大小限制；子进程环境按白名单裁剪，不传入 Provider Key、Vault 或数据库路径。
 
 ### 12.6 Frontend Extension Slot
 
@@ -2092,9 +2100,13 @@ MRR
 Citation Hit Rate
 P50 Latency
 P95 Latency
+total_cases
+successful_cases
+failed_cases
+failure_rate
 ```
 
-Benchmark 参数、Embedding 模型、Reranker、数据集版本和运行环境需要一起记录，保证不同实验结果可以复现。
+失败样本按零分计入质量指标分母，报告同时输出样本构成字段标明实际分母。Benchmark 参数、Embedding 模型、Reranker、数据集版本和运行环境需要一起记录，保证不同实验结果可以复现。
 
 ### 20.3 Agent Benchmark
 
@@ -2321,7 +2333,7 @@ Markdown Workspace
 
 第一阶段 Plugin Runtime 已完成安装、启用、停用、权限和声明式 Tool 注册，建立 Skill 调用 Plugin Tool 的基础链路。Command、Settings 和 MCP 执行不计入第一阶段完成项。
 
-截至 2026-09-01，上述第一阶段后端链路和 Web 联调前端均已完成，第二阶段前置的 Workspace 去 Mock 联调及 Agent Trace 持久化/恢复接口也已完成。当前验证基线为后端 81 项测试、前端 27 项测试及生产构建通过。向量链路当前使用 `HashEmbeddingProvider` 验证工程正确性，真实 Embedding 召回质量不属于该测试结论。
+截至 2026-09-01，上述第一阶段后端链路和 Web 联调前端均已完成；第二阶段前置的 Workspace 去 Mock 联调、Agent Trace 持久化/恢复接口以及 stdio MCP Bridge / Plugin Host 也已完成。当前验证基线为后端 92 项测试、前端 27 项测试及生产构建通过。向量链路当前使用 `HashEmbeddingProvider` 验证工程正确性，真实 Embedding 召回质量不属于该测试结论。
 
 第二阶段在既有 Contract 上接入：
 
@@ -2331,7 +2343,7 @@ Multimodal
 └── pyannote.audio
 
 Extension / Model
-├── MCP Bridge
+├── MCP Bridge（stdio 首版已实现）
 ├── Plugin Command Contribution
 ├── Plugin Settings Contribution
 └── Provider Streaming / Tool Calling / Error Mapping 增强
@@ -2353,7 +2365,7 @@ Frontend Extension
 └── Plugin Settings UI
 ```
 
-上述列表描述第二阶段技术范围，不表示能力已经实现。每项功能必须继续经过现有 Service、Contract、Permission 和 Adapter 边界，不因 Demo 需要在 Vue 组件、Router 或 Agent Runtime 中直接绑定第三方协议。
+上述列表描述第二阶段技术范围，其中 stdio MCP Bridge 已实现，其余能力以各自开发说明的状态为准。每项功能必须继续经过现有 Service、Contract、Permission 和 Adapter 边界，不因 Demo 需要在 Vue 组件、Router 或 Agent Runtime 中直接绑定第三方协议。
 
 第三阶段处理：
 
@@ -2409,7 +2421,7 @@ Sync Server 按独立服务开发和部署，不进入桌面客户端核心启
 
 Python AI Core 未来作为 Tauri Sidecar 运行，当前由开发命令独立启动，FastAPI 提供本地接口。Knowledge Core 管理笔记结构；Retrieval Core 当前通过 FTS5、`HashEmbeddingProvider`、sqlite-vec、RRF 和轻量 Reranker 跑通混合检索，真实 Embedding 与正式 Benchmark 在第二阶段接入；Agent Runtime 使用 Tool Registry 操作知识库和任务，并将扩展 Agent Trace Contract 供可视化和 Benchmark 共用；Skill Runtime 将提示词、工具、权限和检索参数组装为可复用 Agent 配置。
 
-当前 Plugin Runtime 支持 Manifest、生命周期和声明式白名单 Tool Contribution；第二阶段通过 MCP Bridge 接入隔离 Tool，并增加 Command 与 Settings Contribution。Provider Adapter 当前实现 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama，第二阶段按统一行为测试完善 OpenAI Responses、Anthropic Messages 等协议。多模态目标方案使用 faster-whisper、pyannote.audio 和可选 emotion2vec；当前只读取 Host 预生成 transcript。
+当前 Plugin Runtime 支持 Manifest、生命周期和声明式白名单 Tool Contribution，并已通过 stdio MCP Bridge 接入独立进程 Tool、Host 状态与重启接口；Command 与 Settings Contribution 尚待后续阶段实现。Provider Adapter 当前实现 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama，第二阶段按统一行为测试完善 OpenAI Responses、Anthropic Messages 等协议。多模态目标方案使用 faster-whisper、pyannote.audio 和可选 emotion2vec；当前只读取 Host 预生成 transcript。
 
 第二阶段内容输出以 Document AST、Exporter Adapter、Mermaid Renderer 和 Function Plot Renderer 为共同边界，支持 HTML、PDF、DOCX 与静态图导出。Theme Package 使用 Manifest、Design Token 和受限 CSS 实现本地导入；联网主题市场不属于本阶段核心依赖。API Key 在 Web 联调期由 Fernet 开发存储加密保存，桌面版迁移到 Tauri Stronghold。多设备同步的目标方案为独立、可自托管的 Sync Server，目前尚未实现；本地核心功能不依赖 Sync Server。
 

docs/architecture/第二阶段团队分工表.md

@@ -746,8 +746,8 @@ Markdown
 - [ ] faster-whisper 能完成真实音频转写；
 - [ ] pyannote.audio 能生成说话人分段；
 - [ ] 两者能组合生成带时间戳和 Speaker 的 Transcript；
-- [ ] MCP Server 能通过 MCP Bridge 注册 Tool；
-- [ ] Agent 能调用 MCP Tool；
+- [x] MCP Server 能通过 MCP Bridge 注册 Tool；
+- [x] Agent 能调用 MCP Tool；
 - [ ] Plugin Command Contribution 后端可注册；
 - [ ] Plugin Settings Contribution 后端可解析；
 - [ ] Provider Adapter 的 Streaming / Tool Calling / Error Mapping 稳定；

docs/contracts/后端接口契约-开发版.md

@@ -1,6 +1,6 @@
 # 后端接口契约（开发版）
 
-> 更新日期：2026-08-31。本文档记录当前前后端联调使用的已实现接口；机器可读字段、校验规则和响应模型以 FastAPI 运行时生成的 OpenAPI 为准。第二阶段尚未实现的规划接口见 `第二阶段接口契约-开发版.md`，不要将规划路径视为当前服务能力。
+> 更新日期：2026-09-01。本文档记录当前前后端联调使用的已实现接口；机器可读字段、校验规则和响应模型以 FastAPI 运行时生成的 OpenAPI 为准。第二阶段尚未实现的规划接口见 `第二阶段接口契约-开发版.md`，不要将规划路径视为当前服务能力。
 
 ## 契约入口
 
@@ -76,6 +76,8 @@ Web 联调阶段只暴露后端通过 `APP_VAULT_PATH` 配置的单一 Vault，
 | POST | `/api/plugins/{plugin_id}/enable` | 启用 Plugin |
 | POST | `/api/plugins/{plugin_id}/disable` | 停用 Plugin |
 | PUT | `/api/plugins/{plugin_id}/permissions` | 设置 Plugin 已授权权限 |
+| GET | `/api/plugins/{plugin_id}/host` | 获取隔离 MCP Host 状态、工具数和协商信息 |
+| POST | `/api/plugins/{plugin_id}/host/restart` | 重启 MCP Host 并重新发现、校验和注册 Tool |
 | DELETE | `/api/plugins/{plugin_id}` | 卸载 Plugin |
 
 ### Provider
@@ -174,7 +176,7 @@ RunCancelled
 
 ## 当前实现状态
 
-更新至 2026-09-01：后端 81 项回归测试通过。
+更新至 2026-09-01：后端 92 项回归测试通过。
 
 - Chat、Agent Run、Agent Events、Tool 列表、Provider 配置生命周期、模型列表和连接测试已经接入 AI Core。
 - Agent Run/Event 已持久化到 SQLite；SSE 帧携带 sequence `id`，断线后可以回放缺失事件。Trace API 与 Benchmark 共用同一事件事实，并在入库前执行 Secret 脱敏和结果限长。
@@ -183,6 +185,7 @@ RunCancelled
 - Workspace 已接入后端配置的真实 Vault；文件树、笔记读写、文件/目录新建、重命名和删除不再使用前端 Mock Fallback。
 - Note Move 保留 `note_id`；Citation 的字符偏移统一使用 UTF-16 code unit，供浏览器编辑器直接定位。
 - Plugin 启用前必须通过权限接口记录授权，未知权限默认拒绝。
+- 本地 stdio MCP Server 已通过独立子进程接入 Plugin Runtime；Agent 只消费内部 Tool Contract。Host 支持 initialize、分页发现、调用、超时取消、状态查询、重启和异常退出后的 Tool 注销。
 - Attachment Tool 读取 Host 管理的 `attachments` 目录；音频接口读取 Host 生成的转写文本，真实本地语音模型在第二阶段接入。
 - 接入业务模块时保持当前路径和 Contract，不在 Router 中直接实现数据库、Provider 或 Agent 逻辑。
 

docs/contracts/第二阶段接口契约-开发版.md

@@ -2,7 +2,7 @@
 
 > 文档状态：接口冻结草案
 >
-> 更新日期：2026-08-31
+> 更新日期：2026-09-01
 >
 > 依据：`../architecture/第二阶段团队分工表.md`、`../architecture/AI笔记软件技术栈说明-团队版-v2.3.md`、`后端接口契约-开发版.md`
 
@@ -45,8 +45,8 @@
 | Transcription | POST | `/api/media/transcriptions/{job_id}/notes` | 计划新增 | 将 Transcript 写入 Knowledge Core |
 | Agent Trace | GET | `/api/agent/runs/{run_id}/events` | 已实现 | 支持游标恢复并增加模型与权限事件 |
 | Agent Trace | GET | `/api/agent/runs/{run_id}/trace` | 已实现 | 分页读取可回放 Trace 快照 |
-| Plugin Host | GET | `/api/plugins/{plugin_id}/host` | 计划新增 | 获取 MCP Host 健康状态 |
-| Plugin Host | POST | `/api/plugins/{plugin_id}/host/restart` | 计划新增 | 重启异常 Host 并重新发现 Tool |
+| Plugin Host | GET | `/api/plugins/{plugin_id}/host` | 已实现 | 获取 MCP Host 健康状态 |
+| Plugin Host | POST | `/api/plugins/{plugin_id}/host/restart` | 已实现 | 重启异常 Host 并重新发现 Tool |
 | Plugin Command | GET | `/api/plugin-contributions/commands` | 计划新增 | 获取前端可展示的 Command |
 | Plugin Command | POST | `/api/plugin-contributions/commands/{command_id}/execute` | 计划新增 | 受控执行 Command |
 | Plugin Settings | GET | `/api/plugins/{plugin_id}/settings` | 计划新增 | 获取 Schema 与非敏感配置 |
@@ -55,9 +55,9 @@
 | Provider | 现有路径 | `/api/providers/*`、`POST /api/chat` | 扩展 | 补齐协议能力和统一行为 |
 | Retrieval | GET/POST | `/api/index/status`、`/api/index/rebuild` | 扩展 | 暴露 Embedding 兼容状态并安全重建向量 |
 | Benchmark | GET | `/api/benchmarks/datasets` | 计划新增 | 枚举受控 Dataset |
-| Benchmark | POST | `/api/benchmarks/rag/runs` | 计划新增 | 创建 RAG Benchmark |
-| Benchmark | POST | `/api/benchmarks/agent/runs` | 计划新增 | 创建 Agent Benchmark |
-| Benchmark | GET | `/api/benchmarks/runs` | 计划新增 | 分页获取 Benchmark Run |
+| Benchmark | POST | `/api/benchmarks/rag/runs` | 已实现 | 创建 RAG Benchmark |
+| Benchmark | POST | `/api/benchmarks/agent/runs` | 暂缓 | 创建 Agent Benchmark（依赖 Agent Runtime 完成后交付） |
+| Benchmark | GET | `/api/benchmarks/runs` | 已实现 | 分页获取 Benchmark Run |
 | Benchmark | GET/POST | `/api/benchmarks/runs/{run_id}/*` | 计划新增 | 查询、订阅、取消和读取报告 |
 | Export | POST | `/api/exports` | 计划新增 | 创建 HTML/PDF/DOCX 导出任务 |
 | Export | GET | `/api/exports` | 计划新增 | 分页获取导出任务 |
@@ -427,7 +427,31 @@ class McpBridge(Protocol):
     async def stop(self, plugin_id: str) -> None: ...
 ```
 
-首个实现支持本地 `stdio`。Host 负责 initialize、capability negotiation、进程生命周期、超时、取消、stderr 隔离和异常退出后的 Tool 注销。
+首个实现已支持本地 `stdio`，按 MCP `2025-11-25` 发起 initialize，并兼容 `2025-06-18`、`2025-03-26` 和 `2024-11-05` 协商结果。Host 负责 capability negotiation、分页 `tools/list`、进程生命周期、超时取消、stderr 隔离和异常退出后的 Tool 注销。stdio 消息使用 UTF-8 单行 JSON-RPC，并在完整行进入内存前执行有界读取；当前不实现 Streamable HTTP。
+
+MCP Plugin 的 `backend` 增加：
+
+```yaml
+backend:
+  type: mcp
+  transport: stdio
+  command: uvx
+  args: [--isolated, --from, example-mcp==1.2.3, example-mcp]
+  startup_timeout_seconds: 60
+  tool_timeout_seconds: 30
+```
+
+命令通过参数数组直接启动，不经过 Shell。带路径的 executable 必须位于 Plugin 包内；PATH 中的命令可以按名称引用。Python 包形式的 MCP 推荐使用固定版本的 `uvx --isolated --from`，但 `uvx` 只隔离依赖而不是文件/网络/系统调用安全沙箱，非 Python Server 不强制使用。开发模式首次运行未缓存的 uvx 包可能联网解析，因此 startup 示例使用 60 秒；生产安装阶段必须预取并验证，运行阶段不得临时解析依赖。子进程只继承运行所需的系统环境变量，不继承 `OPENAI_API_KEY`、`APP_DB_PATH`、Vault 路径等宿主状态。Secret 注入留给阶段 D 的专用引用接口。
+
+在平台沙箱和可信命令许可完成前，`APP_ENVIRONMENT != development` 时启用 MCP Plugin 必须返回 `403 MCP_TRUST_APPROVAL_REQUIRED`，不得启动进程或注册 Tool。该门禁由后端执行，不能只依赖前端提示或文档约定。
+
+远端 Tool 的可选项目权限放在 MCP `_meta`：
+
+```json
+{ "_meta": { "notesagent/permission": "notes.read" } }
+```
+
+该权限必须属于项目已知权限并同时出现在 Plugin Manifest 中。发现结果必须与 `contributes.tools` 的命名空间 ID 完全一致；校验全部成功后才一次性发布到 Tool Registry。
 
 当前 Web 开发接口 `POST /api/plugins/install` 使用 `package_path`。桌面 Host 接入后，`ExtensionInstallRequest` 增加 `package_id`，由文件选择器产生临时包句柄；`package_path` 只在明确的 development 环境保留并标记 deprecated，生产构建拒绝任意前端路径。
 
@@ -460,6 +484,8 @@ error
 
 `POST /api/plugins/{plugin_id}/host/restart` 返回 `202 OperationResponse`。重启期间先注销旧 Tool，发现和校验全部成功后再一次性发布新 Tool 集合，避免半注册状态。
 
+当前实现还返回协商后的 `protocol_version`、`server_name` 和 `server_version`。单条协议消息上限为 2 MiB，单次 Tool Result 上限为 256 KiB；超限分别按 Host/Result 错误处理。Server 异常退出或发送无效 stdout 消息时，Host 进入 `unhealthy`，Plugin 进入 `error`，相关 Tool 立即注销。取消会同时通知 Server、移除 pending request 并唤醒本地等待线程。Restart 不得把 `installed`、`disabled` 或 `permission_required` Plugin 隐式启用，这些状态必须走 Enable。
+
 ### 7.3 Command Contribution 列表
 
 `GET /api/plugin-contributions/commands?location=command_palette`
@@ -608,6 +634,7 @@ MCP_CAPABILITY_UNSUPPORTED
 MCP_TOOL_SCHEMA_INVALID
 MCP_TOOL_CALL_FAILED
 MCP_TOOL_RESULT_TOO_LARGE
+MCP_TRUST_APPROVAL_REQUIRED
 PLUGIN_COMMAND_NOT_FOUND
 PLUGIN_COMMAND_CONTEXT_INVALID
 PLUGIN_SETTINGS_SCHEMA_INVALID
@@ -790,7 +817,7 @@ Dataset 从仓库或受控导入目录注册。API 不接受调用方提交任
 
 配置快照必须记录 Embedding model ID/version/dimension、Reranker、索引版本、Dataset Hash 和运行环境。
 
-### 9.5 创建 Agent Benchmark
+### 9.5 创建 Agent Benchmark（暂缓，未暴露接口）
 
 `POST /api/benchmarks/agent/runs`
 
@@ -824,12 +851,16 @@ RAG 和 Agent 创建接口均返回 `202 BenchmarkRun`：
   "metrics": null,
   "config_snapshot": {},
   "error": null,
+  "error_code": null,
   "created_at": "2026-08-31T10:30:00Z",
   "started_at": null,
   "completed_at": null
 }
 ```
 
+`status` 取值：`queued` → `running` → `completed` | `failed` | `cancelled`。失败/取消时 `error` 与
+`error_code` 只返回项目错误码与安全消息，不暴露第三方堆栈。
+
 公共接口：
 
 | 方法 | 路径 | 用途 |
@@ -840,6 +871,10 @@ RAG 和 Agent 创建接口均返回 `202 BenchmarkRun`：
 | POST | `/api/benchmarks/runs/{run_id}/cancel` | 取消运行 |
 | GET | `/api/benchmarks/runs/{run_id}/report` | 获取结构化完整报告 |
 
+SSE 事件流（`RunStarted` → `CaseCompleted`* → `RunCompleted` | `RunFailed` | `RunCancelled`）：
+`GET /api/benchmarks/runs/{run_id}/events` 支持 `Last-Event-ID` 与 `?after_sequence=` 游标恢复，
+`RunCompleted` / `RunFailed` / `RunCancelled` 为终止事件，收到后即断流。
+
 ### 9.7 指标 Contract
 
 RAG：
@@ -852,10 +887,17 @@ RAG：
   "mrr": 0.81,
   "citation_hit_rate": 0.89,
   "p50_latency_ms": 24.5,
-  "p95_latency_ms": 67.3
+  "p95_latency_ms": 67.3,
+  "total_cases": 50,
+  "successful_cases": 48,
+  "failed_cases": 2,
+  "failure_rate": 0.04
 }
 ```
 
+失败样本按零分计入质量指标分母，`total_cases` / `successful_cases` / `failed_cases` /
+`failure_rate` 让报告明确实际分母；延迟仅统计成功样本。
+
 Agent：
 
 ```json
@@ -879,8 +921,10 @@ BENCHMARK_DATASET_NOT_FOUND
 BENCHMARK_DATASET_INVALID
 BENCHMARK_CONFIG_INVALID
 BENCHMARK_INDEX_INCOMPATIBLE
+BENCHMARK_CAPACITY_EXCEEDED
 BENCHMARK_RUN_NOT_FOUND
 BENCHMARK_RUN_FAILED
+BENCHMARK_CASE_EVALUATION_FAILED
 ```
 
 ### 9.9 Retrieval Profile 与索引兼容

docs/development/AI-Core与Agent-Core开发说明.md

@@ -2,7 +2,7 @@
 
 > 本文档用于团队开发和模块联调，记录当前已经落地的核心边界与使用方式。
 
-> 更新日期：2026-09-01。第一阶段 AI Core、Agent Core、Extension Core 和 Model Core 主链路已经完成；第二阶段 Agent Trace 持久化和可恢复 SSE 已落地，后端当前回归基线为 81 项测试通过。
+> 更新日期：2026-09-01。第一阶段 AI Core、Agent Core、Extension Core 和 Model Core 主链路已经完成；第二阶段 Agent Trace 持久化、可恢复 SSE、stdio MCP Bridge 与隔离 Plugin Host 已落地，后端当前回归基线为 92 项测试通过。
 
 ## 当前实现
 
@@ -33,12 +33,14 @@ backend/app/
 │   ├── permissions.py   权限策略、确认请求和会话授权
 │   └── builtin_tools.py 无副作用的内置开发 Tool
 ├── extensions/
-│   └── runtime.py       Skill/Plugin Manifest、生命周期、依赖与 Tool Contribution
+│   ├── runtime.py       Skill/Plugin Manifest、生命周期、依赖与 Tool Contribution
+│   └── mcp.py           stdio JSON-RPC、MCP 生命周期、发现、调用与 Host 隔离
 └── container.py         AI Core 依赖组装
 
 backend/extensions/
 ├── skills/knowledge-assistant/   内置知识库 Skill
-└── plugins/text-tools/           内置示例 Plugin
+├── plugins/text-tools/           内置声明式 Plugin
+└── fixtures/mcp-echo/            离线 MCP Server 联调 Fixture
 ```
 
 Router 只负责 HTTP/SSE 与错误转换，不实现 Agent、Tool 或 Provider 业务逻辑。
@@ -57,6 +59,7 @@ Router 只负责 HTTP/SSE 与错误转换，不实现 Agent、Tool 或 Provider
 - SQLite Trace、分页快照与可恢复 SSE；
 - Skill Manifest、Prompt、Tool/Permission/模型能力解析；
 - Plugin Manifest、生命周期和 Tool Contribution；
+- stdio MCP Bridge、隔离进程生命周期、Tool 映射与 Host 健康状态；
 - Skill 调用内置 Tool 与 Plugin Tool；
 - 公共 Contract 和 API 接入。
 
@@ -65,7 +68,7 @@ Router 只负责 HTTP/SSE 与错误转换，不实现 Agent、Tool 或 Provider
 - Note、NoteBlock、Markdown Parser：由 Knowledge Core 提供；
 - FTS5、Vector、RRF、Reranker、Citation：由 Retrieval Core 提供；
 - 文件系统和 API Key 明文读取：由 Rust Host 提供；
-- MCP Plugin Host、Frontend Extension Slot：按技术基线放在第二阶段实现。
+- Frontend Extension Slot 与 Plugin Command/Settings：按第二阶段后续阶段实现。
 
 ## Provider
 
@@ -302,7 +305,7 @@ DELETE /api/skills/{skill_id}
 
 ### Plugin Runtime
 
-第一阶段 Plugin Runtime 完成 Manifest 校验、安装、启用、停用、卸载和 Tool Contribution。第三方代码不会直接 import 到 AI Core；当前 Declarative Plugin Host 只执行宿主实现的白名单 handler，MCP Host 留到第二阶段。
+第一阶段 Plugin Runtime 完成 Manifest 校验、安装、启用、停用、卸载和声明式 Tool Contribution。阶段 C 增加 stdio MCP Bridge：第三方代码不会直接 import 到 AI Core，而由独立子进程运行，通过换行分隔 JSON-RPC 完成 initialize、Tool 发现和调用。
 
 启用 Plugin 时将 Tool 注册到统一 Tool Registry，并标记 `source=plugin`；停用或异常时注销 Tool。启用中的 Skill 依赖某 Plugin Tool 时，Plugin 不能直接卸载。
 
@@ -315,11 +318,15 @@ GET    /api/plugins/{plugin_id}
 POST   /api/plugins/{plugin_id}/enable
 POST   /api/plugins/{plugin_id}/disable
 PUT    /api/plugins/{plugin_id}/permissions
+GET   /api/plugins/{plugin_id}/host
+POST  /api/plugins/{plugin_id}/host/restart
 DELETE /api/plugins/{plugin_id}
 ```
 
 Plugin Manifest 中的权限只是声明，不代表已经授权。带权限的 Plugin 安装后进入 `permission_required`，Host 必须通过权限接口记录用户授权，之后才能启用。JSON Schema 在安装阶段校验，Tool 调用时再次校验实际参数。
 
+MCP Tool 进入 Registry 前统一增加 `<plugin_id>.<remote_name>` 命名空间。Server 声明的 `notesagent/permission` 必须属于已知权限并出现在 Plugin Manifest；发现集合还必须与 Manifest Contribution 完全一致。启用失败会回滚全部 Tool 并关闭子进程，异常退出会把 Plugin 标记为 `error` 并立即注销对应 Tool。详细实现和 Fixture 操作见 [MCP Bridge 与 Plugin Host 开发说明](MCP-Bridge与Plugin-Host开发说明.md)。
+
 内置示例 `text-tools` 注册 `text.uppercase`。内置 `knowledge-assistant` Skill 同时声明 `notes.search` 和 `text.uppercase`，用于验证完整链路：
 
 ```text
@@ -342,4 +349,4 @@ Skill Manifest
 - Task 已持久化到 SQLite；Attachment Tool 读取 Host 管理目录中的 UTF-8 文件。
 - `audio.transcribe` 当前消费 Host 预生成的 transcript；faster-whisper 与说话人分离仍按技术基线在第二阶段接入。
 - Extension 安装记录暂存内存；后续接入持久化 Registry 与版本升级流程。
-- 当前 Plugin Host 只支持内置声明式白名单 handler；MCP Bridge、独立进程健康检查与 UI Contribution 在第二阶段实现。
+- 当前 Plugin Host 支持内置声明式 handler 和本地 stdio MCP Server；Streamable HTTP、OS 级沙箱、Plugin Command/Settings 与 UI Contribution 留在后续阶段。

docs/development/Benchmark开发说明.md

@@ -0,0 +1,76 @@
+# Benchmark 开发说明
+
+> 所属模块：Knowledge / Retrieval Core（后端，负责人 yxx）。RAG Benchmark 已交付；Agent Benchmark 暂缓，待 Agent Runtime 完成后在同一契约下补齐。
+
+## 定位
+
+Benchmark Service 用受控 Dataset 对检索引擎做可复现评测：创建即返回 queued、后台 asyncio.Task 执行、SSE 实时推送进度、结束后产出结构化报告。CLI、测试与前端报告页复用同一 Service，不各自实现指标。
+
+## 接口
+
+| 方法 | 路径 | 用途 |
+| --- | --- | --- |
+| GET | `/api/benchmarks/datasets?kind=rag` | 枚举受控目录下的 Dataset 元信息 |
+| POST | `/api/benchmarks/rag/runs` | 创建 RAG Benchmark（202） |
+| GET | `/api/benchmarks/runs?kind=&status=&limit=&offset=` | 分页获取运行记录 |
+| GET | `/api/benchmarks/runs/{run_id}` | 状态与指标摘要 |
+| GET | `/api/benchmarks/runs/{run_id}/events` | SSE 进度与 Case 结果 |
+| POST | `/api/benchmarks/runs/{run_id}/cancel` | 取消运行 |
+| GET | `/api/benchmarks/runs/{run_id}/report` | 结构化完整报告 |
+
+Agent Benchmark 的 `/api/benchmarks/agent/runs` 未暴露（暂缓），不在 OpenAPI 注册占位接口。
+
+## Dataset
+
+Dataset 来自 `settings.benchmark_datasets_path`（默认 `backend/data/benchmarks`），API 不接受调用方提交任意路径。按文件名 stem 精确匹配 `{dataset_id}.json`，与请求无关文件的损坏（JSON 语法错误、UTF-8 解码错误、顶层非对象）不会阻断加载；只有目标文件本身损坏才返回 `BENCHMARK_DATASET_INVALID`。
+
+RAG Case 结构：`case_id`、`query`、`expected_note_ids`、`expected_block_ids`、`citation_required`、`tags`。`citation_required=true` 时必须声明 `expected_block_ids`，否则无法计算 Citation Hit Rate。
+
+## 运行生命周期
+
+`queued → running → completed | failed | cancelled`。
+
+- 创建时校验索引兼容性：索引非空、Embedding model/dim 与当前引擎一致、vector/hybrid 时向量索引非空；不满足返回 `BENCHMARK_INDEX_INCOMPATIBLE`（409），避免把环境/索引错误误判为检索质量差。
+- 内存注册表上限 `MAX_RUNS=100`，超限只淘汰终态 run；满容量且全为活动 run 时返回 `BENCHMARK_CAPACITY_EXCEEDED`（429）。
+- 失败/取消只向公开响应暴露项目错误码与安全消息，详细异常进入日志，不通过 HTTP/SSE 返回。
+
+## 指标
+
+RAG 按 (mode, case, repeat) 逐样本计算，再按 mode 聚合：
+
+- 质量：`hit_at_1`、`hit_at_5`、`recall_at_k`、`mrr`、`citation_hit_rate`；
+- 延迟：`p50_latency_ms`、`p95_latency_ms`（仅统计成功样本）；
+- 样本构成：`total_cases`、`successful_cases`、`failed_cases`、`failure_rate`。
+
+失败样本按零分计入质量指标分母，报告据此可知实际分母，避免把执行失败误判为检索质量差。
+
+## 事件与 SSE
+
+事件流：`RunStarted → CaseCompleted* → RunCompleted | RunFailed | RunCancelled`。
+
+`GET /api/benchmarks/runs/{run_id}/events` 支持 `Last-Event-ID` 与 `?after_sequence=` 游标恢复（复用 Agent SSE 的解析逻辑），`RunCompleted` / `RunFailed` / `RunCancelled` 为终止事件，收到后断流。
+
+## 错误码
+
+```text
+BENCHMARK_DATASET_NOT_FOUND
+BENCHMARK_DATASET_INVALID
+BENCHMARK_INDEX_INCOMPATIBLE
+BENCHMARK_CAPACITY_EXCEEDED
+BENCHMARK_RUN_NOT_FOUND
+BENCHMARK_RUN_FAILED
+BENCHMARK_CASE_EVALUATION_FAILED
+```
+
+## 配置快照
+
+报告与运行记录保存 `config_snapshot`：dataset hash/version、modes、retrieval 参数、Embedding model/version/dim、Reranker、索引元数据、App 版本与环境、Python 版本，保证不同实验结果可复现。
+
+## 测试
+
+```powershell
+cd backend
+uv run pytest -q
+```
+
+`tests/test_benchmark.py` 覆盖数据集注册与校验、指标纯函数、端到端运行、取消、索引兼容、容量与失败样本聚合；`tests/test_retrieval.py` 覆盖 FTS 阈值与分页 total 一致性。

docs/development/Knowledge与Retrieval-Core开发说明.md

@@ -3,7 +3,7 @@
 > 本文档用于团队开发和模块联调，记录 Knowledge Core / Retrieval Core 已经落地的
 > 模块边界、数据模型、接口与使用方式，对应分工表中的杨星萱。
 
-> 更新日期：2026-09-01。第一阶段 Knowledge/Retrieval 主链路已经完成，并已接入 Agent Tool Registry；完整后端回归基线为 81 项测试通过。
+> 更新日期：2026-09-01。第一阶段 Knowledge/Retrieval 主链路已经完成，并已接入 Agent Tool Registry；完整后端回归基线为 92 项测试通过。
 
 ## 当前实现
 
@@ -198,7 +198,7 @@ cd backend
 uv run pytest -q
 ```
 
-当前后端完整测试共 71 个用例通过（单元 + 端到端）。测试通过 `tests/conftest.py` 的 autouse fixture 把
+当前后端完整测试共 120 个用例通过（单元 + 端到端）。测试通过 `tests/conftest.py` 的 autouse fixture 把
 数据目录/DB/Vault 重定向到临时目录，不读写真实 `backend/data`，任何本机状态下结果确定。
 
 ## 配置
@@ -232,4 +232,6 @@ rag.search
 - Embedding / Reranker 为轻量实现，后续替换为真实模型（接口不变）。
 - 小语料下 hybrid 检索召回偏宽（向量 Top-K 覆盖全部 block），可加相关性阈值收紧。
 - 重建为同步 + 全量，后续接入增量索引与异步任务队列。
-- 检索 Benchmark 待建立。
+- RAG Benchmark 已建立：`POST /api/benchmarks/rag/runs` 创建即返回 queued、后台 Task 执行，
+  通过 SSE 实时推送进度，报告含逐 Case 结果与 `total_cases` / `successful_cases` / `failed_cases` / `failure_rate`。
+- Agent Benchmark 暂缓，待 Agent Runtime 完成后交付。

docs/development/MCP-Bridge与Plugin-Host开发说明.md

@@ -0,0 +1,282 @@
+# MCP Bridge 与 Plugin Host 开发说明
+
+> 更新日期：2026-09-01。本文记录第二阶段阶段 C 已实现的本地 stdio MCP Bridge、隔离 Plugin Host、Tool Contract 转换和离线测试方式。Plugin Command 与 Settings 属于阶段 D，不在本文实现范围内。
+
+## 1. 目标与实现状态
+
+阶段 C 的目标是让外部 MCP Server 进入既有 Plugin、Tool、Permission、Agent 和 Trace 链路，同时避免 Agent Runtime、前端或 Benchmark 直接依赖 MCP 原始消息。
+
+当前链路：
+
+```text
+Plugin Manifest
+→ Plugin Runtime
+→ 独立 stdio MCP Server 进程
+→ initialize / capability negotiation
+→ tools/list 分页发现与校验
+→ NotesAgent ToolDefinition
+→ Tool Registry / Permission Manager
+→ Agent Runtime / Agent Trace
+```
+
+已经实现：
+
+- 本地 stdio 子进程启动、关闭和异常退出检测；
+- UTF-8、换行分隔的 JSON-RPC 2.0 消息；
+- initialize、协议版本与 tools capability 协商；
+- `notifications/initialized`；
+- 分页 `tools/list`；
+- `tools/call`、业务错误与 JSON-RPC 错误转换；
+- 超时和 `notifications/cancelled`；
+- Tool 命名空间、JSON Schema、权限和 Manifest 集合校验；
+- Host 状态查询、重启和异常后的 Tool 自动注销；
+- stderr 隔离、环境变量裁剪、消息及结果大小限制；
+- 无网络、无密钥的确定性 MCP Fixture。
+
+实现依据为 MCP 官方 [Lifecycle 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle)、[Transports 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports) 和 [Tools 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/server/tools)。
+
+## 2. 代码位置
+
+```text
+backend/app/extensions/mcp.py
+    stdio 进程、JSON-RPC、MCP 生命周期、发现、调用和 Host 状态
+
+backend/app/extensions/runtime.py
+    Plugin Manifest、权限、MCP Tool 批量注册/回滚和生命周期集成
+
+backend/app/agent/tools.py
+    内部 Tool 参数校验、结构化执行错误和线程安全 Registry
+
+backend/extensions/fixtures/mcp-echo/
+    确定性 stdio MCP Server 与 Plugin Manifest
+```
+
+## 3. Plugin Manifest
+
+MCP Plugin 的后端配置示例：
+
+```yaml
+id: example-mcp
+name: Example MCP
+version: 1.0.0
+permissions:
+  - notes.read
+contributes:
+  tools:
+    - example-mcp.search
+backend:
+  type: mcp
+  transport: stdio
+  command: uvx
+  args: [--isolated, --from, example-mcp==1.2.3, example-mcp]
+  startup_timeout_seconds: 60
+  tool_timeout_seconds: 30
+```
+
+约束：
+
+- 阶段 C 只接受 `type: mcp` 与 `transport: stdio`；
+- 命令和参数通过数组直接传给 `subprocess.Popen`，不经过 Shell；
+- Python 包形式的 MCP Server 推荐使用 `uvx --isolated --from <package>==<version> <command>`，固定版本并与 NotesAgent 项目环境隔离；
+- Plugin 包内自带且不需要第三方依赖的 Python 脚本可以使用 `python server.py`；Node、Rust 等 Server 继续使用各自受控启动器，因此 Host 不强制所有 MCP 都经过 `uvx`；
+- PATH 中的 executable 使用名称，例如 `uvx`、`python`、`node`；
+- manifest 中带目录的 executable 必须解析到 Plugin 包内部；
+- `contributes.tools` 使用 `<plugin_id>.<remote_name>`；
+- 安装阶段只读 Manifest，不启动第三方进程；
+- 完成用户授权后，`enable` 才启动 Host。
+
+## 4. 生命周期
+
+### 4.1 启动
+
+启用 MCP Plugin 时依次执行：
+
+1. 检查 Plugin 声明权限是否全部获得授权；
+2. 检查 Manifest 声明的 Tool ID 是否与现有 Registry 冲突；
+3. 启动独立 stdio Server；
+4. 发送 `initialize`；
+5. 校验协商版本和 `tools` capability；
+6. 发送 `notifications/initialized`；
+7. 分页读取 `tools/list`；
+8. 校验全部 Tool；
+9. 确认发现集合与 Manifest 完全一致；
+10. 将完整集合注册到 Tool Registry；
+11. Plugin 和 Host 进入 `ready`。
+
+任何步骤失败都会注销本轮已注册 Tool、关闭子进程并把 Plugin 标记为 `error`，不会留下半启用状态。
+
+### 4.2 停止与异常退出
+
+停用、卸载或应用关闭时，先注销 Tool，再关闭 stdin，等待 Server 正常退出。超时后依次 terminate 和 kill。
+
+Server 异常退出、stdout 出现非 JSON-RPC 内容或发送超大协议消息时：
+
+- 未完成请求返回 `PLUGIN_HOST_UNAVAILABLE`；
+- Host 进入 `unhealthy`；
+- Plugin 进入 `error`；
+- 对应 Tool 从 Registry 中立即注销；
+- 用户可以调用 Host Restart 接口重新协商和发现。
+
+Server 发送 `notifications/tools/list_changed` 时不会直接信任新集合。当前实现先把 Host 标记为不健康并注销旧 Tool，要求通过 Restart 重新执行完整发现与校验。
+
+## 5. Tool Contract 转换
+
+MCP Tool：
+
+```json
+{
+  "name": "search",
+  "description": "Search notes",
+  "inputSchema": { "type": "object", "properties": {} },
+  "_meta": { "notesagent/permission": "notes.read" }
+}
+```
+
+进入系统后转换为：
+
+```json
+{
+  "name": "example-mcp.search",
+  "description": "Search notes",
+  "parameters": { "type": "object", "properties": {} },
+  "permission": "notes.read",
+  "source": "plugin"
+}
+```
+
+转换规则：
+
+- 远端名称必须能转换为合法且稳定的项目 Tool ID；
+- `inputSchema` 必须是有效的 object JSON Schema；
+- `additionalProperties`、`patternProperties` 等动态字段先由完整 JSON Schema 校验，Pydantic 参数载体不会再次误拒绝合法字段；
+- `_meta.notesagent/permission` 必须属于项目已知权限；
+- Tool 权限必须同时出现在 Plugin Manifest 中；
+- Agent 仍通过 Tool Registry 执行参数校验、Permission、超时和 Trace；
+- MCP `structuredContent` 存在时映射为内部 output；否则保留为受控 `content` 数组；
+- MCP `isError: true` 映射为 `MCP_TOOL_CALL_FAILED`；
+- 结果超过 256 KiB 映射为 `MCP_TOOL_RESULT_TOO_LARGE`。
+
+## 6. 隔离与安全边界
+
+当前隔离是“独立进程 + 协议边界”，不是完整的操作系统沙箱。
+
+`uvx` 解决的是 Python 工具依赖隔离：它等价于 `uv tool run`，在 uv 缓存中使用可丢弃的独立虚拟环境。它不会限制 Server 读取用户文件、访问网络、创建子进程或调用系统 API，因此不能代替安全沙箱。当前开发模式下，首次 `enable` 尚未缓存的包可能访问包索引，因此示例使用 60 秒启动上限；生产实现不得依赖该行为，必须在用户确认后的安装/更新阶段预取和验证固定版本，运行阶段只启动已准备好的环境。
+
+已经执行的保护：
+
+- 第三方模块不 import 到 AI Core；
+- 子进程 `cwd` 固定为 Plugin 包目录；
+- 不使用 Shell 拼接命令；
+- 不把 Provider API Key、`APP_DB_PATH`、Vault 路径和其他宿主环境变量传入子进程；
+- stderr 与 JSON-RPC stdout 分离，stderr 不进入 API 和 Agent Trace；
+- stdout 只能发送合法 MCP JSON-RPC；
+- stdout 在读取完整行前即应用有界读取，单条协议消息上限 2 MiB；stderr 也按固定大小分块读取；
+- 单次 Tool Result 上限 256 KiB；
+- MCP Tool 不绕过 Permission Manager 和 Agent Tool Timeout。
+- 调用被 Agent 取消时，同时通知 Server 并唤醒本地 pending Queue，阻塞线程不会继续占用线程池直至远端超时。
+
+当前尚未提供容器、受限系统账户、seccomp、Windows AppContainer 或 macOS Sandbox，因此 Plugin 进程仍具有当前操作系统用户授予的一般文件访问能力。正式社区插件分发或“一键安装”前必须完成以下安全门槛：
+
+- 由 Tauri/Rust Host 统一启动进程并提供平台级文件、网络、子进程和资源配额限制；
+- 安装/更新时完整展示 executable 与全部参数，明确警告并要求用户主动确认；
+- 固定包来源和版本，增加包哈希/签名与可信发布者校验；
+- 默认禁止访问 Vault、凭据和宿主环境，只通过声明 Permission 与受控 Host API 授权；
+- 关闭 Host 时终止完整进程树，不只结束直接子进程。
+
+在这些门槛完成前，当前 MCP Host 只适用于内置 Fixture、团队可信插件和开发联调；不得把它描述为可以安全执行任意社区代码。上述安装确认要求遵循 MCP [SEP-1024](https://modelcontextprotocol.io/seps/1024-mcp-client-security-requirements-for-local-server-)；`uvx` 行为依据 uv 官方 [Using tools](https://docs.astral.sh/uv/guides/tools/) 文档。
+
+后端通过 `APP_ENVIRONMENT` 强制该边界：只有 `development` 可以启动当前未沙箱化的 MCP Host；其他环境返回 `403 MCP_TRUST_APPROVAL_REQUIRED`，且不会创建进程或注册 Tool。后续 Tauri/Rust Host 提供沙箱与绑定完整命令摘要的可信许可后，再替换此临时门禁。
+
+## 7. Host API
+
+```http
+GET  /api/plugins/{plugin_id}/host
+POST /api/plugins/{plugin_id}/host/restart
+```
+
+状态响应包含：
+
+```text
+plugin_id
+backend_type / transport
+status
+tools_count
+started_at / last_seen_at
+protocol_version
+server_name / server_version
+error
+```
+
+状态值：
+
+```text
+stopped
+starting
+ready
+unhealthy
+error
+```
+
+Restart 返回 `202 OperationResponse`。接口返回前已完成本地 Host 重启和 Tool 重新发现；`message` 中给出最终 Host 状态。Restart 只用于运行中或异常 Host；用户主动停用、尚未启用或等待授权的 Plugin 返回 `409 PLUGIN_HOST_UNAVAILABLE`，必须通过 Enable 明确启动。
+
+## 8. 离线 Fixture
+
+Fixture 位于：
+
+```text
+backend/extensions/fixtures/mcp-echo
+```
+
+它提供：
+
+- `mcp-fixture.echo`：返回 structuredContent；
+- `mcp-fixture.fail`：返回 `isError: true`；
+- `mcp-fixture.sleep`：验证超时和取消；
+- `mcp-fixture.large`：验证结果大小上限；
+- `mcp-fixture.environment`：验证宿主 Secret/路径没有进入子进程；
+- `mcp-fixture.exit`：验证异常退出、Tool 注销和 Restart。
+
+Fixture 的 `tools/list` 使用两页响应，用于覆盖分页发现。测试还会启动缺少 tools capability、返回无效 Schema/initialize result，以及输出超长无换行 stdout 的变体。
+
+## 9. 验证
+
+```powershell
+cd backend
+uv run python -m compileall -q app
+uv run pytest
+
+cd ../frontend
+pnpm test
+pnpm type-check
+pnpm build
+```
+
+阶段 C 新增测试覆盖：
+
+- initialize、版本和 capability negotiation；
+- 分页 `tools/list` 与命名空间映射；
+- Permission、JSON Schema 与 Contribution 集合；
+- Tool 成功、业务错误、结果过大和超时；
+- Agent 取消后 pending 等待线程及时释放；
+- `additionalProperties` 动态参数保持 JSON Schema 语义；
+- Agent Runtime 调用 MCP Tool 并写入正式 Trace；
+- Secret/Vault 环境隔离；
+- Server 异常退出、Tool 注销和 Host Restart；
+- 缺少 capability、无效 initialize result、无效 MCP Schema 和超长无换行 stdout；
+- disabled Plugin 不会被 Host Restart 隐式重新启用；
+- OpenAPI 发布 Host 状态和重启路径。
+
+## 10. 当前边界与后续阶段
+
+阶段 C 不包含：
+
+- Streamable HTTP MCP transport；
+- Resources、Prompts、Sampling、Elicitation 和 MCP Tasks；
+- Plugin Command 与 Settings Contribution；
+- Secret Reference 注入；
+- Plugin Registry 持久化、签名与社区来源校验；
+- 操作系统级沙箱；
+- 一键安装前的完整命令展示与确认 UI；
+- Tool 列表热更新的无中断替换。
+
+阶段 D 将在当前 Plugin Runtime 上继续增加 Command、Settings、Secret Contract 和命名空间 Storage，不修改 Agent 使用内部 Tool Contract 的原则。

docs/development/前端壳子与接口层开发说明.md

@@ -187,12 +187,12 @@ pnpm build
 ```text
 pnpm build       passed
 pnpm test        27 passed
-uv run pytest    81 passed
+uv run pytest    92 passed
 preview smoke    HTTP 200
 git diff --check passed
 ```
 
-当前前端使用 Vitest 执行 Store、Workspace API Adapter、SSE 恢复游标、文件树、编辑器组件、智能体标签、轻量动效约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题测试；`pnpm build` 同时执行 `vue-tsc -b` 与 Vite 生产构建。后端测试出现过 `.pytest_cache` 无法写入的 Windows 权限警告，不影响 81 项测试结果，也不涉及产品代码。
+当前前端使用 Vitest 执行 Store、Workspace API Adapter、SSE 恢复游标、文件树、编辑器组件、智能体标签、轻量动效约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题测试；`pnpm build` 同时执行 `vue-tsc -b` 与 Vite 生产构建。后端测试出现过 `.pytest_cache` 无法写入的 Windows 权限警告，不影响 92 项测试结果，也不涉及产品代码。
 
 Vite 当前会提示 Chat 与 Workspace 的部分异步 Chunk 超过 500 kB，这是 Milkdown、CodeMirror、KaTeX 和 Shiki 等编辑/渲染依赖带来的性能优化项，不影响构建成功或功能正确性；进入桌面打包前应通过手动分包或更细粒度动态加载继续优化。
 

docs/development/模型提供商与模型发现开发说明.md

@@ -104,4 +104,4 @@ pnpm build
 
 自动化验证覆盖 Provider 预设、OpenAI-Compatible `/models` 请求与鉴权头、模型映射、前端自动刷新、排序去重及按 Provider 隔离错误。生产构建同时执行 Vue 和 TypeScript 类型检查。
 
-当前完整回归基线：后端 81 项测试、前端 27 项测试通过，前端类型检查和生产构建通过。Provider 配置目前仍保存在内存 Registry，AI Core 重启后需要重新创建；凭据密文会保留。OpenAI Responses 与 Anthropic Messages Adapter 尚未实现，设置页正式预设不会使用这两种协议。
+当前完整回归基线：后端 92 项测试、前端 27 项测试通过，前端类型检查和生产构建通过。Provider 配置目前仍保存在内存 Registry，AI Core 重启后需要重新创建；凭据密文会保留。OpenAI Responses 与 Anthropic Messages Adapter 尚未实现，设置页正式预设不会使用这两种协议。

docs/retrospectives/后端全面审阅问题与修复复盘.md

@@ -4,7 +4,7 @@
 > 审阅范围：FastAPI、Knowledge / Retrieval Core、Agent Core、Extension Core、Provider Adapter、公共接口和后端开发文档。
 > 文档用途：记录问题形成原因、实际影响、修复判断和落地方案，供后续开发文档、比赛材料与技术博客使用。
 
-> 2026-09-01 状态补充：本文记录的缺陷均保持修复。此后又加入 Provider 预设、模型发现、DeepSeek/OpenAI 凭据解析、Fernet 加密存储和 Agent Trace 持久化，当前完整后端回归基线为 81 项测试通过。
+> 2026-09-01 状态补充：本文记录的缺陷均保持修复。此后又加入 Provider 预设、模型发现、DeepSeek/OpenAI 凭据解析、Fernet 加密存储、Agent Trace 持久化和 stdio MCP Plugin Host，当前完整后端回归基线为 92 项测试通过。
 
 ## 1. 审阅结论
 

frontend/pnpm-workspace.yaml

@@ -0,0 +1,2 @@
+allowBuilds:
+  esbuild: true

frontend/src/contracts/index.ts

@@ -255,6 +255,22 @@ export type PluginStatus =
   | 'dependency_missing'
   | 'permission_required'
 
+export type PluginHostState = 'stopped' | 'starting' | 'ready' | 'unhealthy' | 'error'
+
+export interface PluginHostStatus {
+  plugin_id: string
+  backend_type: 'mcp' | 'internal_rpc' | 'none'
+  transport: 'stdio' | 'http' | 'none'
+  status: PluginHostState
+  tools_count: number
+  started_at?: string | null
+  last_seen_at?: string | null
+  protocol_version?: string | null
+  server_name?: string | null
+  server_version?: string | null
+  error?: string | null
+}
+
 export interface PluginContribution {
   type: 'tool' | 'command' | 'importer' | 'exporter' | 'sidebar_panel' | 'settings_section'
   id: string
@@ -539,7 +555,14 @@ export interface ApiPlugin {
       panels: string[]
       settings_sections: string[]
     }
-    backend: { type: 'mcp' | 'internal_rpc' | 'none'; transport: 'stdio' | 'http' | 'none' }
+    backend: {
+      type: 'mcp' | 'internal_rpc' | 'none'
+      transport: 'stdio' | 'http' | 'none'
+      command?: string | null
+      args?: string[]
+      startup_timeout_seconds?: number
+      tool_timeout_seconds?: number
+    }
   }
   status: PluginStatus
   enabled: boolean

frontend/src/services/pluginService.ts

@@ -1,5 +1,5 @@
 import apiClient from './apiClient'
-import type { ApiPlugin, OperationResponse, Plugin, PluginContribution } from '@/contracts'
+import type { ApiPlugin, OperationResponse, Plugin, PluginContribution, PluginHostStatus } from '@/contracts'
 
 function toPlugin(plugin: ApiPlugin): Plugin {
   const { manifest } = plugin
@@ -54,6 +54,14 @@ export async function grantPluginPermissions(pluginId: string, permissions: stri
   return toPlugin(await apiClient.put<ApiPlugin>(`/api/plugins/${pluginId}/permissions`, { permissions }))
 }
 
+export async function getPluginHostStatus(pluginId: string): Promise<PluginHostStatus> {
+  return apiClient.get(`/api/plugins/${pluginId}/host`)
+}
+
+export async function restartPluginHost(pluginId: string): Promise<OperationResponse> {
+  return apiClient.post(`/api/plugins/${pluginId}/host/restart`)
+}
+
 export async function uninstallPlugin(pluginId: string): Promise<OperationResponse> {
   return apiClient.delete(`/api/plugins/${pluginId}`)
 }