diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py
index 64f7bf2f7..dd6beb4c5 100644
--- a/app/modules/filemanager/storages/__init__.py
+++ b/app/modules/filemanager/storages/__init__.py
@@ -1,6 +1,6 @@
 from abc import ABCMeta, abstractmethod
 from pathlib import Path, PurePosixPath
-from typing import Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
 from tqdm import tqdm
 
@@ -13,6 +13,8 @@ from app.schemas.file import StorageUsage as _SchemaStorageUsage
 from app.schemas.system import StorageConf as _SchemaStorageConf
 from app.schemas.workflow import FileItem as _SchemaFileItem
 
+_SnapshotInfo = Dict[str, Dict[str, Any]]
+
 
 def transfer_process(path: str) -> Callable[[int | float], None]:
     """
@@ -312,6 +314,150 @@ class StorageBase(metaclass=ABCMeta):
             raise StorageQueryError(f"读取快照路径失败: {fileitem.path} - {error}") from error
         logger.debug(f"Snapshot error for {fileitem.path}: {error}")
 
+    @staticmethod
+    def _snapshot_previous_state(
+        root_path: PurePosixPath,
+        previous_snapshot: Optional[_SnapshotInfo],
+    ) -> Tuple[
+        _SnapshotInfo,
+        Dict[PurePosixPath, set[PurePosixPath]],
+        Dict[PurePosixPath, str],
+    ]:
+        """
+        为增量快照建立旧文件和目录直接子项索引。
+
+        :param root_path: 本轮快照的根路径
+        :param previous_snapshot: 上一轮完整快照
+        :return: 当前文件信息、目录直接子项索引和路径反向索引
+        """
+        files_info = {
+            file_path: file_info
+            for file_path, file_info in (previous_snapshot or {}).items()
+            if PurePosixPath(file_path).is_relative_to(root_path)
+        }
+        previous_children: Dict[PurePosixPath, set[PurePosixPath]] = {}
+        previous_files: Dict[PurePosixPath, str] = {}
+        for file_path in files_info:
+            file_posix_path = PurePosixPath(file_path)
+            previous_files[file_posix_path] = file_path
+            try:
+                relative_parts = file_posix_path.relative_to(root_path).parts
+            except ValueError:
+                # files_info 已经按 root_path 过滤；保留防御性判断，避免
+                # 非标准路径对象让增量索引中断整轮监控。
+                continue
+            for depth in range(len(relative_parts)):
+                directory_path = root_path.joinpath(*relative_parts[:depth])
+                child_path = directory_path / relative_parts[depth]
+                previous_children.setdefault(directory_path, set()).add(child_path)
+        return files_info, previous_children, previous_files
+
+    @staticmethod
+    def _snapshot_remove_previous_subtree(
+        path: PurePosixPath,
+        files_info: _SnapshotInfo,
+        previous_children: Dict[PurePosixPath, set[PurePosixPath]],
+        previous_files: Dict[PurePosixPath, str],
+    ) -> None:
+        """沿旧快照索引移除一个已经消失的直接子树。"""
+        pending = [path]
+        while pending:
+            current_path = pending.pop()
+            old_file_path = previous_files.pop(current_path, None)
+            if old_file_path is not None:
+                files_info.pop(old_file_path, None)
+            pending.extend(previous_children.pop(current_path, ()))
+
+    def _snapshot_remove_deleted_children(
+        self,
+        fileitem: _SchemaFileItem,
+        sub_files: List[_SchemaFileItem],
+        files_info: _SnapshotInfo,
+        previous_children: Dict[PurePosixPath, set[PurePosixPath]],
+        previous_files: Dict[PurePosixPath, str],
+    ) -> None:
+        """
+        清理已确认遍历目录中不再存在的直接子项。
+
+        :param fileitem: 当前已列举的目录
+        :param sub_files: 当前目录返回的直接子项
+        :param files_info: 当前正在构建的快照
+        :param previous_children: 旧快照目录直接子项索引
+        :param previous_files: 旧快照路径反向索引
+        """
+        directory_path = PurePosixPath(fileitem.path)
+        child_paths = {PurePosixPath(sub_file.path) for sub_file in sub_files}
+        for direct_child_path in previous_children.pop(directory_path, ()):
+            if direct_child_path not in child_paths:
+                StorageBase._snapshot_remove_previous_subtree(
+                    direct_child_path, files_info, previous_children, previous_files
+                )
+
+    def _snapshot_file(
+        self,
+        fileitem: _SchemaFileItem,
+        files_info: _SnapshotInfo,
+        previous_children: Dict[PurePosixPath, set[PurePosixPath]],
+        previous_files: Dict[PurePosixPath, str],
+        last_snapshot_time: float,
+        max_depth: int,
+        current_depth: int = 0,
+    ) -> None:
+        """
+        递归读取目录并更新快照文件信息。
+
+        :param fileitem: 当前文件或目录
+        :param files_info: 当前正在构建的快照
+        :param previous_children: 旧快照目录直接子项索引
+        :param previous_files: 旧快照路径反向索引
+        :param last_snapshot_time: 上次快照时间
+        :param max_depth: 最大递归深度
+        :param current_depth: 当前递归深度
+        """
+        try:
+            if fileitem.type == "dir":
+                if current_depth >= max_depth:
+                    return
+
+                # 根目录每轮至少列举一次；子目录按修改时间跳过未变化分支。
+                if (current_depth > 0 and
+                        self.snapshot_check_folder_modtime and
+                        last_snapshot_time and
+                        fileitem.modify_time and
+                        fileitem.modify_time <= last_snapshot_time):
+                    return
+
+                # 只有目录列表成功返回后才清理旧基线，查询异常时保留待下轮重试。
+                sub_files = self._snapshot_list(fileitem)
+                if sub_files is None:
+                    return
+                sub_files = list(sub_files)
+                StorageBase._snapshot_remove_deleted_children(
+                    self,
+                    fileitem, sub_files, files_info, previous_children, previous_files
+                )
+                for sub_file in sub_files:
+                    StorageBase._snapshot_file(
+                        self,
+                        sub_file,
+                        files_info,
+                        previous_children,
+                        previous_files,
+                        last_snapshot_time,
+                        max_depth,
+                        current_depth + 1,
+                    )
+            else:
+                # 始终记录文件完整信息，由 compare_snapshots 负责检测变化。
+                files_info[fileitem.path] = {
+                    'size': fileitem.size or 0,
+                    'modify_time': getattr(fileitem, 'modify_time', 0),
+                    'fileid': getattr(fileitem, 'fileid', None),
+                    'type': fileitem.type
+                }
+        except Exception as error:
+            self._snapshot_read_error(fileitem, error)
+
     def snapshot(self, path: Path, last_snapshot_time: float = None, max_depth: int = 5,
                  previous_snapshot: Optional[Dict[str, Dict]] = None) -> Optional[Dict[str, Dict]]:
         """
@@ -324,74 +470,23 @@ class StorageBase(metaclass=ABCMeta):
         """
         strict_query = self.snapshot_strict_query
         root_path = PurePosixPath(path.as_posix())
-        files_info = {
-            file_path: file_info
-            for file_path, file_info in (previous_snapshot or {}).items()
-            if PurePosixPath(file_path).is_relative_to(root_path)
-        }
-
-        def __remove_deleted_children(_fileitm: _SchemaFileItem,
-                                      sub_files: List[_SchemaFileItem]) -> None:
-            """
-            清理已确认遍历目录中不再存在的直接子项。
-            未变化的子目录仍保留旧基线，避免增量遍历将其误删。
-            """
-            directory_path = PurePosixPath(_fileitm.path)
-            child_paths = {PurePosixPath(sub_file.path) for sub_file in sub_files}
-            for old_file_path in list(files_info):
-                try:
-                    relative_path = PurePosixPath(old_file_path).relative_to(directory_path)
-                except ValueError:
-                    continue
-                if not relative_path.parts:
-                    continue
-                direct_child_path = directory_path / relative_path.parts[0]
-                if direct_child_path not in child_paths:
-                    files_info.pop(old_file_path, None)
-
-        def __snapshot_file(_fileitm: _SchemaFileItem, current_depth: int = 0):
-            """
-            递归获取文件信息
-            """
-            try:
-                if _fileitm.type == "dir":
-                    # 检查递归深度限制
-                    if current_depth >= max_depth:
-                        return
-
-                    # 根目录每轮至少列举一次，用于清理已移走的直接子项；子目录仍按修改时间增量遍历
-                    if (current_depth > 0 and
-                            self.snapshot_check_folder_modtime and
-                            last_snapshot_time and
-                            _fileitm.modify_time and
-                            _fileitm.modify_time <= last_snapshot_time):
-                        return
-
-                    # 只有目录列表成功返回后才清理旧基线，查询异常时继续保留待下轮重试
-                    sub_files = self._snapshot_list(_fileitm)
-                    if sub_files is None:
-                        return
-                    sub_files = list(sub_files)
-                    __remove_deleted_children(_fileitm, sub_files)
-                    for sub_file in sub_files:
-                        __snapshot_file(sub_file, current_depth + 1)
-                else:
-                    # 记录文件的完整信息用于比对（始终包含所有文件，由 compare_snapshots 负责检测变化）
-                    files_info[_fileitm.path] = {
-                        'size': _fileitm.size or 0,
-                        'modify_time': getattr(_fileitm, 'modify_time', 0),
-                        'fileid': getattr(_fileitm, 'fileid', None),
-                        'type': _fileitm.type
-                    }
-
-            except Exception as e:
-                self._snapshot_read_error(_fileitm, e)
+        files_info, previous_children, previous_files = StorageBase._snapshot_previous_state(
+            root_path, previous_snapshot
+        )
 
         try:
             fileitem = self.get_item_strict(path) if strict_query else self.get_item(path)
             if not fileitem:
                 return {}
-            __snapshot_file(fileitem)
+            StorageBase._snapshot_file(
+                self,
+                fileitem,
+                files_info,
+                previous_children,
+                previous_files,
+                last_snapshot_time,
+                max_depth,
+            )
         except StorageQueryError as err:
             logger.warning(f"存储快照失败，保留上次基线: {self.schema}:{path} - {err}")
             return None
diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md
index 61b7474fb..91db547be 100644
--- a/docs/refactor/optimization-checklist.md
+++ b/docs/refactor/optimization-checklist.md
@@ -106,7 +106,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
-| 全量 mypy 历史债务 | 9,177 / 508 文件 | 系统版本查询补齐类型后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
+| 全量 mypy 历史债务 | 9,176 / 508 文件 | 系统版本查询补齐类型后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
 | Ruff 历史诊断 | 505 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json
index 34f54a886..bece4a999 100644
--- a/tests/fixtures/architecture/mypy-baseline.json
+++ b/tests/fixtures/architecture/mypy-baseline.json
@@ -1516,7 +1516,7 @@
     "attr-defined": 4,
     "index": 1,
     "no-untyped-call": 2,
-    "no-untyped-def": 8,
+    "no-untyped-def": 7,
     "return-value": 1,
     "type-arg": 6,
     "var-annotated": 1
diff --git a/tests/test_snapshot_reconciliation_performance.py b/tests/test_snapshot_reconciliation_performance.py
new file mode 100644
index 000000000..9362c99ec
--- /dev/null
+++ b/tests/test_snapshot_reconciliation_performance.py
@@ -0,0 +1,90 @@
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from app import schemas
+from app.modules.filemanager.storages import StorageBase
+
+
+def _directory(path: str, *, modify_time: int = 200):
+    return schemas.FileItem(
+        storage="alist",
+        type="dir",
+        path=f"{path.rstrip('/')}/",
+        name=Path(path).name,
+        modify_time=modify_time,
+    )
+
+
+def _file(path: str, *, size: int = 1):
+    return schemas.FileItem(
+        storage="alist",
+        type="file",
+        path=path,
+        name=Path(path).name,
+        size=size,
+        modify_time=200,
+    )
+
+
+def test_snapshot_removes_a_deleted_directory_without_scanning_other_subtrees():
+    """
+    目录删除只应移除该目录的文件，而不影响其它目录的旧基线。
+    """
+    storage = MagicMock()
+    storage.snapshot_check_folder_modtime = True
+    storage.snapshot_strict_query = False
+    storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item)
+    root = _directory("/mon")
+    keep = _directory("/mon/keep", modify_time=50)
+    storage.get_item.return_value = root
+    storage.list.return_value = [keep]
+
+    previous_snapshot = {
+        "/mon/gone/episode-1.mkv": {"size": 1, "modify_time": 100},
+        "/mon/gone/episode-2.mkv": {"size": 2, "modify_time": 100},
+        "/mon/keep/episode-1.mkv": {"size": 3, "modify_time": 50},
+    }
+
+    snapshot = StorageBase.snapshot(
+        storage,
+        Path("/mon"),
+        last_snapshot_time=100,
+        previous_snapshot=previous_snapshot,
+    )
+
+    assert snapshot == {
+        "/mon/keep/episode-1.mkv": {"size": 3, "modify_time": 50}
+    }
+
+
+def test_snapshot_reconciles_nested_deleted_files_with_bounded_directory_work():
+    """
+    每个被列举目录只需处理自己的直接子项；嵌套文件删除仍应正确收敛。
+    """
+    storage = MagicMock()
+    storage.snapshot_check_folder_modtime = False
+    storage.snapshot_strict_query = False
+    storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item)
+    root = _directory("/mon")
+    season = _directory("/mon/show/Season 1")
+    storage.get_item.return_value = root
+    storage.list.side_effect = [
+        [_directory("/mon/show")],
+        [season],
+        [_file("/mon/show/Season 1/episode-2.mkv", size=20)],
+    ]
+
+    previous_snapshot = {
+        "/mon/show/Season 1/episode-1.mkv": {"size": 10, "modify_time": 100},
+        "/mon/show/Season 1/episode-2.mkv": {"size": 20, "modify_time": 100},
+    }
+
+    snapshot = StorageBase.snapshot(
+        storage,
+        Path("/mon"),
+        previous_snapshot=previous_snapshot,
+    )
+
+    assert snapshot == {
+        "/mon/show/Season 1/episode-2.mkv": {"size": 20, "modify_time": 200, "fileid": None, "type": "file"}
+    }