fix(extension): 收紧插件命令运行时契约

Reviewed-on: #8

README.md

@@ -118,7 +118,7 @@ cd frontend
 pnpm test
 ```
 
-当前回归基线为后端 81 项测试、前端 27 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。
+当前回归基线为后端 126 项测试、前端 29 项测试，且 TypeScript 类型检查和生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。
 
 构建产物位于 `frontend/dist`，该目录不提交到 Git。
 
@@ -132,6 +132,8 @@ 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 映射、状态与错误边界 |
+| [Plugin Command 与 Settings](docs/development/Plugin-Command与Settings开发说明.md) | Command Registry、Settings Schema、Secret 引用与联调边界 |
 | [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

@@ -2,7 +2,7 @@
 
 FastAPI + Pydantic 的本地 AI Core / Agent Core。项目使用 uv 管理依赖和虚拟环境。
 
-当前实现包含 Knowledge/Retrieval、Chat、Agent Runtime、Tool/Permission、Skill/Plugin、Provider Adapter、任务、索引和开发阶段凭据加密存储。Provider 支持 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama；OpenAI Responses、Anthropic Messages、MCP 独立 Host 和真实语音模型仍属于后续阶段。
+当前实现包含 Knowledge/Retrieval、Chat、Agent Runtime、Tool/Permission、Skill/Plugin、stdio MCP Host、Plugin Command/Settings、Provider Adapter、任务、索引和开发阶段凭据加密存储。Provider 支持 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama；OpenAI Responses、Anthropic Messages、操作系统级 Plugin 沙箱和真实语音模型仍属于后续阶段。
 
 ```powershell
 uv sync
@@ -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>` 注入；不要把真实密钥写入仓库。
+当前基线为 126 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入；不要把真实密钥写入仓库。`plugin.*` 是 Plugin Settings 的保留凭据命名空间，通用 Provider 凭据接口不能读写。
 
 团队接口清单见 `../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
@@ -10,6 +11,7 @@ from jsonschema import Draft202012Validator
 from jsonschema.exceptions import ValidationError as JsonSchemaValidationError
 
 from app.contracts import ToolCall, ToolDefinition, ToolResult
+from app.schema_security import reject_external_schema_references
 
 ToolExecutor = Callable[[BaseModel, "ToolExecutionContext"], Any | Awaitable[Any]]
 
@@ -17,6 +19,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 +33,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 +55,40 @@ class ToolRegistry:
         arguments_model: type[BaseModel],
         executor: ToolExecutor,
     ) -> None:
-        if definition.name in self._tools:
+        Draft202012Validator.check_schema(definition.parameters)
-            raise ValueError(f"Tool already registered: {definition.name}")
+        reject_external_schema_references(definition.parameters)
-        self._tools[definition.name] = RegisteredTool(
+        with self._lock:
-            definition=definition,
+            if definition.name in self._tools:
-            arguments_model=arguments_model,
+                raise ValueError(f"Tool already registered: {definition.name}")
-            executor=executor,
+            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:
+        with self._lock:
-            return self._tools[name]
+            try:
-        except KeyError as exc:
+                return self._tools[name]
-            raise ToolNotFoundError(name) from exc
+            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 [
+        with self._lock:
-            item.definition.model_copy(deep=True)
+            return [
-            for name, item in self._tools.items()
+                item.definition.model_copy(deep=True)
-            if names is None or name in names
+                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 +128,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/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,13 @@ def build_container() -> ApplicationContainer:
     tools = ToolRegistry()
     register_builtin_tools(tools)
 
-    plugins = PluginRuntime(tools)
+    plugins = PluginRuntime(
+        tools,
+        credentials=credentials,
+        # 当前 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

@@ -1,6 +1,6 @@
 from datetime import datetime
 from enum import Enum
-from typing import Any, Literal
+from typing import Annotated, Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, SecretStr
 
@@ -417,6 +417,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 +464,194 @@ 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 PluginCommandLocation(str, Enum):
+    command_palette = "command_palette"
+    context_menu = "context_menu"
+    toolbar = "toolbar"
+
+
+class PluginCommand(Contract):
+    command_id: str
+    plugin_id: str
+    title: str
+    description: str = ""
+    icon: str | None = None
+    locations: list[PluginCommandLocation] = Field(default_factory=list)
+    when: list[str] = Field(default_factory=list)
+    parameters: dict[str, Any] = Field(default_factory=dict)
+    enabled: bool = True
+
+
+class PluginCommandListResponse(Contract):
+    items: list[PluginCommand] = Field(default_factory=list)
+
+
+class PluginCommandContext(Contract):
+    vault_id: str | None = None
+    note_id: str | None = None
+    file_path: str | None = None
+    selection: str | None = None
+
+
+class PluginCommandExecuteRequest(Contract):
+    arguments: dict[str, Any] = Field(default_factory=dict)
+    context: PluginCommandContext = Field(default_factory=PluginCommandContext)
+
+
+class PluginNotificationEffectPayload(Contract):
+    level: Literal["info", "success", "warning", "error"] = "info"
+    message: str = Field(min_length=1, max_length=4096)
+
+
+class PluginNavigateEffectPayload(Contract):
+    route: Literal[
+        "vault-entry",
+        "workspace",
+        "search",
+        "chat",
+        "agent",
+        "tasks",
+        "skills",
+        "plugins",
+        "themes",
+        "settings",
+    ]
+
+
+class PluginRefreshEffectPayload(Contract):
+    scope: Literal["workspace", "commands", "settings", "plugins"]
+
+
+class PluginJobEffectPayload(Contract):
+    job_id: str = Field(
+        min_length=1,
+        max_length=128,
+        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$",
+    )
+
+
+class PluginNoEffectPayload(Contract):
+    pass
+
+
+class PluginNoEffect(Contract):
+    type: Literal["none"] = "none"
+    payload: PluginNoEffectPayload = Field(default_factory=PluginNoEffectPayload)
+
+
+class PluginNotificationEffect(Contract):
+    type: Literal["notification"] = "notification"
+    payload: PluginNotificationEffectPayload
+
+
+class PluginNavigateEffect(Contract):
+    type: Literal["navigate"] = "navigate"
+    payload: PluginNavigateEffectPayload
+
+
+class PluginRefreshEffect(Contract):
+    type: Literal["refresh"] = "refresh"
+    payload: PluginRefreshEffectPayload
+
+
+class PluginJobEffect(Contract):
+    type: Literal["job"] = "job"
+    payload: PluginJobEffectPayload
+
+
+PluginCommandEffect = Annotated[
+    PluginNoEffect
+    | PluginNotificationEffect
+    | PluginNavigateEffect
+    | PluginRefreshEffect
+    | PluginJobEffect,
+    Field(discriminator="type"),
+]
+
+PLUGIN_COMMAND_EFFECT_TYPES = (
+    PluginNoEffect,
+    PluginNotificationEffect,
+    PluginNavigateEffect,
+    PluginRefreshEffect,
+    PluginJobEffect,
+)
+
+
+class PluginCommandResult(Contract):
+    command_id: str
+    status: Literal["completed"] = "completed"
+    effect: PluginCommandEffect = Field(default_factory=PluginNoEffect)
+
+
+class PluginSettingType(str, Enum):
+    string = "string"
+    number = "number"
+    boolean = "boolean"
+    select = "select"
+    secret = "secret"
+
+
+class PluginSettingField(Contract):
+    key: str
+    label: str
+    description: str = ""
+    type: PluginSettingType
+    required: bool = False
+    default: Any | None = None
+    minimum: float | None = None
+    maximum: float | None = None
+    options: list[str] = Field(default_factory=list)
+
+
+class PluginSecretState(Contract):
+    configured: bool = False
+
+
+class PluginSettingsSchema(Contract):
+    plugin_id: str
+    schema_version: int = Field(ge=1)
+    fields: list[PluginSettingField] = Field(default_factory=list)
+    values: dict[str, Any] = Field(default_factory=dict)
+    secrets: dict[str, PluginSecretState] = Field(default_factory=dict)
+
+
+class PluginSettingsUpdateRequest(Contract):
+    schema_version: int = Field(ge=1)
+    values: dict[str, Any] = Field(default_factory=dict)
+
+
+class PluginSecretWriteRequest(Contract):
+    secret: SecretStr
+
+
+class PluginSecretStatus(Contract):
+    plugin_id: str
+    key: str
+    configured: bool
+
+
 class PluginPermissionGrantRequest(Contract):
     permissions: list[str] = Field(default_factory=list)
 

backend/app/extensions/__init__.py

@@ -1,8 +1,12 @@
-from app.extensions.runtime import (
+from app.extensions.errors import ExtensionError
-    AgentConfiguration,
+from app.extensions.runtime import AgentConfiguration, PluginRuntime, SkillRuntime
-    ExtensionError,
+from app.extensions.mcp import McpBridge, McpBridgeError
-    PluginRuntime,
-    SkillRuntime,
-)
 
-__all__ = ["AgentConfiguration", "ExtensionError", "PluginRuntime", "SkillRuntime"]
+__all__ = [
+    "AgentConfiguration",
+    "ExtensionError",
+    "McpBridge",
+    "McpBridgeError",
+    "PluginRuntime",
+    "SkillRuntime",
+]

backend/app/extensions/contributions.py

@@ -0,0 +1,847 @@
+"""Plugin Command Registry 与 Settings/Secret 命名空间存储。"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import inspect
+import json
+import math
+import re
+import threading
+from collections import deque
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from time import perf_counter
+from typing import Any, Awaitable, Callable, Literal
+
+from jsonschema import Draft202012Validator
+from jsonschema.exceptions import SchemaError, ValidationError as JsonSchemaValidationError
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from app.config import get_settings
+from app.contracts import (
+    PLUGIN_COMMAND_EFFECT_TYPES,
+    PluginCommand,
+    PluginCommandContext,
+    PluginCommandEffect,
+    PluginCommandLocation,
+    PluginCommandResult,
+    PluginSecretState,
+    PluginSecretStatus,
+    PluginSettingField,
+    PluginSettingType,
+    PluginSettingsSchema,
+)
+from app.extensions.errors import ExtensionError
+from app.providers.credentials import CredentialStoreError, EncryptedCredentialStore
+from app.schema_security import (
+    SchemaReferenceError,
+    reject_external_schema_references,
+)
+
+_CONTRIBUTION_ID = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
+_SETTING_KEY = re.compile(r"^[a-z][a-z0-9._-]{0,127}$")
+_HOST_ICONS = {"bolt", "document", "edit", "link", "refresh", "search", "setting"}
+_WHEN_TOKENS = {
+    "workspace.has_vault",
+    "editor.has_note",
+    "editor.has_selection",
+}
+_CONTEXT_KEYS = {"vault_id", "note_id", "file_path", "selection"}
+_WHEN_CONTEXT = {
+    "workspace.has_vault": "vault_id",
+    "editor.has_note": "note_id",
+    "editor.has_selection": "selection",
+}
+
+
+class PluginCommandSpec(BaseModel):
+    """包内 commands.yaml 的宿主侧声明，不直接暴露 handler。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    command_id: str
+    title: str
+    description: str = ""
+    icon: str | None = None
+    locations: list[PluginCommandLocation] = Field(default_factory=list)
+    when: list[str] = Field(default_factory=list)
+    context: list[Literal["vault_id", "note_id", "file_path", "selection"]] = Field(
+        default_factory=list
+    )
+    parameters: dict[str, Any] = Field(
+        default_factory=lambda: {
+            "type": "object",
+            "properties": {},
+            "additionalProperties": False,
+        }
+    )
+    permission: str | None = None
+    secrets: list[str] = Field(default_factory=list)
+    handler: Literal["echo", "uppercase_selection"] | None = None
+    mcp_tool: str | None = None
+    timeout_seconds: int = Field(default=30, ge=1, le=120)
+
+    @model_validator(mode="after")
+    def validate_execution_target(self) -> "PluginCommandSpec":
+        if (self.handler is None) == (self.mcp_tool is None):
+            raise ValueError("Command must declare exactly one handler or mcp_tool target.")
+        return self
+
+
+CommandExecutor = Callable[
+    [dict[str, Any], dict[str, Any]],
+    PluginCommandEffect | Awaitable[PluginCommandEffect],
+]
+PluginSecretResolver = Callable[[str], str | None]
+
+
+@dataclass(slots=True)
+class _RegisteredCommand:
+    command: PluginCommand
+    spec: PluginCommandSpec
+    executor: CommandExecutor
+
+
+@dataclass(frozen=True, slots=True)
+class PluginCommandAuditEvent:
+    """不记录参数与上下文的轻量审计事件，避免把正文或 Secret 写入日志。"""
+
+    command_id: str
+    plugin_id: str
+    status: Literal["completed", "failed"]
+    duration_ms: int
+    error_code: str | None
+    created_at: datetime
+
+
+class CommandRegistry:
+    """只发布已启用 Plugin 的受控 Command Contribution。"""
+
+    def __init__(self) -> None:
+        self._commands: dict[str, _RegisteredCommand] = {}
+        self._audit: deque[PluginCommandAuditEvent] = deque(maxlen=500)
+        self._lock = threading.RLock()
+
+    def register(
+        self,
+        plugin_id: str,
+        spec: PluginCommandSpec,
+        executor: CommandExecutor,
+    ) -> None:
+        validate_command_spec(plugin_id, spec)
+        command = PluginCommand(
+            command_id=spec.command_id,
+            plugin_id=plugin_id,
+            title=spec.title,
+            description=spec.description,
+            icon=spec.icon,
+            locations=spec.locations,
+            when=spec.when,
+            parameters=spec.parameters,
+            enabled=True,
+        )
+        with self._lock:
+            if spec.command_id in self._commands:
+                raise ExtensionError(
+                    "PLUGIN_COMMAND_CONFLICT",
+                    f"Plugin command is already registered: {spec.command_id}",
+                    status_code=409,
+                    details={"command_id": spec.command_id},
+                )
+            self._commands[spec.command_id] = _RegisteredCommand(command, spec, executor)
+
+    def unregister(self, command_id: str) -> None:
+        with self._lock:
+            self._commands.pop(command_id, None)
+
+    def contains(self, command_id: str) -> bool:
+        with self._lock:
+            return command_id in self._commands
+
+    def list(self, location: PluginCommandLocation | None = None) -> list[PluginCommand]:
+        with self._lock:
+            items = [
+                item.command.model_copy(deep=True)
+                for item in self._commands.values()
+                if location is None or location in item.command.locations
+            ]
+        return sorted(items, key=lambda item: item.command_id)
+
+    def audit_events(self) -> list[PluginCommandAuditEvent]:
+        """返回有界审计快照；事件刻意不包含 arguments/context/effect。"""
+
+        with self._lock:
+            return list(self._audit)
+
+    async def execute(
+        self,
+        command_id: str,
+        arguments: dict[str, Any],
+        context: PluginCommandContext,
+    ) -> PluginCommandResult:
+        with self._lock:
+            registered = self._commands.get(command_id)
+        if registered is None:
+            raise ExtensionError(
+                "PLUGIN_COMMAND_NOT_FOUND",
+                f"Plugin command is not registered or enabled: {command_id}",
+                status_code=404,
+                details={"command_id": command_id},
+            )
+        started_at = perf_counter()
+        try:
+            Draft202012Validator(registered.spec.parameters).validate(arguments)
+        except JsonSchemaValidationError as exc:
+            error = ExtensionError(
+                "PLUGIN_COMMAND_ARGUMENT_INVALID",
+                "Plugin command arguments do not match the declared schema.",
+                details={"command_id": command_id, "path": list(exc.path)},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error from exc
+
+        raw_context = context.model_dump(exclude_none=True)
+        missing = [
+            token
+            for token in registered.spec.when
+            if not raw_context.get(_WHEN_CONTEXT[token])
+        ]
+        if missing:
+            error = ExtensionError(
+                "PLUGIN_COMMAND_CONTEXT_INVALID",
+                "Plugin command context does not satisfy its when conditions.",
+                details={"command_id": command_id, "missing": missing},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error
+        scoped_context = {
+            key: raw_context[key]
+            for key in registered.spec.context
+            if key in raw_context
+        }
+        try:
+            effect = registered.executor(dict(arguments), scoped_context)
+            if inspect.isawaitable(effect):
+                effect = await asyncio.wait_for(
+                    effect, timeout=registered.spec.timeout_seconds
+                )
+        except TimeoutError as exc:
+            error = ExtensionError(
+                "PLUGIN_COMMAND_TIMEOUT",
+                "Plugin command execution timed out.",
+                status_code=504,
+                details={"command_id": command_id},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error from exc
+        except ExtensionError as exc:
+            self._record_audit(registered, started_at, exc.code)
+            raise
+        except Exception as exc:
+            error = ExtensionError(
+                "PLUGIN_COMMAND_EXECUTION_FAILED",
+                "Plugin command execution failed.",
+                status_code=502,
+                details={"command_id": command_id},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error from exc
+        if not isinstance(effect, PLUGIN_COMMAND_EFFECT_TYPES):
+            error = ExtensionError(
+                "PLUGIN_COMMAND_RESULT_INVALID",
+                "Plugin command returned an invalid effect.",
+                status_code=502,
+                details={"command_id": command_id},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error
+        try:
+            encoded_effect = json.dumps(effect.model_dump(mode="json"), ensure_ascii=False)
+        except (TypeError, ValueError) as exc:
+            error = ExtensionError(
+                "PLUGIN_COMMAND_RESULT_INVALID",
+                "Plugin command returned a non-serializable effect.",
+                status_code=502,
+                details={"command_id": command_id},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error from exc
+        if len(encoded_effect.encode("utf-8")) > 64 * 1024:
+            error = ExtensionError(
+                "PLUGIN_COMMAND_RESULT_TOO_LARGE",
+                "Plugin command effect exceeds the 64 KiB response limit.",
+                status_code=502,
+                details={"command_id": command_id},
+            )
+            self._record_audit(registered, started_at, error.code)
+            raise error
+        self._record_audit(registered, started_at, None)
+        return PluginCommandResult(command_id=command_id, effect=effect)
+
+    def _record_audit(
+        self,
+        registered: _RegisteredCommand,
+        started_at: float,
+        error_code: str | None,
+    ) -> None:
+        event = PluginCommandAuditEvent(
+            command_id=registered.command.command_id,
+            plugin_id=registered.command.plugin_id,
+            status="failed" if error_code else "completed",
+            duration_ms=max(0, round((perf_counter() - started_at) * 1000)),
+            error_code=error_code,
+            created_at=datetime.now(UTC),
+        )
+        with self._lock:
+            self._audit.append(event)
+
+
+class PluginSettingsDefinition(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    section_id: str
+    schema_version: int = Field(ge=1)
+    fields: list[PluginSettingField] = Field(default_factory=list)
+
+
+class PluginSettingsStore:
+    """非敏感值写入插件命名空间；Secret 只保存加密凭据引用。"""
+
+    def __init__(self, credentials: EncryptedCredentialStore) -> None:
+        self.credentials = credentials
+        self._lock = threading.RLock()
+
+    @staticmethod
+    def _path() -> Path:
+        return get_settings().data_dir / "plugins" / "settings.json"
+
+    def get(
+        self, plugin_id: str, definition: PluginSettingsDefinition
+    ) -> PluginSettingsSchema:
+        with self._lock:
+            entry = self._entry(self._read(), plugin_id)
+            stored_values = entry.get("values", {})
+            secret_refs = entry.get("secret_refs", {})
+            if not isinstance(stored_values, dict) or not isinstance(secret_refs, dict):
+                raise self._storage_format_error(plugin_id)
+            validated_refs = self._validate_secret_refs(plugin_id, secret_refs)
+            values = {
+                field.key: field.default
+                for field in definition.fields
+                if field.type != PluginSettingType.secret and field.default is not None
+            }
+            allowed_values = {
+                field.key
+                for field in definition.fields
+                if field.type != PluginSettingType.secret
+            }
+            fields = {field.key: field for field in definition.fields}
+            for key, value in stored_values.items():
+                if key not in allowed_values:
+                    continue
+                try:
+                    _validate_setting_value(fields[key], value)
+                except ExtensionError as exc:
+                    raise self._storage_format_error(plugin_id) from exc
+                values[key] = value
+            secrets: dict[str, PluginSecretState] = {}
+            for field in definition.fields:
+                if field.type != PluginSettingType.secret:
+                    continue
+                reference = validated_refs.get(field.key)
+                secrets[field.key] = PluginSecretState(
+                    configured=isinstance(reference, str) and self._has_secret(reference)
+                )
+            return PluginSettingsSchema(
+                plugin_id=plugin_id,
+                schema_version=definition.schema_version,
+                fields=definition.fields,
+                values=values,
+                secrets=secrets,
+            )
+
+    def runtime_values(
+        self, plugin_id: str, definition: PluginSettingsDefinition
+    ) -> dict[str, Any]:
+        """返回可供 Command 使用的完整普通设置，并拦截未配置的必填项。"""
+
+        schema = self.get(plugin_id, definition)
+        missing = [
+            field.key
+            for field in definition.fields
+            if field.required
+            and field.type != PluginSettingType.secret
+            and field.key not in schema.values
+        ]
+        if missing:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_REQUIRED",
+                "Required Plugin settings have not been configured.",
+                status_code=409,
+                details={"plugin_id": plugin_id, "fields": missing},
+            )
+        return schema.values
+
+    def update(
+        self,
+        plugin_id: str,
+        definition: PluginSettingsDefinition,
+        schema_version: int,
+        values: dict[str, Any],
+    ) -> PluginSettingsSchema:
+        if schema_version != definition.schema_version:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_VERSION_CONFLICT",
+                "Plugin settings schema version is out of date.",
+                status_code=409,
+                details={
+                    "plugin_id": plugin_id,
+                    "requested_version": schema_version,
+                    "current_version": definition.schema_version,
+                },
+            )
+        fields = {field.key: field for field in definition.fields}
+        unknown = sorted(set(values) - set(fields))
+        if unknown:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_FIELD_INVALID",
+                "Plugin settings contain unknown fields.",
+                details={"plugin_id": plugin_id, "fields": unknown},
+            )
+        secret_keys = sorted(
+            key for key in values if fields[key].type == PluginSettingType.secret
+        )
+        if secret_keys:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_FIELD_INVALID",
+                "Secret fields must use the dedicated Secret endpoint.",
+                details={"plugin_id": plugin_id, "fields": secret_keys},
+            )
+        for key, value in values.items():
+            _validate_setting_value(fields[key], value)
+
+        with self._lock:
+            data = self._read()
+            entry = self._entry(data, plugin_id, create=True)
+            current = entry.get("values", {})
+            if not isinstance(current, dict):
+                raise self._storage_format_error(plugin_id)
+            entry["values"] = current
+            current.update(values)
+            effective = {
+                field.key: field.default
+                for field in definition.fields
+                if field.type != PluginSettingType.secret and field.default is not None
+            }
+            effective.update(current)
+            missing = [
+                field.key
+                for field in definition.fields
+                if field.required
+                and field.type != PluginSettingType.secret
+                and field.key not in effective
+            ]
+            if missing:
+                raise ExtensionError(
+                    "PLUGIN_SETTINGS_FIELD_INVALID",
+                    "Required Plugin settings are missing.",
+                    details={"plugin_id": plugin_id, "fields": missing},
+                )
+            entry["schema_version"] = definition.schema_version
+            self._write(data)
+        return self.get(plugin_id, definition)
+
+    def put_secret(
+        self,
+        plugin_id: str,
+        definition: PluginSettingsDefinition,
+        key: str,
+        secret: str,
+    ) -> PluginSecretStatus:
+        _secret_field(definition, plugin_id, key)
+        if not secret:
+            raise ExtensionError(
+                "PLUGIN_SECRET_VALUE_INVALID",
+                "Plugin secret cannot be empty.",
+                details={"plugin_id": plugin_id, "key": key},
+            )
+        if len(secret.encode("utf-8")) > 64 * 1024:
+            raise ExtensionError(
+                "PLUGIN_SECRET_VALUE_INVALID",
+                "Plugin secret exceeds the 64 KiB limit.",
+                details={"plugin_id": plugin_id, "key": key},
+            )
+        reference = _secret_reference(plugin_id, key)
+        with self._lock:
+            data = self._read()
+            entry = self._entry(data, plugin_id, create=True)
+            refs = entry.get("secret_refs", {})
+            if not isinstance(refs, dict):
+                raise self._storage_format_error(plugin_id)
+            self._validate_secret_refs(plugin_id, refs)
+            entry["secret_refs"] = refs
+            try:
+                previous = self.credentials.resolve(reference)
+                self.credentials.put(reference, secret)
+            except CredentialStoreError as exc:
+                raise ExtensionError(
+                    "PLUGIN_SECRET_STORE_ERROR", str(exc), status_code=500
+                ) from exc
+            refs[key] = reference
+            entry["schema_version"] = definition.schema_version
+            try:
+                self._write(data)
+            except ExtensionError:
+                # 普通设置落盘失败时恢复凭据旧值，避免产生不可达的新 Secret。
+                try:
+                    if previous is None:
+                        self.credentials.delete(reference)
+                    else:
+                        self.credentials.put(reference, previous)
+                except CredentialStoreError:
+                    pass
+                raise
+        return PluginSecretStatus(plugin_id=plugin_id, key=key, configured=True)
+
+    def delete_secret(
+        self,
+        plugin_id: str,
+        definition: PluginSettingsDefinition,
+        key: str,
+    ) -> PluginSecretStatus:
+        _secret_field(definition, plugin_id, key)
+        with self._lock:
+            data = self._read()
+            entry = self._entry(data, plugin_id)
+            refs = entry.get("secret_refs", {})
+            if not isinstance(refs, dict):
+                raise self._storage_format_error(plugin_id)
+            self._validate_secret_refs(plugin_id, refs)
+            reference = _secret_reference(plugin_id, key)
+            had_reference = refs.pop(key, None) is not None
+            if plugin_id in data and had_reference:
+                self._write(data)
+            try:
+                self.credentials.delete(reference)
+            except CredentialStoreError as exc:
+                if had_reference:
+                    refs[key] = reference
+                    try:
+                        self._write(data)
+                    except ExtensionError as rollback_exc:
+                        raise ExtensionError(
+                            "PLUGIN_STORAGE_ERROR",
+                            "Plugin Secret deletion failed and its reference could not be restored.",
+                            status_code=500,
+                            details={"plugin_id": plugin_id, "key": key},
+                        ) from rollback_exc
+                raise ExtensionError(
+                    "PLUGIN_SECRET_STORE_ERROR", str(exc), status_code=500
+                ) from exc
+        return PluginSecretStatus(plugin_id=plugin_id, key=key, configured=False)
+
+    def resolve_secret(
+        self, plugin_id: str, definition: PluginSettingsDefinition, key: str
+    ) -> str | None:
+        _secret_field(definition, plugin_id, key)
+        with self._lock:
+            entry = self._entry(self._read(), plugin_id)
+            refs = entry.get("secret_refs", {})
+            if not isinstance(refs, dict):
+                raise self._storage_format_error(plugin_id)
+            reference = self._validate_secret_refs(plugin_id, refs).get(key)
+        try:
+            return self.credentials.resolve(reference) if isinstance(reference, str) else None
+        except CredentialStoreError as exc:
+            raise ExtensionError(
+                "PLUGIN_SECRET_STORE_ERROR", str(exc), status_code=500
+            ) from exc
+
+    def remove_plugin(self, plugin_id: str) -> None:
+        with self._lock:
+            data = self._read()
+            entry = data.pop(plugin_id, None)
+            references: list[str] = []
+            if entry is not None and not isinstance(entry, dict):
+                raise self._storage_format_error(plugin_id)
+            if entry is not None:
+                refs = entry.get("secret_refs", {})
+                if not isinstance(refs, dict):
+                    raise self._storage_format_error(plugin_id)
+                references = list(self._validate_secret_refs(plugin_id, refs).values())
+            if entry is not None:
+                self._write(data)
+            try:
+                self.credentials.delete_many(references)
+            except CredentialStoreError as exc:
+                if entry is not None:
+                    data[plugin_id] = entry
+                    try:
+                        self._write(data)
+                    except ExtensionError as rollback_exc:
+                        raise ExtensionError(
+                            "PLUGIN_STORAGE_ERROR",
+                            "Plugin uninstall failed and its Settings namespace could not be restored.",
+                            status_code=500,
+                            details={"plugin_id": plugin_id},
+                        ) from rollback_exc
+                raise ExtensionError(
+                    "PLUGIN_SECRET_STORE_ERROR", str(exc), status_code=500
+                ) from exc
+
+    def _validate_secret_refs(
+        self, plugin_id: str, refs: dict[Any, Any]
+    ) -> dict[str, str]:
+        validated: dict[str, str] = {}
+        for key, reference in refs.items():
+            if (
+                not isinstance(key, str)
+                or not _SETTING_KEY.fullmatch(key)
+                or not isinstance(reference, str)
+                or reference != _secret_reference(plugin_id, key)
+            ):
+                raise self._storage_format_error(plugin_id)
+            validated[key] = reference
+        return validated
+
+    def _has_secret(self, reference: str) -> bool:
+        try:
+            return self.credentials.has(reference)
+        except CredentialStoreError as exc:
+            raise ExtensionError(
+                "PLUGIN_SECRET_STORE_ERROR", str(exc), status_code=500
+            ) from exc
+
+    @staticmethod
+    def _storage_format_error(plugin_id: str) -> ExtensionError:
+        return ExtensionError(
+            "PLUGIN_STORAGE_ERROR",
+            "Plugin settings namespace has an invalid format.",
+            status_code=500,
+            details={"plugin_id": plugin_id},
+        )
+
+    def _entry(
+        self,
+        data: dict[str, dict[str, Any]],
+        plugin_id: str,
+        *,
+        create: bool = False,
+    ) -> dict[str, Any]:
+        entry = data.get(plugin_id)
+        if entry is None:
+            if create:
+                data[plugin_id] = {}
+                return data[plugin_id]
+            return {}
+        if not isinstance(entry, dict):
+            raise self._storage_format_error(plugin_id)
+        return entry
+
+    def _read(self) -> dict[str, dict[str, Any]]:
+        path = self._path()
+        if not path.exists():
+            return {}
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ExtensionError(
+                "PLUGIN_STORAGE_ERROR",
+                "Plugin settings storage cannot be loaded.",
+                status_code=500,
+            ) from exc
+        if not isinstance(value, dict):
+            raise ExtensionError(
+                "PLUGIN_STORAGE_ERROR",
+                "Plugin settings storage has an invalid format.",
+                status_code=500,
+            )
+        return value
+
+    def _write(self, value: dict[str, dict[str, Any]]) -> None:
+        path = self._path()
+        temporary = path.with_suffix(".tmp")
+        try:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            temporary.write_text(
+                json.dumps(value, ensure_ascii=False, sort_keys=True),
+                encoding="utf-8",
+            )
+            temporary.replace(path)
+        except OSError as exc:
+            try:
+                temporary.unlink(missing_ok=True)
+            except OSError:
+                pass
+            raise ExtensionError(
+                "PLUGIN_STORAGE_ERROR",
+                "Plugin settings storage cannot be written.",
+                status_code=500,
+            ) from exc
+
+
+def validate_settings_definition(
+    plugin_id: str, definition: PluginSettingsDefinition
+) -> None:
+    if not _CONTRIBUTION_ID.fullmatch(definition.section_id):
+        raise _settings_schema_error(plugin_id, "Settings section id is invalid.")
+    if not definition.section_id.startswith(f"{plugin_id}."):
+        raise _settings_schema_error(
+            plugin_id, "Settings section id must use the Plugin namespace."
+        )
+    keys: set[str] = set()
+    for field in definition.fields:
+        if not _SETTING_KEY.fullmatch(field.key) or field.key in keys:
+            raise _settings_schema_error(plugin_id, f"Invalid or duplicate setting key: {field.key}")
+        keys.add(field.key)
+        if field.type == PluginSettingType.select and not field.options:
+            raise _settings_schema_error(plugin_id, f"Select setting requires options: {field.key}")
+        if field.type != PluginSettingType.select and field.options:
+            raise _settings_schema_error(plugin_id, f"Only select settings accept options: {field.key}")
+        if field.type != PluginSettingType.number and (
+            field.minimum is not None or field.maximum is not None
+        ):
+            raise _settings_schema_error(plugin_id, f"Only number settings accept bounds: {field.key}")
+        if any(
+            bound is not None and not math.isfinite(bound)
+            for bound in (field.minimum, field.maximum)
+        ):
+            raise _settings_schema_error(
+                plugin_id, f"Number setting bounds must be finite: {field.key}"
+            )
+        if field.minimum is not None and field.maximum is not None and field.minimum > field.maximum:
+            raise _settings_schema_error(plugin_id, f"Setting bounds are reversed: {field.key}")
+        if field.type == PluginSettingType.secret and field.default is not None:
+            raise _settings_schema_error(plugin_id, f"Secret settings cannot declare defaults: {field.key}")
+        if field.default is not None:
+            try:
+                _validate_setting_value(field, field.default)
+            except ExtensionError as exc:
+                raise _settings_schema_error(plugin_id, exc.message) from exc
+
+
+def validate_command_spec(plugin_id: str, spec: PluginCommandSpec) -> None:
+    if not _CONTRIBUTION_ID.fullmatch(spec.command_id) or not spec.command_id.startswith(
+        f"{plugin_id}."
+    ):
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command id must be valid and use the Plugin namespace.",
+            details={"plugin_id": plugin_id, "command_id": spec.command_id},
+        )
+    if not spec.locations:
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command must declare at least one location.",
+            details={"command_id": spec.command_id},
+        )
+    if len(spec.locations) != len(set(spec.locations)):
+        raise ExtensionError("PLUGIN_COMMAND_INVALID", "Plugin command locations must be unique.")
+    if len(spec.when) != len(set(spec.when)) or len(spec.context) != len(set(spec.context)):
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command when/context entries must be unique.",
+        )
+    if len(spec.secrets) != len(set(spec.secrets)):
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command Secret entries must be unique.",
+            details={"command_id": spec.command_id},
+        )
+    unknown_when = sorted(set(spec.when) - _WHEN_TOKENS)
+    if unknown_when:
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command declares unsupported when tokens.",
+            details={"command_id": spec.command_id, "when": unknown_when},
+        )
+    required_context = {_WHEN_CONTEXT[token] for token in spec.when}
+    if not required_context.issubset(set(spec.context)):
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command context must include every field required by when.",
+            details={"command_id": spec.command_id},
+        )
+    if not set(spec.context).issubset(_CONTEXT_KEYS):
+        raise ExtensionError("PLUGIN_COMMAND_INVALID", "Plugin command context is invalid.")
+    if spec.icon and spec.icon not in _HOST_ICONS:
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            "Plugin command icon is not a supported Host icon.",
+            details={"command_id": spec.command_id, "icon": spec.icon},
+        )
+    if spec.parameters.get("type", "object") != "object":
+        raise ExtensionError("PLUGIN_COMMAND_INVALID", "Command parameters must be an object schema.")
+    try:
+        Draft202012Validator.check_schema(spec.parameters)
+        reject_external_schema_references(spec.parameters)
+    except (SchemaReferenceError, SchemaError) as exc:
+        message = exc.message if isinstance(exc, SchemaError) else str(exc)
+        raise ExtensionError(
+            "PLUGIN_COMMAND_INVALID",
+            f"Plugin command parameters contain invalid JSON Schema: {message}",
+        ) from exc
+
+
+def _validate_setting_value(field: PluginSettingField, value: Any) -> None:
+    valid = False
+    if field.type == PluginSettingType.string:
+        valid = isinstance(value, str) and len(value.encode("utf-8")) <= 64 * 1024
+    elif field.type == PluginSettingType.number:
+        valid = (
+            (isinstance(value, int) and not isinstance(value, bool))
+            or (isinstance(value, float) and math.isfinite(value))
+        )
+    elif field.type == PluginSettingType.boolean:
+        valid = isinstance(value, bool)
+    elif field.type == PluginSettingType.select:
+        valid = isinstance(value, str) and value in field.options
+    if not valid:
+        raise ExtensionError(
+            "PLUGIN_SETTINGS_FIELD_INVALID",
+            f"Plugin setting has an invalid value: {field.key}",
+            details={"key": field.key},
+        )
+    if field.type == PluginSettingType.number:
+        if field.minimum is not None and value < field.minimum:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_FIELD_INVALID",
+                f"Plugin setting is below its minimum: {field.key}",
+                details={"key": field.key, "minimum": field.minimum},
+            )
+        if field.maximum is not None and value > field.maximum:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_FIELD_INVALID",
+                f"Plugin setting is above its maximum: {field.key}",
+                details={"key": field.key, "maximum": field.maximum},
+            )
+
+
+def _secret_field(
+    definition: PluginSettingsDefinition, plugin_id: str, key: str
+) -> PluginSettingField:
+    field = next((item for item in definition.fields if item.key == key), None)
+    if field is None or field.type != PluginSettingType.secret:
+        raise ExtensionError(
+            "PLUGIN_SECRET_FIELD_NOT_FOUND",
+            f"Plugin secret field does not exist: {key}",
+            status_code=404,
+            details={"plugin_id": plugin_id, "key": key},
+        )
+    return field
+
+
+def _secret_reference(plugin_id: str, key: str) -> str:
+    digest = hashlib.sha256(f"{plugin_id}\0{key}".encode("utf-8")).hexdigest()
+    return f"plugin.{digest}"
+
+
+def _settings_schema_error(plugin_id: str, message: str) -> ExtensionError:
+    return ExtensionError(
+        "PLUGIN_SETTINGS_SCHEMA_INVALID",
+        message,
+        details={"plugin_id": plugin_id},
+    )

backend/app/extensions/errors.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class ExtensionError(RuntimeError):
+    """Extension Core 对 API 暴露的稳定领域错误。"""
+
+    def __init__(
+        self,
+        code: str,
+        message: str,
+        *,
+        status_code: int = 422,
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.status_code = status_code
+        self.details = details or {}

backend/app/extensions/mcp.py

@@ -0,0 +1,791 @@
+"""本地 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,
+)
+from app.schema_security import (
+    SchemaReferenceError,
+    reject_external_schema_references,
+)
+
+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)
+            reject_external_schema_references(schema)
+        except (SchemaReferenceError, SchemaError) as exc:
+            message = exc.message if isinstance(exc, SchemaError) else str(exc)
+            raise McpBridgeError(
+                "MCP_TOOL_SCHEMA_INVALID",
+                f"Invalid MCP tool schema for {remote_name}: {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,21 +1,44 @@
 from __future__ import annotations
 
 import re
+import threading
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Literal
+from uuid import uuid4
 
 import yaml
 from jsonschema import Draft202012Validator
-from jsonschema.exceptions import SchemaError
+from jsonschema.exceptions import (
-from pydantic import BaseModel, ConfigDict, Field, ValidationError, create_model
+    SchemaError,
+    ValidationError as JsonSchemaValidationError,
+)
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    TypeAdapter,
+    ValidationError,
+    create_model,
+)
 
-from app.agent.tools import ToolExecutionContext, ToolRegistry
+from app.agent.tools import ToolExecutionContext, ToolExecutionError, ToolRegistry
 from app.agent.permissions import KNOWN_PERMISSIONS
 from app.contracts import (
     ModelCapability,
     Plugin,
+    PluginCommand,
+    PluginCommandContext,
+    PluginCommandEffect,
+    PluginNoEffect,
+    PluginNotificationEffect,
+    PluginCommandLocation,
+    PluginCommandResult,
     PluginManifest,
+    PluginHostStatus,
+    PluginSecretStatus,
+    PluginSettingType,
+    PluginSettingsSchema,
     PluginStatus,
     RetrievalConfig,
     Skill,
@@ -23,26 +46,26 @@ from app.contracts import (
     SkillStatus,
     ToolDefinition,
 )
+from app.extensions.contributions import (
+    CommandRegistry,
+    PluginCommandSpec,
+    PluginSecretResolver,
+    PluginSettingsDefinition,
+    PluginSettingsStore,
+    validate_command_spec,
+    validate_settings_definition,
+)
+from app.extensions.errors import ExtensionError
+from app.extensions.mcp import McpBridge, McpBridgeError, McpDiscoveredTool
+from app.providers.credentials import EncryptedCredentialStore
+from app.schema_security import (
+    SchemaReferenceError,
+    reject_external_schema_references,
+)
 
 _EXTENSION_ID = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
 
 
-class ExtensionError(RuntimeError):
-    def __init__(
-        self,
-        code: str,
-        message: str,
-        *,
-        status_code: int = 422,
-        details: dict[str, Any] | None = None,
-    ) -> None:
-        super().__init__(message)
-        self.code = code
-        self.message = message
-        self.status_code = status_code
-        self.details = details or {}
-
-
 @dataclass(frozen=True, slots=True)
 class AgentConfiguration:
     skill_id: str
@@ -235,25 +258,70 @@ class DeclarativePluginHost:
             return {"text": str(values.get("text", "")).upper()}
         raise ExtensionError("PLUGIN_HANDLER_UNSUPPORTED", f"Unsupported handler: {handler}")
 
+    async def execute_command(
+        self,
+        handler: str,
+        arguments: dict[str, Any],
+        context: dict[str, Any],
+        settings: dict[str, Any],
+        resolve_secret: PluginSecretResolver,
+    ) -> PluginCommandEffect:
+        """执行宿主内置的白名单 Command handler，不导入 Plugin Python 代码。"""
+
+        if handler == "echo":
+            message = str(arguments.get("message", context.get("selection", "")))
+            if not message:
+                return PluginNoEffect()
+            return PluginNotificationEffect(
+                payload={"level": "info", "message": message},
+            )
+        if handler == "uppercase_selection":
+            text = str(arguments.get("text", context.get("selection", "")))
+            limit = int(settings.get("result_limit", 100))
+            return PluginNotificationEffect(
+                payload={"level": "success", "message": text[:limit].upper()},
+            )
+        raise ExtensionError(
+            "PLUGIN_HANDLER_UNSUPPORTED", f"Unsupported command handler: {handler}"
+        )
+
 
 @dataclass(slots=True)
 class _PluginRecord:
     plugin: Plugin
     tools: list[DeclarativeToolSpec]
+    commands: list[PluginCommandSpec]
+    settings_definition: PluginSettingsDefinition | None
     package_path: Path
     registered_tools: list[str]
+    registered_commands: list[str]
+    mcp_remote_names: dict[str, str]
+    mcp_command_schemas: dict[str, dict[str, Any]]
 
 
 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,
+        credentials: EncryptedCredentialStore | None = None,
+        *,
+        allow_unsandboxed_mcp: bool = False,
+    ) -> None:
         self.registry = tools
         self.host = host or DeclarativePluginHost()
+        self.mcp = mcp_bridge or McpBridge()
+        self.commands = CommandRegistry()
+        self.settings = PluginSettingsStore(credentials or EncryptedCredentialStore())
+        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 +339,19 @@ class PluginRuntime:
                 status_code=409,
             )
 
-        specs = self._load_tools(root)
+        _validate_backend(manifest)
-        declared = set(manifest.contributes.tools)
+        specs = [] if manifest.backend.type == "mcp" else self._load_tools(root)
-        actual = {spec.name for spec in specs}
+        command_specs = self._load_commands(root)
-        if declared != actual:
+        settings_definition = self._load_settings(root)
-            raise ExtensionError(
+        if manifest.backend.type != "mcp":
-                "PLUGIN_CONTRIBUTION_INVALID",
+            declared = set(manifest.contributes.tools)
-                "plugin.yaml tool contributions must exactly match tools.yaml",
+            actual = {spec.name for spec in specs}
-                details={"declared": sorted(declared), "actual": sorted(actual)},
+            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)
@@ -289,6 +361,89 @@ class PluginRuntime:
                     f"Tool permission is not declared by Plugin: {spec.permission}",
                     details={"tool": spec.name, "permission": spec.permission},
                 )
+        declared_commands = set(manifest.contributes.commands)
+        actual_commands = {spec.command_id for spec in command_specs}
+        if (
+            declared_commands != actual_commands
+            or len(manifest.contributes.commands) != len(declared_commands)
+            or len(command_specs) != len(actual_commands)
+        ):
+            raise ExtensionError(
+                "PLUGIN_CONTRIBUTION_INVALID",
+                "plugin.yaml command contributions must exactly match commands.yaml",
+                details={
+                    "declared": sorted(declared_commands),
+                    "actual": sorted(actual_commands),
+                },
+            )
+        for spec in command_specs:
+            validate_command_spec(manifest.plugin_id, spec)
+            if spec.permission and spec.permission not in manifest.permissions:
+                raise ExtensionError(
+                    "PLUGIN_PERMISSION_UNDECLARED",
+                    f"Command permission is not declared by Plugin: {spec.permission}",
+                    details={"command": spec.command_id, "permission": spec.permission},
+                )
+        declared_sections = set(manifest.contributes.settings_sections)
+        actual_sections = (
+            {settings_definition.section_id} if settings_definition is not None else set()
+        )
+        if (
+            declared_sections != actual_sections
+            or len(manifest.contributes.settings_sections) != len(declared_sections)
+        ):
+            raise ExtensionError(
+                "PLUGIN_CONTRIBUTION_INVALID",
+                "plugin.yaml settings contributions must exactly match settings.yaml",
+                details={
+                    "declared": sorted(declared_sections),
+                    "actual": sorted(actual_sections),
+                },
+            )
+        if settings_definition is not None:
+            validate_settings_definition(manifest.plugin_id, settings_definition)
+        secret_fields = (
+            {
+                field.key
+                for field in settings_definition.fields
+                if field.type == PluginSettingType.secret
+            }
+            if settings_definition is not None
+            else set()
+        )
+        for spec in command_specs:
+            unknown_secrets = sorted(set(spec.secrets) - secret_fields)
+            if unknown_secrets:
+                raise ExtensionError(
+                    "PLUGIN_COMMAND_INVALID",
+                    "Plugin command references undeclared Secret settings.",
+                    details={
+                        "command_id": spec.command_id,
+                        "secrets": unknown_secrets,
+                    },
+                )
+            if spec.secrets and "secrets.use" not in manifest.permissions:
+                raise ExtensionError(
+                    "PLUGIN_PERMISSION_UNDECLARED",
+                    "Commands using Secret settings require the secrets.use permission.",
+                    details={"command_id": spec.command_id},
+                )
+            if spec.mcp_tool is not None:
+                _validate_id("MCP command target", spec.mcp_tool)
+                if manifest.backend.type != "mcp" or not spec.mcp_tool.startswith(
+                    f"{manifest.plugin_id}."
+                ):
+                    raise ExtensionError(
+                        "PLUGIN_COMMAND_INVALID",
+                        "MCP Command target must use the current Plugin namespace.",
+                        details={"command_id": spec.command_id},
+                    )
+                if spec.mcp_tool in manifest.contributes.tools:
+                    raise ExtensionError(
+                        "PLUGIN_COMMAND_INVALID",
+                        "MCP Command target cannot also be exposed as an Agent Tool.",
+                        details={"command_id": spec.command_id},
+                    )
 
         record = _PluginRecord(
             plugin=Plugin(
@@ -300,8 +455,13 @@ class PluginRuntime:
                 ),
             ),
             tools=specs,
+            commands=command_specs,
+            settings_definition=settings_definition,
             package_path=root,
             registered_tools=[],
+            registered_commands=[],
+            mcp_remote_names={},
+            mcp_command_schemas={},
         )
         self._records[manifest.plugin_id] = record
         return record.plugin.model_copy(deep=True)
@@ -313,18 +473,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 +492,20 @@ 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},
+            )
+        if record.settings_definition is not None:
+            self.settings.runtime_values(plugin_id, record.settings_definition)
+        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",
@@ -344,44 +513,229 @@ class PluginRuntime:
                 status_code=409,
                 details={"plugin_id": plugin_id, "tools": conflicts},
             )
+        command_conflicts = [
+            spec.command_id for spec in record.commands if self.commands.contains(spec.command_id)
+        ]
+        if command_conflicts:
+            raise ExtensionError(
+                "PLUGIN_COMMAND_CONFLICT",
+                "Plugin commands are already registered.",
+                status_code=409,
+                details={"plugin_id": plugin_id, "commands": command_conflicts},
+            )
         record.plugin.status = PluginStatus.starting
         try:
-            for spec in record.tools:
+            if record.plugin.manifest.backend.type == "mcp":
-                arguments_model = _arguments_model(spec)
+                discovered = self._start_mcp(record)
+                actual = {item.definition.name for item in discovered}
+                declared = set(declared_tools)
+                command_targets = {
+                    spec.mcp_tool for spec in record.commands if spec.mcp_tool is not None
+                }
+                expected = declared | command_targets
+                if actual != expected:
+                    raise ExtensionError(
+                        "PLUGIN_CONTRIBUTION_INVALID",
+                        "Discovered MCP tools must exactly match Tool and Command targets.",
+                        details={"declared": sorted(expected), "actual": sorted(actual)},
+                    )
+                for item in discovered:
+                    if item.definition.name in declared:
+                        self._register_mcp_tool(record, item)
+                    else:
+                        record.mcp_remote_names[item.definition.name] = item.remote_name
+                        record.mcp_command_schemas[item.definition.name] = (
+                            item.definition.parameters
+                        )
+                        for spec in (
+                            command
+                            for command in record.commands
+                            if command.mcp_tool == item.definition.name
+                        ):
+                            _validate_mcp_command_target_schema(
+                                item.definition.parameters,
+                                spec.command_id,
+                            )
+            else:
+                for spec in record.tools:
+                    arguments_model = _arguments_model(spec)
 
-                async def executor(
+                    async def executor(
-                    arguments: BaseModel,
+                        arguments: BaseModel,
-                    context: ToolExecutionContext,
+                        context: ToolExecutionContext,
-                    _handler: str = spec.handler,
+                        _handler: str = spec.handler,
-                ) -> Any:
+                    ) -> Any:
-                    return await self.host.execute(_handler, arguments, context)
+                        return await self.host.execute(_handler, arguments, context)
 
-                self.registry.register(
+                    self.registry.register(
-                    ToolDefinition(
+                        ToolDefinition(
-                        name=spec.name,
+                            name=spec.name,
-                        description=spec.description,
+                            description=spec.description,
-                        parameters=spec.parameters,
+                            parameters=spec.parameters,
-                        permission=spec.permission,
+                            permission=spec.permission,
-                        source="plugin",
+                            source="plugin",
-                    ),
+                        ),
-                    arguments_model,
+                        arguments_model,
-                    executor,
+                        executor,
-                )
+                    )
-                record.registered_tools.append(spec.name)
+                    record.registered_tools.append(spec.name)
+            for spec in record.commands:
+
+                async def command_executor(
+                    arguments: dict[str, Any],
+                    context: dict[str, Any],
+                    _spec: PluginCommandSpec = spec,
+                    _record: _PluginRecord = record,
+                ) -> PluginCommandEffect:
+                    if (
+                        not _record.plugin.enabled
+                        or _record.plugin.status != PluginStatus.ready
+                    ):
+                        raise ExtensionError(
+                            "PLUGIN_COMMAND_NOT_FOUND",
+                            "Plugin command is not available while its Plugin is inactive.",
+                            status_code=404,
+                            details={"command_id": _spec.command_id},
+                        )
+                    settings = (
+                        self.settings.runtime_values(
+                            _record.plugin.manifest.plugin_id,
+                            _record.settings_definition,
+                        )
+                        if _record.settings_definition is not None
+                        else {}
+                    )
+
+                    def resolve_secret(key: str) -> str | None:
+                        if key not in _spec.secrets:
+                            raise ExtensionError(
+                                "PLUGIN_SECRET_ACCESS_DENIED",
+                                "Command cannot access an undeclared Plugin Secret.",
+                                status_code=403,
+                                details={
+                                    "command_id": _spec.command_id,
+                                    "key": key,
+                                },
+                            )
+                        if "secrets.use" not in _record.plugin.granted_permissions:
+                            raise ExtensionError(
+                                "PLUGIN_SECRET_ACCESS_DENIED",
+                                "Plugin no longer has permission to access Secret settings.",
+                                status_code=403,
+                                details={"command_id": _spec.command_id, "key": key},
+                            )
+                        if _record.settings_definition is None:
+                            return None
+                        value = self.settings.resolve_secret(
+                            _record.plugin.manifest.plugin_id,
+                            _record.settings_definition,
+                            key,
+                        )
+                        field = next(
+                            item
+                            for item in _record.settings_definition.fields
+                            if item.key == key
+                        )
+                        if field.required and value is None:
+                            raise ExtensionError(
+                                "PLUGIN_SECRET_REQUIRED",
+                                "A required Plugin Secret has not been configured.",
+                                status_code=409,
+                                details={"command_id": _spec.command_id, "key": key},
+                            )
+                        return value
+
+                    if _spec.mcp_tool is not None:
+                        remote_name = _record.mcp_remote_names[_spec.mcp_tool]
+                        secret_values = {
+                            key: value
+                            for key in _spec.secrets
+                            if (value := resolve_secret(key)) is not None
+                        }
+                        envelope = _mcp_command_envelope(
+                            _spec,
+                            arguments=arguments,
+                            context=context,
+                            settings=settings,
+                            secrets=secret_values,
+                        )
+                        _validate_mcp_command_envelope(
+                            _record.mcp_command_schemas[_spec.mcp_tool],
+                            envelope,
+                            _spec.command_id,
+                        )
+                        try:
+                            effect = await self.mcp.call_tool(
+                                _record.plugin.manifest.plugin_id,
+                                remote_name,
+                                envelope,
+                                request_id=f"command:{uuid4().hex}",
+                            )
+                        except ToolExecutionError as exc:
+                            raise ExtensionError(
+                                exc.code,
+                                "MCP Command target execution failed.",
+                                status_code=502,
+                                details={"command_id": _spec.command_id},
+                            ) from exc
+                        try:
+                            return TypeAdapter(PluginCommandEffect).validate_python(effect)
+                        except ValidationError as exc:
+                            raise ExtensionError(
+                                "PLUGIN_COMMAND_RESULT_INVALID",
+                                "MCP Command target returned an invalid effect.",
+                                status_code=502,
+                                details={"command_id": _spec.command_id},
+                            ) from exc
+
+                    return await self.host.execute_command(
+                        _spec.handler,
+                        arguments,
+                        context,
+                        settings,
+                        resolve_secret,
+                    )
+
+                self.commands.register(plugin_id, spec, command_executor)
+                record.registered_commands.append(spec.command_id)
         except Exception as exc:
             # 注册过程必须具备回滚语义，防止半启用插件污染全局工具表。
             for name in record.registered_tools:
                 self.registry.unregister(name)
             record.registered_tools.clear()
+            for command_id in record.registered_commands:
+                self.commands.unregister(command_id)
+            record.registered_commands.clear()
+            record.mcp_remote_names.clear()
+            record.mcp_command_schemas.clear()
+            self.mcp.stop(plugin_id)
             record.plugin.status = PluginStatus.error
-            record.plugin.error_message = str(exc)
+            record.plugin.error_message = _safe_extension_message(exc)
-            raise
+            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 +757,174 @@ 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()
+        for command_id in record.registered_commands:
+            self.commands.unregister(command_id)
+        record.registered_commands.clear()
+        record.mcp_remote_names.clear()
+        record.mcp_command_schemas.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 list_commands(
+        self, location: PluginCommandLocation | None = None
+    ) -> list[PluginCommand]:
+        return self.commands.list(location)
+
+    async def execute_command(
+        self,
+        command_id: str,
+        arguments: dict[str, Any],
+        context: PluginCommandContext,
+    ) -> PluginCommandResult:
+        return await self.commands.execute(command_id, arguments, context)
+
+    def get_settings(self, plugin_id: str) -> PluginSettingsSchema:
+        record = self._record(plugin_id)
+        definition = self._settings_definition(record)
+        return self.settings.get(plugin_id, definition)
+
+    def update_settings(
+        self, plugin_id: str, schema_version: int, values: dict[str, Any]
+    ) -> PluginSettingsSchema:
+        record = self._record(plugin_id)
+        definition = self._settings_definition(record)
+        return self.settings.update(plugin_id, definition, schema_version, values)
+
+    def put_setting_secret(
+        self, plugin_id: str, key: str, secret: str
+    ) -> PluginSecretStatus:
+        record = self._record(plugin_id)
+        definition = self._settings_definition(record)
+        return self.settings.put_secret(plugin_id, definition, key, secret)
+
+    def delete_setting_secret(self, plugin_id: str, key: str) -> PluginSecretStatus:
+        record = self._record(plugin_id)
+        definition = self._settings_definition(record)
+        return self.settings.delete_secret(plugin_id, definition, key)
+
+    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()
+        for command_id in record.registered_commands:
+            self.commands.unregister(command_id)
+        record.registered_commands.clear()
+        record.mcp_remote_names.clear()
+        record.mcp_command_schemas.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()
+            for command_id in record.registered_commands:
+                self.commands.unregister(command_id)
+            record.registered_commands.clear()
+            record.mcp_remote_names.clear()
+            record.mcp_command_schemas.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 +933,14 @@ 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)
+        self.settings.remove_plugin(plugin_id)
         del self._records[plugin_id]
 
     def _record(self, plugin_id: str) -> _PluginRecord:
@@ -443,6 +962,52 @@ class PluginRuntime:
         except ValidationError as exc:
             raise _manifest_error("plugin tool", exc) from exc
 
+    @staticmethod
+    def _load_commands(root: Path) -> list[PluginCommandSpec]:
+        path = root / "commands.yaml"
+        if not path.exists():
+            return []
+        raw = _read_yaml(path)
+        items = raw.get("commands", [])
+        if not isinstance(items, list):
+            raise ExtensionError(
+                "EXTENSION_MANIFEST_INVALID",
+                "Invalid plugin command manifest: commands must be an array.",
+            )
+        try:
+            return [
+                PluginCommandSpec.model_validate(item)
+                for item in items
+            ]
+        except ValidationError as exc:
+            raise _manifest_error("plugin command", exc) from exc
+
+    @staticmethod
+    def _load_settings(root: Path) -> PluginSettingsDefinition | None:
+        path = root / "settings.yaml"
+        if not path.exists():
+            return None
+        raw = _read_yaml(path)
+        try:
+            return PluginSettingsDefinition.model_validate(raw)
+        except ValidationError as exc:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_SCHEMA_INVALID",
+                "Invalid Plugin settings schema.",
+                details={"errors": exc.errors(include_url=False)},
+            ) from exc
+
+    @staticmethod
+    def _settings_definition(record: _PluginRecord) -> PluginSettingsDefinition:
+        if record.settings_definition is None:
+            raise ExtensionError(
+                "PLUGIN_SETTINGS_NOT_FOUND",
+                "Plugin does not contribute a Settings section.",
+                status_code=404,
+                details={"plugin_id": record.plugin.manifest.plugin_id},
+            )
+        return record.settings_definition
+
 
 def _package_dir(package_path: str | Path) -> Path:
     root = Path(package_path).expanduser().resolve()
@@ -498,34 +1063,85 @@ 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 _mcp_command_envelope(
+    spec: PluginCommandSpec,
+    *,
+    arguments: dict[str, Any],
+    context: dict[str, Any],
+    settings: dict[str, Any],
+    secrets: dict[str, str],
+) -> dict[str, Any]:
+    return {
+        "_notesagent": {
+            "command_id": spec.command_id,
+            "arguments": arguments,
+            "context": context,
+            "settings": settings,
+            "secrets": secrets,
+        }
+    }
+
+
+def _validate_mcp_command_envelope(
+    schema: dict[str, Any],
+    envelope: dict[str, Any],
+    command_id: str,
+) -> None:
+    """执行前用目标 Tool Schema 校验包含真实业务数据的宿主信封。"""
+
+    try:
+        Draft202012Validator(schema).validate(envelope)
+    except JsonSchemaValidationError as exc:
+        raise ExtensionError(
+            "PLUGIN_COMMAND_TARGET_SCHEMA_MISMATCH",
+            "MCP Command envelope does not match the target inputSchema.",
+            status_code=502,
+            details={"command_id": command_id, "path": list(exc.path)},
+        ) from exc
+
+
+def _validate_mcp_command_target_schema(
+    schema: dict[str, Any], command_id: str
+) -> None:
+    """启用时只检查稳定信封入口，避免用伪造业务值误判合法 Schema。"""
+
+    properties = schema.get("properties")
+    envelope_schema = (
+        properties.get("_notesagent") if isinstance(properties, dict) else None
+    )
+    if not isinstance(envelope_schema, dict) or envelope_schema.get("type") != "object":
+        raise ExtensionError(
+            "PLUGIN_CONTRIBUTION_INVALID",
+            "MCP Command target inputSchema must directly declare "
+            "_notesagent with type object.",
+            details={"command_id": command_id},
+        )
+
+
+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", {})
+    model_name = "PluginArgs_" + re.sub(r"\W+", "_", tool_name)
-    required = set(schema.get("required", []))
+    # 完整 JSON Schema 已在 ToolRegistry 中先行校验。参数载体不重复声明字段，
-    fields: dict[str, tuple[Any, Any]] = {}
+    # 从而完整保留 model_dump、连字符键、联合类型和动态属性等合法 JSON 键值。
-    types = {
+    return create_model(model_name, __config__=ConfigDict(extra="allow"))
-        "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)
 
 
 def _validate_tool_schema(spec: DeclarativeToolSpec) -> None:
     schema = spec.parameters or {"type": "object", "properties": {}}
     try:
         Draft202012Validator.check_schema(schema)
-    except SchemaError as exc:
+        reject_external_schema_references(schema)
+    except (SchemaReferenceError, SchemaError) as exc:
+        message = exc.message if isinstance(exc, SchemaError) else str(exc)
         raise ExtensionError(
             "PLUGIN_TOOL_SCHEMA_INVALID",
-            f"Invalid JSON Schema for tool {spec.name}: {exc.message}",
+            f"Invalid JSON Schema for tool {spec.name}: {message}",
             details={"tool": spec.name},
         ) from exc
     if schema.get("type", "object") != "object" or not isinstance(
@@ -536,3 +1152,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/providers/credentials.py

@@ -13,6 +13,7 @@ from app.config import get_settings
 
 
 _CREDENTIAL_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+_PLUGIN_CREDENTIAL_PREFIX = "plugin."
 
 
 class CredentialStoreError(RuntimeError):
@@ -23,6 +24,15 @@ class CredentialResolver(Protocol):
     def resolve(self, credential_id: str | None) -> str | None: ...
 
 
+def validate_provider_credential_id(credential_id: str | None) -> None:
+    """阻止 Provider 和通用凭据 API 跨入 Plugin 私有命名空间。"""
+
+    if credential_id and credential_id.casefold().startswith(
+        _PLUGIN_CREDENTIAL_PREFIX
+    ):
+        raise CredentialStoreError("Credential namespace is reserved for Plugin settings.")
+
+
 class EnvironmentCredentialResolver:
     """解析由桌面 Host 注入 Sidecar 进程的临时凭证上下文。"""
 
@@ -109,17 +119,26 @@ class EncryptedCredentialStore:
 
     def _write_tokens(self, tokens: dict[str, str]) -> None:
         _, store_path = self._paths()
-        store_path.parent.mkdir(parents=True, exist_ok=True)
-        self._restrict(store_path.parent, 0o700)
         temporary = store_path.with_suffix(".tmp")
-        temporary.write_text(
+        try:
-            json.dumps(tokens, ensure_ascii=True, sort_keys=True),
+            store_path.parent.mkdir(parents=True, exist_ok=True)
-            encoding="utf-8",
+            self._restrict(store_path.parent, 0o700)
-        )
+            temporary.write_text(
-        self._restrict(temporary, 0o600)
+                json.dumps(tokens, ensure_ascii=True, sort_keys=True),
-        # 凭据表同样使用原子替换，确保并发读取只会看到完整 JSON。
+                encoding="utf-8",
-        temporary.replace(store_path)
+            )
-        self._restrict(store_path, 0o600)
+            self._restrict(temporary, 0o600)
+            # 凭据表同样使用原子替换，确保并发读取只会看到完整 JSON。
+            temporary.replace(store_path)
+            self._restrict(store_path, 0o600)
+        except OSError as exc:
+            try:
+                temporary.unlink(missing_ok=True)
+            except OSError:
+                pass
+            raise CredentialStoreError(
+                "Encrypted credential store cannot be written."
+            ) from exc
 
     def put(self, credential_id: str, secret: str) -> None:
         self._validate_id(credential_id)
@@ -158,6 +177,24 @@ class EncryptedCredentialStore:
                 self._write_tokens(tokens)
             return removed
 
+    def delete_many(self, credential_ids: list[str]) -> set[str]:
+        """用一次原子替换删除多个凭据，避免插件卸载只删除部分 Secret。"""
+
+        for credential_id in credential_ids:
+            self._validate_id(credential_id)
+        with self._lock:
+            tokens = self._read_tokens()
+            removed = {
+                credential_id
+                for credential_id in credential_ids
+                if credential_id in tokens
+            }
+            if removed:
+                for credential_id in removed:
+                    del tokens[credential_id]
+                self._write_tokens(tokens)
+            return removed
+
 
 class ChainedCredentialResolver:
     def __init__(self, *resolvers: CredentialResolver) -> None:
@@ -170,3 +207,14 @@ class ChainedCredentialResolver:
             if value:
                 return value
         return None
+
+
+class ProviderCredentialResolver:
+    """Provider 专用防御层，避免配置绕过 HTTP 校验读取 Plugin Secret。"""
+
+    def __init__(self, delegate: CredentialResolver) -> None:
+        self._delegate = delegate
+
+    def resolve(self, credential_id: str | None) -> str | None:
+        validate_provider_credential_id(credential_id)
+        return self._delegate.resolve(credential_id)

backend/app/providers/factory.py

@@ -1,6 +1,6 @@
 from app.contracts import ModelCapability, ProviderConfig, ProviderPreset, ProviderType
 from app.providers.base import ModelProvider
-from app.providers.credentials import CredentialResolver
+from app.providers.credentials import CredentialResolver, ProviderCredentialResolver
 from app.providers.ollama import OllamaProvider
 from app.providers.openai_compatible import OpenAICompatibleProvider
 
@@ -11,7 +11,9 @@ class UnsupportedProviderError(ValueError):
 
 class ProviderFactory:
     def __init__(self, credentials: CredentialResolver) -> None:
-        self.credentials = credentials
+        # ProviderFactory 是所有可配置 Provider 的创建边界，在此统一禁止
+        # Provider 借用 Plugin Secret 引用，避免调用方漏包安全 Resolver。
+        self.credentials = ProviderCredentialResolver(credentials)
 
     def build(self, config: ProviderConfig) -> ModelProvider:
         if config.provider_type in {

backend/app/routes.py

@@ -1,3 +1,4 @@
+import asyncio
 from collections.abc import AsyncIterator
 from datetime import datetime, timezone
 from uuid import uuid4
@@ -32,8 +33,17 @@ from app.contracts import (
     PageMeta,
     PermissionDecisionRequest,
     Plugin,
+    PluginCommandExecuteRequest,
+    PluginCommandListResponse,
+    PluginCommandLocation,
+    PluginCommandResult,
+    PluginHostStatus,
     PluginListResponse,
     PluginPermissionGrantRequest,
+    PluginSecretStatus,
+    PluginSecretWriteRequest,
+    PluginSettingsSchema,
+    PluginSettingsUpdateRequest,
     ProviderConfig,
     ProviderCreateRequest,
     ProviderListResponse,
@@ -65,7 +75,10 @@ from app.extensions import ExtensionError
 from app.providers.registry import ProviderNotFoundError
 from app.providers.factory import UnsupportedProviderError
 from app.providers.base import ProviderError
-from app.providers.credentials import CredentialStoreError
+from app.providers.credentials import (
+    CredentialStoreError,
+    validate_provider_credential_id,
+)
 from app.retrieval.engine import engine
 from app.services import (
     index_service,
@@ -82,6 +95,13 @@ def utc_now() -> datetime:
     return datetime.now(timezone.utc)
 
 
+def validate_public_credential_id(credential_id: str | None) -> None:
+    try:
+        validate_provider_credential_id(credential_id)
+    except CredentialStoreError as exc:
+        raise ApiError(422, "CREDENTIAL_NAMESPACE_RESERVED", str(exc)) from exc
+
+
 def as_sse(event: str, payload: str, *, event_id: int | None = None) -> str:
     id_line = f"id: {event_id}\n" if event_id is not None else ""
     return f"{id_line}event: {event}\ndata: {payload}\n\n"
@@ -130,6 +150,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 +512,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 +521,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 +532,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,10 +571,92 @@ 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")
 
 
+# Plugin Command / Settings Contributions
+@router.get(
+    "/plugin-contributions/commands",
+    response_model=PluginCommandListResponse,
+    tags=["Plugins"],
+)
+async def list_plugin_commands(
+    location: PluginCommandLocation | None = Query(default=None),
+) -> PluginCommandListResponse:
+    return PluginCommandListResponse(items=container.plugins.list_commands(location))
+
+
+@router.post(
+    "/plugin-contributions/commands/{command_id}/execute",
+    response_model=PluginCommandResult,
+    tags=["Plugins"],
+)
+async def execute_plugin_command(
+    command_id: str, request: PluginCommandExecuteRequest
+) -> PluginCommandResult:
+    try:
+        return await container.plugins.execute_command(
+            command_id, request.arguments, request.context
+        )
+    except ExtensionError as exc:
+        raise ApiError(exc.status_code, exc.code, exc.message, exc.details) from exc
+
+
+@router.get(
+    "/plugins/{plugin_id}/settings",
+    response_model=PluginSettingsSchema,
+    tags=["Plugins"],
+)
+async def get_plugin_settings(plugin_id: str) -> PluginSettingsSchema:
+    return extension_call(lambda: container.plugins.get_settings(plugin_id))
+
+
+@router.put(
+    "/plugins/{plugin_id}/settings",
+    response_model=PluginSettingsSchema,
+    tags=["Plugins"],
+)
+async def update_plugin_settings(
+    plugin_id: str, request: PluginSettingsUpdateRequest
+) -> PluginSettingsSchema:
+    return extension_call(
+        lambda: container.plugins.update_settings(
+            plugin_id, request.schema_version, request.values
+        )
+    )
+
+
+@router.put(
+    "/plugins/{plugin_id}/settings/{key}/secret",
+    response_model=PluginSecretStatus,
+    tags=["Plugins"],
+)
+async def put_plugin_setting_secret(
+    plugin_id: str, key: str, request: PluginSecretWriteRequest
+) -> PluginSecretStatus:
+    return extension_call(
+        lambda: container.plugins.put_setting_secret(
+            plugin_id, key, request.secret.get_secret_value()
+        )
+    )
+
+
+@router.delete(
+    "/plugins/{plugin_id}/settings/{key}/secret",
+    response_model=PluginSecretStatus,
+    tags=["Plugins"],
+)
+async def delete_plugin_setting_secret(
+    plugin_id: str, key: str
+) -> PluginSecretStatus:
+    return extension_call(
+        lambda: container.plugins.delete_setting_secret(plugin_id, key)
+    )
+
+
 # Providers
 @router.get(
     "/credentials/{credential_id}",
@@ -527,6 +664,7 @@ async def uninstall_plugin(plugin_id: str) -> OperationResponse:
     tags=["Providers"],
 )
 async def get_credential_status(credential_id: str) -> CredentialStatus:
+    validate_public_credential_id(credential_id)
     try:
         configured = container.credentials.has(credential_id)
     except CredentialStoreError as exc:
@@ -542,6 +680,7 @@ async def get_credential_status(credential_id: str) -> CredentialStatus:
 async def put_credential(
     credential_id: str, request: CredentialWriteRequest
 ) -> CredentialStatus:
+    validate_public_credential_id(credential_id)
     try:
         container.credentials.put(credential_id, request.api_key.get_secret_value())
     except CredentialStoreError as exc:
@@ -555,6 +694,7 @@ async def put_credential(
     tags=["Providers"],
 )
 async def delete_credential(credential_id: str) -> CredentialStatus:
+    validate_public_credential_id(credential_id)
     try:
         container.credentials.delete(credential_id)
     except CredentialStoreError as exc:
@@ -591,6 +731,7 @@ async def get_provider(provider_id: str) -> ProviderConfig:
     tags=["Providers"],
 )
 async def create_provider(request: ProviderCreateRequest) -> ProviderConfig:
+    validate_public_credential_id(request.credential_id)
     config = ProviderConfig(
         provider_id=f"provider_{uuid4().hex}",
         provider_type=request.provider_type,
@@ -634,6 +775,8 @@ async def update_provider(
             "name and enabled cannot be null when explicitly provided.",
         )
     updates = {name: getattr(request, name) for name in fields}
+    if "credential_id" in fields:
+        validate_public_credential_id(request.credential_id)
     config = ProviderConfig.model_validate(
         {**current.model_dump(mode="python"), **updates}
     )
@@ -693,6 +836,7 @@ async def list_provider_models(provider_id: str) -> ProviderModelsResponse:
 async def test_provider(request: ProviderTestRequest) -> ProviderTestResponse:
     registered = configurable_provider_or_404(request.provider_id)
     if request.credential_context_id:
+        validate_public_credential_id(request.credential_context_id)
         temporary_config = registered.config.model_copy(
             update={"credential_id": request.credential_context_id, "enabled": True}
         )

backend/app/schema_security.py

@@ -0,0 +1,60 @@
+"""共享 JSON Schema 安全约束。"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urljoin
+
+from referencing import Registry
+from referencing.exceptions import Unresolvable
+from referencing.jsonschema import DRAFT202012
+
+_SCHEMA_BASE_URI = "https://notesagent.invalid/local-schema"
+
+
+class SchemaReferenceError(ValueError):
+    """Schema 引用不符合宿主的离线、文档内解析约束。"""
+
+
+class ExternalSchemaReferenceError(SchemaReferenceError):
+    def __init__(self, keyword: str, reference: Any) -> None:
+        super().__init__(f"External JSON Schema reference is not allowed: {reference!r}")
+        self.keyword = keyword
+        self.reference = reference
+
+
+class UnresolvableLocalSchemaReferenceError(SchemaReferenceError):
+    def __init__(self, reference: str) -> None:
+        super().__init__(f"Local JSON Schema reference cannot be resolved: {reference!r}")
+        self.reference = reference
+
+
+def reject_external_schema_references(schema: Any) -> None:
+    """只允许可解析的文档内 Fragment，并按 JSON Schema Resource 作用域解析。"""
+
+    root = DRAFT202012.create_resource(schema)
+    root_uri = urljoin(_SCHEMA_BASE_URI, root.id() or "")
+    registry = Registry().with_resource(_SCHEMA_BASE_URI, root).crawl()
+    resolver = registry.resolver(root_uri)
+    _validate_resource_references(root, resolver)
+
+
+def _validate_resource_references(resource, resolver: Any) -> None:
+    contents = resource.contents
+    if isinstance(contents, dict):
+        for keyword in ("$ref", "$dynamicRef"):
+            if keyword not in contents:
+                continue
+            reference = contents[keyword]
+            if not isinstance(reference, str) or not reference.startswith("#"):
+                raise ExternalSchemaReferenceError(keyword, reference)
+            try:
+                resolver.lookup(reference)
+            except Unresolvable as exc:
+                raise UnresolvableLocalSchemaReferenceError(reference) from exc
+
+    for subresource in resource.subresources():
+        _validate_resource_references(
+            subresource,
+            resolver.in_subresource(subresource),
+        )

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

@@ -0,0 +1,20 @@
+commands:
+  - command_id: mcp-fixture.notify
+    title: MCP 通知
+    description: 通过隔离 MCP Host 返回宿主白名单通知 effect。
+    icon: bolt
+    locations:
+      - command_palette
+    when:
+      - editor.has_selection
+    context:
+      - selection
+    secrets:
+      - api_key
+    mcp_tool: mcp-fixture.command
+    parameters:
+      type: object
+      properties:
+        message:
+          type: string
+      additionalProperties: false

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

@@ -0,0 +1,26 @@
+id: mcp-fixture
+name: MCP Fixture
+version: 1.0.0
+description: 阶段 C/D 离线联调 Fixture，覆盖 MCP Tool、Command 与错误边界。
+permissions:
+  - notes.read
+  - secrets.use
+contributes:
+  tools:
+    - mcp-fixture.echo
+    - mcp-fixture.fail
+    - mcp-fixture.sleep
+    - mcp-fixture.large
+    - mcp-fixture.environment
+    - mcp-fixture.exit
+  commands:
+    - mcp-fixture.notify
+  settings_sections:
+    - mcp-fixture.general
+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,248 @@
+"""确定性的 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."),
+    "command": tool(
+        "command",
+        "Execute a NotesAgent Plugin Command envelope.",
+        {"_notesagent": {"type": "object"}},
+    ),
+}
+# 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 == "command":
+        envelope = arguments.get("_notesagent") or {}
+        command_arguments = envelope.get("arguments") or {}
+        context = envelope.get("context") or {}
+        settings = envelope.get("settings") or {}
+        secrets = envelope.get("secrets") or {}
+        if not isinstance(secrets.get("api_key"), str):
+            respond(
+                request_id,
+                {
+                    "content": [{"type": "text", "text": "declared secret missing"}],
+                    "isError": True,
+                },
+            )
+            return
+        message = command_arguments.get("message") or context.get("selection") or ""
+        message = f"{settings.get('message_prefix', '')}{message}"
+        respond(
+            request_id,
+            {
+                "content": [{"type": "text", "text": "command completed"}],
+                "structuredContent": {
+                    "type": "notification",
+                    "payload": {
+                        "level": "success",
+                        "message": str(message),
+                    },
+                },
+                "isError": False,
+            },
+        )
+        return
+    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"],
+                            TOOLS["command"],
+                        ]
+                    },
+                )
+            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/extensions/fixtures/mcp-echo/settings.yaml

@@ -0,0 +1,11 @@
+section_id: mcp-fixture.general
+schema_version: 1
+fields:
+  - key: message_prefix
+    label: Message Prefix
+    type: string
+    default: ""
+  - key: api_key
+    label: Fixture API Key
+    type: secret
+    required: true

backend/extensions/plugins/text-tools/commands.yaml

@@ -0,0 +1,19 @@
+commands:
+  - command_id: text-tools.uppercase-selection
+    title: 转为大写
+    description: 将当前选区或传入文本转换为大写并显示通知。
+    icon: edit
+    locations:
+      - command_palette
+      - context_menu
+    when:
+      - editor.has_selection
+    context:
+      - selection
+    handler: uppercase_selection
+    parameters:
+      type: object
+      properties:
+        text:
+          type: string
+      additionalProperties: false

backend/extensions/plugins/text-tools/plugin.yaml

@@ -6,6 +6,10 @@ permissions: []
 contributes:
   tools:
     - text.uppercase
+  commands:
+    - text-tools.uppercase-selection
+  settings_sections:
+    - text-tools.general
 backend:
   type: internal_rpc
   transport: none

backend/extensions/plugins/text-tools/settings.yaml

@@ -0,0 +1,31 @@
+section_id: text-tools.general
+schema_version: 1
+fields:
+  - key: result_limit
+    label: 结果字符数
+    description: Command 通知中最多保留的字符数。
+    type: number
+    required: true
+    default: 100
+    minimum: 1
+    maximum: 1000
+  - key: label_prefix
+    label: 标签前缀
+    type: string
+    default: ""
+  - key: output_style
+    label: 输出样式
+    type: select
+    default: notification
+    options:
+      - notification
+      - compact
+  - key: enabled_hint
+    label: 显示提示
+    type: boolean
+    default: true
+  - key: api_key
+    label: API Key
+    description: Secret 示例字段；普通 Settings API 永不返回明文。
+    type: secret
+    required: false

backend/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "httpx>=0.28,<1.0",
     "jsonschema>=4.25,<5.0",
     "pyyaml>=6.0,<7.0",
+    "referencing>=0.36,<1.0",
     "sqlite-vec>=0.1.9",
     "uvicorn[standard]>=0.35,<1.0",
 ]

backend/tests/test_api.py

@@ -93,6 +93,12 @@ 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/plugin-contributions/commands",
+        "/api/plugin-contributions/commands/{command_id}/execute",
+        "/api/plugins/{plugin_id}/settings",
+        "/api/plugins/{plugin_id}/settings/{key}/secret",
         "/api/plugins/{plugin_id}/enable",
         "/api/plugins/{plugin_id}/disable",
         "/api/providers/test",

backend/tests/test_credentials.py

@@ -1,16 +1,21 @@
 import asyncio
+from pathlib import Path
 
 import httpx
+import pytest
 
 from app.config import get_settings
 from app.contracts import CredentialWriteRequest
+from app.errors import ApiError
 from app.providers.credentials import (
     ChainedCredentialResolver,
+    CredentialStoreError,
     EncryptedCredentialStore,
     EnvironmentCredentialResolver,
 )
+from app.providers.factory import ProviderFactory
 from app.providers.openai_compatible import OpenAICompatibleProvider
-from app.routes import get_credential_status, put_credential
+from app.routes import delete_credential, get_credential_status, put_credential
 
 
 def test_encrypted_credential_store_round_trip_without_plaintext_on_disk() -> None:
@@ -31,6 +36,33 @@ def test_encrypted_credential_store_round_trip_without_plaintext_on_disk() -> No
     assert store.resolve("deepseek") is None
 
 
+def test_encrypted_credential_store_deletes_multiple_credentials_atomically() -> None:
+    store = EncryptedCredentialStore()
+    store.put("plugin.first", "first")
+    store.put("plugin.second", "second")
+    store.put("openai", "keep")
+
+    removed = store.delete_many(["plugin.first", "plugin.second"])
+
+    assert removed == {"plugin.first", "plugin.second"}
+    assert store.resolve("plugin.first") is None
+    assert store.resolve("plugin.second") is None
+    assert store.resolve("openai") == "keep"
+
+
+def test_credential_write_os_error_uses_stable_store_error(monkeypatch) -> None:
+    store = EncryptedCredentialStore()
+    store.put("existing", "value")
+
+    def fail_replace(_path: Path, _target: Path) -> Path:
+        raise OSError("injected replace failure")
+
+    monkeypatch.setattr(Path, "replace", fail_replace)
+
+    with pytest.raises(CredentialStoreError, match="cannot be written"):
+        store.put("new", "value")
+
+
 def test_credential_api_never_returns_secret() -> None:
     written = asyncio.run(
         put_credential(
@@ -72,3 +104,29 @@ def test_saved_credential_takes_precedence_over_environment_fallback(monkeypatch
     resolver = ChainedCredentialResolver(store, EnvironmentCredentialResolver())
 
     assert resolver.resolve("deepseek") == "saved-key"
+
+
+def test_public_credential_api_rejects_plugin_namespace() -> None:
+    operations = [
+        get_credential_status("plugin.text-tools.api_key"),
+        put_credential(
+            "plugin.text-tools.api_key",
+            CredentialWriteRequest(api_key="must-not-write"),
+        ),
+        delete_credential("plugin.text-tools.api_key"),
+    ]
+    for operation in operations:
+        with pytest.raises(ApiError) as exc:
+            asyncio.run(operation)
+        assert exc.value.code == "CREDENTIAL_NAMESPACE_RESERVED"
+
+    assert EncryptedCredentialStore().resolve("plugin.text-tools.api_key") is None
+
+
+def test_provider_resolver_cannot_read_plugin_secret() -> None:
+    store = EncryptedCredentialStore()
+    store.put("plugin.text-tools.api_key", "private-plugin-secret")
+    resolver = ProviderFactory(store).credentials
+
+    with pytest.raises(CredentialStoreError, match="reserved for Plugin settings"):
+        resolver.resolve("plugin.text-tools.api_key")

backend/tests/test_extension_core.py

@@ -1,4 +1,7 @@
 import asyncio
+import shutil
+import threading
+import time
 
 import pytest
 
@@ -8,18 +11,39 @@ from app.container import build_container
 from app.contracts import (
     AgentRunCreateRequest,
     AgentRunStatus,
+    PluginCommandContext,
     SkillStatus,
     ToolCall,
 )
 from app.extensions import ExtensionError
+from app.extensions.mcp import McpStdioClient
+from app.extensions.runtime import (
+    _arguments_model_from_schema,
+    _validate_mcp_command_target_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", "secrets.use"])
+    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 +321,456 @@ 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 == 7
+        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_mcp_command_target_receives_scoped_context_and_declared_secret(
+    mcp_container,
+) -> None:
+    async def scenario() -> None:
+        mcp_container.plugins.enable("mcp-fixture")
+
+        assert not mcp_container.tools.contains("mcp-fixture.command")
+        with pytest.raises(ExtensionError) as missing:
+            await mcp_container.plugins.execute_command(
+                "mcp-fixture.notify",
+                {},
+                PluginCommandContext(selection="来自选区"),
+            )
+        assert missing.value.code == "PLUGIN_SECRET_REQUIRED"
+
+        mcp_container.plugins.put_setting_secret(
+            "mcp-fixture", "api_key", "mcp-command-secret"
+        )
+        mcp_container.plugins.update_settings(
+            "mcp-fixture", 1, {"message_prefix": "Fixture: "}
+        )
+        result = await mcp_container.plugins.execute_command(
+            "mcp-fixture.notify",
+            {},
+            PluginCommandContext(
+                note_id="must-not-enter-envelope",
+                selection="来自选区",
+            ),
+        )
+
+        assert result.effect.type == "notification"
+        assert result.effect.payload.model_dump() == {
+            "level": "success",
+            "message": "Fixture: 来自选区",
+        }
+        assert "mcp-command-secret" not in repr(
+            mcp_container.plugins.commands.audit_events()
+        )
+
+    run(scenario())
+
+
+def test_mcp_command_target_rejects_incompatible_envelope_schema(tmp_path) -> None:
+    package = tmp_path / "mcp-bad-command"
+    shutil.copytree(MCP_FIXTURE, package)
+    for filename in ("plugin.yaml", "commands.yaml", "settings.yaml"):
+        path = package / filename
+        path.write_text(
+            path.read_text(encoding="utf-8").replace(
+                "mcp-fixture", "mcp-bad-command"
+            ),
+            encoding="utf-8",
+        )
+    server_path = package / "server.py"
+    server_path.write_text(
+        server_path.read_text(encoding="utf-8").replace(
+            '{"_notesagent": {"type": "object"}}',
+            '{"unexpected": {"type": "string"}}',
+        ),
+        encoding="utf-8",
+    )
+    container = build_container()
+    container.plugins.install(package)
+    container.plugins.set_permissions(
+        "mcp-bad-command", ["notes.read", "secrets.use"]
+    )
+    try:
+        with pytest.raises(ExtensionError) as exc:
+            container.plugins.enable("mcp-bad-command")
+        assert exc.value.code == "PLUGIN_CONTRIBUTION_INVALID"
+        assert container.plugins.get("mcp-bad-command").status == "error"
+    finally:
+        container.plugins.shutdown()
+
+
+def test_mcp_command_target_enable_check_only_requires_protocol_marker() -> None:
+    # `not`/`oneOf` 等完整语义由实际调用前的官方 Validator 处理；启用检查
+    # 只确认不可被引用或组合隐藏的稳定宿主入口，避免维护不完整的求解器。
+    _validate_mcp_command_target_schema(
+        {
+            "type": "object",
+            "properties": {
+                "_notesagent": {
+                    "type": "object",
+                    "not": {"type": "object"},
+                }
+            },
+        },
+        "marker.run",
+    )
+
+    invalid_markers = [
+        {
+            "$defs": {"envelope": {"type": "object"}},
+            "properties": {"_notesagent": {"$ref": "#/$defs/envelope"}},
+        },
+        {
+            "allOf": [
+                {"properties": {"_notesagent": {"type": "object"}}},
+            ]
+        },
+    ]
+    for schema in invalid_markers:
+        with pytest.raises(ExtensionError) as exc:
+            _validate_mcp_command_target_schema(schema, "marker.run")
+        assert exc.value.code == "PLUGIN_CONTRIBUTION_INVALID"
+
+
+def test_mcp_command_validates_actual_envelope_before_call(tmp_path) -> None:
+    package = tmp_path / "mcp-runtime-schema"
+    shutil.copytree(MCP_FIXTURE, package)
+    for filename in ("plugin.yaml", "commands.yaml", "settings.yaml"):
+        path = package / filename
+        path.write_text(
+            path.read_text(encoding="utf-8").replace(
+                "mcp-fixture", "mcp-runtime-schema"
+            ),
+            encoding="utf-8",
+        )
+    server_path = package / "server.py"
+    server_path.write_text(
+        server_path.read_text(encoding="utf-8").replace(
+            '{"_notesagent": {"type": "object"}}',
+            '{"_notesagent": {"type": "object", "properties": '
+            '{"arguments": {"type": "object", "maxProperties": 0}, '
+            '"context": {"type": "object", "properties": '
+            '{"selection": {"type": "string"}}, "required": ["selection"]}}, '
+            '"required": ["arguments", "context"]}}',
+        ),
+        encoding="utf-8",
+    )
+    container = build_container()
+    container.plugins.install(package)
+    container.plugins.set_permissions(
+        "mcp-runtime-schema", ["notes.read", "secrets.use"]
+    )
+    try:
+        # context.selection 是 Command 的 when/context 契约保证的真实字段；
+        # 启用期结构检查不得因没有伪造该业务值而拒绝目标 Schema。
+        container.plugins.enable("mcp-runtime-schema")
+        container.plugins.put_setting_secret(
+            "mcp-runtime-schema", "api_key", "configured"
+        )
+        with pytest.raises(ExtensionError) as exc:
+            run(
+                container.plugins.execute_command(
+                    "mcp-runtime-schema.notify",
+                    {"message": "must be rejected locally"},
+                    PluginCommandContext(selection="visible"),
+                )
+            )
+        assert exc.value.code == "PLUGIN_COMMAND_TARGET_SCHEMA_MISMATCH"
+    finally:
+        container.plugins.shutdown()
+
+
+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", "secrets.use"])
+    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 == 7
+        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_plugin_contributions.py

@@ -0,0 +1,653 @@
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+from pydantic import TypeAdapter, ValidationError
+
+from app.agent import ToolRegistry
+from app.config import BACKEND_DIR, get_settings
+from app.container import build_container
+from app.contracts import (
+    PluginCommandContext,
+    PluginCommandEffect,
+    PluginNoEffect,
+    PluginSettingType,
+)
+from app.extensions import ExtensionError, PluginRuntime
+from app.extensions.contributions import _secret_reference
+from app.extensions.runtime import DeclarativePluginHost
+from app.providers.credentials import CredentialStoreError
+
+TEXT_TOOLS = BACKEND_DIR / "extensions" / "plugins" / "text-tools"
+
+
+def run(coroutine):
+    return asyncio.run(coroutine)
+
+
+def test_command_list_filter_and_lifecycle() -> None:
+    container = build_container()
+
+    commands = container.plugins.list_commands()
+    palette = container.plugins.list_commands(location="command_palette")
+
+    assert [item.command_id for item in commands] == ["text-tools.uppercase-selection"]
+    assert palette[0].plugin_id == "text-tools"
+    assert palette[0].icon == "edit"
+    assert palette[0].when == ["editor.has_selection"]
+
+    container.plugins.disable("text-tools")
+    assert container.plugins.list_commands() == []
+    with pytest.raises(ExtensionError) as exc:
+        run(
+            container.plugins.execute_command(
+                "text-tools.uppercase-selection",
+                {},
+                PluginCommandContext(selection="hello"),
+            )
+        )
+    assert exc.value.code == "PLUGIN_COMMAND_NOT_FOUND"
+
+    container.plugins.enable("text-tools")
+    assert len(container.plugins.list_commands()) == 1
+
+
+def test_command_executes_with_scoped_context_and_settings() -> None:
+    container = build_container()
+    container.plugins.update_settings("text-tools", 1, {"result_limit": 4})
+
+    result = run(
+        container.plugins.execute_command(
+            "text-tools.uppercase-selection",
+            {},
+            PluginCommandContext(
+                vault_id="default",
+                note_id="note_private",
+                file_path="private.md",
+                selection="abcdef",
+            ),
+        )
+    )
+
+    assert result.status == "completed"
+    assert result.effect.type == "notification"
+    assert result.effect.payload.model_dump() == {
+        "level": "success",
+        "message": "ABCD",
+    }
+
+
+def test_echo_command_returns_none_for_empty_message() -> None:
+    host = DeclarativePluginHost()
+
+    empty = run(host.execute_command("echo", {}, {}, {}, lambda _: None))
+    populated = run(
+        host.execute_command("echo", {"message": "hello"}, {}, {}, lambda _: None)
+    )
+
+    assert isinstance(empty, PluginNoEffect)
+    assert populated.type == "notification"
+    assert populated.payload.message == "hello"
+
+
+@pytest.mark.parametrize(
+    ("effect_type", "payload"),
+    [
+        ("none", {"unexpected": True}),
+        ("notification", {"level": "debug", "message": "invalid"}),
+        ("navigate", {"route": "https://example.com"}),
+        ("refresh", {"scope": "everything"}),
+        ("job", {"job_id": "invalid job id"}),
+    ],
+)
+def test_command_effect_rejects_untrusted_payloads(effect_type, payload) -> None:
+    with pytest.raises(ValidationError):
+        TypeAdapter(PluginCommandEffect).validate_python(
+            {"type": effect_type, "payload": payload}
+        )
+
+
+def test_command_rejects_missing_context_and_invalid_arguments() -> None:
+    container = build_container()
+
+    with pytest.raises(ExtensionError) as context_error:
+        run(
+            container.plugins.execute_command(
+                "text-tools.uppercase-selection", {}, PluginCommandContext()
+            )
+        )
+    assert context_error.value.code == "PLUGIN_COMMAND_CONTEXT_INVALID"
+
+    with pytest.raises(ExtensionError) as argument_error:
+        run(
+            container.plugins.execute_command(
+                "text-tools.uppercase-selection",
+                {"unknown": True},
+                PluginCommandContext(selection="hello"),
+            )
+        )
+    assert argument_error.value.code == "PLUGIN_COMMAND_ARGUMENT_INVALID"
+
+    audit = container.plugins.commands.audit_events()
+    assert [event.error_code for event in audit[-2:]] == [
+        "PLUGIN_COMMAND_CONTEXT_INVALID",
+        "PLUGIN_COMMAND_ARGUMENT_INVALID",
+    ]
+    # 审计事件不得携带参数、正文选区或返回 effect。
+    assert "hello" not in repr(audit)
+
+
+def test_command_only_receives_declared_context() -> None:
+    class CapturingHost(DeclarativePluginHost):
+        def __init__(self) -> None:
+            self.context = None
+
+        async def execute_command(
+            self, handler, arguments, context, settings, resolve_secret
+        ):
+            self.context = context
+            return PluginNoEffect()
+
+    host = CapturingHost()
+    runtime = PluginRuntime(ToolRegistry(), host=host)
+    runtime.install(TEXT_TOOLS)
+    runtime.enable("text-tools")
+
+    run(
+        runtime.execute_command(
+            "text-tools.uppercase-selection",
+            {},
+            PluginCommandContext(
+                vault_id="default", note_id="note_private", selection="visible"
+            ),
+        )
+    )
+
+    assert host.context == {"selection": "visible"}
+
+
+def test_command_resolves_only_declared_plugin_secrets(tmp_path: Path) -> None:
+    class SecretHost(DeclarativePluginHost):
+        def __init__(self) -> None:
+            self.secret = None
+            self.denied_code = None
+
+        async def execute_command(
+            self, handler, arguments, context, settings, resolve_secret
+        ):
+            self.secret = resolve_secret("api_key")
+            try:
+                resolve_secret("undeclared")
+            except ExtensionError as exc:
+                self.denied_code = exc.code
+            return PluginNoEffect()
+
+    host = SecretHost()
+    runtime = PluginRuntime(ToolRegistry(), host=host)
+    package = tmp_path / "secret-command"
+    package.mkdir()
+    (package / "plugin.yaml").write_text(
+        """
+id: secret-command
+name: Secret Command
+version: 1.0.0
+permissions: [secrets.use]
+contributes:
+  commands: [secret-command.run]
+  settings_sections: [secret-command.general]
+backend:
+  type: internal_rpc
+  transport: none
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "commands.yaml").write_text(
+        """
+commands:
+  - command_id: secret-command.run
+    title: Secret Command
+    locations: [command_palette]
+    secrets: [api_key]
+    handler: echo
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "settings.yaml").write_text(
+        """
+section_id: secret-command.general
+schema_version: 1
+fields:
+  - key: api_key
+    label: API Key
+    type: secret
+""".strip(),
+        encoding="utf-8",
+    )
+    runtime.install(package)
+    runtime.set_permissions("secret-command", ["secrets.use"])
+    runtime.enable("secret-command")
+    runtime.put_setting_secret("secret-command", "api_key", "runtime-only-secret")
+
+    run(
+        runtime.execute_command(
+            "secret-command.run",
+            {},
+            PluginCommandContext(selection="visible"),
+        )
+    )
+
+    assert host.secret == "runtime-only-secret"
+    assert host.denied_code == "PLUGIN_SECRET_ACCESS_DENIED"
+    assert "runtime-only-secret" not in repr(runtime.commands.audit_events())
+
+
+def test_settings_schema_contains_defaults_and_hides_secret() -> None:
+    container = build_container()
+
+    schema = container.plugins.get_settings("text-tools")
+    by_key = {field.key: field for field in schema.fields}
+
+    assert schema.schema_version == 1
+    assert schema.values == {
+        "result_limit": 100,
+        "label_prefix": "",
+        "output_style": "notification",
+        "enabled_hint": True,
+    }
+    assert "api_key" not in schema.values
+    assert schema.secrets["api_key"].configured is False
+    assert by_key["api_key"].type == PluginSettingType.secret
+
+
+def test_settings_update_validates_version_type_bounds_and_secret_boundary() -> None:
+    container = build_container()
+
+    updated = container.plugins.update_settings(
+        "text-tools", 1, {"result_limit": 20, "output_style": "compact"}
+    )
+    assert updated.values["result_limit"] == 20
+    assert updated.values["output_style"] == "compact"
+
+    cases = [
+        (2, {}, "PLUGIN_SETTINGS_VERSION_CONFLICT"),
+        (1, {"result_limit": 0}, "PLUGIN_SETTINGS_FIELD_INVALID"),
+        (1, {"enabled_hint": "yes"}, "PLUGIN_SETTINGS_FIELD_INVALID"),
+        (1, {"output_style": "unknown"}, "PLUGIN_SETTINGS_FIELD_INVALID"),
+        (1, {"api_key": "plaintext"}, "PLUGIN_SETTINGS_FIELD_INVALID"),
+        (1, {"unknown": True}, "PLUGIN_SETTINGS_FIELD_INVALID"),
+    ]
+    for version, values, code in cases:
+        with pytest.raises(ExtensionError) as exc:
+            container.plugins.update_settings("text-tools", version, values)
+        assert exc.value.code == code
+
+
+def test_required_plain_setting_blocks_enable_until_configured(tmp_path: Path) -> None:
+    package = tmp_path / "required-setting"
+    package.mkdir()
+    (package / "plugin.yaml").write_text(
+        """
+id: required-setting
+name: Required Setting
+version: 1.0.0
+contributes:
+  commands: [required-setting.run]
+  settings_sections: [required-setting.general]
+backend:
+  type: internal_rpc
+  transport: none
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "commands.yaml").write_text(
+        """
+commands:
+  - command_id: required-setting.run
+    title: Required Setting
+    locations: [command_palette]
+    handler: echo
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "settings.yaml").write_text(
+        """
+section_id: required-setting.general
+schema_version: 1
+fields:
+  - key: endpoint
+    label: Endpoint
+    type: string
+    required: true
+""".strip(),
+        encoding="utf-8",
+    )
+    runtime = PluginRuntime(ToolRegistry())
+    runtime.install(package)
+
+    with pytest.raises(ExtensionError) as exc:
+        runtime.enable("required-setting")
+    assert exc.value.code == "PLUGIN_SETTINGS_REQUIRED"
+    assert runtime.get("required-setting").status == "installed"
+
+    runtime.update_settings("required-setting", 1, {"endpoint": "local"})
+    assert runtime.enable("required-setting").status == "ready"
+
+
+def test_secret_roundtrip_never_enters_plain_settings_storage() -> None:
+    container = build_container()
+    plaintext = "stage-d-secret-value"
+
+    status = container.plugins.put_setting_secret("text-tools", "api_key", plaintext)
+    schema = container.plugins.get_settings("text-tools")
+    settings_path = get_settings().data_dir / "plugins" / "settings.json"
+    credentials_path = get_settings().data_dir / "credentials" / "credentials.json"
+
+    assert status.configured is True
+    assert schema.secrets["api_key"].configured is True
+    assert "api_key" not in schema.values
+    assert plaintext not in settings_path.read_text(encoding="utf-8")
+    assert plaintext not in credentials_path.read_text(encoding="utf-8")
+    stored_settings = json.loads(settings_path.read_text(encoding="utf-8"))
+    reference = stored_settings["text-tools"]["secret_refs"]["api_key"]
+    assert reference.startswith("plugin.")
+    assert len(reference) == 71
+    assert "text-tools" not in reference and "api_key" not in reference
+    assert container.credentials.resolve(reference) == plaintext
+
+    deleted = container.plugins.delete_setting_secret("text-tools", "api_key")
+    assert deleted.configured is False
+    assert container.credentials.resolve(reference) is None
+
+
+def test_uninstall_removes_plugin_settings_and_secret_namespace() -> None:
+    container = build_container()
+    container.plugins.update_settings("text-tools", 1, {"result_limit": 12})
+    container.plugins.put_setting_secret("text-tools", "api_key", "temporary")
+    settings_path = get_settings().data_dir / "plugins" / "settings.json"
+    reference = json.loads(settings_path.read_text(encoding="utf-8"))[
+        "text-tools"
+    ]["secret_refs"]["api_key"]
+
+    container.plugins.uninstall("text-tools")
+
+    stored = json.loads(settings_path.read_text(encoding="utf-8"))
+    assert "text-tools" not in stored
+    assert container.credentials.resolve(reference) is None
+
+
+def test_plugin_secret_reference_has_fixed_credential_safe_length() -> None:
+    reference = _secret_reference("p" * 512, "k" * 128)
+
+    assert reference.startswith("plugin.")
+    assert len(reference) <= 128
+
+
+def test_tampered_secret_reference_cannot_cross_credential_namespace() -> None:
+    container = build_container()
+    container.credentials.put("openai", "provider-private-secret")
+    settings_path = get_settings().data_dir / "plugins" / "settings.json"
+    settings_path.parent.mkdir(parents=True, exist_ok=True)
+    settings_path.write_text(
+        json.dumps(
+            {
+                "text-tools": {
+                    "schema_version": 1,
+                    "values": {},
+                    "secret_refs": {"api_key": "openai"},
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ExtensionError) as read_error:
+        container.plugins.get_settings("text-tools")
+    with pytest.raises(ExtensionError) as uninstall_error:
+        container.plugins.uninstall("text-tools")
+
+    assert read_error.value.code == "PLUGIN_STORAGE_ERROR"
+    assert uninstall_error.value.code == "PLUGIN_STORAGE_ERROR"
+    assert container.credentials.resolve("openai") == "provider-private-secret"
+
+
+def test_secret_delete_restores_reference_when_credential_delete_fails(
+    monkeypatch,
+) -> None:
+    container = build_container()
+    container.plugins.put_setting_secret("text-tools", "api_key", "keep-me")
+    settings_path = get_settings().data_dir / "plugins" / "settings.json"
+    original = settings_path.read_text(encoding="utf-8")
+    reference = _secret_reference("text-tools", "api_key")
+
+    def fail_delete(_credential_id: str) -> bool:
+        raise CredentialStoreError("injected delete failure")
+
+    monkeypatch.setattr(container.credentials, "delete", fail_delete)
+
+    with pytest.raises(ExtensionError) as exc:
+        container.plugins.delete_setting_secret("text-tools", "api_key")
+
+    assert exc.value.code == "PLUGIN_SECRET_STORE_ERROR"
+    assert settings_path.read_text(encoding="utf-8") == original
+    assert container.credentials.resolve(reference) == "keep-me"
+
+
+def test_uninstall_restores_settings_when_atomic_secret_delete_fails(
+    monkeypatch,
+) -> None:
+    container = build_container()
+    container.plugins.update_settings("text-tools", 1, {"result_limit": 12})
+    container.plugins.put_setting_secret("text-tools", "api_key", "keep-me")
+    settings_path = get_settings().data_dir / "plugins" / "settings.json"
+    original = settings_path.read_text(encoding="utf-8")
+    reference = _secret_reference("text-tools", "api_key")
+
+    def fail_delete_many(_credential_ids: list[str]) -> set[str]:
+        raise CredentialStoreError("injected batch delete failure")
+
+    monkeypatch.setattr(container.credentials, "delete_many", fail_delete_many)
+
+    with pytest.raises(ExtensionError) as exc:
+        container.plugins.uninstall("text-tools")
+
+    assert exc.value.code == "PLUGIN_SECRET_STORE_ERROR"
+    assert settings_path.read_text(encoding="utf-8") == original
+    assert container.credentials.resolve(reference) == "keep-me"
+    assert container.plugins.get("text-tools").manifest.plugin_id == "text-tools"
+
+
+def test_invalid_command_and_settings_manifest_are_rejected(tmp_path: Path) -> None:
+    invalid_command = tmp_path / "invalid-command"
+    invalid_command.mkdir()
+    (invalid_command / "plugin.yaml").write_text(
+        """
+id: invalid-command
+name: Invalid Command
+version: 1.0.0
+contributes:
+  commands: [other.run]
+""".strip(),
+        encoding="utf-8",
+    )
+    (invalid_command / "commands.yaml").write_text(
+        """
+commands:
+  - command_id: other.run
+    title: Invalid
+    locations: [command_palette]
+    handler: echo
+""".strip(),
+        encoding="utf-8",
+    )
+
+    invalid_settings = tmp_path / "invalid-settings"
+    invalid_settings.mkdir()
+    (invalid_settings / "plugin.yaml").write_text(
+        """
+id: invalid-settings
+name: Invalid Settings
+version: 1.0.0
+contributes:
+  settings_sections: [invalid-settings.general]
+""".strip(),
+        encoding="utf-8",
+    )
+    (invalid_settings / "settings.yaml").write_text(
+        """
+section_id: invalid-settings.general
+schema_version: 1
+fields:
+  - key: token
+    label: Token
+    type: secret
+    default: leaked-default
+""".strip(),
+        encoding="utf-8",
+    )
+
+    runtime = PluginRuntime(ToolRegistry())
+    with pytest.raises(ExtensionError) as command_error:
+        runtime.install(invalid_command)
+    assert command_error.value.code == "PLUGIN_COMMAND_INVALID"
+
+    with pytest.raises(ExtensionError) as settings_error:
+        runtime.install(invalid_settings)
+    assert settings_error.value.code == "PLUGIN_SETTINGS_SCHEMA_INVALID"
+
+
+@pytest.mark.parametrize("bound", [".nan", ".inf", "-.inf"])
+def test_non_finite_setting_bounds_are_rejected(tmp_path: Path, bound: str) -> None:
+    package = tmp_path / f"invalid-bound-{bound.replace('.', 'dot').replace('-', 'neg')}"
+    package.mkdir()
+    (package / "plugin.yaml").write_text(
+        """
+id: invalid-bound
+name: Invalid Bound
+version: 1.0.0
+contributes:
+  settings_sections: [invalid-bound.general]
+backend:
+  type: none
+  transport: none
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "settings.yaml").write_text(
+        f"""
+section_id: invalid-bound.general
+schema_version: 1
+fields:
+  - key: limit
+    label: Limit
+    type: number
+    minimum: {bound}
+""".strip(),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ExtensionError) as exc:
+        PluginRuntime(ToolRegistry()).install(package)
+
+    assert exc.value.code == "PLUGIN_SETTINGS_SCHEMA_INVALID"
+    assert "must be finite" in exc.value.message
+
+
+def test_null_command_list_returns_stable_manifest_error(tmp_path: Path) -> None:
+    package = tmp_path / "null-commands"
+    package.mkdir()
+    (package / "plugin.yaml").write_text(
+        """
+id: null-commands
+name: Null Commands
+version: 1.0.0
+contributes:
+  commands: []
+backend:
+  type: internal_rpc
+  transport: none
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "commands.yaml").write_text("commands:\n", encoding="utf-8")
+
+    with pytest.raises(ExtensionError) as exc:
+        PluginRuntime(ToolRegistry()).install(package)
+
+    assert exc.value.code == "EXTENSION_MANIFEST_INVALID"
+
+
+def test_external_command_schema_reference_is_rejected(tmp_path: Path) -> None:
+    package = tmp_path / "external-ref"
+    package.mkdir()
+    (package / "plugin.yaml").write_text(
+        """
+id: external-ref
+name: External Ref
+version: 1.0.0
+contributes:
+  commands: [external-ref.run]
+backend:
+  type: internal_rpc
+  transport: none
+""".strip(),
+        encoding="utf-8",
+    )
+    (package / "commands.yaml").write_text(
+        """
+commands:
+  - command_id: external-ref.run
+    title: External Ref
+    locations: [command_palette]
+    handler: echo
+    parameters:
+      $ref: file:///host/private-schema.json
+""".strip(),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ExtensionError) as exc:
+        PluginRuntime(ToolRegistry()).install(package)
+
+    assert exc.value.code == "PLUGIN_COMMAND_INVALID"
+    assert "External JSON Schema reference" in exc.value.message
+
+
+def test_settings_missing_and_secret_field_errors_are_stable() -> None:
+    container = build_container()
+
+    with pytest.raises(ExtensionError) as missing:
+        container.plugins.get_settings("does-not-exist")
+    assert missing.value.code == "PLUGIN_NOT_FOUND"
+
+    with pytest.raises(ExtensionError) as field:
+        container.plugins.put_setting_secret("text-tools", "result_limit", "secret")
+    assert field.value.code == "PLUGIN_SECRET_FIELD_NOT_FOUND"
+
+    with pytest.raises(ExtensionError) as empty:
+        container.plugins.put_setting_secret("text-tools", "api_key", "")
+    assert empty.value.code == "PLUGIN_SECRET_VALUE_INVALID"
+
+
+def test_corrupted_plugin_settings_namespace_returns_stable_error() -> None:
+    container = build_container()
+    settings_path = get_settings().data_dir / "plugins" / "settings.json"
+    settings_path.parent.mkdir(parents=True, exist_ok=True)
+    settings_path.write_text('{"text-tools": []}', encoding="utf-8")
+
+    with pytest.raises(ExtensionError) as exc:
+        container.plugins.get_settings("text-tools")
+
+    assert exc.value.code == "PLUGIN_STORAGE_ERROR"
+
+    with pytest.raises(ExtensionError) as secret_exc:
+        container.plugins.put_setting_secret("text-tools", "api_key", "must-not-orphan")
+
+    assert secret_exc.value.code == "PLUGIN_STORAGE_ERROR"
+    credentials_path = get_settings().data_dir / "credentials" / "credentials.json"
+    credential_ids = (
+        json.loads(credentials_path.read_text(encoding="utf-8")).keys()
+        if credentials_path.exists()
+        else []
+    )
+    assert not any(item.startswith("plugin.") for item in credential_ids)

backend/tests/test_schema_security.py

@@ -0,0 +1,65 @@
+import pytest
+
+from app.schema_security import (
+    ExternalSchemaReferenceError,
+    UnresolvableLocalSchemaReferenceError,
+    reject_external_schema_references,
+)
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [
+        {"$ref": "file:///host/private-schema.json"},
+        {"properties": {"value": {"$ref": "https://schema.invalid/value.json"}}},
+        {"allOf": [{"$dynamicRef": "https://schema.invalid/dynamic"}]},
+    ],
+)
+def test_external_json_schema_references_are_rejected(schema) -> None:
+    with pytest.raises(ExternalSchemaReferenceError):
+        reject_external_schema_references(schema)
+
+
+def test_local_json_schema_fragment_reference_is_allowed() -> None:
+    reject_external_schema_references(
+        {
+            "$defs": {"value": {"type": "string"}},
+            "properties": {"value": {"$ref": "#/$defs/value"}},
+        }
+    )
+
+
+@pytest.mark.parametrize("reference", ["#/$defs/missing", "#missing-anchor"])
+def test_unresolvable_local_schema_reference_is_rejected(reference: str) -> None:
+    with pytest.raises(UnresolvableLocalSchemaReferenceError):
+        reject_external_schema_references({"type": "object", "$ref": reference})
+
+
+def test_root_reference_cannot_use_anchor_from_nested_schema_resource() -> None:
+    schema = {
+        "$defs": {
+            "nested": {
+                "$id": "nested",
+                "$anchor": "inside",
+                "type": "string",
+            }
+        },
+        "properties": {"value": {"$ref": "#inside"}},
+    }
+
+    with pytest.raises(UnresolvableLocalSchemaReferenceError):
+        reject_external_schema_references(schema)
+
+
+def test_nested_schema_resource_can_resolve_its_own_anchor() -> None:
+    schema = {
+        "$defs": {
+            "nested": {
+                "$id": "nested",
+                "$anchor": "inside",
+                "allOf": [{"$ref": "#inside"}],
+            }
+        }
+    }
+
+    reject_external_schema_references(schema)

backend/uv.lock

@@ -374,6 +374,7 @@ dependencies = [
     { name = "httpx" },
     { name = "jsonschema" },
     { name = "pyyaml" },
+    { name = "referencing" },
     { name = "sqlite-vec" },
     { name = "uvicorn", extra = ["standard"] },
 ]
@@ -390,6 +391,7 @@ requires-dist = [
     { name = "httpx", specifier = ">=0.28,<1.0" },
     { name = "jsonschema", specifier = ">=4.25,<5.0" },
     { name = "pyyaml", specifier = ">=6.0,<7.0" },
+    { name = "referencing", specifier = ">=0.36,<1.0" },
     { name = "sqlite-vec", specifier = ">=0.1.9" },
     { name = "uvicorn", extras = ["standard"], specifier = ">=0.35,<1.0" },
 ]

frontend/src/contracts/index.ts

@@ -255,6 +255,102 @@ 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 type PluginCommandLocation = 'command_palette' | 'context_menu' | 'toolbar'
+
+export interface PluginCommand {
+  command_id: string
+  plugin_id: string
+  title: string
+  description: string
+  icon?: string | null
+  locations: PluginCommandLocation[]
+  when: string[]
+  parameters: Record<string, unknown>
+  enabled: boolean
+}
+
+export interface PluginCommandContext {
+  vault_id?: string | null
+  note_id?: string | null
+  file_path?: string | null
+  selection?: string | null
+}
+
+export type PluginCommandEffect =
+  | { type: 'none'; payload: Record<string, never> }
+  | {
+      type: 'notification'
+      payload: { level: 'info' | 'success' | 'warning' | 'error'; message: string }
+    }
+  | {
+      type: 'navigate'
+      payload: {
+        route:
+          | 'vault-entry'
+          | 'workspace'
+          | 'search'
+          | 'chat'
+          | 'agent'
+          | 'tasks'
+          | 'skills'
+          | 'plugins'
+          | 'themes'
+          | 'settings'
+      }
+    }
+  | { type: 'refresh'; payload: { scope: 'workspace' | 'commands' | 'settings' | 'plugins' } }
+  | { type: 'job'; payload: { job_id: string } }
+
+export interface PluginCommandResult {
+  command_id: string
+  status: 'completed'
+  effect: PluginCommandEffect
+}
+
+export type PluginSettingType = 'string' | 'number' | 'boolean' | 'select' | 'secret'
+
+export interface PluginSettingField {
+  key: string
+  label: string
+  description: string
+  type: PluginSettingType
+  required: boolean
+  default?: unknown
+  minimum?: number | null
+  maximum?: number | null
+  options: string[]
+}
+
+export interface PluginSettingsSchema {
+  plugin_id: string
+  schema_version: number
+  fields: PluginSettingField[]
+  values: Record<string, unknown>
+  secrets: Record<string, { configured: boolean }>
+}
+
+export interface PluginSecretStatus {
+  plugin_id: string
+  key: string
+  configured: boolean
+}
+
 export interface PluginContribution {
   type: 'tool' | 'command' | 'importer' | 'exporter' | 'sidebar_panel' | 'settings_section'
   id: string
@@ -539,7 +635,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.spec.ts

@@ -0,0 +1,85 @@
+// @vitest-environment happy-dom
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import * as pluginService from './pluginService'
+
+function jsonResponse(body: unknown) {
+  return new Response(JSON.stringify(body), {
+    status: 200,
+    headers: { 'Content-Type': 'application/json' },
+  })
+}
+
+beforeEach(() => {
+  vi.stubGlobal('fetch', vi.fn())
+})
+
+afterEach(() => {
+  vi.unstubAllGlobals()
+  vi.restoreAllMocks()
+})
+
+describe('pluginService contribution adapter', () => {
+  it('lists and executes Plugin Commands with scoped wire fields', async () => {
+    const fetchMock = vi.mocked(fetch)
+    fetchMock
+      .mockResolvedValueOnce(jsonResponse({ items: [{ command_id: 'text-tools.uppercase-selection' }] }))
+      .mockResolvedValueOnce(jsonResponse({
+        command_id: 'text-tools.uppercase-selection',
+        status: 'completed',
+        effect: { type: 'notification', payload: { level: 'success', message: 'HELLO' } },
+      }))
+
+    const commands = await pluginService.listPluginCommands('command_palette')
+    const result = await pluginService.executePluginCommand(
+      'text-tools.uppercase-selection',
+      {},
+      { note_id: 'note-1', selection: 'hello' },
+    )
+
+    expect(commands[0].command_id).toBe('text-tools.uppercase-selection')
+    if (result.effect.type !== 'notification') throw new Error('expected notification effect')
+    expect(result.effect.payload.message).toBe('HELLO')
+    expect(fetchMock.mock.calls[0][0]).toBe(
+      '/api/plugin-contributions/commands?location=command_palette',
+    )
+    expect(JSON.parse(String(fetchMock.mock.calls[1][1]?.body))).toEqual({
+      arguments: {},
+      context: { note_id: 'note-1', selection: 'hello' },
+    })
+  })
+
+  it('uses separate Settings and Secret endpoints', async () => {
+    const fetchMock = vi.mocked(fetch)
+    fetchMock
+      .mockResolvedValueOnce(jsonResponse({
+        plugin_id: 'text-tools', schema_version: 1, fields: [],
+        values: { result_limit: 10 }, secrets: { api_key: { configured: false } },
+      }))
+      .mockResolvedValueOnce(jsonResponse({
+        plugin_id: 'text-tools', schema_version: 1, fields: [],
+        values: { result_limit: 20 }, secrets: { api_key: { configured: false } },
+      }))
+      .mockResolvedValueOnce(jsonResponse({ plugin_id: 'text-tools', key: 'api_key', configured: true }))
+      .mockResolvedValueOnce(jsonResponse({ plugin_id: 'text-tools', key: 'api_key', configured: false }))
+
+    await pluginService.getPluginSettings('text-tools')
+    await pluginService.updatePluginSettings('text-tools', 1, { result_limit: 20 })
+    await pluginService.putPluginSecret('text-tools', 'api_key', 'request-only-secret')
+    await pluginService.deletePluginSecret('text-tools', 'api_key')
+
+    expect(fetchMock.mock.calls.map(([url]) => url)).toEqual([
+      '/api/plugins/text-tools/settings',
+      '/api/plugins/text-tools/settings',
+      '/api/plugins/text-tools/settings/api_key/secret',
+      '/api/plugins/text-tools/settings/api_key/secret',
+    ])
+    expect(JSON.parse(String(fetchMock.mock.calls[1][1]?.body))).toEqual({
+      schema_version: 1,
+      values: { result_limit: 20 },
+    })
+    expect(JSON.parse(String(fetchMock.mock.calls[2][1]?.body))).toEqual({
+      secret: 'request-only-secret',
+    })
+    expect(fetchMock.mock.calls[3][1]?.method).toBe('DELETE')
+  })
+})

frontend/src/services/pluginService.ts

@@ -1,5 +1,17 @@
 import apiClient from './apiClient'
-import type { ApiPlugin, OperationResponse, Plugin, PluginContribution } from '@/contracts'
+import type {
+  ApiPlugin,
+  OperationResponse,
+  Plugin,
+  PluginCommand,
+  PluginCommandContext,
+  PluginCommandLocation,
+  PluginCommandResult,
+  PluginContribution,
+  PluginHostStatus,
+  PluginSecretStatus,
+  PluginSettingsSchema,
+} from '@/contracts'
 
 function toPlugin(plugin: ApiPlugin): Plugin {
   const { manifest } = plugin
@@ -54,6 +66,63 @@ 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 listPluginCommands(location?: PluginCommandLocation): Promise<PluginCommand[]> {
+  const query = location ? `?location=${encodeURIComponent(location)}` : ''
+  const response = await apiClient.get<{ items: PluginCommand[] }>(`/api/plugin-contributions/commands${query}`)
+  return response.items
+}
+
+export async function executePluginCommand(
+  commandId: string,
+  argumentsValue: Record<string, unknown> = {},
+  context: PluginCommandContext = {},
+): Promise<PluginCommandResult> {
+  return apiClient.post(`/api/plugin-contributions/commands/${encodeURIComponent(commandId)}/execute`, {
+    arguments: argumentsValue,
+    context,
+  })
+}
+
+export async function getPluginSettings(pluginId: string): Promise<PluginSettingsSchema> {
+  return apiClient.get(`/api/plugins/${encodeURIComponent(pluginId)}/settings`)
+}
+
+export async function updatePluginSettings(
+  pluginId: string,
+  schemaVersion: number,
+  values: Record<string, unknown>,
+): Promise<PluginSettingsSchema> {
+  return apiClient.put(`/api/plugins/${encodeURIComponent(pluginId)}/settings`, {
+    schema_version: schemaVersion,
+    values,
+  })
+}
+
+export async function putPluginSecret(
+  pluginId: string,
+  key: string,
+  secret: string,
+): Promise<PluginSecretStatus> {
+  return apiClient.put(
+    `/api/plugins/${encodeURIComponent(pluginId)}/settings/${encodeURIComponent(key)}/secret`,
+    { secret },
+  )
+}
+
+export async function deletePluginSecret(pluginId: string, key: string): Promise<PluginSecretStatus> {
+  return apiClient.delete(
+    `/api/plugins/${encodeURIComponent(pluginId)}/settings/${encodeURIComponent(key)}/secret`,
+  )
+}
+
 export async function uninstallPlugin(pluginId: string): Promise<OperationResponse> {
   return apiClient.delete(`/api/plugins/${pluginId}`)
 }