fix(runtime): prevent implicit scheduler startup (#6337)

app/scheduler.py

@@ -300,6 +300,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
     }
 
     def __init__(self):
+        """创建调度器状态；后台任务由应用生命周期显式启动。"""
         # 定时服务
         self._scheduler = None
         # 退出事件
@@ -314,10 +315,6 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         self._auth_count = 0
         # 用户认证失败消息发送
         self._auth_message = False
-        # 对账上个进程未收口的 Agent 任务
-        self._reconcile_agent_task_interruptions()
-        # 初始化
-        self.init()
 
     def on_config_changed(self) -> None:
         """
@@ -409,6 +406,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         if settings.DEV:
             return
 
+        # 对账上个进程未收口的 Agent 任务；进程内重复初始化不会重复改写状态。
+        self._reconcile_agent_task_interruptions()
+
         with lock:
             # 各服务的运行状态
             mediaserver_chain = MediaServerChain()

app/startup/scheduler_initializer.py

@@ -9,7 +9,7 @@ def init_scheduler():
     """
     初始化定时器
     """
-    Scheduler()
+    Scheduler().init()
 
 
 def stop_scheduler():

pytest.ini

@@ -11,3 +11,4 @@ filterwarnings =
     ignore:datetime.datetime.utcfromtimestamp\(\) is deprecated:DeprecationWarning
     ignore:'crypt' is deprecated:DeprecationWarning
     ignore:'audioop' is deprecated:DeprecationWarning
+    ignore:There is no current event loop:DeprecationWarning:lark_oapi\.ws\.client

tests/test_agent_task_runs.py

@@ -313,15 +313,10 @@ async def test_query_task_returns_owner_scoped_ten_recent_runs(monkeypatch) -> N
     assert other_run
     assert oper.finish_run(other_run.run_id, success=True, result="其他用户")
 
-    class _SchedulerStub:
-        """为查询工具提供下一次触发时间，避免启动真实 APScheduler。"""
-
-        @staticmethod
-        def get_agent_task_next_run(_task_id):
-            """返回测试任务不需要的下一次触发时间。"""
-            return None
-
-    monkeypatch.setattr("app.scheduler.Scheduler", _SchedulerStub)
+    monkeypatch.setattr(
+        "app.application.scheduling.get_agent_task_next_run",
+        lambda _task_id: None,
+    )
     detail = json.loads(await _build_query_tool(task.user_id).run(task_id=task.id))
     assert detail["total"] == 1
     assert [run["run_id"] for run in detail["tasks"][0]["recent_runs"]] == expected[:10]

tests/test_lifecycle_shutdown.py

@@ -28,6 +28,7 @@ def _patch_lifespan(monkeypatch, *, failing_step: str | None = None) -> dict:
         "init_plugins",
         "init_scheduler",
         "init_monitor",
+        "replay_pending_transfers",
         "init_command",
         "init_workflow",
     ):

tests/test_media_response_models.py

@@ -1,6 +1,6 @@
 import pytest
 from fastapi import FastAPI
-from fastapi.testclient import TestClient
+from httpx import ASGITransport, AsyncClient
 
 from app import schemas
 from app.api.endpoints import mediaserver as mediaserver_endpoint
@@ -9,7 +9,8 @@ from app.domain.context import MediaInfo as CoreMediaInfo
 from app.schemas.types import MediaSource, MediaType
 
 
-def test_media_search_response_preserves_core_collection_fields() -> None:
+@pytest.mark.asyncio
+async def test_media_search_response_preserves_core_collection_fields() -> None:
     """媒体搜索响应模型应保留 Core MediaInfo 合集输出的全部兼容字段。"""
     media = CoreMediaInfo(tmdb_info={
         "id": 42,
@@ -36,7 +37,11 @@ def test_media_search_response_preserves_core_collection_fields() -> None:
 
     app = FastAPI()
     app.include_router(router)
-    response = TestClient(app).get("/media/search")
+    async with AsyncClient(
+        transport=ASGITransport(app=app),
+        base_url="http://testserver",
+    ) as client:
+        response = await client.get("/media/search")
 
     assert response.status_code == 200
     result = response.json()["data"][0]
@@ -52,7 +57,8 @@ def test_media_search_response_preserves_core_collection_fields() -> None:
     assert "anilist_info" in result
 
 
-def test_media_response_accepts_cross_source_credit_shapes() -> None:
+@pytest.mark.asyncio
+async def test_media_response_accepts_cross_source_credit_shapes() -> None:
     """媒体响应应同时保留豆瓣姓名字符串和 TMDB 演职员对象。"""
     router = ResponseAPIRouter()
 
@@ -71,7 +77,11 @@ def test_media_response_accepts_cross_source_credit_shapes() -> None:
 
     app = FastAPI()
     app.include_router(router)
-    response = TestClient(app).get("/recommend")
+    async with AsyncClient(
+        transport=ASGITransport(app=app),
+        base_url="http://testserver",
+    ) as client:
+        response = await client.get("/recommend")
 
     assert response.status_code == 200
     media = response.json()["data"][0]
@@ -83,7 +93,8 @@ def test_media_response_accepts_cross_source_credit_shapes() -> None:
     assert media["directors"][1]["name"] == "导演乙"
 
 
-def test_media_response_accepts_legacy_source_key() -> None:
+@pytest.mark.asyncio
+async def test_media_response_accepts_legacy_source_key() -> None:
     """媒体身份重构前缓存的旧格式条目（source + media_id）应被归一化并正常响应。"""
     router = ResponseAPIRouter()
 
@@ -103,7 +114,11 @@ def test_media_response_accepts_legacy_source_key() -> None:
 
     app = FastAPI()
     app.include_router(router)
-    response = TestClient(app).get("/recommend")
+    async with AsyncClient(
+        transport=ASGITransport(app=app),
+        base_url="http://testserver",
+    ) as client:
+        response = await client.get("/recommend")
 
     assert response.status_code == 200
     media = response.json()["data"][0]

tests/test_music_subscribe.py

@@ -470,7 +470,7 @@ def test_album_subscription_uses_persisted_snapshot_when_remote_detail_is_unavai
     media_chain = Mock()
     media_chain.recognize_media.return_value = None
 
-    with patch("app.chain.subscribe.MediaChain", return_value=media_chain):
+    with patch("app.chain._music.MediaChain", return_value=media_chain):
         restored = SubscribeChain._recognize_music_subscribe(subscribe)
 
     assert restored.music_type == MUSIC_ENTITY_ALBUM
@@ -485,7 +485,7 @@ def test_legacy_music_identity_failure_does_not_guess_entity_from_title():
     media_chain = Mock()
     media_chain.recognize_media.return_value = None
 
-    with patch("app.chain.subscribe.MediaChain", return_value=media_chain):
+    with patch("app.chain._music.MediaChain", return_value=media_chain):
         restored = SubscribeChain._recognize_music_subscribe(subscribe)
 
     assert restored is None
@@ -501,7 +501,7 @@ def test_album_subscription_without_remote_id_uses_persisted_entity_snapshot():
         total_tracks=11,
     )
 
-    with patch("app.chain.subscribe.MediaChain") as media_chain:
+    with patch("app.chain._music.MediaChain") as media_chain:
         restored = SubscribeChain._recognize_music_subscribe(subscribe)
 
     assert restored.music_type == MUSIC_ENTITY_ALBUM
@@ -543,7 +543,7 @@ def test_legacy_music_subscription_rejects_artist_recognition_result():
     media_chain = Mock()
     media_chain.recognize_media.return_value = artist
 
-    with patch("app.chain.subscribe.MediaChain", return_value=media_chain):
+    with patch("app.chain._music.MediaChain", return_value=media_chain):
         restored = SubscribeChain._recognize_music_subscribe(subscribe)
 
     assert restored is None
@@ -569,7 +569,7 @@ def test_album_subscription_preserves_track_count_snapshot_when_remote_omits_it(
     media_chain = Mock()
     media_chain.recognize_media.return_value = remote
 
-    with patch("app.chain.subscribe.MediaChain", return_value=media_chain):
+    with patch("app.chain._music.MediaChain", return_value=media_chain):
         restored = SubscribeChain._recognize_music_subscribe(subscribe)
 
     assert restored is not remote

tests/test_scheduler_cache_expiry.py

@@ -3,6 +3,7 @@ from unittest.mock import Mock
 
 from app import scheduler as scheduler_module
 from app.scheduler import Scheduler
+from app.startup import scheduler_initializer
 
 
 class _BackgroundSchedulerStub:
@@ -22,6 +23,28 @@ class _BackgroundSchedulerStub:
         self.started = True
 
 
+def test_scheduler_constructor_does_not_start_background_jobs(monkeypatch):
+    """取得调度器单例不应绕过应用生命周期启动后台任务。"""
+    init = Mock()
+    monkeypatch.setattr(Scheduler, "init", init)
+
+    scheduler = object.__new__(Scheduler)
+    scheduler.__init__()
+
+    init.assert_not_called()
+    assert scheduler._scheduler is None
+
+
+def test_scheduler_initializer_starts_background_jobs(monkeypatch):
+    """应用启动入口负责显式启动已经构造的调度器。"""
+    scheduler = Mock()
+    monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler))
+
+    scheduler_initializer.init_scheduler()
+
+    scheduler.init.assert_called_once_with()
+
+
 def test_meta_cache_expire_does_not_schedule_bulk_cache_clear(monkeypatch):
     """单条缓存 TTL 不应再被用于注册整批缓存清理任务。"""
     background_scheduler = _BackgroundSchedulerStub()
@@ -68,6 +91,7 @@ def test_meta_cache_expire_does_not_schedule_bulk_cache_clear(monkeypatch):
     scheduler._event = threading.Event()
     scheduler._lock = threading.RLock()
     scheduler._jobs = {}
+    scheduler._agent_task_interruptions_reconciled = True
     scheduler._auth_count = 0
     scheduler._auth_message = False