From d6509f092c7a2a464611886a588820c8ca7ef75d Mon Sep 17 00:00:00 2001 From: Aqr-K <1210498076@qq.com> Date: Sat, 12 Sep 2026 19:48:57 -0400 Subject: [PATCH] =?UTF-8?q?chore(plugin):=20=E6=8A=80=E8=83=BD=E8=A1=A8?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E7=99=BB=E8=AE=B0=E8=A1=A5=E8=BF=9B=E7=94=9F?= =?UTF-8?q?=E6=88=90=E8=84=9A=E6=9C=AC=E5=B9=B6=E8=A1=A5=E8=BF=81=E7=A7=BB?= =?UTF-8?q?=E5=BE=80=E8=BF=94=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 技能文档 skills/database-operation/SKILL.md 的核心表清单是 scripts/generate_agent_skill_docs.py 里 DATABASE_TABLE_GUIDES 的生成物， 前面两笔改动却是手改产物登记 plugininstance，任何人重跑生成器都会把它弄丢。 把条目补进字典，再用生成器重出该文档。 同时补上 agentinvocation：官方 v3 已有该表但字典里没有，生成器因此直接抛 ValueError，不补就根本跑不起来。它的说明照抄文档里现有的那份手写文案， 避免登记动作反而把已有描述改瘦。 迁移测试原本只按单个 revision 直接调 upgrade()，绕过了 Alembic 自己的版本 解析与记账，链接错的 down_revision、少写的 downgrade、重复升级时的重复建表 都测不出来。新增用例改走 alembic.command，在落盘的临时库上从官方 head b2d4f6a8c1e3 升到链尾 487f7e681955 再降回去，验证配置与描述符确实落到 plugininstance、旧 plugin.* 键就地移除而 PluginInstances 键按回滚依据保留。 --- scripts/generate_agent_skill_docs.py | 20 ++ skills/database-operation/SKILL.md | 14 +- ...est_plugin_instance_migration_roundtrip.py | 252 ++++++++++++++++++ 3 files changed, 279 insertions(+), 7 deletions(-) create mode 100644 tests/test_plugin_instance_migration_roundtrip.py diff --git a/scripts/generate_agent_skill_docs.py b/scripts/generate_agent_skill_docs.py index 2e0ba830d..2f4fad226 100644 --- a/scripts/generate_agent_skill_docs.py +++ b/scripts/generate_agent_skill_docs.py @@ -26,6 +26,16 @@ DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { "Tracing Agent history or context restoration by user, session, or update time.", "Owned by the Agent conversation service; do not rewrite message JSON, counters, or ownership.", ), + "agentinvocation": ( + "Durable Agent write invocation identity and last observed outcome, including confirmed asynchronous " + "submission.", + "Inspect an exact principal and session's running, unknown, pending, succeeded, or failed receipts; " + "compare timestamps when diagnosing an interrupted write.", + "Owned by the host's atomic claim and reconciliation path. Never change IDs, fingerprints, claim " + "tokens, or statuses to bypass duplicate protection. Running and unknown records are recovery state; " + "ordinary retention does not delete them. Pending means submission was confirmed, not that the " + "external task finished. Raw arguments and tool output are not stored here.", + ), "agenttask": ( "Stores one-shot or recurring Agent task definitions, triggers, and the latest execution summary.", "Inspecting task ownership, enablement, cron/run_at settings, and the latest result.", @@ -86,6 +96,16 @@ DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { "Diagnosing interrupted installations, rollback conditions, and package or backup presence.", "Owned by the plugin installation state machine; never advance phase or overwrite evidence manually.", ), + "plugininstance": ( + "Stores one row per shared-source plugin runtime instance, covering both clones and the host plugin " + "itself (instance_id equals source_plugin_id, so equality identifies the host and inequality a clone), " + "together with that instance's display overrides, its own log-level override and the moment that " + "override expires, and the plugin's own configuration payload. A clone exists exactly while its row " + "exists, so deleting the row uninstalls the clone and discards its configuration.", + "Diagnosing clone naming and ownership, inspecting what a plugin or one of its clones is configured " + "with, or finding which instance currently overrides the global log level and until when.", + "Owned by the plugin instance, plugin configuration, and plugin log-level APIs; never edit rows directly.", + ), "site": ( "Stores private-tracker URLs, RSS, credentials, rate limits, proxy state, and downloader binding.", "Inspecting enablement, domain, rate limits, or downloader binding with minimal credential exposure.", diff --git a/skills/database-operation/SKILL.md b/skills/database-operation/SKILL.md index 187fe4d68..7442873a1 100644 --- a/skills/database-operation/SKILL.md +++ b/skills/database-operation/SKILL.md @@ -125,8 +125,8 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" ### `agentinvocation` - Purpose: Durable Agent write invocation identity and last observed outcome, including confirmed asynchronous submission. - Useful queries: Inspect an exact principal and session's running, unknown, pending, succeeded, or failed receipts; compare timestamps when diagnosing an interrupted write. -- Write boundary: Owned by the host's atomic claim and reconciliation path. Never change IDs, fingerprints, claim tokens, or statuses to bypass duplicate protection. Running and unknown records are recovery state; ordinary retention does not delete them. Pending means submission was confirmed, not that the external task finished. -- Columns: `id`, `principal_id`, `session_id`, `invocation_id`, `tool_name`, `arguments_digest`, `claim_token`, `status`, `summary`, `created_at`, `updated_at`. Raw arguments and tool output are not stored here. +- Write boundary: Owned by the host's atomic claim and reconciliation path. Never change IDs, fingerprints, claim tokens, or statuses to bypass duplicate protection. Running and unknown records are recovery state; ordinary retention does not delete them. Pending means submission was confirmed, not that the external task finished. Raw arguments and tool output are not stored here. +- Columns: `id`, `principal_id`, `session_id`, `invocation_id`, `tool_name`, `arguments_digest`, `claim_token`, `status`, `summary`, `created_at`, `updated_at` ### `agenttask` - Purpose: Stores one-shot or recurring Agent task definitions, triggers, and the latest execution summary. @@ -207,9 +207,9 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - 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, its own log-level override, 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. +- Purpose: Stores one row per shared-source plugin runtime instance, covering both clones and the host plugin itself (instance_id equals source_plugin_id, so equality identifies the host and inequality a clone), together with that instance's display overrides, its own log-level override and the moment that override expires, and the plugin's 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, inspecting what a plugin or one of its clones is configured with, or finding which instance currently overrides the global log level and until when. +- Write boundary: Owned by the plugin instance, plugin configuration, and plugin log-level APIs; never edit rows directly. - Columns: `id`, `instance_id`, `source_plugin_id`, `plugin_name`, `plugin_desc`, `plugin_icon`, `log_level`, `log_expires_at`, `config_data`, `created_at`, `updated_at` ### `site` @@ -258,7 +258,7 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - Purpose: Stores one durable subscription search task per batch and subscription with leases and execution phases. - Useful queries: Diagnosing queued, running, failed, cancelled, or recovered work and its current site. - Write boundary: Advanced only by the search queue lease state machine; never rewrite leases or terminal states manually. -- Columns: `id`, `task_id`, `batch_id`, `subscription_id`, `active_key`, `source`, `priority`, `position`, `state`, `phase`, `current_site_id`, `attempt_count`, `cancel_requested`, `lease_owner`, `lease_token`, `lease_expires_at`, `available_at`, `created_at`, `updated_at`, `started_at`, `finished_at`, `last_error` +- Columns: `id`, `task_id`, `batch_id`, `subscription_id`, `active_key`, `source`, `priority`, `position`, `state`, `phase`, `current_site_id`, `pending_site_ids`, `attempt_count`, `cancel_requested`, `lease_owner`, `lease_token`, `lease_expires_at`, `available_at`, `created_at`, `updated_at`, `started_at`, `finished_at`, `last_error` ### `subscriptionsitebudget` - Purpose: Stores per-site subscription search concurrency, cooldown, health, and fairness state. @@ -282,7 +282,7 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - Purpose: Stores transfer source, destination, mode, media identity, download linkage, and outcome. - Useful queries: Reviewing success/failure history, destination paths, media classification, and download linkage. - Write boundary: Written by transfer settlement; delete or retry through transfer-history business APIs. -- Columns: `id`, `transfer_task_id`, `transfer_settlement_revision`, `src`, `src_storage`, `src_fileitem`, `dest`, `dest_storage`, `dest_fileitem`, `mode`, `type`, `media_category_id`, `category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `title`, `year`, `media_source`, `media_id`, `music_type`, `total_tracks`, `audio_format`, `audio_lossless`, `bit_depth`, `sample_rate`, `bitrate`, `seasons`, `episodes`, `image`, `downloader`, `download_hash`, `status`, `errmsg`, `date`, `files`, `episode_group` +- Columns: `id`, `transfer_task_id`, `transfer_settlement_revision`, `src`, `src_storage`, `src_fileitem`, `dest`, `dest_storage`, `dest_fileitem`, `mode`, `type`, `media_category_id`, `category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `title`, `year`, `media_source`, `media_id`, `music_type`, `total_tracks`, `audio_format`, `audio_lossless`, `bit_depth`, `sample_rate`, `bitrate`, `seasons`, `episodes`, `image`, `downloader`, `download_hash`, `status`, `errmsg`, `retry_count`, `auto_paused`, `failure_stage`, `recovery_action`, `cleanup_status`, `cleanup_error`, `date`, `files`, `episode_group` ### `transferpending` - Purpose: Durably stores pending transfer input, plans, checkpoints, leases, retries, and manual review state. diff --git a/tests/test_plugin_instance_migration_roundtrip.py b/tests/test_plugin_instance_migration_roundtrip.py new file mode 100644 index 000000000..8c87b7ef4 --- /dev/null +++ b/tests/test_plugin_instance_migration_roundtrip.py @@ -0,0 +1,252 @@ +"""插件实例迁移链在真实 Alembic 运行器上的升降级往返验证。 + +同目录下按单个 revision 调 ``upgrade()`` 的用例只验证单步搬迁逻辑，绕过了 +Alembic 自己的版本解析、顺序编排与 ``alembic_version`` 记账。真实事故恰恰出在 +这一层：链接错的 ``down_revision``、少写的 ``downgrade``、第二次升级时重复建表， +单步用例一个都发现不了。本模块因此走 ``alembic.command``，在一个落盘的临时库上 +从官方 head 一路升到链尾再降回去，验证三个迁移串起来确实可用。 +""" + +from __future__ import annotations + +import json +from configparser import ConfigParser +from pathlib import Path + +import pytest +import sqlalchemy as sa +from alembic import command +from alembic.config import Config + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + +# 官方 v3 的迁移链末端，也是本特性三个迁移的共同起点 +OFFICIAL_HEAD = "b2d4f6a8c1e3" +CHAIN_HEAD = "487f7e681955" +CHAIN = ("281965691a20", "c4e1a7b9d2f6", CHAIN_HEAD) + +_TABLE = "plugininstance" +_LEGACY_KEY = "PluginInstances" + +# 一条分身描述符、两条插件配置（本体一条、分身一条），照搬运行期真实形状 +_LEGACY_INSTANCES = { + "DemoPluginWork": { + "instance_id": "DemoPluginWork", + "source_plugin_id": "DemoPlugin", + "plugin_name": "工作实例", + "plugin_desc": "分身描述", + "plugin_icon": "work.svg", + }, +} +_HOST_CONFIG = {"enabled": True, "cron": "0 1 * * *", "notify": False} +_CLONE_CONFIG = {"enabled": False, "cron": "0 2 * * *", "notify": True} + + +def _instance_table() -> sa.TableClause: + """按列类型声明实例表，使 JSON 配置列读出时自动解码。""" + return 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("log_level", sa.String()), + sa.column("log_expires_at", sa.String()), + sa.column("config_data", sa.JSON()), + ) + + +def _seed_legacy_rows(engine: sa.Engine) -> None: + """写入迁移前的 systemconfig 存量：实例描述符单键与两条 plugin.* 配置键。""" + with engine.begin() as connection: + for key, value in ( + (_LEGACY_KEY, _LEGACY_INSTANCES), + ("plugin.DemoPlugin", _HOST_CONFIG), + ("plugin.DemoPluginWork", _CLONE_CONFIG), + ): + connection.execute( + sa.text("INSERT INTO systemconfig (key, value) VALUES (:key, :value)"), + {"key": key, "value": json.dumps(value)}, + ) + + +def _systemconfig_keys(engine: sa.Engine) -> set[str]: + """读取当前库里的全部 systemconfig 键。""" + with engine.connect() as connection: + return { + row[0] + for row in connection.execute(sa.text("SELECT key FROM systemconfig")) + } + + +def _systemconfig_value(engine: sa.Engine, key: str): + """读取一条 systemconfig 值并解码，键不存在时返回 ``None``。""" + with engine.connect() as connection: + row = connection.execute( + sa.text("SELECT value FROM systemconfig WHERE key = :key"), {"key": key} + ).first() + if row is None: + return None + value = row[0] + return json.loads(value) if isinstance(value, (str, bytes, bytearray)) else value + + +def _instance_rows(engine: sa.Engine) -> dict[str, sa.Row]: + """按实例 ID 读出实例表全部行。""" + table = _instance_table() + with engine.connect() as connection: + return { + row.instance_id: row + for row in connection.execute(sa.select(table)) + } + + +def _current_revision(engine: sa.Engine) -> str | None: + """读取库里记账的当前 revision。""" + with engine.connect() as connection: + if "alembic_version" not in sa.inspect(connection).get_table_names(): + return None + return connection.execute( + sa.text("SELECT version_num FROM alembic_version") + ).scalar() + + +def _table_names(engine: sa.Engine) -> set[str]: + """读取库中已有表名。""" + with engine.connect() as connection: + return set(sa.inspect(connection).get_table_names()) + + +@pytest.fixture +def official_head_database(tmp_path): + """造一个停在官方 head 的落盘 SQLite 库，并塞入迁移前的存量数据。 + + 官方 head 与本分支的 schema 差异只有 ``plugininstance`` 一张表，因此「按当前 + 模型建表但跳过该表，再把版本记账戳到官方 head」与真实的官方库等价；从链条最 + 开头逐条重放做不到，因为官方的初始化流程本身是先 ``create_all`` 再升级，早期 + 迁移并不自带建表。 + + :return: ``(Alembic 配置, 指向该库的 Engine)`` + """ + from app.db.base import Base + from app.db.models import load_all_models + + load_all_models() + database_path = tmp_path / "roundtrip.db" + url = f"sqlite:///{database_path}" + engine = sa.create_engine(url) + Base.metadata.create_all( + engine, + tables=[ + table + for name, table in Base.metadata.tables.items() + if name != _TABLE + ], + ) + + config = Config() + # 不读 ini 文件，避免 env.py 对日志配置段的依赖，与应用启动时的构造方式一致 + config.file_config = ConfigParser(interpolation=None) + config.set_main_option("script_location", str(PROJECT_ROOT / "database")) + config.set_main_option("sqlalchemy.url", url) + command.stamp(config, OFFICIAL_HEAD) + + _seed_legacy_rows(engine) + try: + yield config, engine + finally: + # 建的是整套官方表，单个库文件接近 1MB；pytest 默认还要留最近几轮 tmp_path， + # 攒下来不小，因此释放句柄后立刻删掉，不把回收推给它 + engine.dispose() + database_path.unlink(missing_ok=True) + + +def test_upgrade_from_official_head_lands_legacy_payloads_on_instance_rows( + official_head_database, +) -> None: + """整条链升上来后，描述符与两条配置都应落到实例表，旧配置键就地移除。""" + config, engine = official_head_database + + command.upgrade(config, CHAIN_HEAD) + + assert _current_revision(engine) == CHAIN_HEAD + rows = _instance_rows(engine) + assert set(rows) == {"DemoPlugin", "DemoPluginWork"} + + clone = rows["DemoPluginWork"] + assert clone.source_plugin_id == "DemoPlugin" + assert clone.plugin_name == "工作实例" + assert clone.plugin_desc == "分身描述" + assert clone.plugin_icon == "work.svg" + assert clone.config_data == _CLONE_CONFIG + + host = rows["DemoPlugin"] + # 本体没有描述符条目，靠 plugin. 配置键补建，故身份两列相等 + assert host.source_plugin_id == "DemoPlugin" + assert host.config_data == _HOST_CONFIG + + # 日志等级列由链尾迁移补出，缺省为空表示跟随全局等级 + assert clone.log_level is None and clone.log_expires_at is None + assert host.log_level is None and host.log_expires_at is None + + keys = _systemconfig_keys(engine) + # 同一份配置不能两处各留一份，plugin.* 旧键必须消失 + assert "plugin.DemoPlugin" not in keys + assert "plugin.DemoPluginWork" not in keys + # 描述符单键刻意保留作回滚依据 + assert _systemconfig_value(engine, _LEGACY_KEY) == _LEGACY_INSTANCES + + +def test_downgrade_back_to_official_head_restores_the_legacy_config_keys( + official_head_database, +) -> None: + """降回官方 head 不应报错，实例表整张移除且配置回到旧键，回退路径不丢配置。""" + config, engine = official_head_database + command.upgrade(config, CHAIN_HEAD) + + command.downgrade(config, OFFICIAL_HEAD) + + assert _current_revision(engine) == OFFICIAL_HEAD + assert _TABLE not in _table_names(engine) + assert _systemconfig_value(engine, "plugin.DemoPlugin") == _HOST_CONFIG + assert _systemconfig_value(engine, "plugin.DemoPluginWork") == _CLONE_CONFIG + assert _systemconfig_value(engine, _LEGACY_KEY) == _LEGACY_INSTANCES + + +def test_upgrade_is_idempotent_on_an_already_migrated_database( + official_head_database, +) -> None: + """重复升级与降级后重升都不应报错，且不会把行搬成两份。""" + config, engine = official_head_database + command.upgrade(config, CHAIN_HEAD) + first_pass = _instance_rows(engine) + + # 已在链尾时再升一次：Alembic 层面是空操作，验证不会因重复执行而失败 + command.upgrade(config, CHAIN_HEAD) + assert _current_revision(engine) == CHAIN_HEAD + assert _instance_rows(engine).keys() == first_pass.keys() + + # 降回官方 head 后重升：三个迁移都要在已有存量的库上重跑一遍 + command.downgrade(config, OFFICIAL_HEAD) + command.upgrade(config, CHAIN_HEAD) + + assert _current_revision(engine) == CHAIN_HEAD + replayed = _instance_rows(engine) + assert set(replayed) == {"DemoPlugin", "DemoPluginWork"} + assert replayed["DemoPlugin"].config_data == _HOST_CONFIG + assert replayed["DemoPluginWork"].config_data == _CLONE_CONFIG + assert replayed["DemoPluginWork"].source_plugin_id == "DemoPlugin" + + +def test_chain_applies_one_revision_at_a_time_in_the_declared_order( + official_head_database, +) -> None: + """逐级升级应按声明顺序推进记账，确认三个迁移的链接方向没有接反。""" + config, engine = official_head_database + + for revision in CHAIN: + command.upgrade(config, revision) + assert _current_revision(engine) == revision + + assert _TABLE in _table_names(engine)