diff --git a/DATABASE_MIGRATION_ATOMICITY_TASK.md b/DATABASE_MIGRATION_ATOMICITY_TASK.md new file mode 100644 index 0000000..ce8df5e --- /dev/null +++ b/DATABASE_MIGRATION_ATOMICITY_TASK.md @@ -0,0 +1,54 @@ +# 数据库迁移原子性与 Prompt 外键修复任务 + +## 背景 + +第二次工程复检确认：内容实验账本迁移在事务内调用 `sqlite3.executescript()`，会隐式提交并破坏整体回滚；历史数据库通过 `ALTER TABLE` 新增的 `ai_analysis_runs.prompt_version_id` 没有新建数据库所具备的外键。 + +## 目标 + +- 让内容实验结构、校验与迁移账本处于同一事务，失败时不残留半套表或索引。 +- 为已升级的历史数据库补齐 `prompt_version_id → ai_prompt_versions.id` 外键，并保持 AI Run 数据、既有索引和下游引用。 +- 迁移前生成可恢复备份；数据或结构无法安全证明时 fail-closed，不猜测或删除历史记录。 + +## 允许修改范围 + +- `app/db/database.py` +- `tests/test_schema_migration_ledger.py` +- `docs/DATABASE_SCHEMA.md` +- `DEVELOPMENT_LOG.md`、`NEXT_STEPS.md` 与本任务文件 + +## 禁止修改范围 + +- 活动 SQLite、真实 Provider、Chrome Worker、发布任务和运行中服务。 +- 其他历史兼容迁移的全面重写。 +- 删除或自动清空无法归因的 AI Run。 + +## 已确定实现要求 + +1. 内容实验迁移不得在账本事务内使用 `executescript()`。 +2. 外键修复使用新的迁移版本和 checksum，不改写已发布迁移账本。 +3. 表重建时临时关闭外键仅限该迁移连接，事务提交或回滚后恢复原状态，并执行 `PRAGMA foreign_key_check`。 +4. 保留 AI Run 全部规范字段、显式索引和触发器；发现未知字段、临时迁移表或孤儿 Prompt 引用时拒绝迁移。 +5. 新建库与历史升级库最终都必须具有同一 `NO ACTION` Prompt 外键语义。 + +## 验收标准 + +- 故障注入拒绝创建第二张实验表后，第一张表、索引和账本均不残留；移除故障后可安全重跑。 +- 模拟旧库升级后 AI Run、反馈引用和索引保留，Prompt 外键存在，外键检查为空。 +- 孤儿 Prompt 引用会使迁移整体失败，原表和数据不变，账本不写入。 +- 定向测试、全量测试、Ruff、Compileall、Compose 配置和 `git diff --check` 通过。 + +## 测试命令 + +```powershell +pytest -q tests/test_schema_migration_ledger.py tests/test_database_backup_service.py +pytest -q +ruff check app tests scripts +python -m compileall -q app tests scripts +git diff --check +``` + +## 返回格式 + +- 原子性与外键修复说明、故障注入和重跑证据。 +- 修改文件、测试结果、分支、提交 SHA、远端 SHA 与 PR 状态。 diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index 2f7ca6e..f7bfb14 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1441,3 +1441,10 @@ - 显式片段反馈改为只绑定候选的 `source_analysis_run_id`，并验证 Run 属于同一任务；来源缺失、不存在或跨任务时保留反馈但不归因，绝不回退到当前 active Run。 - Prompt 对比统一使用“官方导入时长 → 候选时长 → 输出片段源时长”的有效时长口径，官方报表时长为空时仍能计算平均观看比例。 - 定向回归由修复前 `55 passed` 增加到 `62 passed`，全量回归 `861 passed`；Ruff、Compileall、5 个 JavaScript 语法检查、20 个 PowerShell 解析检查、三套合并 Compose 配置、`pip check` 和 `git diff --check` 均通过。测试只使用临时 SQLite 和本地 mock，未调用真实 Provider、Chrome 或发布平台，也未修改活动数据库。 + +## 2026-08-30 数据库迁移原子性与 Prompt 外键修复 + +- 内容实验账本迁移不再调用会隐式提交的 `executescript()`；表、索引、结构校验和账本现在处于同一 `BEGIN IMMEDIATE` 事务，第二张表故障注入后第一张表和账本均可完整回滚并安全重跑。 +- 新增独立迁移补齐历史 `ai_analysis_runs.prompt_version_id` 外键，保留 AI Run 数据、显式索引、触发器和下游反馈引用；未知字段、残留临时表或孤儿 Prompt 引用均 fail-closed。 +- 正式数据库仅做只读核对：当前 39 条 AI Run、0 条孤儿 Prompt 引用、0 条现有外键异常，确认具备无损迁移前提；本阶段没有改写活动 SQLite 或重启服务。 +- 迁移与备份定向回归 `28 passed`，迁移失败、重试、外键、孤儿引用与旧 AI Run 专项 `5 passed`，全量回归 `865 passed`；Ruff、Compileall、5 个 JavaScript 语法检查、20 个 PowerShell 解析检查、三套合并 Compose 配置、`pip check` 和 `git diff --check` 均通过。验证只使用隔离临时库；活动主库大小保持 `10,276,864` bytes，但文件时间被现有后台服务持续更新，因此不把 mtime 作为“未写入”证据。 diff --git a/NEXT_STEPS.md b/NEXT_STEPS.md index 648857a..5cafe14 100644 --- a/NEXT_STEPS.md +++ b/NEXT_STEPS.md @@ -1181,3 +1181,11 @@ 3. 对历史候选提交显式反馈时，反馈应归属于候选生成时的 Prompt Run；来源无法证明的旧候选应显示为未归因，不得算到当前 Prompt。 4. 导入抖音官方作品报表后，即使报表没有视频时长，已匹配作品的 Prompt 对比仍应按候选或输出片段时长显示平均观看比例。 5. 本阶段不需要重启服务、修改活动数据库或执行真实投稿；完成自动验收后继续处理数据库迁移原子性。 + +## 2026-08-30 数据库迁移原子性验收 + +1. 先确认数据库迁移 PR 的全量测试、Windows 主机冒烟和 Docker 镜像冒烟全部通过；未经用户确认不合并。 +2. 本阶段自动测试只使用临时 SQLite。正式库应用必须安排维护窗口，先核对 `workflow-before-ai-prompt-version-fk-*` 备份成功，再允许新版本首次启动执行迁移。 +3. 首次启动后检查迁移账本、`PRAGMA foreign_key_list(ai_analysis_runs)`、`PRAGMA foreign_key_check`、39 条既有 AI Run 数量和反馈引用；任一不一致立即停止并使用备份回滚。 +4. 不需要重新调用 AI、重新投稿或改写历史 Prompt；迁移只补结构约束，不改变内容业务数据。 +5. 自动验收完成后继续补异步服务边界与页面主链回归测试。 diff --git a/app/db/database.py b/app/db/database.py index c0f9f4b..24faeee 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -138,6 +138,42 @@ CONTENT_FEEDBACK_LOOP_MIGRATION_CHECKSUM = hashlib.sha256( CONTENT_FEEDBACK_LOOP_MIGRATION_SPEC.encode("utf-8") ).hexdigest() +AI_PROMPT_VERSION_FK_MIGRATION_VERSION = "20260830_01_ai_prompt_version_fk" +AI_PROMPT_VERSION_FK_MIGRATION_NAME = "AI 分析 Prompt 版本外键一致性" +AI_PROMPT_VERSION_FK_MIGRATION_SPEC = "\n".join( + ( + AI_PROMPT_VERSION_FK_MIGRATION_VERSION, + AI_PROMPT_VERSION_FK_MIGRATION_NAME, + "ai_analysis_runs.prompt_version_id->ai_prompt_versions.id", + "on-update-no-action", + "on-delete-no-action", + "preserve-ai-analysis-run-data-indexes-triggers", + "reject-orphan-prompt-version-references", + "foreign-key-check-before-ledger-commit", + ) +) +AI_PROMPT_VERSION_FK_MIGRATION_CHECKSUM = hashlib.sha256( + AI_PROMPT_VERSION_FK_MIGRATION_SPEC.encode("utf-8") +).hexdigest() +AI_ANALYSIS_RUN_COLUMNS = ( + "id", + "task_id", + "run_number", + "provider", + "provider_label", + "model", + "ai_prompt_preset_id", + "ai_prompt_preset_name", + "prompt_version_id", + "prompt_text_sha256", + "requested_clip_count", + "clip_count", + "analysis_summary", + "fallback_notice", + "analysis_payload_json", + "created_at", + "is_active", +) class SchemaMigrationError(RuntimeError): @@ -151,6 +187,7 @@ class SchemaMigration: checksum: str apply: Callable[[sqlite3.Connection], None] verify: Callable[[sqlite3.Connection], None] + requires_foreign_keys_off: bool = False @contextmanager @@ -182,6 +219,9 @@ def init_db() -> None: needs_content_feedback_loop_backup = _requires_content_feedback_loop_migration( settings.database_path ) + needs_ai_prompt_version_fk_backup = _requires_ai_prompt_version_fk_migration( + settings.database_path + ) if needs_long_live_backup: create_schema_migration_backup( settings.database_path, @@ -247,6 +287,18 @@ def init_db() -> None: settings.data_dir / "backups", "content-feedback-loop", ) + if needs_ai_prompt_version_fk_backup and not any( + ( + needs_content_review_backup, + needs_douyin_item_export_backup, + needs_content_feedback_loop_backup, + ) + ): + create_schema_migration_backup( + settings.database_path, + settings.data_dir / "backups", + "ai-prompt-version-fk", + ) with get_connection() as connection: connection.executescript( @@ -1130,6 +1182,61 @@ def _requires_content_feedback_loop_migration(database_path) -> bool: connection.close() +def _requires_ai_prompt_version_fk_migration(database_path) -> bool: + """旧 AI Run 表缺少 Prompt 外键或新迁移账本时，重建前先备份。""" + if not database_path.exists() or database_path.stat().st_size == 0: + return False + connection = None + try: + connection = sqlite3.connect( + f"{database_path.resolve().as_uri()}?mode=ro", + uri=True, + timeout=10, + ) + table_names = { + row[0] + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + } + if "ai_analysis_runs" not in table_names: + return False + ledger_row = None + if "schema_migrations" in table_names: + try: + ledger_row = connection.execute( + "SELECT 1 FROM schema_migrations WHERE version = ? AND checksum = ?", + ( + AI_PROMPT_VERSION_FK_MIGRATION_VERSION, + AI_PROMPT_VERSION_FK_MIGRATION_CHECKSUM, + ), + ).fetchone() + except sqlite3.Error: + return True + return not _has_ai_prompt_version_fk(connection) or ledger_row is None + finally: + if connection is not None: + connection.close() + + +def _has_ai_prompt_version_fk(connection: sqlite3.Connection) -> bool: + for row in connection.execute("PRAGMA foreign_key_list(ai_analysis_runs)").fetchall(): + table_name = row["table"] if isinstance(row, sqlite3.Row) else row[2] + source_column = row["from"] if isinstance(row, sqlite3.Row) else row[3] + target_column = row["to"] if isinstance(row, sqlite3.Row) else row[4] + on_update = row["on_update"] if isinstance(row, sqlite3.Row) else row[5] + on_delete = row["on_delete"] if isinstance(row, sqlite3.Row) else row[6] + if ( + table_name == "ai_prompt_versions" + and source_column == "prompt_version_id" + and target_column == "id" + and str(on_update).upper() == "NO ACTION" + and str(on_delete).upper() == "NO ACTION" + ): + return True + return False + + def _get_table_columns(connection: sqlite3.Connection, table_name: str) -> set[str]: rows = connection.execute(f"PRAGMA table_info({table_name})").fetchall() return {row["name"] for row in rows} @@ -1531,8 +1638,7 @@ def _verify_douyin_item_export_migration(connection: sqlite3.Connection) -> None def _apply_content_feedback_loop_migration(connection: sqlite3.Connection) -> None: - connection.executescript( - """ + schema_sql = """ CREATE TABLE IF NOT EXISTS content_improvement_experiments ( id TEXT PRIMARY KEY, account_id TEXT NOT NULL, @@ -1573,7 +1679,11 @@ def _apply_content_feedback_loop_migration(connection: sqlite3.Connection) -> No CREATE INDEX IF NOT EXISTS idx_content_experiment_items_experiment ON content_improvement_experiment_items(experiment_id, assigned_at DESC); """ - ) + # executescript() 会先隐式提交，账本迁移必须逐条执行以保持整体可回滚。 + for statement in schema_sql.split(";"): + normalized = statement.strip() + if normalized: + connection.execute(normalized) def _verify_content_feedback_loop_migration(connection: sqlite3.Connection) -> None: @@ -1601,6 +1711,164 @@ def _verify_content_feedback_loop_migration(connection: sqlite3.Connection) -> N ) +def _apply_ai_prompt_version_fk_migration(connection: sqlite3.Connection) -> None: + if _has_ai_prompt_version_fk(connection): + return + if connection.execute("PRAGMA foreign_keys").fetchone()[0] != 0: + raise SchemaMigrationError("重建 AI Run 表前未关闭当前连接的外键检查") + + orphan = connection.execute( + """ + SELECT r.id, r.prompt_version_id + FROM ai_analysis_runs r + LEFT JOIN ai_prompt_versions p ON p.id = r.prompt_version_id + WHERE r.prompt_version_id IS NOT NULL + AND TRIM(r.prompt_version_id) != '' + AND p.id IS NULL + ORDER BY r.id + LIMIT 1 + """ + ).fetchone() + if orphan is not None: + raise SchemaMigrationError( + "AI Run 存在无法验证的 Prompt 版本引用，已拒绝自动重建：" + f"{orphan['id']} -> {orphan['prompt_version_id']}" + ) + + table_info = connection.execute("PRAGMA table_info(ai_analysis_runs)").fetchall() + actual_columns = {row["name"] for row in table_info} + expected_columns = set(AI_ANALYSIS_RUN_COLUMNS) + missing_columns = sorted(expected_columns - actual_columns) + unknown_columns = sorted(actual_columns - expected_columns) + if missing_columns: + raise SchemaMigrationError( + "AI Run 表缺少规范字段，已拒绝自动重建：" + ", ".join(missing_columns) + ) + if unknown_columns: + raise SchemaMigrationError( + "AI Run 表存在未知字段，已拒绝自动重建以避免数据丢失：" + + ", ".join(unknown_columns) + ) + + replacement_table = "ai_analysis_runs_prompt_fk_new" + replacement_exists = connection.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?", + (replacement_table,), + ).fetchone() + if replacement_exists is not None: + raise SchemaMigrationError(f"检测到残留临时表 {replacement_table}，已拒绝覆盖") + + schema_objects = connection.execute( + """ + SELECT type, name, sql + FROM sqlite_master + WHERE tbl_name = 'ai_analysis_runs' + AND type IN ('index', 'trigger') + AND sql IS NOT NULL + ORDER BY type, name + """ + ).fetchall() + original_count = connection.execute("SELECT COUNT(*) FROM ai_analysis_runs").fetchone()[0] + connection.execute( + f""" + CREATE TABLE {replacement_table} ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + run_number INTEGER NOT NULL, + provider TEXT NOT NULL, + provider_label TEXT NOT NULL, + model TEXT NOT NULL, + ai_prompt_preset_id TEXT, + ai_prompt_preset_name TEXT, + prompt_version_id TEXT, + prompt_text_sha256 TEXT, + requested_clip_count INTEGER NOT NULL DEFAULT 5, + clip_count INTEGER NOT NULL DEFAULT 0, + analysis_summary TEXT, + fallback_notice TEXT, + analysis_payload_json TEXT NOT NULL, + created_at TEXT NOT NULL, + is_active INTEGER NOT NULL DEFAULT 1, + FOREIGN KEY(task_id) REFERENCES tasks(id), + FOREIGN KEY(prompt_version_id) REFERENCES ai_prompt_versions(id) + ) + """ + ) + columns_sql = ", ".join(AI_ANALYSIS_RUN_COLUMNS) + connection.execute( + f"INSERT INTO {replacement_table} ({columns_sql}) " + f"SELECT {columns_sql} FROM ai_analysis_runs" + ) + copied_count = connection.execute( + f"SELECT COUNT(*) FROM {replacement_table}" + ).fetchone()[0] + if copied_count != original_count: + raise SchemaMigrationError( + f"AI Run 表重建行数不一致：原表 {original_count}，新表 {copied_count}" + ) + + connection.execute("DROP TABLE ai_analysis_runs") + connection.execute(f"ALTER TABLE {replacement_table} RENAME TO ai_analysis_runs") + for schema_object in schema_objects: + connection.execute(schema_object["sql"]) + + +def _verify_ai_prompt_version_fk_migration(connection: sqlite3.Connection) -> None: + if not _has_ai_prompt_version_fk(connection): + raise SchemaMigrationError("AI Run 的 Prompt 版本外键不存在或删除语义不一致") + task_fk_exists = False + for row in connection.execute("PRAGMA foreign_key_list(ai_analysis_runs)").fetchall(): + if ( + row["table"] == "tasks" + and row["from"] == "task_id" + and row["to"] == "id" + and str(row["on_update"]).upper() == "NO ACTION" + and str(row["on_delete"]).upper() == "NO ACTION" + ): + task_fk_exists = True + break + if not task_fk_exists: + raise SchemaMigrationError("AI Run 表重建后丢失任务外键") + + required_indexes = { + "idx_ai_analysis_runs_task_created", + "idx_ai_analysis_runs_prompt_version", + } + actual_indexes = { + row[0] + for row in connection.execute( + "SELECT name FROM sqlite_master " + "WHERE type='index' AND tbl_name='ai_analysis_runs'" + ).fetchall() + } + missing_indexes = sorted(required_indexes - actual_indexes) + if missing_indexes: + raise SchemaMigrationError( + "AI Run 表重建后缺少索引：" + ", ".join(missing_indexes) + ) + + orphan = connection.execute( + """ + SELECT 1 + FROM ai_analysis_runs r + LEFT JOIN ai_prompt_versions p ON p.id = r.prompt_version_id + WHERE r.prompt_version_id IS NOT NULL + AND TRIM(r.prompt_version_id) != '' + AND p.id IS NULL + LIMIT 1 + """ + ).fetchone() + if orphan is not None: + raise SchemaMigrationError("AI Run 表仍存在孤儿 Prompt 版本引用") + violations = connection.execute("PRAGMA foreign_key_check").fetchall() + if violations: + samples = "; ".join( + f"{row[0]} rowid={row[1]} parent={row[2]}" + for row in violations[:5] + ) + raise SchemaMigrationError("外键检查失败，已拒绝记录迁移账本：" + samples) + + def _registered_schema_migrations() -> tuple[SchemaMigration, ...]: return ( SchemaMigration( @@ -1638,6 +1906,14 @@ def _registered_schema_migrations() -> tuple[SchemaMigration, ...]: apply=_apply_content_feedback_loop_migration, verify=_verify_content_feedback_loop_migration, ), + SchemaMigration( + version=AI_PROMPT_VERSION_FK_MIGRATION_VERSION, + name=AI_PROMPT_VERSION_FK_MIGRATION_NAME, + checksum=AI_PROMPT_VERSION_FK_MIGRATION_CHECKSUM, + apply=_apply_ai_prompt_version_fk_migration, + verify=_verify_ai_prompt_version_fk_migration, + requires_foreign_keys_off=True, + ), ) @@ -1648,7 +1924,14 @@ def _run_schema_migrations(connection: sqlite3.Connection) -> None: for migration in _registered_schema_migrations(): version = migration.version + original_foreign_keys = int(connection.execute("PRAGMA foreign_keys").fetchone()[0]) try: + if migration.requires_foreign_keys_off: + connection.execute("PRAGMA foreign_keys = OFF") + if connection.execute("PRAGMA foreign_keys").fetchone()[0] != 0: + raise SchemaMigrationError( + f"数据库迁移 {version} 无法临时关闭当前连接的外键检查" + ) connection.execute("BEGIN IMMEDIATE") _ensure_schema_migrations_table(connection) applied = connection.execute( @@ -1685,6 +1968,15 @@ def _run_schema_migrations(connection: sqlite3.Connection) -> None: if isinstance(exc, SchemaMigrationError): raise raise SchemaMigrationError(f"数据库迁移 {version} 执行失败：{exc}") from exc + finally: + if migration.requires_foreign_keys_off and original_foreign_keys: + if connection.in_transaction: + connection.rollback() + connection.execute("PRAGMA foreign_keys = ON") + if connection.execute("PRAGMA foreign_keys").fetchone()[0] != 1: + raise SchemaMigrationError( + f"数据库迁移 {version} 后无法恢复当前连接的外键检查" + ) def _migrate_tasks_table(connection: sqlite3.Connection) -> None: diff --git a/docs/DATABASE_SCHEMA.md b/docs/DATABASE_SCHEMA.md index 2e5f8d6..60b5b5e 100644 --- a/docs/DATABASE_SCHEMA.md +++ b/docs/DATABASE_SCHEMA.md @@ -1,5 +1,12 @@ # 数据库结构说明 +## 2026-08-30：迁移原子性与 AI Prompt 外键一致性 + +- 新迁移账本版本为 `20260830_01_ai_prompt_version_fk`。历史库缺少 `ai_analysis_runs.prompt_version_id → ai_prompt_versions.id` 外键时，先生成 `workflow-before-ai-prompt-version-fk-*` 在线备份，再在一个事务内重建 AI Run 表。 +- 重建保留全部规范字段、AI Run 数据、显式索引和触发器；`clip_feedback.analysis_run_id` 等下游引用保持不变。发现未知字段、残留临时表或不存在的 Prompt 版本引用时直接拒绝迁移，不猜测、不清空历史数据。 +- Prompt 外键与新建数据库统一使用 `ON UPDATE NO ACTION / ON DELETE NO ACTION`。结构复制、索引恢复、`PRAGMA foreign_key_check` 与账本写入全部成功后才提交；提交或回滚后恢复当前连接原有的外键检查状态。 +- `20260829_02_content_feedback_loop` 的建表和建索引改为逐条执行，避免 `sqlite3.executescript()` 在账本事务内隐式提交；故障时不会残留半套实验结构或错误账本。 + ## 2026-08-29：内容诊断与实验闭环 迁移账本版本为 `20260829_02_content_feedback_loop`。迁移只新增表和索引，不删除或改写现有作品快照、发布记录、排期或 Prompt 版本；应用前继续通过 SQLite Online Backup 生成 `workflow-before-content-feedback-loop-*` 备份。 diff --git a/tests/test_schema_migration_ledger.py b/tests/test_schema_migration_ledger.py index 4b81da8..1ad37b0 100644 --- a/tests/test_schema_migration_ledger.py +++ b/tests/test_schema_migration_ledger.py @@ -38,6 +38,101 @@ def _index_names(connection): } +def _prompt_version_foreign_keys(connection): + return [ + row + for row in connection.execute("PRAGMA foreign_key_list(ai_analysis_runs)").fetchall() + if row["table"] == "ai_prompt_versions" + and row["from"] == "prompt_version_id" + and row["to"] == "id" + ] + + +def _seed_analysis_run(connection, *, prompt_version_id: str) -> None: + connection.execute( + """ + INSERT INTO tasks ( + id, task_name, source_type, status, created_at, updated_at + ) VALUES ('migration-task', '迁移测试任务', 'upload', 'completed', 'now', 'now') + """ + ) + connection.execute( + """ + INSERT INTO ai_analysis_runs ( + id, task_id, run_number, provider, provider_label, model, + prompt_version_id, prompt_text_sha256, requested_clip_count, + clip_count, analysis_payload_json, created_at, is_active + ) VALUES ( + 'migration-run', 'migration-task', 1, 'test', 'Test', 'test-model', + ?, 'prompt-hash', 1, 1, '{}', 'now', 1 + ) + """, + (prompt_version_id,), + ) + connection.execute( + """ + INSERT INTO clip_feedback ( + id, task_id, clip_candidate_id, analysis_run_id, + selection_profile, decision, reason_code, created_at + ) VALUES ( + 'migration-feedback', 'migration-task', 'migration-clip', + 'migration-run', 'general', 'keep', 'worth_publishing', 'now' + ) + """ + ) + + +def _rebuild_ai_runs_without_prompt_fk(connection) -> None: + schema_objects = connection.execute( + """ + SELECT sql FROM sqlite_master + WHERE tbl_name = 'ai_analysis_runs' + AND type IN ('index', 'trigger') + AND sql IS NOT NULL + ORDER BY type, name + """ + ).fetchall() + connection.execute( + """ + CREATE TABLE ai_analysis_runs_without_prompt_fk ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + run_number INTEGER NOT NULL, + provider TEXT NOT NULL, + provider_label TEXT NOT NULL, + model TEXT NOT NULL, + ai_prompt_preset_id TEXT, + ai_prompt_preset_name TEXT, + prompt_version_id TEXT, + prompt_text_sha256 TEXT, + requested_clip_count INTEGER NOT NULL DEFAULT 5, + clip_count INTEGER NOT NULL DEFAULT 0, + analysis_summary TEXT, + fallback_notice TEXT, + analysis_payload_json TEXT NOT NULL, + created_at TEXT NOT NULL, + is_active INTEGER NOT NULL DEFAULT 1, + FOREIGN KEY(task_id) REFERENCES tasks(id) + ) + """ + ) + columns_sql = ", ".join(database_module.AI_ANALYSIS_RUN_COLUMNS) + connection.execute( + f"INSERT INTO ai_analysis_runs_without_prompt_fk ({columns_sql}) " + f"SELECT {columns_sql} FROM ai_analysis_runs" + ) + connection.execute("DROP TABLE ai_analysis_runs") + connection.execute( + "ALTER TABLE ai_analysis_runs_without_prompt_fk RENAME TO ai_analysis_runs" + ) + for schema_object in schema_objects: + connection.execute(schema_object["sql"]) + connection.execute( + "DELETE FROM schema_migrations WHERE version = ?", + (database_module.AI_PROMPT_VERSION_FK_MIGRATION_VERSION,), + ) + + def test_init_records_migration_once_and_switches_unique_index(isolated_database): database_module.init_db() database_module.init_db() @@ -48,7 +143,7 @@ def test_init_records_migration_once_and_switches_unique_index(isolated_database ).fetchall() indexes = _index_names(connection) - assert len(migrations) == 5 + assert len(migrations) == 6 migrations_by_version = {row["version"]: row for row in migrations} publish_migration = migrations_by_version[database_module.PUBLISH_ACTIVE_INDEX_MIGRATION_VERSION] assert publish_migration["name"] == database_module.PUBLISH_ACTIVE_INDEX_MIGRATION_NAME @@ -70,6 +165,15 @@ def test_init_records_migration_once_and_switches_unique_index(isolated_database assert feedback_migration["name"] == database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_NAME assert feedback_migration["checksum"] == database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_CHECKSUM assert feedback_migration["applied_at"] + prompt_fk_migration = migrations_by_version[ + database_module.AI_PROMPT_VERSION_FK_MIGRATION_VERSION + ] + assert prompt_fk_migration["name"] == database_module.AI_PROMPT_VERSION_FK_MIGRATION_NAME + assert ( + prompt_fk_migration["checksum"] + == database_module.AI_PROMPT_VERSION_FK_MIGRATION_CHECKSUM + ) + assert prompt_fk_migration["applied_at"] assert database_module.PUBLISH_ACTIVE_UNIQUE_INDEX_NAME in indexes assert database_module.PUBLISH_ACTIVE_UNIQUE_INDEX_LEGACY_NAME not in indexes assert set(database_module.CONTENT_REVIEW_REQUIRED_INDEXES) <= indexes @@ -87,6 +191,225 @@ def test_init_records_migration_once_and_switches_unique_index(isolated_database } assert set(database_module.DOUYIN_ITEM_EXPORT_COLUMNS) <= item_columns assert set(database_module.CONTENT_FEEDBACK_LOOP_REQUIRED_TABLES) <= tables + with _connect(isolated_database) as connection: + prompt_fks = _prompt_version_foreign_keys(connection) + violations = connection.execute("PRAGMA foreign_key_check").fetchall() + assert len(prompt_fks) == 1 + assert prompt_fks[0]["on_update"] == "NO ACTION" + assert prompt_fks[0]["on_delete"] == "NO ACTION" + assert violations == [] + + +def test_content_feedback_migration_failure_rolls_back_and_can_retry(isolated_database): + database_module.init_db() + with _connect(isolated_database) as connection: + connection.execute( + "DELETE FROM schema_migrations WHERE version = ?", + (database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_VERSION,), + ) + connection.execute("DROP TABLE content_improvement_experiment_items") + connection.execute("DROP TABLE content_improvement_experiments") + connection.commit() + + def deny_second_table(action, object_name, _arg2, _db_name, _source): + if ( + action == sqlite3.SQLITE_CREATE_TABLE + and object_name == "content_improvement_experiment_items" + ): + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + connection.set_authorizer(deny_second_table) + with pytest.raises(database_module.SchemaMigrationError, match="not authorized"): + database_module._run_schema_migrations(connection) + connection.set_authorizer(None) + + tables_after_failure = { + row["name"] + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + indexes_after_failure = _index_names(connection) + ledger_after_failure = connection.execute( + "SELECT COUNT(*) FROM schema_migrations WHERE version = ?", + (database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_VERSION,), + ).fetchone()[0] + assert "content_improvement_experiments" not in tables_after_failure + assert "content_improvement_experiment_items" not in tables_after_failure + assert not set(database_module.CONTENT_FEEDBACK_LOOP_REQUIRED_INDEXES) <= indexes_after_failure + assert ledger_after_failure == 0 + + database_module._run_schema_migrations(connection) + tables_after_retry = { + row["name"] + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + ledger_after_retry = connection.execute( + "SELECT COUNT(*) FROM schema_migrations WHERE version = ?", + (database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_VERSION,), + ).fetchone()[0] + assert set(database_module.CONTENT_FEEDBACK_LOOP_REQUIRED_TABLES) <= tables_after_retry + assert set(database_module.CONTENT_FEEDBACK_LOOP_REQUIRED_INDEXES) <= _index_names( + connection + ) + assert ledger_after_retry == 1 + + +def test_legacy_ai_runs_gain_prompt_fk_without_losing_data(isolated_database): + database_module.init_db() + with _connect(isolated_database) as connection: + prompt_version_id = connection.execute( + "SELECT id FROM ai_prompt_versions ORDER BY created_at, id LIMIT 1" + ).fetchone()[0] + connection.execute("CREATE TABLE migration_run_audit (run_id TEXT NOT NULL)") + connection.execute( + """ + CREATE TRIGGER migration_run_audit_trigger + AFTER INSERT ON ai_analysis_runs + BEGIN + INSERT INTO migration_run_audit (run_id) VALUES (NEW.id); + END + """ + ) + _seed_analysis_run(connection, prompt_version_id=prompt_version_id) + _rebuild_ai_runs_without_prompt_fk(connection) + connection.commit() + assert _prompt_version_foreign_keys(connection) == [] + + database_module.init_db() + + with _connect(isolated_database) as connection: + connection.execute("PRAGMA foreign_keys = ON") + run = connection.execute( + "SELECT id, task_id, prompt_version_id, is_active FROM ai_analysis_runs " + "WHERE id = 'migration-run'" + ).fetchone() + feedback_run_id = connection.execute( + "SELECT analysis_run_id FROM clip_feedback WHERE id = 'migration-feedback'" + ).fetchone()[0] + ledger_count = connection.execute( + "SELECT COUNT(*) FROM schema_migrations WHERE version = ?", + (database_module.AI_PROMPT_VERSION_FK_MIGRATION_VERSION,), + ).fetchone()[0] + prompt_fks = _prompt_version_foreign_keys(connection) + violations = connection.execute("PRAGMA foreign_key_check").fetchall() + run_indexes = _index_names(connection) + trigger_exists = connection.execute( + "SELECT 1 FROM sqlite_master WHERE type='trigger' " + "AND name='migration_run_audit_trigger'" + ).fetchone() + with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY constraint failed"): + connection.execute( + "DELETE FROM ai_prompt_versions WHERE id = ?", + (prompt_version_id,), + ) + + backups = list( + (isolated_database.parent / "backups").glob( + "workflow-before-ai-prompt-version-fk-*.sqlite3" + ) + ) + assert dict(run) == { + "id": "migration-run", + "task_id": "migration-task", + "prompt_version_id": prompt_version_id, + "is_active": 1, + } + assert feedback_run_id == "migration-run" + assert ledger_count == 1 + assert len(prompt_fks) == 1 + assert prompt_fks[0]["on_update"] == "NO ACTION" + assert prompt_fks[0]["on_delete"] == "NO ACTION" + assert violations == [] + assert { + "idx_ai_analysis_runs_task_created", + "idx_ai_analysis_runs_prompt_version", + } <= run_indexes + assert trigger_exists is not None + assert len(backups) == 1 + + +def test_prompt_fk_rebuild_failure_rolls_back_and_restores_foreign_keys(isolated_database): + database_module.init_db() + with _connect(isolated_database) as connection: + prompt_version_id = connection.execute( + "SELECT id FROM ai_prompt_versions ORDER BY created_at, id LIMIT 1" + ).fetchone()[0] + _seed_analysis_run(connection, prompt_version_id=prompt_version_id) + _rebuild_ai_runs_without_prompt_fk(connection) + connection.commit() + connection.execute("PRAGMA foreign_keys = ON") + + def deny_recreated_index(action, object_name, _arg2, _db_name, _source): + if ( + action == sqlite3.SQLITE_CREATE_INDEX + and object_name == "idx_ai_analysis_runs_prompt_version" + ): + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + connection.set_authorizer(deny_recreated_index) + with pytest.raises(database_module.SchemaMigrationError, match="not authorized"): + database_module._run_schema_migrations(connection) + connection.set_authorizer(None) + + assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1 + assert _prompt_version_foreign_keys(connection) == [] + assert connection.execute( + "SELECT prompt_version_id FROM ai_analysis_runs WHERE id = 'migration-run'" + ).fetchone()[0] == prompt_version_id + assert connection.execute( + "SELECT COUNT(*) FROM schema_migrations WHERE version = ?", + (database_module.AI_PROMPT_VERSION_FK_MIGRATION_VERSION,), + ).fetchone()[0] == 0 + assert connection.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' " + "AND name='ai_analysis_runs_prompt_fk_new'" + ).fetchone() is None + + database_module._run_schema_migrations(connection) + assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1 + assert len(_prompt_version_foreign_keys(connection)) == 1 + assert connection.execute( + "SELECT COUNT(*) FROM schema_migrations WHERE version = ?", + (database_module.AI_PROMPT_VERSION_FK_MIGRATION_VERSION,), + ).fetchone()[0] == 1 + assert connection.execute("PRAGMA foreign_key_check").fetchall() == [] + + +def test_prompt_fk_migration_rejects_orphan_reference_without_rewrite(isolated_database): + database_module.init_db() + with _connect(isolated_database) as connection: + prompt_version_id = connection.execute( + "SELECT id FROM ai_prompt_versions ORDER BY created_at, id LIMIT 1" + ).fetchone()[0] + _seed_analysis_run(connection, prompt_version_id=prompt_version_id) + _rebuild_ai_runs_without_prompt_fk(connection) + connection.execute( + "UPDATE ai_analysis_runs SET prompt_version_id = 'missing-prompt-version' " + "WHERE id = 'migration-run'" + ) + connection.commit() + connection.execute("PRAGMA foreign_keys = ON") + + with pytest.raises( + database_module.SchemaMigrationError, + match="无法验证的 Prompt 版本引用", + ): + database_module._run_schema_migrations(connection) + + assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1 + assert _prompt_version_foreign_keys(connection) == [] + assert connection.execute( + "SELECT prompt_version_id FROM ai_analysis_runs WHERE id = 'migration-run'" + ).fetchone()[0] == "missing-prompt-version" + assert connection.execute( + "SELECT COUNT(*) FROM schema_migrations WHERE version = ?", + (database_module.AI_PROMPT_VERSION_FK_MIGRATION_VERSION,), + ).fetchone()[0] == 0 def test_checksum_drift_refuses_startup(isolated_database):