test: isolate lifecycle shutdown contracts

tests/test_agent_lazy_initializer.py

@@ -319,15 +319,22 @@ async def test_stop_retains_running_blocking_tool_until_retry(monkeypatch) -> No
     started = threading.Event()
     release = threading.Event()
 
+    assert reopen_blocking_executors() is True
+
     def _blocking_call() -> str:
         started.set()
         release.wait()
         return "done"
 
+    async def _wait_until_started() -> None:
+        """等待阻塞工具线程启动，避免把永久等待任务留在默认线程池。"""
+        while not started.is_set():
+            await asyncio.sleep(0)
+
     worker = asyncio.create_task(
         MoviePilotTool.run_blocking("web", _blocking_call)
     )
-    assert await asyncio.wait_for(asyncio.to_thread(started.wait), timeout=1)
+    await asyncio.wait_for(_wait_until_started(), timeout=1)
     monkeypatch.setattr(
         agent_initializer,
         "begin_agent_shutdown",

tests/test_agent_lifecycle.py

@@ -15,6 +15,7 @@ from app.agent.orchestrator import (
     AgentManagerUnavailableError,
 )
 from app.agent.memory import MemoryManager
+from app.agent.tools.base import reopen_blocking_executors
 from app.startup.initializers import agent as agent_initializer
 from app.startup.initializers import modules as modules_initializer
 
@@ -144,11 +145,14 @@ async def test_agent_entrypoint_reuses_tasks_and_closes_idempotently(
     assert manager._idle_cleanup_task is idle_cleanup_task
     assert memory_manager.cleanup_task is memory_cleanup_task
 
-    await agent_initializer.stop_agent()
-    await agent_initializer.stop_agent()
-    assert initializer._initialized is False
-    assert manager._idle_cleanup_task is None
-    assert memory_manager.cleanup_task is None
+    try:
+        await agent_initializer.stop_agent()
+        await agent_initializer.stop_agent()
+        assert initializer._initialized is False
+        assert manager._idle_cleanup_task is None
+        assert memory_manager.cleanup_task is None
+    finally:
+        assert reopen_blocking_executors() is True
 
 
 @pytest.mark.anyio

tests/test_api_background_task_registry.py

@@ -189,7 +189,7 @@ def test_seerr_subscribe_uses_task_registry(monkeypatch) -> None:
 
 
 def test_manual_subscription_search_uses_task_registry() -> None:
-    """手工订阅搜索命令应以稳定 owner 提交历史兼容的调度参数。"""
+    """手工订阅搜索命令应以稳定 owner 提交顺序搜索批次。"""
     registry = _TaskRegistry()
     repository = object()
     runtime = SimpleNamespace(
@@ -207,15 +207,10 @@ def test_manual_subscription_search_uses_task_registry() -> None:
 
     function, args, kwargs, owner = registry.calls[0]
     assert found is True
-    assert function is subscription_dependencies.start_scheduler_job
-    assert args == ()
-    assert kwargs == {
-        "job_id": "subscribe_search",
-        "sid": None,
-        "state": "R",
-        "manual": True,
-    }
-    assert owner == "api.subscription.search_schedule"
+    assert function is subscription_dependencies._start_subscription_search_batch
+    assert args == (None, "R")
+    assert kwargs == {}
+    assert owner == "api.subscribe.search"
 
 
 def test_history_ai_redo_uses_task_registry() -> None: