Migrate scan store to shared entities

README.md

@@ -16,7 +16,9 @@
 - 递归扫描一个或多个目录中的常见视频扩展名，默认覆盖 `mp4/mov/m4v/mkv/webm/avi/flv/f4v/ts/m2ts/mts/mpg/mpeg/mpe/mpv/m2v/wmv/asf/3gp/3g2/ogv/ogm/vob/qt/rm/rmvb/mxf/divx`
 - 支持从 `txt` 文件批量导入目录列表
 - 使用 SQLite 落盘扫描状态，并基于 `path + size + mtime` 复用旧结果
+- 数据库存储已改为“扫描批次 + 文件实体”两层模型：同一个未变化文件的 stage 结果跨 run 只保存一份
 - 当同一组 `roots + profile` 的扫描上次未完成时，`scan --resume` 会直接续跑同一个 run，不再把 discovery 结果整轮复制一遍
+- 当后续递归扫描的 roots 完全覆盖旧历史 run，例如先扫 `a\b` 再扫 `a` 时，命中的旧 run 会在当前 `scan --resume` 里自动压扁进当前 run
 - stage1 使用 `ffprobe` 做容器级快速校验
 - `scan` 默认只执行 stage1，把需要详细复核的文件标记为 `pending_stage2`
 - `verify-stage2` 只对 `pending_stage2` 文件使用 `ffmpeg` 做定向采样或完整确认
@@ -145,6 +147,7 @@ tgvideo-check scan "D:\extra-folder" --paths-file .\dirs.txt --db .\scan_state.d
 - `verify-stage2 --full-decode-confirm`：对第二轮强制使用完整解码确认
 - `scan --resume / --no-resume`：控制是否复用旧的第一轮结果
 - `compact-db`：清理同一组 `roots + profile` 下的重复历史 run，并可选执行 `VACUUM`
+- 打开旧版数据库时会自动迁移到新的实体去重存储模型
 
 命令边界：
 
@@ -162,6 +165,7 @@ mp4,mov,m4v,mkv,webm,avi,flv,f4v,ts,m2ts,mts,mpg,mpeg,mpe,mpv,m2v,wmv,asf,3gp,3g
 
 - `tg-hdd` / `tg-fast` 默认不再因为文件后缀直接升级到第二轮，只有 `ffprobe` 已经发现异常信号时才进入 `pending_stage2`
 - 如果你希望对全部已支持格式都走更激进的二次校验，使用 `--profile deep-verify`
+- 自动压扁历史 run 只在 `scan` 的递归模式下触发；如果旧 run 中有文件 identity 已变化，则该旧 run 不会被自动删除
 
 ### `--paths-file` 规则
 
@@ -262,8 +266,23 @@ tgvideo-check compact-db --db .\scan_state.db --run-id 12 --no-dry-run
 
 - 默认保留最新 run；如果你在做中断恢复，通常就保留当前正在续跑的那个 run
 - 这个命令只清理同一组 `roots + profile` 下的重复历史 run，不会顺手删除别的扫描任务
+- 真实删除时还会顺手清掉没有任何 run 引用的孤儿实体与孤儿 stage 结果
 - 默认会在真实删除后执行 `VACUUM`，把 SQLite 文件物理收缩下来
 
+### 旧库迁移
+
+如果你手头已经有旧版 `scan_state.db`，不需要手工导表。当前版本在第一次打开数据库时会自动完成：
+
+- 将旧版按 `file_id` 存储的 `stage1_results / stage2_results` 迁移成按文件实体存储
+- 对同一个 `path + size + mtime + extension` 的重复 stage 结果做去重
+- 保留各个 run 自己的状态行，只把可共享的检测结果合并
+
+如果你还想顺手收缩物理文件，迁移后再跑一次：
+
+```powershell
+tgvideo-check compact-db --db .\scan_state.db --no-dry-run
+```
+
 ### 原因码说明
 
 导出的 `reason_code` / `reason_code_zh` 含义如下：

docs/plans/2026-03-30-entity-based-store-migration.md

@@ -0,0 +1,133 @@
+# Entity-Based Store Migration Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** 将当前按 run 复制的 SQLite 存储改成按文件实体去重存储，并支持把旧库迁移到新模型，保留最有价值的检测结果。
+
+**Architecture:** 保留 `scan_runs` 作为扫描批次概念，引入 `file_entities` 作为跨 run 共享的唯一文件实体，引入 `run_files` 作为“某次扫描里看到了哪个实体”的成员关系。`stage1_results` / `stage2_results` 改为按 `entity_id` 存储，旧 schema 在 `initialize()` 时自动迁移并去重。
+
+**Tech Stack:** Python 3.12+, sqlite3, typer, orjson, pytest
+
+---
+
+### Task 1: 写迁移与实体存储失败测试
+
+**Files:**
+- Modify: `tests/test_resume_logic.py`
+- Modify: `tests/test_cli.py`
+
+**Step 1: 写旧 schema 迁移后实体去重的失败测试**
+
+- 构造旧版 `files/stage1_results/stage2_results` schema
+- 插入两个 run 中同一个 `path+size+mtime` 的重复记录
+- 调用 `ScanStore.initialize()`
+- 断言新 schema 存在，实体表只有 1 条，共享的 `stage1_results/stage2_results` 也只有 1 条
+
+**Step 2: 写新 schema 下复用结果不再复制 `stage1_results` 的失败测试**
+
+- 创建两个 run，插入同一个实体
+- 为第一个 run 写入 stage1 结果
+- 第二个 run 复用后断言 `run_files` 有 2 条，但 `stage1_results` 仍只有 1 条
+
+**Step 3: 写 `compact-db` 会清理孤儿实体/结果的失败测试**
+
+- 造两个重复 run
+- 执行压缩后断言被删除 run 对应的 `run_files` 消失
+- 如果某实体已无任何 run 引用，则其 `file_entities/stage1_results/stage2_results` 被一并清掉
+
+**Step 4: 跑聚焦测试确认失败**
+
+Run: `pytest tests/test_resume_logic.py tests/test_cli.py -k "entity or migrate or compact" -q`
+Expected: FAIL，失败点落在缺失的新 schema / 新行为
+
+### Task 2: 引入新 schema 与兼容模型
+
+**Files:**
+- Modify: `src/tgvideo_check/models.py`
+- Modify: `src/tgvideo_check/store.py`
+
+**Step 1: 定义新的实体/运行时数据模型**
+
+- 在 `models.py` 中补充实体级字段映射所需结构
+- 保持 `StoredFile` 对上层调用接口尽量稳定，`file_id` 继续表示 run 级成员记录 ID
+
+**Step 2: 在 `store.py.initialize()` 中创建新 schema**
+
+- 新建 `file_entities`
+- 新建 `run_files`
+- 将 `stage1_results/stage2_results` 改为 `entity_id` 主键
+- 为实体身份、run 顺序、run 路径建立索引
+
+**Step 3: 在 `initialize()` 中探测并迁移旧 schema**
+
+- 如果仍存在旧版 `files` 表结构，则迁移到新表
+- 迁移后删除旧表，避免双份存储
+- 对重复实体只保留一份 stage 结果，优先保留“最新且信息更完整”的记录
+
+### Task 3: 重写存储读写路径
+
+**Files:**
+- Modify: `src/tgvideo_check/store.py`
+
+**Step 1: 重写 `upsert_files()`**
+
+- 先按 `path+size+mtime+extension` 获取或创建实体
+- 再按 `run_id+path` 写入/更新 `run_files`
+- 复用历史状态时只复用 run 级状态字段，不再复制 stage 结果表
+
+**Step 2: 重写 stage1/stage2 结果写入**
+
+- `record_stage1_result()` 通过 `run_file_id -> entity_id` 写实体结果
+- `record_stage2_result()` 同理
+- 同一个实体跨 run 只保留一条 stage 结果
+
+**Step 3: 重写读路径**
+
+- `list_files()`
+- `list_stage1_debug_rows()`
+- `get_stage1_stdout_text()`
+- `compact_duplicate_runs()`
+
+### Task 4: 调整 CLI 与压缩行为
+
+**Files:**
+- Modify: `src/tgvideo_check/cli.py`
+- Modify: `README.md`
+
+**Step 1: 保持现有 CLI 输出兼容**
+
+- `scan`
+- `verify-stage2`
+- `inspect-stage1`
+- `compact-db`
+
+**Step 2: 让 `compact-db` 在删除 run 后清理孤儿实体与孤儿 stage 结果**
+
+- 压缩后再执行 `VACUUM`
+- 输出删除的 run 数、run_files 数、实体数、stage 结果数
+
+**Step 3: 文档更新**
+
+- 说明数据库已改为实体去重存储
+- 说明旧库会自动迁移
+- 说明 `compact-db` 现在还会清孤儿实体
+
+### Task 5: 验证
+
+**Files:**
+- Test: `tests/test_resume_logic.py`
+- Test: `tests/test_cli.py`
+- Test: `tests/test_integration_resume.py`
+
+**Step 1: 跑新增聚焦测试**
+
+Run: `pytest tests/test_resume_logic.py tests/test_cli.py -k "entity or migrate or compact" -q`
+
+**Step 2: 跑 resume 相关回归**
+
+Run: `pytest tests/test_resume_logic.py tests/test_integration_resume.py tests/test_cli.py -k "resume or compact" -q`
+
+**Step 3: 若环境允许，再跑更大范围回归**
+
+Run: `pytest -q`
+

src/tgvideo_check/cli.py

@@ -400,6 +400,7 @@ def compact_db(
         f"keep_run_id={summary.keep_run_id} "
         f"deleted_runs={summary.deleted_run_count} "
         f"deleted_files={summary.deleted_file_count} "
+        f"deleted_entities={summary.deleted_entity_count} "
         f"deleted_stage1_results={summary.deleted_stage1_count} "
         f"deleted_stage2_results={summary.deleted_stage2_count} "
         f"dry_run={'yes' if summary.dry_run else 'no'} "
@@ -446,6 +447,18 @@ def _execute_stage1_scan(config: ScanConfig) -> list[ReportRow]:
             reuse_previous_results=config.resume,
             reuse_stage2_results=False,
         )
+        compacted_run_ids: tuple[int, ...] = ()
+        if config.resume:
+            compacted_run_ids = store.compact_covered_runs(
+                run_id,
+                root_paths=config.roots,
+                recursive=config.recursive,
+            )
+            if compacted_run_ids:
+                typer.echo(
+                    "Resume: compacted covered historical runs "
+                    f"into run_id={run_id} deleted_run_ids={','.join(str(run) for run in compacted_run_ids)}"
+                )
         progress.discovery_completed(stored_files)
 
         stage1_files = select_pending_stage1_files(stored_files)

src/tgvideo_check/store.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import sqlite3
 from dataclasses import dataclass
 from pathlib import Path
@@ -22,6 +23,7 @@ class RunCompactionSummary:
     keep_run_id: int
     duplicate_run_ids: tuple[int, ...]
     deleted_file_count: int
+    deleted_entity_count: int
     deleted_stage1_count: int
     deleted_stage2_count: int
     dry_run: bool
@@ -38,60 +40,17 @@ class ScanStore:
 
     def initialize(self) -> None:
         with self._connect() as connection:
-            connection.executescript(
+            self._create_base_tables(connection)
-                """
-                CREATE TABLE IF NOT EXISTS scan_runs (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                    profile TEXT NOT NULL,
-                    root_paths_json TEXT NOT NULL,
-                    status TEXT NOT NULL DEFAULT 'running'
-                );
-
-                CREATE TABLE IF NOT EXISTS files (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    run_id INTEGER NOT NULL,
-                    path TEXT NOT NULL,
-                    size INTEGER NOT NULL,
-                    mtime_ns INTEGER NOT NULL,
-                    extension TEXT NOT NULL,
-                    discovery_order INTEGER NOT NULL,
-                    stage1_status TEXT NOT NULL,
-                    stage2_status TEXT,
-                    final_status TEXT,
-                    resumed_from_run_id INTEGER,
-                    reason_code TEXT,
-                    UNIQUE (run_id, path),
-                    FOREIGN KEY (run_id) REFERENCES scan_runs(id)
-                );
-
-                CREATE TABLE IF NOT EXISTS stage1_results (
-                    file_id INTEGER PRIMARY KEY,
-                    exit_code INTEGER,
-                    stdout_text TEXT,
-                    stderr_text TEXT,
-                    elapsed_ms INTEGER,
-                    FOREIGN KEY (file_id) REFERENCES files(id)
-                );
-
-                CREATE TABLE IF NOT EXISTS stage2_results (
-                    file_id INTEGER PRIMARY KEY,
-                    mode TEXT,
-                    sample_points_json TEXT,
-                    exit_code INTEGER,
-                    stderr_text TEXT,
-                    reason_code TEXT,
-                    elapsed_ms INTEGER,
-                    FOREIGN KEY (file_id) REFERENCES files(id)
-                );
-
-                CREATE INDEX IF NOT EXISTS idx_files_identity
-                    ON files(path, size, mtime_ns);
-                CREATE INDEX IF NOT EXISTS idx_files_run_order
-                    ON files(run_id, discovery_order);
-                """
-            )
             self._ensure_column(connection, "files", "reason_code", "TEXT")
+            self._ensure_column(connection, "files", "entity_id", "INTEGER")
+            self._ensure_index(
+                connection,
+                "idx_files_entity_id",
+                "CREATE INDEX IF NOT EXISTS idx_files_entity_id ON files(entity_id)",
+            )
+            self._backfill_file_entities(connection)
+            self._migrate_stage1_results_to_entities(connection)
+            self._migrate_stage2_results_to_entities(connection)
 
     def create_run(self, *, profile: str, root_paths: tuple[Path, ...]) -> int:
         root_paths_json = self._serialize_root_paths(root_paths)
@@ -166,9 +125,17 @@ class ScanStore:
     ) -> list[StoredFile]:
         with self._connect() as connection:
             for candidate in candidates:
+                entity_id = self._ensure_entity(
+                    connection,
+                    path=candidate.path,
+                    size=candidate.size,
+                    mtime_ns=candidate.mtime_ns,
+                    extension=candidate.extension,
+                )
                 current = self._find_current_file(connection, run_id, candidate.path)
                 current_matches = (
                     current is not None
+                    and current["entity_id"] == entity_id
                     and self._row_matches_candidate(current, candidate)
                 )
                 previous = None
@@ -184,7 +151,7 @@ class ScanStore:
                         self._find_previous_match(
                             connection,
                             run_id,
-                            candidate,
+                            entity_id=entity_id,
                             allow_stage2_match=reuse_stage2_results,
                         )
                         if reuse_previous_results
@@ -204,6 +171,7 @@ class ScanStore:
                         size,
                         mtime_ns,
                         extension,
+                        entity_id,
                         discovery_order,
                         stage1_status,
                         stage2_status,
@@ -211,11 +179,12 @@ class ScanStore:
                         resumed_from_run_id,
                         reason_code
                     )
-                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                     ON CONFLICT(run_id, path) DO UPDATE SET
                         size = excluded.size,
                         mtime_ns = excluded.mtime_ns,
                         extension = excluded.extension,
+                        entity_id = excluded.entity_id,
                         discovery_order = excluded.discovery_order,
                         stage1_status = excluded.stage1_status,
                         stage2_status = excluded.stage2_status,
@@ -229,6 +198,7 @@ class ScanStore:
                         candidate.size,
                         candidate.mtime_ns,
                         candidate.extension,
+                        entity_id,
                         candidate.discovery_order,
                         stage1_status,
                         stage2_status,
@@ -237,31 +207,6 @@ class ScanStore:
                         reason_code,
                     ),
                 )
-                current_file_id = (
-                    int(current["file_id"])
-                    if current is not None
-                    else int(
-                        connection.execute(
-                            """
-                            SELECT id
-                            FROM files
-                            WHERE run_id = ? AND path = ?
-                            """,
-                            (run_id, str(candidate.path)),
-                        ).fetchone()["id"]
-                    )
-                )
-
-                if current is not None and not current_matches:
-                    self._clear_file_results(connection, current_file_id)
-
-                if previous is not None:
-                    self._copy_previous_results(
-                        connection,
-                        previous_file_id=previous["file_id"],
-                        current_file_id=current_file_id,
-                        copy_stage2_results=reuse_stage2_results,
-                    )
 
             rows = connection.execute(
                 """
@@ -335,17 +280,18 @@ class ScanStore:
         elapsed_ms: int,
     ) -> None:
         with self._connect() as connection:
+            entity_id = self._get_entity_id_for_file(connection, file_id)
             connection.execute(
                 """
-                INSERT INTO stage1_results(file_id, exit_code, stdout_text, stderr_text, elapsed_ms)
+                INSERT INTO stage1_results(entity_id, exit_code, stdout_text, stderr_text, elapsed_ms)
                 VALUES (?, ?, ?, ?, ?)
-                ON CONFLICT(file_id) DO UPDATE SET
+                ON CONFLICT(entity_id) DO UPDATE SET
                     exit_code = excluded.exit_code,
                     stdout_text = excluded.stdout_text,
                     stderr_text = excluded.stderr_text,
                     elapsed_ms = excluded.elapsed_ms
                 """,
-                (file_id, exit_code, stdout_text, stderr_text, elapsed_ms),
+                (entity_id, exit_code, stdout_text, stderr_text, elapsed_ms),
             )
 
     def record_stage2_status(
@@ -378,10 +324,11 @@ class ScanStore:
         elapsed_ms: int,
     ) -> None:
         with self._connect() as connection:
+            entity_id = self._get_entity_id_for_file(connection, file_id)
             connection.execute(
                 """
                 INSERT INTO stage2_results(
-                    file_id,
+                    entity_id,
                     mode,
                     sample_points_json,
                     exit_code,
@@ -390,7 +337,7 @@ class ScanStore:
                     elapsed_ms
                 )
                 VALUES (?, ?, ?, ?, ?, ?, ?)
-                ON CONFLICT(file_id) DO UPDATE SET
+                ON CONFLICT(entity_id) DO UPDATE SET
                     mode = excluded.mode,
                     sample_points_json = excluded.sample_points_json,
                     exit_code = excluded.exit_code,
@@ -399,7 +346,7 @@ class ScanStore:
                     elapsed_ms = excluded.elapsed_ms
                 """,
                 (
-                    file_id,
+                    entity_id,
                     mode,
                     sample_points_json,
                     exit_code,
@@ -424,9 +371,10 @@ class ScanStore:
         with self._connect() as connection:
             row = connection.execute(
                 """
-                SELECT stdout_text
+                SELECT r.stdout_text
-                FROM stage1_results
+                FROM stage1_results AS r
-                WHERE file_id = ?
+                JOIN files AS f ON f.entity_id = r.entity_id
+                WHERE f.id = ?
                 """,
                 (file_id,),
             ).fetchone()
@@ -480,7 +428,7 @@ class ScanStore:
                 r.stdout_text,
                 r.stderr_text
             FROM files AS f
-            JOIN stage1_results AS r ON r.file_id = f.id
+            JOIN stage1_results AS r ON r.entity_id = f.entity_id
             WHERE f.run_id = ?
         """
         params: list[object] = [run_id]
@@ -526,6 +474,7 @@ class ScanStore:
 
         duplicate_run_ids: tuple[int, ...]
         deleted_file_count = 0
+        deleted_entity_count = 0
         deleted_stage1_count = 0
         deleted_stage2_count = 0
 
@@ -553,58 +502,8 @@ class ScanStore:
                         duplicate_run_ids,
                     ).fetchone()[0]
                 )
-                deleted_stage1_count = int(
-                    connection.execute(
-                        f"""
-                        SELECT COUNT(*)
-                        FROM stage1_results
-                        WHERE file_id IN (
-                            SELECT id
-                            FROM files
-                            WHERE run_id IN ({placeholders})
-                        )
-                        """,
-                        duplicate_run_ids,
-                    ).fetchone()[0]
-                )
-                deleted_stage2_count = int(
-                    connection.execute(
-                        f"""
-                        SELECT COUNT(*)
-                        FROM stage2_results
-                        WHERE file_id IN (
-                            SELECT id
-                            FROM files
-                            WHERE run_id IN ({placeholders})
-                        )
-                        """,
-                        duplicate_run_ids,
-                    ).fetchone()[0]
-                )
 
                 if not dry_run:
-                    connection.execute(
-                        f"""
-                        DELETE FROM stage1_results
-                        WHERE file_id IN (
-                            SELECT id
-                            FROM files
-                            WHERE run_id IN ({placeholders})
-                        )
-                        """,
-                        duplicate_run_ids,
-                    )
-                    connection.execute(
-                        f"""
-                        DELETE FROM stage2_results
-                        WHERE file_id IN (
-                            SELECT id
-                            FROM files
-                            WHERE run_id IN ({placeholders})
-                        )
-                        """,
-                        duplicate_run_ids,
-                    )
                     connection.execute(
                         f"""
                         DELETE FROM files
@@ -619,6 +518,11 @@ class ScanStore:
                         """,
                         duplicate_run_ids,
                     )
+                    (
+                        deleted_entity_count,
+                        deleted_stage1_count,
+                        deleted_stage2_count,
+                    ) = self._delete_orphan_entities(connection)
 
         vacuumed = bool(duplicate_run_ids) and not dry_run and vacuum
         if vacuumed:
@@ -629,12 +533,74 @@ class ScanStore:
             keep_run_id=keep_run_id,
             duplicate_run_ids=duplicate_run_ids,
             deleted_file_count=deleted_file_count,
+            deleted_entity_count=deleted_entity_count,
             deleted_stage1_count=deleted_stage1_count,
             deleted_stage2_count=deleted_stage2_count,
             dry_run=dry_run,
             vacuumed=vacuumed,
         )
 
+    def compact_covered_runs(
+        self,
+        keep_run_id: int,
+        *,
+        root_paths: tuple[Path, ...],
+        recursive: bool,
+    ) -> tuple[int, ...]:
+        if not recursive:
+            return ()
+
+        keep_run = self.get_run(keep_run_id)
+        if keep_run is None:
+            raise ValueError(f"Scan run does not exist: {keep_run_id}")
+
+        with self._connect() as connection:
+            candidate_rows = connection.execute(
+                """
+                SELECT id, root_paths_json
+                FROM scan_runs
+                WHERE profile = ? AND id <> ?
+                ORDER BY id ASC
+                """,
+                (keep_run.profile, keep_run_id),
+            ).fetchall()
+
+            covered_run_ids: list[int] = []
+            for row in candidate_rows:
+                candidate_run_id = int(row["id"])
+                candidate_roots = self._deserialize_root_paths(row["root_paths_json"])
+                if not self._roots_are_covered(candidate_roots, root_paths):
+                    continue
+                if not self._run_files_are_covered(
+                    connection,
+                    keep_run_id=keep_run_id,
+                    covered_run_id=candidate_run_id,
+                ):
+                    continue
+                covered_run_ids.append(candidate_run_id)
+
+            if not covered_run_ids:
+                return ()
+
+            placeholders = ",".join("?" for _ in covered_run_ids)
+            connection.execute(
+                f"""
+                DELETE FROM files
+                WHERE run_id IN ({placeholders})
+                """,
+                tuple(covered_run_ids),
+            )
+            connection.execute(
+                f"""
+                DELETE FROM scan_runs
+                WHERE id IN ({placeholders})
+                """,
+                tuple(covered_run_ids),
+            )
+            self._delete_orphan_entities(connection)
+
+        return tuple(covered_run_ids)
+
     def _rows_to_stored_files(self, rows: list[sqlite3.Row]) -> list[StoredFile]:
         return [
             StoredFile(
@@ -659,9 +625,337 @@ class ScanStore:
         connection.row_factory = sqlite3.Row
         return connection
 
+    def _create_base_tables(self, connection: sqlite3.Connection) -> None:
+        connection.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS scan_runs (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                profile TEXT NOT NULL,
+                root_paths_json TEXT NOT NULL,
+                status TEXT NOT NULL DEFAULT 'running'
+            );
+
+            CREATE TABLE IF NOT EXISTS file_entities (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                path TEXT NOT NULL,
+                size INTEGER NOT NULL,
+                mtime_ns INTEGER NOT NULL,
+                extension TEXT NOT NULL,
+                UNIQUE (path, size, mtime_ns, extension)
+            );
+
+            CREATE TABLE IF NOT EXISTS files (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                run_id INTEGER NOT NULL,
+                path TEXT NOT NULL,
+                size INTEGER NOT NULL,
+                mtime_ns INTEGER NOT NULL,
+                extension TEXT NOT NULL,
+                entity_id INTEGER,
+                discovery_order INTEGER NOT NULL,
+                stage1_status TEXT NOT NULL,
+                stage2_status TEXT,
+                final_status TEXT,
+                resumed_from_run_id INTEGER,
+                reason_code TEXT,
+                UNIQUE (run_id, path),
+                FOREIGN KEY (run_id) REFERENCES scan_runs(id),
+                FOREIGN KEY (entity_id) REFERENCES file_entities(id)
+            );
+
+            CREATE INDEX IF NOT EXISTS idx_files_identity
+                ON files(path, size, mtime_ns);
+            CREATE INDEX IF NOT EXISTS idx_files_run_order
+                ON files(run_id, discovery_order);
+            CREATE INDEX IF NOT EXISTS idx_file_entities_identity
+                ON file_entities(path, size, mtime_ns, extension);
+            """
+        )
+
     def _serialize_root_paths(self, root_paths: tuple[Path, ...]) -> str:
         return orjson.dumps([str(path) for path in root_paths]).decode("utf-8")
 
+    def _deserialize_root_paths(self, root_paths_json: str) -> tuple[Path, ...]:
+        raw_paths = orjson.loads(root_paths_json)
+        return tuple(Path(path) for path in raw_paths)
+
+    def _ensure_entity(
+        self,
+        connection: sqlite3.Connection,
+        *,
+        path: Path,
+        size: int,
+        mtime_ns: int,
+        extension: str,
+    ) -> int:
+        connection.execute(
+            """
+            INSERT INTO file_entities(path, size, mtime_ns, extension)
+            VALUES (?, ?, ?, ?)
+            ON CONFLICT(path, size, mtime_ns, extension) DO NOTHING
+            """,
+            (str(path), size, mtime_ns, extension),
+        )
+        return int(
+            connection.execute(
+                """
+                SELECT id
+                FROM file_entities
+                WHERE path = ? AND size = ? AND mtime_ns = ? AND extension = ?
+                """,
+                (str(path), size, mtime_ns, extension),
+            ).fetchone()["id"]
+        )
+
+    def _backfill_file_entities(self, connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            INSERT OR IGNORE INTO file_entities(path, size, mtime_ns, extension)
+            SELECT DISTINCT path, size, mtime_ns, extension
+            FROM files
+            """
+        )
+        connection.execute(
+            """
+            UPDATE files
+            SET entity_id = (
+                SELECT e.id
+                FROM file_entities AS e
+                WHERE e.path = files.path
+                  AND e.size = files.size
+                  AND e.mtime_ns = files.mtime_ns
+                  AND e.extension = files.extension
+            )
+            WHERE entity_id IS NULL
+            """
+        )
+
+    def _migrate_stage1_results_to_entities(self, connection: sqlite3.Connection) -> None:
+        columns = self._get_table_columns(connection, "stage1_results")
+        if not columns:
+            self._create_stage1_results_table(connection)
+            return
+
+        if "file_id" not in columns:
+            return
+
+        connection.execute("ALTER TABLE stage1_results RENAME TO stage1_results_legacy")
+        self._create_stage1_results_table(connection)
+        self._migrate_legacy_stage1_rows(connection)
+        connection.execute("DROP TABLE stage1_results_legacy")
+
+    def _migrate_stage2_results_to_entities(self, connection: sqlite3.Connection) -> None:
+        columns = self._get_table_columns(connection, "stage2_results")
+        if not columns:
+            self._create_stage2_results_table(connection)
+            return
+
+        if "file_id" not in columns:
+            return
+
+        connection.execute("ALTER TABLE stage2_results RENAME TO stage2_results_legacy")
+        self._create_stage2_results_table(connection)
+        self._migrate_legacy_stage2_rows(connection)
+        connection.execute("DROP TABLE stage2_results_legacy")
+
+    def _create_stage1_results_table(self, connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS stage1_results (
+                entity_id INTEGER PRIMARY KEY,
+                exit_code INTEGER,
+                stdout_text TEXT,
+                stderr_text TEXT,
+                elapsed_ms INTEGER,
+                FOREIGN KEY (entity_id) REFERENCES file_entities(id)
+            )
+            """
+        )
+
+    def _create_stage2_results_table(self, connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS stage2_results (
+                entity_id INTEGER PRIMARY KEY,
+                mode TEXT,
+                sample_points_json TEXT,
+                exit_code INTEGER,
+                stderr_text TEXT,
+                reason_code TEXT,
+                elapsed_ms INTEGER,
+                FOREIGN KEY (entity_id) REFERENCES file_entities(id)
+            )
+            """
+        )
+
+    def _migrate_legacy_stage1_rows(self, connection: sqlite3.Connection) -> None:
+        rows = connection.execute(
+            """
+            SELECT
+                f.entity_id,
+                f.run_id,
+                f.id AS file_id,
+                r.exit_code,
+                r.stdout_text,
+                r.stderr_text,
+                r.elapsed_ms
+            FROM stage1_results_legacy AS r
+            JOIN files AS f ON f.id = r.file_id
+            WHERE f.entity_id IS NOT NULL
+            ORDER BY f.run_id DESC, f.id DESC
+            """
+        ).fetchall()
+        seen: set[int] = set()
+        for row in rows:
+            entity_id = int(row["entity_id"])
+            if entity_id in seen:
+                continue
+            seen.add(entity_id)
+            connection.execute(
+                """
+                INSERT INTO stage1_results(entity_id, exit_code, stdout_text, stderr_text, elapsed_ms)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                (
+                    entity_id,
+                    row["exit_code"],
+                    row["stdout_text"],
+                    row["stderr_text"],
+                    row["elapsed_ms"],
+                ),
+            )
+
+    def _migrate_legacy_stage2_rows(self, connection: sqlite3.Connection) -> None:
+        rows = connection.execute(
+            """
+            SELECT
+                f.entity_id,
+                f.run_id,
+                f.id AS file_id,
+                r.mode,
+                r.sample_points_json,
+                r.exit_code,
+                r.stderr_text,
+                r.reason_code,
+                r.elapsed_ms
+            FROM stage2_results_legacy AS r
+            JOIN files AS f ON f.id = r.file_id
+            WHERE f.entity_id IS NOT NULL
+            ORDER BY f.run_id DESC, f.id DESC
+            """
+        ).fetchall()
+        seen: set[int] = set()
+        for row in rows:
+            entity_id = int(row["entity_id"])
+            if entity_id in seen:
+                continue
+            seen.add(entity_id)
+            connection.execute(
+                """
+                INSERT INTO stage2_results(
+                    entity_id,
+                    mode,
+                    sample_points_json,
+                    exit_code,
+                    stderr_text,
+                    reason_code,
+                    elapsed_ms
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    entity_id,
+                    row["mode"],
+                    row["sample_points_json"],
+                    row["exit_code"],
+                    row["stderr_text"],
+                    row["reason_code"],
+                    row["elapsed_ms"],
+                ),
+            )
+
+    def _get_entity_id_for_file(
+        self,
+        connection: sqlite3.Connection,
+        file_id: int,
+    ) -> int:
+        row = connection.execute(
+            """
+            SELECT entity_id
+            FROM files
+            WHERE id = ?
+            """,
+            (file_id,),
+        ).fetchone()
+        if row is None or row["entity_id"] is None:
+            raise ValueError(f"File row does not have an entity_id: {file_id}")
+        return int(row["entity_id"])
+
+    def _delete_orphan_entities(
+        self,
+        connection: sqlite3.Connection,
+    ) -> tuple[int, int, int]:
+        rows = connection.execute(
+            """
+            SELECT id
+            FROM file_entities
+            WHERE id NOT IN (
+                SELECT DISTINCT entity_id
+                FROM files
+                WHERE entity_id IS NOT NULL
+            )
+            """
+        ).fetchall()
+        orphan_entity_ids = tuple(int(row["id"]) for row in rows)
+        if not orphan_entity_ids:
+            return 0, 0, 0
+
+        placeholders = ",".join("?" for _ in orphan_entity_ids)
+        deleted_stage1_count = int(
+            connection.execute(
+                f"""
+                SELECT COUNT(*)
+                FROM stage1_results
+                WHERE entity_id IN ({placeholders})
+                """,
+                orphan_entity_ids,
+            ).fetchone()[0]
+        )
+        deleted_stage2_count = int(
+            connection.execute(
+                f"""
+                SELECT COUNT(*)
+                FROM stage2_results
+                WHERE entity_id IN ({placeholders})
+                """,
+                orphan_entity_ids,
+            ).fetchone()[0]
+        )
+
+        connection.execute(
+            f"""
+            DELETE FROM stage1_results
+            WHERE entity_id IN ({placeholders})
+            """,
+            orphan_entity_ids,
+        )
+        connection.execute(
+            f"""
+            DELETE FROM stage2_results
+            WHERE entity_id IN ({placeholders})
+            """,
+            orphan_entity_ids,
+        )
+        connection.execute(
+            f"""
+            DELETE FROM file_entities
+            WHERE id IN ({placeholders})
+            """,
+            orphan_entity_ids,
+        )
+        return len(orphan_entity_ids), deleted_stage1_count, deleted_stage2_count
+
     def _find_current_file(
         self,
         connection: sqlite3.Connection,
@@ -677,6 +971,7 @@ class ScanStore:
                 size,
                 mtime_ns,
                 extension,
+                entity_id,
                 stage1_status,
                 stage2_status,
                 final_status,
@@ -703,19 +998,24 @@ class ScanStore:
         self,
         connection: sqlite3.Connection,
         run_id: int,
-        candidate: FileCandidate,
         *,
+        entity_id: int,
         allow_stage2_match: bool,
     ) -> sqlite3.Row | None:
         query = """
-            SELECT id AS file_id, run_id, stage1_status, stage2_status, final_status, reason_code
+            SELECT
+                id AS file_id,
+                entity_id,
+                run_id,
+                stage1_status,
+                stage2_status,
+                final_status,
+                reason_code
             FROM files
-            WHERE path = ? AND size = ? AND mtime_ns = ? AND run_id <> ?
+            WHERE entity_id = ? AND run_id <> ?
         """
         params: tuple[object, ...] = (
-            str(candidate.path),
+            entity_id,
-            candidate.size,
-            candidate.mtime_ns,
             run_id,
         )
         if not allow_stage2_match:
@@ -726,74 +1026,85 @@ class ScanStore:
         """
         return connection.execute(query, params).fetchone()
 
-    def _clear_file_results(
+    def _roots_are_covered(
         self,
-        connection: sqlite3.Connection,
+        covered_roots: tuple[Path, ...],
-        file_id: int,
+        keep_roots: tuple[Path, ...],
-    ) -> None:
+    ) -> bool:
-        connection.execute(
+        return all(
-            """
+            any(
-            DELETE FROM stage1_results
+                self._is_same_or_child_path(candidate_root, keep_root)
-            WHERE file_id = ?
+                for keep_root in keep_roots
-            """,
+            )
-            (file_id,),
+            for candidate_root in covered_roots
-        )
-        connection.execute(
-            """
-            DELETE FROM stage2_results
-            WHERE file_id = ?
-            """,
-            (file_id,),
         )
 
-    def _copy_previous_results(
+    def _run_files_are_covered(
         self,
         connection: sqlite3.Connection,
         *,
-        previous_file_id: int,
+        keep_run_id: int,
-        current_file_id: int,
+        covered_run_id: int,
-        copy_stage2_results: bool,
+    ) -> bool:
-    ) -> None:
+        row = connection.execute(
-        connection.execute(
             """
-            INSERT INTO stage1_results(file_id, exit_code, stdout_text, stderr_text, elapsed_ms)
+            SELECT
-            SELECT ?, exit_code, stdout_text, stderr_text, elapsed_ms
+                COUNT(*) AS total_files,
-            FROM stage1_results
+                COALESCE(
-            WHERE file_id = ?
+                    SUM(
-            ON CONFLICT(file_id) DO UPDATE SET
+                        CASE
-                exit_code = excluded.exit_code,
+                            WHEN EXISTS (
-                stdout_text = excluded.stdout_text,
+                                SELECT 1
-                stderr_text = excluded.stderr_text,
+                                FROM files AS current_files
-                elapsed_ms = excluded.elapsed_ms
+                                WHERE current_files.run_id = ?
+                                  AND current_files.path = covered_files.path
+                                  AND current_files.entity_id = covered_files.entity_id
+                            )
+                            THEN 1
+                            ELSE 0
+                        END
+                    ),
+                    0
+                ) AS matched_files
+            FROM files AS covered_files
+            WHERE covered_files.run_id = ?
             """,
-            (current_file_id, previous_file_id),
+            (keep_run_id, covered_run_id),
-        )
+        ).fetchone()
-        if not copy_stage2_results:
+        if row is None:
-            return
+            return False
-        connection.execute(
+        return int(row["total_files"]) == int(row["matched_files"])
+
+    def _is_same_or_child_path(self, candidate: Path, root: Path) -> bool:
+        candidate_text = self._normalize_compare_path(candidate)
+        root_text = self._normalize_compare_path(root)
+        try:
+            return os.path.commonpath([candidate_text, root_text]) == root_text
+        except ValueError:
+            return False
+
+    def _normalize_compare_path(self, path: Path) -> str:
+        return os.path.normcase(str(path.resolve(strict=False)))
+
+    def _get_table_columns(
+        self,
+        connection: sqlite3.Connection,
+        table_name: str,
+    ) -> set[str]:
+        table_row = connection.execute(
             """
-            INSERT INTO stage2_results(
+            SELECT name
-                file_id,
+            FROM sqlite_master
-                mode,
+            WHERE type = 'table' AND name = ?
-                sample_points_json,
-                exit_code,
-                stderr_text,
-                reason_code,
-                elapsed_ms
-            )
-            SELECT ?, mode, sample_points_json, exit_code, stderr_text, reason_code, elapsed_ms
-            FROM stage2_results
-            WHERE file_id = ?
-            ON CONFLICT(file_id) DO UPDATE SET
-                mode = excluded.mode,
-                sample_points_json = excluded.sample_points_json,
-                exit_code = excluded.exit_code,
-                stderr_text = excluded.stderr_text,
-                reason_code = excluded.reason_code,
-                elapsed_ms = excluded.elapsed_ms
             """,
-            (current_file_id, previous_file_id),
+            (table_name,),
-        )
+        ).fetchone()
+        if table_row is None:
+            return set()
+        return {
+            row["name"]
+            for row in connection.execute(f"PRAGMA table_info({table_name})").fetchall()
+        }
 
     def _ensure_column(
         self,
@@ -802,16 +1113,31 @@ class ScanStore:
         column_name: str,
         column_definition: str,
     ) -> None:
-        columns = {
+        columns = self._get_table_columns(connection, table_name)
-            row["name"]
-            for row in connection.execute(f"PRAGMA table_info({table_name})").fetchall()
-        }
         if column_name in columns:
             return
         connection.execute(
             f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_definition}"
         )
 
+    def _ensure_index(
+        self,
+        connection: sqlite3.Connection,
+        index_name: str,
+        create_sql: str,
+    ) -> None:
+        row = connection.execute(
+            """
+            SELECT name
+            FROM sqlite_master
+            WHERE type = 'index' AND name = ?
+            """,
+            (index_name,),
+        ).fetchone()
+        if row is not None:
+            return
+        connection.execute(create_sql)
+
 
 def _preview_text(value: str | None, *, max_length: int = 160) -> str:
     if not value:

tests/test_cli.py

@@ -2,6 +2,7 @@ import sqlite3
 import time
 from pathlib import Path
 
+import orjson
 from typer.testing import CliRunner
 
 from tgvideo_check.cli import app
@@ -642,6 +643,168 @@ def test_scan_resume_does_not_reuse_previous_stage2_results(tmp_path, monkeypatc
     assert probe_calls == 1
 
 
+def test_scan_compacts_covered_historical_run_when_parent_root_is_scanned(
+    tmp_path: Path, monkeypatch
+) -> None:
+    parent = tmp_path / "a"
+    nested = parent / "b"
+    sibling = parent / "c"
+    nested.mkdir(parents=True)
+    sibling.mkdir()
+    legacy_video = nested / "legacy.mp4"
+    fresh_video = sibling / "fresh.mp4"
+    legacy_video.write_bytes(b"legacy")
+    fresh_video.write_bytes(b"fresh")
+    db_path = tmp_path / "scan_state.db"
+
+    store = ScanStore(db_path)
+    store.initialize()
+    legacy_run = store.create_run(profile="tg-hdd", root_paths=(nested,))
+    legacy_candidate = FileCandidate(
+        path=legacy_video,
+        size=legacy_video.stat().st_size,
+        mtime_ns=legacy_video.stat().st_mtime_ns,
+        extension="mp4",
+        discovery_order=0,
+    )
+    [legacy_stored] = store.upsert_files(legacy_run, [legacy_candidate])
+    store.record_stage1_result(
+        legacy_stored.file_id,
+        exit_code=0,
+        stdout_text='{"format":{"format_name":"mp4","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
+        stderr_text="",
+        elapsed_ms=1,
+    )
+    store.record_stage1_status(
+        legacy_stored.file_id,
+        stage1_status="passed_stage1",
+        final_status="passed_stage1",
+        reason_code="passed_stage1",
+    )
+    store.update_run_status(legacy_run, status="completed")
+
+    probe_calls: list[Path] = []
+
+    def fake_ffprobe(path: Path) -> ProbeExecutionResult:
+        probe_calls.append(path)
+        return ProbeExecutionResult(
+            status="completed",
+            command=("ffprobe", str(path)),
+            exit_code=0,
+            stdout_text='{"format":{"format_name":"mp4","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
+            stderr_text="",
+            elapsed_ms=1,
+        )
+
+    monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_ffprobe)
+
+    result = runner.invoke(
+        app,
+        [
+            "scan",
+            str(parent),
+            "--db",
+            str(db_path),
+        ],
+    )
+
+    assert result.exit_code == 0
+
+    with sqlite3.connect(db_path) as connection:
+        run_count = connection.execute("SELECT COUNT(*) FROM scan_runs").fetchone()[0]
+        file_count = connection.execute("SELECT COUNT(*) FROM files").fetchone()[0]
+        stage1_count = connection.execute("SELECT COUNT(*) FROM stage1_results").fetchone()[0]
+        remaining_root_paths_json = connection.execute(
+            "SELECT root_paths_json FROM scan_runs"
+        ).fetchone()[0]
+        legacy_row = connection.execute(
+            """
+            SELECT stage1_status, final_status
+            FROM files
+            WHERE path = ?
+            """,
+            (str(legacy_video),),
+        ).fetchone()
+
+    assert run_count == 1
+    assert file_count == 2
+    assert stage1_count == 2
+    assert orjson.loads(remaining_root_paths_json) == [str(parent)]
+    assert legacy_row == ("passed_stage1", "passed_stage1")
+    assert probe_calls == [fresh_video]
+
+
+def test_scan_does_not_compact_historical_run_when_nested_file_changed(
+    tmp_path: Path, monkeypatch
+) -> None:
+    parent = tmp_path / "a"
+    nested = parent / "b"
+    nested.mkdir(parents=True)
+    legacy_video = nested / "legacy.mp4"
+    legacy_video.write_bytes(b"legacy")
+    db_path = tmp_path / "scan_state.db"
+
+    store = ScanStore(db_path)
+    store.initialize()
+    legacy_run = store.create_run(profile="tg-hdd", root_paths=(nested,))
+    legacy_candidate = FileCandidate(
+        path=legacy_video,
+        size=legacy_video.stat().st_size,
+        mtime_ns=legacy_video.stat().st_mtime_ns,
+        extension="mp4",
+        discovery_order=0,
+    )
+    [legacy_stored] = store.upsert_files(legacy_run, [legacy_candidate])
+    store.record_stage1_result(
+        legacy_stored.file_id,
+        exit_code=0,
+        stdout_text='{"format":{"format_name":"mp4","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
+        stderr_text="",
+        elapsed_ms=1,
+    )
+    store.record_stage1_status(
+        legacy_stored.file_id,
+        stage1_status="passed_stage1",
+        final_status="passed_stage1",
+        reason_code="passed_stage1",
+    )
+    store.update_run_status(legacy_run, status="completed")
+
+    time.sleep(0.01)
+    legacy_video.write_bytes(b"legacy-changed")
+
+    def fake_ffprobe(path: Path) -> ProbeExecutionResult:
+        return ProbeExecutionResult(
+            status="completed",
+            command=("ffprobe", str(path)),
+            exit_code=0,
+            stdout_text='{"format":{"format_name":"mp4","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
+            stderr_text="",
+            elapsed_ms=1,
+        )
+
+    monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_ffprobe)
+
+    result = runner.invoke(
+        app,
+        [
+            "scan",
+            str(parent),
+            "--db",
+            str(db_path),
+        ],
+    )
+
+    assert result.exit_code == 0
+
+    with sqlite3.connect(db_path) as connection:
+        run_count = connection.execute("SELECT COUNT(*) FROM scan_runs").fetchone()[0]
+        stage1_count = connection.execute("SELECT COUNT(*) FROM stage1_results").fetchone()[0]
+
+    assert run_count == 2
+    assert stage1_count == 2
+
+
 def test_compact_db_command_keeps_selected_run_and_removes_duplicate_runs(tmp_path: Path) -> None:
     root = tmp_path / "videos"
     root.mkdir()

tests/test_resume_logic.py

@@ -16,6 +16,55 @@ def make_candidate(path: Path, *, order: int, size: int = 10, mtime_ns: int = 10
     )
 
 
+def create_legacy_schema(db_path: Path) -> None:
+    with sqlite3.connect(db_path) as connection:
+        connection.executescript(
+            """
+            CREATE TABLE scan_runs (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                profile TEXT NOT NULL,
+                root_paths_json TEXT NOT NULL,
+                status TEXT NOT NULL DEFAULT 'running'
+            );
+
+            CREATE TABLE files (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                run_id INTEGER NOT NULL,
+                path TEXT NOT NULL,
+                size INTEGER NOT NULL,
+                mtime_ns INTEGER NOT NULL,
+                extension TEXT NOT NULL,
+                discovery_order INTEGER NOT NULL,
+                stage1_status TEXT NOT NULL,
+                stage2_status TEXT,
+                final_status TEXT,
+                resumed_from_run_id INTEGER,
+                reason_code TEXT,
+                UNIQUE (run_id, path)
+            );
+
+            CREATE TABLE stage1_results (
+                file_id INTEGER PRIMARY KEY,
+                exit_code INTEGER,
+                stdout_text TEXT,
+                stderr_text TEXT,
+                elapsed_ms INTEGER
+            );
+
+            CREATE TABLE stage2_results (
+                file_id INTEGER PRIMARY KEY,
+                mode TEXT,
+                sample_points_json TEXT,
+                exit_code INTEGER,
+                stderr_text TEXT,
+                reason_code TEXT,
+                elapsed_ms INTEGER
+            );
+            """
+        )
+
+
 def test_initialize_creates_expected_tables(tmp_path: Path) -> None:
     store = ScanStore(tmp_path / "scan_state.db")
 
@@ -29,7 +78,7 @@ def test_initialize_creates_expected_tables(tmp_path: Path) -> None:
             )
         }
 
-    assert {"scan_runs", "files", "stage1_results", "stage2_results"} <= table_names
+    assert {"scan_runs", "files", "file_entities", "stage1_results", "stage2_results"} <= table_names
 
 
 def test_upsert_files_persists_candidates_in_discovery_order(tmp_path: Path) -> None:
@@ -69,6 +118,13 @@ def test_upsert_files_reuses_previous_status_when_identity_matches(tmp_path: Pat
     assert resumed.final_status == "passed_stage1"
     assert resumed.resumed_from_run_id == first_run
 
+    with sqlite3.connect(store.db_path) as connection:
+        stage1_count = connection.execute(
+            "SELECT COUNT(*) FROM stage1_results"
+        ).fetchone()[0]
+
+    assert stage1_count == 0
+
 
 def test_upsert_files_can_skip_reusing_previous_stage2_results(tmp_path: Path) -> None:
     store = ScanStore(tmp_path / "scan_state.db")
@@ -170,12 +226,16 @@ def test_upsert_files_same_run_preserves_existing_status_without_duplication(
             "SELECT COUNT(*) FROM files WHERE run_id = ?",
             (run_id,),
         ).fetchone()[0]
+        entity_count = connection.execute(
+            "SELECT COUNT(DISTINCT entity_id) FROM files WHERE run_id = ?",
+            (run_id,),
+        ).fetchone()[0]
         stage1_count = connection.execute(
-            "SELECT COUNT(*) FROM stage1_results WHERE file_id = ?",
+            "SELECT COUNT(*) FROM stage1_results",
-            (stored.file_id,),
         ).fetchone()[0]
 
     assert file_count == 1
+    assert entity_count == 1
     assert stage1_count == 1
 
 
@@ -202,6 +262,11 @@ def test_upsert_files_same_run_changed_file_clears_stale_stage_results(
         final_status="passed_stage1",
         reason_code="passed_stage1",
     )
+    with sqlite3.connect(store.db_path) as connection:
+        original_entity_id = connection.execute(
+            "SELECT entity_id FROM files WHERE id = ?",
+            (stored.file_id,),
+        ).fetchone()[0]
 
     changed = FileCandidate(
         path=file_path,
@@ -218,14 +283,199 @@ def test_upsert_files_same_run_changed_file_clears_stale_stage_results(
     assert refreshed.reason_code is None
 
     with sqlite3.connect(store.db_path) as connection:
-        stage1_count = connection.execute(
+        current_entity_id = connection.execute(
-            "SELECT COUNT(*) FROM stage1_results WHERE file_id = ?",
+            "SELECT entity_id FROM files WHERE id = ?",
             (stored.file_id,),
         ).fetchone()[0]
-        stage2_count = connection.execute(
+        new_entity_stage1_count = connection.execute(
-            "SELECT COUNT(*) FROM stage2_results WHERE file_id = ?",
+            "SELECT COUNT(*) FROM stage1_results WHERE entity_id = ?",
-            (stored.file_id,),
+            (current_entity_id,),
+        ).fetchone()[0]
+        old_entity_stage1_count = connection.execute(
+            "SELECT COUNT(*) FROM stage1_results WHERE entity_id = ?",
+            (original_entity_id,),
+        ).fetchone()[0]
+        new_entity_stage2_count = connection.execute(
+            "SELECT COUNT(*) FROM stage2_results WHERE entity_id = ?",
+            (current_entity_id,),
         ).fetchone()[0]
 
-    assert stage1_count == 0
+    assert current_entity_id != original_entity_id
-    assert stage2_count == 0
+    assert new_entity_stage1_count == 0
+    assert old_entity_stage1_count == 1
+    assert new_entity_stage2_count == 0
+
+
+def test_initialize_migrates_legacy_schema_to_entity_storage_without_duplicate_stage_results(
+    tmp_path: Path,
+) -> None:
+    db_path = tmp_path / "scan_state.db"
+    create_legacy_schema(db_path)
+
+    with sqlite3.connect(db_path) as connection:
+        connection.execute(
+            """
+            INSERT INTO scan_runs(id, profile, root_paths_json, status)
+            VALUES (1, 'tg-hdd', '["root"]', 'failed')
+            """
+        )
+        connection.execute(
+            """
+            INSERT INTO scan_runs(id, profile, root_paths_json, status)
+            VALUES (2, 'tg-hdd', '["root"]', 'running')
+            """
+        )
+        connection.execute(
+            """
+            INSERT INTO files(
+                id, run_id, path, size, mtime_ns, extension, discovery_order,
+                stage1_status, stage2_status, final_status, resumed_from_run_id, reason_code
+            )
+            VALUES
+                (11, 1, ?, 12, 999, 'mp4', 0, 'passed_stage1', NULL, 'passed_stage1', NULL, 'passed_stage1'),
+                (12, 2, ?, 12, 999, 'mp4', 0, 'passed_stage1', NULL, 'passed_stage1', 1, 'passed_stage1')
+            """,
+            (str(tmp_path / "clip.mp4"), str(tmp_path / "clip.mp4")),
+        )
+        connection.execute(
+            """
+            INSERT INTO stage1_results(file_id, exit_code, stdout_text, stderr_text, elapsed_ms)
+            VALUES
+                (11, 0, '{"legacy":"first"}', '', 10),
+                (12, 0, '{"legacy":"latest"}', '', 12)
+            """
+        )
+
+    store = ScanStore(db_path)
+    store.initialize()
+
+    with sqlite3.connect(db_path) as connection:
+        table_names = {
+            row[0]
+            for row in connection.execute(
+                "SELECT name FROM sqlite_master WHERE type = 'table'"
+            )
+        }
+        entity_count = connection.execute(
+            "SELECT COUNT(*) FROM file_entities"
+        ).fetchone()[0]
+        file_rows = connection.execute(
+            "SELECT COUNT(*) FROM files"
+        ).fetchone()[0]
+        stage1_rows = connection.execute(
+            "SELECT COUNT(*) FROM stage1_results"
+        ).fetchone()[0]
+        migrated_stdout = connection.execute(
+            "SELECT stdout_text FROM stage1_results"
+        ).fetchone()[0]
+        entity_links = connection.execute(
+            "SELECT COUNT(DISTINCT entity_id) FROM files"
+        ).fetchone()[0]
+
+    assert "file_entities" in table_names
+    assert entity_count == 1
+    assert file_rows == 2
+    assert stage1_rows == 1
+    assert migrated_stdout == '{"legacy":"latest"}'
+    assert entity_links == 1
+
+
+def test_stage1_results_are_shared_across_runs_by_entity(tmp_path: Path) -> None:
+    store = ScanStore(tmp_path / "scan_state.db")
+    store.initialize()
+
+    first_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
+    candidate = make_candidate(tmp_path / "clip.mp4", order=0, size=12, mtime_ns=999)
+    [first] = store.upsert_files(first_run, [candidate])
+    store.record_stage1_result(
+        first.file_id,
+        exit_code=0,
+        stdout_text='{"shared":"yes"}',
+        stderr_text="",
+        elapsed_ms=10,
+    )
+    store.record_stage1_status(
+        first.file_id,
+        stage1_status="passed_stage1",
+        final_status="passed_stage1",
+        reason_code="passed_stage1",
+    )
+
+    second_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
+    [second] = store.upsert_files(second_run, [candidate])
+
+    assert second.stage1_status == "passed_stage1"
+    assert second.final_status == "passed_stage1"
+
+    with sqlite3.connect(store.db_path) as connection:
+        stage1_rows = connection.execute(
+            "SELECT COUNT(*) FROM stage1_results"
+        ).fetchone()[0]
+        distinct_entities = connection.execute(
+            "SELECT COUNT(DISTINCT entity_id) FROM files"
+        ).fetchone()[0]
+
+    assert stage1_rows == 1
+    assert distinct_entities == 1
+
+
+def test_compact_duplicate_runs_removes_orphan_entities_and_stage_results(
+    tmp_path: Path,
+) -> None:
+    store = ScanStore(tmp_path / "scan_state.db")
+    store.initialize()
+
+    first_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
+    common = make_candidate(tmp_path / "common.mp4", order=0, size=12, mtime_ns=999)
+    stale_only = make_candidate(tmp_path / "stale.mp4", order=1, size=13, mtime_ns=1000)
+    first_rows = store.upsert_files(first_run, [common, stale_only])
+    for item, stdout_text in zip(first_rows, ('{"file":"common"}', '{"file":"stale"}'), strict=True):
+        store.record_stage1_result(
+            item.file_id,
+            exit_code=0,
+            stdout_text=stdout_text,
+            stderr_text="",
+            elapsed_ms=10,
+        )
+        store.record_stage1_status(
+            item.file_id,
+            stage1_status="passed_stage1",
+            final_status="passed_stage1",
+            reason_code="passed_stage1",
+        )
+
+    second_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
+    [second_common] = store.upsert_files(second_run, [common])
+    store.record_stage1_status(
+        second_common.file_id,
+        stage1_status="passed_stage1",
+        final_status="passed_stage1",
+        reason_code="passed_stage1",
+    )
+
+    summary = store.compact_duplicate_runs(second_run, dry_run=False, vacuum=False)
+
+    with sqlite3.connect(store.db_path) as connection:
+        remaining_runs = connection.execute(
+            "SELECT COUNT(*) FROM scan_runs"
+        ).fetchone()[0]
+        remaining_files = connection.execute(
+            "SELECT COUNT(*) FROM files"
+        ).fetchone()[0]
+        remaining_entities = connection.execute(
+            "SELECT COUNT(*) FROM file_entities"
+        ).fetchone()[0]
+        remaining_stage1 = connection.execute(
+            "SELECT COUNT(*) FROM stage1_results"
+        ).fetchone()[0]
+        remaining_stdout = {
+            row[0]
+            for row in connection.execute("SELECT stdout_text FROM stage1_results")
+        }
+
+    assert summary.deleted_run_count == 1
+    assert remaining_runs == 1
+    assert remaining_files == 1
+    assert remaining_entities == 1
+    assert remaining_stage1 == 1
+    assert remaining_stdout == {'{"file":"common"}'}