feat(plugin): 插件配置与实例描述符迁入独立表

插件配置此前寄存在 systemconfig 的 plugin.<实例ID> 单键下，实例描述符则整份挤在
PluginInstances 这一个 JSON 键里，两者都不成立：

- plugin.<ID> 是裸字符串 key，而仓库规则本身禁止用裸字符串做 SystemConfig key，
  只允许先定义 SystemConfigKey 枚举项；插件 ID 由用户安装决定，永远枚举不出常量。
  条目数随安装量增长，混在系统设置表里会把主程序自己的设置项淹掉。
- 那个键存的其实是实例 ID，表里没有任何一列说得出它属于哪个插件，想列出某个插件
  的全部实例配置只能靠字符串前缀去猜。
- 实例描述符整份读出再整份写回，改一个分身要重写全部分身。

改为 plugininstance 一实例一行：instance_id 与 source_plugin_id 构成身份，本体与
分身由两者是否相等派生而不设模式列——模式列会是这个等式的冗余副本，两者一旦失步
同一行就会在不同读取口被判成不同角色。展示信息与业务参数同存一行，属于同一个
生命周期，分表只会让建分身、删分身退化成两张表之间的协调问题。

读写路由落在 SystemConfigOper.get/set/delete，而不是在各个调用方各改一处：第三方
插件可能直接用 self.systemconfig.get("plugin.xxx") 读写自己的配置，只改
PluginConfigStore 与 _PluginBase.get_config/update_config 必然漏掉它们。路由只做
前缀识别，插入、更新与空本体行回收都委托 PluginInstanceOper，不在配置层重抄一遍。

PluginInstances 旧键迁移后不删，留作回滚依据，并以
SystemConfigKey.PluginInstancesImported 标志防止重复导入；判据不能是「表当前为空」，
否则用户把分身全部删光后，下次启动会把它们整批导回来。

迁移链：b2d4f6a8c1e3 -> 281965691a20（3.0.34 建表并搬描述符）
-> c4e1a7b9d2f6（3.0.35 加 config_data 并搬 plugin.* 配置）。

依赖基线与启动模块数随两个新模块重算，架构文档与数据库技能表目录同步登记新表。

app/db/models/__init__.py

@@ -22,6 +22,10 @@ _MODEL_EXPORTS = {
         "app.db.models.plugininstallation",
         "PluginInstallation",
     ),
+    "PluginInstance": (
+        "app.db.models.plugininstance",
+        "PluginInstance",
+    ),
     "PluginIdentity": (
         "app.db.models.pluginidentity",
         "PluginIdentity",

app/db/models/plugininstance.py

@@ -0,0 +1,87 @@
+"""共享源码插件的实例描述符与配置持久化模型。"""
+
+from __future__ import annotations
+
+from typing import Any, List, Optional, Self, cast
+
+from sqlalchemy import JSON, Index, String, UniqueConstraint, select
+from sqlalchemy.orm import Mapped, Session, mapped_column
+
+from app.db.base import Base, get_id_column
+
+
+class PluginInstance(Base):
+    """持久化一个共享源码插件的运行实例，一实例一行。
+
+    ``instance_id`` 既是分身的实例 ID，也可以等于 ``source_plugin_id`` 表示源插件
+    本体自身；两者相等即本体，不等即分身，不再另设模式列——该列会是从这个等式复制
+    出来的冗余副本，两者一旦失步就会让同一行在不同读取口被判成不同角色。
+
+    身份（``instance_id`` 与 ``source_plugin_id``）之外的列都是这个实例的设置：
+    展示信息与业务参数同属一个生命周期，应当随卸载一起消失、随重建一起重来，因此
+    存放在同一行而非分表协调。
+
+    业务参数存进 ``config_data`` 而不是 ``systemconfig`` 的 ``plugin.<实例ID>`` 单键：
+    那个键由插件 ID 决定、条目数随安装量增长，混在系统设置里会把主程序自己的设置项
+    淹掉；而且它存的其实是实例 ID，表里没有任何一列说得出它属于哪个插件，想列出
+    某个插件的全部实例配置就只能靠字符串前缀去猜。
+
+    表名由 ``Base`` 按类名自动派生为小写 ``plugininstance``。
+    """
+
+    id = get_id_column()
+    instance_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    source_plugin_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    plugin_name: Mapped[Optional[str]] = mapped_column(String(255))
+    plugin_desc: Mapped[Optional[str]] = mapped_column(String(255))
+    plugin_icon: Mapped[Optional[str]] = mapped_column(String(255))
+    config_data: Mapped[Optional[Any]] = mapped_column(JSON)
+    created_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    updated_at: Mapped[str] = mapped_column(String(40), nullable=False)
+
+    __table_args__ = (
+        UniqueConstraint("instance_id", name="uq_plugininstance_instance_id"),
+        Index("ix_plugininstance_source_plugin_id", "source_plugin_id"),
+    )
+
+    @property
+    def is_host(self) -> bool:
+        """该行是否为源插件本体自身，而非共享其源码的分身。"""
+        return bool(self.instance_id == self.source_plugin_id)
+
+    @property
+    def carries_only_identity(self) -> bool:
+        """除一对身份列外是否已不承载任何设置。
+
+        本体行会在插件首次存配置时被隐式建出，配置被删掉后若各列皆空就只剩身份列，
+        留着会让「有哪些插件登记过本体设置」的枚举逐步失真，因而可以回收。分身行不
+        适用：分身的存在本身就由这一行表达，清空设置不等于删除分身。
+        """
+        return not any(
+            (
+                self.config_data is not None,
+                self.plugin_name,
+                self.plugin_desc,
+                self.plugin_icon,
+            )
+        )
+
+    @classmethod
+    def get_by_instance_id(cls, db: Session, instance_id: str) -> Optional[Self]:
+        """在调用方 Session 中按实例 ID 查询单行。"""
+        return cast(
+            Optional[Self],
+            db.execute(select(cls).where(cls.instance_id == instance_id)).scalars().first(),
+        )
+
+    @classmethod
+    def list_by_source_plugin_id(cls, db: Session, source_plugin_id: str) -> List[Self]:
+        """在调用方 Session 中列出某个源插件名下的全部实例，含本体与各个分身。"""
+        return cast(
+            List[Self],
+            list(
+                db.execute(
+                    select(cls).where(cls.source_plugin_id == source_plugin_id)
+                ).scalars()
+            ),
+        )

app/db/oper/plugininstance.py

@@ -0,0 +1,165 @@
+"""插件实例的数据访问原语。"""
+
+from __future__ import annotations
+
+import copy
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.db.base import DbOper
+from app.db.models.plugininstance import PluginInstance
+
+
+def _now() -> str:
+    """返回实例行使用的 ISO-8601 时间戳。"""
+    return datetime.now(timezone.utc).isoformat()
+
+
+class PluginInstanceOper(DbOper):
+    """在调用方 Session 或独占事务中查询并暂存插件实例。
+
+    一行由 ``instance_id`` 唯一确定，``source_plugin_id`` 指向提供代码的源插件；
+    两者相等即源插件本体，不等即共享其源码的分身。
+    """
+
+    def get(self, instance_id: str) -> Optional[PluginInstance]:
+        """按实例 ID 查询单行，分身与本体共用同一张表和同一个查询。"""
+        return self._execute_sync_query(
+            lambda session: PluginInstance.get_by_instance_id(session, instance_id)
+        )
+
+    def list_by_source(self, source_plugin_id: str) -> list[PluginInstance]:
+        """按源插件 ID 列举其全部实例，含分身与本体。"""
+        return list(
+            self._execute_sync_query(
+                lambda session: PluginInstance.list_by_source_plugin_id(
+                    session, source_plugin_id
+                )
+            )
+            or []
+        )
+
+    def list_all(self) -> list[PluginInstance]:
+        """列举全部实例，供目录投影与兜底导入判空使用。"""
+        return list(
+            self._execute_sync_query(
+                lambda session: session.execute(select(PluginInstance)).scalars().all()
+            )
+        )
+
+    def save(self, **fields: Any) -> PluginInstance:
+        """按 ``instance_id`` 新增或更新一行，只写入本次给出的列。
+
+        查询与写入收在同一事务内，避免两个并发的首次写入各自读到空、双双插入而
+        撞上 ``instance_id`` 唯一键。
+
+        已存在的行不允许改写 ``source_plugin_id``：它连同 ``instance_id`` 构成这一行
+        的身份，改写会把整行的归属换掉，同时把本体与分身的角色判定一并翻转，只可能
+        来自调用方错把分身自身实例 ID 当作源插件 ID 使用，因而在持久化层直接拒绝，
+        不依赖上层纪律。
+
+        :param fields: 实例字段，须含 ``instance_id``
+        :return: 写入后的实例行
+        :raise ValueError: 已存在的行与本次写入的 ``source_plugin_id`` 不一致
+        """
+        now = _now()
+        instance_id = fields["instance_id"]
+
+        def stage(session: Session) -> PluginInstance:
+            """在同一事务内查询并新增或更新实例行。"""
+            existing = PluginInstance.get_by_instance_id(session, instance_id)
+            if existing is None:
+                record = PluginInstance(**fields, created_at=now, updated_at=now)
+                session.add(record)
+                return record
+            incoming_source = fields.get("source_plugin_id")
+            if incoming_source is not None and existing.source_plugin_id != incoming_source:
+                raise ValueError(
+                    f"插件实例 {instance_id} 已归属于 {existing.source_plugin_id}，"
+                    f"不能改写为 {incoming_source}"
+                )
+            for key, value in {**fields, "updated_at": now}.items():
+                setattr(existing, key, value)
+            return existing
+
+        return self._execute_sync_write(stage)
+
+    def delete(self, instance_id: str) -> bool:
+        """按实例 ID 删除整行，返回删除前是否存在。
+
+        这会连同该实例的业务参数与展示信息一起清除：分身的存在本身就由这一行表达，
+        删行即卸载分身。
+        """
+
+        def stage(session: Session) -> bool:
+            """在同一事务内查询并删除，避免读写跨两个独立事务。"""
+            existing = PluginInstance.get_by_instance_id(session, instance_id)
+            if existing is None:
+                return False
+            session.delete(existing)
+            return True
+
+        return bool(self._execute_sync_write(stage))
+
+    def save_config_data(
+        self,
+        *,
+        instance_id: str,
+        source_plugin_id: str,
+        config_data: Any,
+    ) -> Optional[bool]:
+        """写入业务参数，保留该行已有的身份与展示信息。
+
+        :param instance_id: 实例 ID
+        :param source_plugin_id: 该行不存在时用于建行的源插件 ID
+        :param config_data: 业务参数
+        :return: True 已写入，None 值未变化无需写入
+        """
+
+        def stage(session: Session) -> Optional[bool]:
+            """在同一事务内创建或更新业务参数。"""
+            record = PluginInstance.get_by_instance_id(session, instance_id)
+            if record is None:
+                now = _now()
+                session.add(
+                    PluginInstance(
+                        instance_id=instance_id,
+                        source_plugin_id=source_plugin_id,
+                        config_data=copy.deepcopy(config_data),
+                        created_at=now,
+                        updated_at=now,
+                    )
+                )
+                return True
+            if record.config_data == config_data:
+                return None
+            record.config_data = copy.deepcopy(config_data)
+            record.updated_at = _now()
+            return True
+
+        return self._execute_sync_write(stage)
+
+    def clear_config_data(self, instance_id: str) -> bool:
+        """清空某实例的业务参数，保留其身份与展示信息。
+
+        清空后若该行是本体且各列皆空，整行一并移除，避免只剩一对身份列的空行堆积。
+
+        :param instance_id: 实例 ID
+        :return: 该行存在并已处理
+        """
+
+        def stage(session: Session) -> bool:
+            """在同一事务内清空业务参数并回收空的本体行。"""
+            record = PluginInstance.get_by_instance_id(session, instance_id)
+            if record is None:
+                return False
+            record.config_data = None
+            record.updated_at = _now()
+            if record.is_host and record.carries_only_identity:
+                session.delete(record)
+            return True
+
+        return bool(self._execute_sync_write(stage))

app/db/oper/systemconfig.py

@@ -7,7 +7,9 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
+from app.db.models.plugininstance import PluginInstance
 from app.db.models.systemconfig import SystemConfig
+from app.db.oper.plugininstance import PluginInstanceOper
 from app.foundation.singleton import Singleton
 from app.schemas.types import SystemConfigKey
 
@@ -18,6 +20,19 @@ class SystemConfigOper(DbOper, metaclass=Singleton):
     """
     系统配置管理
     """
+    # 插件配置沿用 plugin.<实例ID> 这个对外契约，但物理上存进插件实例表。路由必须落在
+    # 这一层：_PluginBase 之外还有第三方插件直接拿 SystemConfigOper 读写这个键，在每个
+    # 调用方各改一处必然漏掉它们。
+    PLUGIN_CONFIG_KEY_PREFIX = "plugin."
+
+    @classmethod
+    def _plugin_id_of(cls, key: str) -> Optional[str]:
+        """把 plugin.<实例ID> 解析为实例 ID；非插件配置键返回 None。"""
+        if not key.startswith(cls.PLUGIN_CONFIG_KEY_PREFIX):
+            return None
+        instance_id = key[len(cls.PLUGIN_CONFIG_KEY_PREFIX):]
+        return instance_id or None
+
     def __init__(self):
         """初始化空快照，数据库加载由启动组合根显式执行。"""
         super().__init__()
@@ -32,10 +47,20 @@ class SystemConfigOper(DbOper, metaclass=Singleton):
             items = SystemConfig.list(db) if db is not None else self._execute_sync_query(
                 SystemConfig.list
             )
+            instances = PluginInstance.list(db) if db is not None else self._execute_sync_query(
+                PluginInstance.list
+            )
             snapshot = {
                 item.key: copy.deepcopy(item.value)
                 for item in items
             }
+            # 插件配置以同一套 plugin.<实例ID> 键进入快照，读取端感知不到换表
+            snapshot.update({
+                f"{self.PLUGIN_CONFIG_KEY_PREFIX}{item.instance_id}":
+                    copy.deepcopy(item.config_data)
+                for item in instances
+                if item.config_data is not None
+            })
             with self._snapshot_lock:
                 self.__SYSTEMCONF = snapshot
                 self._loaded = True
@@ -76,6 +101,17 @@ class SystemConfigOper(DbOper, metaclass=Singleton):
             key = key.value
         self._require_loaded()
         with self._write_lock:
+            instance_id = self._plugin_id_of(key)
+            if instance_id is not None:
+                # 分身的实例行由分身服务先行建出，这里建不出分身；能落到建行分支的
+                # 只有本体自身，源插件因而就是它自己
+                result = PluginInstanceOper().save_config_data(
+                    instance_id=instance_id,
+                    source_plugin_id=instance_id,
+                    config_data=value,
+                )
+                self._publish_value(key, value)
+                return result
 
             def write(db):
                 """在当前事务中创建或更新配置记录。"""
@@ -169,6 +205,13 @@ class SystemConfigOper(DbOper, metaclass=Singleton):
             key = key.value
         self._require_loaded()
         with self._write_lock:
+            instance_id = self._plugin_id_of(key)
+            if instance_id is not None:
+                # 只清业务参数：这一行还承载着实例身份与展示信息，整行删掉会把
+                # 「删一份配置」变成「删掉这个实例」
+                PluginInstanceOper().clear_config_data(instance_id)
+                self._publish_delete(key)
+                return True
 
             def delete(db):
                 """在当前事务中删除配置记录。"""

app/runtime/extensions/plugin/runtime.py

@@ -27,6 +27,7 @@ from app.runtime.extensions.plugin.projection import PluginProjection
 from app.runtime.extensions.plugin.registry import PluginRegistry
 from app.runtime.extensions.plugin.storage import (
     PluginConfigStore,
+    PluginInstanceDirectory,
     PluginInstanceStore,
     PluginStorage,
 )
@@ -89,6 +90,7 @@ class PluginRuntimeEnvironment:
 
     plugins_root: Path
     storage: Callable[[], PluginStorage]
+    instance_directory: Callable[[], PluginInstanceDirectory]
     system: Callable[[], PluginSystemServices]
     database: Callable[[], PluginDatabase]
     catalog_factory: PluginCatalogFactory
@@ -134,7 +136,10 @@ def build_plugin_runtime(
 ) -> PluginRuntime:
     """按依赖顺序构造唯一插件运行时，各业务能力仍由对应 owner 实现。"""
     registry = PluginRegistry()
-    instances = PluginInstanceStore(storage=environment.storage)
+    instances = PluginInstanceStore(
+        storage=environment.storage,
+        directory=environment.instance_directory,
+    )
     configs = PluginConfigStore(
         storage=environment.storage,
         database=environment.database,

app/runtime/extensions/plugin/storage.py

@@ -148,15 +148,126 @@ class PluginConfigStore:
         return True
 
 
+InstanceReader = Callable[[str], "PluginInstance | None"]
+InstanceLister = Callable[[], "list[PluginInstance]"]
+InstanceSourceLister = Callable[[str], "list[PluginInstance]"]
+InstanceWriter = Callable[["PluginInstance"], None]
+InstanceDeleter = Callable[[str], bool]
+
+
+def _empty_instance_get(_instance_id: str) -> PluginInstance | None:
+    """组合根尚未装配时返回空实例描述。"""
+    return None
+
+
+def _empty_instance_list() -> list[PluginInstance]:
+    """组合根尚未装配时返回空实例列表。"""
+    return []
+
+
+def _empty_instance_list_by_source(_source_plugin_id: str) -> list[PluginInstance]:
+    """组合根尚未装配时返回空实例列表。"""
+    return []
+
+
+def _ignore_instance_save(_instance: PluginInstance) -> None:
+    """组合根尚未装配时忽略实例描述写入。"""
+
+
+def _ignore_instance_delete(_instance_id: str) -> bool:
+    """组合根尚未装配时报告实例描述未删除。"""
+    return False
+
+
+class PluginInstanceDirectory:
+    """封装插件实例表的持久化能力。
+
+    分身与源插件本体共用同一张表、同一套读写原语，两者靠 ``instance_id`` 是否等于
+    ``source_plugin_id`` 区分；本类不做角色过滤，角色隔离由调用方
+    （``PluginInstanceStore``）负责，因为只有调用方知道当前服务的是分身清单还是
+    本体自身的那一行。
+    """
+
+    def __init__(
+            self,
+            *,
+            get: InstanceReader = _empty_instance_get,
+            list_all: InstanceLister = _empty_instance_list,
+            list_by_source: InstanceSourceLister = _empty_instance_list_by_source,
+            save: InstanceWriter = _ignore_instance_save,
+            delete: InstanceDeleter = _ignore_instance_delete,
+    ) -> None:
+        """保存由启动组合根提供的实例表读写函数。"""
+        self._get = get
+        self._list_all = list_all
+        self._list_by_source = list_by_source
+        self._save = save
+        self._delete = delete
+
+    def get(self, instance_id: str) -> PluginInstance | None:
+        """按实例 ID 读取单条描述，不区分分身与本体。"""
+        return self._get(instance_id)
+
+    def list_all(self) -> list[PluginInstance]:
+        """列出表中全部描述，不区分分身与本体。"""
+        return self._list_all()
+
+    def list_by_source(self, source_plugin_id: str) -> list[PluginInstance]:
+        """按源插件 ID 列出其全部描述，不区分分身与本体。"""
+        return self._list_by_source(source_plugin_id)
+
+    def save(self, instance: PluginInstance) -> None:
+        """新增或更新一条描述，以 ``instance_id`` 为稳定键。"""
+        self._save(instance)
+
+    def delete(self, instance_id: str) -> bool:
+        """按实例 ID 删除一行，连同其配置，返回删除前是否存在。"""
+        return self._delete(instance_id)
+
+
 class PluginInstanceStore:
-    """管理虚拟插件实例描述，并隔离兼容清单与新实例清单。"""
+    """管理共享源码的分身实例描述，并把源插件本体自身的那一行隔离在视图之外。
 
-    def __init__(self, *, storage: Callable[[], "PluginStorage"]) -> None:
-        """保存延迟解析的持久化端口，便于启动组合根后装配。"""
+    两类记录同存一张表，靠 ``instance_id`` 是否等于 ``source_plugin_id`` 区分：
+    本类只服务分身，本体行（它承载插件自身的业务参数）读不到也改不到。
+    """
+
+    def __init__(
+            self,
+            *,
+            storage: Callable[[], "PluginStorage"],
+            directory: Callable[[], PluginInstanceDirectory],
+    ) -> None:
+        """保存独立表持久化端口，以及旧 systemconfig 单键端口供兜底导入使用。"""
         self._storage = storage
+        self._directory = directory
+        self._bootstrap_checked = False
 
-    def all(self) -> dict[str, PluginInstance]:
-        """读取全部有效实例，忽略损坏项以免阻断存量插件启动。"""
+    def _ensure_bootstrapped(self) -> None:
+        """旧 systemconfig 单键的内容向独立表兜底导入一次，且只导入一次。
+
+        alembic 迁移已经搬过一轮，这里兜的是「库结构升级后又用旧版本写过分身」这类
+        回滚往返。导入完成后落一个持久化标记，判据不能是「表当前为空」：旧键刻意保留
+        作回滚依据、从不清理，用户把分身全部删光后表就会重新变空，下次进程启动会把
+        已删除的分身整批导回来，删一次复活一次。进程内另外维护一个已检查标志，避免
+        每次访问都为此多打一次查询。
+
+        :raise Exception: 导入失败时向上抛出，不吞掉持久化层错误
+        """
+        if self._bootstrap_checked:
+            return
+        self._bootstrap_checked = True
+        storage = self._storage()
+        if storage.read(SystemConfigKey.PluginInstancesImported):
+            return
+        directory = self._directory()
+        if not directory.list_all():
+            for instance in self._legacy_instances().values():
+                directory.save(instance)
+        storage.write(SystemConfigKey.PluginInstancesImported, True)
+
+    def _legacy_instances(self) -> dict[str, PluginInstance]:
+        """解析旧 systemconfig 单键里的实例描述，兼容历史字典与列表两种载荷形态。"""
         raw_instances = self._storage().read(SystemConfigKey.PluginInstances) or {}
         if isinstance(raw_instances, list):
             entries = {
@@ -180,43 +291,65 @@ class PluginInstanceStore:
                 continue
         return instances
 
+    def all(self) -> dict[str, PluginInstance]:
+        """读取全部登记的分身实例，不含源插件本体自身的那一行。"""
+        self._ensure_bootstrapped()
+        return {
+            record.instance_id: record
+            for record in self._directory().list_all()
+            if not record.is_host
+        }
+
     def get(self, instance_id: str) -> PluginInstance | None:
-        """读取指定实例描述。"""
-        return self.all().get(instance_id)
+        """读取指定分身实例描述；本体自身的那一行不会从这里返回。"""
+        self._ensure_bootstrapped()
+        record = self._directory().get(instance_id)
+        return record if record is not None and not record.is_host else None
 
     def save(self, instance: PluginInstance) -> None:
-        """新增或更新实例描述，并以实例 ID 作为稳定持久化键。"""
-        instances = self.all()
-        instances[instance.instance_id] = instance
-        self._write(instances)
+        """新增或更新分身实例描述，并以实例 ID 作为稳定持久化键。
+
+        分身的实例 ID 必须区别于其源插件 ID：两者相等的那一行表示的是本体自身，
+        按分身写入会把本体承载的业务参数顶掉。
+        """
+        self._ensure_bootstrapped()
+        if instance.is_host:
+            raise ValueError(
+                f"分身实例 {instance.instance_id} 的 ID 不能等于其源插件 ID"
+            )
+        self._directory().save(instance)
 
     def delete(self, instance_id: str) -> bool:
-        """删除指定实例描述，返回删除前是否存在。"""
-        instances = self.all()
-        removed = instances.pop(instance_id, None)
-        if removed is None:
+        """删除指定分身实例描述连同其配置，返回删除前是否存在。"""
+        self._ensure_bootstrapped()
+        record = self.get(instance_id)
+        if record is None:
             return False
-        self._write(instances)
-        return True
+        return self._directory().delete(record.instance_id)
 
     def for_source(self, source_plugin_id: str) -> list[PluginInstance]:
-        """按持久化顺序返回引用同一源码插件的全部实例。"""
+        """按持久化顺序返回引用同一源码插件的全部分身，不含本体自身那一行。"""
+        self._ensure_bootstrapped()
         return [
-            instance
-            for instance in self.all().values()
-            if instance.source_plugin_id == source_plugin_id
+            record
+            for record in self._directory().list_by_source(source_plugin_id)
+            if not record.is_host
         ]
 
-    def _write(self, instances: dict[str, PluginInstance]) -> None:
-        """把模型映射序列化为普通字典，避免存储层依赖 Pydantic。"""
-        payload = {
-            instance_id: instance.model_dump(mode="json")
-            for instance_id, instance in instances.items()
-        }
-        self._storage().write(SystemConfigKey.PluginInstances, payload)
-
 
 _plugin_storage = PluginStorage()
+_plugin_instance_directory = PluginInstanceDirectory()
+
+
+def configure_plugin_instance_directory(directory: PluginInstanceDirectory) -> None:
+    """由启动组合根替换插件实例表持久化实现。"""
+    global _plugin_instance_directory
+    _plugin_instance_directory = directory
+
+
+def get_plugin_instance_directory() -> PluginInstanceDirectory:
+    """返回当前插件实例表持久化端口。"""
+    return _plugin_instance_directory
 
 
 def configure_plugin_storage(storage: PluginStorage) -> None:

app/schemas/plugin.py

@@ -5,6 +5,7 @@ from typing import Dict, List, Literal, Optional, Union
 from pydantic import AfterValidator as _AfterValidator
 from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator
 from pydantic import PrivateAttr as _PrivateAttr
+from pydantic import computed_field as _computed_field
 
 from app.schemas.common import JsonData
 
@@ -59,7 +60,19 @@ class PluginInstance(BaseModel):
     plugin_name: Optional[str] = Field(default=None, description="实例展示名称")
     plugin_desc: Optional[str] = Field(default=None, description="实例展示描述")
     plugin_icon: Optional[str] = Field(default=None, description="实例展示图标")
-    mode: Literal["virtual"] = Field(default="virtual", description="实例实现模式")
+
+    @property
+    def is_host(self) -> bool:
+        """该实例是否为源插件本体自身，而非共享其源码的分身。"""
+        return self.instance_id == self.source_plugin_id
+
+    @_computed_field(  # type: ignore[prop-decorator, misc]
+        description="实例实现模式：virtual 为共享源码的分身，host 为源插件本体自身",
+    )
+    @property
+    def mode(self) -> Literal["virtual", "host"]:
+        """由一对身份 ID 派生实例角色，而非另存一份可能失步的副本。"""
+        return "host" if self.is_host else "virtual"
 
 
 class Plugin(BaseModel):

app/schemas/types.py

@@ -384,8 +384,10 @@ class SystemConfigKey(Enum):
     UserCustomCSS = "UserCustomCSS"
     # 用户已安装的插件
     UserInstalledPlugins = "UserInstalledPlugins"
-    # 共享源码插件的虚拟运行实例
+    # 共享源码插件的虚拟运行实例（已迁移到独立表，本键保留作回滚依据）
     PluginInstances = "PluginInstances"
+    # 上面那个旧键是否已完成向独立表的兜底导入，避免用户删光分身后被重新导回
+    PluginInstancesImported = "PluginInstancesImported"
     # 插件文件夹分组配置
     PluginFolders = "PluginFolders"
     # 默认电影订阅规则

app/startup/initializers/plugins.py

@@ -77,7 +77,9 @@ from app.application.plugin.transaction import (
 )
 from app.application.scheduling import update_plugin_job
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
+from app.db.models.plugininstance import PluginInstance as PluginInstanceRecord
 from app.db.oper.plugindata import PluginDataOper
+from app.db.oper.plugininstance import PluginInstanceOper
 from app.db.plugin.registry import (
     destroy_database,
     ensure_database,
@@ -115,8 +117,11 @@ from app.runtime.extensions.plugin.runtime import (
     build_plugin_runtime,
 )
 from app.runtime.extensions.plugin.storage import (
+    PluginInstanceDirectory,
     PluginStorage,
+    configure_plugin_instance_directory,
     configure_plugin_storage,
+    get_plugin_instance_directory,
     get_plugin_storage,
 )
 from app.runtime.extensions.plugin.system import (
@@ -129,7 +134,7 @@ from app.runtime.loop import main_loop_registry
 from app.runtime.resources import acquire_managed_resource
 from app.runtime.settings import get_runtime_setting
 from app.schemas.exception import PluginMutationRejectedError
-from app.schemas.plugin import PluginRuntimeStatus
+from app.schemas.plugin import PluginInstance, PluginRuntimeStatus
 from app.schemas.types import SystemConfigKey
 from app.startup.composition.plugin import (
     compose_plugin_market,
@@ -155,6 +160,59 @@ def _delete_plugin_data(plugin_id: str) -> None:
         session.close()
 
 
+def _plugin_instance_from_record(record: PluginInstanceRecord) -> PluginInstance:
+    """把插件实例表的 ORM 行投影为运行时端口使用的 Pydantic 描述。
+
+    只投影描述符各列：业务参数走 plugin.<实例ID> 配置读取口，运行时端口拿到的应当是
+    一份实例身份与展示信息的视图。
+    """
+    return PluginInstance(
+        instance_id=record.instance_id,
+        source_plugin_id=record.source_plugin_id,
+        plugin_name=record.plugin_name,
+        plugin_desc=record.plugin_desc,
+        plugin_icon=record.plugin_icon,
+    )
+
+
+def _save_plugin_instance_record(instance: PluginInstance) -> None:
+    """把运行时实例描述写入插件实例表，以实例 ID 为稳定键做新增或更新。
+
+    业务参数不在此列：它由插件自身通过 plugin.<实例ID> 配置口写入、不进运行时描述，
+    原样写回会把用户刚存的配置覆盖成空。
+    """
+    PluginInstanceOper().save(
+        instance_id=instance.instance_id,
+        source_plugin_id=instance.source_plugin_id,
+        plugin_name=instance.plugin_name,
+        plugin_desc=instance.plugin_desc,
+        plugin_icon=instance.plugin_icon,
+    )
+
+
+def _build_plugin_instance_directory() -> PluginInstanceDirectory:
+    """把插件实例表端口装配到 db 层实现。"""
+    oper = PluginInstanceOper()
+
+    def _get(instance_id: str) -> PluginInstance | None:
+        """按实例 ID 查询实例并投影为运行时描述。"""
+        record = oper.get(instance_id)
+        return _plugin_instance_from_record(record) if record is not None else None
+
+    return PluginInstanceDirectory(
+        get=_get,
+        list_all=lambda: [
+            _plugin_instance_from_record(record) for record in oper.list_all()
+        ],
+        list_by_source=lambda source_plugin_id: [
+            _plugin_instance_from_record(record)
+            for record in oper.list_by_source(source_plugin_id)
+        ],
+        save=_save_plugin_instance_record,
+        delete=oper.delete,
+    )
+
+
 def _build_plugin_database() -> PluginDatabase:
     """把插件自有数据库端口装配到 db 层的建库、释放与销毁实现。"""
     return PluginDatabase(
@@ -180,6 +238,7 @@ def build_plugin_runtime_graph(host: PluginRuntimeHost) -> PluginRuntime:
         PluginRuntimeEnvironment(
             plugins_root=Path(get_runtime_setting('ROOT_PATH')) / "app" / "plugins",
             storage=lambda: get_plugin_storage(),
+            instance_directory=lambda: get_plugin_instance_directory(),
             system=lambda: get_plugin_system(),
             database=lambda: get_plugin_database(),
             catalog_factory=lambda mapper: _build_plugin_catalog(mapper),
@@ -424,6 +483,7 @@ def configure_plugin_services() -> None:
         delete_data=_delete_plugin_data,
     ))
     configure_plugin_database(_build_plugin_database())
+    configure_plugin_instance_directory(_build_plugin_instance_directory())
 
 
 def _register_plugin_runtime(plugin_id: str) -> None:

database/versions/281965691a20_3_0_34.py

@@ -0,0 +1,144 @@
+"""3.0.34 插件实例描述符迁入独立表。
+
+Revision ID: 281965691a20
+Revises: b2d4f6a8c1e3
+Create Date: 2026-09-02
+"""
+
+from datetime import datetime, timezone
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "281965691a20"
+down_revision = "b2d4f6a8c1e3"
+branch_labels = None
+depends_on = None
+
+_TABLE = "plugininstance"
+_LEGACY_KEY = "PluginInstances"
+
+
+def _id_column(dialect_name: str) -> sa.Column:
+    """保持 PostgreSQL Identity 与 SQLite 整数主键的当前模型语义一致。"""
+    if dialect_name == "postgresql":
+        return sa.Column(
+            "id",
+            sa.Integer(),
+            sa.Identity(start=1, cycle=True),
+            nullable=False,
+        )
+    return sa.Column("id", sa.Integer(), nullable=False)
+
+
+def _legacy_entries(connection: sa.engine.Connection) -> list[dict]:
+    """读取旧 systemconfig 单键里的实例描述，兼容历史字典与列表两种载荷形态。
+
+    :param connection: 当前迁移事务连接
+    :return: 已补全 ``instance_id`` 的实例字典列表，损坏项直接跳过
+    """
+    systemconfig = sa.table(
+        "systemconfig",
+        sa.column("key", sa.String()),
+        sa.column("value", sa.JSON()),
+    )
+    row = connection.execute(
+        sa.select(systemconfig.c.value).where(systemconfig.c.key == _LEGACY_KEY)
+    ).first()
+    raw = row[0] if row else None
+    if isinstance(raw, dict):
+        entries = []
+        for instance_id, payload in raw.items():
+            if not isinstance(payload, dict) or not instance_id:
+                continue
+            merged = dict(payload)
+            merged.setdefault("instance_id", instance_id)
+            entries.append(merged)
+        return entries
+    if isinstance(raw, list):
+        return [
+            item
+            for item in raw
+            if isinstance(item, dict) and item.get("instance_id")
+        ]
+    return []
+
+
+def upgrade() -> None:
+    """建立插件实例描述符表，并把旧 systemconfig 单键的现有内容逐条搬入。
+
+    实例描述符原本挤在 ``PluginInstances`` 这一个 JSON 单键里：改任何一个分身都要
+    读出整份载荷再整份写回，条目数随分身数量增长，而表里没有任何一列说得出某一条
+    属于哪个源插件。独立成表之后一实例一行，归属由 ``source_plugin_id`` 显式表达。
+
+    本体与分身不设模式列，角色由 ``instance_id`` 是否等于 ``source_plugin_id`` 派生：
+    模式列会是这个等式的冗余副本，两者一旦失步，同一行就会在不同读取口被判成不同角色。
+
+    只搬迁，不删除原 systemconfig 键：原键留作回滚依据，运行期兜底导入也据此
+    在表为空而旧键非空时补一次导入。
+    """
+    inspector = sa.inspect(op.get_bind())
+    if _TABLE in inspector.get_table_names():
+        return
+    op.create_table(
+        _TABLE,
+        _id_column(op.get_bind().dialect.name),
+        sa.Column("instance_id", sa.String(length=128), nullable=False),
+        sa.Column("source_plugin_id", sa.String(length=128), nullable=False),
+        sa.Column("plugin_name", sa.String(length=255), nullable=True),
+        sa.Column("plugin_desc", sa.String(length=255), nullable=True),
+        sa.Column("plugin_icon", sa.String(length=255), nullable=True),
+        sa.Column("created_at", sa.String(length=40), nullable=False),
+        sa.Column("updated_at", sa.String(length=40), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("instance_id", name="uq_plugininstance_instance_id"),
+    )
+    op.create_index(
+        "ix_plugininstance_source_plugin_id",
+        _TABLE,
+        ["source_plugin_id"],
+    )
+
+    connection = op.get_bind()
+    entries = _legacy_entries(connection)
+    if not entries:
+        return
+    table = sa.table(
+        _TABLE,
+        sa.column("instance_id", sa.String()),
+        sa.column("source_plugin_id", sa.String()),
+        sa.column("plugin_name", sa.String()),
+        sa.column("plugin_desc", sa.String()),
+        sa.column("plugin_icon", sa.String()),
+        sa.column("created_at", sa.String()),
+        sa.column("updated_at", sa.String()),
+    )
+    now = datetime.now(timezone.utc).isoformat()
+    rows = []
+    seen_instance_ids: set[str] = set()
+    for entry in entries:
+        instance_id = entry.get("instance_id")
+        source_plugin_id = entry.get("source_plugin_id")
+        if not instance_id or not source_plugin_id or instance_id in seen_instance_ids:
+            continue
+        seen_instance_ids.add(instance_id)
+        rows.append({
+            "instance_id": instance_id,
+            "source_plugin_id": source_plugin_id,
+            "plugin_name": entry.get("plugin_name"),
+            "plugin_desc": entry.get("plugin_desc"),
+            "plugin_icon": entry.get("plugin_icon"),
+            "created_at": now,
+            "updated_at": now,
+        })
+    if rows:
+        connection.execute(table.insert(), rows)
+
+
+def downgrade() -> None:
+    """删除插件实例描述符表，不触碰原 systemconfig 单键。"""
+    inspector = sa.inspect(op.get_bind())
+    if _TABLE not in inspector.get_table_names():
+        return
+    op.drop_index("ix_plugininstance_source_plugin_id", table_name=_TABLE)
+    op.drop_table(_TABLE)

database/versions/c4e1a7b9d2f6_3_0_35.py

@@ -0,0 +1,140 @@
+"""3.0.35 插件配置迁入插件实例表。
+
+Revision ID: c4e1a7b9d2f6
+Revises: 281965691a20
+Create Date: 2026-09-11
+"""
+
+import json
+from datetime import datetime, timezone
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "c4e1a7b9d2f6"
+down_revision = "281965691a20"
+branch_labels = None
+depends_on = None
+
+_TABLE = "plugininstance"
+_LEGACY_PREFIX = "plugin."
+
+
+def _table_names(connection) -> set:
+    """读取当前数据库已有表名。"""
+    return set(sa.inspect(connection).get_table_names())
+
+
+def _column_names(connection) -> set:
+    """读取实例表已有列名；表不存在时为空集。"""
+    if _TABLE not in _table_names(connection):
+        return set()
+    return {column["name"] for column in sa.inspect(connection).get_columns(_TABLE)}
+
+
+def _decode(value):
+    """把 JSON 列还原为 Python 值，兼容驱动直接返回原始字符串的情形。"""
+    if isinstance(value, (str, bytes, bytearray)):
+        try:
+            return json.loads(value)
+        except (TypeError, ValueError):
+            return value
+    return value
+
+
+def upgrade() -> None:
+    """把每实例的业务参数收拢到它自己那一行上。
+
+    插件配置此前寄存在 systemconfig 的 plugin.<实例ID> 单键下：键由插件 ID 决定、
+    条目数随安装量增长，混在系统设置里会把主程序自己的设置项淹掉；而且那个键存的
+    其实是实例 ID，表里没有任何一列说得出它属于哪个插件，想列出某插件的全部实例
+    配置就只能靠字符串前缀去猜。
+
+    配置与实例身份同存一行，而不是另起一张同键的表：展示信息与业务参数都是这个
+    实例的设置，同一个生命周期，分表只会让「建分身、删分身」退化成两张表之间的
+    协调问题。
+    """
+    connection = op.get_bind()
+    now = datetime.now(timezone.utc).isoformat()
+    columns = _column_names(connection)
+    if not columns:
+        return
+
+    if "config_data" not in columns:
+        op.add_column(_TABLE, sa.Column("config_data", sa.JSON(), nullable=True))
+
+    if "systemconfig" not in _table_names(connection):
+        return
+
+    existing = {
+        row[0]
+        for row in connection.execute(
+            sa.text(f"SELECT instance_id FROM {_TABLE}")
+        ).fetchall()
+    }
+    legacy = connection.execute(
+        sa.text("SELECT key, value FROM systemconfig WHERE key LIKE :pattern"),
+        {"pattern": f"{_LEGACY_PREFIX}%"},
+    ).fetchall()
+    migrated_keys = []
+    for key, value in legacy:
+        instance_id = key[len(_LEGACY_PREFIX):]
+        if not instance_id:
+            continue
+        payload = json.dumps(_decode(value))
+        if instance_id in existing:
+            connection.execute(
+                sa.text(
+                    f"UPDATE {_TABLE} SET config_data = :config_data, updated_at = :now "
+                    "WHERE instance_id = :instance_id"
+                ),
+                {"config_data": payload, "instance_id": instance_id, "now": now},
+            )
+        else:
+            # 没有实例行的只可能是本体自身：分身的行在建分身时就已落盘
+            connection.execute(
+                sa.text(
+                    f"INSERT INTO {_TABLE} "
+                    "(instance_id, source_plugin_id, config_data, created_at, updated_at) "
+                    "VALUES (:instance_id, :instance_id, :config_data, :now, :now)"
+                ),
+                {"instance_id": instance_id, "config_data": payload, "now": now},
+            )
+            existing.add(instance_id)
+        migrated_keys.append(key)
+
+    # 同一份配置不能在两处各留一份，否则读取端按哪个都不对
+    for key in migrated_keys:
+        connection.execute(
+            sa.text("DELETE FROM systemconfig WHERE key = :key"),
+            {"key": key},
+        )
+
+
+def downgrade() -> None:
+    """把业务参数搬回 systemconfig 原键，并移除新增列，回退路径上不丢配置。"""
+    connection = op.get_bind()
+    columns = _column_names(connection)
+    if "config_data" not in columns:
+        return
+
+    if "systemconfig" in _table_names(connection):
+        rows = connection.execute(
+            sa.text(
+                f"SELECT instance_id, config_data FROM {_TABLE} WHERE config_data IS NOT NULL"
+            )
+        ).fetchall()
+        for instance_id, config_data in rows:
+            key = f"{_LEGACY_PREFIX}{instance_id}"
+            exists = connection.execute(
+                sa.text("SELECT 1 FROM systemconfig WHERE key = :key"),
+                {"key": key},
+            ).fetchone()
+            if not exists:
+                connection.execute(
+                    sa.text("INSERT INTO systemconfig (key, value) VALUES (:key, :value)"),
+                    {"key": key, "value": json.dumps(_decode(config_data))},
+                )
+
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.drop_column("config_data")

docs/architecture-overview.md

@@ -756,8 +756,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 1010 |
-| 内部导入边 | 8,586 |
+| Python 模块 | 1012 |
+| 内部导入边 | 8,598 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/refactor/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 1010 / 8,586 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与终端作用域模块的受控依赖 |
+| 宿主 Python 模块 / 内部依赖边 | 1012 / 8,598 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与终端作用域模块的受控依赖 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

docs/rules/10-data-and-persistent.md

@@ -20,6 +20,7 @@ Models are SQLAlchemy declarative classes. Each model maps to one database table
 | `Message` | Message log |
 | `PluginData` | Plugin-persisted data |
 | `PluginIdentity` | Installed physical-plugin source binding and payload provenance |
+| `PluginInstance` | Per-instance plugin descriptor and configuration, one row per instance |
 | `PassKey` | Passkey authentication records |
 | `Workflow` | Workflow definitions |
 
@@ -63,6 +64,7 @@ directly in chain, module, or endpoint code.
 | `MessageOper` | `oper/message.py` |
 | `PluginDataOper` | `oper/plugindata.py` |
 | `PluginIdentityOper` | `oper/pluginidentity.py` |
+| `PluginInstanceOper` | `oper/plugininstance.py` |
 | `SiteOper` | `oper/site.py` |
 | `SubscribeHistoryOper` | `oper/subscribehistory.py` |
 | `SubscribeOper` | `oper/subscribe.py` |
@@ -441,4 +443,4 @@ can be accepted only once without Application knowing the configured backend.
 - `settings.API_TOKEN` and other secret fields must not be included in log output or API responses.
 - The `config list --show-secrets` flag exists specifically to gate secret visibility in the CLI.
 
-*Last Updated: 2026-09-02*
+*Last Updated: 2026-09-12*

skills/database-operation/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: database-operation
-version: 7
+version: 8
 description: >-
   Use this skill when you need to inspect, query, maintain, or carefully modify
   the MoviePilot database. This skill uses the bundled scripts/mp-db.py helper,
@@ -206,6 +206,12 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123"
 - Write boundary: Owned by the plugin installation state machine; never advance phase or overwrite evidence manually.
 - Columns: `id`, `transaction_id`, `plugin_id`, `phase`, `membership_before`, `membership_target`, `identity_before_revision`, `identity_target_revision`, `package_existed`, `persistent_backup_existed`, `created_at`, `updated_at`, `schema_version`
 
+### `plugininstance`
+- Purpose: Stores one row per shared-source plugin runtime instance, covering both clones and the host plugin itself (instance_id equals source_plugin_id), together with that instance's display overrides and its own configuration payload. A clone exists exactly while its row exists, so deleting the row uninstalls the clone and discards its configuration.
+- Useful queries: Diagnosing clone naming and ownership, or inspecting what a plugin or one of its clones is configured with.
+- Write boundary: Owned by the plugin instance and plugin configuration APIs; never edit rows directly.
+- Columns: `id`, `instance_id`, `source_plugin_id`, `plugin_name`, `plugin_desc`, `plugin_icon`, `config_data`, `created_at`, `updated_at`
+
 ### `site`
 - Purpose: Stores private-tracker URLs, RSS, credentials, rate limits, proxy state, and downloader binding.
 - Useful queries: Inspecting enablement, domain, rate limits, or downloader binding with minimal credential exposure.

tests/conftest.py

@@ -219,7 +219,10 @@ def configure_plugin_system_services():
         PluginRuntimeEnvironment,
         build_plugin_runtime,
     )
-    from app.runtime.extensions.plugin.storage import get_plugin_storage
+    from app.runtime.extensions.plugin.storage import (
+        get_plugin_instance_directory,
+        get_plugin_storage,
+    )
     from app.runtime.extensions.plugin.system import get_plugin_system
     from app.runtime.extensions.service import ServiceConfigHelper
 
@@ -234,6 +237,7 @@ def configure_plugin_system_services():
             PluginRuntimeEnvironment(
                 plugins_root=settings.ROOT_PATH / "app" / "plugins",
                 storage=get_plugin_storage,
+                instance_directory=get_plugin_instance_directory,
                 system=get_plugin_system,
                 database=get_plugin_database,
                 catalog_factory=lambda mapper: (

tests/fixtures/architecture/dependency-baseline.json

@@ -1077,8 +1077,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8586,
-  "edge_sha256": "8351a247a58263a1dca5912b44c66734b0c4aa91c2fbfe8d51b566291f7dfee3",
+  "edge_count": 8598,
+  "edge_sha256": "980206679f2d82d37521d9e8798188d27b123ec1b085f621a9b281023b8cd0df",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -5872,6 +5872,8 @@
     "app.db.models.pluginidentity -> app.db.base",
     "app.db.models.plugininstallation -> app.db",
     "app.db.models.plugininstallation -> app.db.base",
+    "app.db.models.plugininstance -> app.db",
+    "app.db.models.plugininstance -> app.db.base",
     "app.db.models.site -> app.db",
     "app.db.models.site -> app.db.base",
     "app.db.models.siteicon -> app.db",
@@ -5969,6 +5971,10 @@
     "app.db.oper.pluginidentity -> app.db.base",
     "app.db.oper.pluginidentity -> app.db.models",
     "app.db.oper.pluginidentity -> app.db.models.pluginidentity",
+    "app.db.oper.plugininstance -> app.db",
+    "app.db.oper.plugininstance -> app.db.base",
+    "app.db.oper.plugininstance -> app.db.models",
+    "app.db.oper.plugininstance -> app.db.models.plugininstance",
     "app.db.oper.query -> app.db",
     "app.db.oper.query -> app.db.base",
     "app.db.oper.query -> app.schemas",
@@ -6013,7 +6019,10 @@
     "app.db.oper.systemconfig -> app.db",
     "app.db.oper.systemconfig -> app.db.base",
     "app.db.oper.systemconfig -> app.db.models",
+    "app.db.oper.systemconfig -> app.db.models.plugininstance",
     "app.db.oper.systemconfig -> app.db.models.systemconfig",
+    "app.db.oper.systemconfig -> app.db.oper",
+    "app.db.oper.systemconfig -> app.db.oper.plugininstance",
     "app.db.oper.systemconfig -> app.foundation",
     "app.db.oper.systemconfig -> app.foundation.singleton",
     "app.db.oper.systemconfig -> app.schemas",
@@ -9356,8 +9365,11 @@
     "app.startup.initializers.plugins -> app.application.scheduling",
     "app.startup.initializers.plugins -> app.application.site",
     "app.startup.initializers.plugins -> app.db",
+    "app.startup.initializers.plugins -> app.db.models",
+    "app.startup.initializers.plugins -> app.db.models.plugininstance",
     "app.startup.initializers.plugins -> app.db.oper",
     "app.startup.initializers.plugins -> app.db.oper.plugindata",
+    "app.startup.initializers.plugins -> app.db.oper.plugininstance",
     "app.startup.initializers.plugins -> app.db.plugin",
     "app.startup.initializers.plugins -> app.db.plugin.registry",
     "app.startup.initializers.plugins -> app.db.session",
@@ -9667,7 +9679,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 1010,
+  "module_count": 1012,
   "modules": [
     "app",
     "app.adapters",
@@ -10178,6 +10190,7 @@
     "app.db.models.plugindata",
     "app.db.models.pluginidentity",
     "app.db.models.plugininstallation",
+    "app.db.models.plugininstance",
     "app.db.models.site",
     "app.db.models.siteicon",
     "app.db.models.sitestatistic",
@@ -10204,6 +10217,7 @@
     "app.db.oper.passkey",
     "app.db.oper.plugindata",
     "app.db.oper.pluginidentity",
+    "app.db.oper.plugininstance",
     "app.db.oper.query",
     "app.db.oper.site",
     "app.db.oper.subscribe",

tests/fixtures/architecture/startup-performance-baseline.json

@@ -6,7 +6,7 @@
   "repeat": 3,
   "targets": {
     "app.startup.lifecycle": {
-      "loaded_app_module_count": 547,
+      "loaded_app_module_count": 549,
       "max_ms": 1293.338,
       "median_ms": 1156.239,
       "min_ms": 1102.806,
@@ -17,7 +17,7 @@
       ]
     },
     "app.factory": {
-      "loaded_app_module_count": 559,
+      "loaded_app_module_count": 561,
       "max_ms": 1127.911,
       "median_ms": 1122.382,
       "min_ms": 1119.221,
@@ -28,7 +28,7 @@
       ]
     },
     "app.main": {
-      "loaded_app_module_count": 561,
+      "loaded_app_module_count": 563,
       "max_ms": 1188.652,
       "median_ms": 1183.509,
       "min_ms": 1174.522,

tests/test_builtin_skill_boundaries.py

@@ -115,7 +115,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
     expected_versions = {
         "browser-use": "3",
         "command-dispatch": "2",
-        "database-operation": "7",
+        "database-operation": "8",
         "feedback-issue": "9",
         "moviepilot-api": "31",
         "moviepilot-update": "5",

tests/test_plugin_config_migration.py

@@ -0,0 +1,178 @@
+"""插件配置迁入插件实例表的 Alembic 迁移测试。"""
+
+from __future__ import annotations
+
+import importlib
+import json
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+MIGRATION_MODULE = "database.versions.c4e1a7b9d2f6_3_0_35"
+
+
+def _bind_migration(monkeypatch, connection):
+    """把迁移绑定到隔离数据库连接。"""
+    migration = importlib.import_module(MIGRATION_MODULE)
+    context = MigrationContext.configure(connection)
+    monkeypatch.setattr(migration, "op", Operations(context))
+    return migration
+
+
+def _create_legacy_schema(connection: sa.engine.Connection) -> None:
+    """建出迁移前的系统设置表与实例表，并写入两类数据各若干条。
+
+    实例表按上一条迁移 281965691a20 建出的形状，不含业务参数列。
+    """
+    sa.Table(
+        "systemconfig",
+        sa.MetaData(),
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("key", sa.String(), nullable=False),
+        sa.Column("value", sa.JSON()),
+    ).create(connection)
+    connection.execute(
+        sa.text("INSERT INTO systemconfig (key, value) VALUES (:k, :v)"),
+        [
+            {"k": "plugin.DemoPlugin", "v": json.dumps({"enable": True, "token": "keep"})},
+            {"k": "plugin.DemoPluginwork", "v": json.dumps({"enable": False})},
+            {"k": "UserInstalledPlugins", "v": json.dumps(["DemoPlugin", "BareInstalled"])},
+        ],
+    )
+
+    sa.Table(
+        "plugininstance",
+        sa.MetaData(),
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("instance_id", sa.String(128), nullable=False, unique=True),
+        sa.Column("source_plugin_id", sa.String(128), nullable=False),
+        sa.Column("plugin_name", sa.String(255)),
+        sa.Column("plugin_desc", sa.String(255)),
+        sa.Column("plugin_icon", sa.String(255)),
+        sa.Column("created_at", sa.String(40), nullable=False),
+        sa.Column("updated_at", sa.String(40), nullable=False),
+    ).create(connection)
+    connection.execute(
+        sa.text(
+            "INSERT INTO plugininstance "
+            "(instance_id, source_plugin_id, plugin_name, created_at, updated_at) "
+            "VALUES (:iid, :sid, :name, 'x', 'x')"
+        ),
+        [{"iid": "DemoPluginwork", "sid": "DemoPlugin", "name": "分身甲"}],
+    )
+
+
+def _instance_rows(connection: sa.engine.Connection) -> dict:
+    """读取实例表内容，兼容驱动把 JSON 列返回为字符串的情形。"""
+    rows = connection.execute(
+        sa.text(
+            "SELECT instance_id, source_plugin_id, plugin_name, config_data "
+            "FROM plugininstance"
+        )
+    ).fetchall()
+    return {
+        row[0]: {
+            "source_plugin_id": row[1],
+            "plugin_name": row[2],
+            "config_data": json.loads(row[3]) if isinstance(row[3], str) else row[3],
+        }
+        for row in rows
+    }
+
+
+def _system_keys(connection: sa.engine.Connection) -> set:
+    """读取系统设置表现存的键。"""
+    return {
+        row[0]
+        for row in connection.execute(sa.text("SELECT key FROM systemconfig")).fetchall()
+    }
+
+
+def _columns(connection: sa.engine.Connection) -> set:
+    """读取实例表当前列名。"""
+    return {column["name"] for column in sa.inspect(connection).get_columns("plugininstance")}
+
+
+def test_migration_moves_plugin_config_onto_the_instance_rows(monkeypatch):
+    """插件配置搬到实例行上，原行一并删除，真·系统设置留在原地。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.connect() as connection:
+        _create_legacy_schema(connection)
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+
+        rows = _instance_rows(connection)
+        assert rows["DemoPlugin"]["config_data"] == {"enable": True, "token": "keep"}
+        # 本体没有实例行时按自身建出，归属就是它自己
+        assert rows["DemoPlugin"]["source_plugin_id"] == "DemoPlugin"
+        # 分身已有实例行，配置并到那一行上而不是另起一行
+        assert rows["DemoPluginwork"]["config_data"] == {"enable": False}
+        assert rows["DemoPluginwork"]["source_plugin_id"] == "DemoPlugin"
+        # 同一份配置不能在两张表里各留一份，否则读取端按哪个都不对
+        assert _system_keys(connection) == {"UserInstalledPlugins"}
+
+
+def test_migration_keeps_the_display_fields_already_on_the_instance(monkeypatch):
+    """展示信息本就长在实例行上，补列搬配置时不得把它丢掉。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.connect() as connection:
+        _create_legacy_schema(connection)
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+
+        assert _instance_rows(connection)["DemoPluginwork"]["plugin_name"] == "分身甲"
+
+
+def test_migration_does_not_backfill_rows_for_plugins_without_config(monkeypatch):
+    """装了却从没存过配置的插件不该凭空多出一行：没有配置就没有要搬的东西。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.connect() as connection:
+        _create_legacy_schema(connection)
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+
+        assert set(_instance_rows(connection)) == {"DemoPlugin", "DemoPluginwork"}
+
+
+def test_migration_is_idempotent_on_repeated_upgrade(monkeypatch):
+    """重复升级不得因唯一键撞车而失败，也不得把配置复制成两份。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.connect() as connection:
+        _create_legacy_schema(connection)
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+        migration.upgrade()
+
+        rows = _instance_rows(connection)
+        assert len(rows) == 2
+        assert rows["DemoPlugin"]["config_data"] == {"enable": True, "token": "keep"}
+
+
+def test_downgrade_puts_plugin_config_back_into_system_settings(monkeypatch):
+    """回滚把配置搬回原键并移除新增列，回退路径上不丢配置。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.connect() as connection:
+        _create_legacy_schema(connection)
+        migration = _bind_migration(monkeypatch, connection)
+        migration.upgrade()
+
+        migration.downgrade()
+
+        assert _system_keys(connection) == {
+            "UserInstalledPlugins",
+            "plugin.DemoPlugin",
+            "plugin.DemoPluginwork",
+        }
+        assert "config_data" not in _columns(connection)
+        # 实例行本身留着：回滚只把配置搬走，不该顺手删掉分身
+        assert set(
+            row[0]
+            for row in connection.execute(
+                sa.text("SELECT instance_id FROM plugininstance")
+            ).fetchall()
+        ) == {"DemoPlugin", "DemoPluginwork"}

tests/test_plugin_config_table.py

@@ -0,0 +1,213 @@
+"""插件配置落在插件实例表上的存放与读写路由测试。"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.db.models.plugininstance import PluginInstance
+from app.db.models.systemconfig import SystemConfig
+from app.db.oper.plugininstance import PluginInstanceOper
+from app.db.oper.systemconfig import SystemConfigOper
+from app.db.session import SessionFactory
+
+PROBE_PLUGIN_ID = "PytestPluginConfigProbe"
+PROBE_KEY = f"plugin.{PROBE_PLUGIN_ID}"
+PROBE_CLONE_ID = f"{PROBE_PLUGIN_ID}work"
+PROBE_CLONE_KEY = f"plugin.{PROBE_CLONE_ID}"
+
+
+@pytest.fixture(name="system_config")
+def fixture_system_config():
+    """提供系统配置入口，并在用例结束后精确删除本次写入的行。
+
+    测试库全局共享且没有按用例重置的 fixture，残留会按文件名字母序污染后续用例。
+    """
+    oper = SystemConfigOper()
+    instances = PluginInstanceOper()
+    try:
+        yield oper
+    finally:
+        oper.delete(PROBE_KEY)
+        oper.delete(PROBE_CLONE_KEY)
+        instances.delete(PROBE_PLUGIN_ID)
+        instances.delete(PROBE_CLONE_ID)
+
+
+def _row_counts(instance_id: str, key: str) -> tuple[int, int]:
+    """返回该插件配置分别在实例表与系统设置表中的行数。"""
+    session = SessionFactory()
+    try:
+        in_instance_table = PluginInstance.get_by_instance_id(session, instance_id) is not None
+        in_system_table = SystemConfig.get_by_key(session, key) is not None
+        return int(in_instance_table), int(in_system_table)
+    finally:
+        session.close()
+
+
+def test_plugin_config_is_written_to_the_instance_table(system_config: SystemConfigOper):
+    """插件配置落进 plugininstance 表，不再占用主程序设置表的行。"""
+    system_config.set(PROBE_KEY, {"enable": True, "token": "probe"})
+
+    assert _row_counts(PROBE_PLUGIN_ID, PROBE_KEY) == (1, 0)
+
+
+def test_plugin_config_reads_back_through_the_same_key(system_config: SystemConfigOper):
+    """读取仍按 plugin.<ID> 键进行：这是插件侧的既有契约，不能因换表而改变。"""
+    system_config.set(PROBE_KEY, {"enable": False, "token": "probe"})
+
+    assert system_config.get(PROBE_KEY) == {"enable": False, "token": "probe"}
+
+
+def test_plugin_config_enters_the_snapshot_under_the_same_key(
+    system_config: SystemConfigOper,
+):
+    """重新加载快照后仍按 plugin.<ID> 键读得到，换表对读取端不可见。"""
+    system_config.set(PROBE_KEY, {"enable": True})
+
+    system_config.load_snapshot()
+
+    assert system_config.get(PROBE_KEY) == {"enable": True}
+
+
+def test_non_plugin_settings_still_live_in_system_config(system_config: SystemConfigOper):
+    """真·系统设置不受影响，路由只认 plugin. 前缀。"""
+    probe_key = "PytestPluginConfigProbeSystemKey"
+    try:
+        system_config.set(probe_key, {"kept": True})
+        session = SessionFactory()
+        try:
+            assert SystemConfig.get_by_key(session, probe_key) is not None
+            assert PluginInstance.get_by_instance_id(session, probe_key) is None
+        finally:
+            session.close()
+    finally:
+        system_config.delete(probe_key)
+
+
+def test_deleting_config_of_a_bare_host_reclaims_the_empty_row(
+    system_config: SystemConfigOper,
+):
+    """只承载一份配置的本体行在配置删掉后就是空壳，一并回收。"""
+    system_config.set(PROBE_KEY, {"enable": True})
+    assert _row_counts(PROBE_PLUGIN_ID, PROBE_KEY) == (1, 0)
+
+    system_config.delete(PROBE_KEY)
+
+    assert _row_counts(PROBE_PLUGIN_ID, PROBE_KEY) == (0, 0)
+    assert system_config.get(PROBE_KEY) is None
+
+
+def test_deleting_config_keeps_the_instance_that_still_carries_settings(
+    system_config: SystemConfigOper,
+):
+    """删一份配置不得把实例本身删掉：同一行还带着展示信息等设置。
+
+    配置与实例身份合表之后，删配置若照旧删行，「删一份配置」就会静默变成
+    「删掉这个实例」，展示信息一并消失。
+    """
+    instances = PluginInstanceOper()
+    instances.save(
+        instance_id=PROBE_PLUGIN_ID,
+        source_plugin_id=PROBE_PLUGIN_ID,
+        plugin_name="本体展示名",
+    )
+    system_config.set(PROBE_KEY, {"enable": True})
+
+    system_config.delete(PROBE_KEY)
+
+    survivor = instances.get(PROBE_PLUGIN_ID)
+    assert survivor is not None
+    assert survivor.plugin_name == "本体展示名"
+    assert survivor.config_data is None
+
+
+def test_deleting_a_clone_config_never_removes_the_clone_itself(
+    system_config: SystemConfigOper,
+):
+    """分身的存在由它那一行表达，清空配置不等于删掉这个分身。"""
+    instances = PluginInstanceOper()
+    instances.save(instance_id=PROBE_CLONE_ID, source_plugin_id=PROBE_PLUGIN_ID)
+    system_config.set(PROBE_CLONE_KEY, {"token": "probe"})
+
+    system_config.delete(PROBE_CLONE_KEY)
+
+    survivor = instances.get(PROBE_CLONE_ID)
+    assert survivor is not None
+    assert survivor.config_data is None
+
+
+def test_clone_config_records_its_source_plugin_so_instances_can_be_listed(
+    system_config: SystemConfigOper,
+):
+    """分身配置要记下它属于哪个源插件，否则只能靠字符串前缀去猜归属。
+
+    扁平键时代配置行只按实例 ID 命名，想列出「某插件的全部实例配置」无从下手。
+    """
+    instances = PluginInstanceOper()
+    instances.save(instance_id=PROBE_CLONE_ID, source_plugin_id=PROBE_PLUGIN_ID)
+
+    system_config.set(PROBE_KEY, {"host": True})
+    system_config.set(PROBE_CLONE_KEY, {"clone": True})
+
+    owned = instances.list_by_source(PROBE_PLUGIN_ID)
+    assert {record.instance_id for record in owned} == {PROBE_PLUGIN_ID, PROBE_CLONE_ID}
+    # 本体自身的 source_plugin_id 等于 instance_id；分身则指向源插件
+    clone_record = instances.get(PROBE_CLONE_ID)
+    assert clone_record.source_plugin_id == PROBE_PLUGIN_ID
+    assert clone_record.is_host is False
+
+
+def test_writing_config_does_not_clobber_the_display_fields_on_the_same_row(
+    system_config: SystemConfigOper,
+):
+    """插件自己存配置不得抹掉宿主写在同一行上的展示信息。"""
+    instances = PluginInstanceOper()
+    instances.save(
+        instance_id=PROBE_CLONE_ID,
+        source_plugin_id=PROBE_PLUGIN_ID,
+        plugin_name="分身甲",
+    )
+
+    system_config.set(PROBE_CLONE_KEY, {"enable": True})
+
+    record = instances.get(PROBE_CLONE_ID)
+    assert record.plugin_name == "分身甲"
+    assert record.config_data == {"enable": True}
+
+
+def test_descriptor_save_port_keeps_the_config_already_on_the_row(
+    system_config: SystemConfigOper,
+):
+    """组合根的实例落盘端口只写描述符各列，不得把同一行上的配置抹成空。
+
+    描述符视图里根本没有业务参数，若把它整份写回，用户刚存的配置会在下一次
+    改名或重载时静默消失。
+    """
+    from app.schemas.plugin import PluginInstance as PluginInstanceSchema
+    from app.startup.initializers.plugins import _save_plugin_instance_record
+
+    system_config.set(PROBE_CLONE_KEY, {"token": "keep"})
+
+    _save_plugin_instance_record(
+        PluginInstanceSchema(
+            instance_id=PROBE_CLONE_ID,
+            source_plugin_id=PROBE_CLONE_ID,
+            plugin_name="分身甲",
+        )
+    )
+
+    record = PluginInstanceOper().get(PROBE_CLONE_ID)
+    assert record is not None
+    assert record.plugin_name == "分身甲"
+    assert record.config_data == {"token": "keep"}
+
+
+def test_saving_an_instance_under_a_different_source_is_rejected(
+    system_config: SystemConfigOper,
+):
+    """实例归属是身份的一半，改写它等于偷换整行，持久化层直接拒绝。"""
+    instances = PluginInstanceOper()
+    instances.save(instance_id=PROBE_CLONE_ID, source_plugin_id=PROBE_PLUGIN_ID)
+
+    with pytest.raises(ValueError):
+        instances.save(instance_id=PROBE_CLONE_ID, source_plugin_id="PytestOtherSource")

tests/test_plugin_helper.py

@@ -856,7 +856,7 @@ class TestPluginHelper:
         _patch_catalog_settings(monkeypatch, VERSION_FLAG="v2")
         monkeypatch.setattr(
             "app.runtime.extensions.plugin.storage._plugin_storage",
-            SimpleNamespace(read=lambda _key: []),
+            SimpleNamespace(read=lambda _key: [], write=lambda _key, _value: None),
         )
         monkeypatch.setattr(
             "app.runtime.extensions.plugin.manager._site_auth_level_provider",
@@ -928,7 +928,7 @@ class TestPluginHelper:
         )
         monkeypatch.setattr(
             "app.runtime.extensions.plugin.storage._plugin_storage",
-            SimpleNamespace(read=lambda _key: []),
+            SimpleNamespace(read=lambda _key: [], write=lambda _key, _value: None),
         )
         monkeypatch.setattr(
             "app.runtime.extensions.plugin.manager._site_auth_level_provider",
@@ -1154,7 +1154,8 @@ class TestPluginHelper:
             SimpleNamespace(
                 read=lambda key: ["DemoPlugin"]
                 if key == SystemConfigKey.UserInstalledPlugins
-                else None
+                else None,
+                write=lambda _key, _value: None,
             ),
         )
 

tests/test_plugin_instance_migration.py

@@ -0,0 +1,219 @@
+"""插件实例描述符表 Alembic 迁移测试。"""
+
+from __future__ import annotations
+
+import importlib
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+from app.db.models.systemconfig import SystemConfig
+
+MIGRATION_MODULE = "database.versions.281965691a20_3_0_34"
+
+# 本迁移建出的列集合；业务参数列由后续 c4e1a7b9d2f6 迁移补上，这里不与当前完整模型
+# 比较，否则每次给该表加列都要回头改这条断言。
+EXPECTED_COLUMNS = {
+    "id",
+    "instance_id",
+    "source_plugin_id",
+    "plugin_name",
+    "plugin_desc",
+    "plugin_icon",
+    "created_at",
+    "updated_at",
+}
+
+
+def _bind_migration(monkeypatch, connection):
+    """把迁移绑定到隔离数据库连接。"""
+    migration = importlib.import_module(MIGRATION_MODULE)
+    context = MigrationContext.configure(connection)
+    monkeypatch.setattr(migration, "op", Operations(context))
+    return migration
+
+
+def _seed_legacy_key(connection: sa.engine.Connection, value) -> None:
+    """写入旧 systemconfig 单键，模拟迁移前的实例描述存量数据。"""
+    connection.execute(
+        sa.insert(SystemConfig.__table__).values(key="PluginInstances", value=value)
+    )
+
+
+def test_plugin_instance_migration_migrates_legacy_dict_payload_and_keeps_source_key(
+    monkeypatch,
+) -> None:
+    """字典载荷应逐条搬入新表，原 systemconfig 键保留不删，且可重复升级与完整回滚。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        SystemConfig.__table__.create(connection)
+        _seed_legacy_key(
+            connection,
+            {
+                "DemoPluginWork": {
+                    "instance_id": "DemoPluginWork",
+                    "source_plugin_id": "DemoPlugin",
+                    "plugin_name": "工作实例",
+                    "plugin_icon": "work.svg",
+                },
+            },
+        )
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+        migration.upgrade()
+
+        inspector = sa.inspect(connection)
+        assert "plugininstance" in inspector.get_table_names()
+        columns = {column["name"] for column in inspector.get_columns("plugininstance")}
+        assert columns == EXPECTED_COLUMNS
+        unique_constraints = {
+            constraint["name"]: tuple(constraint["column_names"])
+            for constraint in inspector.get_unique_constraints("plugininstance")
+        }
+        assert unique_constraints["uq_plugininstance_instance_id"] == ("instance_id",)
+        indexes = {index["name"] for index in inspector.get_indexes("plugininstance")}
+        assert "ix_plugininstance_source_plugin_id" in indexes
+
+        table = sa.Table("plugininstance", sa.MetaData(), autoload_with=connection)
+        rows = connection.execute(sa.select(table)).mappings().all()
+        assert len(rows) == 1
+        row = rows[0]
+        assert row["instance_id"] == "DemoPluginWork"
+        assert row["source_plugin_id"] == "DemoPlugin"
+        assert row["plugin_name"] == "工作实例"
+        assert row["plugin_icon"] == "work.svg"
+
+        legacy_row = connection.execute(
+            sa.select(SystemConfig.value).where(SystemConfig.key == "PluginInstances")
+        ).scalar_one()
+        assert legacy_row == {
+            "DemoPluginWork": {
+                "instance_id": "DemoPluginWork",
+                "source_plugin_id": "DemoPlugin",
+                "plugin_name": "工作实例",
+                "plugin_icon": "work.svg",
+            },
+        }
+
+        migration.downgrade()
+        assert "plugininstance" not in sa.inspect(connection).get_table_names()
+        assert connection.execute(
+            sa.select(SystemConfig.value).where(SystemConfig.key == "PluginInstances")
+        ).scalar_one() == legacy_row
+
+        migration.upgrade()
+        restored = sa.Table("plugininstance", sa.MetaData(), autoload_with=connection)
+        restored_rows = connection.execute(sa.select(restored)).mappings().all()
+        assert len(restored_rows) == 1
+        assert restored_rows[0]["instance_id"] == "DemoPluginWork"
+
+
+def test_plugin_instance_migration_drops_the_redundant_mode_field(monkeypatch) -> None:
+    """旧载荷里的 mode 是身份等式的冗余副本，新表不再为它建列。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        SystemConfig.__table__.create(connection)
+        _seed_legacy_key(
+            connection,
+            [
+                {
+                    "instance_id": "DemoPluginWork",
+                    "source_plugin_id": "DemoPlugin",
+                    "mode": "virtual",
+                },
+            ],
+        )
+        migration = _bind_migration(monkeypatch, connection)
+        migration.upgrade()
+
+        columns = {
+            column["name"]
+            for column in sa.inspect(connection).get_columns("plugininstance")
+        }
+        assert "mode" not in columns
+        check_constraints = {
+            constraint["name"]
+            for constraint in sa.inspect(connection).get_check_constraints("plugininstance")
+        }
+        assert "ck_plugininstance_mode" not in check_constraints
+
+
+def test_plugin_instance_migration_migrates_legacy_list_payload(monkeypatch) -> None:
+    """历史列表载荷同样应逐条搬入新表。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        SystemConfig.__table__.create(connection)
+        _seed_legacy_key(
+            connection,
+            [
+                {"instance_id": "DemoPluginWork", "source_plugin_id": "DemoPlugin"},
+                {"instance_id": "DemoPluginBackup", "source_plugin_id": "DemoPlugin"},
+            ],
+        )
+        migration = _bind_migration(monkeypatch, connection)
+        migration.upgrade()
+
+        table = sa.Table("plugininstance", sa.MetaData(), autoload_with=connection)
+        instance_ids = {
+            row["instance_id"]
+            for row in connection.execute(sa.select(table)).mappings().all()
+        }
+        assert instance_ids == {"DemoPluginWork", "DemoPluginBackup"}
+
+
+def test_plugin_instance_migration_skips_malformed_legacy_entries(monkeypatch) -> None:
+    """缺失必填字段或非字典条目必须被跳过，不得中断迁移。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        SystemConfig.__table__.create(connection)
+        _seed_legacy_key(
+            connection,
+            [
+                {"instance_id": "MissingSource"},
+                {"source_plugin_id": "DemoPlugin"},
+                "not-a-dict",
+                {"instance_id": "DemoPluginWork", "source_plugin_id": "DemoPlugin"},
+            ],
+        )
+        migration = _bind_migration(monkeypatch, connection)
+        migration.upgrade()
+
+        table = sa.Table("plugininstance", sa.MetaData(), autoload_with=connection)
+        rows = connection.execute(sa.select(table)).mappings().all()
+        assert [row["instance_id"] for row in rows] == ["DemoPluginWork"]
+
+
+def test_plugin_instance_migration_without_legacy_key_creates_empty_table(
+    monkeypatch,
+) -> None:
+    """旧键缺失或为空时应正常建表，不产生任何数据行。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        SystemConfig.__table__.create(connection)
+        migration = _bind_migration(monkeypatch, connection)
+        migration.upgrade()
+
+        assert "plugininstance" in sa.inspect(connection).get_table_names()
+        table = sa.Table("plugininstance", sa.MetaData(), autoload_with=connection)
+        assert connection.execute(sa.select(table)).mappings().all() == []
+
+
+def test_plugin_instance_migration_accepts_fresh_current_schema(monkeypatch) -> None:
+    """create_all 已建当前表时重复升级不得创建冲突对象。"""
+    from app.db.models.plugininstance import PluginInstance
+
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        SystemConfig.__table__.create(connection)
+        PluginInstance.__table__.create(connection)
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+        migration.upgrade()
+
+        assert {
+            column["name"]
+            for column in sa.inspect(connection).get_columns("plugininstance")
+        } == {column.name for column in PluginInstance.__table__.columns}

tests/test_plugin_instance_oper_listing.py

@@ -0,0 +1,74 @@
+"""插件实例列举在独占事务下的结果物化测试。"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import delete
+from sqlalchemy.orm import Session
+
+from app.db.models.plugininstance import PluginInstance
+from app.db.oper.plugininstance import PluginInstanceOper
+from app.db.uow import run_sync_transaction
+
+_INSTANCE_IDS = ("DemoPlugin", "DemoPlugin@clone1", "OtherPlugin@clone1")
+
+
+@pytest.fixture(autouse=True)
+def purge_written_rows():
+    """删除用例写入的行，测试库在整个会话内共享，残留会干扰后续用例。"""
+    yield
+
+    def purge(session: Session) -> None:
+        """按本文件使用的固定标识精确删除，不触碰其他用例的数据。"""
+        session.execute(
+            delete(PluginInstance).where(
+                PluginInstance.instance_id.in_(_INSTANCE_IDS)
+            )
+        )
+
+    run_sync_transaction(purge)
+
+
+def test_list_all_materializes_rows_before_session_close():
+    """表中有记录时 list_all 必须返回已物化的实例，而不是关闭会话后的游标。"""
+    oper = PluginInstanceOper()
+    oper.save(instance_id="DemoPlugin", source_plugin_id="DemoPlugin")
+    oper.save(instance_id="DemoPlugin@clone1", source_plugin_id="DemoPlugin")
+
+    records = oper.list_all()
+
+    assert {record.instance_id for record in records} >= {
+        "DemoPlugin",
+        "DemoPlugin@clone1",
+    }
+
+
+def test_list_by_source_materializes_rows_before_session_close():
+    """按源插件列举同样要在会话关闭前取完行。"""
+    oper = PluginInstanceOper()
+    oper.save(instance_id="OtherPlugin@clone1", source_plugin_id="OtherPlugin")
+
+    records = oper.list_by_source("OtherPlugin")
+
+    assert [record.instance_id for record in records] == ["OtherPlugin@clone1"]
+
+
+def test_save_rejects_flipping_an_existing_row_between_virtual_and_host():
+    """分身与本体互不转换，改写已存在行的归属必须在持久化层被拒绝。
+
+    角色由 instance_id 是否等于 source_plugin_id 派生，把归属改成实例 ID 自身正是
+    把一个分身就地变成本体，只可能来自调用方错把分身实例 ID 当作源插件 ID 使用。
+    """
+    oper = PluginInstanceOper()
+    oper.save(instance_id="DemoPlugin@clone1", source_plugin_id="DemoPlugin")
+
+    with pytest.raises(ValueError):
+        oper.save(
+            instance_id="DemoPlugin@clone1",
+            source_plugin_id="DemoPlugin@clone1",
+        )
+
+    preserved = oper.get("DemoPlugin@clone1")
+    assert preserved is not None
+    assert preserved.is_host is False
+    assert preserved.source_plugin_id == "DemoPlugin"

tests/test_plugin_local_sync.py

@@ -163,6 +163,7 @@ def _set_installed_plugins(monkeypatch, plugin_ids: list[str]) -> None:
         read=lambda key: plugin_ids
         if key == SystemConfigKey.UserInstalledPlugins
         else None,
+        write=lambda _key, _value: None,
     )
     monkeypatch.setattr(
         "app.runtime.extensions.plugin.storage._plugin_storage",

tests/test_plugin_virtual_instances.py

@@ -3,13 +3,40 @@
 import sys
 from types import ModuleType, SimpleNamespace
 
+import pytest
+
 from app.runtime.extensions.plugin.clone import PluginCloneService
 from app.runtime.extensions.plugin.loader import PluginLoader
-from app.runtime.extensions.plugin.storage import PluginInstanceStore, PluginStorage
+from app.runtime.extensions.plugin.storage import (
+    PluginInstanceDirectory,
+    PluginInstanceStore,
+    PluginStorage,
+)
 from app.schemas.plugin import PluginInstance, PluginRuntimeStatus
 from app.schemas.types import SystemConfigKey
 
 
+def _make_directory() -> tuple[PluginInstanceDirectory, dict[str, PluginInstance]]:
+    """构造进程内插件实例表，供分身持久化测试使用。
+
+    返回背后的字典，用例据此断言写入落在哪一行上，而不是只看端口回报的结果。
+    """
+    records: dict[str, PluginInstance] = {}
+
+    directory = PluginInstanceDirectory(
+        get=records.get,
+        list_all=lambda: list(records.values()),
+        list_by_source=lambda source_plugin_id: [
+            record
+            for record in records.values()
+            if record.source_plugin_id == source_plugin_id
+        ],
+        save=lambda instance: records.__setitem__(instance.instance_id, instance),
+        delete=lambda instance_id: records.pop(instance_id, None) is not None,
+    )
+    return directory, records
+
+
 def _logger() -> SimpleNamespace:
     """提供加载器测试所需的最小日志对象。"""
     return SimpleNamespace(
@@ -27,7 +54,8 @@ def test_instance_store_keeps_virtual_instances_out_of_installed_list():
         read=values.get,
         write=lambda key, value: values.__setitem__(key, value),
     )
-    store = PluginInstanceStore(storage=lambda: storage)
+    directory, _records = _make_directory()
+    store = PluginInstanceStore(storage=lambda: storage, directory=lambda: directory)
 
     instance = PluginInstance(
         instance_id="DemoPluginWork",
@@ -43,6 +71,87 @@ def test_instance_store_keeps_virtual_instances_out_of_installed_list():
     assert store.all() == {}
 
 
+def test_host_rows_do_not_leak_into_the_clone_listing():
+    """本体自身那一行与分身共用一张表，但不得出现在分身清单里。
+
+    本体行承载的是插件自己的业务参数，混进分身清单会让「我的插件」里多出一张
+    指向插件自身的分身卡片。
+    """
+    values = {SystemConfigKey.UserInstalledPlugins: ["DemoPlugin"]}
+    storage = PluginStorage(
+        read=values.get,
+        write=lambda key, value: values.__setitem__(key, value),
+    )
+    directory, records = _make_directory()
+    store = PluginInstanceStore(storage=lambda: storage, directory=lambda: directory)
+    clone = PluginInstance(instance_id="DemoPluginWork", source_plugin_id="DemoPlugin")
+    store.save(clone)
+    records["DemoPlugin"] = PluginInstance(
+        instance_id="DemoPlugin",
+        source_plugin_id="DemoPlugin",
+    )
+
+    assert store.all() == {"DemoPluginWork": clone}
+    assert store.for_source("DemoPlugin") == [clone]
+    assert store.get("DemoPlugin") is None
+    assert store.delete("DemoPlugin") is False
+    assert "DemoPlugin" in records
+
+
+def test_saving_a_clone_whose_id_equals_its_source_is_rejected():
+    """两者相等的那一行表示本体自身，按分身写入会把本体承载的配置顶掉。"""
+    values: dict = {}
+    storage = PluginStorage(
+        read=values.get,
+        write=lambda key, value: values.__setitem__(key, value),
+    )
+    directory, records = _make_directory()
+    store = PluginInstanceStore(storage=lambda: storage, directory=lambda: directory)
+
+    with pytest.raises(ValueError):
+        store.save(
+            PluginInstance(instance_id="DemoPlugin", source_plugin_id="DemoPlugin")
+        )
+
+    assert records == {}
+
+
+def test_legacy_instances_are_not_reimported_after_being_deleted():
+    """删光分身后重启不得从旧 systemconfig 键把它们导回来。
+
+    旧键刻意保留作回滚依据、从不清理，若以「表为空」作为兜底导入的判据，
+    用户每删光一次分身、下次启动就会复活一次。
+    """
+    values = {
+        SystemConfigKey.UserInstalledPlugins: ["DemoPlugin"],
+        SystemConfigKey.PluginInstances: {
+            "DemoPluginWork": {
+                "instance_id": "DemoPluginWork",
+                "source_plugin_id": "DemoPlugin",
+            }
+        },
+    }
+    storage = PluginStorage(
+        read=values.get,
+        write=lambda key, value: values.__setitem__(key, value),
+    )
+    directory, _records = _make_directory()
+
+    # 首次访问：旧键内容被导入独立表
+    first = PluginInstanceStore(storage=lambda: storage, directory=lambda: directory)
+    assert set(first.all()) == {"DemoPluginWork"}
+
+    # 用户删光全部分身，表重新变空
+    assert first.delete("DemoPluginWork") is True
+    assert first.all() == {}
+
+    # 进程重启：新建 store 重新走一次兜底导入判定
+    second = PluginInstanceStore(storage=lambda: storage, directory=lambda: directory)
+
+    assert second.all() == {}
+    assert values[SystemConfigKey.PluginInstances] is not None
+
+
 def test_loader_executes_each_instance_in_an_isolated_module_namespace(
     tmp_path,
     monkeypatch,