diff --git a/.github/workflows/plugin-api-v2.yml b/.github/workflows/plugin-api-v2.yml
deleted file mode 100644
index 63d3e66..0000000
--- a/.github/workflows/plugin-api-v2.yml
+++ /dev/null
@@ -1,28 +0,0 @@
-name: plugin-api-v2
-
-on:
-  pull_request:
-  push:
-    branches:
-      - master
-
-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..ff6a040
--- /dev/null
+++ b/.github/workflows/plugin-api-v3.yml
@@ -0,0 +1,69 @@
+name: plugin-api-v3
+
+on:
+  pull_request:
+  push:
+    branches:
+      - master
+  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 Feed runtime
+        run: |
+          python -m venv mcp/.venv
+          mcp/.venv/bin/python -m pip install \
+            -r .akashic-core/requirements.txt \
+            -r mcp/requirements.txt \
+            -r .akashic-core/requirements-dev.txt \
+            pytest pytest-asyncio
+      - name: Run Feed unit tests
+        env:
+          AKASHIC_AGENT_ROOT: .akashic-core
+          PYTHONPATH: .akashic-core:mcp/.venv/lib/python3.13/site-packages
+        run: mcp/.venv/bin/python -m pytest -q mcp/tests tests
+      - name: Run pyright
+        env:
+          AKASHIC_AGENT_ROOT: .akashic-core
+          PYTHONPATH: .akashic-core:mcp/.venv/lib/python3.13/site-packages
+        run: mcp/.venv/bin/pyright plugin.py mcp/run_mcp.py mcp/src mcp/scripts scripts
+      - name: Compile Python sources
+        run: python -m compileall -q plugin.py mcp/run_mcp.py mcp/src mcp/scripts scripts tests
+      - name: Check diff formatting
+        run: git diff --check
diff --git a/README.md b/README.md
index 7e79d92..7d25fca 100644
--- a/README.md
+++ b/README.md
@@ -1,79 +1,71 @@
 # feed-mcp
 
-`feed-mcp` 是一个 Aka 插件试点仓库，打包了三类能力：
+Feed 是一个 Akashic Plugin API v3 插件，提供 RSS 订阅管理、Feed MCP 和
+`subscriptions` 主动内容源，同时保留 `skills/` 下的 Feed 技能。
 
-- `lifecycle`: 最小 `FeedPlugin`
-- `skills`: `feed-manage` 与 `rsshub-route-finder`
-- `mcp`: feed 订阅查询、缓存自刷新与主动事件读取
-
-查询语义（1.3.0 起）：`feed_query` 只读缓存，不主动触发拉取；缓存 freshness 由 MCP lifespan 的后台 `FeedPoller` 按 `poll_ttl_seconds`（默认 300s）周期刷新，或通过 `poll_feeds` 显式触发。
+`feed_query` 只读缓存，不主动触发拉取；缓存 freshness 由 MCP lifespan 的后台
+`FeedPoller` 按 `poll_ttl_seconds`（默认 300s）周期刷新，或通过 `poll_feeds`
+显式触发。
 
 目录结构：
 
 ```text
 feed-mcp
+├─ akashic.plugin.toml
 ├─ plugin.py
+├─ scripts/migrate_v2_data.py
 ├─ skills/
-│  ├─ feed-manage/
-│  └─ rsshub-route-finder/
 └─ mcp/
    ├─ run_mcp.py
    └─ src/
 ```
 
-本仓库用于验证：
-
-- `plugin.py` 程序化声明生命周期、skills、MCP 与主动信息源
-- `~/.akashic-plugin/cache` 下的 installed plugin 装载
-- skill 软链接
-- 插件程序化 MCP 注册
-
-运行时目录：
+`plugin.py` 只执行 `apply(ctx, config)` 声明，不启动进程、不访问网络、不读写
+插件数据。Core 从静态 manifest 准备 MCP runtime，并按配置注册
+`subscriptions` 主动事件源；`skill_roots = ("skills",)` 保持原有技能装载路径。
 
-```text
-~/.akashic-plugin
-├─ cache/
-│  └─ <marketplace>/feed/<version>/
-│     ├─ plugin.py
-│     ├─ skills/
-│     └─ mcp/
-└─ data/
-   └─ feed-<marketplace>/
-      ├─ feed_mcp.sqlite3
-      ├─ source_scores.json
-      ├─ feed_cache.db
-      ├─ feed_mcp.runtime.log
-      ├─ feed_mcp.runtime.log.1
-      ├─ feed_mcp.runtime.log.2
-      └─ feed_mcp.runtime.log.3
-```
+正式运行数据位于 Core 分配的 `plugin-data/feed-<marketplace>/`：
 
-边界约定：
+- `feed_mcp.sqlite3`：订阅、条目、确认和轮询状态
+- `source_scores.json`、`feed_cache.db`：v2 历史运行数据（如存在）
+- 运行日志只通过 MCP stderr 输出，不创建 `feed_mcp.runtime.log`
 
-- `cache` 只放代码包与依赖环境，可被新版本替换
-- `data` 只放运行时状态与历史数据，升级时保留
-- 仓库本身不提交 sqlite、日志、运行态缓存
-- 运行日志按 5MB 轮转，最多保留 3 个历史文件
+候选验证使用 `FEED_BACKEND=recording`：
 
-当前 feed 的持久化方式：
+- `get_proactive_events` 固定返回 `{"status":"empty"}`
+- 不启动 `FeedPoller`，不访问 RSS/RSSHub、不连接 SQLite
+- `acknowledge_events` 在 recording 后端 fail-loud
+- candidate 只开放只读的 `get_proactive_events`
 
-- 新增/取消订阅通过 `feed_manage` 直接读写 sqlite `sources`
-- 历史内容保存在 sqlite `items`
-- 主动推送确认状态保存在 sqlite `acked_items`
-- 轮询状态保存在 sqlite `poll_state`
+正式主动端口使用明确的 typed 结果：拉取返回 `empty` 或 `items`，确认只有全部
+请求 ID 持久成功时才返回 `committed`；异常和部分确认返回 `failure`，不会伪装
+为成功。
 
-缓存 freshness：
+## 从 v2 迁移
 
-- MCP 进程通过 FastMCP lifespan 启动唯一后台 poller
-- 启动后立即刷新一次；首次主动事件读取会等待该刷新完成
-- 后续按 `feed_mcp.json` 的 `poll_ttl_seconds` 定时刷新
-- `get_proactive_events` 只读取稳定缓存，不承担网络抓取
-- SQLite 使用 WAL 和 busy timeout，轮询写入不会阻塞缓存快照读取
+先停止占用 workspace 的 Akashic runtime，再运行：
 
-首次迁移行为：
+```bash
+PYTHONPATH=/path/to/akashic-agent \
+python scripts/migrate_v2_data.py \
+  --workspace /path/to/workspace \
+  --marketplace github
+```
 
-- 插件首次启动时，如果 `$AKA_PLUGIN_DATA_DIR/feed_mcp.sqlite3` 不存在
-- 会尝试从旧目录复制历史数据
-  - `$AKASHIC_WORKSPACE/mcp/feed-mcp/`
-  - `$AKASHIC_WORKSPACE/backups/feed-plugin-migration-*/feed-mcp/`
-- 迁移的是运行态数据，不是把数据打包进仓库
+迁移脚本持有 workspace 独占锁，按 `mcp/feed-mcp/` primary、再按
+`backups/feed-plugin-migration-*/feed-mcp/` 最新备份顺序选择第一个含数据的源，
+并保留源目录。`feed_mcp.sqlite3` 和其他 SQLite 数据使用在线 backup 后执行
+integrity check；目标存在不同内容时直接失败。
+
+最终 receipt 写入
+`plugin-data/feed-<marketplace>/.feed-v2-migration.json`，逐文件记录
+`source_missing`、`target_only`、`verified` 或 `copied`、源路径、SHA-256、大小和
+SQLite integrity。进程内发布失败会回滚本次新增文件；进程崩溃后重跑会清理残留
+staging、核对同内容目标并完成发布。源数据保留作为 recovery source；receipt
+不属于候选验证输入。
+
+完整外网 RSS E2E 不属于本插件工作流。v3 workflow 固定 Core
+`78e50d4dfb3f4348fff37d55d9c9bdd0e002164d` 与 contracts
+`4dd69dd621e029e51e99aa428443fa3a4ec1f6cf`，执行插件单元测试、pyright、
+`compileall` 和 `git diff --check`，并以空订阅库走真实 Manager、stdio MCP、
+committed proactive source lease 与 terminate cleanup。
diff --git a/akashic.plugin.toml b/akashic.plugin.toml
new file mode 100644
index 0000000..e18dcbd
--- /dev/null
+++ b/akashic.plugin.toml
@@ -0,0 +1,27 @@
+schema_version = 1
+name = "feed"
+version = "3.0.0"
+api_version = 3
+entrypoint = "plugin.py"
+
+[[python]]
+requirements = "mcp/requirements.txt"
+
+[validation]
+exclude_data_paths = [
+  "feed_mcp.sqlite3",
+  "feed_mcp.sqlite3-wal",
+  "feed_mcp.sqlite3-shm",
+  "source_scores.json",
+  "feed_cache.db",
+  "feed_cache.db-wal",
+  "feed_cache.db-shm",
+  ".feed-v2-migration.json",
+]
+
+[[mcp]]
+name = "feed"
+command = ["python", "mcp/run_mcp.py"]
+required_tools = ["get_proactive_events", "acknowledge_events"]
+candidate_read_only_tools = ["get_proactive_events"]
+candidate_env = {FEED_BACKEND = "recording"}
diff --git a/mcp/run_mcp.py b/mcp/run_mcp.py
index 96e9145..3a2a6c7 100755
--- a/mcp/run_mcp.py
+++ b/mcp/run_mcp.py
@@ -1,36 +1,27 @@
 #!/usr/bin/env python3
+from __future__ import annotations
+
 import logging
 import os
 import sys
-from logging.handlers import RotatingFileHandler
 from pathlib import Path
 
 
 def _runtime_dir() -> Path:
+    """返回 Core 注入的数据目录，但不创建或读取其中的文件。"""
+
     raw = os.environ.get("AKA_PLUGIN_DATA_DIR", "").strip()
     if not raw:
         raise RuntimeError("feed MCP 缺少 AKA_PLUGIN_DATA_DIR")
-    path = Path(raw).expanduser()
-    path.mkdir(parents=True, exist_ok=True)
-    return path
+    return Path(raw).expanduser()
 
 
-def _setup_logging(runtime_dir: Path) -> None:
-    runtime_log = runtime_dir / "feed_mcp.runtime.log"
-    runtime_log.parent.mkdir(parents=True, exist_ok=True)
+def _setup_logging() -> None:
+    """把 MCP 日志绑定到 stderr，不创建运行日志文件。"""
 
     formatter = logging.Formatter(
         "%(asctime)s %(levelname)-8s %(name)s | %(message)s"
     )
-    file_handler = RotatingFileHandler(
-        runtime_log,
-        maxBytes=5 * 1024 * 1024,
-        backupCount=3,
-        encoding="utf-8",
-    )
-    file_handler.setLevel(logging.INFO)
-    file_handler.setFormatter(formatter)
-
     stream_handler = logging.StreamHandler(sys.stderr)
     stream_handler.setLevel(logging.INFO)
     stream_handler.setFormatter(formatter)
@@ -38,25 +29,24 @@ def _setup_logging(runtime_dir: Path) -> None:
     root = logging.getLogger()
     root.setLevel(logging.INFO)
     root.handlers.clear()
-    root.addHandler(file_handler)
     root.addHandler(stream_handler)
 
 
 def main() -> None:
-    # 1. 切换到脚本目录，保证相对路径（sqlite/json）稳定。
+    # 1. 校验 Core 注入的数据目录变量，但不创建或读取运行态文件。
+    _runtime_dir()
+
+    # 2. 切换到脚本目录，保证相对代码路径稳定。
     script_dir = Path(__file__).resolve().parent
     os.chdir(script_dir)
     if str(script_dir) not in sys.path:
         sys.path.insert(0, str(script_dir))
 
-    # 2. 初始化日志（落盘 + stderr）。
-    _setup_logging(_runtime_dir())
-
-    # 3. 启动 MCP stdio 服务。
+    # 3. 启动 MCP stdio 服务；日志只经过 stderr。
+    _setup_logging()
     from src.mcp_bridge import create_mcp_server
 
-    mcp = create_mcp_server()
-    mcp.run(transport="stdio")
+    create_mcp_server().run(transport="stdio")
 
 
 if __name__ == "__main__":
diff --git a/mcp/src/feed_backend.py b/mcp/src/feed_backend.py
index 60f839e..c94fbf1 100644
--- a/mcp/src/feed_backend.py
+++ b/mcp/src/feed_backend.py
@@ -100,8 +100,8 @@ def _connect(cfg: FeedMcpConfig) -> sqlite3.Connection:
     cfg.db_path.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(cfg.db_path, timeout=30)
     conn.row_factory = sqlite3.Row
-    conn.execute("PRAGMA journal_mode=WAL")
     conn.execute("PRAGMA busy_timeout=30000")
+    conn.execute("PRAGMA journal_mode=WAL")
     conn.execute(
         """
         CREATE TABLE IF NOT EXISTS sources (
@@ -1022,75 +1022,6 @@ def feed_manage(action: str, name: str = "", url: str = "", source_type: str = "
         conn.close()
 
 
-def sync_legacy_subscriptions(json_path: str) -> dict[str, int]:
-    cfg = load_config()
-    conn = _connect(cfg)
-    inserted = 0
-    updated = 0
-    try:
-        # 1. 从旧 feeds.json 读取订阅列表，兼容历史本地 feed 配置。
-        path = Path(json_path).expanduser()
-        if not path.exists():
-            return {"inserted": 0, "updated": 0}
-        raw = json.loads(path.read_text(encoding="utf-8"))
-        if not isinstance(raw, list):
-            return {"inserted": 0, "updated": 0}
-        # 2. 按 URL 对齐到 feed-mcp 的 sources 表，避免迁移时重复插入。
-        for item in raw:
-            if not isinstance(item, dict):
-                continue
-            name = str(item.get("name") or "").strip()
-            url = _normalize_source_url(str(item.get("url") or "").strip())
-            if not name or not url:
-                continue
-            row = conn.execute(
-                "SELECT id FROM sources WHERE url = ? LIMIT 1",
-                (url,),
-            ).fetchone()
-            payload = (
-                str(item.get("type") or "rss"),
-                name,
-                url,
-                str(item.get("note") or "").strip() or None,
-                1 if bool(item.get("enabled", True)) else 0,
-                cfg.poll_ttl_seconds,
-                str(item.get("added_at") or _now().isoformat()),
-                _now().isoformat(),
-            )
-            if row is None:
-                source_id = str(item.get("id") or uuid.uuid4())
-                conn.execute(
-                    """
-                    INSERT INTO sources (id, type, name, url, note, enabled, poll_interval_seconds, added_at, updated_at)
-                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-                    """,
-                    (source_id, *payload),
-                )
-                inserted += 1
-                continue
-            conn.execute(
-                """
-                UPDATE sources
-                SET type = ?, name = ?, note = ?, enabled = ?, poll_interval_seconds = ?, updated_at = ?
-                WHERE id = ?
-                """,
-                (
-                    payload[0],
-                    payload[1],
-                    payload[3],
-                    payload[4],
-                    payload[5],
-                    payload[7],
-                    str(row["id"]),
-                ),
-            )
-            updated += 1
-        conn.commit()
-        return {"inserted": inserted, "updated": updated}
-    finally:
-        conn.close()
-
-
 def feed_query(
     action: str,
     source: str = "",
diff --git a/mcp/src/mcp_bridge.py b/mcp/src/mcp_bridge.py
index d2434cc..61e52a7 100644
--- a/mcp/src/mcp_bridge.py
+++ b/mcp/src/mcp_bridge.py
@@ -1,22 +1,31 @@
 from __future__ import annotations
 
-# pyright: reportMissingImports=false
-
 import asyncio
 import json
 import logging
+import os
 from contextlib import asynccontextmanager
-from typing import AsyncIterator, List, Optional
+from typing import Any, AsyncIterator
 
 from mcp.server.fastmcp import FastMCP
 
-from src import feed_backend
-
 logger = logging.getLogger(__name__)
 
 
+def _recording_backend() -> bool:
+    return os.environ.get("FEED_BACKEND", "").strip().lower() == "recording"
+
+
+def _live_backend() -> Any:
+    if _recording_backend():
+        raise RuntimeError("feed recording backend 不允许访问正式 Feed 后端")
+    from src import feed_backend
+
+    return feed_backend
+
+
 class FeedPoller:
-    """在后台维护 Feed 缓存刷新，不阻塞缓存读取。"""
+    """在正式运行时后台维护 Feed 缓存刷新。"""
 
     def __init__(self) -> None:
         self._lock = asyncio.Lock()
@@ -32,27 +41,60 @@ async def stop(self) -> None:
         self._stop.set()
         if self._task is None:
             return
-        self._task.cancel()
-        await asyncio.gather(self._task, return_exceptions=True)
+        task = self._task
+        task.cancel()
+        caller_cancelled = False
+        while not task.done():
+            try:
+                await asyncio.shield(task)
+            except asyncio.CancelledError:
+                if not task.done():
+                    caller_cancelled = True
+            except Exception:
+                break
         self._task = None
+        if not task.cancelled():
+            task.result()
+        if caller_cancelled:
+            raise asyncio.CancelledError
 
     async def poll_now(self) -> None:
         async with self._lock:
-            await asyncio.to_thread(feed_backend.poll_feeds_only)
+            worker = asyncio.create_task(
+                asyncio.to_thread(_live_backend().poll_feeds_only),
+                name="feed-cache-poll-worker",
+            )
+            try:
+                await asyncio.shield(worker)
+            except asyncio.CancelledError:
+                while not worker.done():
+                    try:
+                        await asyncio.shield(worker)
+                    except asyncio.CancelledError:
+                        continue
+                    except Exception:
+                        break
+                error = worker.exception()
+                if error is not None:
+                    logger.error(
+                        "[feed] poll worker 在取消收束期间失败",
+                        exc_info=(type(error), error, error.__traceback__),
+                    )
+                raise
 
     async def _run(self) -> None:
         """首次立即刷新，随后按缓存 TTL 持续刷新。"""
 
-        # 1. 首次刷新在后台执行，读取方始终可以使用现有缓存
+        # 1. 首次刷新失败必须暴露，同时保留后续重试能力。
         try:
             await self.poll_now()
         except Exception:
             logger.exception("[feed] 首次缓存刷新失败")
 
-        # 2. 后续失败显式记录，并保留下一轮重试能力
+        # 2. 正式后端按配置周期刷新，recording 不会进入此生命周期。
         while not self._stop.is_set():
             try:
-                interval = feed_backend.load_config().poll_ttl_seconds
+                interval = _live_backend().load_config().poll_ttl_seconds
             except Exception:
                 logger.exception("[feed] 读取轮询配置失败")
                 interval = 60
@@ -68,10 +110,15 @@ async def _run(self) -> None:
 
 
 def create_mcp_server() -> FastMCP:
-    poller = FeedPoller()
+    """创建 Feed MCP，并让 recording 生命周期保持零轮询、零数据库访问。"""
+
+    poller = None if _recording_backend() else FeedPoller()
 
     @asynccontextmanager
     async def lifespan(_: FastMCP) -> AsyncIterator[None]:
+        if poller is None:
+            yield None
+            return
         await poller.start()
         try:
             yield None
@@ -88,11 +135,9 @@ def feed_manage(
         source_type: str = "rss",
         note: str = "",
     ) -> str:
-        """管理 RSS 订阅源：添加、删除、列出订阅。支持 rss add / 添加订阅 / 订阅管理 / 取消订阅。
+        """管理 RSS 订阅源：添加、删除、列出订阅。"""
 
-        action: list（列出所有订阅）/ add（添加新订阅）/ remove（删除订阅）
-        """
-        return feed_backend.feed_manage(
+        return _live_backend().feed_manage(
             action=action,
             name=name,
             url=url,
@@ -109,11 +154,9 @@ def feed_query(
         page: int = 1,
         page_size: int = 20,
     ) -> str:
-        """查询 RSS 订阅内容，获取最近新闻、最新文章、最新资讯、rss查询。
+        """查询 RSS 订阅内容。"""
 
-        action: latest（最近内容）/ search（关键词搜索）/ sources（列出信息来源）
-        """
-        return feed_backend.feed_query(
+        return _live_backend().feed_query(
             action=action,
             source=source,
             keyword=keyword,
@@ -124,30 +167,107 @@ def feed_query(
 
     @mcp.tool()
     async def poll_feeds() -> str:
-        try:
-            await poller.poll_now()
-            return "ok"
-        except Exception as e:
-            logger.exception("[feed] poll_feeds 系统级失败")
-            return f"error: {e}"
+        if poller is None:
+            raise RuntimeError("feed recording backend 不允许轮询")
+        await poller.poll_now()
+        return "ok"
 
     @mcp.tool()
-    async def get_proactive_events(offset: int = 0, limit: int = 50) -> str:
+    async def get_proactive_events(
+        offset: int = 0,
+        limit: int = 50,
+        cursor: str | None = None,
+    ) -> str:
         events = await asyncio.to_thread(
-            feed_backend.get_proactive_events,
+            _fetch_proactive_events,
             offset=offset,
             limit=limit,
+            cursor=cursor,
         )
-        return json.dumps(
-            events,
-            ensure_ascii=False,
-        )
+        return json.dumps(events, ensure_ascii=False)
 
     @mcp.tool()
-    def acknowledge_events(event_ids: List[str], feedback: Optional[str] = None) -> str:
+    def acknowledge_events(
+        event_ids: list[str], feedback: str | None = None
+    ) -> str:
         return json.dumps(
-            feed_backend.acknowledge_events(event_ids, feedback=feedback),
+            _acknowledge_proactive_events(event_ids, feedback=feedback),
             ensure_ascii=False,
         )
 
     return mcp
+
+
+def _proactive_fetch_payload(
+    events: list[dict[str, Any]],
+    *,
+    cursor: str | None = None,
+) -> dict[str, Any]:
+    """把正式后端结果编码成 Core 可识别的 typed empty/items。"""
+
+    if not events:
+        return {"status": "empty"}
+    payload: dict[str, Any] = {"status": "items", "items": events}
+    if cursor is not None:
+        payload["cursor"] = cursor
+    return payload
+
+
+def _fetch_proactive_events(
+    *,
+    offset: int = 0,
+    limit: int = 50,
+    cursor: str | None = None,
+) -> dict[str, Any]:
+    """recording 固定返回 typed empty，正式运行才读取 Feed 数据库。"""
+
+    if _recording_backend():
+        return {"status": "empty"}
+    if limit < 1:
+        raise ValueError("Feed proactive limit 必须大于零")
+    if cursor is not None:
+        if offset != 0:
+            raise ValueError("Feed proactive cursor 不能与 offset 同时使用")
+        prefix = "feed-offset:"
+        if not cursor.startswith(prefix) or not cursor[len(prefix) :].isdigit():
+            raise ValueError("Feed proactive cursor 无效")
+        offset = int(cursor[len(prefix) :])
+    backend = _live_backend()
+    events = backend.get_proactive_events(offset=offset, limit=limit + 1)
+    has_more = len(events) > limit
+    return _proactive_fetch_payload(
+        events[:limit],
+        cursor=f"feed-offset:{offset + limit}" if has_more else None,
+    )
+
+
+def _acknowledge_proactive_events(
+    requested: list[str], *, feedback: str | None = None
+) -> dict[str, Any]:
+    """只有全部请求 ID 持久确认后才编码 committed。"""
+
+    if not requested:
+        return {"status": "skipped", "reason": "no_ids"}
+    if _recording_backend():
+        raise RuntimeError("feed recording backend 不允许确认事件")
+    result = _live_backend().acknowledge_events(requested, feedback=feedback)
+    return _proactive_ack_payload(requested, result)
+
+
+def _proactive_ack_payload(
+    requested: list[str], result: dict[str, list[str]]
+) -> dict[str, Any]:
+    """把 Feed ack 结果转换为完整 committed 或明确 failure。"""
+
+    if not requested:
+        return {"status": "skipped", "reason": "no_ids"}
+    acknowledged = list(result.get("acknowledged", []))
+    failed = list(result.get("failed", []))
+    if failed or acknowledged != requested:
+        return {
+            "status": "failure",
+            "error": "feed ack 未完整提交",
+            "retryable": True,
+            "failed_ids": failed,
+        }
+    return {"status": "committed", "ids": acknowledged}
diff --git a/mcp/tests/conftest.py b/mcp/tests/conftest.py
index 57f5082..a35df14 100644
--- a/mcp/tests/conftest.py
+++ b/mcp/tests/conftest.py
@@ -11,9 +11,11 @@
     os.environ.get("AKASHIC_AGENT_ROOT", "").strip()
     or repo_root.parents[1] / "akasic-agent"
 )
-for path in (repo_root, repo_root / "mcp", agent_root):
+for path in (repo_root, agent_root):
     if str(path) not in sys.path:
         sys.path.insert(0, str(path))
+if str(repo_root / "mcp") not in sys.path:
+    sys.path.append(str(repo_root / "mcp"))
 
 _test_data_dir = tempfile.TemporaryDirectory(prefix="feed-plugin-tests-")
 if not os.environ.get("AKA_PLUGIN_DATA_DIR", "").strip():
diff --git a/mcp/tests/test_poll_lifecycle.py b/mcp/tests/test_poll_lifecycle.py
index e95a516..ca27dd1 100644
--- a/mcp/tests/test_poll_lifecycle.py
+++ b/mcp/tests/test_poll_lifecycle.py
@@ -1,24 +1,22 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 from types import SimpleNamespace
 
+import pytest
+
 from src import mcp_bridge
 
 
 def test_poller_refreshes_immediately_and_continues(monkeypatch) -> None:
     calls: list[int] = []
 
-    monkeypatch.setattr(
-        mcp_bridge.feed_backend,
-        "poll_feeds_only",
-        lambda: calls.append(len(calls) + 1),
-    )
-    monkeypatch.setattr(
-        mcp_bridge.feed_backend,
-        "load_config",
-        lambda: SimpleNamespace(poll_ttl_seconds=0.01),
+    backend = SimpleNamespace(
+        poll_feeds_only=lambda: calls.append(len(calls) + 1),
+        load_config=lambda: SimpleNamespace(poll_ttl_seconds=0.01),
     )
+    monkeypatch.setattr(mcp_bridge, "_live_backend", lambda: backend)
 
     async def scenario() -> None:
         poller = mcp_bridge.FeedPoller()
@@ -49,12 +47,11 @@ def poll() -> None:
         if attempts == 1:
             raise OSError("feed database unavailable")
 
-    monkeypatch.setattr(mcp_bridge.feed_backend, "poll_feeds_only", poll)
-    monkeypatch.setattr(
-        mcp_bridge.feed_backend,
-        "load_config",
-        lambda: SimpleNamespace(poll_ttl_seconds=0.01),
+    backend = SimpleNamespace(
+        poll_feeds_only=poll,
+        load_config=lambda: SimpleNamespace(poll_ttl_seconds=0.01),
     )
+    monkeypatch.setattr(mcp_bridge, "_live_backend", lambda: backend)
 
     async def scenario() -> None:
         poller = mcp_bridge.FeedPoller()
@@ -69,3 +66,67 @@ async def scenario() -> None:
             await poller.stop()
 
     asyncio.run(scenario())
+
+
+def test_poller_stop_waits_for_inflight_thread(monkeypatch) -> None:
+    entered = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def poll() -> None:
+        entered.set()
+        release.wait(timeout=5)
+        finished.set()
+
+    backend = SimpleNamespace(
+        poll_feeds_only=poll,
+        load_config=lambda: SimpleNamespace(poll_ttl_seconds=60),
+    )
+    monkeypatch.setattr(mcp_bridge, "_live_backend", lambda: backend)
+
+    async def scenario() -> None:
+        poller = mcp_bridge.FeedPoller()
+        await poller.start()
+        await asyncio.to_thread(entered.wait, 5)
+        stop = asyncio.create_task(poller.stop())
+        await asyncio.sleep(0)
+        assert not stop.done()
+        assert not finished.is_set()
+        release.set()
+        await stop
+        assert finished.is_set()
+
+    asyncio.run(scenario())
+
+
+def test_poller_stop_finishes_worker_before_restoring_cancellation(monkeypatch) -> None:
+    entered = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def poll() -> None:
+        entered.set()
+        release.wait(timeout=5)
+        finished.set()
+
+    backend = SimpleNamespace(
+        poll_feeds_only=poll,
+        load_config=lambda: SimpleNamespace(poll_ttl_seconds=60),
+    )
+    monkeypatch.setattr(mcp_bridge, "_live_backend", lambda: backend)
+
+    async def scenario() -> None:
+        poller = mcp_bridge.FeedPoller()
+        await poller.start()
+        await asyncio.to_thread(entered.wait, 5)
+        stop = asyncio.create_task(poller.stop())
+        await asyncio.sleep(0)
+        stop.cancel()
+        await asyncio.sleep(0)
+        assert not stop.done()
+        release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await stop
+        assert finished.is_set()
+
+    asyncio.run(scenario())
diff --git a/mcp/tests/test_runtime_paths.py b/mcp/tests/test_runtime_paths.py
index 72a92a3..5ac411d 100644
--- a/mcp/tests/test_runtime_paths.py
+++ b/mcp/tests/test_runtime_paths.py
@@ -3,11 +3,10 @@
 import sqlite3
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from types import SimpleNamespace
 
 import pytest
 
-from plugin import FeedPlugin
+import plugin
 from run_mcp import _runtime_dir
 from src import feed_backend
 from src.feed_backend import _runtime_root, load_config
@@ -25,15 +24,11 @@ def test_runtime_entrypoints_reject_missing_data_dir(
         load_config()
 
 
-def test_activate_rejects_missing_context_paths(tmp_path: Path) -> None:
-    plugin = FeedPlugin()
-    plugin.context = SimpleNamespace(data_dir=None, workspace=tmp_path)
-    with pytest.raises(RuntimeError, match="数据目录"):
-        plugin.activate()
-
-    plugin.context = SimpleNamespace(data_dir=tmp_path, workspace=None)
-    with pytest.raises(RuntimeError, match="workspace"):
-        plugin.activate()
+def test_v3_module_keeps_skill_root_and_identity_exports() -> None:
+    assert plugin.api_version == 3
+    assert plugin.name == "feed"
+    assert plugin.version == "3.0.0"
+    assert plugin.skill_roots == ("skills",)
 
 
 def test_concurrent_legacy_connections_share_one_schema_migration(
@@ -43,10 +38,11 @@ def test_concurrent_legacy_connections_share_one_schema_migration(
     monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(tmp_path))
     config = feed_backend.load_config()
 
-    # 1. 建立真实旧表，让所有连接都必须走 ADD COLUMN 迁移
+    # 1. 建立真实旧表，让所有连接都必须走 ADD COLUMN 迁移。
     config.db_path.parent.mkdir(parents=True, exist_ok=True)
     with sqlite3.connect(config.db_path) as connection:
-        connection.execute("""
+        connection.execute(
+            """
             CREATE TABLE items (
                 event_id TEXT PRIMARY KEY,
                 source_id TEXT NOT NULL,
@@ -61,9 +57,10 @@ def test_concurrent_legacy_connections_share_one_schema_migration(
                 emitted_at TEXT,
                 content_hash TEXT NOT NULL
             )
-            """)
+            """
+        )
 
-    # 2. 并发模拟 lifespan poller 与首个 MCP 调用同时启动
+    # 2. 并发模拟正式 poller 与首个 MCP 调用同时启动。
     def connect_once() -> set[str]:
         connection = feed_backend._connect(config)
         try:
@@ -77,6 +74,6 @@ def connect_once() -> set[str]:
     with ThreadPoolExecutor(max_workers=8) as executor:
         schemas = list(executor.map(lambda _: connect_once(), range(24)))
 
-    # 3. 每个连接都必须看到同一个完整迁移终态
+    # 3. 每个连接都必须看到同一个完整迁移终态。
     required = {"author", "interest_ok", "interest_scored_at"}
     assert all(required <= schema for schema in schemas)
diff --git a/plugin.py b/plugin.py
index 5dec829..860057d 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 FeedProactiveConfig(BaseModel):
@@ -17,84 +19,43 @@ class FeedConfig(BaseModel):
     proactive: FeedProactiveConfig = Field(default_factory=FeedProactiveConfig)
 
 
-class FeedPlugin(Plugin):
-    api_version = 2
-    name = "feed"
-    version = "1.3.1"
-    desc = "Feed MCP plugin"
-    ConfigModel = FeedConfig
-
-    @classmethod
-    def skill_roots(cls) -> tuple[str, ...]:
-        return ("skills",)
-
-    @classmethod
-    def mcp_servers(cls) -> list[McpServerSpec]:
-        return [
-            McpServerSpec(
-                name="feed",
-                command=("python", "mcp/run_mcp.py"),
-                candidate_read_only_tools=(
-                    "feed_query",
-                    "get_proactive_events",
-                ),
-            )
-        ]
-
-    def proactive_sources(self) -> list[ProactiveSourceSpec]:
-        config = cast(FeedConfig, self.context.config)
-        if not config.proactive.enabled:
-            return []
-        return [
-            ProactiveSourceSpec(
-                id="subscriptions",
+api_version = 3
+name = "feed"
+version = "3.0.0"
+desc = "Feed MCP plugin"
+Config = FeedConfig
+inject = (MCP_SERVERS, PROACTIVE_COMPONENTS)
+skill_roots = ("skills",)
+
+
+async def apply(ctx: Context, config: object) -> None:
+    """注册 Feed MCP 与可选的订阅主动事件源。"""
+
+    if not isinstance(config, FeedConfig):
+        raise TypeError("feed config 必须是 FeedConfig")
+
+    # 1. 只声明 MCP；apply 本身不启动进程、不访问网络或插件数据。
+    await ctx.require(MCP_SERVERS).register(
+        ctx,
+        McpServerDefinition(
+            name="feed",
+            command=("python", "mcp/run_mcp.py"),
+            required_tools=("get_proactive_events", "acknowledge_events"),
+            candidate_read_only_tools=("get_proactive_events",),
+            candidate_env={"FEED_BACKEND": "recording"},
+        ),
+    )
+
+    # 2. 主动能力由用户配置决定是否发布。
+    if config.proactive.enabled:
+        await ctx.require(PROACTIVE_COMPONENTS).register(
+            ctx,
+            ProactiveSourceDefinition(
+                name="subscriptions",
                 channels=("content",),
-                server="feed",
+                mcp_server="feed",
                 fetch_tool="get_proactive_events",
                 ack_tool="acknowledge_events",
                 fetch_page_size=50,
-            )
-        ]
-
-    def activate(self) -> None:
-        data_dir = self.context.data_dir
-        workspace = self.context.workspace
-        if data_dir is None:
-            raise RuntimeError("feed 缺少插件数据目录")
-        if workspace is None:
-            raise RuntimeError("feed 缺少 workspace")
-        data_dir.mkdir(parents=True, exist_ok=True)
-        if (data_dir / "feed_mcp.sqlite3").exists():
-            return
-        for source_dir in _legacy_feed_dirs(workspace):
-            copied = _copy_legacy_state(source_dir, data_dir)
-            if copied:
-                return
-
-
-def _legacy_feed_dirs(workspace: Path) -> list[Path]:
-    result: list[Path] = []
-    primary = workspace / "mcp" / "feed-mcp"
-    if primary.exists():
-        result.append(primary)
-    backups_root = workspace / "backups"
-    if backups_root.exists():
-        result.extend(
-            sorted(
-                backups_root.glob("feed-plugin-migration-*/feed-mcp"),
-                reverse=True,
-            )
+            ),
         )
-    return result
-
-
-def _copy_legacy_state(source_dir: Path, data_dir: Path) -> bool:
-    copied = False
-    for name in ("feed_mcp.sqlite3", "source_scores.json", "feed_cache.db"):
-        source = source_dir / name
-        target = data_dir / name
-        if not source.exists() or target.exists():
-            continue
-        shutil.copy2(source, target)
-        copied = True
-    return copied
diff --git a/pyrightconfig.json b/pyrightconfig.json
new file mode 100644
index 0000000..17e8cd5
--- /dev/null
+++ b/pyrightconfig.json
@@ -0,0 +1,19 @@
+{
+  "include": [
+    "plugin.py",
+    "mcp/run_mcp.py",
+    "mcp/src",
+    "scripts",
+    "tests"
+  ],
+  "exclude": ["**/__pycache__"],
+  "venvPath": "mcp",
+  "venv": ".venv",
+  "executionEnvironments": [
+    {
+      "root": ".",
+      "pythonVersion": "3.13",
+      "extraPaths": ["mcp", ".akashic-core"]
+    }
+  ]
+}
diff --git a/scripts/migrate_v2_data.py b/scripts/migrate_v2_data.py
new file mode 100755
index 0000000..d35695c
--- /dev/null
+++ b/scripts/migrate_v2_data.py
@@ -0,0 +1,419 @@
+#!/usr/bin/env python3
+"""把 Feed 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 typing import Any
+
+from agent.plugins.manifest import (
+    ensure_workspace_plugin_data_dir,
+    validate_workspace_plugin_data_path,
+)
+from bootstrap.workspace_lock import WorkspaceInstanceLock
+
+
+_DATA_FILES = (
+    "feed_mcp.sqlite3",
+    "source_scores.json",
+    "feed_cache.db",
+)
+_RECEIPT = ".feed-v2-migration.json"
+_BACKUP_GLOB = "feed-plugin-migration-*/feed-mcp"
+_TARGET_STATUSES = {"target_only", "verified", "copied"}
+
+
+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 integrity receipt，不修改数据库。"""
+
+    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"Feed SQLite 完整性检查失败: {path} ({result})")
+    return "ok"
+
+
+def _is_sqlite_file(path: Path) -> bool:
+    """按文件名和 SQLite magic 判断是否必须执行数据库迁移。"""
+
+    if path.name == "feed_mcp.sqlite3":
+        return True
+    with path.open("rb") as stream:
+        return stream.read(16) == b"SQLite format 3\x00"
+
+
+def _copy_sqlite(source: Path, destination: Path) -> str:
+    """用 SQLite 在线备份生成一致副本，并校验源和目标。"""
+
+    source_integrity = _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()
+    target_integrity = _sqlite_integrity(destination)
+    if source_integrity != target_integrity:
+        raise sqlite3.DatabaseError(
+            f"Feed SQLite integrity receipt 不一致: {source} -> {destination}"
+        )
+    return target_integrity
+
+
+def _source_candidates(workspace: Path) -> tuple[tuple[Path, str], ...]:
+    """按 primary、最新 backup 顺序返回安全的 v2 数据目录候选。"""
+
+    root = workspace.resolve()
+    primary_root = workspace / "mcp"
+    backups_root = workspace / "backups"
+    candidates: list[tuple[Path, str]] = []
+
+    # 1. primary 或 backup 根穿过符号链接都属于不可信迁移输入。
+    if primary_root.is_symlink():
+        raise FileNotFoundError(f"Feed v2 数据目录不安全: {primary_root}")
+    primary = primary_root / "feed-mcp"
+    if primary.exists() or primary.is_symlink():
+        if primary.is_symlink() or not primary.is_dir():
+            raise FileNotFoundError(f"Feed v2 数据目录不存在或不安全: {primary}")
+        if not primary.resolve().is_relative_to(root):
+            raise FileNotFoundError(f"Feed v2 数据目录越界: {primary}")
+        candidates.append((primary, "mcp/feed-mcp"))
+
+    if backups_root.is_symlink():
+        raise FileNotFoundError(f"Feed v2 备份目录不安全: {backups_root}")
+    if backups_root.is_dir():
+        backups = sorted(
+            backups_root.glob(_BACKUP_GLOB),
+            key=lambda item: item.parent.name,
+            reverse=True,
+        )
+        for backup in backups:
+            if backup.is_symlink() or not backup.is_dir():
+                raise FileNotFoundError(f"Feed v2 备份目录不存在或不安全: {backup}")
+            if not backup.resolve().is_relative_to(root):
+                raise FileNotFoundError(f"Feed v2 备份目录越界: {backup}")
+            relative = backup.relative_to(root).as_posix()
+            candidates.append((backup, relative))
+    return tuple(candidates)
+
+
+def _has_data_file(source: Path) -> bool:
+    return any(
+        (source / name).exists() or (source / name).is_symlink()
+        for name in _DATA_FILES
+    )
+
+
+def _select_source(workspace: Path) -> tuple[Path, str]:
+    """选择第一个含有 v2 文件的 primary 或最新 backup。"""
+
+    for source, relative in _source_candidates(workspace):
+        if _has_data_file(source):
+            return source, relative
+    raise FileNotFoundError(
+        "Feed v2 primary 与 backups 都没有可迁移文件"
+    )
+
+
+def _stage_files(
+    source: Path,
+    source_relative: str,
+    staging: Path,
+) -> tuple[dict[str, object], ...]:
+    """复制选定 v2 源的全部数据文件并返回内容 receipt。"""
+
+    entries: list[dict[str, object]] = []
+    for name in _DATA_FILES:
+        source_file = source / name
+        source_name = f"{source_relative}/{name}"
+        if not source_file.exists() and not source_file.is_symlink():
+            entries.append(
+                {
+                    "name": name,
+                    "status": "source_missing",
+                    "source": source_name,
+                }
+            )
+            continue
+        if source_file.is_symlink() or not source_file.is_file():
+            raise ValueError(f"Feed v2 数据不是普通文件: {source_file}")
+        integrity: str | None = None
+        if _is_sqlite_file(source_file):
+            integrity = _copy_sqlite(source_file, staging / name)
+        else:
+            shutil.copy2(source_file, staging / name)
+        entry: dict[str, object] = {
+            "name": name,
+            "status": "staged",
+            "source": source_name,
+            "sha256": _digest(staging / name),
+            "size": (staging / name).stat().st_size,
+        }
+        if integrity is not None:
+            entry["integrity"] = integrity
+            entry["sqlite_integrity"] = integrity
+        entries.append(entry)
+    return tuple(entries)
+
+
+def _record_target_file(entry: dict[str, object], destination: Path) -> None:
+    entry["sha256"] = _digest(destination)
+    entry["size"] = destination.stat().st_size
+    if _is_sqlite_file(destination):
+        integrity = _sqlite_integrity(destination)
+        entry["integrity"] = integrity
+        entry["sqlite_integrity"] = integrity
+
+
+def _validate_targets(target: Path, entries: tuple[dict[str, object], ...]) -> None:
+    """发布前拒绝覆盖不同内容，并收束崩溃留下的同内容文件。"""
+
+    for entry in entries:
+        name = str(entry["name"])
+        destination = target / name
+        if destination.is_symlink():
+            raise ValueError(f"Feed v3 目标不得是符号链接: {destination}")
+        if entry["status"] == "source_missing":
+            if not destination.exists():
+                continue
+            if not destination.is_file():
+                raise FileExistsError(f"Feed v3 目标不是普通文件: {destination}")
+            entry["status"] = "target_only"
+            _record_target_file(entry, destination)
+            continue
+        if not destination.exists():
+            entry["status"] = "copied"
+            continue
+        if not destination.is_file():
+            raise FileExistsError(f"Feed v3 目标不是普通文件: {destination}")
+        raw_size = entry.get("size")
+        raw_digest = entry.get("sha256")
+        if not isinstance(raw_size, int) or isinstance(raw_size, bool):
+            raise ValueError(f"Feed migration entry 缺少文件大小: {destination}")
+        if not isinstance(raw_digest, str):
+            raise ValueError(f"Feed migration entry 缺少文件 hash: {destination}")
+        expected_size = raw_size
+        expected_digest = raw_digest
+        if (
+            destination.stat().st_size != expected_size
+            or _digest(destination) != expected_digest
+        ):
+            raise FileExistsError(f"Feed v3 目标已存在且内容不同: {destination}")
+        entry["status"] = "verified"
+        if _is_sqlite_file(destination):
+            integrity = _sqlite_integrity(destination)
+            if entry.get("integrity") != integrity:
+                raise sqlite3.DatabaseError(
+                    f"Feed SQLite integrity receipt 不一致: {destination}"
+                )
+
+    if all(entry["status"] == "source_missing" for entry in entries):
+        raise FileNotFoundError("Feed v2 与 v3 数据目录都没有可迁移文件")
+
+
+def _publish(
+    staging: Path,
+    target: Path,
+    entries: tuple[dict[str, object], ...],
+    receipt: dict[str, object],
+) -> None:
+    """发布本事务创建的文件，失败时只回滚本事务新增文件。"""
+
+    published: list[Path] = []
+    receipt_path = target / _RECEIPT
+    try:
+        for entry in entries:
+            name = str(entry["name"])
+            destination = target / name
+            if entry["status"] != "copied":
+                continue
+            published.append(destination)
+            os.replace(staging / name, destination)
+        staged_receipt = staging / _RECEIPT
+        staged_receipt.write_text(
+            json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+        published.append(receipt_path)
+        os.replace(staged_receipt, receipt_path)
+    except BaseException:
+        for path in reversed(published):
+            path.unlink(missing_ok=True)
+        raise
+
+
+def _remove_stale_staging(workspace: Path) -> None:
+    """清理上次进程崩溃遗留且未发布的 Feed staging。"""
+
+    root = workspace / "plugin-data"
+    if root.is_symlink():
+        raise ValueError(f"Feed plugin-data 目录不得是符号链接: {root}")
+    if not root.is_dir():
+        return
+    for path in root.glob(".feed-v2-migrate-*"):
+        if path.is_symlink() or path.is_file():
+            path.unlink()
+        elif path.is_dir():
+            shutil.rmtree(path)
+
+
+def _validate_receipt_file(
+    item: object,
+    *,
+    target: Path,
+    source_relative: str,
+) -> None:
+    if not isinstance(item, dict):
+        raise ValueError("Feed migration receipt 文件记录无效")
+    name = item.get("name")
+    status = item.get("status")
+    if name not in _DATA_FILES or status not in {"source_missing", *_TARGET_STATUSES}:
+        raise ValueError("Feed migration receipt 文件记录无效")
+    if item.get("source") != f"{source_relative}/{name}":
+        raise ValueError("Feed migration receipt recovery source 无效")
+    destination = target / str(name)
+    if status == "source_missing":
+        if destination.exists() or destination.is_symlink():
+            raise ValueError(f"Feed migration receipt 目标漂移: {destination}")
+        return
+    digest = item.get("sha256")
+    size = item.get("size")
+    if (
+        not isinstance(digest, str)
+        or len(digest) != 64
+        or any(character not in "0123456789abcdef" for character in digest)
+        or not isinstance(size, int)
+        or isinstance(size, bool)
+        or size < 0
+    ):
+        raise ValueError("Feed migration receipt 文件证据无效")
+    if destination.is_symlink() or not destination.is_file():
+        raise ValueError(f"Feed migration receipt 目标缺失: {destination}")
+    if destination.stat().st_size != size or _digest(destination) != digest:
+        raise ValueError(f"Feed migration receipt 目标内容漂移: {destination}")
+    if _is_sqlite_file(destination):
+        if item.get("integrity") != "ok" or item.get("sqlite_integrity") != "ok":
+            raise ValueError("Feed migration receipt 缺少 SQLite integrity")
+        _sqlite_integrity(destination)
+
+
+def _has_valid_receipt(
+    path: Path,
+    *,
+    target: Path,
+    marketplace: str,
+    source_relative: str,
+) -> bool:
+    """严格复核已有迁移 receipt 与正式 target 的内容。"""
+
+    if not path.exists() and not path.is_symlink():
+        return False
+    if path.is_symlink() or not path.is_file():
+        raise ValueError(f"Feed migration receipt 不是普通文件: {path}")
+    value: Any = json.loads(path.read_text(encoding="utf-8"))
+    files = value.get("files") if isinstance(value, dict) else None
+    expected_target = f"plugin-data/feed-{marketplace}"
+    expected_recovery = {"kind": "retained_source", "path": source_relative}
+    if (
+        not isinstance(value, dict)
+        or value.get("schema_version") != 1
+        or value.get("source") != source_relative
+        or value.get("target") != expected_target
+        or value.get("recovery") != expected_recovery
+        or not isinstance(files, list)
+        or [item.get("name") for item in files if isinstance(item, dict)]
+        != list(_DATA_FILES)
+    ):
+        raise ValueError(f"Feed migration receipt 无效: {path}")
+    for item in files:
+        _validate_receipt_file(
+            item,
+            target=target,
+            source_relative=source_relative,
+        )
+    return True
+
+
+def migrate_v2_data(*, workspace: Path, marketplace: str) -> Path:
+    """持有 workspace 独占锁迁移 Feed 数据并写最终 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:
+    """在 workspace 独占区间准备、校验并发布一次迁移。"""
+
+    if not marketplace or not marketplace.replace("-", "").isalnum():
+        raise ValueError(f"Feed marketplace 无效: {marketplace!r}")
+    source, source_relative = _select_source(workspace)
+    target = workspace / "plugin-data" / f"feed-{marketplace}"
+    validate_workspace_plugin_data_path(target, workspace)
+    _remove_stale_staging(workspace)
+    receipt_path = target / _RECEIPT
+    if _has_valid_receipt(
+        receipt_path,
+        target=target,
+        marketplace=marketplace,
+        source_relative=source_relative,
+    ):
+        return receipt_path
+
+    staging = workspace / "plugin-data" / f".feed-v2-migrate-{uuid.uuid4().hex}"
+    created_target = not target.exists()
+    ensure_workspace_plugin_data_dir(staging, workspace)
+    try:
+        entries = _stage_files(source, source_relative, staging)
+        ensure_workspace_plugin_data_dir(target, workspace)
+        _validate_targets(target, entries)
+        receipt: dict[str, object] = {
+            "schema_version": 1,
+            "source": source_relative,
+            "target": f"plugin-data/feed-{marketplace}",
+            "recovery": {
+                "kind": "retained_source",
+                "path": source_relative,
+            },
+            "files": entries,
+        }
+        _publish(staging, target, entries, receipt)
+    finally:
+        if staging.exists():
+            shutil.rmtree(staging)
+        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", default="github")
+    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/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..c286825
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+
+repo_root = Path(__file__).resolve().parents[1]
+agent_root = Path(os.environ["AKASHIC_AGENT_ROOT"])
+for path in (repo_root, agent_root):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+if str(repo_root / "mcp") not in sys.path:
+    sys.path.append(str(repo_root / "mcp"))
diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py
new file mode 100644
index 0000000..4f46b3c
--- /dev/null
+++ b/tests/test_manager_integration.py
@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import shutil
+import sqlite3
+import sys
+from pathlib import Path
+
+import pytest
+from agent.plugin_composition.proactive import FetchEmpty
+from agent.plugins.generation_activity_host import ActivityHost
+from agent.plugins.generation_proactive_host import (
+    ProactiveActivityAdapter,
+    ProactiveRuntimeBinding,
+)
+from agent.plugins.manager import PluginManager
+from bus.event_bus import EventBus
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _stage_plugin(tmp_path: Path) -> Path:
+    """复制可执行插件，并复用当前测试解释器的依赖环境。"""
+
+    source = tmp_path / "plugins" / "feed"
+    shutil.copytree(
+        ROOT,
+        source,
+        ignore=shutil.ignore_patterns(
+            ".git",
+            ".akashic-core",
+            ".pytest_cache",
+            ".venv",
+            "__pycache__",
+            "tests",
+        ),
+    )
+    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_feed_fetches_empty_and_drains(
+    tmp_path: Path,
+) -> None:
+    """走真实 stdio 与 exact source lease，并证明空库运行不访问外部 Feed。"""
+
+    # 1. staging 中没有订阅源，正式 poller 只会初始化临时数据库。
+    plugin_root = _stage_plugin(tmp_path)
+    workspace = tmp_path / "workspace"
+    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)
+
+    # 2. 通过 committed Activity binding 调用真实 MCP source。
+    snapshot = None
+    generation_id = None
+    lease = None
+    try:
+        await manager.load_all()
+        snapshot = manager.current_snapshot
+        assert snapshot is not None and snapshot.mcp_server_registry is not None
+        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"
+        assert "get_proactive_events" in runtime.mcp.server("feed").tool_names
+
+        binding = activity.active
+        assert binding is not None
+        proactive = binding.child_bindings["proactive_components"]
+        assert isinstance(proactive, ProactiveRuntimeBinding)
+        lease = manager.snapshot_store.lease(snapshot.snapshot_id)
+        result = await proactive.source("subscriptions").fetch(lease)
+        assert isinstance(result, FetchEmpty)
+    finally:
+        if lease is not None:
+            await lease.release()
+        await manager.terminate_all()
+
+    # 3. formal SQLite 完整，terminate 后 runtime、Root 与 activity 全释放。
+    database_path = workspace / "plugin-data" / "feed-builtin" / "feed_mcp.sqlite3"
+    with sqlite3.connect(database_path) as database:
+        assert database.execute("PRAGMA integrity_check").fetchone() == ("ok",)
+    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 == ()
diff --git a/tests/test_mcp_v3.py b/tests/test_mcp_v3.py
new file mode 100644
index 0000000..b282ff0
--- /dev/null
+++ b/tests/test_mcp_v3.py
@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import logging
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+
+MCP_BRIDGE_PATH = Path(__file__).resolve().parents[1] / "mcp" / "src" / "mcp_bridge.py"
+RUN_MCP_PATH = Path(__file__).resolve().parents[1] / "mcp" / "run_mcp.py"
+
+
+def _load_module(path: Path, name: str):
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_recording_fetch_ack_and_lifespan_are_zero_write(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    bridge = _load_module(MCP_BRIDGE_PATH, "feed_test_mcp_bridge")
+    monkeypatch.setenv("FEED_BACKEND", "recording")
+    monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(tmp_path))
+
+    class UnexpectedPoller:
+        def __init__(self) -> None:
+            raise AssertionError("recording 不得创建 FeedPoller")
+
+    monkeypatch.setattr(bridge, "FeedPoller", UnexpectedPoller)
+    monkeypatch.setattr(
+        bridge,
+        "_live_backend",
+        lambda: (_ for _ in ()).throw(AssertionError("recording 不得加载后端")),
+    )
+
+    server = bridge.create_mcp_server()
+    assert server is not None
+    assert bridge._fetch_proactive_events() == {"status": "empty"}
+    with pytest.raises(RuntimeError, match="不允许确认"):
+        bridge._acknowledge_proactive_events(["event-1"])
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_live_results_are_explicit_typed_payloads(monkeypatch: pytest.MonkeyPatch) -> None:
+    bridge = _load_module(MCP_BRIDGE_PATH, "feed_test_mcp_bridge_live")
+    monkeypatch.delenv("FEED_BACKEND", raising=False)
+    backend = SimpleNamespace(
+        get_proactive_events=lambda **_: [],
+        acknowledge_events=lambda ids, feedback=None: {
+            "acknowledged": list(ids),
+            "failed": [],
+        },
+    )
+    monkeypatch.setattr(bridge, "_live_backend", lambda: backend)
+
+    assert bridge._fetch_proactive_events() == {"status": "empty"}
+    backend.get_proactive_events = lambda **_: [{"event_id": "one", "kind": "content"}]
+    assert bridge._fetch_proactive_events() == {
+        "status": "items",
+        "items": [{"event_id": "one", "kind": "content"}],
+    }
+    assert bridge._acknowledge_proactive_events(["one"]) == {
+        "status": "committed",
+        "ids": ["one"],
+    }
+    assert bridge._proactive_ack_payload(
+        ["one", "two"], {"acknowledged": ["one"], "failed": ["two"]}
+    )["status"] == "failure"
+    assert bridge._proactive_ack_payload(
+        [], {"acknowledged": [], "failed": []}
+    ) == {"status": "skipped", "reason": "no_ids"}
+
+
+def test_proactive_cursor_returns_every_event_once(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    bridge = _load_module(MCP_BRIDGE_PATH, "feed_test_mcp_bridge_pages")
+    monkeypatch.delenv("FEED_BACKEND", raising=False)
+    events = [{"event_id": f"event-{index}", "kind": "content"} for index in range(51)]
+
+    def fetch(*, offset: int, limit: int):
+        return events[offset : offset + limit]
+
+    monkeypatch.setattr(
+        bridge,
+        "_live_backend",
+        lambda: SimpleNamespace(get_proactive_events=fetch),
+    )
+
+    first = bridge._fetch_proactive_events(limit=50)
+    assert first["cursor"] == "feed-offset:50"
+    second = bridge._fetch_proactive_events(limit=50, cursor=first["cursor"])
+    combined = [*first["items"], *second["items"]]
+    assert [item["event_id"] for item in combined] == [
+        f"event-{index}" for index in range(51)
+    ]
+    assert "cursor" not in second
+
+
+def test_runner_configures_stderr_without_runtime_log(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    runner = _load_module(RUN_MCP_PATH, "feed_test_run_mcp")
+    monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(tmp_path))
+    runner._setup_logging()
+    assert all(
+        not isinstance(handler, logging.FileHandler)
+        for handler in logging.getLogger().handlers
+    )
+    assert list(tmp_path.iterdir()) == []
diff --git a/tests/test_migrate_v2_data.py b/tests/test_migrate_v2_data.py
new file mode 100644
index 0000000..ee99f6b
--- /dev/null
+++ b/tests/test_migrate_v2_data.py
@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+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, relative: str = "mcp/feed-mcp") -> Path:
+    source = workspace / relative
+    source.mkdir(parents=True)
+    (source / "source_scores.json").write_text('{"source": "kept"}\n', encoding="utf-8")
+    with sqlite3.connect(source / "feed_cache.db") as database:
+        database.execute("CREATE TABLE cache (value TEXT NOT NULL)")
+        database.execute("INSERT INTO cache VALUES ('kept')")
+        database.commit()
+    with sqlite3.connect(source / "feed_mcp.sqlite3") as database:
+        database.execute("CREATE TABLE receipts (value TEXT NOT NULL)")
+        database.execute("INSERT INTO receipts VALUES ('kept')")
+        database.commit()
+    return source
+
+
+def test_primary_precedes_latest_backup_and_source_is_retained(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    primary = _legacy_data(workspace)
+    backup = _legacy_data(
+        workspace,
+        "backups/feed-plugin-migration-20260817-120000/feed-mcp",
+    )
+    (backup / "source_scores.json").write_text("backup\n", encoding="utf-8")
+
+    receipt_path = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    target = workspace / "plugin-data" / "feed-github"
+
+    assert receipt["source"] == "mcp/feed-mcp"
+    assert receipt["recovery"] == {
+        "kind": "retained_source",
+        "path": "mcp/feed-mcp",
+    }
+    assert [item["status"] for item in receipt["files"]] == [
+        "copied",
+        "copied",
+        "copied",
+    ]
+    assert receipt["files"][0]["integrity"] == "ok"
+    assert (primary / "feed_mcp.sqlite3").is_file()
+    assert (target / "source_scores.json").read_text(encoding="utf-8") == '{"source": "kept"}\n'
+    with sqlite3.connect(target / "feed_mcp.sqlite3") as database:
+        assert database.execute("PRAGMA integrity_check").fetchone() == ("ok",)
+
+
+def test_latest_backup_is_selected_when_primary_has_no_data(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    older = _legacy_data(
+        workspace,
+        "backups/feed-plugin-migration-20260816-120000/feed-mcp",
+    )
+    latest = _legacy_data(
+        workspace,
+        "backups/feed-plugin-migration-20260817-120000/feed-mcp",
+    )
+    (older / "source_scores.json").write_text("older\n", encoding="utf-8")
+    (latest / "source_scores.json").write_text("latest\n", encoding="utf-8")
+
+    receipt_path = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert receipt["source"] == (
+        "backups/feed-plugin-migration-20260817-120000/feed-mcp"
+    )
+    assert (
+        receipt_path.parent / "source_scores.json"
+    ).read_text(encoding="utf-8") == "latest\n"
+
+
+def test_conflict_fails_without_changing_source_or_target(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+    target = workspace / "plugin-data" / "feed-github"
+    target.mkdir(parents=True)
+    (target / "source_scores.json").write_text("current\n", encoding="utf-8")
+
+    with pytest.raises(FileExistsError, match="内容不同"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
+
+    assert (source / "source_scores.json").read_text(encoding="utf-8") == '{"source": "kept"}\n'
+    assert not (target / migration._RECEIPT).exists()
+    assert list((workspace / "plugin-data").glob(".feed-v2-migrate-*")) == []
+
+
+def test_source_missing_target_only_is_recorded_and_verified(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+    (source / "source_scores.json").unlink()
+    target = workspace / "plugin-data" / "feed-github"
+    target.mkdir(parents=True)
+    (target / "source_scores.json").write_text("target-only\n", encoding="utf-8")
+
+    receipt_path = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    entries = {
+        item["name"]: item
+        for item in json.loads(receipt_path.read_text(encoding="utf-8"))["files"]
+    }
+    assert entries["source_scores.json"]["status"] == "target_only"
+    assert entries["source_scores.json"]["size"] == len("target-only\n")
+    assert entries["feed_mcp.sqlite3"]["status"] == "copied"
+
+
+def test_in_process_publish_failure_rolls_back_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" / "feed-github").exists()
+    assert list((workspace / "plugin-data").glob(".feed-v2-migrate-*")) == []
+
+
+def test_post_replace_cancellation_rolls_back_published_file(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+    original_replace = os.replace
+    calls = 0
+
+    def cancel_after_first_replace(source_path: Path, destination: Path) -> None:
+        nonlocal calls
+        original_replace(source_path, destination)
+        calls += 1
+        if calls == 1:
+            raise KeyboardInterrupt("injected post-replace cancellation")
+
+    monkeypatch.setattr(migration.os, "replace", cancel_after_first_replace)
+    with pytest.raises(KeyboardInterrupt, match="post-replace"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
+
+    target = workspace / "plugin-data" / "feed-github"
+    assert not target.exists()
+    assert (source / "feed_mcp.sqlite3").is_file()
+    assert list((workspace / "plugin-data").glob(".feed-v2-migrate-*")) == []
+
+
+def test_crash_partial_publish_is_reconciled_on_rerun(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    source = _legacy_data(workspace)
+    plugin_data = workspace / "plugin-data"
+    target = plugin_data / "feed-github"
+    stale = plugin_data / ".feed-v2-migrate-crashed"
+    target.mkdir(parents=True)
+    stale.mkdir(parents=True)
+    (target / "source_scores.json").write_bytes(
+        (source / "source_scores.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["source_scores.json"] == "verified"
+    assert statuses["feed_mcp.sqlite3"] == "copied"
+    assert not stale.exists()
+
+
+def test_process_crash_after_replace_is_reconciled_on_restart(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    _legacy_data(workspace)
+    script = f"""
+from pathlib import Path
+import os
+from scripts import migrate_v2_data as migration
+
+original_replace = migration.os.replace
+
+def crash_after_replace(source, destination):
+    original_replace(source, destination)
+    os._exit(137)
+
+migration.os.replace = crash_after_replace
+migration.migrate_v2_data(workspace=Path({str(workspace)!r}), marketplace="github")
+"""
+
+    crashed = subprocess.run(
+        [sys.executable, "-c", script],
+        cwd=Path(__file__).parents[1],
+        check=False,
+    )
+    assert crashed.returncode == 137
+
+    receipt = migration.migrate_v2_data(
+        workspace=workspace,
+        marketplace="github",
+    )
+    statuses = {
+        item["name"]: item["status"]
+        for item in json.loads(receipt.read_text(encoding="utf-8"))["files"]
+    }
+    assert sorted(statuses.values()) == ["copied", "copied", "verified"]
+    assert list((workspace / "plugin-data").glob(".feed-v2-migrate-*")) == []
+
+
+def test_symlink_and_receipt_drift_are_rejected(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    outside = tmp_path / "outside"
+    _legacy_data(outside)
+    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-2"
+    _legacy_data(workspace)
+    receipt_path = migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    target_file = receipt_path.parent / "source_scores.json"
+    target_file.write_text("drifted\n", encoding="utf-8")
+    with pytest.raises(ValueError, match="内容漂移"):
+        migration.migrate_v2_data(workspace=workspace, marketplace="github")
+
+
+def test_migration_requires_idle_workspace(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    _legacy_data(workspace)
+    lock = WorkspaceInstanceLock(workspace)
+    lock.acquire()
+    try:
+        with pytest.raises(RuntimeError, match="其他 runtime 占用"):
+            migration.migrate_v2_data(workspace=workspace, marketplace="github")
+    finally:
+        lock.release()
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
new file mode 100644
index 0000000..0ba824d
--- /dev/null
+++ b/tests/test_plugin.py
@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import inspect
+from pathlib import Path
+
+import pytest
+import plugin
+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.static_manifest import load_static_plugin_manifest
+from agent.plugins.composable import ComposablePlugin
+from agent.plugins.manager import _copy_validation_data
+from plugin import FeedConfig, FeedProactiveConfig
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_pure_v3_exports_and_exact_apply() -> None:
+    assert plugin.api_version == 3
+    assert plugin.name == "feed"
+    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",)
+
+
+@pytest.mark.asyncio
+async def test_apply_registers_mcp_and_source_without_data_writes(tmp_path: Path) -> None:
+    root = CompositionRoot("feed: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_dir = tmp_path / "plugin-data"
+    await root.mount(
+        ComposablePlugin.from_module(plugin),
+        name="feed",
+        runtime=PluginRuntime(
+            plugin_id="feed",
+            plugin_dir=ROOT,
+            data_dir=data_dir,
+            workspace=tmp_path / "workspace",
+            config=FeedConfig(
+                proactive=FeedProactiveConfig(enabled=True),
+            ),
+        ),
+    )
+
+    mcp = _freeze_plugin_mcp_servers(servers, root.instance_token)["feed"].definition
+    source = _freeze_plugin_proactive_components(
+        components,
+        root.instance_token,
+        {"feed": "feed:test"},
+    ).source("subscriptions")
+    assert mcp.candidate_env == {"FEED_BACKEND": "recording"}
+    assert mcp.candidate_read_only_tools == ("get_proactive_events",)
+    assert source is not None
+    assert source.definition.channels == ("content",)
+    assert source.definition.mcp_server == "feed"
+    assert not data_dir.exists()
+    await root.dispose()
+
+
+@pytest.mark.asyncio
+async def test_disabled_proactive_omits_source(tmp_path: Path) -> None:
+    root = CompositionRoot("feed:disabled")
+    servers = PluginMcpServers(root.instance_token)
+    components = PluginProactiveComponents(root.instance_token)
+    await root.context.provide(MCP_SERVERS, servers)
+    await root.context.provide(PROACTIVE_COMPONENTS, components)
+    await root.mount(
+        ComposablePlugin.from_module(plugin),
+        name="feed",
+        runtime=PluginRuntime(
+            plugin_id="feed",
+            plugin_dir=ROOT,
+            data_dir=tmp_path / "plugin-data",
+            workspace=tmp_path / "workspace",
+            config=FeedConfig(
+                proactive=FeedProactiveConfig(enabled=False),
+            ),
+        ),
+    )
+    catalog = _freeze_plugin_proactive_components(
+        components,
+        root.instance_token,
+        {"feed": "feed:disabled"},
+    )
+    assert catalog.sources == {}
+    await root.dispose()
+
+
+def test_static_manifest_freezes_recording_and_data_exclusions() -> None:
+    manifest = load_static_plugin_manifest(Path(__file__).resolve().parents[1])
+
+    assert manifest.name == "feed"
+    assert manifest.version == "3.0.0"
+    assert manifest.api_version == 3
+    assert manifest.requirements == ("mcp/requirements.txt",)
+    assert manifest.exclude_data_paths == (
+        "feed_mcp.sqlite3",
+        "feed_mcp.sqlite3-wal",
+        "feed_mcp.sqlite3-shm",
+        "source_scores.json",
+        "feed_cache.db",
+        "feed_cache.db-wal",
+        "feed_cache.db-shm",
+        ".feed-v2-migration.json",
+    )
+    assert len(manifest.mcp_servers) == 1
+    server = manifest.mcp_servers[0]
+    assert server.required_tools == ("get_proactive_events", "acknowledge_events")
+    assert server.candidate_read_only_tools == ("get_proactive_events",)
+    assert server.candidate_env == (("FEED_BACKEND", "recording"),)
+
+
+def test_candidate_copy_excludes_sqlite_and_sidecars(tmp_path: Path) -> None:
+    manifest = load_static_plugin_manifest(ROOT)
+    source = tmp_path / "workspace" / "plugin-data" / "feed-builtin"
+    source.mkdir(parents=True)
+    for name in (
+        "feed_mcp.sqlite3",
+        "feed_mcp.sqlite3-wal",
+        "feed_mcp.sqlite3-shm",
+        "feed_cache.db",
+        "feed_cache.db-wal",
+        "feed_cache.db-shm",
+        "source_scores.json",
+        ".feed-v2-migration.json",
+    ):
+        (source / name).write_text(f"secret:{name}", encoding="utf-8")
+    (source / "candidate-visible.txt").write_text("visible", encoding="utf-8")
+    target = tmp_path / "validation" / "feed"
+
+    inventory = _copy_validation_data(  # pyright: ignore[reportPrivateUsage]
+        source,
+        target,
+        manifest.exclude_data_paths,
+    )
+
+    assert inventory == ("candidate-visible.txt",)
+    assert (target / "candidate-visible.txt").read_text(encoding="utf-8") == "visible"