diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py
index 0fd70b3c..98b44244 100644
--- a/app/api/endpoints/system.py
+++ b/app/api/endpoints/system.py
@@ -1339,12 +1339,7 @@ def restart_system(_: User = Depends(get_current_active_superuser)):
     """
     if not SystemHelper.can_restart():
         return schemas.Response(success=False, message="当前运行环境不支持重启操作！")
-    # 标识停止事件
-    global_vars.stop_system()
-    # 执行重启
     ret, msg = SystemHelper.restart()
-    if not ret:
-        global_vars.resume_system()
     return schemas.Response(success=ret, message=msg)
 
 
@@ -1362,11 +1357,7 @@ def upgrade_system(
     if not SystemHelper.can_restart():
         return schemas.Response(success=False, message="当前运行环境不支持升级操作！")
 
-    # 标识停止事件
-    global_vars.stop_system()
     ret, msg = SystemHelper.upgrade(mode=mode or "release")
-    if not ret:
-        global_vars.resume_system()
     return schemas.Response(success=ret, message=msg)
 
 
diff --git a/app/chain/system.py b/app/chain/system.py
index c3bc5cf6..59abaab9 100644
--- a/app/chain/system.py
+++ b/app/chain/system.py
@@ -38,8 +38,6 @@ class SystemChain(ChainBase):
         """
         重启系统
         """
-        from app.core.config import global_vars
-
         if channel and userid:
             self.post_message(Notification(
                 channel=channel,
@@ -54,8 +52,6 @@ class SystemChain(ChainBase):
             }, self._restart_file)
         # 主动备份一次插件
         self.backup_plugins()
-        # 设置停止标志，通知所有模块准备停止
-        global_vars.stop_system()
         # 重启
         SystemHelper.restart()
 
diff --git a/app/core/config.py b/app/core/config.py
index 3d7a5f6d..06f52260 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -1211,12 +1211,6 @@ class GlobalVar(object):
         """
         self.STOP_EVENT.set()
 
-    def resume_system(self):
-        """
-        恢复系统运行标记。
-        """
-        self.STOP_EVENT.clear()
-
     @property
     def is_system_stopped(self):
         """
diff --git a/app/main.py b/app/main.py
index 9cc53f29..400cc75e 100644
--- a/app/main.py
+++ b/app/main.py
@@ -30,16 +30,31 @@ elif SystemUtils.is_frozen():
     sys.stderr = open(os.devnull, 'w')
 
 from app.factory import app
-from app.core.config import settings
+from app.core.config import global_vars, settings
 from app.db.init import init_db, update_db
 
 # 设置进程名
 setproctitle.setproctitle(settings.PROJECT_NAME)
 
+
+class MoviePilotServer(uvicorn.Server):
+    """在 Uvicorn 开始优雅退出前发布应用协作停止标志"""
+
+    def handle_exit(self, sig, frame) -> None:
+        global_vars.stop_system()
+        super().handle_exit(sig, frame)
+
+
 # uvicorn服务
-Server = uvicorn.Server(Config(app, host=settings.HOST, port=settings.PORT,
-                               reload=settings.DEV, workers=multiprocessing.cpu_count() * 2 + 1,
-                               timeout_graceful_shutdown=60))
+Server = MoviePilotServer(Config(app, host=settings.HOST, port=settings.PORT,
+                                 reload=settings.DEV, workers=multiprocessing.cpu_count() * 2 + 1,
+                                 timeout_graceful_shutdown=60))
+
+
+def request_shutdown() -> None:
+    """发布协作停止标志并请求 Uvicorn 退出"""
+    global_vars.stop_system()
+    Server.should_exit = True
 
 
 def start_tray():
@@ -64,8 +79,8 @@ def start_tray():
         """
         退出程序
         """
+        request_shutdown()
         TrayIcon.stop()
-        Server.should_exit = True
 
     import pystray
 
@@ -93,10 +108,11 @@ def signal_handler(signum, frame):
     信号处理函数，用于优雅停止服务
     """
     print(f"收到信号 {signum}，开始优雅停止服务...")
-    Server.should_exit = True
+    request_shutdown()
 
 
-if __name__ == '__main__':
+def run_application() -> None:
+    """初始化进程并启动 API 服务"""
     # 注册信号处理器
     signal.signal(signal.SIGTERM, signal_handler)
     signal.signal(signal.SIGINT, signal_handler)
@@ -109,3 +125,7 @@ if __name__ == '__main__':
     update_db()
     # 启动API服务
     Server.run()
+
+
+if __name__ == '__main__':
+    run_application()
diff --git a/app/modules/thetvdb/__init__.py b/app/modules/thetvdb/__init__.py
index de4e2711..d7d9baa5 100644
--- a/app/modules/thetvdb/__init__.py
+++ b/app/modules/thetvdb/__init__.py
@@ -114,7 +114,6 @@ class TheTvDbModule(_ModuleBase):
         return 4
 
     def stop(self):
-        logger.info("TheTvDbModule 停止。正在清除 TVDB 会话。")
         with self.__auth_lock:
             self.tvdb = None
 
diff --git a/app/startup/lifecycle.py b/app/startup/lifecycle.py
index a40d1525..3d114271 100644
--- a/app/startup/lifecycle.py
+++ b/app/startup/lifecycle.py
@@ -1,5 +1,7 @@
 import asyncio
+import inspect
 from contextlib import asynccontextmanager
+from typing import Callable
 
 from fastapi import FastAPI
 
@@ -20,7 +22,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.log import logger, 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
@@ -56,6 +58,16 @@ async def init_extra():
     await MoviePilotServerHelper.async_report_usage()
 
 
+async def run_shutdown_step(name: str, callback: Callable[[], object]) -> None:
+    """隔离单个关闭阶段的异常，确保后续资源仍有机会释放"""
+    try:
+        result = callback()
+        if inspect.isawaitable(result):
+            await result
+    except Exception as err:
+        logger.error(f"关闭{name}失败：{err}")
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """
@@ -90,6 +102,7 @@ async def lifespan(app: FastAPI):
         yield
     finally:
         print("Shutting down...")
+        global_vars.stop_system()
         # 取消同步插件任务
         try:
             sync_plugins_task.cancel()
@@ -100,22 +113,19 @@ async def lifespan(app: FastAPI):
             print(str(e))
         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()
+                await run_shutdown_step(
+                    "插件备份", lambda: SystemChain().backup_plugins()
+                )
+                await run_shutdown_step("工作流", stop_workflow)
+                await run_shutdown_step("命令服务", stop_command)
+                await run_shutdown_step("监控器", stop_monitor)
+                await run_shutdown_step("定时器", stop_scheduler)
+                await run_shutdown_step("插件", stop_plugins)
+            await run_shutdown_step("模块服务", stop_modules)
+            await run_shutdown_step(
+                "共享异步 HTTP 连接池",
+                aclose_shared_async_transports,
+            )
         finally:
             # 日志最后关闭，确保其他组件的收尾信息已写入文件
             LoggerManager.shutdown()
diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py
index 840cfd3d..e8a66f54 100644
--- a/app/startup/modules_initializer.py
+++ b/app/startup/modules_initializer.py
@@ -1,4 +1,6 @@
+import inspect
 import sys
+from typing import Callable
 
 from app.helper.redis import RedisHelper, AsyncRedisHelper
 
@@ -129,29 +131,27 @@ async def stop_modules():
     """
     服务关闭
     """
-    # 停止AI智能体
-    await stop_agent()
-    # 停止模块
-    ModuleManager().stop()
-    # 停止事件消费
-    EventManager().stop()
-    # 停止虚拟显示
-    DisplayHelper().stop()
-    # 停止 DoH 服务
-    DohHelper().shutdown()
-    # 停止线程池
-    ThreadHelper().shutdown()
-    # 停止消息服务
-    stop_message()
-    # 关闭Redis缓存连接
-    RedisHelper().close()
-    await AsyncRedisHelper().close()
-    # 停止数据库连接
-    await close_database()
-    # 停止前端服务
-    stop_frontend()
-    # 清理临时文件
-    clear_temp()
+    async def run_step(name: str, callback: Callable[[], object]) -> None:
+        """单个模块资源关闭失败时继续执行后续阶段"""
+        try:
+            result = callback()
+            if inspect.isawaitable(result):
+                await result
+        except Exception as err:
+            logger.error(f"关闭{name}失败：{err}")
+
+    await run_step("AI智能体", stop_agent)
+    await run_step("模块", lambda: ModuleManager().stop())
+    await run_step("事件消费", lambda: EventManager().stop())
+    await run_step("虚拟显示", lambda: DisplayHelper().stop())
+    await run_step("DoH服务", lambda: DohHelper().shutdown())
+    await run_step("线程池", lambda: ThreadHelper().shutdown())
+    await run_step("消息服务", stop_message)
+    await run_step("Redis缓存连接", lambda: RedisHelper().close())
+    await run_step("异步Redis缓存连接", lambda: AsyncRedisHelper().close())
+    await run_step("数据库连接", close_database)
+    await run_step("前端服务", stop_frontend)
+    await run_step("临时文件", clear_temp)
 
 
 def init_modules():
diff --git a/app/utils/http.py b/app/utils/http.py
index dbf8ddf2..12662d5e 100644
--- a/app/utils/http.py
+++ b/app/utils/http.py
@@ -76,6 +76,14 @@ _REQUESTS_RETRY_IDEMPOTENT_METHODS = ("GET", "HEAD", "OPTIONS")
 _pending_eviction_tasks: set[asyncio.Task] = set()
 
 
+def _discard_pending_eviction_task(task: asyncio.Task) -> None:
+    """从跨线程共享集合移除已完成的 transport 关闭任务"""
+    with _shared_async_transports_lock:
+        _pending_eviction_tasks.discard(task)
+    if not task.cancelled() and (error := task.exception()):
+        logger.debug(f"LRU 淘汰共享 transport 时关闭失败: {error!r}")
+
+
 def _get_shared_async_transport(
     proxy: Optional[str],
     verify: Union[bool, str],
@@ -140,8 +148,9 @@ def _get_shared_async_transport(
         try:
             task = loop.create_task(evicted_transport.aclose())
             # 强引用避免 task 仅被 loop 弱持有而触发 "Task was destroyed but pending"
-            _pending_eviction_tasks.add(task)
-            task.add_done_callback(_pending_eviction_tasks.discard)
+            with _shared_async_transports_lock:
+                _pending_eviction_tasks.add(task)
+            task.add_done_callback(_discard_pending_eviction_task)
         except Exception as e:  # pragma: no cover - 防御性
             logger.debug(f"LRU 淘汰共享 transport 时调度关闭失败: {e!r}")
 
@@ -160,17 +169,27 @@ async def aclose_shared_async_transports() -> None:
     # 弹出而非 get+clear，避免外层 dict 残留空 OrderedDict 占位
     with _shared_async_transports_lock:
         per_loop = _shared_async_transports.pop(loop, None)
-    if not per_loop:
+        pending_evictions = [
+            task
+            for task in _pending_eviction_tasks
+            if task.get_loop() is loop
+        ]
+    transports = list(per_loop.values()) if per_loop else []
+    if per_loop:
+        per_loop.clear()
+    if not transports and not pending_evictions:
         return
-    transports = list(per_loop.values())
-    per_loop.clear()
     # 并行关闭：每个 transport 的 TLS close_notify 各占一个 RTT，
     # 顺序等待会线性放大 shutdown 耗时；return_exceptions 让单点失败
     # 不影响其他 transport 的释放
     results = await asyncio.gather(
-        *(t.aclose() for t in transports), return_exceptions=True
+        *pending_evictions,
+        *(t.aclose() for t in transports),
+        return_exceptions=True,
     )
-    for result in results:
+    with _shared_async_transports_lock:
+        _pending_eviction_tasks.difference_update(pending_evictions)
+    for result in results[len(pending_evictions):]:
         if isinstance(result, BaseException):
             logger.debug(f"关闭共享 AsyncHTTPTransport 失败: {result!r}")
 
diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py
index c984a3f5..34b9c7d4 100644
--- a/tests/test_lifecycle_shutdown.py
+++ b/tests/test_lifecycle_shutdown.py
@@ -1,16 +1,28 @@
 import asyncio
+import signal
+import threading
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from fastapi import FastAPI
 
-from app.startup import lifecycle
+from app.startup import lifecycle, modules_initializer
+from app.utils import http as http_utils
 
 
-def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch):
-    """前置关闭步骤失败时仍应关闭 Logger"""
+def _assert_completed_once(mock: MagicMock) -> None:
+    if isinstance(mock, AsyncMock):
+        mock.assert_awaited_once_with()
+    else:
+        mock.assert_called_once_with()
+
+
+def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict:
+    """隔离 lifespan 的外部依赖，并按名称注入一个关闭失败"""
     monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False)
     monkeypatch.setattr(lifecycle.global_vars, "set_loop", MagicMock())
+    monkeypatch.setattr(lifecycle.global_vars, "stop_system", MagicMock())
+
     for name in (
         "init_routers",
         "init_modules",
@@ -19,28 +31,375 @@ def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch):
         "init_monitor",
         "init_command",
         "init_workflow",
+    ):
+        monkeypatch.setattr(lifecycle, name, MagicMock())
+
+    system_chain = MagicMock()
+    monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain))
+    monkeypatch.setattr(lifecycle, "init_extra", AsyncMock())
+
+    shutdown_steps = {
+        "backup_plugins": system_chain.backup_plugins,
+        "stop_workflow": MagicMock(),
+        "stop_command": MagicMock(),
+        "stop_monitor": MagicMock(),
+        "stop_scheduler": MagicMock(),
+        "stop_plugins": MagicMock(),
+        "stop_modules": AsyncMock(),
+        "close_http": AsyncMock(),
+    }
+    for name in (
         "stop_workflow",
         "stop_command",
         "stop_monitor",
         "stop_scheduler",
         "stop_plugins",
     ):
-        monkeypatch.setattr(lifecycle, name, MagicMock())
+        monkeypatch.setattr(lifecycle, name, shutdown_steps[name])
+    monkeypatch.setattr(lifecycle, "stop_modules", shutdown_steps["stop_modules"])
+    monkeypatch.setattr(
+        lifecycle,
+        "aclose_shared_async_transports",
+        shutdown_steps["close_http"],
+    )
+
+    if failing_step:
+        shutdown_steps[failing_step].side_effect = RuntimeError(
+            f"{failing_step} failed"
+        )
 
-    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)
+    shutdown_steps["logger"] = logger_shutdown
+    return shutdown_steps
+
+
+@pytest.mark.parametrize(
+    "failing_step",
+    [
+        "backup_plugins",
+        "stop_workflow",
+        "stop_command",
+        "stop_monitor",
+        "stop_scheduler",
+        "stop_plugins",
+        "stop_modules",
+        "close_http",
+    ],
+)
+def test_lifespan_continues_after_each_shutdown_owner_failure(
+    monkeypatch,
+    failing_step,
+):
+    """任一关闭阶段失败都不能跳过后续资源所有者"""
+    shutdown_steps = _patch_lifespan(monkeypatch, failing_step=failing_step)
 
     async def run_lifespan():
-        with pytest.raises(RuntimeError, match="backup failed"):
-            async with lifecycle.lifespan(FastAPI()):
-                pass
+        async with lifecycle.lifespan(FastAPI()):
+            pass
 
     asyncio.run(run_lifespan())
 
-    logger_shutdown.assert_called_once_with()
+    lifecycle.global_vars.stop_system.assert_called_once_with()
+    for step in shutdown_steps.values():
+        _assert_completed_once(step)
+
+
+def test_uvicorn_signal_publishes_stop_before_server_exit(monkeypatch):
+    """Uvicorn 接管系统信号时必须先发布协作停止标志"""
+    from app import main
+
+    calls = []
+    monkeypatch.setattr(main.global_vars, "stop_system", lambda: calls.append("stop"))
+    monkeypatch.setattr(
+        main.uvicorn.Server,
+        "handle_exit",
+        lambda _self, _sig, _frame: calls.append("uvicorn"),
+    )
+
+    server = object.__new__(main.MoviePilotServer)
+    server.handle_exit(signal.SIGTERM, None)
+
+    assert calls == ["stop", "uvicorn"]
+
+
+def test_application_preserves_stop_requested_before_startup(monkeypatch):
+    """启动流程不能清除初始化前已经发布的退出请求"""
+    from app import main
+
+    stop_event = threading.Event()
+    stop_event.set()
+    monkeypatch.setattr(main.global_vars, "STOP_EVENT", stop_event)
+    calls = []
+    monkeypatch.setattr(
+        main.signal,
+        "signal",
+        lambda *_args: calls.append("signal"),
+    )
+    monkeypatch.setattr(main, "start_tray", lambda: calls.append("tray"))
+    monkeypatch.setattr(main, "init_db", lambda: calls.append("init_db"))
+    monkeypatch.setattr(main, "update_db", lambda: calls.append("update_db"))
+    monkeypatch.setattr(main.Server, "run", lambda: calls.append("server"))
+
+    main.run_application()
+
+    assert stop_event.is_set()
+    assert calls == [
+        "signal",
+        "signal",
+        "tray",
+        "init_db",
+        "update_db",
+        "server",
+    ]
+
+
+def test_uvicorn_preserves_stop_requested_before_serve(monkeypatch):
+    """Uvicorn 启动不能清除数据库初始化阶段已经发布的停止请求"""
+    from app import main
+
+    stop_event = threading.Event()
+    monkeypatch.setattr(main.global_vars, "STOP_EVENT", stop_event)
+    main.global_vars.stop_system()
+
+    async def serve(_self, sockets=None):
+        assert main.global_vars.is_system_stopped
+
+    monkeypatch.setattr(main.uvicorn.Server, "serve", serve)
+    server = object.__new__(main.MoviePilotServer)
+    asyncio.run(server.serve())
+
+
+@pytest.mark.parametrize("endpoint_name", ["restart_system", "upgrade_system"])
+@pytest.mark.parametrize(
+    "initially_stopped",
+    [False, True],
+    ids=["running", "stopping"],
+)
+def test_restart_endpoint_failure_preserves_stop_state(
+    monkeypatch,
+    endpoint_name,
+    initially_stopped,
+):
+    """重启或升级失败不能发布或撤销停止请求"""
+    from app.api.endpoints import system
+
+    stop_event = threading.Event()
+    if initially_stopped:
+        stop_event.set()
+    monkeypatch.setattr(system.global_vars, "STOP_EVENT", stop_event)
+    monkeypatch.setattr(system.SystemHelper, "can_restart", MagicMock(return_value=True))
+    monkeypatch.setattr(
+        system.SystemHelper,
+        "restart" if endpoint_name == "restart_system" else "upgrade",
+        MagicMock(return_value=(False, "restart failed")),
+    )
+
+    if endpoint_name == "restart_system":
+        response = system.restart_system(None)
+    else:
+        response = system.upgrade_system(None, None)
+
+    assert not response.success
+    assert stop_event.is_set() is initially_stopped
+
+
+def test_command_restart_failure_does_not_publish_stop_request(monkeypatch):
+    """命令重启失败时进程仍在运行，不能提前发布停止请求"""
+    from app.chain.system import SystemChain
+    from app.core.config import global_vars
+
+    stop_event = threading.Event()
+    monkeypatch.setattr(global_vars, "STOP_EVENT", stop_event)
+    monkeypatch.setattr(SystemChain, "backup_plugins", MagicMock())
+    restart = MagicMock(return_value=(False, "restart failed"))
+    monkeypatch.setattr("app.chain.system.SystemHelper.restart", restart)
+
+    chain = object.__new__(SystemChain)
+    chain.restart(channel=None, userid=None)
+
+    restart.assert_called_once_with()
+    assert not stop_event.is_set()
+
+
+def test_stop_modules_continues_after_internal_owner_failures(monkeypatch):
+    """模块关闭编排中的多个失败不能阻断其余清理"""
+    stop_agent = AsyncMock(side_effect=RuntimeError("agent failed"))
+    monkeypatch.setattr(modules_initializer, "stop_agent", stop_agent)
+    dependencies = _patch_module_shutdown_dependencies(monkeypatch)
+    dependencies["module"].side_effect = RuntimeError("module failed")
+
+    asyncio.run(modules_initializer.stop_modules())
+
+    stop_agent.assert_awaited_once_with()
+    for dependency in dependencies.values():
+        _assert_completed_once(dependency)
+
+
+def _patch_module_shutdown_dependencies(monkeypatch) -> dict:
+    """替换 stop_modules 的资源所有者，避免测试启动真实后台服务"""
+    dependencies = {}
+    for name, method_name in (
+        ("ModuleManager", "stop"),
+        ("EventManager", "stop"),
+        ("DisplayHelper", "stop"),
+        ("DohHelper", "shutdown"),
+        ("ThreadHelper", "shutdown"),
+        ("RedisHelper", "close"),
+    ):
+        instance = MagicMock()
+        setattr(instance, method_name, MagicMock())
+        monkeypatch.setattr(
+            modules_initializer,
+            name,
+            MagicMock(return_value=instance),
+        )
+        key = name.removesuffix("Helper").removesuffix("Manager").lower()
+        dependencies[key] = getattr(instance, method_name)
+
+    for name in ("stop_message", "stop_frontend", "clear_temp"):
+        dependency = MagicMock()
+        monkeypatch.setattr(modules_initializer, name, dependency)
+        dependencies[name] = dependency
+
+    async_redis = MagicMock()
+    async_redis.close = AsyncMock()
+    monkeypatch.setattr(
+        modules_initializer,
+        "AsyncRedisHelper",
+        MagicMock(return_value=async_redis),
+    )
+    dependencies["async_redis"] = async_redis.close
+    close_database = AsyncMock()
+    monkeypatch.setattr(modules_initializer, "close_database", close_database)
+    dependencies["close_database"] = close_database
+    return dependencies
+
+
+def test_shared_http_close_waits_for_real_lru_eviction(monkeypatch):
+    """最终 HTTP 关闭必须等待真实 LRU 淘汰任务并消费其异常"""
+
+    class FakeTransport:
+        created = []
+
+        def __init__(self, **_kwargs):
+            self.close_started = asyncio.Event()
+            self.release_close = asyncio.Event()
+            self.closed = False
+            self.fail_on_close = not self.created
+            if not self.fail_on_close:
+                self.release_close.set()
+            self.created.append(self)
+
+        async def aclose(self):
+            self.close_started.set()
+            await self.release_close.wait()
+            self.closed = True
+            if self.fail_on_close:
+                raise RuntimeError("eviction close failed")
+
+    monkeypatch.setattr(http_utils, "_MAX_SHARED_TRANSPORTS_PER_LOOP", 1)
+    monkeypatch.setattr(http_utils.httpx, "AsyncHTTPTransport", FakeTransport)
+    debug = MagicMock()
+    monkeypatch.setattr(http_utils.logger, "debug", debug)
+
+    async def run_test():
+        transport_kwargs = {
+            "proxy": None,
+            "verify": True,
+            "http2": False,
+            "max_keepalive_connections": 1,
+            "max_connections": 1,
+        }
+        evicted_transport = http_utils._get_shared_async_transport(
+            **transport_kwargs,
+            keepalive_expiry=1,
+        )
+        active_transport = http_utils._get_shared_async_transport(
+            **transport_kwargs,
+            keepalive_expiry=2,
+        )
+        await asyncio.wait_for(evicted_transport.close_started.wait(), timeout=1)
+
+        loop = asyncio.get_running_loop()
+        with http_utils._shared_async_transports_lock:
+            eviction_tasks = [
+                task
+                for task in http_utils._pending_eviction_tasks
+                if task.get_loop() is loop
+            ]
+        assert len(eviction_tasks) == 1
+
+        close_task = asyncio.create_task(http_utils.aclose_shared_async_transports())
+        await asyncio.sleep(0)
+        try:
+            assert not close_task.done()
+            evicted_transport.release_close.set()
+            await close_task
+            await asyncio.sleep(0)
+            assert eviction_tasks[0].done()
+            assert evicted_transport.closed
+            assert active_transport.closed
+            with http_utils._shared_async_transports_lock:
+                assert not any(
+                    task.get_loop() is loop
+                    for task in http_utils._pending_eviction_tasks
+                )
+        finally:
+            evicted_transport.release_close.set()
+            active_transport.release_close.set()
+            await asyncio.gather(close_task, return_exceptions=True)
+            await http_utils.aclose_shared_async_transports()
+
+    asyncio.run(run_test())
+
+    debug.assert_any_call(
+        "LRU 淘汰共享 transport 时关闭失败: "
+        "RuntimeError('eviction close failed')"
+    )
+
+
+def test_shared_http_close_ignores_eviction_from_other_loop():
+    """当前事件循环关闭不能等待其他循环持有的淘汰任务"""
+    ready = threading.Event()
+    release = threading.Event()
+    failures = []
+    state = {}
+
+    def run_foreign_loop():
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        async def delayed_close():
+            while not release.is_set():
+                await asyncio.sleep(0.01)
+
+        task = loop.create_task(delayed_close())
+        state["task"] = task
+        with http_utils._shared_async_transports_lock:
+            http_utils._pending_eviction_tasks.add(task)
+        task.add_done_callback(http_utils._discard_pending_eviction_task)
+        ready.set()
+        try:
+            loop.run_until_complete(task)
+            loop.run_until_complete(asyncio.sleep(0))
+        except BaseException as err:
+            failures.append(err)
+        finally:
+            with http_utils._shared_async_transports_lock:
+                http_utils._pending_eviction_tasks.discard(task)
+            loop.close()
+
+    thread = threading.Thread(target=run_foreign_loop)
+    thread.start()
+    try:
+        assert ready.wait(timeout=2)
+        asyncio.run(http_utils.aclose_shared_async_transports())
+        assert thread.is_alive()
+        assert not state["task"].done()
+    finally:
+        release.set()
+        thread.join(timeout=2)
+
+    assert not thread.is_alive()
+    assert not failures