From e1d7918297766d10be62590d47d21dc756823127 Mon Sep 17 00:00:00 2001
From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com>
Date: Sun, 23 Aug 2026 17:17:59 +0800
Subject: [PATCH] =?UTF-8?q?fix(runtime):=20=E6=94=B6=E5=8F=A3=E8=B0=83?=
 =?UTF-8?q?=E5=BA=A6=E4=B8=8E=E4=BA=8B=E4=BB=B6=E5=BC=82=E6=AD=A5=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=20(#6415)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/endpoints/site.py                     |  10 +-
 app/runtime/event/dispatch.py                 |  18 +-
 app/runtime/events.py                         | 163 +++-
 app/scheduler.py                              | 832 ++++++++++++++----
 app/startup/modules_initializer.py            |   2 +-
 app/startup/scheduler_initializer.py          |   4 +-
 .../adr/0007-background-action-reliability.md |   4 +-
 .../backend-architecture-next-stage.md        |   4 +-
 .../architecture/dependency-baseline.json     |  37 +-
 tests/test_agent_scheduled_tasks.py           |  73 ++
 tests/test_database_backup_scheduler.py       |   7 +
 tests/test_event_dispatch_snapshot.py         | 173 +++-
 tests/test_lifecycle_shutdown.py              |   2 +-
 tests/test_plugin_local_sync.py               |   5 +
 tests/test_scheduler_cache_expiry.py          |  11 +-
 tests/test_scheduler_lifecycle.py             | 724 +++++++++++++++
 tests/test_scheduler_progress.py              |  22 +-
 tests/test_site_reset_scheduler.py            |  60 ++
 18 files changed, 1937 insertions(+), 214 deletions(-)
 create mode 100644 tests/test_scheduler_lifecycle.py
 create mode 100644 tests/test_site_reset_scheduler.py

diff --git a/app/api/endpoints/site.py b/app/api/endpoints/site.py
index 59ee4819e..e1dcae9dc 100644
--- a/app/api/endpoints/site.py
+++ b/app/api/endpoints/site.py
@@ -185,6 +185,7 @@ async def cookie_cloud_sync(
 
 @router.get("/reset", summary="重置站点", response_model=_SchemaResponse[None])
 async def reset(
+    task_registry: Annotated[TaskRegistry, Depends(get_background_task_registry)],
     command: SiteMutationCommand = Depends(get_site_mutation_command),
     _: ApiPrincipal = Depends(get_current_active_superuser_async),
 ) -> Any:
@@ -194,9 +195,12 @@ async def reset(
     result = await command.reset()
     await get_configured_system_config().async_set(SystemConfigKey.IndexerSites, [])
     await get_configured_system_config().async_set(SystemConfigKey.RssSites, [])
-    # 启动定时服务
-    Scheduler().start("cookiecloud", manual=True)
-    # 插件站点删除
+    resolve_background_task_registry(task_registry).create_sync(
+        Scheduler().start,
+        job_id="cookiecloud",
+        owner="api.site.reset",
+        manual=True,
+    )
     return _SchemaResponse(success=result.success, message="站点已重置！")
 
 
diff --git a/app/runtime/event/dispatch.py b/app/runtime/event/dispatch.py
index 78f4064d9..a7adcc4c3 100644
--- a/app/runtime/event/dispatch.py
+++ b/app/runtime/event/dispatch.py
@@ -29,6 +29,7 @@ class EventDispatcher:
         event_loop: Callable[[], Any],
         event_factory: Callable[..., Any],
         error_handler: Callable[..., None],
+        async_handle_sink: Callable[[Any], bool] | None = None,
     ) -> None:
         """注入注册表、绑定器、执行器和错误策略回调。"""
         self._registry = registry
@@ -37,6 +38,7 @@ class EventDispatcher:
         self._event_loop = event_loop
         self._event_factory = event_factory
         self._error_handler = error_handler
+        self._async_handle_sink = async_handle_sink
 
     def dispatch_chain(self, event: Any) -> bool:
         """同步按优先级顺序执行链式事件快照。"""
@@ -119,10 +121,18 @@ class EventDispatcher:
                 correlation_id=event.correlation_id,
             )
             if inspect.iscoroutinefunction(handler):
-                asyncio.run_coroutine_threadsafe(
-                    self.safe_invoke_async(handler, isolated),
-                    self._event_loop(),
-                )
+                coroutine = self.safe_invoke_async(handler, isolated)
+                if self._async_handle_sink:
+                    self._async_handle_sink(coroutine)
+                    continue
+                try:
+                    asyncio.run_coroutine_threadsafe(coroutine, self._event_loop())
+                except RuntimeError:
+                    coroutine.close()
+                    logger.warning(
+                        "事件 %s 的异步处理器无法投递，事件循环已停止",
+                        event.event_type,
+                    )
             else:
                 self._executor().submit(
                     self.safe_invoke_sync,
diff --git a/app/runtime/events.py b/app/runtime/events.py
index 908e3435e..f4b387061 100644
--- a/app/runtime/events.py
+++ b/app/runtime/events.py
@@ -1,7 +1,10 @@
+import asyncio
+import concurrent.futures
 import random
 import threading
 import traceback
 import uuid
+from dataclasses import dataclass
 from queue import Empty, PriorityQueue
 from typing import Callable, Dict, List, Optional, Tuple, Union, Any, Type
 
@@ -28,6 +31,15 @@ DEFAULT_EVENT_PRIORITY = 10  # 事件的默认优先级
 MIN_EVENT_CONSUMER_THREADS = 1  # 最小事件消费者线程数
 INITIAL_EVENT_QUEUE_IDLE_TIMEOUT_SECONDS = 1  # 事件队列空闲时的初始超时时间（秒）
 MAX_EVENT_QUEUE_IDLE_TIMEOUT_SECONDS = 5  # 事件队列空闲时的最大超时时间（秒）
+_EVENT_STOP_SENTINEL = object()
+
+
+@dataclass(slots=True)
+class _EventAsyncHandle:
+    """记录异步广播的取消代理和真实完成信号。"""
+
+    handle: concurrent.futures.Future[Any]
+    completion: concurrent.futures.Future[Any]
 
 
 class Event:
@@ -109,6 +121,10 @@ class EventManager(metaclass=Singleton):
         self.__lock = threading.Lock()
         # 退出事件
         self.__event = threading.Event()
+        # 广播异步处理器的生命周期由事件总线自己持有，避免关闭后仍向主循环运行。
+        self.__lifecycle_lock = threading.RLock()
+        self.__lifecycle_state = "new"
+        self.__async_handles: Dict[int, _EventAsyncHandle] = {}
         # 由上层管理器注册的处理器实例解析器
         self.__handler_instance_resolvers: Dict[str, HandlerInstanceResolver] = {}
         # 由启动组合层注入的错误通知回调
@@ -138,6 +154,7 @@ class EventManager(metaclass=Singleton):
             event_loop=lambda: global_vars.loop,
             event_factory=Event,
             error_handler=lambda **kwargs: self.__handle_event_error(**kwargs),
+            async_handle_sink=self.__register_async_handle,
         )
 
     def register_handler_instance_resolver(
@@ -165,8 +182,16 @@ class EventManager(metaclass=Singleton):
         """
         开始广播事件处理线程
         """
+        with self.__lifecycle_lock:
+            if self.__lifecycle_state == "running":
+                return
+            if self.__lifecycle_state == "stopping":
+                logger.warning("事件处理仍在停止，忽略重复启动")
+                return
+            self.__lifecycle_state = "running"
+            self.__event.set()
+            self.__consumer_threads = []
         # 启动消费者线程用于处理广播事件
-        self.__event.set()
         for _ in range(MIN_EVENT_CONSUMER_THREADS):
             thread = threading.Thread(target=self.__broadcast_consumer_loop, daemon=True)
             thread.start()
@@ -177,14 +202,140 @@ class EventManager(metaclass=Singleton):
         停止广播事件处理线程
         """
         logger.info("正在停止事件处理...")
-        self.__event.clear()  # 停止广播事件处理
+        consumer_threads = self.__begin_stop()
         try:
-            # 通过遍历保存的线程来等待它们完成
-            for consumer_thread in self.__consumer_threads:
-                consumer_thread.join()
+            self.__join_consumer_threads(consumer_threads)
+            self.__discard_stop_sentinels()
+            self.__cancel_async_handles()
             logger.info("事件处理停止完成")
         except Exception as e:
             logger.error(f"停止事件处理线程出错：{str(e)} - {traceback.format_exc()}")
+        finally:
+            with self.__lifecycle_lock:
+                self.__lifecycle_state = "stopped"
+                self.__consumer_threads = []
+
+    async def stop_async(self) -> None:
+        """停止广播消费者并等待已投递的异步处理器收口。"""
+        logger.info("正在停止事件处理...")
+        consumer_threads = self.__begin_stop()
+        try:
+            if consumer_threads:
+                await asyncio.to_thread(
+                    self.__join_consumer_threads,
+                    consumer_threads,
+                )
+            self.__discard_stop_sentinels()
+            with self.__lifecycle_lock:
+                handles = tuple(self.__async_handles.values())
+            for handle in handles:
+                handle.handle.cancel()
+            if handles:
+                await asyncio.gather(
+                    *(
+                        asyncio.shield(asyncio.wrap_future(handle.completion))
+                        for handle in handles
+                    ),
+                    return_exceptions=True,
+                )
+            logger.info("事件处理停止完成")
+        except Exception as e:
+            logger.error(f"停止事件处理线程出错：{str(e)} - {traceback.format_exc()}")
+            raise
+        with self.__lifecycle_lock:
+            self.__lifecycle_state = "stopped"
+            self.__consumer_threads = []
+
+    def __begin_stop(self) -> tuple[threading.Thread, ...]:
+        """关闭提交入口并唤醒消费者线程。"""
+        with self.__lifecycle_lock:
+            self.__lifecycle_state = "stopping"
+            self.__event.clear()
+            consumer_threads = tuple(self.__consumer_threads)
+            if consumer_threads:
+                self.__event_queue.put((float("-inf"), _EVENT_STOP_SENTINEL))
+        return consumer_threads
+
+    @staticmethod
+    def __join_consumer_threads(consumer_threads: tuple[threading.Thread, ...]) -> None:
+        """在线程池或同步兼容入口中等待事件消费者退出。"""
+        for consumer_thread in consumer_threads:
+            consumer_thread.join()
+
+    def __discard_stop_sentinels(self) -> None:
+        """清理仅用于唤醒消费者的标记，保留尚未消费的业务事件。"""
+        pending = []
+        while True:
+            try:
+                item = self.__event_queue.get_nowait()
+            except Empty:
+                break
+            if item[1] is not _EVENT_STOP_SENTINEL:
+                pending.append(item)
+        for item in pending:
+            self.__event_queue.put(item)
+
+    def __cancel_async_handles(self) -> None:
+        """请求取消所有仍由事件总线持有的异步处理器。"""
+        with self.__lifecycle_lock:
+            handles = tuple(self.__async_handles.values())
+        for handle in handles:
+            handle.handle.cancel()
+
+    def __register_async_handle(
+            self,
+            coroutine: Any,
+    ) -> bool:
+        """在同一生命周期临界区提交并登记异步广播处理器。"""
+        with self.__lifecycle_lock:
+            if self.__lifecycle_state != "running":
+                coroutine.close()
+                return False
+            completion: concurrent.futures.Future[Any] = concurrent.futures.Future()
+            started = threading.Event()
+
+            async def _tracked() -> None:
+                started.set()
+                try:
+                    result = await coroutine
+                except asyncio.CancelledError:
+                    if not completion.done():
+                        completion.cancel()
+                except Exception as err:
+                    if not completion.done():
+                        completion.set_exception(err)
+                else:
+                    if not completion.done():
+                        completion.set_result(result)
+
+            tracked = _tracked()
+            try:
+                handle = asyncio.run_coroutine_threadsafe(tracked, global_vars.loop)
+            except RuntimeError:
+                tracked.close()
+                coroutine.close()
+                logger.warning("异步事件处理器无法投递，事件循环已停止")
+                return False
+            self.__async_handles[id(completion)] = _EventAsyncHandle(
+                handle=handle,
+                completion=completion,
+            )
+
+            def _complete_unstarted_submission(
+                    submitted: concurrent.futures.Future[Any],
+            ) -> None:
+                if submitted.cancelled() and not started.is_set():
+                    coroutine.close()
+                    completion.cancel()
+
+            handle.add_done_callback(_complete_unstarted_submission)
+        completion.add_done_callback(self.__remove_async_handle)
+        return True
+
+    def __remove_async_handle(self, handle: concurrent.futures.Future[Any]) -> None:
+        """异步处理器完成后移除其 owner 句柄。"""
+        with self.__lifecycle_lock:
+            self.__async_handles.pop(id(handle), None)
 
     def check(self, etype: Union[EventType, ChainEventType]) -> bool:
         """
@@ -429,6 +580,8 @@ class EventManager(metaclass=Singleton):
         while self.__event.is_set():
             try:
                 priority, event = self.__event_queue.get(timeout=rate_limiter.current_wait)
+                if event is _EVENT_STOP_SENTINEL:
+                    break
                 record_metric(
                     "event.queue.depth",
                     self.__event_queue.qsize(),
diff --git a/app/scheduler.py b/app/scheduler.py
index 8fd963e6d..1e2642cbf 100644
--- a/app/scheduler.py
+++ b/app/scheduler.py
@@ -1,6 +1,5 @@
 import asyncio
-from concurrent.futures import CancelledError as ConcurrentCancelledError
-from concurrent.futures import Future as ConcurrentFuture
+import concurrent.futures
 import gc
 import hashlib
 import inspect
@@ -8,6 +7,7 @@ import multiprocessing
 import threading
 import time
 import traceback
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import Callable, Optional, Dict, Any, List
 
@@ -57,6 +57,19 @@ from app.runtime.observability import record_metric
 
 lock = threading.Lock()
 SCHEDULER_PROGRESS_PREFIX = "scheduler"
+
+
+@dataclass(slots=True)
+class _SchedulerHandle:
+    """记录调度器提交到事件循环的执行句柄及其 job generation。"""
+
+    job_id: str
+    generation: int
+    loop: asyncio.AbstractEventLoop
+    handle: asyncio.Future[Any] | concurrent.futures.Future[Any]
+    completion: asyncio.Future[Any] | concurrent.futures.Future[Any]
+
+
 # Agent 自主定时任务前缀下沉到 application 门面，此处保留兼容导出。
 from app.application.scheduling import (  # noqa: E402
     AGENT_TASK_JOB_PREFIX,
@@ -125,21 +138,32 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         self._lock = threading.RLock()
         # 各服务的运行状态
         self._jobs = {}
+        # 生命周期门禁与事件循环句柄由调度器实例独立持有。
+        self._lifecycle_state = "new"
+        self._handles: dict[int, _SchedulerHandle] = {}
+        self._job_generations: dict[str, int] = {}
+        # 运行所有权独立于可热重建的任务定义，避免重载期间同 ID 任务并行执行。
+        self._active_job_generations: dict[str, set[int]] = {}
+        self._agent_task_reservations: dict[str, int] = {}
         # 进程启动时只对账一次，配置热重载不得改写仍在执行的任务状态
         self._agent_task_interruptions_reconciled = False
         # 用户认证失败次数
         self._auth_count = 0
         # 用户认证失败消息发送
         self._auth_message = False
-        # 记录由 Scheduler 提交到事件循环的协程，避免 stop() 后继续悬挂。
-        self._async_tasks: set[asyncio.Task[Any] | ConcurrentFuture[Any]] = set()
-        self._accepting_async_tasks = True
 
-    def on_config_changed(self) -> None:
+    async def on_config_changed(self) -> None:
         """
         配置变更后重新初始化定时服务。
         """
-        self.init()
+        reload_started, scheduler = self._begin_reload()
+        if not reload_started:
+            return
+        await asyncio.to_thread(self._shutdown_scheduler_sync, scheduler)
+        with self._lock:
+            if self._lifecycle_state != "reloading":
+                return
+        self.init(_already_stopped=True)
 
     def get_reload_name(self) -> str:
         """
@@ -147,6 +171,243 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         """
         return "定时服务"
 
+    def _accepting_submissions(self) -> bool:
+        """判断调度器是否仍允许提交新的运行实例。"""
+        return self._lifecycle_state in {"starting", "running"}
+
+    def _next_job_generation(self, job_id: str) -> int:
+        """为同一 job 的下一次注册分配单调 generation。"""
+        generation = self._job_generations.get(job_id, 0) + 1
+        self._job_generations[job_id] = generation
+        return generation
+
+    def _assign_job_generation(self, job_id: str, job: dict[str, Any]) -> None:
+        """把注册 generation 写入可变运行时状态。"""
+        job["_generation"] = self._next_job_generation(job_id)
+
+    def _is_job_active(self, job_id: str) -> bool:
+        """判断任一 generation 的同 ID 任务是否仍在真实执行。"""
+        return bool(self._active_job_generations.get(job_id))
+
+    def _release_job_generation(self, job_id: str, generation: int) -> None:
+        """在任务真实收尾后释放对应 generation 的运行所有权。"""
+        active_generations = self._active_job_generations.get(job_id)
+        if not active_generations:
+            return
+        active_generations.discard(generation)
+        if not active_generations:
+            self._active_job_generations.pop(job_id, None)
+
+    def _finish_unsubmitted_job(
+            self,
+            job_id: str,
+            job: dict[str, Any],
+            generation: int,
+            error: Optional[str],
+    ) -> None:
+        """收尾协程无法提交时同步释放任务状态和运行所有权。"""
+        finished_at = self._format_time()
+        metric_started_at = None
+        with self._lock:
+            if generation not in self._active_job_generations.get(job_id, set()):
+                return
+            current_job = self._jobs.get(job_id)
+            if current_job is job and current_job.get("_generation", 0) == generation:
+                JobExecutionState.finish(job, finished_at, error)
+                metric_started_at = job.pop("_metric_started_at", None)
+            self._release_job_generation(job_id, generation)
+        if metric_started_at is not None:
+            record_metric(
+                "scheduler.job.duration",
+                time.perf_counter() - metric_started_at,
+                owner=str(job.get("owner", "unknown")),
+                outcome="success" if error is None else "error",
+            )
+
+    def _remove_handle(
+            self,
+            handle: asyncio.Future[Any] | concurrent.futures.Future[Any],
+    ) -> None:
+        """执行句柄完成后从 owner registry 移除。"""
+        with self._lock:
+            self._handles.pop(id(handle), None)
+
+    def _register_handle(
+            self,
+            job_id: str,
+            generation: int,
+            loop: asyncio.AbstractEventLoop,
+            handle: asyncio.Future[Any] | concurrent.futures.Future[Any],
+            completion: asyncio.Future[Any] | concurrent.futures.Future[Any] | None = None,
+    ) -> bool:
+        """登记调度器拥有的句柄；关闭竞态下拒绝并取消新句柄。"""
+        if completion is None:
+            completion = handle
+        with self._lock:
+            if not self._accepts_handle(job_id, generation):
+                if isinstance(handle, concurrent.futures.Future):
+                    handle.cancel()
+                elif loop.is_running():
+                    loop.call_soon_threadsafe(handle.cancel)
+                else:
+                    handle.cancel()
+                return False
+            self._handles[id(completion)] = _SchedulerHandle(
+                job_id=job_id,
+                generation=generation,
+                loop=loop,
+                handle=handle,
+                completion=completion,
+            )
+        completion.add_done_callback(self._remove_handle)
+        return True
+
+    def _accepts_handle(self, job_id: str, generation: int) -> bool:
+        """判断新句柄是否属于当前运行期或热重载中的既有任务。"""
+        if self._accepting_submissions():
+            return True
+        current_job = self._jobs.get(job_id)
+        return bool(
+            self._lifecycle_state == "reloading"
+            and current_job is not None
+            and current_job.get("_generation", 0) == generation
+            and current_job.get("running")
+        )
+
+    @staticmethod
+    def _cancel_handle(handle: _SchedulerHandle) -> None:
+        """从句柄所属线程安全地请求取消。"""
+        target = handle.handle
+        if isinstance(target, concurrent.futures.Future):
+            target.cancel()
+            return
+        if target.done():
+            return
+        if target.get_loop().is_running():
+            target.get_loop().call_soon_threadsafe(target.cancel)
+        else:
+            target.cancel()
+
+    @staticmethod
+    async def _wait_handle(handle: _SchedulerHandle) -> None:
+        """等待取消请求到达协程 finally，而不是只等待提交代理变为 cancelled。"""
+        target = handle.completion
+        if isinstance(target, concurrent.futures.Future):
+            await asyncio.shield(asyncio.wrap_future(target))
+            return
+        if target.get_loop() is asyncio.get_running_loop():
+            await asyncio.shield(target)
+
+    async def _await_cancelled_handles(
+            self,
+            handles: tuple[_SchedulerHandle, ...],
+    ) -> None:
+        """等待已投递协程结束，关闭总预算由应用生命周期统一控制。"""
+        if not handles:
+            return
+        await asyncio.gather(
+            *(self._wait_handle(handle) for handle in handles),
+            return_exceptions=True,
+        )
+
+    @staticmethod
+    def _track_cross_thread_completion(
+            coro: Any,
+            completion: concurrent.futures.Future[Any],
+            started: threading.Event,
+    ) -> Any:
+        """把跨线程提交代理与协程真实终态分离。"""
+        async def _tracked() -> None:
+            started.set()
+            try:
+                result = await coro
+            except asyncio.CancelledError:
+                if not completion.done():
+                    completion.cancel()
+            except Exception as err:
+                if not completion.done():
+                    completion.set_exception(err)
+            else:
+                if not completion.done():
+                    completion.set_result(result)
+
+        return _tracked()
+
+    def _submit_cross_thread(
+            self,
+            coro: Any,
+            *,
+            target_loop: asyncio.AbstractEventLoop,
+            job_id: str,
+            generation: int,
+            on_unstarted_cancel: Optional[Callable[[], None]] = None,
+    ) -> bool:
+        """向主循环提交协程，并以独立完成信号跟踪真实收尾。"""
+        completion: concurrent.futures.Future[Any] = concurrent.futures.Future()
+        handle: concurrent.futures.Future[Any] = concurrent.futures.Future()
+        started = threading.Event()
+        tracked = self._track_cross_thread_completion(coro, completion, started)
+        task_lock = threading.Lock()
+        target_task: asyncio.Task[Any] | None = None
+
+        def complete_target_task(task: asyncio.Task[Any]) -> None:
+            if task.cancelled() and not started.is_set():
+                if on_unstarted_cancel:
+                    on_unstarted_cancel()
+                if not completion.done():
+                    completion.cancel()
+            elif not completion.done():
+                error = task.exception()
+                if error is None:
+                    completion.set_result(None)
+                else:
+                    completion.set_exception(error)
+            if not handle.done():
+                handle.set_result(None)
+
+        def start_on_target_loop() -> None:
+            nonlocal target_task
+            with task_lock:
+                if handle.cancelled():
+                    tracked.close()
+                    coro.close()
+                    if on_unstarted_cancel:
+                        on_unstarted_cancel()
+                    completion.cancel()
+                    return
+                target_task = target_loop.create_task(tracked)
+                target_task.add_done_callback(complete_target_task)
+
+        def cancel_target_task(submitted: concurrent.futures.Future[Any]) -> None:
+            if not submitted.cancelled():
+                return
+            with task_lock:
+                task = target_task
+            if task is not None and not task.done():
+                target_loop.call_soon_threadsafe(task.cancel)
+
+        with self._lock:
+            if not self._accepts_handle(job_id, generation):
+                tracked.close()
+                coro.close()
+                return False
+            try:
+                target_loop.call_soon_threadsafe(start_on_target_loop)
+            except RuntimeError:
+                tracked.close()
+                coro.close()
+                return False
+
+            registered = self._register_handle(
+                job_id=job_id,
+                generation=generation,
+                loop=target_loop,
+                handle=handle,
+                completion=completion,
+            )
+            handle.add_done_callback(cancel_target_task)
+        return registered
+
     @staticmethod
     def _get_mediaserver_sync_interval(
             mediaserver: _SchemaMediaServerConf,
@@ -227,13 +488,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             return
 
         job_id = "database_backup"
-        self._jobs[job_id] = JobSpec(
+        job = JobSpec(
             job_id,
             "数据库备份",
             self.database_backup,
             "database",
             recovery=JobRecoveryPolicy.DURABLE_QUEUE,
         ).to_runtime_state()
+        self._assign_job_generation(job_id, job)
+        self._jobs[job_id] = job
         self._scheduler.add_job(
             self.start,
             trigger=TimerUtils.build_schedule_trigger(
@@ -247,24 +510,29 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             replace_existing=True,
         )
 
-    def init(self) -> None:
+    def init(self, *, _already_stopped: bool = False) -> None:
         """
         初始化定时服务
         """
 
         config = get_scheduler_runtime_config()
         # 停止定时服务
-        self.stop()
-        self._accepting_async_tasks = True
+        if not _already_stopped:
+            self.stop()
 
         # 调试模式不启动定时服务
         if config.dev:
+            with self._lock:
+                self._lifecycle_state = "stopped"
             return
 
         # 对账上个进程未收口的 Agent 任务；进程内重复初始化不会重复改写状态。
         self._reconcile_agent_task_interruptions()
 
         with lock:
+            with self._lock:
+                self._event.clear()
+                self._lifecycle_state = "starting"
             # 各服务的运行状态
             mediaserver_chain = MediaServerChain()
             self._jobs = JobCatalog([
@@ -289,6 +557,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 JobSpec("agent_heartbeat", "智能体定时任务", self.agent_heartbeat, "agent"),
                 JobSpec("usage_report", "安装版本统计上报", MoviePilotServerHelper.report_usage, "server"),
             ]).runtime_states()
+            for job_id, job in self._jobs.items():
+                self._assign_job_generation(job_id, job)
 
             self._scheduler = BackgroundScheduler(
                 timezone=config.timezone,
@@ -296,13 +566,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             )
 
             self._register_database_backup_job(config)
-            self._jobs["outbox_dispatch"] = JobSpec(
+            outbox_job = JobSpec(
                 "outbox_dispatch",
                 "恢复待投递副作用",
                 dispatch_pending_outbox,
                 "outbox",
                 recovery=JobRecoveryPolicy.DURABLE_QUEUE,
             ).to_runtime_state()
+            self._assign_job_generation("outbox_dispatch", outbox_job)
+            self._jobs["outbox_dispatch"] = outbox_job
             self._scheduler.add_job(
                 self.start,
                 "interval",
@@ -336,13 +608,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             )
             for mediaserver_schedule in mediaserver_schedules:
                 job_id = mediaserver_schedule["id"]
-                self._jobs[job_id] = JobSpec(
+                job = JobSpec(
                     job_id,
                     mediaserver_schedule["name"],
                     mediaserver_chain.sync,
                     "mediaserver",
                     kwargs={"server": mediaserver_schedule["server"]},
                 ).to_runtime_state()
+                self._assign_job_generation(job_id, job)
+                self._jobs[job_id] = job
                 self._scheduler.add_job(
                     self.start,
                     "interval",
@@ -559,18 +833,32 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
 
             # 启动定时服务
             self._scheduler.start()
+            with self._lock:
+                self._lifecycle_state = "running"
 
     def __prepare_job(self, job_id: str) -> Optional[dict]:
         """
         准备定时任务
         """
-        if not getattr(self, "_accepting_async_tasks", True):
-            return None
         started_at = self._format_time()
         with self._lock:
+            if not self._accepting_submissions():
+                return None
+            reservation_owner = self._agent_task_reservations.get(job_id)
+            if reservation_owner is not None:
+                if reservation_owner != threading.get_ident():
+                    return None
+                self._agent_task_reservations.pop(job_id, None)
             job = self._jobs.get(job_id)
             if not job:
                 return None
+            if self._is_job_active(job_id):
+                logger.warning(f"定时任务 {job_id} - {job.get('name')} 正在运行 ...")
+                record_metric(
+                    "scheduler.job.overlap_skip",
+                    owner=str(job.get("owner", "unknown")),
+                )
+                return None
             if not JobExecutionState.begin(job, started_at):
                 logger.warning(f"定时任务 {job_id} - {job.get('name')} 正在运行 ...")
                 record_metric(
@@ -578,6 +866,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                     owner=str(job.get("owner", "unknown")),
                 )
                 return None
+            generation = job.get("_generation", 0)
+            self._active_job_generations.setdefault(job_id, set()).add(generation)
             job["_metric_started_at"] = time.perf_counter()
         progress = ProgressHelper(self._get_progress_key(job_id))
         progress.start()
@@ -586,6 +876,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             text=f"{job.get('name') or job_id} 开始执行 ...",
             data={
                 "id": job_id,
+                "_generation": job.get("_generation", 0),
                 "name": job.get("name"),
                 "provider": job.get("provider_name", "[系统]"),
                 "status": "running",
@@ -600,6 +891,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
     async def __finish_job(
             self,
             job_id: str,
+            job: dict,
+            generation: int,
             success: bool = True,
             error: Optional[str] = None,
     ) -> None:
@@ -607,37 +900,43 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         完成定时任务
         """
         finished_at = self._format_time()
-        job = None
         with self._lock:
-            job = self._jobs.get(job_id)
-            if job:
-                JobExecutionState.finish(job, finished_at, error)
-                metric_started_at = job.pop("_metric_started_at", None)
-                if metric_started_at is not None:
-                    record_metric(
-                        "scheduler.job.duration",
-                        time.perf_counter() - metric_started_at,
-                        owner=str(job.get("owner", "unknown")),
-                        outcome="success" if success else "error",
-                    )
+            current_job = self._jobs.get(job_id)
+            if current_job is not job or current_job.get("_generation", 0) != generation:
+                self._release_job_generation(job_id, generation)
+                return
+            JobExecutionState.finish(job, finished_at, error)
+            metric_started_at = job.pop("_metric_started_at", None)
+            if metric_started_at is not None:
+                record_metric(
+                    "scheduler.job.duration",
+                    time.perf_counter() - metric_started_at,
+                    owner=str(job.get("owner", "unknown")),
+                    outcome="success" if success else "error",
+                )
         job_name = job.get("name") if job else job_id
         # 收尾可能发生在事件循环上（__run_coro_job），使用异步进度后端避免阻塞
         progress = AsyncProgressHelper(self._get_progress_key(job_id))
         current_progress = await progress.get() or {}
         progress_value = 100 if success else current_progress.get("value", 0)
-        await progress.end(
-            text=f"{job_name} {'执行完成' if success else '执行失败'}",
-            data={
-                "id": job_id,
-                "name": job_name,
-                "provider": job.get("provider_name", "[系统]") if job else None,
-                "status": "success" if success else "failed",
-                "success": success,
-                "finished_at": finished_at,
-                "error": error,
-            },
-            value=progress_value,
-        )
+        try:
+            await progress.end(
+                text=f"{job_name} {'执行完成' if success else '执行失败'}",
+                data={
+                    "id": job_id,
+                    "_generation": generation,
+                    "name": job_name,
+                    "provider": job.get("provider_name", "[系统]") if job else None,
+                    "status": "success" if success else "failed",
+                    "success": success,
+                    "finished_at": finished_at,
+                    "error": error,
+                },
+                value=progress_value,
+            )
+        finally:
+            with self._lock:
+                self._release_job_generation(job_id, generation)
 
     def get_progress(self, job_id: str) -> Optional[_SchemaScheduleProgress]:
         """
@@ -649,14 +948,24 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             job = self._jobs.get(job_id)
             job_name = job.get("name") if job else job_id
             provider_name = job.get("provider_name", "[系统]") if job else None
-            running = bool(job.get("running")) if job else False
+            running = bool(
+                job and (self._is_job_active(job_id) or job.get("running"))
+            )
             last_started_at = job.get("last_started_at") if job else None
             last_finished_at = job.get("last_finished_at") if job else None
             last_error = job.get("last_error") if job else None
         detail = ProgressHelper(self._get_progress_key(job_id)).get() or {}
         if not job and not detail:
             return None
-        data = detail.get("data") or {}
+        data = dict(detail.get("data") or {})
+        progress_generation = data.pop("_generation", None)
+        if (
+                job
+                and progress_generation is not None
+                and progress_generation != job.get("_generation", 0)
+        ):
+            detail = {}
+            data = {}
         value = detail.get("value", 0)
         try:
             value = float(value)
@@ -687,7 +996,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             job = self._jobs.get(job_id)
             job_name = job.get("name") if job else job_id
             provider_name = job.get("provider_name", "[系统]") if job else None
-            running = bool(job.get("running")) if job else False
+            running = bool(
+                job and (self._is_job_active(job_id) or job.get("running"))
+            )
             last_started_at = job.get("last_started_at") if job else None
             last_finished_at = job.get("last_finished_at") if job else None
             last_error = job.get("last_error") if job else None
@@ -695,7 +1006,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         detail = await AsyncProgressHelper(self._get_progress_key(job_id)).get() or {}
         if not job and not detail:
             return None
-        data = detail.get("data") or {}
+        data = dict(detail.get("data") or {})
+        progress_generation = data.pop("_generation", None)
+        if (
+                job
+                and progress_generation is not None
+                and progress_generation != job.get("_generation", 0)
+        ):
+            detail = {}
+            data = {}
         value = detail.get("value", 0)
         try:
             value = float(value)
@@ -742,6 +1061,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         """
         构建传递给定时任务内部的进度更新回调。
         """
+        generation = job.get("_generation", 0)
 
         def update_progress(
                 value: Optional[float] = None,
@@ -753,6 +1073,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             """
             progress_data = {
                 "id": job_id,
+                "_generation": generation,
                 "name": job.get("name"),
                 "provider": job.get("provider_name", "[系统]"),
                 "status": "running",
@@ -763,6 +1084,13 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             key = self._get_progress_key(job_id)
 
             async def _update() -> None:
+                with self._lock:
+                    current_job = self._jobs.get(job_id)
+                    if (
+                            current_job is not job
+                            or current_job.get("_generation", 0) != generation
+                    ):
+                        return
                 # 异步后端更新，避免任务函数在事件循环内调用回调时阻塞
                 await AsyncProgressHelper(key).update(
                     value=value,
@@ -772,7 +1100,11 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
 
             # 回调可能在事件循环内（async 任务）或线程池中（sync 任务）被调用，
             # 统一经事件循环提交；无运行中循环时同步执行兜底
-            self._submit_to_loop(_update())
+            self._submit_to_loop(
+                _update(),
+                job_id=job_id,
+                generation=job.get("_generation", 0),
+            )
 
         return update_progress
 
@@ -801,15 +1133,22 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             return str(result[1]) if len(result) > 1 and result[1] else "定时任务返回失败"
         return None
 
-    async def __run_coro_job(self, coro, job_id: str, job: dict) -> None:
+    async def __run_coro_job(
+            self,
+            coro_factory: Callable[[], Any],
+            job_id: str,
+            job: dict,
+            generation: Optional[int] = None,
+    ) -> None:
         """
         在当前事件循环内执行协程定时任务并在真实完成后收敛状态。
         """
+        generation = job.get("_generation", 0) if generation is None else generation
         success = True
         error = None
         try:
             result = await JobExecutionState.await_result(
-                coro,
+                coro_factory(),
                 timeout_seconds=job.get("timeout_seconds"),
             )
             error = self.__get_result_error(result)
@@ -828,79 +1167,102 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             self.__handle_job_error(job_id=job_id, job=job, error=err)
         finally:
             # 协程收尾在事件循环上完成，同步路径（线程池/调用线程）提交到事件循环执行
-            await self.__finish_job(job_id=job_id, success=success, error=error)
+            await self.__finish_job(
+                job_id=job_id,
+                job=job,
+                generation=generation,
+                success=success,
+                error=error,
+            )
 
-    def _track_async_task(
-            self,
-            task: asyncio.Task[Any] | ConcurrentFuture[Any],
-    ) -> asyncio.Task[Any] | ConcurrentFuture[Any]:
-        """登记 Scheduler 自有协程任务，并在完成后移除和消费异常。"""
-        tasks = getattr(self, "_async_tasks", None)
-        if tasks is None:
-            tasks = set()
-            self._async_tasks = tasks
-
-        def _discard(done: asyncio.Task[Any] | ConcurrentFuture[Any]) -> None:
-            """释放已完成任务，并避免跨线程 Future 产生未取回异常。"""
-            with self._lock:
-                tasks.discard(done)
-            try:
-                done.exception()
-            except (asyncio.CancelledError, ConcurrentCancelledError):
-                pass
-
-        with self._lock:
-            tasks.add(task)
-            task.add_done_callback(_discard)
-        return task
-
-    def _create_async_task(self, coro: Any) -> bool:
-        """在当前事件循环创建并登记任务，返回是否已异步接管。"""
-        if not getattr(self, "_accepting_async_tasks", True):
-            coro.close()
-            return False
-        self._track_async_task(asyncio.create_task(coro))
-        return True
-
-    def start(self, job_id: str, *args, **kwargs) -> None:
+    def start(self, job_id: str, *args, **kwargs) -> bool:
         """
         启动定时服务
         """
 
-        def __start_coro(coro) -> bool:
+        def __start_coro(
+                coro_factory: Callable[[], Any],
+                generation: int,
+        ) -> tuple[bool, bool]:
             """
-            启动协程，返回是否由异步回调自行收敛任务状态。
+            启动协程，返回是否异步收尾以及本次提交是否被接受。
             """
             try:
                 running_loop = asyncio.get_running_loop()
             except RuntimeError:
                 running_loop = None
-            target_loop = global_vars.loop
-            if running_loop:
-                return self._create_async_task(
-                    self.__run_coro_job(coro=coro, job_id=job_id, job=job)
-                )
-            if target_loop and target_loop.is_running():
-                if not getattr(self, "_accepting_async_tasks", True):
-                    coro.close()
-                    return False
-                self._track_async_task(
-                    asyncio.run_coroutine_threadsafe(
-                        self.__run_coro_job(coro=coro, job_id=job_id, job=job),
-                        target_loop,
+            target_loop = global_vars.CURRENT_EVENT_LOOP
+            target_loop_available = (
+                target_loop is not None
+                and target_loop.is_running()
+                and not target_loop.is_closed()
+            )
+            if running_loop and (not target_loop_available or running_loop is target_loop):
+                with self._lock:
+                    if not self._accepts_handle(job_id, generation):
+                        return False, False
+                    handle = running_loop.create_task(
+                        self.__run_coro_job(
+                            coro_factory=coro_factory,
+                            job_id=job_id,
+                            job=job,
+                            generation=generation,
+                        ),
                     )
+                    registered = self._register_handle(
+                        job_id=job_id,
+                        generation=generation,
+                        loop=running_loop,
+                        handle=handle,
+                    )
+
+                    def _finish_cancelled_before_start(
+                            submitted: asyncio.Future[Any],
+                    ) -> None:
+                        if submitted.cancelled():
+                            self._finish_unsubmitted_job(
+                                job_id=job_id,
+                                job=job,
+                                generation=generation,
+                                error="任务未提交",
+                            )
+
+                    handle.add_done_callback(_finish_cancelled_before_start)
+                    return registered, registered
+            if target_loop_available:
+                wrapped = self.__run_coro_job(
+                    coro_factory=coro_factory,
+                    job_id=job_id,
+                    job=job,
+                    generation=generation,
                 )
-                return True
-            asyncio.run(coro)
-            return False
+                submitted = self._submit_cross_thread(
+                    wrapped,
+                    target_loop=target_loop,
+                    job_id=job_id,
+                    generation=generation,
+                    on_unstarted_cancel=lambda: self._finish_unsubmitted_job(
+                        job_id=job_id,
+                        job=job,
+                        generation=generation,
+                        error="任务未提交",
+                    ),
+                )
+                return submitted, submitted
+            if self._lifecycle_state in {"stopping", "stopped"}:
+                return False, False
+            asyncio.run(coro_factory())
+            return False, True
 
         # 获取定时任务
         job = self.__prepare_job(job_id)
         if not job:
-            return
+            return False
+        generation = job.get("_generation", 0)
         success = True
         error = None
         deferred_finish = False
+        accepted = True
         # 开始运行
         try:
             if not kwargs:
@@ -916,7 +1278,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             run_in_process = job.get("run_in_process", False)
             if inspect.iscoroutinefunction(func):
                 # 协程函数
-                deferred_finish = __start_coro(func(*args, **kwargs))
+                deferred_finish, accepted = __start_coro(
+                    lambda: func(*args, **kwargs), generation
+                )
             elif run_in_process:
                 # 多进程运行
                 p = multiprocessing.Process(
@@ -936,71 +1300,100 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             self.__handle_job_error(job_id=job_id, job=job, error=e)
         finally:
             if not deferred_finish:
-                # 同步上下文执行异步收尾：优先提交到当前/全局事件循环，无循环时新建循环
-                self._submit_to_loop(self.__finish_job(
-                    job_id=job_id, success=success, error=error
-                ))
+                def finish_without_loop() -> None:
+                    self._finish_unsubmitted_job(
+                        job_id=job_id,
+                        job=job,
+                        generation=generation,
+                        error=error if accepted else "任务未提交",
+                    )
 
-    def _submit_to_loop(self, coro: Any) -> None:
+                # 同步上下文执行异步收尾：优先提交到当前/全局事件循环，无循环时新建循环
+                finish_submitted = self._submit_to_loop(
+                    self.__finish_job(
+                        job_id=job_id,
+                        job=job,
+                        generation=generation,
+                        success=success,
+                        error=error,
+                    ),
+                    job_id=job_id,
+                    generation=generation,
+                    on_unstarted_cancel=finish_without_loop,
+                )
+                if not finish_submitted:
+                    finish_without_loop()
+        return accepted
+
+    def _submit_to_loop(
+            self,
+            coro: Any,
+            *,
+            job_id: Optional[str] = None,
+            generation: int = 0,
+            on_unstarted_cancel: Optional[Callable[[], None]] = None,
+    ) -> bool:
         """
         把协程提交到事件循环执行，兼容以下调用环境：
-        - 已在事件循环内（async 任务内部）：排队为独立任务，避免阻塞
-        - 外部线程且全局循环在运行：跨线程提交，非阻塞
+        - 应用主循环可用：统一由主循环拥有任务和关闭顺序
+        - 仅调用方循环可用：在当前循环排队为独立任务
         - 无运行中循环（测试/CLI）：新建循环同步执行，确保进度不丢失
+
+        带有 job 标识的句柄由 Scheduler 自己持有，关闭时可以取消并等待。
         """
         try:
             running_loop = asyncio.get_running_loop()
         except RuntimeError:
             running_loop = None
-        if running_loop:
-            self._create_async_task(coro)
-        elif global_vars.loop and global_vars.loop.is_running():
-            if not getattr(self, "_accepting_async_tasks", True):
-                coro.close()
-                return
-            self._track_async_task(
-                asyncio.run_coroutine_threadsafe(coro, global_vars.loop)
-            )
+        target_loop = global_vars.CURRENT_EVENT_LOOP
+        target_loop_available = (
+            target_loop is not None
+            and target_loop.is_running()
+            and not target_loop.is_closed()
+        )
+        if running_loop and (not target_loop_available or running_loop is target_loop):
+            if job_id is not None:
+                with self._lock:
+                    if not self._accepts_handle(job_id, generation):
+                        coro.close()
+                        return False
+                    handle = running_loop.create_task(coro)
+                    registered = self._register_handle(
+                        job_id=job_id,
+                        generation=generation,
+                        loop=running_loop,
+                        handle=handle,
+                    )
+                    if on_unstarted_cancel:
+                        handle.add_done_callback(
+                            lambda submitted: (
+                                on_unstarted_cancel()
+                                if submitted.cancelled()
+                                else None
+                            )
+                        )
+                    return registered
+            else:
+                running_loop.create_task(coro)
+                return True
+        elif target_loop_available:
+            if job_id is not None:
+                return self._submit_cross_thread(
+                    coro,
+                    target_loop=target_loop,
+                    job_id=job_id,
+                    generation=generation,
+                    on_unstarted_cancel=on_unstarted_cancel,
+                )
+            else:
+                asyncio.run_coroutine_threadsafe(coro, target_loop)
+                return True
+        elif self._lifecycle_state in {"stopping", "stopped"}:
+            coro.close()
+            return False
         else:
             asyncio.run(coro)
-
-    def _cancel_async_tasks(self) -> tuple[asyncio.Task[Any] | ConcurrentFuture[Any], ...]:
-        """停止接收新协程并请求取消现有 Scheduler 任务。"""
-        with self._lock:
-            self._accepting_async_tasks = False
-            tasks = tuple(getattr(self, "_async_tasks", ()))
-        for task in tasks:
-            if isinstance(task, asyncio.Task):
-                loop = task.get_loop()
-                if loop.is_running():
-                    loop.call_soon_threadsafe(task.cancel)
-                else:
-                    task.cancel()
-            else:
-                task.cancel()
-        return tasks
-
-    async def async_stop(self, *, timeout_seconds: float = 30.0) -> None:
-        """异步关闭 Scheduler，并在有限预算内等待其协程任务收口。"""
-        self.stop()
-        tasks = tuple(getattr(self, "_async_tasks", ()))
-        if not tasks:
-            return
-        awaitables = []
-        for task in tasks:
-            if isinstance(task, asyncio.Future):
-                awaitables.append(task)
-            else:
-                awaitables.append(asyncio.wrap_future(task))
-        try:
-            await asyncio.wait_for(
-                asyncio.gather(*awaitables, return_exceptions=True),
-                timeout=timeout_seconds,
-            )
-        except asyncio.TimeoutError:
-            logger.error("等待定时器协程任务收口超时")
-            for task in tasks:
-                task.cancel()
+            return True
 
     @staticmethod
     def _get_agent_task_job_id(task_id: int) -> str:
@@ -1017,10 +1410,21 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         job_id = self._get_agent_task_job_id(task_id)
         with self._lock:
             job = self._jobs.get(job_id)
-            if not job or job.get("running"):
+            if (
+                    not self._accepting_submissions()
+                    or not job
+                    or self._is_job_active(job_id)
+                    or job.get("running")
+                    or job_id in self._agent_task_reservations
+            ):
                 return False
-        self.start(job_id, task_id=task_id, trigger_source="manual")
-        return True
+            self._agent_task_reservations[job_id] = threading.get_ident()
+        try:
+            result = self.start(job_id, task_id=task_id, trigger_source="manual")
+            return result is not False
+        finally:
+            with self._lock:
+                self._agent_task_reservations.pop(job_id, None)
 
     def init_agent_task_jobs(self) -> None:
         """
@@ -1087,7 +1491,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
 
         job_id = self._get_agent_task_job_id(task_id)
         with self._lock:
-            self._jobs[job_id] = JobSpec(
+            job = JobSpec(
                 job_id,
                 task.name,
                 self.execute_agent_task,
@@ -1095,6 +1499,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 recovery=JobRecoveryPolicy.NEXT_SCHEDULE,
                 kwargs={"task_id": task_id},
             ).to_runtime_state()
+            self._assign_job_generation(job_id, job)
+            self._jobs[job_id] = job
             self._jobs[job_id]["provider_name"] = "[Agent]"
             # 已开始的一次任务在重启后结果未知，只保留显式执行入口，不能按
             # 过期触发时间自动重放可能已经发生的外部副作用。
@@ -1312,13 +1718,15 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         with self._lock:
             try:
                 job_id = f"workflow-{workflow.id}"
-                self._jobs[job_id] = JobSpec(
+                job = JobSpec(
                     job_id,
                     workflow.name,
                     WorkflowChain().process,
                     "workflow",
                 ).to_runtime_state()
-                self._jobs[job_id]["provider_name"] = "工作流"
+                self._assign_job_generation(job_id, job)
+                job["provider_name"] = "工作流"
+                self._jobs[job_id] = job
                 self._scheduler.add_job(
                     self.start,
                     trigger=CronTrigger.from_crontab(workflow.timer),
@@ -1362,17 +1770,19 @@ 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] = JobSpec(
+                    job = JobSpec(
                         job_id,
                         service["name"],
                         service["func"],
                         f"plugin:{pid}",
                         kwargs=service.get("func_kwargs") or {},
                     ).to_runtime_state()
-                    self._jobs[job_id].update(
+                    self._assign_job_generation(job_id, job)
+                    job.update(
                         pid=pid,
                         provider_name=plugin_name,
                     )
+                    self._jobs[job_id] = job
                     self._scheduler.add_job(
                         self.start,
                         service["trigger"],
@@ -1414,7 +1824,11 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             for job_id, service in self._jobs.items():
                 name = service.get("name")
                 provider_name = service.get("provider_name")
-                if service.get("running") and name and provider_name:
+                if (
+                        (self._is_job_active(job_id) or service.get("running"))
+                        and name
+                        and provider_name
+                ):
                     if job_id not in added:
                         added.append(job_id)
                     progress = self.get_progress(job_id)
@@ -1441,7 +1855,11 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 if not service:
                     continue
                 # 任务状态
-                status = "正在运行" if service.get("running") else "等待"
+                status = (
+                    "正在运行"
+                    if self._is_job_active(job_id) or service.get("running")
+                    else "等待"
+                )
                 # 下次运行时间
                 next_run = TimerUtils.time_difference(job.next_run_time)
                 progress = self.get_progress(job_id)
@@ -1471,7 +1889,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                         id=job_id,
                         name=service.get("name"),
                         provider=service.get("provider_name", "[系统]"),
-                        status="等待",
+                        status=(
+                            "正在运行" if self._is_job_active(job_id) else "等待"
+                        ),
                         progress=progress.value if progress else 0,
                         progress_text=progress.text if progress else None,
                         progress_enable=progress.enable if progress else False,
@@ -1480,23 +1900,77 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 )
             return schedulers
 
-    def stop(self):
+    def _begin_stop(self) -> tuple[Any, tuple[_SchedulerHandle, ...]]:
+        """关闭提交入口并摘出当前调度器与其拥有的异步句柄。"""
+        with self._lock:
+            self._lifecycle_state = "stopping"
+            self._event.set()
+            scheduler = self._scheduler
+            self._scheduler = None
+            self._agent_task_reservations.clear()
+            handles = tuple(self._handles.values())
+        if scheduler:
+            try:
+                scheduler.remove_all_jobs()
+            except Exception as err:
+                logger.error("移除定时任务失败：%s", err)
+        return scheduler, handles
+
+    def _begin_reload(self) -> tuple[bool, Any]:
+        """停止旧计划的提交入口，保留已开始任务直到其自然完成。"""
+        with self._lock:
+            if (
+                    global_vars.is_system_stopped
+                    or self._lifecycle_state in {"stopping", "reloading"}
+            ):
+                return False, None
+            self._lifecycle_state = "reloading"
+            self._event.set()
+            scheduler = self._scheduler
+            self._scheduler = None
+            self._agent_task_reservations.clear()
+        if scheduler:
+            try:
+                scheduler.remove_all_jobs()
+            except Exception as err:
+                logger.error("移除定时任务失败：%s", err)
+        return True, scheduler
+
+    @staticmethod
+    def _shutdown_scheduler_sync(scheduler: Any) -> None:
+        """等待 APScheduler 自有线程池停止。"""
+        if scheduler and scheduler.running:
+            scheduler.shutdown()
+
+    def stop(self) -> None:
         """
-        关闭定时服务
+        关闭定时服务的同步兼容入口。
+
+        应用生命周期使用 ``stop_async``，以便等待事件循环中的协程句柄；同步
+        调用方仍可请求取消并等待 APScheduler 自有线程池收口。
         """
-        self._cancel_async_tasks()
         with lock:
             try:
-                if self._scheduler:
-                    logger.info("正在停止定时任务...")
-                    self._event.set()
-                    self._scheduler.remove_all_jobs()
-                    if self._scheduler.running:
-                        self._scheduler.shutdown()
-                    self._scheduler = None
-                    logger.info("定时任务停止完成")
-            except Exception as e:
-                logger.error(f"停止定时任务失败：：{str(e)} - {traceback.format_exc()}")
+                scheduler, handles = self._begin_stop()
+                for handle in handles:
+                    self._cancel_handle(handle)
+                self._shutdown_scheduler_sync(scheduler)
+                with self._lock:
+                    self._lifecycle_state = "stopped"
+                logger.info("定时任务停止完成")
+            except Exception as err:
+                logger.error(f"停止定时任务失败：{err} - {traceback.format_exc()}")
+
+    async def stop_async(self) -> None:
+        """关闭调度器并等待已投递协程收口。"""
+        scheduler, handles = self._begin_stop()
+        for handle in handles:
+            self._cancel_handle(handle)
+        await asyncio.to_thread(self._shutdown_scheduler_sync, scheduler)
+        await self._await_cancelled_handles(handles)
+        with self._lock:
+            self._lifecycle_state = "stopped"
+        logger.info("定时任务停止完成")
 
     @staticmethod
     def clear_cache():
diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py
index ce9d4edc1..e69827f31 100644
--- a/app/startup/modules_initializer.py
+++ b/app/startup/modules_initializer.py
@@ -580,7 +580,7 @@ async def stop_modules():
 
     await run_step("AI智能体", stop_agent)
     await run_step("模块", lambda: ModuleManager().shutdown())
-    await run_step("事件消费", lambda: EventManager().stop())
+    await run_step("事件消费", lambda: EventManager().stop_async())
     await run_step("浏览器会话", close_browser_sessions)
     await run_step("托管资源", stop_managed_resources)
     await run_step("DoH服务", lambda: DohHelper().shutdown())
diff --git a/app/startup/scheduler_initializer.py b/app/startup/scheduler_initializer.py
index 5d9688085..6430dd6a9 100644
--- a/app/startup/scheduler_initializer.py
+++ b/app/startup/scheduler_initializer.py
@@ -16,7 +16,7 @@ def init_scheduler():
 
 def stop_scheduler():
     """
-    停止定时器；生命周期事件循环中返回有限等待的兼容协程。
+    停止定时器；生命周期事件循环中返回可等待的收口协程。
     """
     scheduler = Scheduler()
     try:
@@ -24,7 +24,7 @@ def stop_scheduler():
     except RuntimeError:
         scheduler.stop()
         return None
-    return scheduler.async_stop()
+    return scheduler.stop_async()
 
 
 def restart_scheduler():
diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md
index a91c848d6..f738f84ca 100644
--- a/docs/adr/0007-background-action-reliability.md
+++ b/docs/adr/0007-background-action-reliability.md
@@ -86,8 +86,8 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同
 - IMDb 同步 `clear_cache()` ABI 在事件循环内触发的异步缓存清理登记为
   `module.imdb.cache_clear`；同步调用方式和无运行事件循环时的立即清理行为保持不变，宿主关停后不再
   接受新的清理任务。
-- Scheduler 的协程作业与异步进度收尾由 Scheduler 自有任务集合持有；同步 `start()` / `stop()` ABI 保持，
-  生命周期关闭入口额外等待有限预算，跨线程提交的 Future 也会在停止时收到取消请求。
+- Scheduler 的协程作业与异步进度收尾由 Scheduler 自有句柄表持有；同步 `start()` / `stop()` ABI 保持，
+  生命周期关闭入口等待目标事件循环确认真实收尾，跨线程取消代理不作为任务完成凭据。
 
 ### Transfer pending / 文件整理
 
diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md
index e36b4a244..cd3fe46ed 100644
--- a/docs/refactor/backend-architecture-next-stage.md
+++ b/docs/refactor/backend-architecture-next-stage.md
@@ -839,8 +839,8 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas
 - IMDb 同步清缓存兼容入口在运行事件循环时改由 TaskRegistry 登记异步缓存清理任务，owner 为
   `module.imdb.cache_clear`；同步签名、模块调用方式和无事件循环时的立即清理语义保持不变。
 - Scheduler 的协程作业和异步进度收尾不再使用无主 `create_task` 或丢弃跨线程 Future；由 Scheduler 自有
-  任务集合登记、停止时取消，生命周期入口通过异步兼容包装器在有限预算内等待收口，保留旧同步
-  `Scheduler.start()` / `Scheduler.stop()` 与插件调度 ABI。
+  句柄表登记并在停止时取消，completion 只在目标事件循环确认真实收尾后完成。关闭总预算由宿主生命周期
+  统一控制，保留旧同步 `Scheduler.start()` / `Scheduler.stop()` 与插件调度 ABI。
 
 #### ARCH-251：用现有数据库做首个 durable side-effect pilot
 
diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json
index 9ea776895..60aefa3e1 100644
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6451,
-  "edge_sha256": "9075d8717e384580cd6a41bc36685438770db4a8d8f18c57e6c494f32937113a",
+  "edge_count": 6479,
+  "edge_sha256": "a65f8d4024e2299b37510359c7ffea91219f0aa0ed9673b74a3eb22d51d6c67f",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -1510,6 +1510,8 @@
     "app.api.context -> app.application.subscription.delete",
     "app.api.context -> app.application.subscription.identity",
     "app.api.context -> app.application.subscription.mutation",
+    "app.api.context -> app.runtime",
+    "app.api.context -> app.runtime.tasks",
     "app.api.context -> app.startup",
     "app.api.context -> app.startup.context",
     "app.api.dependencies.agent -> app.api",
@@ -1599,6 +1601,7 @@
     "app.api.dependencies.subscription -> app.runtime",
     "app.api.dependencies.subscription -> app.runtime.events",
     "app.api.dependencies.subscription -> app.runtime.log",
+    "app.api.dependencies.subscription -> app.runtime.tasks",
     "app.api.dependencies.subscription -> app.schemas",
     "app.api.dependencies.subscription -> app.schemas.types",
     "app.api.dependencies.subscription -> app.startup",
@@ -1682,6 +1685,7 @@
     "app.api.endpoints.anthropic -> app.agent",
     "app.api.endpoints.anthropic -> app.agent.runtime_loader",
     "app.api.endpoints.anthropic -> app.api",
+    "app.api.endpoints.anthropic -> app.api.context",
     "app.api.endpoints.anthropic -> app.api.endpoints",
     "app.api.endpoints.anthropic -> app.api.endpoints.openai",
     "app.api.endpoints.anthropic -> app.api.openai_utils",
@@ -1689,6 +1693,8 @@
     "app.api.endpoints.anthropic -> app.api.presentation.sse",
     "app.api.endpoints.anthropic -> app.application",
     "app.api.endpoints.anthropic -> app.application.configuration",
+    "app.api.endpoints.anthropic -> app.runtime",
+    "app.api.endpoints.anthropic -> app.runtime.tasks",
     "app.api.endpoints.anthropic -> app.schemas",
     "app.api.endpoints.anthropic -> app.schemas.openai",
     "app.api.endpoints.auth -> app.api",
@@ -1833,6 +1839,7 @@
     "app.api.endpoints.history -> app.runtime.config",
     "app.api.endpoints.history -> app.runtime.log",
     "app.api.endpoints.history -> app.runtime.progress",
+    "app.api.endpoints.history -> app.runtime.tasks",
     "app.api.endpoints.history -> app.schemas",
     "app.api.endpoints.history -> app.schemas.common",
     "app.api.endpoints.history -> app.schemas.history",
@@ -1945,6 +1952,7 @@
     "app.api.endpoints.message -> app.adapters.web.security",
     "app.api.endpoints.message -> app.adapters.web.security.access",
     "app.api.endpoints.message -> app.api",
+    "app.api.endpoints.message -> app.api.context",
     "app.api.endpoints.message -> app.api.dependencies",
     "app.api.endpoints.message -> app.api.dependencies.agent",
     "app.api.endpoints.message -> app.api.dependencies.auth",
@@ -1961,6 +1969,7 @@
     "app.api.endpoints.message -> app.runtime.extensions",
     "app.api.endpoints.message -> app.runtime.extensions.service_config",
     "app.api.endpoints.message -> app.runtime.log",
+    "app.api.endpoints.message -> app.runtime.tasks",
     "app.api.endpoints.message -> app.schemas",
     "app.api.endpoints.message -> app.schemas.message",
     "app.api.endpoints.message -> app.schemas.response",
@@ -2029,11 +2038,14 @@
     "app.api.endpoints.openai -> app.agent.contracts",
     "app.api.endpoints.openai -> app.agent.runtime_loader",
     "app.api.endpoints.openai -> app.api",
+    "app.api.endpoints.openai -> app.api.context",
     "app.api.endpoints.openai -> app.api.openai_utils",
     "app.api.endpoints.openai -> app.api.presentation",
     "app.api.endpoints.openai -> app.api.presentation.sse",
     "app.api.endpoints.openai -> app.application",
     "app.api.endpoints.openai -> app.application.configuration",
+    "app.api.endpoints.openai -> app.runtime",
+    "app.api.endpoints.openai -> app.runtime.tasks",
     "app.api.endpoints.openai -> app.schemas",
     "app.api.endpoints.openai -> app.schemas.openai",
     "app.api.endpoints.openai -> app.schemas.types",
@@ -2048,6 +2060,7 @@
     "app.api.endpoints.plugin -> app.adapters.web.security",
     "app.api.endpoints.plugin -> app.adapters.web.security.access",
     "app.api.endpoints.plugin -> app.api",
+    "app.api.endpoints.plugin -> app.api.context",
     "app.api.endpoints.plugin -> app.api.dependencies",
     "app.api.endpoints.plugin -> app.api.dependencies.auth",
     "app.api.endpoints.plugin -> app.api.dependencies.plugin",
@@ -2069,6 +2082,7 @@
     "app.api.endpoints.plugin -> app.runtime.extensions.plugin",
     "app.api.endpoints.plugin -> app.runtime.extensions.plugin.contracts",
     "app.api.endpoints.plugin -> app.runtime.log",
+    "app.api.endpoints.plugin -> app.runtime.tasks",
     "app.api.endpoints.plugin -> app.schemas",
     "app.api.endpoints.plugin -> app.schemas.common",
     "app.api.endpoints.plugin -> app.schemas.exception",
@@ -2122,6 +2136,7 @@
     "app.api.endpoints.site -> app.adapters.web.security",
     "app.api.endpoints.site -> app.adapters.web.security.access",
     "app.api.endpoints.site -> app.api",
+    "app.api.endpoints.site -> app.api.context",
     "app.api.endpoints.site -> app.api.dependencies",
     "app.api.endpoints.site -> app.api.dependencies.auth",
     "app.api.endpoints.site -> app.api.dependencies.site",
@@ -2145,6 +2160,7 @@
     "app.api.endpoints.site -> app.domain.site",
     "app.api.endpoints.site -> app.runtime",
     "app.api.endpoints.site -> app.runtime.log",
+    "app.api.endpoints.site -> app.runtime.tasks",
     "app.api.endpoints.site -> app.schemas",
     "app.api.endpoints.site -> app.schemas.common",
     "app.api.endpoints.site -> app.schemas.response",
@@ -2180,6 +2196,7 @@
     "app.api.endpoints.subscribe -> app.adapters.web.security",
     "app.api.endpoints.subscribe -> app.adapters.web.security.access",
     "app.api.endpoints.subscribe -> app.api",
+    "app.api.endpoints.subscribe -> app.api.context",
     "app.api.endpoints.subscribe -> app.api.dependencies",
     "app.api.endpoints.subscribe -> app.api.dependencies.auth",
     "app.api.endpoints.subscribe -> app.api.dependencies.subscription",
@@ -2201,6 +2218,7 @@
     "app.api.endpoints.subscribe -> app.domain.metainfo",
     "app.api.endpoints.subscribe -> app.runtime",
     "app.api.endpoints.subscribe -> app.runtime.events",
+    "app.api.endpoints.subscribe -> app.runtime.tasks",
     "app.api.endpoints.subscribe -> app.schemas",
     "app.api.endpoints.subscribe -> app.schemas.common",
     "app.api.endpoints.subscribe -> app.schemas.event",
@@ -2347,9 +2365,12 @@
     "app.api.endpoints.webhook -> app.adapters.web.security",
     "app.api.endpoints.webhook -> app.adapters.web.security.access",
     "app.api.endpoints.webhook -> app.api",
+    "app.api.endpoints.webhook -> app.api.context",
     "app.api.endpoints.webhook -> app.api.response",
     "app.api.endpoints.webhook -> app.chain",
     "app.api.endpoints.webhook -> app.chain.webhook",
+    "app.api.endpoints.webhook -> app.runtime",
+    "app.api.endpoints.webhook -> app.runtime.tasks",
     "app.api.endpoints.webhook -> app.schemas",
     "app.api.endpoints.webhook -> app.schemas.response",
     "app.api.endpoints.workflow -> app.adapters",
@@ -2555,6 +2576,8 @@
     "app.application.mediaserver -> app.schemas.mediaserver",
     "app.application.mediaserver -> app.schemas.system",
     "app.application.mediaserver -> app.schemas.types",
+    "app.application.messaging.agent -> app.runtime",
+    "app.application.messaging.agent -> app.runtime.tasks",
     "app.application.messaging.agent -> app.schemas",
     "app.application.messaging.agent -> app.schemas.types",
     "app.application.messaging.chat -> app.application",
@@ -3619,7 +3642,6 @@
     "app.db.oper.passkey -> app.db.base",
     "app.db.oper.passkey -> app.db.models",
     "app.db.oper.passkey -> app.db.models.passkey",
-    "app.db.oper.passkey -> app.db.uow",
     "app.db.oper.plugindata -> app.db",
     "app.db.oper.plugindata -> app.db.base",
     "app.db.oper.plugindata -> app.db.models",
@@ -4300,6 +4322,7 @@
     "app.modules.imdb.api -> app.runtime.cache",
     "app.modules.imdb.api -> app.runtime.log",
     "app.modules.imdb.api -> app.runtime.settings",
+    "app.modules.imdb.api -> app.runtime.tasks",
     "app.modules.indexer -> app.application",
     "app.modules.indexer -> app.application.site",
     "app.modules.indexer -> app.application.site.health",
@@ -5626,6 +5649,7 @@
     "app.runtime.extensions.plugin_manager -> app.foundation.version",
     "app.runtime.extensions.plugin_manager -> app.runtime",
     "app.runtime.extensions.plugin_manager -> app.runtime.events",
+    "app.runtime.extensions.plugin_manager -> app.runtime.execution",
     "app.runtime.extensions.plugin_manager -> app.runtime.extensions",
     "app.runtime.extensions.plugin_manager -> app.runtime.extensions.plugin",
     "app.runtime.extensions.plugin_manager -> app.runtime.extensions.plugin.access",
@@ -5994,6 +6018,8 @@
     "app.startup.context -> app.application.subscription.identity",
     "app.startup.context -> app.application.subscription.mutation",
     "app.startup.context -> app.application.workflow",
+    "app.startup.context -> app.runtime",
+    "app.startup.context -> app.runtime.tasks",
     "app.startup.database -> app.adapters",
     "app.startup.database -> app.adapters.system",
     "app.startup.database -> app.adapters.system.backup",
@@ -6057,6 +6083,7 @@
     "app.startup.lifecycle -> app.runtime.log",
     "app.startup.lifecycle -> app.runtime.settings",
     "app.startup.lifecycle -> app.runtime.state",
+    "app.startup.lifecycle -> app.runtime.tasks",
     "app.startup.lifecycle -> app.runtime.topology",
     "app.startup.lifecycle -> app.startup",
     "app.startup.lifecycle -> app.startup.cache_initializer",
@@ -6171,6 +6198,7 @@
     "app.startup.modules_initializer -> app.runtime.observability",
     "app.startup.modules_initializer -> app.runtime.settings",
     "app.startup.modules_initializer -> app.runtime.state",
+    "app.startup.modules_initializer -> app.runtime.tasks",
     "app.startup.modules_initializer -> app.runtime.thread",
     "app.startup.modules_initializer -> app.scheduler",
     "app.startup.modules_initializer -> app.schemas",
@@ -6468,7 +6496,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 799,
+  "module_count": 800,
   "modules": [
     "app",
     "app.adapters",
@@ -7161,6 +7189,7 @@
     "app.runtime.scheduling",
     "app.runtime.settings",
     "app.runtime.state",
+    "app.runtime.tasks",
     "app.runtime.thread",
     "app.runtime.topology",
     "app.scheduler",
diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py
index eaaac8c6c..ad8456b7e 100644
--- a/tests/test_agent_scheduled_tasks.py
+++ b/tests/test_agent_scheduled_tasks.py
@@ -127,9 +127,15 @@ def _add_agent_task(trigger_type: str, trigger_value: str, prefix: str):
 def _build_agent_task_scheduler(reconcile: bool = False) -> Scheduler:
     """构造不启动后台线程的 Agent 任务调度器。"""
     scheduler = object.__new__(Scheduler)
+    scheduler._event = threading.Event()
     scheduler._lock = threading.RLock()
     scheduler._jobs = {}
     scheduler._scheduler = BackgroundScheduler(timezone=settings.TZ)
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
     scheduler._agent_task_interruptions_reconciled = False
     if reconcile:
         scheduler._reconcile_agent_task_interruptions()
@@ -316,6 +322,11 @@ def test_scheduler_registers_and_removes_agent_task_job() -> None:
     scheduler._lock = threading.RLock()
     scheduler._jobs = {}
     scheduler._scheduler = BackgroundScheduler(timezone=settings.TZ)
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
 
     next_run_at = scheduler.update_agent_task_job(task.id)
     job_id = scheduler._get_agent_task_job_id(task.id)
@@ -642,6 +653,63 @@ async def test_scheduler_config_reload_does_not_interrupt_running_agent_task() -
     manager.process_message.assert_not_awaited()
 
 
+@pytest.mark.anyio
+async def test_scheduler_config_reload_preserves_active_agent_task(
+        monkeypatch,
+) -> None:
+    """配置热重载只替换后续计划，已开始的 AgentTask 仍按真实结果收口。"""
+    task = _add_agent_task("cron", "0 * * * *", "reload-active")
+    scheduler = _build_agent_task_scheduler()
+    scheduler.init_agent_task_jobs()
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def process_message(**_kwargs) -> str:
+        started.set()
+        await release.wait()
+        return "执行完成"
+
+    manager = SimpleNamespace(
+        execute_scheduled_task=AgentManager.execute_scheduled_task,
+        process_message=process_message,
+    )
+    manager.execute_scheduled_task = AgentManager.execute_scheduled_task.__get__(manager)
+    monkeypatch.setattr(
+        "app.agent.runtime_loader.get_running_agent_manager",
+        lambda: manager,
+    )
+    monkeypatch.setattr(
+        scheduler,
+        "init",
+        Mock(side_effect=lambda **_kwargs: setattr(
+            scheduler,
+            "_lifecycle_state",
+            "running",
+        )),
+    )
+
+    job_id = scheduler._get_agent_task_job_id(task.id)
+    assert scheduler.start(job_id) is True
+    await asyncio.wait_for(started.wait(), timeout=1)
+    assert AgentTaskOper().get(task.id).last_status == "running"
+
+    await scheduler.on_config_changed()
+    assert AgentTaskOper().get(task.id).last_status == "running"
+    assert scheduler._handles
+
+    release.set()
+
+    async def wait_until_released() -> None:
+        while scheduler._handles:
+            await asyncio.sleep(0)
+
+    await asyncio.wait_for(wait_until_released(), timeout=1)
+
+    completed = AgentTaskOper().get(task.id)
+    assert completed.last_status == "success"
+    assert completed.last_result == "执行完成"
+
+
 def test_scheduler_restart_keeps_interrupted_cron_future_schedule() -> None:
     """周期任务中断后只保留下次正常调度，不抹掉本轮中断事实。"""
     task = _add_agent_task("cron", "0 * * * *", "restart-cron")
@@ -699,6 +767,11 @@ def test_scheduler_starts_registered_agent_task_without_waiting() -> None:
             "running": False,
         }
     }
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
     scheduler.start = Mock()
 
     assert scheduler.start_agent_task(7) is True
diff --git a/tests/test_database_backup_scheduler.py b/tests/test_database_backup_scheduler.py
index d876dc756..a05041792 100644
--- a/tests/test_database_backup_scheduler.py
+++ b/tests/test_database_backup_scheduler.py
@@ -1,6 +1,7 @@
 """数据库备份与宿主调度器的接入合同。"""
 
 import ast
+import threading
 from dataclasses import replace
 from pathlib import Path
 from unittest.mock import Mock
@@ -27,6 +28,12 @@ def _scheduler() -> Scheduler:
     scheduler = object.__new__(Scheduler)
     scheduler._scheduler = _SchedulerStub()
     scheduler._jobs = {}
+    scheduler._lock = threading.RLock()
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
     return scheduler
 
 
diff --git a/tests/test_event_dispatch_snapshot.py b/tests/test_event_dispatch_snapshot.py
index ee425919f..4630b74f2 100644
--- a/tests/test_event_dispatch_snapshot.py
+++ b/tests/test_event_dispatch_snapshot.py
@@ -1,7 +1,12 @@
-"""事件调度订阅快照的并发回归测试。"""
+"""事件调度订阅快照和生命周期回归测试。"""
+
+import asyncio
+import threading
 
 import pytest
 
+from app.runtime.config import global_vars
+from app.runtime import events as events_module
 from app.runtime.events import Event, eventmanager
 from app.schemas.types import ChainEventType, EventType
 
@@ -37,6 +42,26 @@ def isolated_eventmanager(monkeypatch):
         "_EventManager__executor",
         _ImmediateExecutor(),
     )
+    monkeypatch.setattr(
+        eventmanager,
+        "_EventManager__event",
+        threading.Event(),
+    )
+    monkeypatch.setattr(
+        eventmanager,
+        "_EventManager__consumer_threads",
+        [],
+    )
+    monkeypatch.setattr(
+        eventmanager,
+        "_EventManager__lifecycle_state",
+        "new",
+    )
+    monkeypatch.setattr(
+        eventmanager,
+        "_EventManager__async_handles",
+        {},
+    )
     return eventmanager
 
 
@@ -158,3 +183,149 @@ async def test_async_chain_dispatch_uses_subscription_snapshot(
     calls.clear()
     assert await dispatch(Event(ChainEventType.NameRecognize, {})) is True
     assert calls == ["mutating", "late"]
+
+
+@pytest.mark.asyncio
+async def test_async_broadcast_handles_are_cancelled_on_shutdown(isolated_eventmanager):
+    """事件总线关闭时必须取消并收口已投递的异步广播处理器。"""
+    global_vars.set_loop(asyncio.get_running_loop())
+    handler_count = 5
+    active = 0
+    cancelled = 0
+    all_active = asyncio.Event()
+
+    async def handler(_event):
+        nonlocal active, cancelled
+        active += 1
+        if active == handler_count:
+            all_active.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            cancelled += 1
+            raise
+
+    isolated_eventmanager.add_event_listener(EventType.ConfigChanged, handler)
+    isolated_eventmanager.start()
+    consumer_threads = tuple(isolated_eventmanager._EventManager__consumer_threads)
+    for _ in range(handler_count):
+        isolated_eventmanager.send_event(
+            EventType.ConfigChanged,
+            {"key": {"shutdown"}},
+        )
+
+    await asyncio.wait_for(all_active.wait(), timeout=2)
+    assert len(isolated_eventmanager._EventManager__async_handles) == handler_count
+
+    await isolated_eventmanager.stop_async()
+    await asyncio.sleep(0)
+
+    assert cancelled == handler_count
+    assert isolated_eventmanager._EventManager__async_handles == {}
+    assert not any(
+        thread.is_alive()
+        for thread in consumer_threads
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_broadcast_shutdown_waits_for_handler_cleanup(
+        isolated_eventmanager,
+) -> None:
+    """提交代理变为 cancelled 后仍须等待处理器 finally 真正完成。"""
+    global_vars.set_loop(asyncio.get_running_loop())
+    started = asyncio.Event()
+    cancelling = asyncio.Event()
+    cleanup_release = asyncio.Event()
+
+    async def handler(_event):
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            cancelling.set()
+            await cleanup_release.wait()
+            raise
+
+    isolated_eventmanager.add_event_listener(EventType.ConfigChanged, handler)
+    isolated_eventmanager.start()
+    isolated_eventmanager.send_event(EventType.ConfigChanged, {"key": {"shutdown"}})
+    await asyncio.wait_for(started.wait(), timeout=2)
+
+    stop_task = asyncio.create_task(isolated_eventmanager.stop_async())
+    await asyncio.wait_for(cancelling.wait(), timeout=1)
+    await asyncio.sleep(0)
+    assert not stop_task.done()
+    assert isolated_eventmanager._EventManager__lifecycle_state == "stopping"
+
+    stop_task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await stop_task
+    assert isolated_eventmanager._EventManager__async_handles
+    assert isolated_eventmanager._EventManager__lifecycle_state == "stopping"
+
+    cleanup_release.set()
+
+    async def wait_until_released() -> None:
+        while isolated_eventmanager._EventManager__async_handles:
+            await asyncio.sleep(0)
+
+    await asyncio.wait_for(wait_until_released(), timeout=1)
+    await isolated_eventmanager.stop_async()
+    assert isolated_eventmanager._EventManager__async_handles == {}
+    assert isolated_eventmanager._EventManager__lifecycle_state == "stopped"
+
+
+@pytest.mark.asyncio
+async def test_async_broadcast_submission_is_registered_before_stop_snapshot(
+        isolated_eventmanager,
+        monkeypatch,
+) -> None:
+    """事件处理器提交和 owner 登记不得被关闭快照从中切开。"""
+    global_vars.set_loop(asyncio.get_running_loop())
+    submission_entered = threading.Event()
+    submission_release = threading.Event()
+    real_submit = events_module.asyncio.run_coroutine_threadsafe
+
+    def delayed_submit(coroutine, loop):
+        handle = real_submit(coroutine, loop)
+        submission_entered.set()
+        submission_release.wait(timeout=1)
+        return handle
+
+    monkeypatch.setattr(
+        events_module.asyncio,
+        "run_coroutine_threadsafe",
+        delayed_submit,
+    )
+    isolated_eventmanager.start()
+
+    async def handler() -> None:
+        await asyncio.Event().wait()
+
+    submit_thread = threading.Thread(
+        target=isolated_eventmanager._EventManager__register_async_handle,
+        args=(handler(),),
+    )
+    submit_thread.start()
+    assert await asyncio.to_thread(submission_entered.wait, 1)
+
+    stop_result = []
+    stop_thread = threading.Thread(
+        target=lambda: stop_result.append(
+            isolated_eventmanager._EventManager__begin_stop()
+        )
+    )
+    stop_thread.start()
+    await asyncio.sleep(0.02)
+    assert stop_thread.is_alive()
+
+    submission_release.set()
+    await asyncio.to_thread(submit_thread.join, 1)
+    await asyncio.to_thread(stop_thread.join, 1)
+    assert not submit_thread.is_alive()
+    assert not stop_thread.is_alive()
+    assert len(isolated_eventmanager._EventManager__async_handles) == 1
+
+    await isolated_eventmanager.stop_async()
+    assert isolated_eventmanager._EventManager__async_handles == {}
diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py
index 8e73c39c3..caeed2e16 100644
--- a/tests/test_lifecycle_shutdown.py
+++ b/tests/test_lifecycle_shutdown.py
@@ -694,7 +694,7 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict:
     dependencies = {}
     for name, method_name in (
         ("ModuleManager", "shutdown"),
-        ("EventManager", "stop"),
+        ("EventManager", "stop_async"),
         ("DohHelper", "shutdown"),
         ("ThreadHelper", "shutdown"),
         ("RedisHelper", "close"),
diff --git a/tests/test_plugin_local_sync.py b/tests/test_plugin_local_sync.py
index 7cd5b0724..335354e0f 100644
--- a/tests/test_plugin_local_sync.py
+++ b/tests/test_plugin_local_sync.py
@@ -122,6 +122,11 @@ def _build_scheduler_for_plugin_reload(jobs: dict, backend) -> Scheduler:
     scheduler._lock = threading.RLock()
     scheduler._jobs = jobs
     scheduler._scheduler = backend
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
     return scheduler
 
 
diff --git a/tests/test_scheduler_cache_expiry.py b/tests/test_scheduler_cache_expiry.py
index 5c19e0b64..de0a51490 100644
--- a/tests/test_scheduler_cache_expiry.py
+++ b/tests/test_scheduler_cache_expiry.py
@@ -54,13 +54,13 @@ def test_scheduler_initializer_stop_preserves_sync_abi(monkeypatch):
 
     assert scheduler_initializer.stop_scheduler() is None
     scheduler.stop.assert_called_once_with()
-    scheduler.async_stop.assert_not_called()
+    scheduler.stop_async.assert_not_called()
 
 
 def test_scheduler_initializer_stop_awaits_in_running_loop(monkeypatch):
     """生命周期事件循环中的停止入口应返回可等待的异步收口。"""
     scheduler = Mock()
-    scheduler.async_stop = AsyncMock()
+    scheduler.stop_async = AsyncMock()
     monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler))
 
     async def scenario():
@@ -69,7 +69,7 @@ def test_scheduler_initializer_stop_awaits_in_running_loop(monkeypatch):
         await result
 
     asyncio.run(scenario())
-    scheduler.async_stop.assert_awaited_once_with()
+    scheduler.stop_async.assert_awaited_once_with()
     scheduler.stop.assert_not_called()
 
 
@@ -117,6 +117,11 @@ def test_clear_cache_is_manual_only(monkeypatch):
     scheduler._event = threading.Event()
     scheduler._lock = threading.RLock()
     scheduler._jobs = {}
+    scheduler._lifecycle_state = "new"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
     scheduler._agent_task_interruptions_reconciled = True
     scheduler._auth_count = 0
     scheduler._auth_message = False
diff --git a/tests/test_scheduler_lifecycle.py b/tests/test_scheduler_lifecycle.py
new file mode 100644
index 000000000..90b99f5d2
--- /dev/null
+++ b/tests/test_scheduler_lifecycle.py
@@ -0,0 +1,724 @@
+"""Scheduler 任务句柄、generation 与 AgentTask reservation 回归。"""
+
+import asyncio
+import gc
+import threading
+import warnings
+
+import pytest
+
+from app import scheduler as scheduler_module
+from app.runtime.config import global_vars
+from app.scheduler import Scheduler
+
+
+class _ProgressStub:
+    """隔离 scheduler 生命周期测试的同步进度后端。"""
+
+    def __init__(self, _key: str) -> None:
+        """接收进度键但不连接外部后端。"""
+
+    def start(self) -> None:
+        """记录进度开始。"""
+
+    def update(self, **_kwargs) -> None:
+        """忽略中间进度。"""
+
+    def get(self):
+        """返回空的历史进度。"""
+        return None
+
+
+class _AsyncProgressStub:
+    """隔离 scheduler 生命周期测试的异步进度后端。"""
+
+    def __init__(self, _key: str) -> None:
+        """接收进度键但不连接外部后端。"""
+
+    async def get(self):
+        """返回空的历史进度。"""
+        return None
+
+    async def update(self, **_kwargs) -> None:
+        """忽略中间进度。"""
+
+    async def end(self, **_kwargs) -> None:
+        """记录终态但不访问外部缓存。"""
+
+
+def _scheduler(job_id: str, func) -> Scheduler:
+    """构造已启动但不拥有 APScheduler 线程的实例。"""
+    scheduler = object.__new__(Scheduler)
+    scheduler._scheduler = None
+    scheduler._event = threading.Event()
+    scheduler._lock = threading.RLock()
+    scheduler._jobs = {
+        job_id: {
+            "name": "生命周期测试",
+            "provider_name": "测试",
+            "func": func,
+            "running": False,
+            "_generation": 1,
+        }
+    }
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {job_id: 1}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
+    return scheduler
+
+
+@pytest.mark.anyio
+async def test_stop_async_cancels_and_awaits_scheduler_owned_job(monkeypatch) -> None:
+    """关闭后已投递协程必须取消并完成收尾，不得遗留 owner 句柄。"""
+    started = asyncio.Event()
+    cleaned = asyncio.Event()
+
+    async def job():
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cleaned.set()
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub)
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", _AsyncProgressStub)
+    scheduler = _scheduler("lifecycle-job", job)
+
+    assert scheduler.start("lifecycle-job") is True
+    await asyncio.wait_for(started.wait(), timeout=1)
+    assert scheduler._handles
+
+    await scheduler.stop_async()
+
+    assert cleaned.is_set()
+    assert scheduler._jobs["lifecycle-job"]["running"] is False
+    assert scheduler._jobs["lifecycle-job"]["last_error"] == "任务已取消"
+    assert scheduler._handles == {}
+    assert scheduler._lifecycle_state == "stopped"
+
+
+@pytest.mark.anyio
+async def test_foreign_loop_submission_runs_on_main_loop_and_finishes_before_stop(
+        monkeypatch,
+) -> None:
+    """自建事件循环提交的任务仍由应用主循环拥有并完成取消收尾。"""
+    main_loop = asyncio.get_running_loop()
+    started = asyncio.Event()
+    cancelling = asyncio.Event()
+    cleanup_release = asyncio.Event()
+    execution_loop = None
+
+    async def job() -> None:
+        nonlocal execution_loop
+        execution_loop = asyncio.get_running_loop()
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            cancelling.set()
+            await cleanup_release.wait()
+            raise
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub)
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", _AsyncProgressStub)
+    monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", main_loop)
+    scheduler = _scheduler("foreign-loop-job", job)
+
+    def submit_from_foreign_loop() -> bool:
+        async def submit() -> bool:
+            return scheduler.start("foreign-loop-job")
+
+        return asyncio.run(submit())
+
+    assert await asyncio.to_thread(submit_from_foreign_loop) is True
+    await asyncio.wait_for(started.wait(), timeout=1)
+    assert execution_loop is main_loop
+
+    stop_task = asyncio.create_task(scheduler.stop_async())
+    await asyncio.wait_for(cancelling.wait(), timeout=1)
+    await asyncio.sleep(0)
+    assert not stop_task.done()
+    assert scheduler._lifecycle_state == "stopping"
+
+    stop_task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await stop_task
+    assert scheduler._handles
+    assert scheduler._lifecycle_state == "stopping"
+
+    cleanup_release.set()
+
+    async def wait_until_released() -> None:
+        while scheduler._handles:
+            await asyncio.sleep(0)
+
+    await asyncio.wait_for(wait_until_released(), timeout=1)
+    await scheduler.stop_async()
+    assert scheduler._handles == {}
+    assert scheduler._lifecycle_state == "stopped"
+
+
+@pytest.mark.anyio
+async def test_cross_thread_submission_is_registered_before_stop_snapshot(
+        monkeypatch,
+) -> None:
+    """跨线程提交与 owner 登记必须对关闭快照表现为同一原子操作。"""
+    main_loop = asyncio.get_running_loop()
+    registration_entered = threading.Event()
+    registration_release = threading.Event()
+
+    async def job() -> None:
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub)
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", _AsyncProgressStub)
+    monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", main_loop)
+    scheduler = _scheduler("atomic-submit", job)
+    register_handle = scheduler._register_handle
+
+    def delayed_register(**kwargs) -> bool:
+        registration_entered.set()
+        registration_release.wait(timeout=1)
+        return register_handle(**kwargs)
+
+    monkeypatch.setattr(scheduler, "_register_handle", delayed_register)
+    submit_thread = threading.Thread(target=scheduler.start, args=("atomic-submit",))
+    submit_thread.start()
+    assert await asyncio.to_thread(registration_entered.wait, 1)
+
+    stop_result = []
+    stop_thread = threading.Thread(target=lambda: stop_result.append(scheduler._begin_stop()))
+    stop_thread.start()
+    await asyncio.sleep(0.02)
+    assert stop_thread.is_alive()
+
+    registration_release.set()
+    await asyncio.to_thread(submit_thread.join, 1)
+    await asyncio.to_thread(stop_thread.join, 1)
+    assert not submit_thread.is_alive()
+    assert not stop_thread.is_alive()
+    assert len(stop_result[0][1]) == 1
+
+    for handle in stop_result[0][1]:
+        scheduler._cancel_handle(handle)
+    await scheduler._await_cancelled_handles(stop_result[0][1])
+
+
+@pytest.mark.anyio
+async def test_submit_to_loop_tracks_internal_progress_or_finish_tasks() -> None:
+    """进度和收尾协程也必须归 Scheduler 所有并可在关闭时收口。"""
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def pending() -> None:
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    scheduler = _scheduler("internal-task", lambda: None)
+    scheduler._submit_to_loop(
+        pending(),
+        job_id="internal-task",
+        generation=1,
+    )
+
+    await asyncio.wait_for(started.wait(), timeout=1)
+    assert len(scheduler._handles) == 1
+
+    await scheduler.stop_async()
+
+    assert cancelled.is_set()
+    assert scheduler._handles == {}
+
+
+@pytest.mark.anyio
+async def test_sync_job_callback_and_finish_handles_are_owned(monkeypatch) -> None:
+    """同步任务回投的进度与收尾句柄都必须纳入关闭收口。"""
+    update_started = asyncio.Event()
+    finish_started = asyncio.Event()
+    gate = asyncio.Event()
+    cancelled = 0
+
+    class BlockingProgress:
+        """让进度和收尾停在异步后端，便于验证 owner registry。"""
+
+        def __init__(self, _key: str) -> None:
+            pass
+
+        async def update(self, **_kwargs) -> None:
+            nonlocal cancelled
+            update_started.set()
+            try:
+                await gate.wait()
+            except asyncio.CancelledError:
+                cancelled += 1
+                raise
+
+        async def get(self):
+            nonlocal cancelled
+            finish_started.set()
+            try:
+                await gate.wait()
+            except asyncio.CancelledError:
+                cancelled += 1
+                raise
+            return None
+
+        async def end(self, **_kwargs) -> None:
+            pass
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub)
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", BlockingProgress)
+    monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", asyncio.get_running_loop())
+
+    def job(progress_callback) -> None:
+        progress_callback(value=50)
+
+    scheduler = _scheduler("callback-handles", job)
+    await asyncio.to_thread(scheduler.start, "callback-handles")
+    await asyncio.wait_for(
+        asyncio.gather(update_started.wait(), finish_started.wait()),
+        timeout=1,
+    )
+
+    assert len(scheduler._handles) == 2
+
+    await scheduler.stop_async()
+
+    assert cancelled == 2
+    assert scheduler._handles == {}
+
+
+@pytest.mark.anyio
+async def test_stale_progress_cannot_update_replaced_job(monkeypatch) -> None:
+    """旧 generation 的延迟进度不得写入新注册的同 ID 任务。"""
+    updates = []
+
+    class RecordingProgress:
+        def __init__(self, _key: str) -> None:
+            pass
+
+        async def update(self, **kwargs) -> None:
+            updates.append(kwargs)
+
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", RecordingProgress)
+    scheduler = _scheduler("generation-progress", lambda: None)
+    old_job = scheduler._jobs["generation-progress"]
+    callback = scheduler._Scheduler__build_progress_callback(
+        "generation-progress",
+        old_job,
+    )
+    scheduler._jobs["generation-progress"] = {
+        "name": "新一代",
+        "provider_name": "测试",
+        "running": True,
+        "_generation": 2,
+    }
+
+    callback(value=42, text="旧进度")
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+
+    assert updates == []
+    assert scheduler._handles == {}
+
+
+@pytest.mark.anyio
+async def test_replaced_job_keeps_active_state_without_stale_progress(monkeypatch) -> None:
+    """同 ID 新 generation 显示真实运行态，但不继承旧任务进度详情。"""
+    detail = {}
+
+    class RecordingProgress:
+        def __init__(self, _key: str) -> None:
+            pass
+
+        def start(self) -> None:
+            pass
+
+        def update(self, **kwargs) -> None:
+            detail.update(kwargs)
+
+        def get(self):
+            return detail
+
+    class RecordingAsyncProgress:
+        def __init__(self, _key: str) -> None:
+            pass
+
+        async def get(self):
+            return detail
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", RecordingProgress)
+    monkeypatch.setattr(
+        scheduler_module,
+        "AsyncProgressHelper",
+        RecordingAsyncProgress,
+    )
+    scheduler = _scheduler("generation-cache", lambda: None)
+    old_job = scheduler._Scheduler__prepare_job("generation-cache")
+    assert old_job is not None
+    assert detail["data"]["_generation"] == 1
+
+    scheduler._jobs["generation-cache"] = {
+        "name": "新一代",
+        "provider_name": "测试",
+        "running": False,
+        "_generation": 2,
+    }
+
+    progress = scheduler.get_progress("generation-cache")
+    assert progress is not None
+    assert progress.status == "running"
+    assert progress.enable is True
+    assert progress.value == 0
+    assert "_generation" not in progress.data
+    async_progress = await scheduler.aget_progress("generation-cache")
+    assert async_progress is not None
+    assert async_progress.status == "running"
+    assert async_progress.enable is True
+    assert async_progress.value == 0
+    assert "_generation" not in async_progress.data
+
+
+@pytest.mark.anyio
+async def test_stale_generation_cannot_finish_replaced_job(monkeypatch) -> None:
+    """旧 generation 收尾不得改写同 ID 的新任务状态或进度。"""
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", _AsyncProgressStub)
+    scheduler = _scheduler("generation-job", lambda: None)
+    old_job = scheduler._jobs["generation-job"]
+    old_job["running"] = True
+    new_job = {
+        "name": "新一代",
+        "provider_name": "测试",
+        "running": True,
+        "_generation": 2,
+    }
+    scheduler._jobs["generation-job"] = new_job
+
+    await scheduler._Scheduler__finish_job(
+        job_id="generation-job",
+        job=old_job,
+        generation=1,
+        success=True,
+    )
+
+    assert new_job["running"] is True
+    assert "last_finished_at" not in new_job
+    assert old_job["running"] is True
+
+
+def test_agent_task_manual_start_has_single_reservation() -> None:
+    """并发手动触发同一 AgentTask 时只能有一个调用获得 reservation。"""
+    scheduler = _scheduler("agent-task-1", lambda: None)
+    scheduler._jobs["agent-task-1"].update(
+        name="AgentTask",
+        owner="agent",
+    )
+    entered = threading.Event()
+    release = threading.Event()
+    results = []
+
+    def start(*_args, **_kwargs):
+        entered.set()
+        release.wait(timeout=1)
+        return True
+
+    scheduler.start = start
+
+    first = threading.Thread(
+        target=lambda: results.append(scheduler.start_agent_task(1)),
+    )
+    first.start()
+    assert entered.wait(timeout=1)
+    second = scheduler.start_agent_task(1)
+    release.set()
+    first.join(timeout=1)
+
+    assert second is False
+    assert results == [True]
+    assert scheduler._agent_task_reservations == {}
+
+
+def test_scheduler_rejects_new_submission_after_stop() -> None:
+    """进入 stopping/stopped 后不得再从旧 scheduler 提交任务。"""
+    scheduler = _scheduler("stopped-job", lambda: None)
+    scheduler._lifecycle_state = "stopping"
+
+    assert scheduler.start("stopped-job") is False
+    assert scheduler._jobs["stopped-job"]["running"] is False
+
+
+@pytest.mark.anyio
+async def test_config_reload_does_not_restart_scheduler_during_shutdown() -> None:
+    """系统关闭开始后到达的配置事件不得重新打开调度入口。"""
+    scheduler = _scheduler("shutdown-reload", lambda: None)
+    scheduler._lifecycle_state = "stopping"
+    scheduler.init = lambda **_kwargs: pytest.fail("关闭阶段不得重新初始化调度器")
+
+    await scheduler.on_config_changed()
+
+    assert scheduler._lifecycle_state == "stopping"
+
+
+@pytest.mark.anyio
+async def test_concurrent_config_reload_waits_for_old_scheduler_shutdown(
+        monkeypatch,
+) -> None:
+    """并发配置事件合并为一次重建，旧调度线程池结束前不得启动新实例。"""
+    shutdown_started = threading.Event()
+    shutdown_release = threading.Event()
+
+    class BlockingScheduler:
+        running = True
+
+        @staticmethod
+        def remove_all_jobs() -> None:
+            pass
+
+        @staticmethod
+        def shutdown() -> None:
+            shutdown_started.set()
+            shutdown_release.wait(timeout=1)
+
+    scheduler = _scheduler("reload-once", lambda: None)
+    scheduler._scheduler = BlockingScheduler()
+    init_calls = 0
+
+    def init(**_kwargs) -> None:
+        nonlocal init_calls
+        init_calls += 1
+        scheduler._lifecycle_state = "running"
+
+    monkeypatch.setattr(scheduler, "init", init)
+    first = asyncio.create_task(scheduler.on_config_changed())
+    assert await asyncio.to_thread(shutdown_started.wait, 1)
+
+    await scheduler.on_config_changed()
+    assert init_calls == 0
+    assert scheduler._lifecycle_state == "reloading"
+
+    shutdown_release.set()
+    await asyncio.wait_for(first, timeout=1)
+    assert init_calls == 1
+    assert scheduler._lifecycle_state == "running"
+
+
+@pytest.mark.anyio
+async def test_config_reload_preserves_overlap_guard_across_job_generations(
+        monkeypatch,
+) -> None:
+    """热重载替换任务定义后，同 ID 旧任务结束前不得启动新 generation。"""
+    started = asyncio.Event()
+    release = asyncio.Event()
+    finished = asyncio.Event()
+    run_count = 0
+
+    async def job() -> None:
+        nonlocal run_count
+        run_count += 1
+        if run_count == 1:
+            started.set()
+            await release.wait()
+            finished.set()
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub)
+    monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", _AsyncProgressStub)
+    scheduler = _scheduler("reload-overlap", job)
+
+    class ActiveScheduler:
+        """提供列表接口所需的最小 APScheduler 状态。"""
+
+        running = True
+
+        @staticmethod
+        def get_jobs() -> list:
+            """当前用例只关注正在运行任务，不提供后续计划。"""
+            return []
+
+    def init(**_kwargs) -> None:
+        replacement = {
+            "name": "生命周期测试",
+            "provider_name": "测试",
+            "func": job,
+            "running": False,
+        }
+        scheduler._assign_job_generation("reload-overlap", replacement)
+        scheduler._jobs = {"reload-overlap": replacement}
+        scheduler._scheduler = ActiveScheduler()
+        scheduler._lifecycle_state = "running"
+
+    monkeypatch.setattr(scheduler, "init", init)
+
+    assert scheduler.start("reload-overlap") is True
+    await asyncio.wait_for(started.wait(), timeout=1)
+    await scheduler.on_config_changed()
+
+    progress = scheduler.get_progress("reload-overlap")
+    assert progress is not None
+    assert progress.status == "running"
+    assert progress.enable is True
+    listed = scheduler.list()
+    assert len(listed) == 1
+    assert listed[0].id == "reload-overlap"
+    assert listed[0].status == "正在运行"
+    assert scheduler.start("reload-overlap") is False
+    assert run_count == 1
+    assert len(scheduler._handles) == 1
+
+    release.set()
+    await asyncio.wait_for(finished.wait(), timeout=1)
+
+    async def wait_until_released() -> None:
+        while scheduler._active_job_generations or scheduler._handles:
+            await asyncio.sleep(0)
+
+    await asyncio.wait_for(wait_until_released(), timeout=1)
+    assert scheduler.start("reload-overlap") is True
+
+    async def wait_until_second_run_finishes() -> None:
+        while scheduler._active_job_generations or scheduler._handles:
+            await asyncio.sleep(0)
+
+    await asyncio.wait_for(wait_until_second_run_finishes(), timeout=1)
+    assert run_count == 2
+
+
+def test_stop_between_prepare_and_submission_releases_active_generation(
+        monkeypatch,
+) -> None:
+    """关闭插入准备与提交之间时，不得遗留未实际运行的 generation。"""
+    calls = 0
+    scheduler = _scheduler("stop-race", None)
+
+    async def job() -> None:
+        nonlocal calls
+        calls += 1
+
+    scheduler._jobs["stop-race"]["func"] = job
+    original_prepare = scheduler._Scheduler__prepare_job
+
+    def prepare_then_stop(job_id: str):
+        prepared = original_prepare(job_id)
+        scheduler._begin_stop()
+        return prepared
+
+    monkeypatch.setattr(scheduler, "_Scheduler__prepare_job", prepare_then_stop)
+
+    assert scheduler.start("stop-race") is False
+    assert calls == 0
+    assert scheduler._handles == {}
+    assert scheduler._active_job_generations == {}
+    assert scheduler._jobs["stop-race"]["running"] is False
+    assert scheduler._jobs["stop-race"]["last_error"] == "任务未提交"
+
+    monkeypatch.setattr(scheduler, "_Scheduler__prepare_job", original_prepare)
+    scheduler._lifecycle_state = "running"
+    assert scheduler.start("stop-race") is True
+    assert calls == 1
+    assert scheduler._active_job_generations == {}
+
+
+@pytest.mark.anyio
+async def test_cross_thread_rejection_closes_unstarted_business_coroutine(
+        monkeypatch,
+) -> None:
+    """跨线程提交被关闭门禁拒绝时，包装与业务协程都必须释放。"""
+    main_loop = asyncio.get_running_loop()
+    prepared = threading.Event()
+    release = threading.Event()
+    calls = 0
+    scheduler = _scheduler("cross-thread-stop-race", None)
+
+    async def job() -> None:
+        nonlocal calls
+        calls += 1
+
+    scheduler._jobs["cross-thread-stop-race"]["func"] = job
+    monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", main_loop)
+    original_prepare = scheduler._Scheduler__prepare_job
+
+    def prepare_then_wait(job_id: str):
+        result = original_prepare(job_id)
+        prepared.set()
+        release.wait(timeout=1)
+        return result
+
+    monkeypatch.setattr(scheduler, "_Scheduler__prepare_job", prepare_then_wait)
+
+    with warnings.catch_warnings(record=True) as captured:
+        warnings.simplefilter("always", RuntimeWarning)
+        start_task = asyncio.create_task(
+            asyncio.to_thread(scheduler.start, "cross-thread-stop-race")
+        )
+        assert await asyncio.to_thread(prepared.wait, 1)
+        scheduler._begin_stop()
+        release.set()
+        assert await asyncio.wait_for(start_task, timeout=1) is False
+        gc.collect()
+
+    assert calls == 0
+    assert scheduler._active_job_generations == {}
+    assert scheduler._handles == {}
+    assert not any("was never awaited" in str(item.message) for item in captured)
+
+
+def test_cancelled_cross_thread_proxy_waits_for_target_loop_cleanup(
+        monkeypatch,
+) -> None:
+    """跨线程代理提前取消后，真实完成信号必须等待目标循环清理。"""
+    target_loop = asyncio.new_event_loop()
+    loop_blocked = threading.Event()
+    loop_release = threading.Event()
+    loop_drained = threading.Event()
+    loop_errors = []
+    loop_thread = threading.Thread(target=target_loop.run_forever)
+    loop_thread.start()
+
+    def block_target_loop() -> None:
+        loop_blocked.set()
+        loop_release.wait(timeout=1)
+
+    target_loop.set_exception_handler(
+        lambda _loop, context: loop_errors.append(context)
+    )
+    target_loop.call_soon_threadsafe(block_target_loop)
+    assert loop_blocked.wait(timeout=1)
+
+    scheduler = _scheduler("cancel-before-start", None)
+    calls = 0
+
+    async def business() -> None:
+        nonlocal calls
+        calls += 1
+
+    scheduler._jobs["cancel-before-start"]["func"] = business
+    monkeypatch.setattr(global_vars, "CURRENT_EVENT_LOOP", target_loop)
+
+    with warnings.catch_warnings(record=True) as captured:
+        warnings.simplefilter("always", RuntimeWarning)
+        try:
+            assert scheduler.start("cancel-before-start") is True
+            scheduler_handle = next(iter(scheduler._handles.values()))
+            scheduler._cancel_handle(scheduler_handle)
+            assert not scheduler_handle.completion.done()
+
+            loop_release.set()
+            target_loop.call_soon_threadsafe(loop_drained.set)
+            assert loop_drained.wait(timeout=1)
+            assert scheduler_handle.completion.done()
+            gc.collect()
+        finally:
+            target_loop.call_soon_threadsafe(target_loop.stop)
+            loop_thread.join(timeout=1)
+            target_loop.close()
+
+    assert calls == 0
+    assert loop_errors == []
+    assert scheduler._active_job_generations == {}
+    assert scheduler._handles == {}
+    assert not any("was never awaited" in str(item.message) for item in captured)
diff --git a/tests/test_scheduler_progress.py b/tests/test_scheduler_progress.py
index 465ecf7ae..ea5c9486f 100644
--- a/tests/test_scheduler_progress.py
+++ b/tests/test_scheduler_progress.py
@@ -14,8 +14,6 @@ def _build_scheduler(job_id, func):
     scheduler._scheduler = None
     scheduler._event = threading.Event()
     scheduler._lock = threading.RLock()
-    scheduler._async_tasks = set()
-    scheduler._accepting_async_tasks = True
     scheduler._jobs = {
         job_id: {
             "name": "测试定时服务",
@@ -24,6 +22,11 @@ def _build_scheduler(job_id, func):
             "running": False,
         }
     }
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
     return scheduler
 
 
@@ -161,7 +164,7 @@ def test_scheduler_records_cancelled_async_job_as_failed():
     async def run_task():
         job = scheduler._Scheduler__prepare_job(job_id)
         with pytest.raises(asyncio.CancelledError):
-            await scheduler._Scheduler__run_coro_job(task(), job_id, job)
+            await scheduler._Scheduler__run_coro_job(task, job_id, job)
 
     scheduler = _build_scheduler(job_id, task)
     asyncio.run(run_task())
@@ -173,7 +176,7 @@ def test_scheduler_records_cancelled_async_job_as_failed():
     assert progress.error == "任务已取消"
 
 
-def test_scheduler_async_stop_cancels_owned_async_jobs():
+def test_scheduler_stop_async_cancels_owned_async_jobs():
     """Scheduler 关停应取消并等待自身登记的异步作业。"""
     job_id = f"test-owned-task-{uuid4()}"
     started = asyncio.Event()
@@ -194,13 +197,13 @@ def test_scheduler_async_stop_cancels_owned_async_jobs():
         """在当前事件循环启动并收口异步作业。"""
         scheduler.start(job_id)
         await started.wait()
-        assert len(scheduler._async_tasks) == 1
-        await scheduler.async_stop(timeout_seconds=1)
+        assert len(scheduler._handles) == 1
+        await scheduler.stop_async()
 
     asyncio.run(run_task())
 
     assert cancelled.is_set()
-    assert scheduler._async_tasks == set()
+    assert scheduler._handles == {}
 
 
 def test_scheduler_returns_none_for_unknown_job():
@@ -209,5 +212,10 @@ def test_scheduler_returns_none_for_unknown_job():
     scheduler = object.__new__(Scheduler)
     scheduler._lock = threading.RLock()
     scheduler._jobs = {}
+    scheduler._lifecycle_state = "running"
+    scheduler._handles = {}
+    scheduler._job_generations = {}
+    scheduler._active_job_generations = {}
+    scheduler._agent_task_reservations = {}
 
     assert scheduler.get_progress(job_id) is None
diff --git a/tests/test_site_reset_scheduler.py b/tests/test_site_reset_scheduler.py
new file mode 100644
index 000000000..c12820284
--- /dev/null
+++ b/tests/test_site_reset_scheduler.py
@@ -0,0 +1,60 @@
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from app.api.endpoints import site as site_endpoint
+from app.application.site.mutation import SiteMutationResult
+from app.runtime.tasks import TaskRegistry
+
+
+class _TaskRegistry(TaskRegistry):
+    """记录站点重置提交的同步后台任务。"""
+
+    def __init__(self) -> None:
+        """初始化任务记录。"""
+        super().__init__()
+        self.calls: list[tuple] = []
+
+    def create_sync(self, function, *args, owner: str, **kwargs) -> None:
+        """保存函数、参数和 owner，避免端点测试启动真实任务。"""
+        self.calls.append((function, args, kwargs, owner))
+
+
+@pytest.mark.asyncio
+async def test_reset_submits_cookiecloud_after_site_transaction(monkeypatch):
+    """站点重置提交 CookieCloud 后台任务，不在请求事件循环内直接执行。"""
+    command = Mock()
+    command.reset = AsyncMock(return_value=SiteMutationResult(success=True))
+    task_registry = _TaskRegistry()
+    scheduler = Mock()
+    system_config = Mock()
+    system_config.async_set = AsyncMock()
+
+    monkeypatch.setattr(site_endpoint, "Scheduler", Mock(return_value=scheduler))
+    monkeypatch.setattr(
+        site_endpoint,
+        "get_configured_system_config",
+        Mock(return_value=system_config),
+    )
+
+    response = await site_endpoint.reset(
+        task_registry=task_registry,
+        command=command,
+        _=Mock(),
+    )
+
+    assert response.success is True
+    command.reset.assert_awaited_once_with()
+    scheduler.start.assert_not_called()
+    system_config.async_set.assert_any_await(
+        site_endpoint.SystemConfigKey.IndexerSites, []
+    )
+    system_config.async_set.assert_any_await(site_endpoint.SystemConfigKey.RssSites, [])
+    assert task_registry.calls == [
+        (
+            scheduler.start,
+            (),
+            {"job_id": "cookiecloud", "manual": True},
+            "api.site.reset",
+        )
+    ]