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

app/db/__init__.py

@@ -1,12 +1,60 @@
 import asyncio
 from typing import Any, Generator, List, Optional, Self, Tuple, AsyncGenerator, Union
 
-from sqlalchemy import NullPool, QueuePool, and_, create_engine, inspect, text, select, delete, Column, Integer, \
+from sqlalchemy import NullPool, QueuePool, and_, create_engine, event, inspect, text, select, delete, Column, Integer, \
     Sequence, Identity
+from sqlalchemy.engine import Engine as SQLAlchemyEngine, ExceptionContext
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.orm import Session, as_declarative, declared_attr, scoped_session, sessionmaker
 
 from app.core.config import settings
+from app.log import logger
+
+
+def _database_error_metadata(error: BaseException) -> Optional[dict[str, Any]]:
+    """提取 SQLite 与 PostgreSQL 驱动提供的稳定错误分类字段。"""
+    metadata = {"error_type": type(error).__name__}
+
+    # DBAPI 驱动字段并不共享统一类型，动态读取可同时兼容 sqlite3、psycopg2 与 asyncpg。
+    sqlite_errorcode = getattr(error, "sqlite_errorcode", None)
+    sqlite_errorname = getattr(error, "sqlite_errorname", None)
+    if sqlite_errorcode is not None or sqlite_errorname:
+        if sqlite_errorcode is not None:
+            metadata["error_code"] = sqlite_errorcode
+        if sqlite_errorname:
+            metadata["error_name"] = sqlite_errorname
+        return metadata
+
+    sqlstate = getattr(error, "sqlstate", None) or getattr(error, "pgcode", None)
+    if not sqlstate:
+        sqlstate = getattr(getattr(error, "diag", None), "sqlstate", None)
+    if sqlstate:
+        metadata["sqlstate"] = sqlstate
+        return metadata
+
+    return None
+
+
+def _log_database_error(exception_context: ExceptionContext) -> None:
+    """记录非敏感驱动错误码，并保持 SQLAlchemy 原有异常传播。"""
+    metadata = _database_error_metadata(exception_context.original_exception)
+    if not metadata:
+        return
+
+    dialect = exception_context.dialect
+    fields = {
+        "database": dialect.name,
+        "driver": dialect.driver,
+        **metadata,
+    }
+    logger.error(
+        "数据库驱动异常：" + ", ".join(f"{key}={value}" for key, value in fields.items())
+    )
+
+
+def _register_database_error_logging(engine: SQLAlchemyEngine) -> None:
+    """为主程序 Engine 注册统一的底层驱动错误诊断。"""
+    event.listen(engine, "handle_error", _log_database_error)
 
 
 def get_id_column():
@@ -71,6 +119,7 @@ def _get_sqlite_engine(is_async: bool = False):
 
         # 创建数据库引擎
         engine = create_engine(**_db_kwargs)
+        _register_database_error_logging(engine)
 
         # 设置WAL模式
         _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE"
@@ -91,6 +140,7 @@ def _get_sqlite_engine(is_async: bool = False):
         }
         # 创建异步数据库引擎
         async_engine = create_async_engine(**_db_kwargs)
+        _register_database_error_logging(async_engine.sync_engine)
 
         # 设置WAL模式
         _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE"
@@ -146,6 +196,7 @@ def _get_postgresql_engine(is_async: bool = False):
 
         # 创建数据库引擎
         engine = create_engine(**_db_kwargs)
+        _register_database_error_logging(engine)
         print(f"PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}")
 
         return engine
@@ -163,6 +214,7 @@ def _get_postgresql_engine(is_async: bool = False):
         }
         # 创建异步数据库引擎
         async_engine = create_async_engine(**_db_kwargs)
+        _register_database_error_logging(async_engine.sync_engine)
         print(f"Async PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}")
 
         return async_engine

app/helper/doh.py

@@ -18,8 +18,10 @@ from app.log import logger
 from app.utils.mixins import ConfigReloadMixin
 from app.utils.singleton import Singleton
 
-# 定义一个全局线程池执行器
-_executor = concurrent.futures.ThreadPoolExecutor()
+# DoH 关闭时需要释放线程池；保持惰性创建可避免未启用 DoH 时占用进程级资源
+_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
+_executor_lock = Lock()
+_doh_enabled = False
 
 # 定义默认的DoH配置
 _doh_timeout = 5
@@ -29,11 +31,21 @@ _doh_lock = Lock()
 _orig_getaddrinfo = socket.getaddrinfo
 
 
+def _get_executor_locked() -> concurrent.futures.ThreadPoolExecutor:
+    """在持有执行器锁时按需获取 DoH 查询线程池"""
+    global _executor
+    if _executor is None:
+        _executor = concurrent.futures.ThreadPoolExecutor()
+    return _executor
+
+
 def enable_doh(enable: bool) -> None:
     """
     对 socket.getaddrinfo 进行补丁
     """
 
+    global _doh_enabled
+
     def _patched_getaddrinfo(host: str, *args, **kwargs):
         """
         socket.getaddrinfo的补丁版本。
@@ -47,9 +59,15 @@ def enable_doh(enable: bool) -> None:
             logger.info(f"已解析 [{host}] 为 [{ip}] (缓存)")
             return _orig_getaddrinfo(ip, *args, **kwargs)
         # 使用DoH解析主机
-        futures = []
-        for resolver in settings.DOH_RESOLVERS.split(","):
-            futures.append(_executor.submit(_doh_query, resolver, host))
+        with _executor_lock:
+            if not _doh_enabled:
+                return _orig_getaddrinfo(host, *args, **kwargs)
+            executor = _get_executor_locked()
+            # 一次解析的任务必须在同一临界区提交完，避免关闭过程中部分任务落入新线程池
+            futures = [
+                executor.submit(_doh_query, resolver, host)
+                for resolver in settings.DOH_RESOLVERS.split(",")
+            ]
         for future in concurrent.futures.as_completed(futures):
             ip = future.result()
             if ip is not None:
@@ -60,11 +78,9 @@ def enable_doh(enable: bool) -> None:
                 break
         return _orig_getaddrinfo(host, *args, **kwargs)
 
-    if enable:
-        # 替换 socket.getaddrinfo 方法
-        socket.getaddrinfo = _patched_getaddrinfo
-    else:
-        socket.getaddrinfo = _orig_getaddrinfo
+    with _executor_lock:
+        _doh_enabled = enable
+        socket.getaddrinfo = _patched_getaddrinfo if enable else _orig_getaddrinfo
 
 
 class DohHelper(ConfigReloadMixin, metaclass=Singleton):
@@ -77,14 +93,31 @@ class DohHelper(ConfigReloadMixin, metaclass=Singleton):
         enable_doh(settings.DOH_ENABLE)
 
     def on_config_changed(self) -> None:
+        if not settings.DOH_ENABLE:
+            self.shutdown()
+            return
         with _doh_lock:
             # DOH配置有变动的情况下，清空缓存
             _doh_cache.clear()
-        enable_doh(settings.DOH_ENABLE)
+        enable_doh(True)
 
     def get_reload_name(self) -> str:
         return 'DoH'
 
+    def shutdown(self) -> None:
+        """恢复系统 DNS 并释放 DoH 查询线程池"""
+        global _executor, _doh_enabled
+        with _executor_lock:
+            _doh_enabled = False
+            socket.getaddrinfo = _orig_getaddrinfo
+            executor = _executor
+            _executor = None
+        with _doh_lock:
+            _doh_cache.clear()
+        if executor:
+            executor.shutdown(wait=True)
+
+
 def _doh_query(resolver: str, host: str) -> Optional[str]:
     """
     使用给定的DoH解析器查询给定主机的IP地址。

app/helper/message.py

@@ -605,6 +605,7 @@ class MessageQueueManager(metaclass=SingletonClass):
         self.check_interval = check_interval
 
         self._running = True
+        self._stop_event = threading.Event()
         self.thread = threading.Thread(target=self._monitor_loop, daemon=True)
         self.thread.start()
 
@@ -752,13 +753,15 @@ class MessageQueueManager(metaclass=SingletonClass):
                         logger.info(f"队列剩余消息：{self.queue.qsize()}")
                     except queue.Empty:
                         break
-            time.sleep(self.check_interval)
+            if self._stop_event.wait(self.check_interval):
+                break
 
     def stop(self) -> None:
         """
         停止队列管理器
         """
         self._running = False
+        self._stop_event.set()
         logger.info("正在停止消息队列...")
         self.thread.join()
         logger.info("消息队列已停止")
@@ -841,7 +844,8 @@ def stop_message():
     """
     停止消息服务
     """
-    # 停止消息队列
-    MessageQueueManager().stop()
-    # 关闭消息演染器
-    TemplateHelper().close()
+    # 只关闭已启动的服务，避免清理路径反向创建后台线程和缓存
+    if queue_manager := MessageQueueManager.get_existing_instance():
+        queue_manager.stop()
+    if template_helper := TemplateHelper.get_existing_instance():
+        template_helper.close()

app/log.py

@@ -124,7 +124,7 @@ class NonBlockingFileHandler:
     """
     _instance = None
     _lock = threading.Lock()
-    _rotating_handlers = {}
+    _stop_sentinel = object()
 
     def __new__(cls):
         if cls._instance is None:
@@ -138,6 +138,9 @@ class NonBlockingFileHandler:
             return
 
         self._initialized = True
+        self._state_lock = threading.RLock()
+        self._handlers_lock = threading.Lock()
+        self._rotating_handlers = {}
         self._write_queue = queue.Queue(maxsize=log_settings.ASYNC_FILE_QUEUE_SIZE)
         self._executor = ThreadPoolExecutor(max_workers=log_settings.ASYNC_FILE_WORKERS,
                                             thread_name_prefix="LogWriter")
@@ -151,27 +154,28 @@ class NonBlockingFileHandler:
         """
         获取或创建RotatingFileHandler实例
         """
-        if file_path not in self._rotating_handlers:
-            # 确保目录存在
-            file_path.parent.mkdir(parents=True, exist_ok=True)
+        with self._handlers_lock:
+            if file_path not in self._rotating_handlers:
+                # 确保目录存在
+                file_path.parent.mkdir(parents=True, exist_ok=True)
 
-            # 创建RotatingFileHandler
-            handler = RotatingFileHandler(
-                filename=str(file_path),
-                maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES,
-                backupCount=log_settings.LOG_BACKUP_COUNT,
-                encoding='utf-8'
-            )
+                # 创建RotatingFileHandler
+                handler = RotatingFileHandler(
+                    filename=str(file_path),
+                    maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES,
+                    backupCount=log_settings.LOG_BACKUP_COUNT,
+                    encoding='utf-8'
+                )
 
-            # 设置格式化器
-            formatter = logging.Formatter(log_settings.LOG_FILE_FORMAT)
-            handler.setFormatter(formatter)
+                # 设置格式化器
+                formatter = logging.Formatter(log_settings.LOG_FILE_FORMAT)
+                handler.setFormatter(formatter)
 
-            self._rotating_handlers[file_path] = handler
+                self._rotating_handlers[file_path] = handler
 
-        return self._rotating_handlers[file_path]
+            return self._rotating_handlers[file_path]
 
-    def write_log(self, level: str, message: str, file_path: Path):
+    def write_log(self, level: str, message: str, file_path: Path) -> None:
         """
         写入日志 - 自动检测协程环境并使用合适的方式
         """
@@ -181,8 +185,11 @@ class NonBlockingFileHandler:
         if self._is_in_event_loop():
             # 在协程环境中，使用非阻塞方式
             self._write_non_blocking(entry)
-        else:
-            # 不在协程环境中，直接同步写入
+            return
+        with self._state_lock:
+            if not self._running:
+                return
+            # 不在协程环境中，持锁同步写入，避免关闭文件处理器时仍有写操作进行
             self._write_sync(entry)
 
     @staticmethod
@@ -196,15 +203,19 @@ class NonBlockingFileHandler:
         except RuntimeError:
             return False
 
-    def _write_non_blocking(self, entry: LogEntry):
+    def _write_non_blocking(self, entry: LogEntry) -> bool:
         """
         非阻塞写入（用于协程环境）
         """
-        try:
-            self._write_queue.put_nowait(entry)
-        except queue.Full:
-            # 队列满时，使用线程池处理
-            self._executor.submit(self._write_sync, entry)
+        with self._state_lock:
+            if not self._running:
+                return False
+            try:
+                self._write_queue.put_nowait(entry)
+            except queue.Full:
+                # 队列满时，使用线程池处理
+                self._executor.submit(self._write_sync, entry)
+            return True
 
     @staticmethod
     def _write_sync(entry: LogEntry):
@@ -215,8 +226,7 @@ class NonBlockingFileHandler:
             # 获取RotatingFileHandler实例
             handler = NonBlockingFileHandler()._get_rotating_handler(entry.file_path)
 
-            # 使用RotatingFileHandler的emit方法，只传递原始消息
-            handler.emit(logging.LogRecord(
+            handler.handle(logging.LogRecord(
                 name='',
                 level=getattr(logging, entry.level.upper(), logging.INFO),
                 pathname='',
@@ -235,22 +245,28 @@ class NonBlockingFileHandler:
         """
         后台批量写入线程
         """
-        while self._running:
+        while True:
             try:
                 # 收集一批日志条目
                 batch = []
+                should_stop = False
                 end_time = time.time() + log_settings.WRITE_TIMEOUT
 
                 while len(batch) < log_settings.BATCH_WRITE_SIZE and time.time() < end_time:
                     try:
                         remaining_time = max(0, end_time - time.time())
                         entry = self._write_queue.get(timeout=remaining_time)
+                        if entry is self._stop_sentinel:
+                            should_stop = True
+                            break
                         batch.append(entry)
                     except queue.Empty:
                         break
 
                 if batch:
                     self._write_batch(batch)
+                if should_stop:
+                    break
 
             except Exception as e:
                 print(f"批量写入线程错误: {e}")
@@ -275,8 +291,7 @@ class NonBlockingFileHandler:
 
                 # 批量写入
                 for entry in entries:
-                    # 使用RotatingFileHandler的emit方法，只传递原始消息
-                    handler.emit(logging.LogRecord(
+                    handler.handle(logging.LogRecord(
                         name='',
                         level=getattr(logging, entry.level.upper(), logging.INFO),
                         pathname='',
@@ -294,15 +309,23 @@ class NonBlockingFileHandler:
 
     def shutdown(self):
         """
-        关闭文件处理器
+        排空异步日志并关闭文件处理器
         """
-        self._running = False
-        if hasattr(self, '_write_thread'):
-            self._write_thread.join(timeout=5)
+        with self._state_lock:
+            if not self._running:
+                return
+            self._running = False
+            if hasattr(self, '_write_thread') and self._write_thread.is_alive():
+                # 状态锁保证停止标记之后不会再有生产者入队
+                self._write_queue.put(self._stop_sentinel)
+        if hasattr(self, '_write_thread') and self._write_thread.is_alive():
+            self._write_thread.join()
         if self._executor:
             self._executor.shutdown(wait=True)
 
-        # 清理缓存
+        for handler in self._rotating_handlers.values():
+            handler.flush()
+            handler.close()
         self._rotating_handlers.clear()
 
 

app/modules/telegram/telegram.py

@@ -279,8 +279,6 @@ class Telegram:
     @staticmethod
     def _telegramify_item_text(item: Text) -> str:
         """将 telegramify 文本片段转换为 Telegram MarkdownV2 字符串。"""
-        if hasattr(item, "content"):
-            return item.content
         if entities_to_markdownv2:
             return entities_to_markdownv2(item.text, item.entities)
         return standardize(item.text)
@@ -290,8 +288,6 @@ class Telegram:
         """将 telegramify 文本或媒体片段转换为 Telegram MarkdownV2 caption。"""
         if isinstance(item, Text):
             return Telegram._telegramify_item_text(item)
-        if hasattr(item, "caption"):
-            return item.caption
         if entities_to_markdownv2:
             return entities_to_markdownv2(item.caption_text, item.caption_entities)
         return standardize(item.caption_text)

app/startup/lifecycle.py

@@ -20,6 +20,7 @@ from app.chain.system import SystemChain
 from app.core.config import global_vars, settings
 from app.helper.server import MoviePilotServerHelper
 from app.helper.system import SystemHelper
+from app.log import LoggerManager
 from app.startup.command_initializer import init_command, stop_command, restart_command
 from app.startup.modules_initializer import init_modules, stop_modules
 from app.startup.monitor_initializer import stop_monitor, init_monitor
@@ -97,20 +98,24 @@ async def lifespan(app: FastAPI):
             pass
         except Exception as e:
             print(str(e))
-        if not settings.MOVIEPILOT_SAFE_MODE:
-            # 备份插件
-            SystemChain().backup_plugins()
-            # 停止工作流
-            stop_workflow()
-            # 停止命令
-            stop_command()
-            # 停止监控器
-            stop_monitor()
-            # 停止定时器
-            stop_scheduler()
-            # 停止插件
-            stop_plugins()
-        # 停止模块
-        await stop_modules()
-        # 关闭共享的异步 HTTP 连接池，释放底层连接资源
-        await aclose_shared_async_transports()
+        try:
+            if not settings.MOVIEPILOT_SAFE_MODE:
+                # 备份插件
+                SystemChain().backup_plugins()
+                # 停止工作流
+                stop_workflow()
+                # 停止命令
+                stop_command()
+                # 停止监控器
+                stop_monitor()
+                # 停止定时器
+                stop_scheduler()
+                # 停止插件
+                stop_plugins()
+            # 停止模块
+            await stop_modules()
+            # 关闭共享的异步 HTTP 连接池，释放底层连接资源
+            await aclose_shared_async_transports()
+        finally:
+            # 日志最后关闭，确保其他组件的收尾信息已写入文件
+            LoggerManager.shutdown()

app/startup/modules_initializer.py

@@ -137,6 +137,8 @@ async def stop_modules():
     EventManager().stop()
     # 停止虚拟显示
     DisplayHelper().stop()
+    # 停止 DoH 服务
+    DohHelper().shutdown()
     # 停止线程池
     ThreadHelper().shutdown()
     # 停止消息服务

app/testing/network_guard.py

@@ -9,6 +9,8 @@ fixture 一并识别，autouse 自动作用于每个用例，无需逐用例改
 """
 from __future__ import annotations
 
+import ipaddress
+
 import pytest
 
 # 本地回环/通配地址放行，其余主机一律视为真实出站；getaddrinfo 的 host 可能为 str 或 bytes
@@ -20,21 +22,45 @@ def block_real_network(monkeypatch):
     """防御纵深：拦截对非本地主机的真实出站，强制测试零真实网络。
 
     补在各用例自身 mock 之上：某用例万一漏 mock 外部依赖（TMDB / LLM 目录 / 下载器 /
-    媒体服务器 / 任意外链），其真实 DNS 解析会在此被拦并报错，而非静默发请求。本地回环放行
-    （sqlite 等）。asyncio 默认解析器经线程池调用 ``socket.getaddrinfo``，故拦此一处即覆盖
-    同步与异步出站。``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。
+    媒体服务器 / 任意外链），其 DNS 解析或 socket 连接会被拦截。本地回环放行（sqlite 等）。
+    所有拦截记录会在用例收尾再次断言，避免业务代码捕获网络异常后让漏 mock 的用例静默通过。
+    ``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。
     """
     import socket
 
     _real_getaddrinfo = socket.getaddrinfo
+    _real_connect = socket.socket.connect
+    attempts = []
+
+    def _is_allowed_host(host) -> bool:
+        normalized = host.decode() if isinstance(host, (bytes, bytearray)) else host
+        if normalized is None or normalized in _ALLOWED_NETWORK_HOSTS:
+            return True
+        try:
+            address = ipaddress.ip_address(str(normalized).split("%", 1)[0])
+            return address.is_loopback or address.is_unspecified
+        except ValueError:
+            return False
+
+    def _blocked(operation: str, host):
+        attempts.append((operation, host))
+        raise RuntimeError(
+            f"测试禁止真实出站网络：尝试通过 {operation} 访问 {host!r}；请 mock 对应外部依赖"
+        )
 
     def _guarded_getaddrinfo(host, *args, **kwargs):
-        normalized = host.decode() if isinstance(host, (bytes, bytearray)) else host
-        if normalized is not None and normalized not in _ALLOWED_NETWORK_HOSTS:
-            raise RuntimeError(
-                f"测试禁止真实出站网络：尝试解析 {normalized!r}；请 mock 对应外部依赖"
-            )
+        if not _is_allowed_host(host):
+            _blocked("DNS", host)
         return _real_getaddrinfo(host, *args, **kwargs)
 
+    def _guarded_connect(sock, address):
+        if isinstance(address, tuple) and address and not _is_allowed_host(address[0]):
+            _blocked("socket", address[0])
+        return _real_connect(sock, address)
+
     monkeypatch.setattr(socket, "getaddrinfo", _guarded_getaddrinfo)
+    monkeypatch.setattr(socket.socket, "connect", _guarded_connect)
     yield
+    if attempts:
+        details = ", ".join(f"{operation}:{host}" for operation, host in attempts)
+        pytest.fail(f"测试期间发生真实出站网络尝试：{details}")

app/utils/singleton.py

@@ -10,6 +10,11 @@ class Singleton(abc.ABCMeta, type):
 
     _instances: dict = {}
 
+    def get_existing_instance(cls, *args, **kwargs):
+        """按相同参数返回已创建实例，不触发初始化"""
+        key = (cls, args, frozenset(kwargs.items()))
+        return cls._instances.get(key)
+
     def __call__(cls, *args, **kwargs):
         key = (cls, args, frozenset(kwargs.items()))
         if key not in cls._instances:
@@ -31,6 +36,10 @@ class SingletonClass(abc.ABCMeta, type):
 
     _instances: dict = {}
 
+    def get_existing_instance(cls):
+        """返回已创建实例，不触发初始化"""
+        return cls._instances.get(cls)
+
     def __call__(cls, *args, **kwargs):
         if cls not in cls._instances:
             cls._instances[cls] = super().__call__(*args, **kwargs)

docker/entrypoint.sh

@@ -20,6 +20,8 @@ function WARN() {
     echo -e "${WARN} ${1}"
 }
 
+ENTRYPOINT_START_TIME="$(date +%s)"
+
 function normalize_env_value() {
     printf '%s' "${1:-}" | tr '[:upper:]' '[:lower:]'
 }
@@ -57,6 +59,42 @@ function run_package_command() {
     fi
 }
 
+function wait_backend_ready() {
+    local entrypoint_start_time="${1:-$(date +%s)}"
+    local backend_start_time="${2:-$(date +%s)}"
+    local python_pid="${3:-}"
+    local backend_port="${PORT:-3001}"
+    local web_port="${NGINX_PORT:-3000}"
+    local timeout="${MOVIEPILOT_BACKEND_READY_TIMEOUT:-300}"
+    local ready_url="http://127.0.0.1:${backend_port}/api/v1/system/global?token=moviepilot"
+    local deadline
+    if ! [[ "${timeout}" =~ ^[0-9]+$ ]] || [ "$((10#${timeout}))" -le 0 ]; then
+        WARN "→ MOVIEPILOT_BACKEND_READY_TIMEOUT=${timeout} 无效，使用默认 300 秒。"
+        timeout=300
+    else
+        timeout=$((10#${timeout}))
+    fi
+    deadline=$(( $(date +%s) + timeout ))
+
+    while [ "$(date +%s)" -lt "${deadline}" ]; do
+        if [ -n "${python_pid}" ] && ! kill -0 "${python_pid}" >/dev/null 2>&1; then
+            WARN "→ 后端服务启动完成探测已停止：后端进程已退出。"
+            return 1
+        fi
+
+        if curl -fsS --max-time 2 "${ready_url}" >/dev/null 2>&1; then
+            local now
+            now="$(date +%s)"
+            INFO "→ MoviePilot Web 已可访问，启动总耗时 $(( now - entrypoint_start_time )) 秒，后端就绪耗时 $(( now - backend_start_time )) 秒，后端端口 ${backend_port}，前端端口 ${web_port}。"
+            return 0
+        fi
+        sleep 1
+    done
+
+    WARN "→ 后端服务启动完成探测超时，已等待 ${timeout} 秒，后端端口 ${backend_port}，继续等待进程日志..."
+    return 1
+}
+
 # 环境变量补全
 # 优先级: 系统环境变量 -> .env 文件 (即使为空字符串) -> 预设默认值
 # 精准适配 Python 端 set_key (quote_mode="always", 单引号包裹, \' 转义)
@@ -480,12 +518,14 @@ umask "${UMASK}"
 
 # 启动后端服务
 INFO "→ 启动后端服务..."
+BACKEND_START_TIME="$(date +%s)"
 if [ "${START_NOGOSU:-false}" = "true" ]; then
     "${VENV_PATH}/bin/python3" app/main.py > /dev/stdout 2> /dev/stderr &
 else
     gosu moviepilot:moviepilot "${VENV_PATH}/bin/python3" app/main.py > /dev/stdout 2> /dev/stderr &
 fi
 PYTHON_PID=$!
+wait_backend_ready "${ENTRYPOINT_START_TIME}" "${BACKEND_START_TIME}" "${PYTHON_PID}" &
 
 # 等待 Python 进程退出。
 # 如果收到信号，trap 会中断 wait，并执行 graceful_exit。

docs/pr-agent.md

@@ -1,86 +1,50 @@
 # PR-Agent 使用说明
 
-本仓库通过 GitHub Actions 运行开源 PR-Agent，用于自动维护 PR 摘要、发布行内代码审查建议，并在每轮审查后发布一条简短的 Code Review 总结评论。
+本仓库通过 GitHub Actions 运行 PR-Agent，帮助贡献者维护 PR 摘要、获取代码审查结果和提出 PR 相关问题。
 
-## 触发方式
+## 自动执行
 
-`.github/workflows/pr-agent.yml` 监听：
+同仓分支和来自 fork 的 PR 都会自动执行 PR-Agent。
 
-- `pull_request_target`：PR 打开、重新打开、标记 ready、请求 review、推送新 commit 时自动运行。
-- `issue_comment`：允许身份在 PR 评论里写允许的命令时手动运行。
+PR 在以下场景会自动处理：
 
-PR 事件会自动执行受控审查，包含同仓 PR 和 fork PR。允许身份也可以在 PR 评论中使用允许的命令触发受控审查。
-允许身份包括 `OWNER`、`MEMBER`、`COLLABORATOR`、`CONTRIBUTOR` 和 `FIRST_TIME_CONTRIBUTOR`。
+- 打开或重新打开 PR。
+- 将草稿 PR 标记为可审查。
+- 请求审查。
+- 每次推送新的 commit。
 
-## Workflow 权限
+PR 带有 `skip pr-agent` 标签，或标题以 `[Auto]`、`Auto` 开头时，自动和手工路径都会跳过。
 
-workflow 设置了最小可用权限：
+## 手工命令
 
-- `contents: read`：读取仓库内容和 PR diff。
-- `pull-requests: write`：更新 PR 描述、发布 PR Review 或修改 PR 相关元数据。
-- `issues: write`：PR 评论在 GitHub API 中属于 issue comments，手动命令和总结评论需要该权限。
-
-没有开启 `contents: write`。当前配置不让 PR-Agent 往仓库推代码或提交 changelog，因此不需要内容写权限。
-
-## 自动行为
-
-PR 事件默认自动执行：
-
-- `/describe`：更新 PR Body 中的 `PR-Agent 摘要` / `PR-Agent Summary` 标记区域，保留用户原始描述。
-- `/improve`：发布 GitHub 行内代码审查建议，不发布 PR-Agent 建议表格。
-
-workflow 会在 `/improve` 后发布一条普通 PR 评论：
-
-- 评论标题为 `## Code Review`。
-- 如果本轮有新增行内建议，会基于这些建议生成自然语言总结。
-- 如果本轮没有新增行内建议，直接发布无更多反馈的简短总结。
-- 下一次运行前会删除上一条 PR-Agent Code Review 总结评论，避免评论堆叠，同时保留新的通知事件。
-
-## 常用评论命令
-
-以下身份可在 PR 评论中使用：
-
-- `OWNER`：仓库所有者。
-- `MEMBER`：组织仓库中的组织成员。
-- `COLLABORATOR`：仓库协作者。
-- `CONTRIBUTOR`：曾经向仓库提交并合入过代码的贡献者。
-- `FIRST_TIME_CONTRIBUTOR`：首次向仓库贡献 PR 的用户。
+在 PR 的普通讨论评论中使用以下命令：
 
 ```text
 /describe
-/improve
+/review
 /ask 这次改动有没有遗漏权限校验？
 ```
 
-评论触发依赖 `issue_comment` 事件。普通 issue 评论、Bot 评论、非允许身份评论、以及不以允许命令开头的评论都会跳过。
+- `/describe`：更新 PR Body 内按语言显示的 `PR-Agent 摘要` 或 `PR-Agent Summary`，并保留贡献者原有的 PR 描述。
+- `/review`：发起一次代码审查。
+- `/ask ...`：就当前 PR 提问，回复会发布在普通 PR 评论中。
 
-## 输出约定
+手工命令仅允许以下 GitHub 身份关联的用户使用：`OWNER`、`MEMBER`、`COLLABORATOR`、`CONTRIBUTOR`、`FIRST_TIME_CONTRIBUTOR`。
 
-PR-Agent 配置集中在 `.github/workflows/pr-agent.yml` 中维护。公开说明只描述用户可见行为：
+新建的合法命令评论会触发执行；编辑后仍为合法命令的评论也会触发。编辑普通讨论评论不会调用模型。
 
-- 根据 PR 标题和用户原始描述自动选择中文或英文；无法识别时默认中文。
-- 保留用户原始 PR 描述，只更新 PR Body 中的 PR-Agent 标记区域。
-- 不使用 PR-Agent 的 Reviewer Guide 输出。
-- 不输出 PR Type、额外标签、图表或 describe 评论。
-- 只发布值得维护者处理的问题型行内建议。
-- 行内建议可使用 GitHub suggestion 形式，便于直接采纳。
-- 没有建议时不发布 PR-Agent 建议表格，只保留简短的 Code Review 总结评论。
+## 审查结果
 
-行内建议可使用风险前缀：
+`/describe` 的结果位于 PR Body 中按语言显示的 `PR-Agent 摘要` 或 `PR-Agent Summary` 区域，用于概览本次变更。
 
-- `🔴 **High Risk**:`：高风险问题。
-- `🟡 **Medium Risk**:`：中风险问题。
-- `🔵 **Low Risk**:`：低风险问题。
+`/review` 和自动审查会通过原生 GitHub Review 发布，结果位于 Review 页签，标题固定为 `PR-Agent Code Review`。审查摘要包含可点击的 `文件:行号` 链接；可定位到本次变更的具体问题会在对应代码行以行内评论呈现，无法行内定位的问题仍通过摘要中的链接呈现。
 
-可按需再启用的工具配置：
+审查不会额外创建专用的摘要评论。未发现需要处理的问题时，Review 会显示：
 
-- `[pr_update_changelog]`：配合 `/update_changelog` 生成 changelog 建议。
-- `[pr_add_docs]`：配合 `/add_docs` 生成文档建议。
-- `[pr_test]`：配合 `/test` 生成测试建议；它不会替代仓库自己的测试命令。
-- `[pr_questions]`：配合 `/ask ...` 回答 PR 相关问题。
+> 本次变更无需提出审查意见，暂无其他反馈。
 
 ## 安全边界
 
-PR-Agent 依赖的 Docker 镜像在 workflow 中固定版本号和 digest，不使用浮动的 `latest` 或仅依赖可变 tag。
+workflow 使用固定 digest 的 PR-Agent 容器镜像，不使用浮动标签。自动审查通过 `pull_request_target` 在目标仓库上下文中读取 PR 信息，但不会 checkout 或执行 PR 分支代码。
 
-当前使用 `pull_request_target` 支持 PR 自动审查，但 workflow 不 checkout 或执行 PR 分支代码，只运行固定 digest 的 PR-Agent 容器并通过 GitHub API 读取 PR diff。`issue_comment` 属于 base repo 事件，因此评论命令只允许指定身份触发。
+权限保持最小化：只授予读取仓库内容所需的 `contents: read`，以及更新 PR Body、发布 Review 和回复 PR 评论所需的写权限；不会向仓库推送代码或创建提交。

pytest.ini

@@ -7,9 +7,5 @@ timeout_method = thread
 # 让本仓自身的新告警更醒目。本仓代码引发的告警一律不在此忽略，应在源码/用例处修复。
 filterwarnings =
     ignore:datetime.datetime.utcfromtimestamp\(\) is deprecated:DeprecationWarning
-    ignore:websockets.legacy is deprecated:DeprecationWarning
-    ignore:websockets.InvalidStatusCode is deprecated:DeprecationWarning
-    ignore:pkg_resources is deprecated as an API:DeprecationWarning
-    ignore:Deprecated call to .pkg_resources.declare_namespace:DeprecationWarning
     ignore:'crypt' is deprecated:DeprecationWarning
     ignore:'audioop' is deprecated:DeprecationWarning

tests/conftest.py

@@ -16,9 +16,11 @@ prepare_backend()
 from app.testing.network_guard import block_real_network  # noqa: E402,F401
 
 
-def _report_session_cleanup_error(name: str, err: Exception) -> None:
-    """测试收尾清理失败只记录诊断，不覆盖原始 pytest 退出状态。"""
+def _report_session_cleanup_error(session, name: str, err: Exception) -> None:
+    """记录收尾错误；原测试绿色时将会话标记为失败。"""
     sys.stderr.write(f"\npytest session cleanup failed: {name}: {err!r}\n")
+    if session.exitstatus == 0:
+        session.exitstatus = 1
 
 
 def pytest_sessionfinish(session, exitstatus):
@@ -28,21 +30,27 @@ def pytest_sessionfinish(session, exitstatus):
 
         shutdown_blocking_executors(cancel_futures=True)
     except Exception as err:
-        _report_session_cleanup_error("agent blocking executors", err)
+        _report_session_cleanup_error(session, "agent blocking executors", err)
 
     try:
         from app.helper.thread import ThreadHelper
-        from app.utils.singleton import Singleton
 
-        helper = Singleton._instances.get((ThreadHelper, (), frozenset()))
+        helper = ThreadHelper.get_existing_instance()
         if helper:
             helper.shutdown()
     except Exception as err:
-        _report_session_cleanup_error("thread helper", err)
+        _report_session_cleanup_error(session, "thread helper", err)
+
+    try:
+        from app.helper.message import stop_message
+
+        stop_message()
+    except Exception as err:
+        _report_session_cleanup_error(session, "message service", err)
 
     try:
         from app.log import LoggerManager
 
         LoggerManager.shutdown()
     except Exception as err:
-        _report_session_cleanup_error("logger manager", err)
+        _report_session_cleanup_error(session, "logger manager", err)

tests/test_db_error_diagnostics.py

@@ -0,0 +1,106 @@
+import asyncio
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import OperationalError
+
+import app.db as db_module
+
+
+class _SqliteError(Exception):
+    """模拟 sqlite3 异常暴露的扩展错误字段。"""
+
+    sqlite_errorcode = 266
+    sqlite_errorname = "SQLITE_IOERR_READ"
+
+
+class _PsycopgError(Exception):
+    """模拟 psycopg2 异常暴露的 SQLSTATE 字段。"""
+
+    pgcode = "40001"
+
+
+class _AsyncpgError(Exception):
+    """模拟 asyncpg 适配异常暴露的 SQLSTATE 字段。"""
+
+    sqlstate = "23505"
+
+
+@pytest.mark.parametrize(
+    ("error", "expected"),
+    [
+        (
+            _SqliteError("disk I/O error"),
+            {
+                "error_type": "_SqliteError",
+                "error_code": 266,
+                "error_name": "SQLITE_IOERR_READ",
+            },
+        ),
+        (
+            _PsycopgError("serialization failure"),
+            {
+                "error_type": "_PsycopgError",
+                "sqlstate": "40001",
+            },
+        ),
+        (
+            _AsyncpgError("duplicate key"),
+            {
+                "error_type": "_AsyncpgError",
+                "sqlstate": "23505",
+            },
+        ),
+    ],
+)
+def test_database_error_metadata_extracts_driver_codes(error, expected) -> None:
+    """诊断元数据应兼容 SQLite、psycopg2 与 asyncpg 的稳定错误字段。"""
+    assert db_module._database_error_metadata(error) == expected
+
+
+def test_database_error_listener_omits_statement_and_parameters(monkeypatch) -> None:
+    """数据库错误日志不得包含 SQL、参数或驱动返回的原始消息。"""
+    messages = []
+    engine = create_engine("sqlite:///:memory:")
+    monkeypatch.setattr("app.db.logger.error", messages.append)
+    db_module._register_database_error_logging(engine)
+
+    with pytest.raises(OperationalError):
+        with engine.connect() as connection:
+            connection.execute(
+                text("SELECT * FROM missing_table WHERE token = :token"),
+                {"token": "private-token"},
+            )
+
+    assert len(messages) == 1
+    assert "database=sqlite" in messages[0]
+    assert "driver=pysqlite" in messages[0]
+    assert "error_code=1" in messages[0]
+    assert "error_name=SQLITE_ERROR" in messages[0]
+    assert "missing_table" not in messages[0]
+    assert "private-token" not in messages[0]
+
+
+def test_async_database_engine_logs_driver_error_metadata(monkeypatch) -> None:
+    """异步 Engine 应通过底层 sync engine 记录驱动错误码。"""
+    messages = []
+    monkeypatch.setattr("app.db.logger.error", messages.append)
+
+    async def query_missing_table() -> None:
+        async with db_module.AsyncEngine.connect() as connection:
+            await connection.execute(text("SELECT * FROM async_missing_table"))
+
+    with pytest.raises(OperationalError):
+        asyncio.run(query_missing_table())
+
+    assert len(messages) == 1
+    assert "database=sqlite" in messages[0]
+    assert "driver=aiosqlite" in messages[0]
+    assert "error_code=1" in messages[0]
+    assert "error_name=SQLITE_ERROR" in messages[0]
+    assert "async_missing_table" not in messages[0]
+
+
+def test_database_error_metadata_ignores_unclassified_errors() -> None:
+    """没有驱动错误码时不应制造无效诊断日志。"""
+    assert db_module._database_error_metadata(RuntimeError("plain failure")) is None

tests/test_docker_entrypoint_permissions.py

@@ -78,6 +78,26 @@ def _run_permission_case(tmp_path: Path, body: str, env: dict[str, str] | None =
     return chown_log.read_text(encoding="utf-8") if chown_log.exists() else ""
 
 
+def _run_entrypoint_case(tmp_path: Path, body: str, env: dict[str, str] | None = None) -> str:
+    functions = _write_entrypoint_functions(tmp_path)
+    case_env = {
+        **os.environ,
+        "ENTRYPOINT_FUNCTIONS": str(functions),
+    }
+    if env:
+        case_env.update(env)
+
+    script = textwrap.dedent(
+        f"""\
+        set -euo pipefail
+        source "${{ENTRYPOINT_FUNCTIONS}}"
+        {body}
+        """
+    )
+    result = subprocess.run(["bash", "-c", script], check=True, env=case_env, text=True, capture_output=True)
+    return result.stdout
+
+
 def test_image_paths_are_not_chowned_by_default_regardless_of_owner(tmp_path: Path) -> None:
     log = _run_permission_case(
         tmp_path,
@@ -170,3 +190,56 @@ def test_runtime_writable_paths_are_still_corrected(tmp_path: Path) -> None:
     assert not any(line.startswith("-R ") and ".cloakbrowser" in line for line in lines)
     assert not any(f"{tmp_path}/app " in line for line in lines)
     assert not any(f"{tmp_path}/public" in line for line in lines)
+
+
+def test_backend_ready_log_uses_configured_ports(tmp_path: Path) -> None:
+    curl_log = tmp_path / "curl.log"
+    output = _run_entrypoint_case(
+        tmp_path,
+        """
+        INFO() { printf '[INFO] %s\\n' "$1"; }
+        curl() {
+          printf '%s\\n' "$*" > "${CURL_LOG}"
+          return 0
+        }
+        PORT=4321 NGINX_PORT=8765 wait_backend_ready 1 2 "$$"
+        """,
+        env={"CURL_LOG": str(curl_log)},
+    )
+
+    assert curl_log.read_text(encoding="utf-8") == (
+        "-fsS --max-time 2 http://127.0.0.1:4321/api/v1/system/global?token=moviepilot\n"
+    )
+    assert "MoviePilot Web 已可访问" in output
+    assert "后端就绪耗时" in output
+    assert "后端端口 4321" in output
+    assert "前端端口 8765" in output
+
+
+def test_backend_ready_timeout_falls_back_to_default_for_invalid_value(tmp_path: Path) -> None:
+    output = _run_entrypoint_case(
+        tmp_path,
+        """
+        WARN() { printf '[WARN] %s\\n' "$1"; }
+        curl() { return 1; }
+        MOVIEPILOT_BACKEND_READY_TIMEOUT=invalid wait_backend_ready 1 2 999999 || true
+        """,
+    )
+
+    assert "MOVIEPILOT_BACKEND_READY_TIMEOUT=invalid 无效，使用默认 300 秒" in output
+    assert "后端服务启动完成探测已停止：后端进程已退出" in output
+
+
+def test_backend_ready_timeout_accepts_leading_zero_decimal(tmp_path: Path) -> None:
+    output = _run_entrypoint_case(
+        tmp_path,
+        """
+        INFO() { printf '[INFO] %s\\n' "$1"; }
+        WARN() { printf '[WARN] %s\\n' "$1"; }
+        curl() { return 0; }
+        MOVIEPILOT_BACKEND_READY_TIMEOUT=08 wait_backend_ready 1 2 "$$"
+        """,
+    )
+
+    assert "MOVIEPILOT_BACKEND_READY_TIMEOUT=08 无效" not in output
+    assert "MoviePilot Web 已可访问" in output

tests/test_doh_helper.py

@@ -3,6 +3,61 @@ import socket
 from app.helper import doh
 
 
+def test_doh_executor_is_lazy_and_shutdown_restores_socket(monkeypatch):
+    """DoH 线程池按需创建，并在模块关闭时恢复系统 DNS"""
+    original_getaddrinfo = socket.getaddrinfo
+    helper = object.__new__(doh.DohHelper)
+    monkeypatch.setattr(doh.settings, "DOH_DOMAINS", "example.com")
+    monkeypatch.setattr(doh.settings, "DOH_RESOLVERS", "resolver.test")
+    monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: "203.0.113.7")
+    monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: [])
+
+    try:
+        helper.shutdown()
+        assert doh._executor is None
+
+        doh.enable_doh(True)
+        socket.getaddrinfo("example.com", None)
+        executor = doh._executor
+        assert executor is not None
+
+        helper.shutdown()
+
+        assert doh._executor is None
+        assert socket.getaddrinfo is doh._orig_getaddrinfo
+        assert getattr(executor, "_shutdown", False)
+    finally:
+        helper.shutdown()
+        socket.getaddrinfo = original_getaddrinfo
+
+
+def test_doh_config_reload_disables_and_closes_executor(monkeypatch):
+    """热更新关闭 DoH 时恢复系统 DNS 并释放已创建的线程池"""
+    original_getaddrinfo = socket.getaddrinfo
+    helper = object.__new__(doh.DohHelper)
+    monkeypatch.setattr(doh.settings, "DOH_DOMAINS", "example.com")
+    monkeypatch.setattr(doh.settings, "DOH_RESOLVERS", "resolver.test")
+    monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: "203.0.113.7")
+    monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: [])
+
+    try:
+        helper.shutdown()
+        doh.enable_doh(True)
+        socket.getaddrinfo("example.com", None)
+        executor = doh._executor
+        assert executor is not None
+        monkeypatch.setattr(doh.settings, "DOH_ENABLE", False)
+
+        helper.on_config_changed()
+
+        assert doh._executor is None
+        assert getattr(executor, "_shutdown", False)
+        assert socket.getaddrinfo is doh._orig_getaddrinfo
+    finally:
+        helper.shutdown()
+        socket.getaddrinfo = original_getaddrinfo
+
+
 def test_enable_doh_reuses_cached_host_resolution(monkeypatch):
     """
     同一 DoH 域名第二次解析应命中缓存，避免重复请求远端解析器。
@@ -33,6 +88,7 @@ def test_enable_doh_reuses_cached_host_resolution(monkeypatch):
         socket.getaddrinfo("example.com", None)
         socket.getaddrinfo("example.com", None)
     finally:
+        object.__new__(doh.DohHelper).shutdown()
         socket.getaddrinfo = original_getaddrinfo
         with doh._doh_lock:
             doh._doh_cache.clear()

tests/test_emby_dashboard_links.py

@@ -102,6 +102,7 @@ class EmbyDashboardLinksTest(unittest.TestCase):
         with (
             patch.object(client, "_Emby__get_emby_librarys") as librarys,
             patch.object(client, "_Emby__get_local_image_by_id") as image_by_id,
+            patch.object(client, "get_items_count", return_value=0),
         ):
             librarys.return_value = [
                 {

tests/test_lifecycle_shutdown.py

@@ -0,0 +1,46 @@
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi import FastAPI
+
+from app.startup import lifecycle
+
+
+def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch):
+    """前置关闭步骤失败时仍应关闭 Logger"""
+    monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False)
+    monkeypatch.setattr(lifecycle.global_vars, "set_loop", MagicMock())
+    for name in (
+        "init_routers",
+        "init_modules",
+        "init_plugins",
+        "init_scheduler",
+        "init_monitor",
+        "init_command",
+        "init_workflow",
+        "stop_workflow",
+        "stop_command",
+        "stop_monitor",
+        "stop_scheduler",
+        "stop_plugins",
+    ):
+        monkeypatch.setattr(lifecycle, name, MagicMock())
+
+    system_chain = MagicMock()
+    system_chain.backup_plugins.side_effect = RuntimeError("backup failed")
+    monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain))
+    monkeypatch.setattr(lifecycle, "init_extra", AsyncMock())
+    monkeypatch.setattr(lifecycle, "stop_modules", AsyncMock())
+    monkeypatch.setattr(lifecycle, "aclose_shared_async_transports", AsyncMock())
+    logger_shutdown = MagicMock()
+    monkeypatch.setattr(lifecycle.LoggerManager, "shutdown", logger_shutdown)
+
+    async def run_lifespan():
+        with pytest.raises(RuntimeError, match="backup failed"):
+            async with lifecycle.lifespan(FastAPI()):
+                pass
+
+    asyncio.run(run_lifespan())
+
+    logger_shutdown.assert_called_once_with()

tests/test_log_shutdown.py

@@ -0,0 +1,149 @@
+import threading
+import time
+from unittest.mock import MagicMock
+
+from app.log import LogEntry, NonBlockingFileHandler, log_settings
+
+
+def test_non_blocking_file_handler_shutdown_wakes_writer_and_closes_handlers(tmp_path):
+    """日志关闭应立即唤醒空闲写线程，并关闭所有已打开的文件处理器"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    log_handler = handler._get_rotating_handler(tmp_path / "shutdown.log")
+
+    try:
+        started_at = time.monotonic()
+        handler.shutdown()
+        elapsed = time.monotonic() - started_at
+
+        assert elapsed < 1
+        assert not handler._write_thread.is_alive()
+        assert log_handler.stream is None
+        assert handler._write_non_blocking(
+            LogEntry("info", "late-message", tmp_path / "shutdown.log")
+        ) is False
+        assert handler._write_queue.empty()
+    finally:
+        if handler._write_thread.is_alive():
+            handler._running = False
+            handler._write_thread.join(timeout=5)
+        if log_handler.stream is not None:
+            log_handler.close()
+        NonBlockingFileHandler._instance = original_instance
+
+
+def test_non_blocking_file_handler_shutdown_drains_queued_batches(monkeypatch, tmp_path):
+    """停止标记之前已进入队列的日志应跨批次全部写完"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    monkeypatch.setattr(log_settings, "BATCH_WRITE_SIZE", 2)
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    written = []
+    monkeypatch.setattr(
+        handler,
+        "_write_batch",
+        lambda batch: written.extend(entry.message for entry in batch),
+    )
+
+    try:
+        for index in range(5):
+            handler._write_non_blocking(
+                LogEntry("info", f"message-{index}", tmp_path / "drain.log")
+            )
+
+        handler.shutdown()
+
+        assert written == [f"message-{index}" for index in range(5)]
+        assert not handler._write_thread.is_alive()
+    finally:
+        if handler._write_thread.is_alive():
+            handler._running = False
+            handler._write_queue.put(handler._stop_sentinel)
+            handler._write_thread.join(timeout=5)
+        NonBlockingFileHandler._instance = original_instance
+
+
+def test_non_blocking_file_handler_creates_one_handler_for_concurrent_first_write(monkeypatch, tmp_path):
+    """同一路径首次并发写入时只创建并关闭一个文件处理器"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    first_created = threading.Event()
+    second_started = threading.Event()
+    release_first = threading.Event()
+    created_handlers = []
+    results = []
+
+    class ProbeHandler:
+        def __init__(self, **kwargs):
+            self.closed = False
+            created_handlers.append(self)
+            if len(created_handlers) == 1:
+                first_created.set()
+                release_first.wait(timeout=2)
+
+        @staticmethod
+        def setFormatter(formatter):
+            pass
+
+        @staticmethod
+        def flush():
+            pass
+
+        def close(self):
+            self.closed = True
+
+    monkeypatch.setattr("app.log.RotatingFileHandler", ProbeHandler)
+    file_path = tmp_path / "concurrent.log"
+
+    def get_handler(started=None):
+        if started:
+            started.set()
+        results.append(handler._get_rotating_handler(file_path))
+
+    first = threading.Thread(target=get_handler)
+    second = threading.Thread(target=get_handler, args=(second_started,))
+    try:
+        first.start()
+        assert first_created.wait(timeout=1)
+        second.start()
+        assert second_started.wait(timeout=1)
+        time.sleep(0.05)
+        release_first.set()
+        first.join(timeout=2)
+        second.join(timeout=2)
+
+        assert len(created_handlers) == 1
+        assert results[0] is results[1]
+
+        handler.shutdown()
+        assert created_handlers[0].closed is True
+    finally:
+        release_first.set()
+        first.join(timeout=2)
+        second.join(timeout=2)
+        handler.shutdown()
+        NonBlockingFileHandler._instance = original_instance
+
+
+def test_non_blocking_file_handler_uses_handler_lock(monkeypatch, tmp_path):
+    """日志写入通过 Handler 入口串行化 emit 与 rollover"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    log_handler = MagicMock()
+    monkeypatch.setattr(handler, "_get_rotating_handler", MagicMock(return_value=log_handler))
+
+    try:
+        handler._write_sync(LogEntry("info", "message", tmp_path / "locked.log"))
+
+        log_handler.handle.assert_called_once()
+        log_handler.emit.assert_not_called()
+    finally:
+        handler.shutdown()
+        NonBlockingFileHandler._instance = original_instance

tests/test_media_interaction.py

@@ -20,6 +20,16 @@ def clear_media_interactions():
     plugin_input_interaction_manager.clear()
 
 
+@pytest.fixture(autouse=True)
+def mock_default_media_search():
+    """未显式验证搜索结果的消息路由用例不访问真实媒体元数据服务"""
+    with patch(
+        "app.chain.media.MediaChain.search",
+        side_effect=lambda title: (_build_meta(title), []),
+    ):
+        yield
+
+
 def _build_meta(name: str) -> MetaBase:
     """构造媒体识别元数据。"""
     meta = MetaBase(name)

tests/test_message_queue_shutdown.py

@@ -0,0 +1,30 @@
+import time
+
+from app.helper.message import MessageQueueManager, TemplateHelper, stop_message
+from app.utils.singleton import SingletonClass
+
+
+def test_message_queue_stop_wakes_idle_monitor(monkeypatch):
+    """消息队列停止时应唤醒空闲监控线程，不等待完整检查周期"""
+    monkeypatch.setattr(MessageQueueManager, "init_config", lambda self: None)
+    manager = object.__new__(MessageQueueManager)
+    manager.__init__(check_interval=10)
+
+    started_at = time.monotonic()
+    manager.stop()
+    elapsed = time.monotonic() - started_at
+
+    assert elapsed < 1
+    assert not manager.thread.is_alive()
+
+
+def test_stop_message_does_not_initialize_absent_services(monkeypatch):
+    """消息服务未初始化时，关闭入口不应为了清理而创建后台资源"""
+    monkeypatch.setattr(SingletonClass, "_instances", {})
+
+    assert MessageQueueManager.get_existing_instance() is None
+    assert TemplateHelper.get_existing_instance() is None
+    stop_message()
+
+    assert MessageQueueManager not in SingletonClass._instances
+    assert TemplateHelper not in SingletonClass._instances

tests/test_network_guard.py

@@ -0,0 +1,22 @@
+import socket
+
+import pytest
+
+from app.testing.network_guard import block_real_network
+
+
+def test_network_guard_fails_when_blocked_attempt_is_swallowed(monkeypatch):
+    """业务代码即使捕获网络异常，网络守卫仍应在用例收尾报告失败"""
+    fixture = block_real_network.__wrapped__(monkeypatch)
+    next(fixture)
+
+    try:
+        try:
+            socket.getaddrinfo("external.example", 443)
+        except RuntimeError:
+            pass
+
+        with pytest.raises(pytest.fail.Exception, match="external.example"):
+            next(fixture)
+    finally:
+        monkeypatch.undo()

tests/test_singleton.py

@@ -0,0 +1,29 @@
+from app.utils.singleton import Singleton, SingletonClass
+
+
+def test_singleton_class_can_read_existing_instance_without_creating(monkeypatch):
+    """按类单例可以只读取已存在实例"""
+
+    class Example(metaclass=SingletonClass):
+        pass
+
+    monkeypatch.setattr(SingletonClass, "_instances", {})
+
+    assert Example.get_existing_instance() is None
+    instance = Example()
+    assert Example.get_existing_instance() is instance
+
+
+def test_parameterized_singleton_can_read_matching_instance_without_creating(monkeypatch):
+    """参数化单例按相同参数读取已存在实例"""
+
+    class Example(metaclass=Singleton):
+        def __init__(self, name):
+            self.name = name
+
+    monkeypatch.setattr(Singleton, "_instances", {})
+
+    assert Example.get_existing_instance("first") is None
+    instance = Example("first")
+    assert Example.get_existing_instance("first") is instance
+    assert Example.get_existing_instance("second") is None

tests/test_subscribe_endpoint.py

@@ -362,7 +362,7 @@ class SubscribeEndpointTest(TestCase):
         with patch(
             "app.api.endpoints.subscribe.Subscribe.async_list_by_username",
             new=AsyncMock(return_value=owned),
-        ):
+        ), patch("app.api.endpoints.subscribe.Scheduler") as scheduler_cls:
             response = asyncio.run(
                 search_subscribes(
                     background_tasks=background_tasks,
@@ -376,6 +376,7 @@ class SubscribeEndpointTest(TestCase):
             [task["kwargs"]["sid"] for task in background_tasks.tasks],
             [17, 18],
         )
+        self.assertEqual(scheduler_cls.return_value.start.call_count, 0)
 
     def test_subscribe_files_hides_other_user_row(self):
         """

tests/test_telegram.py

@@ -3,6 +3,7 @@
 Telegram 模块单元测试（pytest 原生）。
 """
 import json
+import warnings
 from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch
 
@@ -286,13 +287,29 @@ def test_send_msg_markdown_escaping(telegram):
     assert send_kwargs["text"].startswith("*测试标题*\n")
 
 
-def test_telegramify_new_content_fields_are_used_directly():
-    """新版telegramify对象应直接使用已渲染的MarkdownV2字段"""
-    text_item = SimpleNamespace(content="已转义\\_文本")
-    file_item = SimpleNamespace(caption="已转义\\_说明")
+def test_telegramify_current_fields_are_used_directly():
+    """telegramify 对象直接使用当前 MarkdownV2 字段"""
+    from telegramify_markdown.content import ContentTrace, File, Text
 
-    assert Telegram._telegramify_item_text(text_item) == "已转义\\_文本"
-    assert Telegram._telegramify_item_caption(file_item) == "已转义\\_说明"
+    text_item = Text(
+        text="已转义_文本",
+        entities=[],
+        content_trace=ContentTrace(source_type="test"),
+    )
+    file_item = File(
+        file_name="test.txt",
+        file_data=b"test",
+        caption_text="已转义_说明",
+        caption_entities=[],
+        content_trace=ContentTrace(source_type="test"),
+    )
+
+    with warnings.catch_warnings(record=True) as warning_records:
+        warnings.simplefilter("always")
+        assert Telegram._telegramify_item_text(text_item) == "已转义\\_文本"
+        assert Telegram._telegramify_item_caption(file_item) == "已转义\\_说明"
+
+    assert not warning_records
 
 
 def test_send_msg_with_html_parse_mode_keeps_html(telegram):

tests/test_web_agent_stream.py

@@ -183,7 +183,8 @@ def test_build_web_agent_command_items_returns_slash_commands():
 
 def test_build_web_agent_command_items_includes_sites_command():
     """WebAgent 命令建议应包含内建站点管理命令。"""
-    commands = _build_web_agent_command_items()
+    with patch("app.command.Scheduler"), patch("app.command.ThreadHelper"):
+        commands = _build_web_agent_command_items()
 
     assert any(command["command"] == "/sites" for command in commands)
 

version.py

@@ -1,2 +1,2 @@
-APP_VERSION = 'v2.14.2'
-FRONTEND_VERSION = 'v2.14.2'
+APP_VERSION = 'v2.14.3'
+FRONTEND_VERSION = 'v2.14.3'