refactor: dataize scheduler job contracts

app/application/scheduling.py

@@ -9,7 +9,10 @@ Scheduler 实现由 startup 组合根在导入期注册，避免 application 层
     agent.tools / api.endpoints -> application.scheduling <- startup（注册 Scheduler 类）
 """
 
-from typing import Any, List, Optional
+import asyncio
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any, Awaitable, Callable, List, Optional
 
 # Agent 自主定时任务在运行时调度器中的任务 ID 前缀。
 AGENT_TASK_JOB_PREFIX = "agent-task"
@@ -18,6 +21,103 @@ AGENT_TASK_JOB_PREFIX = "agent-task"
 _scheduler_class: Any = None
 
 
+class JobOverlapPolicy(StrEnum):
+    """描述同一 job 已运行时的新触发处理策略。"""
+
+    SKIP = "skip"
+
+
+class JobRecoveryPolicy(StrEnum):
+    """描述进程重启后是否以及如何重建执行意图。"""
+
+    NEXT_SCHEDULE = "next_schedule"
+    DURABLE_QUEUE = "durable_queue"
+    MANUAL_ONLY = "manual_only"
+
+
+@dataclass(frozen=True, slots=True)
+class JobSpec:
+    """数据化声明一个业务 job 的执行和恢复合同。"""
+
+    job_id: str
+    name: str
+    func: Callable[..., Any]
+    owner: str
+    overlap: JobOverlapPolicy = JobOverlapPolicy.SKIP
+    timeout_seconds: int | None = None
+    manual: bool = False
+    recovery: JobRecoveryPolicy = JobRecoveryPolicy.NEXT_SCHEDULE
+    kwargs: dict[str, Any] = field(default_factory=dict)
+
+    def to_runtime_state(self) -> dict[str, Any]:
+        """生成兼容 Scheduler Facade 的可变执行状态。"""
+        return {
+            "name": self.name,
+            "func": self.func,
+            "owner": self.owner,
+            "overlap": self.overlap.value,
+            "timeout_seconds": self.timeout_seconds,
+            "manual": self.manual,
+            "recovery": self.recovery.value,
+            "kwargs": dict(self.kwargs),
+            "running": False,
+        }
+
+
+class JobCatalog:
+    """保存唯一 job ID 到声明的映射。"""
+
+    def __init__(self, specs: list[JobSpec]) -> None:
+        """拒绝重复 ID，并冻结供 Scheduler 初始化的声明集合。"""
+        self._specs = {spec.job_id: spec for spec in specs}
+        if len(self._specs) != len(specs):
+            raise ValueError("JobSpec job_id 不得重复")
+
+    def runtime_states(self) -> dict[str, dict[str, Any]]:
+        """返回兼容旧 Scheduler `_jobs` 的全新状态字典。"""
+        return {
+            job_id: spec.to_runtime_state()
+            for job_id, spec in self._specs.items()
+        }
+
+
+class JobExecutionState:
+    """集中维护兼容 job 字典的 overlap 与终态字段。"""
+
+    @staticmethod
+    def begin(job: dict[str, Any], started_at: str) -> bool:
+        """按 overlap policy 尝试进入 running，已运行时返回 False。"""
+        if job.get("running") and job.get(
+            "overlap", JobOverlapPolicy.SKIP.value
+        ) == JobOverlapPolicy.SKIP.value:
+            return False
+        job.update(
+            running=True,
+            last_started_at=started_at,
+            last_finished_at=None,
+            last_error=None,
+        )
+        return True
+
+    @staticmethod
+    def finish(job: dict[str, Any], finished_at: str, error: str | None) -> None:
+        """写入成功或失败终态并释放 running 标记。"""
+        job.update(
+            running=False,
+            last_finished_at=finished_at,
+            last_error=error,
+        )
+
+    @staticmethod
+    async def await_result(
+        awaitable: Awaitable[Any], timeout_seconds: int | None
+    ) -> Any:
+        """等待协程任务；声明了超时时由 asyncio 负责取消底层任务。"""
+        if timeout_seconds is None:
+            return await awaitable
+        return await asyncio.wait_for(awaitable, timeout=timeout_seconds)
+
+
 def register_scheduler_class(scheduler_class: Any) -> None:
     """注册 Scheduler 类（组合根在导入期调用）。"""
     global _scheduler_class

app/scheduler.py

@@ -49,7 +49,13 @@ from app.runtime.scheduling import TimerUtils
 lock = threading.Lock()
 SCHEDULER_PROGRESS_PREFIX = "scheduler"
 # Agent 自主定时任务前缀下沉到 application 门面，此处保留兼容导出。
-from app.application.scheduling import AGENT_TASK_JOB_PREFIX  # noqa: E402
+from app.application.scheduling import (  # noqa: E402
+    AGENT_TASK_JOB_PREFIX,
+    JobCatalog,
+    JobExecutionState,
+    JobRecoveryPolicy,
+    JobSpec,
+)
 
 
 class SchedulerChain(ChainBase):
@@ -206,11 +212,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             return
 
         job_id = "database_backup"
-        self._jobs[job_id] = {
-            "name": "数据库备份",
-            "func": self.database_backup,
-            "running": False,
-        }
+        self._jobs[job_id] = JobSpec(
+            job_id,
+            "数据库备份",
+            self.database_backup,
+            "database",
+            recovery=JobRecoveryPolicy.DURABLE_QUEUE,
+        ).to_runtime_state()
         self._scheduler.add_job(
             self.start,
             trigger=TimerUtils.build_schedule_trigger(
@@ -242,112 +250,28 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         with lock:
             # 各服务的运行状态
             mediaserver_chain = MediaServerChain()
-            self._jobs = {
-                "cookiecloud": {
-                    "name": "同步CookieCloud站点",
-                    "func": SiteChain().sync_cookies,
-                    "running": False,
-                },
-                "mediaserver_sync": {
-                    "name": "同步媒体服务器",
-                    "func": mediaserver_chain.sync,
-                    "running": False,
-                },
-                "subscribe_tmdb": {
-                    "name": "订阅元数据更新",
-                    "func": SubscribeChain().check,
-                    "running": False,
-                },
-                "subscribe_search": {
-                    "name": "订阅搜索补全",
-                    "func": SubscribeChain().search,
-                    "running": False,
-                    "kwargs": {"state": "R"},
-                },
-                "new_subscribe_search": {
-                    "name": "新增订阅搜索",
-                    "func": SubscribeChain().search,
-                    "running": False,
-                    "kwargs": {"state": "N"},
-                },
-                "subscribe_refresh": {
-                    "name": "订阅刷新",
-                    "func": SubscribeChain().refresh,
-                    "running": False,
-                },
-                "subscribe_follow": {
-                    "name": "关注的订阅分享",
-                    "func": SubscribeChain().follow,
-                    "running": False,
-                },
-                "transfer": {
-                    "name": "下载文件整理",
-                    "func": TransferChain().process,
-                    "running": False,
-                },
-                "clear_cache": {
-                    "name": "缓存清理",
-                    "func": self.clear_cache,
-                    "running": False,
-                    "manual": True,
-                },
-                "data_cleanup": {
-                    "name": "数据表清理",
-                    "func": SchedulerChain().cleanup,
-                    "running": False,
-                },
-                "user_auth": {
-                    "name": "用户认证检查",
-                    "func": self.user_auth,
-                    "running": False,
-                },
-                "scheduler_job": {
-                    "name": "公共定时服务",
-                    "func": SchedulerChain().scheduler_job,
-                    "running": False,
-                },
-                "random_wallpager": {
-                    "name": "壁纸缓存",
-                    "func": WallpaperHelper().get_wallpapers,
-                    "running": False,
-                },
-                "sitedata_refresh": {
-                    "name": "站点数据刷新",
-                    "func": SiteChain().refresh_userdatas,
-                    "running": False,
-                },
-                "recommend_refresh": {
-                    "name": "推荐缓存",
-                    "func": RecommendChain().refresh_recommend,
-                    "running": False,
-                },
-                "plugin_market_refresh": {
-                    "name": "插件市场缓存",
-                    "func": PluginManager().async_get_online_plugins,
-                    "running": False,
-                    "kwargs": {"force": True},
-                },
-                "subscribe_calendar_cache": {
-                    "name": "订阅日历缓存",
-                    "func": SubscribeChain().cache_calendar,
-                    "running": False,
-                },
-                "full_gc": {
-                    "name": "主动内存回收",
-                    "func": self.full_gc,
-                    "running": False,
-                },
-                "agent_heartbeat": {
-                    "name": "智能体定时任务",
-                    "func": self.agent_heartbeat,
-                    "running": False,
-                },
-                "usage_report": {
-                    "name": "安装版本统计上报",
-                    "func": MoviePilotServerHelper.report_usage,
-                    "running": False,
-                },
-            }
+            self._jobs = JobCatalog([
+                JobSpec("cookiecloud", "同步CookieCloud站点", SiteChain().sync_cookies, "site"),
+                JobSpec("mediaserver_sync", "同步媒体服务器", mediaserver_chain.sync, "mediaserver"),
+                JobSpec("subscribe_tmdb", "订阅元数据更新", SubscribeChain().check, "subscription"),
+                JobSpec("subscribe_search", "订阅搜索补全", SubscribeChain().search, "subscription", kwargs={"state": "R"}),
+                JobSpec("new_subscribe_search", "新增订阅搜索", SubscribeChain().search, "subscription", kwargs={"state": "N"}),
+                JobSpec("subscribe_refresh", "订阅刷新", SubscribeChain().refresh, "subscription"),
+                JobSpec("subscribe_follow", "关注的订阅分享", SubscribeChain().follow, "subscription"),
+                JobSpec("transfer", "下载文件整理", TransferChain().process, "transfer", recovery=JobRecoveryPolicy.DURABLE_QUEUE),
+                JobSpec("clear_cache", "缓存清理", self.clear_cache, "runtime", manual=True, recovery=JobRecoveryPolicy.MANUAL_ONLY),
+                JobSpec("data_cleanup", "数据表清理", SchedulerChain().cleanup, "database"),
+                JobSpec("user_auth", "用户认证检查", self.user_auth, "security"),
+                JobSpec("scheduler_job", "公共定时服务", SchedulerChain().scheduler_job, "module"),
+                JobSpec("random_wallpager", "壁纸缓存", WallpaperHelper().get_wallpapers, "image"),
+                JobSpec("sitedata_refresh", "站点数据刷新", SiteChain().refresh_userdatas, "site"),
+                JobSpec("recommend_refresh", "推荐缓存", RecommendChain().refresh_recommend, "recommend"),
+                JobSpec("plugin_market_refresh", "插件市场缓存", PluginManager().async_get_online_plugins, "plugin", kwargs={"force": True}),
+                JobSpec("subscribe_calendar_cache", "订阅日历缓存", SubscribeChain().cache_calendar, "subscription"),
+                JobSpec("full_gc", "主动内存回收", self.full_gc, "runtime"),
+                JobSpec("agent_heartbeat", "智能体定时任务", self.agent_heartbeat, "agent"),
+                JobSpec("usage_report", "安装版本统计上报", MoviePilotServerHelper.report_usage, "server"),
+            ]).runtime_states()
 
             self._scheduler = BackgroundScheduler(
                 timezone=settings.TZ,
@@ -355,11 +279,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             )
 
             self._register_database_backup_job()
-            self._jobs["outbox_dispatch"] = {
-                "name": "恢复待投递副作用",
-                "func": dispatch_pending_outbox,
-                "running": False,
-            }
+            self._jobs["outbox_dispatch"] = JobSpec(
+                "outbox_dispatch",
+                "恢复待投递副作用",
+                dispatch_pending_outbox,
+                "outbox",
+                recovery=JobRecoveryPolicy.DURABLE_QUEUE,
+            ).to_runtime_state()
             self._scheduler.add_job(
                 self.start,
                 "interval",
@@ -393,12 +319,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             )
             for mediaserver_schedule in mediaserver_schedules:
                 job_id = mediaserver_schedule["id"]
-                self._jobs[job_id] = {
-                    "name": mediaserver_schedule["name"],
-                    "func": mediaserver_chain.sync,
-                    "running": False,
-                    "kwargs": {"server": mediaserver_schedule["server"]},
-                }
+                self._jobs[job_id] = JobSpec(
+                    job_id,
+                    mediaserver_schedule["name"],
+                    mediaserver_chain.sync,
+                    "mediaserver",
+                    kwargs={"server": mediaserver_schedule["server"]},
+                ).to_runtime_state()
                 self._scheduler.add_job(
                     self.start,
                     "interval",
@@ -632,13 +559,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             job = self._jobs.get(job_id)
             if not job:
                 return None
-            if job.get("running"):
+            if not JobExecutionState.begin(job, started_at):
                 logger.warning(f"定时任务 {job_id} - {job.get('name')} 正在运行 ...")
                 return None
-            self._jobs[job_id]["running"] = True
-            self._jobs[job_id]["last_started_at"] = started_at
-            self._jobs[job_id]["last_finished_at"] = None
-            self._jobs[job_id]["last_error"] = None
         progress = ProgressHelper(self._get_progress_key(job_id))
         progress.start()
         progress.update(
@@ -671,9 +594,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         with self._lock:
             job = self._jobs.get(job_id)
             if job:
-                job["running"] = False
-                job["last_finished_at"] = finished_at
-                job["last_error"] = error
+                JobExecutionState.finish(job, finished_at, error)
         job_name = job.get("name") if job else job_id
         # 收尾可能发生在事件循环上（__run_coro_job），使用异步进度后端避免阻塞
         progress = AsyncProgressHelper(self._get_progress_key(job_id))
@@ -862,9 +783,16 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         success = True
         error = None
         try:
-            result = await coro
+            result = await JobExecutionState.await_result(
+                coro,
+                timeout_seconds=job.get("timeout_seconds"),
+            )
             error = self.__get_result_error(result)
             success = error is None
+        except asyncio.TimeoutError as err:
+            success = False
+            error = f"任务执行超时（{job.get('timeout_seconds')} 秒）"
+            self.__handle_job_error(job_id=job_id, job=job, error=err)
         except asyncio.CancelledError:
             success = False
             error = "任务已取消"
@@ -1050,13 +978,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
 
         job_id = self._get_agent_task_job_id(task_id)
         with self._lock:
-            self._jobs[job_id] = {
-                "name": task.name,
-                "provider_name": "[Agent]",
-                "func": self.execute_agent_task,
-                "running": False,
-                "kwargs": {"task_id": task_id},
-            }
+            self._jobs[job_id] = JobSpec(
+                job_id,
+                task.name,
+                self.execute_agent_task,
+                "agent",
+                recovery=JobRecoveryPolicy.NEXT_SCHEDULE,
+                kwargs={"task_id": task_id},
+            ).to_runtime_state()
+            self._jobs[job_id]["provider_name"] = "[Agent]"
             # 已开始的一次任务在重启后结果未知，只保留显式执行入口，不能按
             # 过期触发时间自动重放可能已经发生的外部副作用。
             if manual_only:
@@ -1272,12 +1202,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         with self._lock:
             try:
                 job_id = f"workflow-{workflow.id}"
-                self._jobs[job_id] = {
-                    "func": WorkflowChain().process,
-                    "name": workflow.name,
-                    "provider_name": "工作流",
-                    "running": False,
-                }
+                self._jobs[job_id] = JobSpec(
+                    job_id,
+                    workflow.name,
+                    WorkflowChain().process,
+                    "workflow",
+                ).to_runtime_state()
+                self._jobs[job_id]["provider_name"] = "工作流"
                 self._scheduler.add_job(
                     self.start,
                     trigger=CronTrigger.from_crontab(workflow.timer),
@@ -1321,14 +1252,17 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                     sid = f"{pid}_{service['id']}"
                     job_id = sid.split("|")[0]
                     self.remove_plugin_job(pid, job_id)
-                    self._jobs[job_id] = {
-                        "func": service["func"],
-                        "name": service["name"],
-                        "pid": pid,
-                        "provider_name": plugin_name,
-                        "kwargs": service.get("func_kwargs") or {},
-                        "running": False,
-                    }
+                    self._jobs[job_id] = JobSpec(
+                        job_id,
+                        service["name"],
+                        service["func"],
+                        f"plugin:{pid}",
+                        kwargs=service.get("func_kwargs") or {},
+                    ).to_runtime_state()
+                    self._jobs[job_id].update(
+                        pid=pid,
+                        provider_name=plugin_name,
+                    )
                     self._scheduler.add_job(
                         self.start,
                         service["trigger"],

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

@@ -705,6 +705,18 @@ app/scheduler.py # APScheduler 兼容 Facade
 
 **步骤**：先把 job 定义数据化，再提执行状态；不在第一步替换 APScheduler。每个 job 必须声明 overlap policy、timeout、manual、recovery 和 owner。
 
+**实施记录（2026-08-21）**：
+
+- `app.application.scheduling` 新增 `JobSpec`、`JobCatalog`、`JobExecutionState` 以及 overlap/recovery 枚举；
+  系统、媒体服务器、Agent、工作流、插件和 outbox 动态任务均由同一合同生成兼容运行状态。
+- 保留 APScheduler 和既有 `Scheduler` Facade；重入判断、开始/结束/失败状态统一由 execution state 收敛，
+  job 状态稳定暴露 owner、overlap、timeout、manual、recovery 五项策略。
+- coroutine job 的非空 timeout 使用 `asyncio.wait_for`，超时会取消底层任务并记录明确终态；同步 job 默认
+  `timeout=None`，避免用线程强杀制造不可控的半完成副作用。一次性 Agent 任务重启后保持 manual-only，durable
+  outbox/备份/整理与 next-schedule 任务的恢复语义可审计。
+- 61 个 Scheduler、Agent 定时任务、备份、进度和媒体服务器专项测试通过，覆盖重复 ID、overlap skip、
+  timeout cancel、restart/manual recovery 与兼容状态字段。
+
 ### 阶段 6：可观测性、类型和复杂度预算
 
 #### ARCH-260：统一 request/correlation ID

tests/test_scheduler_contracts.py

@@ -0,0 +1,88 @@
+"""Scheduler 声明、catalog 与 execution state 测试。"""
+
+import asyncio
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.application.scheduling import (
+    JobCatalog,
+    JobExecutionState,
+    JobOverlapPolicy,
+    JobRecoveryPolicy,
+    JobSpec,
+)
+
+
+def test_job_spec_exports_complete_legacy_state() -> None:
+    """每个声明必须显式暴露 owner、overlap、timeout、manual 和 recovery。"""
+    state = JobSpec(
+        "outbox",
+        "恢复副作用",
+        MagicMock(),
+        "outbox",
+        timeout_seconds=30,
+        recovery=JobRecoveryPolicy.DURABLE_QUEUE,
+    ).to_runtime_state()
+
+    assert state["owner"] == "outbox"
+    assert state["overlap"] == JobOverlapPolicy.SKIP.value
+    assert state["timeout_seconds"] == 30
+    assert state["manual"] is False
+    assert state["recovery"] == JobRecoveryPolicy.DURABLE_QUEUE.value
+
+
+def test_job_catalog_rejects_duplicate_ids() -> None:
+    """重复 job ID 在 APScheduler 注册前即失败。"""
+    spec = JobSpec("same", "相同", MagicMock(), "test")
+
+    with pytest.raises(ValueError, match="不得重复"):
+        JobCatalog([spec, spec])
+
+
+def test_execution_state_skips_overlap_and_records_terminal_state() -> None:
+    """统一 execution state 保持旧的跳过重入并记录失败终态。"""
+    state = JobSpec("job", "任务", MagicMock(), "test").to_runtime_state()
+
+    assert JobExecutionState.begin(state, "start") is True
+    assert JobExecutionState.begin(state, "second") is False
+    JobExecutionState.finish(state, "finish", "failed")
+
+    assert state["running"] is False
+    assert state["last_started_at"] == "start"
+    assert state["last_finished_at"] == "finish"
+    assert state["last_error"] == "failed"
+
+
+@pytest.mark.asyncio
+async def test_execution_state_cancels_coroutine_after_timeout() -> None:
+    """声明的 timeout 必须取消协程并向 Facade 抛出超时。"""
+    cancelled = asyncio.Event()
+
+    async def wait_forever() -> None:
+        """模拟只能由取消信号结束的协程任务。"""
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    with pytest.raises(asyncio.TimeoutError):
+        await JobExecutionState.await_result(wait_forever(), timeout_seconds=0.01)
+
+    assert cancelled.is_set()
+
+
+def test_manual_job_declares_restart_policy() -> None:
+    """只允许人工触发的任务必须显式声明重启后不自动补跑。"""
+    state = JobSpec(
+        "manual",
+        "人工任务",
+        MagicMock(),
+        "runtime",
+        manual=True,
+        recovery=JobRecoveryPolicy.MANUAL_ONLY,
+    ).to_runtime_state()
+
+    assert state["manual"] is True
+    assert state["recovery"] == JobRecoveryPolicy.MANUAL_ONLY.value