diff --git a/.github/workflows/plugin-api-v2.yml b/.github/workflows/plugin-api-v2.yml
deleted file mode 100644
index 7c1876b..0000000
--- a/.github/workflows/plugin-api-v2.yml
+++ /dev/null
@@ -1,28 +0,0 @@
-name: plugin-api-v2
-
-on:
-  pull_request:
-  push:
-    branches:
-      - main
-
-permissions:
-  contents: read
-
-jobs:
-  contract:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/checkout@v4
-        with:
-          repository: akashic-plugins/plugin-contracts
-          ref: 24543445c7b99ca63fcd90b5828f754a148b184c
-          path: .plugin-contracts
-      - uses: actions/setup-python@v5
-        with:
-          python-version: "3.13"
-      - name: Check Plugin API v2
-        env:
-          PYTHONPATH: .plugin-contracts
-        run: python -m akashic_plugin_contracts check plugin.py
diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml
new file mode 100644
index 0000000..8d0e2be
--- /dev/null
+++ b/.github/workflows/plugin-api-v3.yml
@@ -0,0 +1,67 @@
+name: plugin-api-v3
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  contract:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/checkout@v4
+        with:
+          repository: akashic-plugins/plugin-contracts
+          ref: 4dd69dd621e029e51e99aa428443fa3a4ec1f6cf
+          path: .plugin-contracts
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+      - name: Check Plugin API v3
+        env:
+          PYTHONPATH: .plugin-contracts
+        run: python -m akashic_plugin_contracts check plugin.py
+
+  plugin-tests:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/checkout@v4
+        with:
+          repository: kachofugetsu09/akashic-agent
+          ref: 3005f838bcd96e2cbc58616aede46e4f39df4523
+          path: .akashic-core
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+          cache: pip
+          cache-dependency-path: |
+            .akashic-core/requirements.txt
+            mcp/requirements.txt
+      - name: Stage exact Steam runtime
+        run: |
+          python -m venv mcp/.venv
+          mcp/.venv/bin/python -m pip install \
+            -r .akashic-core/requirements.txt \
+            -r .akashic-core/requirements-dev.txt \
+            -r mcp/requirements.txt
+      - name: Verify Steam v3 composition and migration
+        env:
+          AKASHIC_AGENT_ROOT: .akashic-core
+          PYTHONPATH: .akashic-core:mcp:mcp/.venv/lib/python3.13/site-packages
+        run: mcp/.venv/bin/python -m pytest -q mcp/tests tests
+      - name: Check changed v3 sources
+        env:
+          PYTHONPATH: .akashic-core:mcp
+        run: mcp/.venv/bin/pyright plugin.py mcp/runtime_config.py mcp/run_mcp.py scripts tests
+      - name: Compile Python sources
+        run: python -m compileall -q plugin.py mcp scripts tests
+      - name: Check diff formatting
+        run: git diff --check
diff --git a/README.md b/README.md
index dbe9914..b2fae2f 100644
--- a/README.md
+++ b/README.md
@@ -18,7 +18,7 @@ Restart Akashic after install.
 Runtime data lives in:
 
 ```text
-~/.akashic-plugin/data/steam-<marketplace>/
+<workspace>/plugin-data/steam-<marketplace>/
 ```
 
 Common files:
@@ -43,4 +43,22 @@ Create `steam_mcp_config.json` in the plugin data directory:
 `get_steam_context` 每次读取实时在线状态，并在历史游戏时长快照超过
 `snapshot_interval_seconds` 时自动刷新。空的最近游玩列表也会记录快照批次，避免重复刷新。
 
-When migrating from the old workspace MCP, the plugin copies the old config and cache files automatically on first startup.
+## v2 data migration
+
+v3 不会在插件加载时隐式复制正式数据。停止 Akashic 后显式执行：
+
+```bash
+PYTHONPATH=/path/to/akashic-agent \
+python scripts/migrate_v2_data.py \
+  --workspace /path/to/workspace \
+  --marketplace github
+```
+
+迁移保留 `mcp/steam-mcp` 原文件，在
+`plugin-data/steam-<marketplace>/.steam-v2-migration.json` 写入 hash 与 SQLite
+完整性证据。进程内失败会回滚本次新增文件；进程崩溃后重跑会清理 staging，
+并只接纳已经发布且内容完全相同的文件。
+
+候选验证使用无凭证、无外网、无数据库的 recording backend；正式 MCP
+只从自己的 `plugin-data` 读取 `steam_mcp_config.json`，不读取 ambient
+`STEAM_API_KEY` 或 `STEAM_ID`。
diff --git a/akashic.plugin.toml b/akashic.plugin.toml
new file mode 100644
index 0000000..ccef591
--- /dev/null
+++ b/akashic.plugin.toml
@@ -0,0 +1,24 @@
+schema_version = 1
+name = "steam"
+version = "3.0.0"
+api_version = 3
+entrypoint = "plugin.py"
+
+[[python]]
+requirements = "mcp/requirements.txt"
+
+[validation]
+exclude_data_paths = [
+  "steam_mcp_config.json",
+  "steam_user_cache.json",
+  "steam_app_cache.json",
+  "steam_proactive.sqlite3",
+  ".steam-v2-migration.json",
+]
+
+[[mcp]]
+name = "steam"
+command = ["python", "mcp/run_mcp.py"]
+required_tools = ["get_steam_context"]
+candidate_read_only_tools = ["get_steam_context"]
+candidate_env = {STEAM_BACKEND = "recording"}
diff --git a/mcp/http_client.py b/mcp/http_client.py
index e8ad13f..84e1550 100644
--- a/mcp/http_client.py
+++ b/mcp/http_client.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-import os
 from pathlib import Path
 from typing import Any
 from urllib.error import HTTPError, URLError
@@ -65,10 +64,6 @@ def get(
             raise SteamApiError(f"Steam API request failed: {exc.reason}") from exc
 
     def _load_api_key(self) -> str:
-        env_api_key = os.environ.get("STEAM_API_KEY", "").strip()
-        if env_api_key:
-            return env_api_key
-
         if not self.config_path.exists():
             raise SteamApiError(
                 f"Steam API key is required. Create `{self.config_path}` and set `steam_api_key`."
diff --git a/mcp/run_mcp.py b/mcp/run_mcp.py
index 27d5fe3..d3cb30d 100644
--- a/mcp/run_mcp.py
+++ b/mcp/run_mcp.py
@@ -10,6 +10,15 @@ def main() -> None:
     if str(script_dir) not in sys.path:
         sys.path.insert(0, str(script_dir))
 
+    backend = os.environ.get("STEAM_BACKEND", "formal").strip().lower()
+    if backend not in {"formal", "recording"}:
+        raise RuntimeError(f"未知 STEAM_BACKEND: {backend}")
+    if backend == "formal":
+        from runtime_config import load_runtime_config
+
+        data_root = Path(os.environ["AKA_PLUGIN_DATA_DIR"]).resolve()
+        _ = load_runtime_config(data_root / "steam_mcp_config.json")
+
     from steam_mcp import mcp
 
     mcp.run(transport="stdio")
diff --git a/mcp/runtime_config.py b/mcp/runtime_config.py
new file mode 100644
index 0000000..02eac76
--- /dev/null
+++ b/mcp/runtime_config.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True, slots=True)
+class SteamRuntimeConfig:
+    steam_api_key: str
+    steam_id: str
+    snapshot_interval_seconds: int
+
+
+def load_runtime_config(path: Path) -> SteamRuntimeConfig:
+    """读取并校验 formal Steam runtime 配置。"""
+
+    # 1. 配置只来自 formal plugin-data，不接受 ambient secret
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError as error:
+        raise RuntimeError("Steam formal runtime 缺少 steam_mcp_config.json") from error
+    except json.JSONDecodeError as error:
+        raise RuntimeError("steam_mcp_config.json 不是合法 JSON") from error
+    if not isinstance(raw, dict):
+        raise RuntimeError("steam_mcp_config.json 根节点必须是 object")
+
+    # 2. 启动前建立完整 credential 与用户身份不变量
+    api_key = raw.get("steam_api_key")
+    steam_id = raw.get("steam_id")
+    if not isinstance(api_key, str) or not api_key.strip():
+        raise RuntimeError("steam_mcp_config.json 缺少 steam_api_key")
+    if not isinstance(steam_id, str) or not steam_id.strip():
+        raise RuntimeError("steam_mcp_config.json 缺少 steam_id")
+    interval = raw.get("snapshot_interval_seconds", 6 * 3600)
+    if not isinstance(interval, int) or isinstance(interval, bool) or interval < 300:
+        raise RuntimeError("snapshot_interval_seconds 必须是大于等于 300 的整数")
+    return SteamRuntimeConfig(api_key.strip(), steam_id.strip(), interval)
diff --git a/mcp/steam_mcp.py b/mcp/steam_mcp.py
index 98ccb18..afe4413 100644
--- a/mcp/steam_mcp.py
+++ b/mcp/steam_mcp.py
@@ -13,6 +13,7 @@
 mcp = FastMCP("steam-web-api")
 RUNTIME_DIR = Path(os.environ.get("AKA_PLUGIN_DATA_DIR", "").strip() or Path.cwd())
 RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
+BACKEND = os.environ.get("STEAM_BACKEND", "formal").strip().lower()
 http_client = HttpClient(config_path=str(RUNTIME_DIR / "steam_mcp_config.json"))
 SUPPORTED_FORMATS = {"json", "xml", "vdf"}
 MAX_STEAM_IDS_PER_REQUEST = 100
@@ -633,14 +634,30 @@ def get_steam_context() -> dict:
     """获取用户 Steam 游戏活动的持久上下文，供 proactive engine 注入 background_context。
     返回近两周游戏时长、历史对比、当前在线状态等结构化数据。
     """
+    if BACKEND == "recording":
+        return {
+            "items": [
+                {
+                    "presence": "unknown",
+                    "interruptibility": 0.4,
+                    "confidence": 0.0,
+                    "transition": "",
+                    "recording": True,
+                }
+            ]
+        }
     import steam_proactive
-    return steam_proactive.get_context()
+
+    return {"items": [steam_proactive.get_context()]}
 
 
 @mcp.tool()
 def take_steam_snapshot() -> dict:
     """拉取并存储一次 Steam 游戏数据快照。通常由定时任务调用，也可手动触发。"""
+    if BACKEND == "recording":
+        raise RuntimeError("recording backend 禁止写 Steam snapshot")
     import steam_proactive
+
     return steam_proactive.take_snapshot()
 
 
diff --git a/mcp/steam_proactive.py b/mcp/steam_proactive.py
index cb92d0c..44a49cc 100644
--- a/mcp/steam_proactive.py
+++ b/mcp/steam_proactive.py
@@ -38,12 +38,6 @@ def _load_config() -> dict:
             raise ValueError("steam_mcp_config.json 根节点必须是 object")
     else:
         loaded = {}
-    steam_api_key = os.environ.get("STEAM_API_KEY", "").strip()
-    steam_id = os.environ.get("STEAM_ID", "").strip()
-    if steam_api_key:
-        loaded["steam_api_key"] = steam_api_key
-    if steam_id:
-        loaded["steam_id"] = steam_id
     return loaded
 
 
diff --git a/mcp/tests/test_v3_runtime.py b/mcp/tests/test_v3_runtime.py
new file mode 100644
index 0000000..3b15461
--- /dev/null
+++ b/mcp/tests/test_v3_runtime.py
@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import importlib
+import json
+import sys
+
+import pytest
+
+from runtime_config import load_runtime_config
+
+
+def test_formal_runtime_config_requires_plugin_data_credentials(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    config_path = tmp_path / "steam_mcp_config.json"
+    config_path.write_text(
+        json.dumps({"steam_id": "user", "snapshot_interval_seconds": 3600}),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("STEAM_API_KEY", "ambient-secret")
+
+    with pytest.raises(RuntimeError, match="缺少 steam_api_key"):
+        load_runtime_config(config_path)
+
+
+def test_formal_runtime_config_accepts_complete_file(tmp_path) -> None:
+    config_path = tmp_path / "steam_mcp_config.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "steam_api_key": "formal-secret",
+                "steam_id": "user",
+                "snapshot_interval_seconds": 3600,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    config = load_runtime_config(config_path)
+
+    assert config.steam_api_key == "formal-secret"
+    assert config.steam_id == "user"
+    assert config.snapshot_interval_seconds == 3600
+
+
+def test_recording_context_never_reads_formal_config_or_creates_database(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    data_root = tmp_path / "candidate-data"
+    monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(data_root))
+    monkeypatch.setenv("STEAM_BACKEND", "recording")
+    monkeypatch.setenv("STEAM_API_KEY", "ambient-secret")
+    monkeypatch.setenv("STEAM_ID", "ambient-user")
+    sys.modules.pop("steam_mcp", None)
+    sys.modules.pop("steam_proactive", None)
+
+    module = importlib.import_module("steam_mcp")
+    result = module.get_steam_context()
+
+    assert result == {
+        "items": [
+            {
+                "presence": "unknown",
+                "interruptibility": 0.4,
+                "confidence": 0.0,
+                "transition": "",
+                "recording": True,
+            }
+        ]
+    }
+    assert "steam_proactive" not in sys.modules
+    assert not (data_root / "steam_mcp_config.json").exists()
+    assert not (data_root / "steam_proactive.sqlite3").exists()
diff --git a/plugin.py b/plugin.py
index 38b9289..5317ea4 100644
--- a/plugin.py
+++ b/plugin.py
@@ -1,12 +1,14 @@
 from __future__ import annotations
 
-import shutil
-from pathlib import Path
-from typing import cast
-
 from pydantic import BaseModel, Field
 
-from agent.plugins import McpServerSpec, Plugin, ProactiveSourceSpec
+from agent.plugin_composition import (
+    MCP_SERVERS,
+    PROACTIVE_COMPONENTS,
+    Context,
+    McpServerDefinition,
+    ProactiveSourceDefinition,
+)
 
 
 class SteamProactiveConfig(BaseModel):
@@ -17,73 +19,41 @@ class SteamConfig(BaseModel):
     proactive: SteamProactiveConfig = Field(default_factory=SteamProactiveConfig)
 
 
-class SteamPlugin(Plugin):
-    api_version = 2
-    name = "steam"
-    version = "1.1.0"
-    desc = "Steam MCP plugin"
-    ConfigModel = SteamConfig
-
-    @classmethod
-    def skill_roots(cls) -> tuple[str, ...]:
-        return ("skills",)
-
-    @classmethod
-    def mcp_servers(cls) -> list[McpServerSpec]:
-        return [
-            McpServerSpec(
-                name="steam",
-                command=("python", "mcp/run_mcp.py"),
-            )
-        ]
-
-    def proactive_sources(self) -> list[ProactiveSourceSpec]:
-        config = cast(SteamConfig, self.context.config)
-        if not config.proactive.enabled:
-            return []
-        return [
-            ProactiveSourceSpec(
-                id="presence",
+api_version = 3
+name = "steam"
+version = "3.0.0"
+desc = "Steam MCP plugin"
+Config = SteamConfig
+inject = (MCP_SERVERS, PROACTIVE_COMPONENTS)
+skill_roots = ("skills",)
+
+
+async def apply(ctx: Context, config: object) -> None:
+    """声明 Steam MCP 与可选的主动上下文源。"""
+
+    if not isinstance(config, SteamConfig):
+        raise TypeError("steam config 必须是 SteamConfig")
+
+    # 1. MCP 由 Core staged Python runtime 启动，candidate 仅开放 recording 上下文
+    await ctx.require(MCP_SERVERS).register(
+        ctx,
+        McpServerDefinition(
+            name="steam",
+            command=("python", "mcp/run_mcp.py"),
+            required_tools=("get_steam_context",),
+            candidate_read_only_tools=("get_steam_context",),
+            candidate_env={"STEAM_BACKEND": "recording"},
+        ),
+    )
+
+    # 2. 主动源只消费明确的 FetchItems/FetchEmpty 结果
+    if config.proactive.enabled:
+        await ctx.require(PROACTIVE_COMPONENTS).register(
+            ctx,
+            ProactiveSourceDefinition(
+                name="presence",
                 channels=("context",),
-                server="steam",
+                mcp_server="steam",
                 fetch_tool="get_steam_context",
-            )
-        ]
-
-    def activate(self) -> None:
-        data_dir = self.context.data_dir
-        workspace = self.context.workspace
-        if data_dir is None or workspace is None:
-            return
-        data_dir.mkdir(parents=True, exist_ok=True)
-        if _has_state(data_dir):
-            return
-        _copy_legacy_state(workspace / "mcp" / "steam-mcp", data_dir)
-
-
-def _has_state(data_dir: Path) -> bool:
-    for name in (
-        "steam_mcp_config.json",
-        "steam_user_cache.json",
-        "steam_app_cache.json",
-        "steam_proactive.sqlite3",
-    ):
-        if (data_dir / name).exists():
-            return True
-    return False
-
-
-def _copy_legacy_state(source_dir: Path, data_dir: Path) -> None:
-    if not source_dir.exists():
-        return
-    for name in (
-        "steam_mcp_config.json",
-        "steam_user_cache.json",
-        "steam_app_cache.json",
-        "steam_proactive.sqlite3",
-    ):
-        source = source_dir / name
-        target = data_dir / name
-        if not source.exists() or target.exists():
-            continue
-        shutil.copy2(source, target)
+            ),
+        )
diff --git a/pyrightconfig.json b/pyrightconfig.json
new file mode 100644
index 0000000..7bf803e
--- /dev/null
+++ b/pyrightconfig.json
@@ -0,0 +1,18 @@
+{
+  "include": [
+    "plugin.py",
+    "mcp",
+    "scripts",
+    "tests"
+  ],
+  "exclude": ["**/__pycache__"],
+  "venvPath": "mcp",
+  "venv": ".venv",
+  "executionEnvironments": [
+    {
+      "root": ".",
+      "pythonVersion": "3.13",
+      "extraPaths": ["mcp", ".akashic-core"]
+    }
+  ]
+}
diff --git a/scripts/__init__.py b/scripts/__init__.py
new file mode 100644
index 0000000..bd8dd68
--- /dev/null
+++ b/scripts/__init__.py
@@ -0,0 +1 @@
+"""Steam 运维脚本。"""
diff --git a/scripts/migrate_v2_data.py b/scripts/migrate_v2_data.py
new file mode 100644
index 0000000..2b839ad
--- /dev/null
+++ b/scripts/migrate_v2_data.py
@@ -0,0 +1,293 @@
+#!/usr/bin/env python3
+"""把 Steam v2 workspace 数据非破坏迁移到 v3 plugin-data。"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shutil
+import sqlite3
+import uuid
+from contextlib import closing
+from pathlib import Path
+
+from agent.plugins.manifest import (
+    ensure_workspace_plugin_data_dir,
+    validate_workspace_plugin_data_path,
+)
+from bootstrap.workspace_lock import WorkspaceInstanceLock
+
+
+_DATA_FILES = (
+    "steam_mcp_config.json",
+    "steam_user_cache.json",
+    "steam_app_cache.json",
+    "steam_proactive.sqlite3",
+)
+_RECEIPT = ".steam-v2-migration.json"
+
+
+def _digest(path: Path) -> str:
+    value = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            value.update(chunk)
+    return value.hexdigest()
+
+
+def _sqlite_integrity(path: Path) -> str:
+    """只读校验 SQLite 文件。"""
+
+    uri = f"{path.resolve().as_uri()}?mode=ro"
+    with closing(sqlite3.connect(uri, uri=True)) as database:
+        result = database.execute("PRAGMA integrity_check").fetchone()
+    if result != ("ok",):
+        raise sqlite3.DatabaseError(f"Steam SQLite 完整性检查失败: {path} ({result})")
+    return "ok"
+
+
+def _copy_sqlite(source: Path, destination: Path) -> None:
+    """用 SQLite backup 生成一致副本。"""
+
+    _sqlite_integrity(source)
+    uri = f"{source.resolve().as_uri()}?mode=ro"
+    with closing(sqlite3.connect(uri, uri=True)) as source_db:
+        with closing(sqlite3.connect(destination)) as destination_db:
+            source_db.backup(destination_db, pages=256, sleep=0.1)
+            destination_db.commit()
+    _sqlite_integrity(destination)
+
+
+def _stage(source: Path, staging: Path) -> tuple[dict[str, object], ...]:
+    """复制全部现存 v2 文件并生成内容证据。"""
+
+    entries: list[dict[str, object]] = []
+    for name in _DATA_FILES:
+        source_file = source / name
+        if not source_file.exists() and not source_file.is_symlink():
+            entries.append({"name": name, "status": "source_missing"})
+            continue
+        if source_file.is_symlink() or not source_file.is_file():
+            raise ValueError(f"Steam v2 数据不是普通文件: {source_file}")
+        staged_file = staging / name
+        if name.endswith(".sqlite3"):
+            _copy_sqlite(source_file, staged_file)
+        else:
+            shutil.copy2(source_file, staged_file)
+        entry: dict[str, object] = {
+            "name": name,
+            "status": "staged",
+            "sha256": _digest(staged_file),
+            "size": staged_file.stat().st_size,
+        }
+        if name.endswith(".sqlite3"):
+            entry["sqlite_integrity"] = "ok"
+        entries.append(entry)
+    return tuple(entries)
+
+
+def _record_target(entry: dict[str, object], destination: Path) -> None:
+    entry["sha256"] = _digest(destination)
+    entry["size"] = destination.stat().st_size
+    if destination.name.endswith(".sqlite3"):
+        entry["sqlite_integrity"] = _sqlite_integrity(destination)
+
+
+def _validate_targets(target: Path, entries: tuple[dict[str, object], ...]) -> None:
+    """拒绝覆盖不同内容，并收束进程崩溃留下的同内容文件。"""
+
+    for entry in entries:
+        destination = target / str(entry["name"])
+        if destination.is_symlink():
+            raise ValueError(f"Steam v3 目标不得是符号链接: {destination}")
+        if entry["status"] == "source_missing":
+            if not destination.exists():
+                continue
+            if not destination.is_file():
+                raise FileExistsError(f"Steam v3 目标不是普通文件: {destination}")
+            entry["status"] = "target_only"
+            _record_target(entry, destination)
+            continue
+        if not destination.exists():
+            entry["status"] = "copied"
+            continue
+        if (
+            not destination.is_file()
+            or destination.stat().st_size != entry["size"]
+            or _digest(destination) != entry["sha256"]
+        ):
+            raise FileExistsError(f"Steam v3 目标已存在且内容不同: {destination}")
+        entry["status"] = "verified"
+        if destination.name.endswith(".sqlite3"):
+            _sqlite_integrity(destination)
+    if all(entry["status"] == "source_missing" for entry in entries):
+        raise FileNotFoundError("Steam v2 与 v3 数据目录都没有可迁移文件")
+
+
+def _publish(
+    staging: Path,
+    target: Path,
+    entries: tuple[dict[str, object], ...],
+    receipt: dict[str, object],
+) -> None:
+    """发布本事务创建的文件，进程内失败时完整回滚。"""
+
+    published: list[Path] = []
+    try:
+        for entry in entries:
+            if entry["status"] != "copied":
+                continue
+            destination = target / str(entry["name"])
+            os.replace(staging / str(entry["name"]), destination)
+            published.append(destination)
+        staged_receipt = staging / _RECEIPT
+        staged_receipt.write_text(
+            json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+        receipt_path = target / _RECEIPT
+        os.replace(staged_receipt, receipt_path)
+        published.append(receipt_path)
+    except BaseException:
+        for path in reversed(published):
+            path.unlink(missing_ok=True)
+        raise
+
+
+def _remove_crash_staging(workspace: Path) -> None:
+    """清理上次进程崩溃留下的未发布 staging。"""
+
+    plugin_data = workspace / "plugin-data"
+    if plugin_data.is_symlink():
+        raise ValueError(f"Steam plugin-data 目录不得是符号链接: {plugin_data}")
+    if not plugin_data.is_dir():
+        return
+    for path in plugin_data.glob(".steam-v2-migrate-*"):
+        if path.is_symlink() or path.is_file():
+            path.unlink()
+        elif path.is_dir():
+            shutil.rmtree(path)
+
+
+def _valid_receipt(path: Path, *, target: Path, marketplace: str) -> bool:
+    """验证已有最终 receipt 及其目标文件。"""
+
+    if not path.exists() and not path.is_symlink():
+        return False
+    if path.is_symlink() or not path.is_file():
+        raise ValueError(f"Steam migration receipt 不是普通文件: {path}")
+    value = json.loads(path.read_text(encoding="utf-8"))
+    files = value.get("files") if isinstance(value, dict) else None
+    if (
+        not isinstance(value, dict)
+        or value.get("schema_version") != 1
+        or value.get("source") != "mcp/steam-mcp"
+        or value.get("target") != f"plugin-data/steam-{marketplace}"
+        or value.get("recovery")
+        != {"kind": "retained_source", "path": "mcp/steam-mcp"}
+        or not isinstance(files, list)
+        or [item.get("name") for item in files if isinstance(item, dict)]
+        != list(_DATA_FILES)
+    ):
+        raise ValueError(f"Steam migration receipt 无效: {path}")
+    for item in files:
+        if not isinstance(item, dict) or item.get("status") not in {
+            "source_missing",
+            "target_only",
+            "verified",
+            "copied",
+        }:
+            raise ValueError(f"Steam migration receipt 无效: {path}")
+        destination = target / str(item.get("name"))
+        if item["status"] == "source_missing":
+            if destination.exists() or destination.is_symlink():
+                raise ValueError(f"Steam migration receipt 目标漂移: {destination}")
+            continue
+        digest = item.get("sha256")
+        size = item.get("size")
+        if (
+            not isinstance(digest, str)
+            or len(digest) != 64
+            or not isinstance(size, int)
+            or isinstance(size, bool)
+            or destination.is_symlink()
+            or not destination.is_file()
+            or destination.stat().st_size != size
+            or _digest(destination) != digest
+        ):
+            raise ValueError(f"Steam migration receipt 目标内容漂移: {destination}")
+        if destination.name.endswith(".sqlite3"):
+            if item.get("sqlite_integrity") != "ok":
+                raise ValueError(f"Steam migration receipt 缺少 SQLite integrity: {path}")
+            _sqlite_integrity(destination)
+    return True
+
+
+def migrate_v2_data(*, workspace: Path, marketplace: str) -> Path:
+    """持有 workspace 独占锁迁移 Steam 数据并写最终 receipt。"""
+
+    workspace = workspace.expanduser().resolve()
+    lock = WorkspaceInstanceLock(workspace)
+    lock.acquire()
+    try:
+        return _migrate_locked(workspace=workspace, marketplace=marketplace)
+    finally:
+        lock.release()
+
+
+def _migrate_locked(*, workspace: Path, marketplace: str) -> Path:
+    """在独占区间准备、校验并发布一次迁移。"""
+
+    if not marketplace or not marketplace.replace("-", "").isalnum():
+        raise ValueError(f"Steam marketplace 无效: {marketplace!r}")
+    mcp_root = workspace / "mcp"
+    source = mcp_root / "steam-mcp"
+    if (
+        mcp_root.is_symlink()
+        or source.is_symlink()
+        or not source.is_dir()
+        or not source.resolve().is_relative_to(workspace)
+    ):
+        raise FileNotFoundError(f"Steam v2 数据目录不存在或不安全: {source}")
+    target = workspace / "plugin-data" / f"steam-{marketplace}"
+    validate_workspace_plugin_data_path(target, workspace)
+    _remove_crash_staging(workspace)
+    receipt_path = target / _RECEIPT
+    if _valid_receipt(receipt_path, target=target, marketplace=marketplace):
+        return receipt_path
+
+    staging = workspace / "plugin-data" / f".steam-v2-migrate-{uuid.uuid4().hex}"
+    created_target = not target.exists()
+    ensure_workspace_plugin_data_dir(staging, workspace)
+    try:
+        entries = _stage(source, staging)
+        ensure_workspace_plugin_data_dir(target, workspace)
+        _validate_targets(target, entries)
+        receipt: dict[str, object] = {
+            "schema_version": 1,
+            "source": "mcp/steam-mcp",
+            "target": f"plugin-data/steam-{marketplace}",
+            "recovery": {"kind": "retained_source", "path": "mcp/steam-mcp"},
+            "files": entries,
+        }
+        _publish(staging, target, entries, receipt)
+    finally:
+        shutil.rmtree(staging, ignore_errors=True)
+        if created_target and target.is_dir() and not any(target.iterdir()):
+            target.rmdir()
+    return receipt_path
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--workspace", type=Path, required=True)
+    parser.add_argument("--marketplace", required=True)
+    args = parser.parse_args()
+    print(migrate_v2_data(workspace=args.workspace, marketplace=args.marketplace))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py
new file mode 100644
index 0000000..05d6c0f
--- /dev/null
+++ b/tests/test_manager_integration.py
@@ -0,0 +1,250 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+from agent.plugins.artifacts import ArtifactPointer, write_pointers
+from agent.plugins.generation_activity_host import ActivityHost
+from agent.plugins.generation_proactive_host import ProactiveActivityAdapter
+from agent.plugins.manifest import write_plugin_manifest
+from agent.plugins.manager import PluginManager
+from bus.event_bus import EventBus
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _stage_installed_plugin(tmp_path: Path) -> Path:
+    """为 installed candidate 测试准备 stable/latest 两个 immutable artifact。"""
+
+    # 1. 复制完整插件 artifact，复用当前测试解释器作为隔离 MCP runtime。
+    plugin_base = tmp_path / "home" / "cache" / "github" / "steam"
+    artifacts = plugin_base / ".artifacts"
+    runtime = Path(sys.executable).parent.parent
+    for label in ("stable", "candidate"):
+        artifact = artifacts / label
+        shutil.copytree(
+            ROOT,
+            artifact,
+            ignore=shutil.ignore_patterns(
+                ".git",
+                ".akashic-core",
+                ".pytest_cache",
+                "__pycache__",
+                ".venv",
+            ),
+        )
+        (artifact / "mcp" / ".venv").symlink_to(runtime, target_is_directory=True)
+        if label == "candidate":
+            (artifact / ".candidate-marker").write_text("candidate\n", encoding="utf-8")
+
+    # 2. stable 与 latest 指向不同 artifact，显式打开 installed 插件。
+    write_pointers(
+        plugin_base,
+        stable=ArtifactPointer(".artifacts/stable"),
+        latest=ArtifactPointer(".artifacts/stable"),
+    )
+    write_plugin_manifest(
+        {"steam@github": True},
+        plugins_home=tmp_path / "home",
+    )
+    return plugin_base
+
+
+def _stage_plugin(tmp_path: Path) -> Path:
+    """复制可执行插件，并复用当前测试解释器的依赖环境。"""
+
+    source = tmp_path / "plugins" / "steam"
+    source.mkdir(parents=True)
+    for relative in (
+        "plugin.py",
+        "akashic.plugin.toml",
+        "mcp/requirements.txt",
+        "mcp/run_mcp.py",
+        "mcp/runtime_config.py",
+        "mcp/steam_mcp.py",
+        "mcp/steam_proactive.py",
+        "mcp/http_client.py",
+    ):
+        target = source / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(ROOT / relative, target)
+    shutil.copytree(ROOT / "skills", source / "skills")
+    runtime = Path(sys.executable).parent.parent
+    (source / "mcp" / ".venv").symlink_to(runtime, target_is_directory=True)
+    return source
+
+
+@pytest.mark.asyncio
+async def test_manager_boots_formal_steam_without_network_calls_and_drains(
+    tmp_path: Path,
+) -> None:
+    """启动真实 stdio handshake，并证明 formal data 与 runtime 精确回收。"""
+
+    # 1. 只提供测试专用 formal 配置；测试不调用任何 Steam tool
+    plugin_root = _stage_plugin(tmp_path)
+    workspace = tmp_path / "workspace"
+    data_root = workspace / "plugin-data" / "steam-builtin"
+    data_root.mkdir(parents=True)
+    config = data_root / "steam_mcp_config.json"
+    config.write_text(
+        json.dumps(
+            {
+                "steam_api_key": "test-only",
+                "steam_id": "76561198000000000",
+                "snapshot_interval_seconds": 3600,
+            }
+        ),
+        encoding="utf-8",
+    )
+    config_digest = hashlib.sha256(config.read_bytes()).hexdigest()
+
+    # 2. 走真实 Manager/Host formal publication，只观察 tools/list
+    manager = PluginManager(
+        plugin_dirs=[plugin_root.parent],
+        event_bus=EventBus(),
+        tool_registry=None,
+        workspace=workspace,
+        installed_cache_root=tmp_path / "cache",
+    )
+    adapter = ProactiveActivityAdapter(manager.composition_generation_host)
+    activity = ActivityHost((adapter,))
+    manager.bind_activity_host(activity)
+    snapshot = None
+    generation_id = None
+    route = None
+    try:
+        await manager.load_all()
+        snapshot = manager.current_snapshot
+        assert snapshot is not None and snapshot.mcp_server_registry is not None
+        assert tuple(snapshot.mcp_server_registry) == ("steam",)
+        generation = next(iter(snapshot.generations.values()))
+        generation_id = generation.generation_id
+        runtime = manager.composition_generation_host.get(generation_id)
+        assert runtime is not None and runtime.mode == "formal"
+        assert runtime.mcp is not None and runtime.mcp.state == "ready"
+        server = runtime.mcp.server("steam")
+        route = server.route()
+        assert route.mode == "formal"
+        assert "get_steam_context" in route.tool_names
+        assert "take_steam_snapshot" in route.tool_names
+        logs = list(server.logs().stdout) + list(server.logs().stderr)
+        assert not any("CallToolRequest" in line or '"tools/call"' in line for line in logs)
+        assert adapter.source_fetch_invocations == 0
+        assert hashlib.sha256(config.read_bytes()).hexdigest() == config_digest
+    finally:
+        if route is not None:
+            await route.aclose()
+        await manager.terminate_all()
+
+    # 3. terminate 后 exact generation、Activity、Root effects 全部释放
+    assert activity.active is None
+    assert manager.composition_generation_host.get(generation_id) is None
+    assert snapshot is not None and snapshot.composition_root is not None
+    assert snapshot.composition_root.receipt().effects == ()
+    assert snapshot.composition_root.topology_view().listeners == ()
+
+
+@pytest.mark.asyncio
+async def test_manager_candidate_publish_switches_formal_steam_and_cleans_validation(
+    tmp_path: Path,
+) -> None:
+    """验证 installed Steam candidate 能 formalize、promote 并清理隔离资源。"""
+
+    # 1. 准备 stable formal 配置与 latest candidate artifact。
+    plugin_base = _stage_installed_plugin(tmp_path)
+    workspace = tmp_path / "workspace"
+    data_root = workspace / "plugin-data" / "steam-github"
+    data_root.mkdir(parents=True)
+    (data_root / "steam_mcp_config.json").write_text(
+        json.dumps(
+            {
+                "steam_api_key": "test-only",
+                "steam_id": "76561198000000000",
+                "snapshot_interval_seconds": 3600,
+            }
+        ),
+        encoding="utf-8",
+    )
+    manager = PluginManager(
+        plugin_dirs=[tmp_path / "builtin-plugins"],
+        event_bus=EventBus(),
+        tool_registry=None,
+        workspace=workspace,
+        installed_cache_root=plugin_base.parent.parent,
+    )
+    adapter = ProactiveActivityAdapter(manager.composition_generation_host)
+    activity = ActivityHost((adapter,))
+    manager.bind_activity_host(activity)
+    stable_snapshot = None
+    candidate = None
+    try:
+        # 2. 先启动 stable formal runtime，再走 candidate -> latest_ready。
+        await manager.load_all()
+        stable_snapshot = manager.current_snapshot
+        assert stable_snapshot is not None
+        stable_root = stable_snapshot.composition_root
+        assert stable_root is not None
+        assert stable_snapshot.proactive_component_catalog is not None
+        assert (
+            stable_snapshot.proactive_component_catalog.root_instance_token
+            is stable_root.instance_token
+        )
+        write_pointers(
+            plugin_base,
+            stable=ArtifactPointer(".artifacts/stable"),
+            latest=ArtifactPointer(".artifacts/candidate"),
+        )
+        candidate = await manager.prepare_candidate("steam@github")
+        assert candidate is not None and candidate.runtime_snapshot is not None
+        assert candidate.validation_workspace is not None
+        candidate_root = candidate.runtime_snapshot.composition_root
+        assert candidate_root is not None
+        assert candidate.runtime_snapshot.proactive_component_catalog is not None
+        assert (
+            candidate.runtime_snapshot.proactive_component_catalog.root_instance_token
+            is candidate_root.instance_token
+        )
+
+        # 3. publish 重新 formalize 到最终 Root，再 promote 并观察验证目录清理。
+        ready_result = await manager.publish_prepared("steam@github")
+        assert ready_result["publication_state"] == "latest_ready"
+        ready = manager.ready_candidate
+        assert ready is not None
+        assert ready is candidate
+        ready_snapshot = ready.runtime_snapshot
+        assert ready_snapshot is not None and ready_snapshot.composition_root is not None
+        assert ready_snapshot.proactive_component_catalog is not None
+        assert (
+            ready_snapshot.proactive_component_catalog.root_instance_token
+            is ready_snapshot.composition_root.instance_token
+        )
+        validation_root = candidate.validation_workspace.parent
+        assert validation_root.exists()
+
+        promoted = await manager.switch_ready("steam@github")
+        assert promoted["publication_state"] == "promoted"
+        final_snapshot = manager.current_snapshot
+        assert final_snapshot is not None and final_snapshot.composition_root is not None
+        assert final_snapshot.proactive_component_catalog is not None
+        assert (
+            final_snapshot.proactive_component_catalog.root_instance_token
+            is final_snapshot.composition_root.instance_token
+        )
+        assert not validation_root.exists()
+        assert manager.ready_candidate is None
+        assert json.loads((plugin_base / ".pointers.json").read_text()) == {
+            "latest": ".artifacts/candidate",
+            "stable": ".artifacts/candidate",
+        }
+    finally:
+        await manager.terminate_all()
+
+    assert candidate is not None
+    assert activity.active is None
+    assert manager.composition_generation_host.get(candidate.generation_id) is None
diff --git a/tests/test_migrate_v2_data.py b/tests/test_migrate_v2_data.py
new file mode 100644
index 0000000..766094a
--- /dev/null
+++ b/tests/test_migrate_v2_data.py
@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+from pathlib import Path
+
+import pytest
+from bootstrap.workspace_lock import WorkspaceInstanceLock
+from scripts import migrate_v2_data as migration
+
+
+def _legacy_data(workspace: Path) -> Path:
+    source = workspace / "mcp" / "steam-mcp"
+    source.mkdir(parents=True)
+    (source / "steam_mcp_config.json").write_text(
+        json.dumps({"steam_api_key": "secret", "steam_id": "user"}),
+        encoding="utf-8",
+    )
+    (source / "steam_user_cache.json").write_text('{"user": "kept"}\n', encoding="utf-8")
+    with sqlite3.connect(source / "steam_proactive.sqlite3") as database:
+        database.execute("CREATE TABLE snapshots (value TEXT NOT NULL)")
+        database.execute("INSERT INTO snapshots VALUES ('kept')")
+        database.commit()
+    return source
+
+
+def test_migration_preserves_source_and_publishes_verified_receipt(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+
+    receipt_path = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    target = workspace / "plugin-data" / "steam-github"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+
+    assert receipt["recovery"] == {
+        "kind": "retained_source",
+        "path": "mcp/steam-mcp",
+    }
+    assert [item["status"] for item in receipt["files"]] == [
+        "copied",
+        "copied",
+        "source_missing",
+        "copied",
+    ]
+    assert (source / "steam_mcp_config.json").is_file()
+    assert (target / "steam_mcp_config.json").read_bytes() == (
+        source / "steam_mcp_config.json"
+    ).read_bytes()
+    with sqlite3.connect(target / "steam_proactive.sqlite3") as database:
+        assert database.execute("PRAGMA integrity_check").fetchone() == ("ok",)
+        assert database.execute("SELECT value FROM snapshots").fetchone() == ("kept",)
+    assert migration.migrate_v2_data(
+        workspace=workspace,
+        marketplace="github",
+    ) == receipt_path
+
+
+def test_in_process_failure_rolls_back_only_new_files(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace = tmp_path / "workspace"
+    _legacy_data(workspace)
+    original_replace = os.replace
+    calls = 0
+
+    def fail_second_publish(source: Path, destination: Path) -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise OSError("injected publish failure")
+        original_replace(source, destination)
+
+    monkeypatch.setattr(migration.os, "replace", fail_second_publish)
+    with pytest.raises(OSError, match="injected publish failure"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
+
+    assert not (workspace / "plugin-data" / "steam-github").exists()
+    assert list((workspace / "plugin-data").glob(".steam-v2-migrate-*")) == []
+
+
+def test_process_crash_partial_publish_is_reconciled_on_restart(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+    plugin_data = workspace / "plugin-data"
+    target = plugin_data / "steam-github"
+    stale = plugin_data / ".steam-v2-migrate-crashed"
+    target.mkdir(parents=True)
+    stale.mkdir()
+    (target / "steam_mcp_config.json").write_bytes(
+        (source / "steam_mcp_config.json").read_bytes()
+    )
+    (stale / "orphan").write_text("partial", encoding="utf-8")
+
+    receipt_path = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    statuses = {
+        item["name"]: item["status"]
+        for item in json.loads(receipt_path.read_text(encoding="utf-8"))["files"]
+    }
+    assert statuses["steam_mcp_config.json"] == "verified"
+    assert statuses["steam_proactive.sqlite3"] == "copied"
+    assert not stale.exists()
+
+
+def test_conflict_and_busy_workspace_leave_both_trees_unchanged(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+    target = workspace / "plugin-data" / "steam-github"
+    target.mkdir(parents=True)
+    (target / "steam_mcp_config.json").write_text("current\n", encoding="utf-8")
+
+    with pytest.raises(FileExistsError, match="内容不同"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    assert json.loads((source / "steam_mcp_config.json").read_text())["steam_api_key"] == "secret"
+    assert (target / "steam_mcp_config.json").read_text(encoding="utf-8") == "current\n"
+    assert not (target / migration._RECEIPT).exists()
+
+    clean_workspace = tmp_path / "busy-workspace"
+    _legacy_data(clean_workspace)
+    lock = WorkspaceInstanceLock(clean_workspace)
+    lock.acquire()
+    try:
+        with pytest.raises(RuntimeError, match="其他 runtime 占用"):
+            migration.migrate_v2_data(workspace=clean_workspace, marketplace="github")
+    finally:
+        lock.release()
+    assert not (clean_workspace / "plugin-data").exists()
+
+
+def test_symlink_source_and_final_receipt_drift_fail_loud(tmp_path: Path) -> None:
+    outside = tmp_path / "outside"
+    _legacy_data(outside)
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    (workspace / "mcp").symlink_to(outside / "mcp", target_is_directory=True)
+    with pytest.raises(FileNotFoundError, match="不安全"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
+
+    workspace = tmp_path / "workspace-drift"
+    _legacy_data(workspace)
+    receipt = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    config = receipt.parent / "steam_mcp_config.json"
+    config.write_text('{"steam_api_key":"changed","steam_id":"user"}', encoding="utf-8")
+    with pytest.raises(ValueError, match="目标内容漂移"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
new file mode 100644
index 0000000..57be110
--- /dev/null
+++ b/tests/test_plugin.py
@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import inspect
+from pathlib import Path
+
+import plugin
+import pytest
+from agent.plugin_composition import (
+    MCP_SERVERS,
+    PROACTIVE_COMPONENTS,
+    CompositionRoot,
+    PluginProactiveComponents,
+    PluginRuntime,
+)
+from agent.plugin_composition.mcp_slots import (
+    PluginMcpServers,
+    _freeze_plugin_mcp_servers,
+)
+from agent.plugin_composition.proactive import _freeze_plugin_proactive_components
+from agent.plugins.composable import ComposablePlugin
+from agent.plugins.static_manifest import load_static_plugin_manifest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_pure_v3_exports_and_exact_apply() -> None:
+    assert plugin.api_version == 3
+    assert plugin.name == "steam"
+    assert plugin.version == "3.0.0"
+    assert plugin.skill_roots == ("skills",)
+    assert tuple(inspect.signature(plugin.apply).parameters) == ("ctx", "config")
+    assert ComposablePlugin.from_module(plugin).skill_roots == ("skills",)
+    assert not hasattr(plugin, "SteamPlugin")
+
+
+@pytest.mark.asyncio
+async def test_apply_registers_mcp_and_proactive_without_data_writes(
+    tmp_path: Path,
+) -> None:
+    root = CompositionRoot("steam:test")
+    servers = PluginMcpServers(root.instance_token)
+    components = PluginProactiveComponents(root.instance_token)
+    await root.context.provide(MCP_SERVERS, servers)
+    await root.context.provide(PROACTIVE_COMPONENTS, components)
+    data_root = tmp_path / "plugin-data"
+    await root.mount(
+        ComposablePlugin.from_module(plugin),
+        name="steam",
+        runtime=PluginRuntime(
+            plugin_id="steam",
+            plugin_dir=ROOT,
+            data_dir=data_root,
+            workspace=tmp_path / "workspace",
+            config=plugin.SteamConfig(),
+        ),
+    )
+
+    server = _freeze_plugin_mcp_servers(
+        servers,
+        root.instance_token,
+    )["steam"].definition
+    source = _freeze_plugin_proactive_components(
+        components,
+        root.instance_token,
+        {"steam": "steam:test"},
+    ).source("presence")
+    assert server.command == ("python", "mcp/run_mcp.py")
+    assert server.required_tools == ("get_steam_context",)
+    assert server.candidate_read_only_tools == ("get_steam_context",)
+    assert server.candidate_env == {"STEAM_BACKEND": "recording"}
+    assert source is not None
+    assert source.definition.mcp_server == "steam"
+    assert source.definition.fetch_tool == "get_steam_context"
+    assert not data_root.exists()
+    await root.dispose()
+
+
+def test_static_manifest_matches_module_and_recording_contract() -> None:
+    manifest = load_static_plugin_manifest(ROOT)
+
+    assert manifest.name == plugin.name == "steam"
+    assert manifest.version == plugin.version == "3.0.0"
+    assert manifest.api_version == plugin.api_version == 3
+    assert manifest.requirements == ("mcp/requirements.txt",)
+    assert manifest.exclude_data_paths == (
+        "steam_mcp_config.json",
+        "steam_user_cache.json",
+        "steam_app_cache.json",
+        "steam_proactive.sqlite3",
+        ".steam-v2-migration.json",
+    )
+    server = manifest.mcp_servers[0]
+    assert server.required_tools == ("get_steam_context",)
+    assert server.candidate_read_only_tools == ("get_steam_context",)
+    assert server.candidate_env == (("STEAM_BACKEND", "recording"),)