Merge remote-tracking branch 'origin/v3' into v3

# Conflicts:
#	tests/fixtures/architecture/dependency-baseline.json

.trivyignore.yaml

@@ -4,58 +4,68 @@ vulnerabilities:
       - Python
     purls:
       - pkg:pypi/msgpack@1.1.2
+    expired_at: 2026-11-20
     statement: The finding belongs to the base image's system pip and is not imported by MoviePilot.
   - id: CVE-2025-47273
     paths:
       - Python
     purls:
       - pkg:pypi/setuptools@70.3.0
+    expired_at: 2026-11-20
     statement: The finding belongs to the base image's system pip and is not used for dependency installation.
   - id: CVE-2026-33818
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-39821
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-46600
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-56853
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-56858
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-56859
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-56860
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.
   - id: CVE-2026-56862
     paths:
       - usr/bin/rclone
     purls:
       - pkg:golang/stdlib@v1.26.5
+    expired_at: 2026-11-20
     statement: The official rclone binary has no patched release for this embedded Go runtime yet.

app/adapters/system/backup/database.py

@@ -46,6 +46,73 @@ class ProcessRunner(Protocol):
         """执行命令并返回结果。"""
 
 
+def verify_database_backup(
+    artifact: Path,
+    *,
+    db_type: str,
+    runner: ProcessRunner = subprocess.run,
+    tool_resolver: Callable[[str], str | None] = shutil.which,
+    pg_restore: str = "pg_restore",
+) -> DatabaseBackupCheck:
+    """在不访问活动数据库的前提下校验一个受管备份文件。"""
+    if db_type == "sqlite":
+        method = "PRAGMA integrity_check"
+        try:
+            # 正式备份不会再变化，immutable 可避免只读校验创建 WAL 旁路文件。
+            uri = f"{artifact.resolve().as_uri()}?mode=ro&immutable=1"
+            with closing(sqlite3.connect(uri, uri=True)) as connection:
+                rows = connection.execute("PRAGMA integrity_check").fetchall()
+        except sqlite3.Error as error:
+            return DatabaseBackupCheck(False, method, str(error))
+        valid = bool(rows) and all(row[0] == "ok" for row in rows)
+        detail = None if valid else "; ".join(str(row[0]) for row in rows)
+        return DatabaseBackupCheck(valid, method, detail)
+
+    if db_type == "postgresql":
+        method = "pg_restore --list"
+        executable = _require_tool(pg_restore, tool_resolver)
+        result = runner(
+            [executable, "--list", str(artifact)],
+            env=_postgres_environment(),
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        valid = result.returncode == 0 and bool(result.stdout.strip())
+        detail = None if valid else f"pg_restore 退出码 {result.returncode}"
+        return DatabaseBackupCheck(valid, method, detail)
+
+    raise ValueError(f"不支持的数据库备份类型：{db_type}")
+
+
+def _require_tool(
+    executable: str,
+    tool_resolver: Callable[[str], str | None],
+) -> str:
+    resolved = tool_resolver(executable)
+    if resolved is None:
+        raise RuntimeError(
+            f"未找到 {executable}，请安装与服务端同主版本或更高的 "
+            "PostgreSQL client 并加入 PATH"
+        )
+    return resolved
+
+
+def _postgres_environment(
+    *,
+    password: str | None = None,
+    sslmode: str | None = None,
+) -> dict[str, str]:
+    environment = dict(os.environ)
+    environment.pop("PGPASSWORD", None)
+    environment.pop("PGSSLMODE", None)
+    if password:
+        environment["PGPASSWORD"] = password
+    if sslmode:
+        environment["PGSSLMODE"] = sslmode
+    return environment
+
+
 class SQLiteBackupBackend:
     """使用 SQLite 在线备份 API 管理活动文件数据库。"""
 
@@ -71,17 +138,7 @@ class SQLiteBackupBackend:
 
     def verify(self, artifact: Path) -> DatabaseBackupCheck:
         """通过 SQLite integrity_check 校验备份内容。"""
-        method = "PRAGMA integrity_check"
-        try:
-            # 已发布前的临时快照不会再变化；immutable 避免 WAL 模式为只读校验创建旁路文件。
-            uri = f"{artifact.resolve().as_uri()}?mode=ro&immutable=1"
-            with closing(sqlite3.connect(uri, uri=True)) as connection:
-                rows = connection.execute("PRAGMA integrity_check").fetchall()
-        except sqlite3.Error as error:
-            return DatabaseBackupCheck(False, method, str(error))
-        valid = bool(rows) and all(row[0] == "ok" for row in rows)
-        detail = None if valid else "; ".join(str(row[0]) for row in rows)
-        return DatabaseBackupCheck(valid, method, detail)
+        return verify_database_backup(artifact, db_type=self.db_type)
 
     def restore(self, artifact: Path) -> None:
         """在 CLI 离线进程中原子替换活动 SQLite 文件。"""
@@ -137,14 +194,13 @@ class PostgreSQLBackupBackend:
 
     def verify(self, artifact: Path) -> DatabaseBackupCheck:
         """通过 pg_restore 目录读取校验 custom-format 归档。"""
-        method = "pg_restore --list"
-        result = self._run(
-            [self._require_tool(self._pg_restore), "--list", str(artifact)],
-            include_password=False,
+        return verify_database_backup(
+            artifact,
+            db_type=self.db_type,
+            runner=self._runner,
+            tool_resolver=self._tool_resolver,
+            pg_restore=self._pg_restore,
         )
-        valid = result.returncode == 0 and bool(result.stdout.strip())
-        detail = None if valid else f"pg_restore 退出码 {result.returncode}"
-        return DatabaseBackupCheck(valid, method, detail)
 
     def restore(self, artifact: Path) -> None:
         """在 CLI 离线进程中覆盖当前 PostgreSQL 数据库内容。"""
@@ -189,21 +245,16 @@ class PostgreSQLBackupBackend:
         )
 
     def _require_tool(self, executable: str) -> str:
-        resolved = self._tool_resolver(executable)
-        if resolved is None:
-            raise RuntimeError(
-                f"未找到 {executable}，请安装与服务端同主版本或更高的 "
-                "PostgreSQL client 并加入 PATH"
-            )
-        return resolved
+        return _require_tool(executable, self._tool_resolver)
 
     def _environment(self, *, include_password: bool) -> dict[str, str]:
-        environment = dict(os.environ)
-        environment.pop("PGPASSWORD", None)
-        environment.pop("PGSSLMODE", None)
-        if include_password and self._engine.url.password:
-            environment["PGPASSWORD"] = str(self._engine.url.password)
+        password = (
+            str(self._engine.url.password)
+            if include_password and self._engine.url.password
+            else None
+        )
         sslmode = self._engine.url.query.get("sslmode")
-        if sslmode:
-            environment["PGSSLMODE"] = str(sslmode)
-        return environment
+        return _postgres_environment(
+            password=password,
+            sslmode=str(sslmode) if sslmode else None,
+        )

app/application/formatting.py

@@ -7,7 +7,7 @@ from typing import Dict, Iterable, List, Match, Optional, Tuple, Union
 
 import anitopy
 
-from app.runtime.config import settings
+from app.application.configuration import get_chain_runtime_config_snapshot
 from app.domain.metainfo import MetaInfoPath
 from app.domain.meta.metabase import MetaBase
 from app.runtime.log import logger
@@ -580,11 +580,12 @@ class EpisodeFormatRuleHelper:
     def _get_file_kind(item: FileItem) -> str:
         """按扩展名把样本归类为视频、字幕、音频或其他。"""
         extension = f".{(item.extension or '').lower().lstrip('.')}" if item.extension else ""
-        if extension in settings.RMT_MEDIAEXT:
+        config = get_chain_runtime_config_snapshot()
+        if extension in config.video_extensions:
             return "media"
-        if extension in settings.RMT_SUBEXT:
+        if extension in config.subtitle_extensions:
             return "subtitle"
-        if extension in settings.RMT_AUDIOEXT:
+        if extension in config.audio_extensions:
             return "audio"
         return "other"
 

app/doctor/checks.py

@@ -18,6 +18,8 @@ from urllib.request import Request, urlopen
 
 import psutil
 
+from app.adapters.system.backup.database import verify_database_backup
+from app.adapters.system.backup.files import BackupFiles
 from app.runtime.config import settings
 from app.runtime.topology import process_topology_issue
 from app.doctor.models import DoctorFinding, DoctorFindingStatus, DoctorReport, DoctorSeverity
@@ -824,6 +826,116 @@ def _check_database(runner: DoctorRunnerProtocol) -> None:
         _check_postgresql_database(runner)
     else:
         _check_sqlite_database(runner)
+    _check_database_backups(runner)
+
+
+def _check_database_backups(runner: DoctorRunnerProtocol) -> None:
+    """列举并离线校验与当前数据库类型匹配的受管备份。"""
+    db_type = "postgresql" if settings.DB_TYPE.lower() == "postgresql" else "sqlite"
+    try:
+        paths = BackupFiles(settings.DATABASE_BACKUP_PATH).list()
+    except OSError as error:
+        runner.add(
+            finding_id="database.backup_recovery",
+            severity=DoctorSeverity.Error,
+            status=DoctorFindingStatus.Failed,
+            title="数据库备份目录无法读取",
+            detail=str(error),
+            recommendation="检查数据库备份目录是否存在且当前用户具有读取权限。",
+            context={"db_type": db_type},
+        )
+        return
+
+    matching = [path for path in paths if BackupFiles.database_type(path.name) == db_type]
+    mismatched = [path.name for path in paths if path not in matching]
+    if not paths:
+        runner.add(
+            finding_id="database.backup_recovery",
+            severity=DoctorSeverity.Info,
+            status=DoctorFindingStatus.Skipped,
+            title="未找到受管数据库备份",
+            detail=f"当前数据库类型为 {db_type}，备份目录中没有正式备份文件。",
+            recommendation="可执行 `moviepilot database backup` 创建一次可校验备份。",
+            affects_report_status=False,
+            context={"db_type": db_type, "backups": []},
+        )
+        return
+
+    if not matching:
+        runner.add(
+            finding_id="database.backup_recovery",
+            severity=DoctorSeverity.Warn,
+            status=DoctorFindingStatus.Degraded,
+            title="没有匹配当前数据库类型的备份",
+            detail=f"当前数据库类型为 {db_type}，仅找到其他类型备份：{', '.join(mismatched)}。",
+            recommendation="确认数据库类型配置，或执行 `moviepilot database backup` 创建当前类型备份。",
+            affects_report_status=False,
+            context={"db_type": db_type, "backups": [], "mismatched": mismatched},
+        )
+        return
+
+    backups = []
+    for path in matching:
+        try:
+            size = path.stat().st_size
+            verification = verify_database_backup(path, db_type=db_type)
+            valid = verification.valid
+            method = verification.method
+            detail = verification.detail
+        except (OSError, RuntimeError, ValueError) as error:
+            size = None
+            valid = False
+            method = "unavailable"
+            detail = str(error)
+        backups.append(
+            {
+                "name": path.name,
+                "db_type": db_type,
+                "size": size,
+                "valid": valid,
+                "method": method,
+                "detail": detail,
+            }
+        )
+
+    valid_backups = [backup for backup in backups if backup["valid"]]
+    context = {
+        "db_type": db_type,
+        "backups": backups,
+        "mismatched": mismatched,
+    }
+    if valid_backups:
+        newest = valid_backups[0]
+        command = f"moviepilot database restore {newest['name']} --confirm"
+        runner.add(
+            finding_id="database.backup_recovery",
+            severity=DoctorSeverity.Info,
+            status=DoctorFindingStatus.Ok,
+            title="存在可还原的数据库备份",
+            detail=f"已校验 {len(backups)} 个 {db_type} 备份，其中 {len(valid_backups)} 个可用。",
+            recommendation=f"需要恢复时先停止 MoviePilot，再执行 `{command}`。",
+            context={**context, "restore_command": command},
+        )
+        return
+
+
+    failures = "; ".join(
+        f"{backup['name']}: {backup['detail'] or backup['method']}"
+        for backup in backups
+    )
+    runner.add(
+        finding_id="database.backup_recovery",
+        severity=DoctorSeverity.Error,
+        status=DoctorFindingStatus.Failed,
+        title="数据库备份均未通过校验",
+        detail=(
+            f"已检查 {len(backups)} 个 {db_type} 备份，没有可直接还原的文件。"
+            f"校验结果：{failures}"
+        ),
+        recommendation="根据校验详情处理备份文件或 PostgreSQL client，再重新运行 doctor。",
+        affects_report_status=False,
+        context=context,
+    )
 
 
 def _check_frontend_assets(runner: DoctorRunnerProtocol) -> None:

docs/refactor/backend-architecture-next-stage.md

@@ -989,19 +989,15 @@ Outbox adapter、DB 装饰器、Base 与 UoW，strict 清单扩大到 37 个源
 
 **实施记录（2026-08-21）**：
 
-- 新增 `scripts/architecture/async_blocking.py`，AST 扫描 `app/api`、`app/agent`、`app/application` 的
-  async 函数，覆盖直接 `open`/Path 读写遍历、`requests`、`time.sleep`、同步 `subprocess` 和目录遍历。
-- scanner 通过局部类型流识别 `aiofiles` 与 `anyio.AsyncPath`，不会把正确异步 I/O 记成债务；baseline
-  只允许调用减少/删除，新增或次数增长均在 CI architecture job 失败。
-- Web Agent 上传已从 `Path.open/write/unlink` 改为 `aiofiles` 写入和统一 `run_in_threadpool` 清理；当前仅保留
-  ActivityLog 为保证 `O_EXCL` 原子创建使用的一处 `os.open` 精确债务，不泛化豁免整个文件或目录。
+- `scripts/architecture/async_blocking.py` 扫描 canonical 主程序目录和顶层运行入口中的 async 函数，覆盖
+  同步 HTTP、Oper、Path、`shutil`、`subprocess`、`os`、`time.sleep` 与 `open`。
+- scanner 按 import 来源、局部别名、互斥分支和嵌套函数定义点解析符号；`AsyncRequestUtils`、
+  `anyio.Path`、延迟回调及受控 worker 内执行的同步函数不记为 async 直接阻塞。函数和 lambda 的默认值、
+  decorator 等定义时表达式仍在所在 async 执行体中检查。
+- baseline 只允许调用减少或删除，新增调用及次数增长均使 CI architecture job 失败；当前记录 10 条已确认
+  存量，包括 8 条文件元数据访问、1 条目录删除和 1 条同步 Oper 读取，由后续数据库与文件 adapter 叶迁移。
 - pytest 全局启用 `asyncio_debug`，专项测试验证实际 loop debug 状态；AST ratchet 与 46 个 Agent 流式回归
   通过。同步第三方 Module 仍由 dispatcher 的 `app.runtime.execution.run_in_threadpool` 兼容。
-- 2026-08-22 扫描范围扩大到 `app/chain`、`app/modules`、`app/startup` 与 `app/scheduler.py`；扩大后未发现
-  新存量，仍只保留 ActivityLog 的一处原子 `os.open` 精确债务，并由测试锁定扫描根目录。
-- 扫描进一步覆盖 `adapters/db/doctor/domain/foundation/monitor/runtime/schemas/workflow` 及 CLI、Command、
-  Factory、Main 顶层入口，明确排除插件源码和 SDK。AsyncPath 条件派生识别已修正；Release zip 解压读取和
-  ActivityLog `O_EXCL` 独占创建移入线程池，扩围后 async 阻塞 baseline 从 1 降为 0。
 
 ## 6. 推荐执行队列
 

scripts/architecture/async_blocking.py

@@ -6,7 +6,8 @@ import argparse
 import ast
 import json
 from collections import Counter
-from collections.abc import Iterator
+from collections.abc import Iterator, Sequence
+from dataclasses import dataclass
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -33,159 +34,545 @@ SCAN_ROOTS = (
     "app/main.py",
     "app/scheduler.py",
 )
-BLOCKING_EXACT = {
-    "open",
-    "time.sleep",
-    "subprocess.call",
-    "subprocess.check_call",
-    "subprocess.check_output",
-    "subprocess.Popen",
-    "subprocess.run",
-    "os.listdir",
-    "os.scandir",
-    "os.walk",
-    "requests.delete",
-    "requests.get",
-    "requests.head",
-    "requests.patch",
-    "requests.post",
-    "requests.put",
-    "requests.request",
+
+_SYNC_HTTP_METHODS = {
+    "delete_res",
+    "get",
+    "get_json",
+    "get_res",
+    "get_stream",
+    "post",
+    "post_json",
+    "post_res",
+    "put",
+    "put_res",
+    "request",
+    "response_manager",
 }
-BLOCKING_ATTRIBUTES = {
+_REQUESTS_METHODS = {
+    "delete",
+    "get",
+    "head",
+    "patch",
+    "post",
+    "put",
+    "request",
+}
+_PATH_IO_METHODS = {
+    "exists",
     "glob",
+    "is_dir",
+    "is_file",
     "iterdir",
+    "mkdir",
     "open",
     "read_bytes",
     "read_text",
+    "rename",
     "rglob",
+    "stat",
+    "unlink",
     "write_bytes",
     "write_text",
 }
+_SHUTIL_METHODS = {
+    "copy",
+    "copyfile",
+    "copytree",
+    "make_archive",
+    "move",
+    "rmtree",
+    "unpack_archive",
+    "which",
+}
+_SUBPROCESS_METHODS = {
+    "Popen",
+    "call",
+    "check_call",
+    "check_output",
+    "run",
+}
+_OS_IO_METHODS = {"listdir", "scandir", "walk"}
+_SYSTEM_CONFIG_MEMORY_READS = {
+    "app.db.oper.SystemConfigOper.all",
+    "app.db.oper.SystemConfigOper.get",
+    "app.db.oper.systemconfig.SystemConfigOper.all",
+    "app.db.oper.systemconfig.SystemConfigOper.get",
+}
 
 
-def _call_name(node: ast.Call) -> str:
-    """把简单名称和属性调用还原为点分文本。"""
-    parts = []
-    target: ast.expr = node.func
-    while isinstance(target, ast.Attribute):
-        parts.append(target.attr)
-        target = target.value
-    if isinstance(target, ast.Name):
-        parts.append(target.id)
-    return ".".join(reversed(parts))
+@dataclass(frozen=True)
+class _Binding:
+    """描述由明确 import 或局部赋值建立的符号来源。"""
+
+    family: str
+    qualified_name: str
+    kind: str = "module"
 
 
-def _root_name(expression: ast.expr) -> str | None:
-    """返回属性调用最左侧的变量名。"""
-    while isinstance(expression, ast.Attribute):
-        expression = expression.value
-    return expression.id if isinstance(expression, ast.Name) else None
+_UNKNOWN_BINDING = _Binding("unknown", "", "unknown")
+_BLOCKING_FAMILIES = {
+    "os",
+    "requests",
+    "shutil",
+    "subprocess",
+    "sync_http",
+    "sync_oper",
+    "sync_path",
+    "time",
+}
 
 
-class _AsyncPathCollector(ast.NodeVisitor):
-    """用局部数据流识别 anyio AsyncPath 变量及其派生值。"""
+def _binding_for_qualified(qualified_name: str) -> _Binding:
+    """按稳定模块路径识别门禁关心的符号族。"""
+    if qualified_name == "app.adapters.network.http.RequestUtils":
+        return _Binding("sync_http", qualified_name, "class")
+    if qualified_name == "app.adapters.network.http.AsyncRequestUtils":
+        return _Binding("async_http", qualified_name, "class")
+    if qualified_name in {"requests.Session", "requests.sessions.Session"}:
+        return _Binding("sync_http", qualified_name, "class")
+    if qualified_name == "pathlib.Path":
+        return _Binding("sync_path", qualified_name, "class")
+    if qualified_name == "anyio.Path":
+        return _Binding("async_path", qualified_name, "class")
+    if qualified_name.startswith("app.db.oper.") and qualified_name.endswith("Oper"):
+        return _Binding("sync_oper", qualified_name, "class")
+    if qualified_name.startswith("requests."):
+        return _Binding("requests", qualified_name, "callable")
+    if qualified_name.startswith("shutil."):
+        return _Binding("shutil", qualified_name, "callable")
+    if qualified_name.startswith("subprocess."):
+        return _Binding("subprocess", qualified_name, "callable")
+    if qualified_name.startswith("os."):
+        return _Binding("os", qualified_name, "callable")
+    if qualified_name.startswith("time."):
+        return _Binding("time", qualified_name, "callable")
+    return _Binding("module", qualified_name)
 
-    def __init__(self, function: ast.AsyncFunctionDef) -> None:
-        """从参数注解初始化 AsyncPath 变量集合。"""
-        self.paths = {
-            argument.arg
-            for argument in (*function.args.posonlyargs, *function.args.args)
-            if argument.annotation and "AsyncPath" in ast.unparse(argument.annotation)
-        }
-        self.path_collections: set[str] = set()
+
+def _resolve_binding(
+    expression: ast.expr,
+    bindings: dict[str, _Binding],
+) -> _Binding | None:
+    """解析明确 import、构造、属性访问和简单路径派生。"""
+    if isinstance(expression, ast.Name):
+        return bindings.get(expression.id)
+    if isinstance(expression, ast.Attribute):
+        base = _resolve_binding(expression.value, bindings)
+        if not base or base.kind in {"collection", "unknown"}:
+            return None
+        qualified_name = f"{base.qualified_name}.{expression.attr}"
+        if base.kind == "module":
+            return _binding_for_qualified(qualified_name)
+        return _Binding(base.family, qualified_name, "callable")
+    if isinstance(expression, ast.Call):
+        target = _resolve_binding(expression.func, bindings)
+        if target and target.kind == "class":
+            return _Binding(target.family, target.qualified_name, "instance")
+        return None
+    if isinstance(expression, ast.BinOp) and isinstance(expression.op, ast.Div):
+        left = _resolve_binding(expression.left, bindings)
+        if left and left.family in {"async_path", "sync_path"}:
+            return left
+    if isinstance(expression, ast.IfExp):
+        return _merge_binding_options(
+            "",
+            (
+                _resolve_binding(expression.body, bindings),
+                _resolve_binding(expression.orelse, bindings),
+            ),
+        )
+    return None
+
+
+def _merge_binding_options(
+    name: str,
+    options: Sequence[_Binding | None],
+) -> _Binding | None:
+    """控制流合流时保留任一分支可能进入的同步阻塞类型。"""
+    if options and all(option == options[0] for option in options):
+        return options[0]
+    blocking = sorted(
+        (
+            option
+            for option in options
+            if option and option.family in _BLOCKING_FAMILIES
+        ),
+        key=lambda item: (item.family, item.qualified_name, item.kind),
+    )
+    if blocking:
+        return blocking[0]
+    if name == "open" and any(option is None for option in options):
+        return None
+    return _UNKNOWN_BINDING
+
+
+def _merge_binding_states(
+    states: Sequence[dict[str, _Binding]],
+) -> dict[str, _Binding]:
+    """合并互斥控制流的局部符号表。"""
+    names = set().union(*(state.keys() for state in states))
+    merged: dict[str, _Binding] = {}
+    for name in names:
+        binding = _merge_binding_options(name, tuple(state.get(name) for state in states))
+        if binding:
+            merged[name] = binding
+    return merged
+
+
+def _load_oper_methods(root: Path) -> dict[str, set[str]]:
+    """从 Oper 源码建立同步方法索引，避免按方法名猜测数据库调用。"""
+    methods: dict[str, set[str]] = {}
+    oper_root = root / "app/db/oper"
+    if not oper_root.is_dir():
+        return methods
+    for path in sorted(oper_root.glob("*.py")):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        module = ".".join(path.relative_to(root).with_suffix("").parts)
+        for node in tree.body:
+            if not isinstance(node, ast.ClassDef) or not node.name.endswith("Oper"):
+                continue
+            qualified_name = f"{module}.{node.name}"
+            sync_methods = {
+                child.name
+                for child in node.body
+                if isinstance(child, ast.FunctionDef)
+                and not child.name.startswith("__")
+            }
+            methods[qualified_name] = sync_methods
+            methods[f"app.db.oper.{node.name}"] = sync_methods
+    return methods
+
+
+class _ImportCollector(ast.NodeVisitor):
+    """收集模块级 import，作为每个 async 函数的初始符号表。"""
+
+    def __init__(self) -> None:
+        self.bindings: dict[str, _Binding] = {}
+
+    def visit_Import(self, node: ast.Import) -> None:
+        for alias in node.names:
+            local_name = alias.asname or alias.name.split(".", 1)[0]
+            qualified_name = alias.name if alias.asname else local_name
+            self.bindings[local_name] = _binding_for_qualified(qualified_name)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        if node.level or not node.module:
+            return
+        for alias in node.names:
+            if alias.name == "*":
+                continue
+            local_name = alias.asname or alias.name
+            qualified_name = f"{node.module}.{alias.name}"
+            self.bindings[local_name] = _binding_for_qualified(qualified_name)
+
+    def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None:
+        if isinstance(target, ast.Name):
+            self.bindings[target.id] = binding or _UNKNOWN_BINDING
+        elif isinstance(target, (ast.Tuple, ast.List)):
+            for element in target.elts:
+                self._bind_target(element, None)
 
     def visit_Assign(self, node: ast.Assign) -> None:
-        """识别 AsyncPath 构造和已知路径的 `/` 派生赋值。"""
-        is_async_path = self._is_async_path_expression(node.value)
-        if is_async_path:
-            for target in node.targets:
-                if isinstance(target, ast.Name):
-                    self.paths.add(target.id)
-        self.generic_visit(node)
-
-    def _is_async_path_expression(self, expression: ast.expr) -> bool:
-        """识别 AsyncPath 构造及任意层级的 `/` 路径派生表达式。"""
-        if isinstance(expression, ast.Call):
-            return _call_name(expression).endswith("AsyncPath")
-        if isinstance(expression, ast.Name):
-            return expression.id in self.paths
-        if isinstance(expression, ast.BinOp):
-            return self._is_async_path_expression(expression.left)
-        if isinstance(expression, ast.IfExp):
-            return (
-                self._is_async_path_expression(expression.body)
-                and self._is_async_path_expression(expression.orelse)
-            )
-        return False
+        binding = _resolve_binding(node.value, self.bindings)
+        for target in node.targets:
+            self._bind_target(target, binding)
 
     def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
-        """识别 `list[AsyncPath]` 等路径集合。"""
-        if isinstance(node.target, ast.Name):
-            annotation = ast.unparse(node.annotation)
-            if "AsyncPath" in annotation:
-                if "list" in annotation or "List" in annotation:
-                    self.path_collections.add(node.target.id)
-                else:
-                    self.paths.add(node.target.id)
-        self.generic_visit(node)
-
-    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
-        """AsyncPath.iterdir 产出的元素仍是 AsyncPath。"""
-        if isinstance(node.target, ast.Name) and isinstance(node.iter, ast.Call):
-            receiver = _root_name(node.iter.func)
-            if receiver in self.paths:
-                self.paths.add(node.target.id)
-        self.generic_visit(node)
-
-    def visit_For(self, node: ast.For) -> None:
-        """从 `list[AsyncPath]` 迭代得到的元素仍是 AsyncPath。"""
-        if (
-            isinstance(node.target, ast.Name)
-            and isinstance(node.iter, ast.Name)
-            and node.iter.id in self.path_collections
-        ):
-            self.paths.add(node.target.id)
-        self.generic_visit(node)
+        binding = _resolve_binding(node.value, self.bindings) if node.value else None
+        self._bind_target(node.target, binding)
 
     def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        """不分析嵌套同步函数。"""
+        return
 
     def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
-        """不分析嵌套异步函数。"""
+        return
+
+    def visit_ClassDef(self, node: ast.ClassDef) -> None:
+        return
 
 
 class _AsyncCallVisitor(ast.NodeVisitor):
-    """只收集一个 async 函数本体中的阻塞调用，不进入嵌套函数定义。"""
+    """用局部符号传播识别一个 async 函数直接执行的阻塞调用。"""
 
-    def __init__(self, async_paths: set[str]) -> None:
-        """初始化违规计数器和异步文件对象白名单。"""
+    def __init__(
+        self,
+        function: ast.FunctionDef | ast.AsyncFunctionDef,
+        module_bindings: dict[str, _Binding],
+        oper_methods: dict[str, set[str]],
+        *,
+        record_calls: bool,
+    ) -> None:
         self.calls: Counter[str] = Counter()
-        self._async_paths = async_paths
+        self.nested_functions: list[
+            tuple[ast.FunctionDef | ast.AsyncFunctionDef, dict[str, _Binding]]
+        ] = []
+        self._bindings = dict(module_bindings)
+        self._oper_methods = oper_methods
+        self._record_calls = record_calls
+        self._bind_arguments(function)
+
+    def _bind_arguments(
+        self,
+        function: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        arguments = (
+            *function.args.posonlyargs,
+            *function.args.args,
+            *function.args.kwonlyargs,
+        )
+        for argument in arguments:
+            binding = self._annotation_binding(argument.annotation)
+            if binding:
+                self._bindings[argument.arg] = binding
+            else:
+                self._bindings[argument.arg] = _UNKNOWN_BINDING
+
+    def _annotation_binding(self, annotation: ast.expr | None) -> _Binding | None:
+        if annotation is None:
+            return None
+        if isinstance(annotation, ast.Subscript):
+            container = ast.unparse(annotation.value).rsplit(".", 1)[-1]
+            elements = (
+                annotation.slice.elts
+                if isinstance(annotation.slice, ast.Tuple)
+                else (annotation.slice,)
+            )
+            for element in elements:
+                binding = self._annotation_binding(element)
+                if not binding:
+                    continue
+                if container in {"list", "List", "Sequence", "set", "tuple"}:
+                    return _Binding(
+                        binding.family,
+                        binding.qualified_name,
+                        "collection",
+                    )
+                if container in {"Annotated", "Optional", "Union"}:
+                    return binding
+            return None
+        if isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr):
+            return self._annotation_binding(annotation.left) or self._annotation_binding(
+                annotation.right
+            )
+        binding = _resolve_binding(annotation, self._bindings)
+        if binding and binding.kind == "class":
+            return _Binding(binding.family, binding.qualified_name, "instance")
+        return None
+
+    def _resolve(self, expression: ast.expr) -> _Binding | None:
+        return _resolve_binding(expression, self._bindings)
+
+    @staticmethod
+    def _call_label(binding: _Binding) -> str:
+        parts = binding.qualified_name.split(".")
+        if binding.family == "sync_oper" and len(parts) >= 2:
+            return ".".join(parts[-2:])
+        if binding.family in {"async_path", "sync_path"} and parts:
+            return f"Path.{parts[-1]}"
+        if binding.family == "sync_http" and len(parts) >= 2:
+            return ".".join(parts[-2:])
+        if binding.family in {"os", "requests", "shutil", "subprocess", "time"}:
+            return ".".join(parts[-2:])
+        return binding.qualified_name
+
+    def _record_call(self, node: ast.Call, binding: _Binding | None) -> None:
+        if not self._record_calls:
+            return
+        if not binding:
+            if isinstance(node.func, ast.Name) and node.func.id == "open":
+                self.calls["open"] += 1
+            return
+        method = binding.qualified_name.rsplit(".", 1)[-1]
+        blocked = False
+        if binding.family == "sync_http":
+            blocked = binding.kind == "callable" and method in _SYNC_HTTP_METHODS
+        elif binding.family == "requests":
+            blocked = method in _REQUESTS_METHODS
+        elif binding.family == "sync_path":
+            blocked = binding.kind == "callable" and method in _PATH_IO_METHODS
+        elif binding.family == "shutil":
+            blocked = method in _SHUTIL_METHODS
+        elif binding.family == "subprocess":
+            blocked = method in _SUBPROCESS_METHODS
+        elif binding.family == "os":
+            blocked = method in _OS_IO_METHODS
+        elif binding.family == "time":
+            blocked = method == "sleep"
+        elif binding.family == "sync_oper" and binding.kind == "callable":
+            class_name, method_name = binding.qualified_name.rsplit(".", 1)
+            blocked = (
+                method_name in self._oper_methods.get(class_name, set())
+                and binding.qualified_name not in _SYSTEM_CONFIG_MEMORY_READS
+            )
+        if blocked:
+            self.calls[self._call_label(binding)] += 1
+
+    def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None:
+        if isinstance(target, ast.Name):
+            self._bindings[target.id] = binding or _UNKNOWN_BINDING
+        elif isinstance(target, (ast.Tuple, ast.List)):
+            for element in target.elts:
+                self._bind_target(element, None)
+
+    def visit_Import(self, node: ast.Import) -> None:
+        collector = _ImportCollector()
+        collector.visit_Import(node)
+        self._bindings.update(collector.bindings)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        collector = _ImportCollector()
+        collector.visit_ImportFrom(node)
+        self._bindings.update(collector.bindings)
+
+    def visit_Assign(self, node: ast.Assign) -> None:
+        self.visit(node.value)
+        binding = self._resolve(node.value)
+        for target in node.targets:
+            self._bind_target(target, binding)
+
+    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
+        if node.value:
+            self.visit(node.value)
+        binding = self._resolve(node.value) if node.value else None
+        binding = binding or self._annotation_binding(node.annotation)
+        self._bind_target(node.target, binding)
+
+    def visit_AugAssign(self, node: ast.AugAssign) -> None:
+        self.visit(node.value)
+        self._bind_target(node.target, None)
+
+    def _visit_branch(
+        self,
+        statements: Sequence[ast.stmt],
+        initial: dict[str, _Binding],
+    ) -> dict[str, _Binding]:
+        saved = self._bindings
+        self._bindings = dict(initial)
+        for statement in statements:
+            self.visit(statement)
+        result = self._bindings
+        self._bindings = saved
+        return result
+
+    def visit_If(self, node: ast.If) -> None:
+        self.visit(node.test)
+        initial = dict(self._bindings)
+        body = self._visit_branch(node.body, initial)
+        alternate = self._visit_branch(node.orelse, initial) if node.orelse else initial
+        self._bindings = _merge_binding_states((body, alternate))
+
+    def visit_For(self, node: ast.For | ast.AsyncFor) -> None:
+        self.visit(node.iter)
+        initial = dict(self._bindings)
+        collection = self._resolve(node.iter)
+        element = (
+            _Binding(collection.family, collection.qualified_name, "instance")
+            if collection and collection.kind == "collection"
+            else None
+        )
+        self._bind_target(node.target, element)
+        for statement in node.body:
+            self.visit(statement)
+        iterated = dict(self._bindings)
+        completed = self._visit_branch(node.orelse, iterated)
+        self._bindings = _merge_binding_states((initial, completed))
+
+    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
+        self.visit_For(node)
+
+    def _visit_comprehension(
+        self,
+        generators: Sequence[ast.comprehension],
+        outputs: Sequence[ast.expr],
+    ) -> None:
+        saved = self._bindings
+        self._bindings = dict(saved)
+        for generator in generators:
+            self.visit(generator.iter)
+            collection = self._resolve(generator.iter)
+            element = (
+                _Binding(collection.family, collection.qualified_name, "instance")
+                if collection and collection.kind == "collection"
+                else None
+            )
+            self._bind_target(generator.target, element)
+            for condition in generator.ifs:
+                self.visit(condition)
+        for output in outputs:
+            self.visit(output)
+        self._bindings = saved
+
+    def visit_ListComp(self, node: ast.ListComp) -> None:
+        self._visit_comprehension(node.generators, (node.elt,))
+
+    def visit_SetComp(self, node: ast.SetComp) -> None:
+        self._visit_comprehension(node.generators, (node.elt,))
+
+    def visit_GeneratorExp(self, node: ast.GeneratorExp) -> None:
+        self._visit_comprehension(node.generators, (node.elt,))
+
+    def visit_DictComp(self, node: ast.DictComp) -> None:
+        self._visit_comprehension(node.generators, (node.key, node.value))
+
+    def visit_NamedExpr(self, node: ast.NamedExpr) -> None:
+        self.visit(node.value)
+        self._bind_target(node.target, self._resolve(node.value))
 
     def visit_Call(self, node: ast.Call) -> None:
-        """记录命中阻塞名单的调用并继续遍历参数表达式。"""
-        name = _call_name(node)
-        attribute = name.rsplit(".", 1)[-1]
-        receiver = _root_name(node.func)
-        async_safe = name.startswith("aiofiles.") or receiver in self._async_paths
-        if not async_safe and (
-            name in BLOCKING_EXACT or attribute in BLOCKING_ATTRIBUTES
-        ):
-            self.calls[name or attribute] += 1
+        binding = self._resolve(node.func)
+        self._record_call(node, binding)
+        if isinstance(node.func, ast.Lambda):
+            self._visit_lambda_defaults(node.func)
+            for argument in node.args:
+                self.visit(argument)
+            for keyword in node.keywords:
+                self.visit(keyword.value)
+            self.visit(node.func.body)
+            return
         self.generic_visit(node)
 
+    def _visit_definition_expressions(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        for decorator in node.decorator_list:
+            self.visit(decorator)
+        for default in node.args.defaults:
+            self.visit(default)
+        for keyword_default in node.args.kw_defaults:
+            if keyword_default:
+                self.visit(keyword_default)
+
+    def _visit_nested_function(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        self._visit_definition_expressions(node)
+        self.nested_functions.append((node, dict(self._bindings)))
+        self._bindings[node.name] = _UNKNOWN_BINDING
+
     def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        """嵌套同步函数不属于外层 async 的直接执行体。"""
+        self._visit_nested_function(node)
 
     def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
-        """嵌套异步函数由模块级收集器单独治理。"""
+        self._visit_nested_function(node)
+
+    def _visit_lambda_defaults(self, node: ast.Lambda) -> None:
+        for default in node.args.defaults:
+            self.visit(default)
+        for keyword_default in node.args.kw_defaults:
+            if keyword_default:
+                self.visit(keyword_default)
+
+    def visit_Lambda(self, node: ast.Lambda) -> None:
+        self._visit_lambda_defaults(node)
 
 
-def _async_functions(
+def _root_functions(
     tree: ast.Module,
 ) -> Iterator[tuple[str, ast.AsyncFunctionDef]]:
-    """产出模块顶层及类直接拥有的 async 函数限定名。"""
+    """产出模块顶层和类直接拥有的 async 入口。"""
     for node in tree.body:
         if isinstance(node, ast.AsyncFunctionDef):
             yield node.name, node
@@ -195,24 +582,63 @@ def _async_functions(
                     yield f"{node.name}.{method.name}", method
 
 
-def collect_async_blocking(root: Path = PROJECT_ROOT) -> dict[str, int]:
+def _scan_paths(root: Path, scan_roots: Sequence[str | Path]) -> Iterator[Path]:
+    """按稳定顺序产出存在的扫描目标。"""
+    for scan_root in scan_roots:
+        target = root / scan_root
+        if target.is_file():
+            yield target
+        elif target.is_dir():
+            yield from sorted(target.rglob("*.py"))
+
+
+def collect_async_blocking(
+    root: Path = PROJECT_ROOT,
+    scan_roots: Sequence[str | Path] = SCAN_ROOTS,
+) -> dict[str, int]:
     """扫描关键目录并以文件、函数、调用名聚合存量次数。"""
     debt: Counter[str] = Counter()
-    for scan_root in SCAN_ROOTS:
-        target = root / scan_root
-        paths = [target] if target.is_file() else sorted(target.rglob("*.py"))
-        for path in paths:
-            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
-            relative = path.relative_to(root).as_posix()
-            for qualname, function in _async_functions(tree):
-                collector = _AsyncPathCollector(function)
-                for statement in function.body:
-                    collector.visit(statement)
-                visitor = _AsyncCallVisitor(collector.paths)
-                for statement in function.body:
-                    visitor.visit(statement)
+    oper_methods = _load_oper_methods(root)
+    for path in _scan_paths(root, scan_roots):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        imports = _ImportCollector()
+        for statement in tree.body:
+            imports.visit(statement)
+        relative = path.relative_to(root).as_posix()
+        pending: list[
+            tuple[
+                str,
+                ast.FunctionDef | ast.AsyncFunctionDef,
+                dict[str, _Binding],
+            ]
+        ] = [
+            (qualname, function, imports.bindings)
+            for qualname, function in _root_functions(tree)
+        ]
+        pending_index = 0
+        while pending_index < len(pending):
+            qualname, function, lexical_bindings = pending[pending_index]
+            pending_index += 1
+            is_async = isinstance(function, ast.AsyncFunctionDef)
+            visitor = _AsyncCallVisitor(
+                function,
+                lexical_bindings,
+                oper_methods,
+                record_calls=is_async,
+            )
+            for statement in function.body:
+                visitor.visit(statement)
+            if is_async:
                 for call_name, count in visitor.calls.items():
                     debt[f"{relative}:{qualname}:{call_name}"] += count
+            pending.extend(
+                (
+                    f"{qualname}.{nested.name}",
+                    nested,
+                    bindings,
+                )
+                for nested, bindings in visitor.nested_functions
+            )
     return dict(sorted(debt.items()))
 
 

tests/fixtures/architecture/async-blocking-baseline.json

@@ -1 +1,12 @@
-{}
+{
+  "app/agent/tools/impl/_plugin_tool_utils.py:uninstall_plugin_runtime:shutil.rmtree": 1,
+  "app/agent/tools/impl/scrape_metadata.py:ScrapeMetadataTool.run:Path.exists": 1,
+  "app/agent/tools/impl/scrape_metadata.py:ScrapeMetadataTool.run:Path.is_dir": 1,
+  "app/chain/media.py:MediaChain._async_music_album_dir_fallback:Path.exists": 1,
+  "app/chain/media.py:MediaChain._async_music_album_dir_fallback:Path.is_file": 1,
+  "app/chain/media.py:MediaChain.async_recognize_music_album_directory:Path.is_dir": 1,
+  "app/modules/acoustid/__init__.py:AcoustIdModule.async_identify_music_by_fingerprint:Path.is_file": 1,
+  "app/modules/discord/discord.py:Discord._send_file:Path.exists": 1,
+  "app/modules/discord/discord.py:Discord._send_file:Path.is_file": 1,
+  "app/scheduler.py:Scheduler.execute_agent_task:AgentTaskOper.get": 1
+}

tests/fixtures/architecture/configuration-debt-baseline.json

@@ -9,7 +9,7 @@
     "root": "app"
   },
   "settings_imports": {
-    "count": 136,
+    "count": 135,
     "files": [
       "app/adapters/cache/backends.py",
       "app/adapters/cache/redis.py",
@@ -47,7 +47,6 @@
       "app/agent/tools/impl/send_voice_message.py",
       "app/agent/tools/impl/update_agent_task.py",
       "app/agent/tools/impl/update_system_settings.py",
-      "app/application/formatting.py",
       "app/application/maintenance.py",
       "app/application/rss.py",
       "app/application/security/auth.py",

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6393,
-  "edge_sha256": "ea375f19071a37a9c72bd01ce9fa4a070e64ad7f14a87d2e5013b5c7440680cd",
+  "edge_count": 6397,
+  "edge_sha256": "801b80e33c626aecf70e5320a2c1c81db401549faa652e718be7386bac593f43",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -2494,12 +2494,13 @@
     "app.application.downloader -> app.schemas",
     "app.application.downloader -> app.schemas.system",
     "app.application.downloader -> app.schemas.types",
+    "app.application.formatting -> app.application",
+    "app.application.formatting -> app.application.configuration",
     "app.application.formatting -> app.domain",
     "app.application.formatting -> app.domain.meta",
     "app.application.formatting -> app.domain.meta.metabase",
     "app.application.formatting -> app.domain.metainfo",
     "app.application.formatting -> app.runtime",
-    "app.application.formatting -> app.runtime.config",
     "app.application.formatting -> app.runtime.log",
     "app.application.formatting -> app.schemas",
     "app.application.formatting -> app.schemas.transfer",
@@ -3663,6 +3664,9 @@
     "app.db.session -> app.runtime.observability",
     "app.doctor.checks -> app.adapters",
     "app.doctor.checks -> app.adapters.system",
+    "app.doctor.checks -> app.adapters.system.backup",
+    "app.doctor.checks -> app.adapters.system.backup.database",
+    "app.doctor.checks -> app.adapters.system.backup.files",
     "app.doctor.checks -> app.adapters.system.host",
     "app.doctor.checks -> app.doctor",
     "app.doctor.checks -> app.doctor.models",

tests/test_async_blocking_gate.py

@@ -1,10 +1,33 @@
 """async 阻塞调用 ratchet 与 debug 模式测试。"""
 
 import asyncio
+import textwrap
+from pathlib import Path
 
 import pytest
 
-from scripts.architecture.async_blocking import SCAN_ROOTS, compare_async_blocking
+from scripts.architecture.async_blocking import (
+    SCAN_ROOTS,
+    collect_async_blocking,
+    compare_async_blocking,
+)
+
+
+def _scan_source(
+    tmp_path: Path,
+    source: str,
+    *,
+    oper_sources: dict[str, str] | None = None,
+) -> dict[str, int]:
+    """构造最小仓库并通过公开扫描入口验证源码，而非测试内部 AST 细节。"""
+    api_root = tmp_path / "app/api"
+    api_root.mkdir(parents=True)
+    (api_root / "sample.py").write_text(textwrap.dedent(source), encoding="utf-8")
+    for filename, oper_source in (oper_sources or {}).items():
+        oper_path = tmp_path / "app/db/oper" / filename
+        oper_path.parent.mkdir(parents=True, exist_ok=True)
+        oper_path.write_text(textwrap.dedent(oper_source), encoding="utf-8")
+    return collect_async_blocking(tmp_path, scan_roots=("app/api",))
 
 
 def test_async_blocking_scan_covers_runtime_entrypoints() -> None:
@@ -51,6 +74,276 @@ def test_async_blocking_ratchet_allows_removal_and_rejects_growth() -> None:
     assert any("新增" in problem for problem in problems)
 
 
+@pytest.mark.parametrize(
+    ("source", "expected"),
+    [
+        (
+            """
+            from app.adapters.network.http import RequestUtils as RU
+
+            async def load():
+                client = RU()
+                alias = client
+                return alias.get_res("https://example.com")
+            """,
+            {"app/api/sample.py:load:RequestUtils.get_res": 1},
+        ),
+        (
+            """
+            import app.adapters.network.http as http
+
+            async def submit():
+                return http.RequestUtils().post_res("https://example.com")
+            """,
+            {"app/api/sample.py:submit:RequestUtils.post_res": 1},
+        ),
+        (
+            """
+            from pathlib import Path as P
+
+            async def inspect_file():
+                path = P("payload") / "item.json"
+                return path.exists()
+            """,
+            {"app/api/sample.py:inspect_file:Path.exists": 1},
+        ),
+        (
+            """
+            import shutil as files
+            from subprocess import run as run_process
+
+            async def cleanup():
+                files.rmtree("payload")
+                run_process(["true"])
+            """,
+            {
+                "app/api/sample.py:cleanup:shutil.rmtree": 1,
+                "app/api/sample.py:cleanup:subprocess.run": 1,
+            },
+        ),
+        (
+            """
+            import os as operating_system
+            import time as clock
+            from requests import Session as HttpSession
+
+            async def legacy_io():
+                open("payload")
+                operating_system.listdir(".")
+                clock.sleep(0.1)
+                HttpSession().get("https://example.com")
+            """,
+            {
+                "app/api/sample.py:legacy_io:Session.get": 1,
+                "app/api/sample.py:legacy_io:open": 1,
+                "app/api/sample.py:legacy_io:os.listdir": 1,
+                "app/api/sample.py:legacy_io:time.sleep": 1,
+            },
+        ),
+        (
+            """
+            from pathlib import Path
+
+            async def outer():
+                async def inner():
+                    return Path("payload").read_text()
+                return await inner()
+            """,
+            {"app/api/sample.py:outer.inner:Path.read_text": 1},
+        ),
+        (
+            """
+            from pathlib import Path
+
+            async def inspect_files(files: list[Path]):
+                for file in files:
+                    if file.is_file():
+                        return file
+                return None
+            """,
+            {"app/api/sample.py:inspect_files:Path.is_file": 1},
+        ),
+    ],
+)
+def test_async_blocking_scan_resolves_imports_aliases_and_nested_async(
+    tmp_path: Path,
+    source: str,
+    expected: dict[str, int],
+) -> None:
+    """别名、简单局部传播和嵌套 async 都必须进入真实扫描结果。"""
+    assert _scan_source(tmp_path, source) == expected
+
+
+def test_async_blocking_scan_uses_oper_source_method_kinds(tmp_path: Path) -> None:
+    """Oper 仅按源码中真实同步方法报告，不依赖方法名前缀猜测。"""
+    source = """
+        from app.db.oper import SiteOper
+        from app.db.oper.site import SiteOper as SO
+
+        async def load(oper: SO):
+            oper.list()
+            return await oper.get_by_id(1)
+
+        async def load_from_facade():
+            return SiteOper().list()
+    """
+    oper_sources = {
+        "site.py": """
+            class SiteOper:
+                def list(self):
+                    return []
+
+                async def get_by_id(self, site_id):
+                    return None
+        """,
+    }
+
+    assert _scan_source(tmp_path, source, oper_sources=oper_sources) == {
+        "app/api/sample.py:load:SiteOper.list": 1,
+        "app/api/sample.py:load_from_facade:SiteOper.list": 1,
+    }
+
+
+def test_async_blocking_scan_exempts_async_apis_and_memory_reads(
+    tmp_path: Path,
+) -> None:
+    """异步实现、受控 worker 与内存配置读取不得成为阻塞债务。"""
+    source = """
+        import asyncio
+        import anyio
+        import subprocess
+        from anyio import Path as AsyncPath
+        from app.agent.tools.base import run_agent_blocking
+        from app.adapters.network.http import AsyncRequestUtils
+        from app.db.oper.systemconfig import SystemConfigOper
+        from app.runtime.execution import run_in_threadpool as runtime_worker
+        from fastapi.concurrency import run_in_threadpool
+
+        async def load(config: SystemConfigOper):
+            await AsyncRequestUtils().get_res("https://example.com")
+            await AsyncPath("payload").exists()
+            await run_in_threadpool(lambda: subprocess.run(["true"]))
+            await runtime_worker(lambda: subprocess.run(["true"]))
+            await run_agent_blocking("plugin", lambda: subprocess.run(["true"]))
+            await asyncio.to_thread(lambda: subprocess.run(["true"]))
+            await anyio.to_thread.run_sync(lambda: subprocess.run(["true"]))
+            deferred = lambda: subprocess.run(["true"])
+            assert deferred
+            return config.get("key")
+
+        def ordinary_sync():
+            subprocess.run(["true"])
+    """
+    oper_sources = {
+        "systemconfig.py": """
+            class SystemConfigOper:
+                def get(self, key):
+                    return key
+        """,
+    }
+
+    assert _scan_source(tmp_path, source, oper_sources=oper_sources) == {}
+
+
+def test_async_blocking_scan_checks_worker_arguments_evaluated_on_loop(
+    tmp_path: Path,
+) -> None:
+    """worker 调用前同步求值的普通参数仍在事件循环中执行。"""
+    source = """
+        import asyncio
+        from pathlib import Path
+
+        async def load():
+            await asyncio.to_thread(print, Path("payload").read_text())
+    """
+
+    assert _scan_source(tmp_path, source) == {
+        "app/api/sample.py:load:Path.read_text": 1,
+    }
+
+
+def test_async_blocking_scan_merges_branch_bindings_conservatively(
+    tmp_path: Path,
+) -> None:
+    """任一互斥分支可能产生同步对象时，合流调用仍属于阻塞风险。"""
+    source = """
+        from anyio import Path as AsyncPath
+        from pathlib import Path
+
+        async def load(use_sync: bool):
+            if use_sync:
+                target = Path("payload")
+            else:
+                target = AsyncPath("payload")
+            return target.read_text()
+    """
+
+    assert _scan_source(tmp_path, source) == {
+        "app/api/sample.py:load:Path.read_text": 1,
+    }
+
+
+def test_nested_async_inherits_bindings_from_definition_scope(tmp_path: Path) -> None:
+    """嵌套 async 使用定义点已有的局部 import 和别名。"""
+    source = """
+        async def outer():
+            from pathlib import Path as LocalPath
+            alias = LocalPath
+
+            async def inner():
+                return alias("payload").read_text()
+
+            return await inner()
+    """
+
+    assert _scan_source(tmp_path, source) == {
+        "app/api/sample.py:outer.inner:Path.read_text": 1,
+    }
+
+
+def test_definition_time_expressions_remain_in_async_execution_body(
+    tmp_path: Path,
+) -> None:
+    """延迟函数体不扫描，但默认值和 decorator 在定义时立即求值。"""
+    source = """
+        import asyncio
+        from pathlib import Path
+
+        def register(value):
+            return lambda function: function
+
+        async def outer():
+            await asyncio.to_thread(
+                lambda value=Path("lambda").read_text(): value
+            )
+
+            @register(Path("decorator").read_text())
+            async def inner(value=Path("default").read_text()):
+                return value
+
+            return await inner()
+    """
+
+    assert _scan_source(tmp_path, source) == {
+        "app/api/sample.py:outer:Path.read_text": 3,
+    }
+
+
+def test_local_shadowing_does_not_reuse_import_or_builtin_bindings(
+    tmp_path: Path,
+) -> None:
+    """参数和 comprehension target 会遮蔽同名 builtin 或导入符号。"""
+    source = """
+        from pathlib import Path
+
+        async def invoke(open, items):
+            open()
+            return [Path.exists() for Path in items]
+    """
+
+    assert _scan_source(tmp_path, source) == {}
+
+
 @pytest.mark.asyncio
 async def test_asyncio_debug_is_enabled_for_async_tests() -> None:
     """专项异步测试必须启用慢 callback 和阻塞诊断所需的 debug 模式。"""

tests/test_database_backup_adapters.py

@@ -6,12 +6,14 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import Mock
 
+import pytest
 from sqlalchemy import create_engine
 from sqlalchemy.engine import make_url
 
 from app.adapters.system.backup.database import (
     PostgreSQLBackupBackend,
     SQLiteBackupBackend,
+    verify_database_backup,
 )
 
 
@@ -34,6 +36,15 @@ def test_sqlite_backup_includes_committed_wal_data(tmp_path: Path) -> None:
     engine.dispose()
 
 
+def test_sqlite_backup_can_be_verified_without_active_engine(tmp_path: Path) -> None:
+    """Doctor 等离线入口不应为校验备份而构造活动数据库引擎。"""
+    artifact = tmp_path / "backup.db"
+    with sqlite3.connect(artifact) as connection:
+        connection.execute("CREATE TABLE entries (value TEXT NOT NULL)")
+
+    assert verify_database_backup(artifact, db_type="sqlite").valid is True
+
+
 def test_sqlite_restore_replaces_database_and_removes_old_wal_files(tmp_path: Path) -> None:
     source = tmp_path / "user.db"
     backup = tmp_path / "backup.db"
@@ -66,6 +77,11 @@ class _Runner:
         return subprocess.CompletedProcess(command, 0, stdout, "")
 
 
+class _FailedRunner:
+    def __call__(self, command, **_kwargs):
+        return subprocess.CompletedProcess(command, 1, "", "invalid archive")
+
+
 def _postgres_backend(runner: _Runner) -> PostgreSQLBackupBackend:
     engine = SimpleNamespace(
         url=make_url(
@@ -114,6 +130,55 @@ def test_postgresql_verify_and_restore_use_pg_restore(tmp_path: Path) -> None:
     assert restore_kwargs["env"]["PGPASSWORD"] == "secret"
 
 
+def test_postgresql_backup_can_be_verified_without_active_engine(tmp_path: Path) -> None:
+    """PostgreSQL 归档校验只依赖 pg_restore，不连接活动数据库。"""
+    runner = _Runner()
+    artifact = tmp_path / "backup.dump"
+    artifact.write_bytes(b"PGDMP")
+
+    result = verify_database_backup(
+        artifact,
+        db_type="postgresql",
+        runner=runner,
+        tool_resolver=lambda executable: executable,
+    )
+
+    assert result.valid is True
+    command, kwargs = runner.calls[0]
+    assert command == ["pg_restore", "--list", str(artifact)]
+    assert "PGPASSWORD" not in kwargs["env"]
+    assert "PGSSLMODE" not in kwargs["env"]
+
+
+def test_postgresql_offline_verify_rejects_invalid_archive(tmp_path: Path) -> None:
+    """pg_restore 无法读取归档目录时备份必须判定为无效。"""
+    artifact = tmp_path / "backup.dump"
+    artifact.write_bytes(b"invalid")
+
+    result = verify_database_backup(
+        artifact,
+        db_type="postgresql",
+        runner=_FailedRunner(),
+        tool_resolver=lambda executable: executable,
+    )
+
+    assert result.valid is False
+    assert result.detail == "pg_restore 退出码 1"
+
+
+def test_postgresql_offline_verify_reports_missing_client(tmp_path: Path) -> None:
+    """缺少 pg_restore 时离线校验应给出可执行的安装提示。"""
+    artifact = tmp_path / "backup.dump"
+    artifact.write_bytes(b"PGDMP")
+
+    with pytest.raises(RuntimeError, match="PostgreSQL client"):
+        verify_database_backup(
+            artifact,
+            db_type="postgresql",
+            tool_resolver=lambda _executable: None,
+        )
+
+
 def test_postgresql_source_install_reports_missing_native_client() -> None:
     runner = _Runner()
     engine = SimpleNamespace(

tests/test_doctor.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta
+import sqlite3
 from types import SimpleNamespace
 
 from app.runtime.config import settings
-from app.doctor import checks, run_doctor
+from app.doctor import checks
 from app.doctor.formatters import format_json_report, format_text_report
 from app.doctor.models import DoctorFinding, DoctorFindingStatus, DoctorSeverity
-from app.doctor.runner import DoctorRunner
+from app.doctor.runner import DoctorRunner, run_doctor
 
 
 def _current_log_timestamp() -> str:
@@ -33,6 +34,98 @@ def test_doctor_report_has_stable_json_shape(tmp_path, monkeypatch):
     assert any(item["id"] == "runtime.paths" for item in payload["findings"])
 
 
+def test_doctor_reports_valid_backup_when_sqlite_database_is_corrupt(
+    tmp_path,
+    monkeypatch,
+):
+    """主数据库损坏时 Doctor 仍应离线校验备份并给出还原命令。"""
+    monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
+    (tmp_path / "user.db").write_bytes(b"not a sqlite database")
+    backup_dir = settings.DATABASE_BACKUP_PATH
+    backup_dir.mkdir(parents=True)
+    backup = backup_dir / "sqlite_20260822_030000.db"
+    with sqlite3.connect(backup) as connection:
+        connection.execute("CREATE TABLE entries (value TEXT NOT NULL)")
+    (backup_dir / "sqlite_20260822_040000.db").write_bytes(b"invalid newer backup")
+
+    runner = DoctorRunner()
+    checks._check_database(runner)
+
+    assert runner.report.find("database.sqlite_open_failed") is not None
+    finding = runner.report.find("database.backup_recovery")
+    assert finding is not None
+    assert finding.status == DoctorFindingStatus.Ok
+    assert finding.context["backups"][0]["valid"] is False
+    assert finding.context["backups"][1]["valid"] is True
+    assert finding.context["restore_command"] == (
+        "moviepilot database restore sqlite_20260822_030000.db --confirm"
+    )
+    assert finding.context["restore_command"] in finding.recommendation
+
+
+def test_doctor_distinguishes_missing_and_mismatched_backups(tmp_path, monkeypatch):
+    """无备份与仅存在其他数据库类型备份应生成不同诊断结论。"""
+    monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
+    runner = DoctorRunner()
+    checks._check_database_backups(runner)
+    missing = runner.report.find("database.backup_recovery")
+    assert missing is not None
+    assert missing.status == DoctorFindingStatus.Skipped
+
+    backup_dir = settings.DATABASE_BACKUP_PATH
+    backup_dir.mkdir(parents=True)
+    (backup_dir / "postgresql_20260822_030000.dump").write_bytes(b"PGDMP")
+    runner = DoctorRunner()
+    checks._check_database_backups(runner)
+    mismatched = runner.report.find("database.backup_recovery")
+    assert mismatched is not None
+    assert mismatched.status == DoctorFindingStatus.Degraded
+    assert mismatched.context["mismatched"] == ["postgresql_20260822_030000.dump"]
+
+
+def test_doctor_reports_invalid_backup_without_modifying_it(tmp_path, monkeypatch):
+    """Doctor --fix 也只校验备份，不覆盖或删除无效文件。"""
+    monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
+    backup_dir = settings.DATABASE_BACKUP_PATH
+    backup_dir.mkdir(parents=True)
+    backup = backup_dir / "sqlite_20260822_030000.db"
+    original = b"invalid sqlite backup"
+    backup.write_bytes(original)
+
+    runner = DoctorRunner(fix=True)
+    checks._check_database_backups(runner)
+
+    finding = runner.report.find("database.backup_recovery")
+    assert finding is not None
+    assert finding.status == DoctorFindingStatus.Failed
+    assert finding.affects_report_status is False
+    assert finding.context["backups"][0]["valid"] is False
+    assert backup.read_bytes() == original
+    assert runner.report.status.value == "healthy"
+
+
+def test_doctor_exposes_missing_pg_restore_in_text_finding(tmp_path, monkeypatch):
+    """PostgreSQL 离线校验工具缺失时应直接告诉用户如何补齐。"""
+    def missing_pg_restore(*_args, **_kwargs):
+        raise RuntimeError("未找到 pg_restore，请安装 PostgreSQL client 并加入 PATH")
+
+    monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
+    monkeypatch.setattr(settings, "DB_TYPE", "postgresql")
+    backup_dir = settings.DATABASE_BACKUP_PATH
+    backup_dir.mkdir(parents=True)
+    (backup_dir / "postgresql_20260822_030000.dump").write_bytes(b"PGDMP")
+    monkeypatch.setattr(checks, "verify_database_backup", missing_pg_restore)
+
+    runner = DoctorRunner()
+    checks._check_database_backups(runner)
+
+    finding = runner.report.find("database.backup_recovery")
+    assert finding is not None
+    assert finding.status == DoctorFindingStatus.Failed
+    assert "未找到 pg_restore" in finding.detail
+    assert "PostgreSQL client" in format_text_report(runner.report)
+
+
 def test_doctor_formatters_include_status_and_finding(tmp_path, monkeypatch):
     """doctor 文本和 JSON 格式化应展示状态与诊断项。"""
     monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))

tests/test_episode_format_helper.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -22,8 +23,12 @@ def _make_file(name: str, size: int = 150 * 1024 * 1024) -> FileItem:
 @pytest.fixture(autouse=True)
 def _patch_media_exts(monkeypatch):
     monkeypatch.setattr(
-        "app.application.formatting.settings.RMT_MEDIAEXT",
-        [".mkv", ".mp4"],
+        "app.application.formatting.get_chain_runtime_config_snapshot",
+        lambda: SimpleNamespace(
+            video_extensions=(".mkv", ".mp4"),
+            subtitle_extensions=(".srt", ".ass"),
+            audio_extensions=(".flac", ".mp3"),
+        ),
     )
 
 

tests/test_release_supply_chain.py

@@ -1,5 +1,6 @@
 """正式镜像发布的供应链门禁合同。"""
 
+from datetime import date
 from pathlib import Path
 
 from ruamel.yaml import YAML
@@ -8,6 +9,7 @@ from ruamel.yaml import YAML
 ROOT = Path(__file__).resolve().parents[1]
 DOCKERFILE = ROOT / "docker" / "Dockerfile"
 RELEASE_WORKFLOW = ROOT / ".github" / "workflows" / "build-v3.yml"
+TRIVY_IGNORE = ROOT / ".trivyignore.yaml"
 
 
 def _load_workflow() -> dict:
@@ -94,6 +96,18 @@ def test_release_scans_both_architectures_before_registry_login_and_publish() ->
     assert last_scan < names.index("Publish multi-architecture image")
 
 
+def test_vulnerability_ignores_are_scoped_justified_and_time_bounded() -> None:
+    """漏洞豁免必须限定制品范围，并保留复查期限和接受理由。"""
+    yaml = YAML(typ="safe")
+    vulnerabilities = yaml.load(TRIVY_IGNORE.read_text(encoding="utf-8"))["vulnerabilities"]
+
+    for vulnerability in vulnerabilities:
+        assert vulnerability["paths"]
+        assert vulnerability["purls"]
+        assert vulnerability["statement"]
+        assert isinstance(vulnerability["expired_at"], date)
+
+
 def test_publish_reuses_scanned_architecture_caches_without_refreshing_base() -> None:
     """发布构建复用已扫描候选缓存，不得在扫描后重新拉取未审计基础镜像。"""
     workflow = _load_workflow()