fix(plugin): 卸载先停后删、销毁全程持锁，PostgreSQL 会话按插件 schema 解析原生 SQL

卸载编排改为先停插件再删数据：插件的停机钩子取一次自有库句柄就能把刚删除的数据重新
建出来，注册表因此不再需要销毁墓碑，release 恢复为「只释放连接、保留数据」的单一语义。
停止同时注销插件类，其后的删除一律按 force 执行。

destroy_database 的摘句柄、关连接与删载体全部收进同一把锁：原先在锁外删除载体，并发的
get_database 能在间隙注册一个指向同一载体的新句柄，随后的删除会把它一并抹掉。

PostgreSQL 句柄在派生引擎上注册 begin 监听器，按事务执行 SET LOCAL search_path，让
schema_translate_map 覆盖不到的原生 SQL 同样解析到插件自己的 schema；监听器只挂在
OptionEngine 上，宿主引擎与其它插件的连接不受影响。

app/api/endpoints/plugin.py

@@ -979,14 +979,15 @@ def uninstall_plugin(plugin_id: str, _: ApiPrincipal = Depends(get_current_activ
             remove_plugin_job(plugin_id)
             # 判断是否为分身
             plugin_class = plugin_manager.plugins.get(plugin_id)
+            # 删除必须晚于停止：停机钩子会重建刚删的自有库；停止同时注销插件类，故删除一律按 force
+            plugin_manager.stop(plugin_id)
             if virtual_instance:
                 plugin_manager.delete_plugin_config(plugin_id, force=True)
                 plugin_manager.delete_plugin_data(plugin_id, force=True)
                 plugin_manager.delete_plugin_instance(plugin_id)
             elif getattr(plugin_class, "is_clone", False):
-                # 如果是分身插件，则删除分身数据和配置
-                plugin_manager.delete_plugin_config(plugin_id)
-                plugin_manager.delete_plugin_data(plugin_id)
+                plugin_manager.delete_plugin_config(plugin_id, force=True)
+                plugin_manager.delete_plugin_data(plugin_id, force=True)
                 # 分身物理目录只能由包文件 owner 删除。
                 if plugin_manager.remove_plugin_package(plugin_id):
                     plugin_manager.plugins.pop(plugin_id, None)

app/application/plugin/management.py

@@ -331,13 +331,17 @@ async def uninstall_plugin_runtime(plugin_id: str) -> dict[str, Any]:
         was_clone = bool(getattr(plugin_class, "is_clone", False))
         clone_files_removed = False
 
+        # 删除数据必须发生在插件停止之后：插件的停机钩子只要取一次自有库句柄，就会把刚
+        # 删除的数据重新建出来。停止同时注销插件类，故其后的删除一律按 force 执行
+        plugin_manager.stop(plugin_id)
+
         if virtual_instance:
             plugin_manager.delete_plugin_config(plugin_id, force=True)
             plugin_manager.delete_plugin_data(plugin_id, force=True)
             plugin_manager.delete_plugin_instance(plugin_id)
         elif was_clone:
-            plugin_manager.delete_plugin_config(plugin_id)
-            plugin_manager.delete_plugin_data(plugin_id)
+            plugin_manager.delete_plugin_config(plugin_id, force=True)
+            plugin_manager.delete_plugin_data(plugin_id, force=True)
             try:
                 clone_files_removed = await to_thread.run_sync(
                     plugin_manager.remove_plugin_package,

app/db/plugin/container.py

@@ -18,7 +18,9 @@ class PluginDatabaseHandle:
 
     SQLite 下引擎由本句柄独占，``owns_engine`` 为真；PostgreSQL 下引擎是宿主引擎按
     ``schema_translate_map`` 派生的外观，``owns_engine`` 为假——只有前者可以 dispose，
-    后者一旦 dispose 会连累宿主与其它插件仍在使用的同一个连接池。
+    后者一旦 dispose 会连累宿主与其它插件仍在使用的同一个连接池。PostgreSQL 下本句柄的
+    会话与连接在每个事务开始时把 ``search_path`` 限定到插件 schema，未限定的原生 SQL 因此
+    同样解析到插件自己的表。
     """
 
     plugin_id: str

app/db/plugin/registry.py

@@ -3,10 +3,11 @@
 from __future__ import annotations
 
 import threading
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from pathlib import Path
 
-from sqlalchemy import text
+from sqlalchemy import event, text
+from sqlalchemy.engine import Connection
 from sqlalchemy.orm import scoped_session, sessionmaker
 
 from app.db.engine import apply_sqlite_journal_mode, build_sqlite_engine, get_engine
@@ -28,12 +29,11 @@ __all__ = [
 ]
 
 # 插件的启动、停止与热重载分别来自调度线程、文件监控线程与 HTTP 线程，句柄的建立与
-# 释放必须串行，否则同一插件会同时存在两个各持一份连接池的引擎
+# 释放必须串行，否则同一插件会同时存在两个各持一份连接池的引擎。销毁的关连接与删载体
+# 同样在锁内完成：只要在删载体前放锁，并发的 get_database 就能注册一个指向同一载体的
+# 新句柄，紧接着的删除会把这个新句柄的库连同数据一起抹掉
 _lock = threading.RLock()
 _handles: dict[str, PluginDatabaseHandle] = {}
-# 宿主的卸载编排先删插件数据、后停插件，插件的停机钩子取一次句柄就能把刚销毁的库重新
-# 建出来。销毁在此留下标记：其后的释放改为再次销毁，重新建库（ensure）则撤销标记
-_destroyed: set[str] = set()
 
 
 def _is_postgresql() -> bool:
@@ -42,12 +42,32 @@ def _is_postgresql() -> bool:
     return db_type.lower() == "postgresql"
 
 
+def _search_path_setter(schema: str) -> Callable[[Connection], None]:
+    """
+    构造把事务内未限定标识符解析到插件 schema 的 ``begin`` 监听器。
+
+    ``schema_translate_map`` 只改写 SQLAlchemy 生成的 schema 感知语句，``text()`` 一类
+    的原生 SQL 不在其列；不动 ``search_path``，插件按合同写的未限定原生 SQL 会落到
+    ``public``。``SET LOCAL`` 的作用域随事务结束，不会经连接池把插件的解析根泄漏给宿主
+    或其它插件。
+    :param schema: 插件 schema 名
+    :return: 绑定该 schema 的 ``begin`` 事件监听器
+    """
+
+    def _apply_search_path(connection: Connection) -> None:
+        """在事务开始时把该连接的未限定标识符解析根切到插件 schema。"""
+        connection.exec_driver_sql(f'SET LOCAL search_path TO "{schema}", public')
+
+    return _apply_search_path
+
+
 def _build_handle(plugin_id: str) -> PluginDatabaseHandle:
     """
     构造插件的数据库句柄。
 
     SQLite 每插件一个独立库文件，句柄独占引擎；PostgreSQL 复用宿主引擎并按
-    ``schema_translate_map`` 派生出限定单一 schema 的外观，句柄不拥有该引擎。
+    ``schema_translate_map`` 派生出限定单一 schema 的外观，句柄不拥有该引擎，并在派生
+    引擎上按事务限定 ``search_path``，让未限定的原生 SQL 同样落在插件自己的 schema。
     :param plugin_id: 插件标识
     :return: 新建的数据库句柄
     """
@@ -57,6 +77,9 @@ def _build_handle(plugin_id: str) -> PluginDatabaseHandle:
         with host_engine.begin() as connection:
             connection.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"'))
         engine = host_engine.execution_options(schema_translate_map={None: schema})
+        # 监听器只挂在派生引擎上：OptionEngine 自带 dispatch，仅单向 _join 宿主引擎已有
+        # 的监听器，宿主与其它插件的连接不会因此改变解析根
+        event.listen(engine, "begin", _search_path_setter(schema))
         db_path = None
         owns_engine = False
     else:
@@ -102,7 +125,9 @@ def get_database(plugin_id: str) -> PluginDatabaseHandle:
     """
     取插件的数据库句柄，句柄不存在时按需建立。
 
-    不要求插件先声明模型：只打算执行原生 SQL 的插件同样可以直接取会话。
+    不要求插件先声明模型：只打算执行原生 SQL 的插件同样可以直接取会话。PostgreSQL 下
+    句柄的会话与连接在每个事务开始时把 ``search_path`` 限定到插件 schema，未限定的原生
+    SQL 因此同样解析到插件自己的表。
     :param plugin_id: 插件标识
     :return: 数据库句柄
     """
@@ -146,8 +171,6 @@ def ensure_database(
     :param migrations: 插件声明的 Alembic 迁移目录
     :raise FileNotFoundError: 声明的迁移目录不存在
     """
-    with _lock:
-        _destroyed.discard(plugin_id)
     if migrations is not None:
         # 目录校验必须早于建句柄：alembic 找不到 script_location 时抛错，而句柄已经把
         # 库文件建了出来，插件下次启动面对的是一个既没有表、也没有版本号的空库
@@ -169,24 +192,20 @@ def release_database(plugin_id: str) -> None:
     释放插件的数据库连接，保留全部数据。
 
     只 dispose 句柄自己拥有的引擎；PostgreSQL 下句柄只是宿主引擎的外观，这里不会、
-    也不能触碰宿主连接池。插件的数据库已被销毁、却又在停机钩子里重新取过句柄时，本次
-    释放改为再次销毁：那份重新建出来的库不属于任何仍然存在的插件，此后再无人回收。
+    也不能触碰宿主连接池。关连接与摘句柄在同一把锁内完成，句柄一旦离开注册表就不会再
+    被任何线程取到一个正在关闭的连接池。
     :param plugin_id: 插件标识
     """
     with _lock:
         handle = _handles.pop(plugin_id, None)
-        was_destroyed = plugin_id in _destroyed
-        _destroyed.discard(plugin_id)
         if handle is not None:
             _close_handle(handle)
-    if was_destroyed:
-        _remove_storage(plugin_id, handle)
 
 
 def release_all_databases() -> None:
     """释放全部插件的数据库连接，供进程关停使用。"""
     with _lock:
-        plugin_ids = list(dict.fromkeys((*_handles, *_destroyed)))
+        plugin_ids = list(_handles)
     for plugin_id in plugin_ids:
         try:
             release_database(plugin_id)
@@ -232,14 +251,14 @@ def destroy_database(plugin_id: str) -> None:
     """
     销毁插件的数据库：SQLite 删除库文件与 -wal/-shm 边车，PostgreSQL 丢弃对应 schema。
 
-    不可逆，调用方必须确认处在「删除插件数据」而非「停止插件」的路径上。失败只记日志：
-    删除数据是一次收尾操作，把文件系统或数据库的清理故障升级成异常，只会让已经删掉的
-    宿主数据与仍然存在的插件库停在不一致的中间态。
+    不可逆，调用方必须确认处在「删除插件数据」而非「停止插件」的路径上。摘句柄、关连接
+    与删载体全程持同一把锁，销毁与同一插件的句柄重建因此互斥，删除不会落到并发建出来的
+    新句柄头上。失败只记日志：删除数据是一次收尾操作，把文件系统或数据库的清理故障升级
+    成异常，只会让已经删掉的宿主数据与仍然存在的插件库停在不一致的中间态。
     :param plugin_id: 插件标识
     """
     with _lock:
         handle = _handles.pop(plugin_id, None)
-        _destroyed.add(plugin_id)
         if handle is not None:
             _close_handle(handle)
         _remove_storage(plugin_id, handle)

app/plugins/__init__.py

@@ -306,6 +306,8 @@ class _PluginBase(metaclass=ABCMeta):
         """
         获取插件自有数据库句柄，用于取会话读写插件自有表
         句柄不存在时按需建立，不要求先声明模型
+        PostgreSQL 下句柄的会话与连接在每个事务开始时把 search_path 限定到本插件 schema，
+        未限定的原生 SQL 因此同样解析到插件自己的表，不会落到 public
         :param plugin_id: 插件ID
         """
         if not plugin_id:

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

@@ -272,7 +272,11 @@ runtime, `_PluginBase.get_database()` returns a `PluginDatabaseHandle` for
 opening sessions against the plugin's own engine. Declaring nothing is not the
 same as never having a database: `get_database()` creates the SQLite file (or
 the PostgreSQL schema) on first call, so a plugin that only runs raw SQL still
-gets an isolated database.
+gets an isolated database. Under PostgreSQL the handle's sessions and
+connections issue `SET LOCAL search_path` at the start of every transaction,
+so unqualified raw SQL resolves to the plugin's own schema rather than
+`public`; `SET LOCAL` ends with the transaction and never leaks back to the
+host through the shared pool.
 
 Lifecycle is strictly ensure/release/destroy: plugin start calls `ensure`
 after `init_plugin()`; stop, reload and remove call `release` only, which
@@ -286,11 +290,10 @@ SCHEMA ... CASCADE`). Stopping or uninstalling an ordinary plugin never
 destroys its database, mirroring the existing `plugindata` retention
 semantics.
 
-Uninstall destroys the database before the plugin is stopped, so a stop hook
-calling `get_database()` recreates it. A destroyed plugin id is remembered
-until its next `release` — which destroys the recreated database instead of
-merely releasing it — or its next `ensure`, which clears the mark because the
-plugin is legitimately starting again.
+Uninstall stops the plugin before deleting anything: a stop hook that calls
+`get_database()` would otherwise recreate the database that was just
+destroyed. Because stopping also unregisters the plugin class, the deletions
+that follow are issued with `force=True`.
 
 `db_query` / `db_update` keep their existing automatic-Session fallback bound
 to the **host** `ScopedSession()`; they are not aware of plugin-owned
@@ -423,4 +426,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-01*
+*Last Updated: 2026-09-02*

tests/test_db_plugin_framework.py

@@ -5,12 +5,13 @@ from __future__ import annotations
 import asyncio
 import threading
 from pathlib import Path
+from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
+from sqlalchemy import create_engine, event, text
 from sqlalchemy import inspect as sa_inspect
-from sqlalchemy import text
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, scoped_session, sessionmaker
 
 import app.db.engine as engine_module
 import app.db.plugin.locator as locator_module
@@ -29,18 +30,14 @@ from app.db.plugin.locator import (
 
 @pytest.fixture(autouse=True)
 def _isolate_plugin_databases():
-    """快照插件数据库句柄与销毁标记，用例结束后释放残留句柄并还原快照。"""
+    """快照插件数据库句柄，用例结束后释放残留句柄并还原快照。"""
     handles = dict(registry_module._handles)
-    destroyed = set(registry_module._destroyed)
     registry_module._handles.clear()
-    registry_module._destroyed.clear()
     yield
     for plugin_id in list(registry_module._handles):
         registry_module.release_database(plugin_id)
     registry_module._handles.clear()
     registry_module._handles.update(handles)
-    registry_module._destroyed.clear()
-    registry_module._destroyed.update(destroyed)
 
 
 def _raise_dispose() -> None:
@@ -48,6 +45,24 @@ def _raise_dispose() -> None:
     raise RuntimeError("dispose failed")
 
 
+def _borrowed_engine_handle(engine: Any) -> PluginDatabaseHandle:
+    """
+    构造一个不拥有引擎的句柄，用于验证 PostgreSQL 分支上的连接路由。
+    :param engine: 句柄借用的引擎
+    :return: owns_engine 为假的数据库句柄
+    """
+    session_factory = sessionmaker(bind=engine)
+    return PluginDatabaseHandle(
+        plugin_id="demo",
+        engine=engine,
+        session_factory=session_factory,
+        scoped_session_factory=scoped_session(session_factory),
+        db_path=None,
+        schema="plugin_demo",
+        owns_engine=False,
+    )
+
+
 def _write_migration_directory(root: Path) -> Path:
     """
     写出一个最小可用的 Alembic script_location，只含一条建表迁移。
@@ -130,7 +145,10 @@ def sqlite_backend(monkeypatch):
 def postgresql_backend(monkeypatch):
     """把宿主数据库类型固定为 PostgreSQL，并用替身覆盖宿主引擎。"""
     host_engine = MagicMock(name="host_engine")
-    derived_engine = MagicMock(name="derived_engine")
+    # 派生引擎必须是真实引擎：注册 begin 监听器要求一个真实的 SQLAlchemy 事件目标，
+    # 替身对象会被 event.listen 拒绝
+    borrowed_engine = create_engine("sqlite://")
+    derived_engine = borrowed_engine.execution_options()
     host_engine.execution_options.return_value = derived_engine
     monkeypatch.setattr(
         registry_module,
@@ -138,7 +156,8 @@ def postgresql_backend(monkeypatch):
         lambda key, default=None: "postgresql" if key == "DB_TYPE" else default,
     )
     monkeypatch.setattr(registry_module, "get_engine", lambda: host_engine)
-    return host_engine, derived_engine
+    yield host_engine, derived_engine
+    borrowed_engine.dispose()
 
 
 def test_plugin_declarative_base_returns_a_fresh_metadata_per_call():
@@ -318,25 +337,32 @@ def test_postgresql_handle_does_not_own_the_host_engine(postgresql_backend):
     )
 
 
-def test_postgresql_release_never_disposes_the_host_engine(postgresql_backend):
+def test_postgresql_release_never_disposes_the_host_engine(postgresql_backend, monkeypatch):
     """release 在 PostgreSQL 下不得 dispose 派生引擎，也不得触碰宿主引擎。"""
     host_engine, derived_engine = postgresql_backend
+    disposed: list[str] = []
+    monkeypatch.setattr(derived_engine, "dispose", lambda: disposed.append("derived"))
     registry_module.get_database("demo")
     registry_module.release_database("demo")
-    derived_engine.dispose.assert_not_called()
+    assert disposed == []
     host_engine.dispose.assert_not_called()
 
 
-def test_postgresql_destroy_drops_the_schema_and_keeps_the_host_engine(postgresql_backend):
+def test_postgresql_destroy_drops_the_schema_and_keeps_the_host_engine(
+    postgresql_backend,
+    monkeypatch,
+):
     """destroy 在 PostgreSQL 下丢弃对应 schema，且不 dispose 任何引擎。"""
     host_engine, derived_engine = postgresql_backend
+    disposed: list[str] = []
+    monkeypatch.setattr(derived_engine, "dispose", lambda: disposed.append("derived"))
     handle = registry_module.get_database("demo")
     schema = handle.schema
     registry_module.destroy_database("demo")
     connection = host_engine.begin.return_value.__enter__.return_value
     executed = [str(call.args[0]) for call in connection.execute.call_args_list]
     assert any("DROP SCHEMA" in sql and schema in sql for sql in executed)
-    derived_engine.dispose.assert_not_called()
+    assert disposed == []
     host_engine.dispose.assert_not_called()
 
 
@@ -486,41 +512,6 @@ def test_release_closes_the_thread_local_session(plugin_data_root, sqlite_backen
     assert handle.scoped_session_factory.registry.has() is False
 
 
-def test_release_after_destroy_removes_a_database_revived_by_a_stop_hook(
-    plugin_data_root,
-    sqlite_backend,
-):
-    """销毁后停机钩子取句柄让库文件复活，随后的 release 负责把它删掉并清除标记。"""
-    db_path = registry_module.get_database("demo").db_path
-    registry_module.destroy_database("demo")
-    assert not db_path.exists()
-
-    revived = registry_module.get_database("demo")
-    assert revived.db_path.exists()
-
-    registry_module.release_database("demo")
-
-    assert not db_path.exists()
-    assert registry_module._destroyed == set()
-
-
-def test_ensure_clears_the_destroyed_mark_so_a_restart_keeps_its_data(
-    plugin_data_root,
-    sqlite_backend,
-):
-    """重新建库即撤销销毁标记，此后的 release 只释放连接、不再删库。"""
-    registry_module.get_database("demo")
-    registry_module.destroy_database("demo")
-
-    registry_module.ensure_database("demo")
-    assert registry_module._destroyed == set()
-
-    handle = registry_module.get_database("demo")
-    registry_module.release_database("demo")
-
-    assert handle.db_path.exists()
-
-
 def test_missing_migrations_directory_is_rejected_before_any_file_is_created(
     plugin_data_root,
     sqlite_backend,
@@ -615,7 +606,6 @@ def test_run_migrations_upgrades_a_sqlite_plugin_database(plugin_data_root, sqli
 
 
 def test_run_migrations_routes_the_postgresql_connection_through_the_handle(
-    postgresql_backend,
     monkeypatch,
     tmp_path,
 ):
@@ -628,7 +618,7 @@ def test_run_migrations_routes_the_postgresql_connection_through_the_handle(
         captured["revision"] = revision
 
     monkeypatch.setattr(migration_module, "upgrade", _record_upgrade)
-    handle = registry_module.get_database("demo")
+    handle = _borrowed_engine_handle(MagicMock(name="derived_engine"))
 
     migration_module.run_migrations(handle, _write_migration_directory(tmp_path))
 
@@ -636,3 +626,150 @@ def test_run_migrations_routes_the_postgresql_connection_through_the_handle(
     assert captured["connection"] is connection
     assert captured["revision"] == "head"
     connection.commit.assert_called_once()
+
+
+def test_release_after_destroy_keeps_a_database_rebuilt_by_a_stop_hook(
+    plugin_data_root,
+    sqlite_backend,
+):
+    """销毁后重新建出的库属于仍在运行的插件，普通停止只释放连接、不得再删一次。"""
+    registry_module.get_database("demo")
+    registry_module.destroy_database("demo")
+
+    rebuilt = registry_module.get_database("demo")
+    session = rebuilt.session()
+    try:
+        session.execute(text("CREATE TABLE kept (id INTEGER PRIMARY KEY)"))
+        session.commit()
+    finally:
+        session.close()
+
+    registry_module.release_database("demo")
+
+    assert rebuilt.db_path.exists()
+    assert "demo" not in registry_module._handles
+
+
+def test_destroy_blocks_a_concurrent_handle_rebuild_until_the_carrier_is_removed(
+    plugin_data_root,
+    sqlite_backend,
+    monkeypatch,
+):
+    """销毁未删完载体前并发的取句柄取不到结果，新句柄因此不会指向被删掉的载体。"""
+    first = registry_module.get_database("demo")
+    entered_removal = threading.Event()
+    resume_removal = threading.Event()
+    original_remove_storage = registry_module._remove_storage
+
+    def _blocking_remove_storage(plugin_id, handle):
+        """在删除载体前挂住销毁流程，制造并发窗口。"""
+        entered_removal.set()
+        assert resume_removal.wait(10)
+        original_remove_storage(plugin_id, handle)
+
+    monkeypatch.setattr(registry_module, "_remove_storage", _blocking_remove_storage)
+    rebuilt: list[PluginDatabaseHandle] = []
+
+    def _rebuild() -> None:
+        """在销毁进行中重新取句柄。"""
+        rebuilt.append(registry_module.get_database("demo"))
+
+    destroyer = threading.Thread(target=registry_module.destroy_database, args=("demo",))
+    destroyer.start()
+    assert entered_removal.wait(10)
+    rebuilder = threading.Thread(target=_rebuild)
+    rebuilder.start()
+    rebuilder.join(0.5)
+
+    assert rebuilder.is_alive()
+    assert rebuilt == []
+
+    resume_removal.set()
+    destroyer.join(10)
+    rebuilder.join(10)
+
+    assert not rebuilder.is_alive()
+    assert rebuilt and rebuilt[0] is not first
+    assert rebuilt[0].db_path.exists()
+
+
+def test_search_path_setter_binds_the_quoted_plugin_schema():
+    """监听器在事务开始时执行 SET LOCAL search_path，schema 名带引号且回落 public。"""
+    connection = MagicMock(name="connection")
+
+    registry_module._search_path_setter("plugin_demo")(connection)
+
+    executed = connection.exec_driver_sql.call_args.args[0]
+    assert executed == 'SET LOCAL search_path TO "plugin_demo", public'
+
+
+def test_begin_listener_on_a_derived_engine_never_reaches_the_host_engine(tmp_path):
+    """派生引擎上的 begin 监听器只对派生连接生效，宿主引擎的连接不触发。"""
+    host_engine = create_engine(f"sqlite:///{tmp_path / 'host.db'}")
+    derived_engine = host_engine.execution_options()
+    fired: list[str] = []
+
+    def _listener(connection) -> None:
+        """记录触发并在同一连接上执行一条无害语句，验证不会递归或报错。"""
+        fired.append("begin")
+        connection.exec_driver_sql("SELECT 1")
+
+    event.listen(derived_engine, "begin", _listener)
+
+    with derived_engine.connect() as connection:
+        connection.execute(text("SELECT 1"))
+        connection.commit()
+    assert fired == ["begin"]
+
+    with host_engine.connect() as connection:
+        connection.execute(text("SELECT 1"))
+        connection.commit()
+    assert fired == ["begin"]
+
+    session = sessionmaker(bind=derived_engine)()
+    try:
+        assert session.execute(text("SELECT 1")).scalar() == 1
+        session.commit()
+    finally:
+        session.close()
+    assert fired == ["begin", "begin"]
+
+    host_engine.dispose()
+
+
+def test_postgresql_handle_binds_a_search_path_listener_to_the_derived_engine(
+    postgresql_backend,
+    monkeypatch,
+):
+    """PostgreSQL 建句柄时把插件 schema 的监听器挂到派生引擎，而不是宿主引擎。"""
+    _host_engine, derived_engine = postgresql_backend
+    bound: list[tuple] = []
+    build_listener = registry_module._search_path_setter
+
+    def _record(schema: str):
+        """记录被绑定的 schema 与生成的监听器。"""
+        listener = build_listener(schema)
+        bound.append((schema, listener))
+        return listener
+
+    monkeypatch.setattr(registry_module, "_search_path_setter", _record)
+
+    handle = registry_module.get_database("demo")
+
+    assert [schema for schema, _ in bound] == [handle.schema]
+    assert event.contains(derived_engine, "begin", bound[0][1])
+
+
+def test_alembic_migrations_trigger_the_begin_event_on_a_borrowed_engine(tmp_path):
+    """借用引擎的迁移在连接上首次执行即触发 begin，search_path 监听器因此覆盖 alembic。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'host.db'}").execution_options()
+    begins: list[str] = []
+    event.listen(engine, "begin", lambda _connection: begins.append("begin"))
+    handle = _borrowed_engine_handle(engine)
+
+    migration_module.run_migrations(handle, _write_migration_directory(tmp_path))
+
+    assert begins
+    assert "notes" in set(sa_inspect(engine).get_table_names())
+
+    engine.dispose()

tests/test_plugin_database_hooks.py

@@ -46,18 +46,14 @@ class _SamplePlugin(_PluginBase):
 
 @pytest.fixture(autouse=True)
 def _isolate_plugin_databases():
-    """快照插件数据库句柄与销毁标记，用例结束后释放残留句柄并还原快照。"""
+    """快照插件数据库句柄，用例结束后释放残留句柄并还原快照。"""
     handles = dict(registry_module._handles)
-    destroyed = set(registry_module._destroyed)
     registry_module._handles.clear()
-    registry_module._destroyed.clear()
     yield
     for plugin_id in list(registry_module._handles):
         registry_module.release_database(plugin_id)
     registry_module._handles.clear()
     registry_module._handles.update(handles)
-    registry_module._destroyed.clear()
-    registry_module._destroyed.update(destroyed)
 
 
 @pytest.fixture

tests/test_plugin_database_lifecycle.py

@@ -7,10 +7,15 @@ import asyncio
 import logging
 from pathlib import Path
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+import app.api.endpoints.plugin as plugin_endpoint
+import app.application.plugin.folders as plugin_folders
+import app.application.plugin.management as plugin_management
+import app.application.plugin.routes as plugin_routes
+import app.application.scheduling as scheduling_module
 import app.db.engine as engine_module
 import app.db.session as session_module
 import app.runtime.extensions.plugin.database as database_module
@@ -18,7 +23,8 @@ import app.startup.initializers.plugins as plugins_initializer
 from app.runtime.extensions.plugin.database import PluginDatabase
 from app.runtime.extensions.plugin.lifecycle import PluginLifecycle
 from app.runtime.extensions.plugin.storage import PluginConfigStore, PluginStorage
-from app.schemas.plugin import PluginRuntimeStatus
+from app.schemas.plugin import PluginInstance, PluginRuntimeStatus
+from app.schemas.types import SystemConfigKey
 
 
 class ModelA:
@@ -454,3 +460,115 @@ def test_stop_all_releases_plugins_that_never_reached_the_running_registry():
     lifecycle.stop()
 
     assert ("release", "DemoPlugin") in calls
+
+
+def test_stopping_an_already_stopped_plugin_stays_idempotent():
+    """卸载编排先停后删，删除之后的 remove_plugin 会再停一次，二次停止必须无害。"""
+    calls: list[tuple] = []
+    plugin_cls = _make_plugin_class("DemoPlugin")
+    lifecycle = _build_lifecycle(
+        load_plugins=lambda *_a, **_kw: [plugin_cls],
+        installed_plugins=lambda: ["DemoPlugin"],
+        database=lambda: _recording_database(calls),
+    )
+    lifecycle.start("DemoPlugin")
+    lifecycle.stop("DemoPlugin")
+    calls.clear()
+
+    lifecycle.stop("DemoPlugin")
+
+    assert calls == [("release", "DemoPlugin")]
+    assert lifecycle._classes == {}
+    assert lifecycle._running == {}
+
+
+def _uninstall_manager() -> MagicMock:
+    """构造卸载分身所需的插件管理器替身，按调用顺序记录全部方法调用。"""
+    plugin_manager = MagicMock()
+    plugin_manager.get_plugin_instance.return_value = None
+    plugin_manager.get_plugin_source_instances.return_value = []
+    plugin_manager.plugins = {"DemoPluginwork": MagicMock(is_clone=True)}
+    plugin_manager.remove_plugin_package.return_value = True
+    return plugin_manager
+
+
+def _assert_stop_precedes_deletion(plugin_manager: MagicMock) -> None:
+    """断言插件先停止再删除数据，且删除按 force 执行。"""
+    method_names = [name for name, _args, _kwargs in plugin_manager.mock_calls]
+    assert method_names.index("stop") < method_names.index("delete_plugin_config")
+    assert method_names.index("stop") < method_names.index("delete_plugin_data")
+    plugin_manager.stop.assert_called_once_with("DemoPluginwork")
+    plugin_manager.delete_plugin_config.assert_called_once_with(
+        "DemoPluginwork",
+        force=True,
+    )
+    plugin_manager.delete_plugin_data.assert_called_once_with(
+        "DemoPluginwork",
+        force=True,
+    )
+
+
+def test_http_uninstall_stops_the_plugin_before_deleting_its_data(monkeypatch):
+    """HTTP 卸载先停插件再删数据，停机钩子因此无法重建刚销毁的自有库。"""
+    plugin_manager = _uninstall_manager()
+    config = MagicMock()
+    config.get.return_value = ["DemoPluginwork"]
+    monkeypatch.setattr(plugin_endpoint, "get_plugin_manager", lambda: plugin_manager)
+    monkeypatch.setattr(plugin_endpoint, "get_configured_system_config", lambda: config)
+    monkeypatch.setattr(plugin_endpoint, "remove_plugin_api", MagicMock())
+    monkeypatch.setattr(plugin_endpoint, "remove_plugin_job", MagicMock())
+    monkeypatch.setattr(plugin_endpoint, "remove_plugin_from_folders", MagicMock())
+
+    result = plugin_endpoint.uninstall_plugin("DemoPluginwork", None)
+
+    assert result.success is True
+    _assert_stop_precedes_deletion(plugin_manager)
+
+
+def test_runtime_uninstall_stops_the_plugin_before_deleting_its_data(monkeypatch):
+    """运行态卸载编排同样先停后删，两条卸载路径的数据删除时机保持一致。"""
+    plugin_manager = _uninstall_manager()
+    config = MagicMock()
+    config.get.return_value = ["DemoPluginwork"]
+    config.async_set = AsyncMock()
+    monkeypatch.setattr(plugin_management, "get_plugin_manager", lambda: plugin_manager)
+    monkeypatch.setattr(
+        plugin_management,
+        "get_configured_system_config",
+        lambda: config,
+    )
+    monkeypatch.setattr(plugin_routes, "remove_plugin_api", MagicMock())
+    monkeypatch.setattr(scheduling_module, "remove_plugin_job", MagicMock())
+    monkeypatch.setattr(plugin_folders, "remove_plugin_from_folders", MagicMock())
+
+    result = asyncio.run(plugin_management.uninstall_plugin_runtime("DemoPluginwork"))
+
+    assert result == {"was_clone": True, "clone_files_removed": True}
+    _assert_stop_precedes_deletion(plugin_manager)
+
+
+def test_uninstall_virtual_instance_also_stops_before_deleting(monkeypatch):
+    """虚拟实例卸载同样先停后删，force 删除不受插件类注销影响。"""
+    plugin_manager = MagicMock()
+    plugin_manager.get_plugin_instance.return_value = PluginInstance(
+        instance_id="DemoPluginwork",
+        source_plugin_id="DemoPlugin",
+    )
+    plugin_manager.get_plugin_source_instances.return_value = []
+    plugin_manager.plugins = {}
+    config = MagicMock()
+    config.get.return_value = ["DemoPlugin"]
+    monkeypatch.setattr(plugin_endpoint, "get_plugin_manager", lambda: plugin_manager)
+    monkeypatch.setattr(plugin_endpoint, "get_configured_system_config", lambda: config)
+    monkeypatch.setattr(plugin_endpoint, "remove_plugin_api", MagicMock())
+    monkeypatch.setattr(plugin_endpoint, "remove_plugin_job", MagicMock())
+    monkeypatch.setattr(plugin_endpoint, "remove_plugin_from_folders", MagicMock())
+
+    result = plugin_endpoint.uninstall_plugin("DemoPluginwork", None)
+
+    assert result.success is True
+    config.set.assert_called_once_with(
+        SystemConfigKey.UserInstalledPlugins,
+        ["DemoPlugin"],
+    )
+    _assert_stop_precedes_deletion(plugin_manager)