fix(scheduler): close async task lifecycle

app/scheduler.py

@@ -1,4 +1,6 @@
 import asyncio
+from concurrent.futures import CancelledError as ConcurrentCancelledError
+from concurrent.futures import Future as ConcurrentFuture
 import gc
 import hashlib
 import inspect
@@ -129,6 +131,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         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:
         """
@@ -250,6 +255,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         config = get_scheduler_runtime_config()
         # 停止定时服务
         self.stop()
+        self._accepting_async_tasks = True
 
         # 调试模式不启动定时服务
         if config.dev:
@@ -558,6 +564,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         """
         准备定时任务
         """
+        if not getattr(self, "_accepting_async_tasks", True):
+            return None
         started_at = self._format_time()
         with self._lock:
             job = self._jobs.get(job_id)
@@ -822,6 +830,38 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             # 协程收尾在事件循环上完成，同步路径（线程池/调用线程）提交到事件循环执行
             await self.__finish_job(job_id=job_id, 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:
         """
         启动定时服务
@@ -837,12 +877,18 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 running_loop = None
             target_loop = global_vars.loop
             if running_loop:
-                asyncio.create_task(self.__run_coro_job(coro=coro, job_id=job_id, job=job))
-                return True
+                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():
-                asyncio.run_coroutine_threadsafe(
-                    self.__run_coro_job(coro=coro, job_id=job_id, job=job),
-                    target_loop,
+                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,
+                    )
                 )
                 return True
             asyncio.run(coro)
@@ -895,8 +941,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                     job_id=job_id, success=success, error=error
                 ))
 
-    @staticmethod
-    def _submit_to_loop(coro: Any) -> None:
+    def _submit_to_loop(self, coro: Any) -> None:
         """
         把协程提交到事件循环执行，兼容以下调用环境：
         - 已在事件循环内（async 任务内部）：排队为独立任务，避免阻塞
@@ -908,12 +953,55 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         except RuntimeError:
             running_loop = None
         if running_loop:
-            asyncio.create_task(coro)
+            self._create_async_task(coro)
         elif global_vars.loop and global_vars.loop.is_running():
-            asyncio.run_coroutine_threadsafe(coro, global_vars.loop)
+            if not getattr(self, "_accepting_async_tasks", True):
+                coro.close()
+                return
+            self._track_async_task(
+                asyncio.run_coroutine_threadsafe(coro, global_vars.loop)
+            )
         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()
+
     @staticmethod
     def _get_agent_task_job_id(task_id: int) -> str:
         """生成 Agent 自主定时任务的调度器 Job ID。"""
@@ -1396,6 +1484,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         """
         关闭定时服务
         """
+        self._cancel_async_tasks()
         with lock:
             try:
                 if self._scheduler:

app/startup/scheduler_initializer.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from app.application.scheduling import register_scheduler_class
 from app.scheduler import Scheduler
 
@@ -14,9 +16,15 @@ def init_scheduler():
 
 def stop_scheduler():
     """
-    停止定时器
+    停止定时器；生命周期事件循环中返回有限等待的兼容协程。
     """
-    Scheduler().stop()
+    scheduler = Scheduler()
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        scheduler.stop()
+        return None
+    return scheduler.async_stop()
 
 
 def restart_scheduler():

docs/adr/0007-background-action-reliability.md

@@ -86,6 +86,8 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同
 - IMDb 同步 `clear_cache()` ABI 在事件循环内触发的异步缓存清理登记为
   `module.imdb.cache_clear`；同步调用方式和无运行事件循环时的立即清理行为保持不变，宿主关停后不再
   接受新的清理任务。
+- Scheduler 的协程作业与异步进度收尾由 Scheduler 自有任务集合持有；同步 `start()` / `stop()` ABI 保持，
+  生命周期关闭入口额外等待有限预算，跨线程提交的 Future 也会在停止时收到取消请求。
 
 ### Transfer pending / 文件整理
 

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

@@ -838,6 +838,9 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas
   原语义；未启动完整 lifespan 的协议校验和旧直接调用通过兼容依赖回退到默认登记器。
 - IMDb 同步清缓存兼容入口在运行事件循环时改由 TaskRegistry 登记异步缓存清理任务，owner 为
   `module.imdb.cache_clear`；同步签名、模块调用方式和无事件循环时的立即清理语义保持不变。
+- Scheduler 的协程作业和异步进度收尾不再使用无主 `create_task` 或丢弃跨线程 Future；由 Scheduler 自有
+  任务集合登记、停止时取消，生命周期入口通过异步兼容包装器在有限预算内等待收口，保留旧同步
+  `Scheduler.start()` / `Scheduler.stop()` 与插件调度 ABI。
 
 #### ARCH-251：用现有数据库做首个 durable side-effect pilot
 

tests/test_scheduler_cache_expiry.py

@@ -1,5 +1,6 @@
+import asyncio
 import threading
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 
 from app import scheduler as scheduler_module
 from app.scheduler import Scheduler
@@ -46,6 +47,32 @@ def test_scheduler_initializer_starts_background_jobs(monkeypatch):
     scheduler.init.assert_called_once_with()
 
 
+def test_scheduler_initializer_stop_preserves_sync_abi(monkeypatch):
+    """同步调用停止入口仍应立即关闭 Scheduler，不返回 coroutine。"""
+    scheduler = Mock()
+    monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler))
+
+    assert scheduler_initializer.stop_scheduler() is None
+    scheduler.stop.assert_called_once_with()
+    scheduler.async_stop.assert_not_called()
+
+
+def test_scheduler_initializer_stop_awaits_in_running_loop(monkeypatch):
+    """生命周期事件循环中的停止入口应返回可等待的异步收口。"""
+    scheduler = Mock()
+    scheduler.async_stop = AsyncMock()
+    monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler))
+
+    async def scenario():
+        result = scheduler_initializer.stop_scheduler()
+        assert result is not None
+        await result
+
+    asyncio.run(scenario())
+    scheduler.async_stop.assert_awaited_once_with()
+    scheduler.stop.assert_not_called()
+
+
 def test_clear_cache_is_manual_only(monkeypatch):
     """缓存清理任务应仅手动执行，不注册到调度器自动运行。"""
     background_scheduler = _BackgroundSchedulerStub()

tests/test_scheduler_progress.py

@@ -11,7 +11,11 @@ from app.scheduler import Scheduler
 def _build_scheduler(job_id, func):
     """构造不启动 APScheduler 的定时服务测试对象。"""
     scheduler = object.__new__(Scheduler)
+    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": "测试定时服务",
@@ -169,6 +173,36 @@ def test_scheduler_records_cancelled_async_job_as_failed():
     assert progress.error == "任务已取消"
 
 
+def test_scheduler_async_stop_cancels_owned_async_jobs():
+    """Scheduler 关停应取消并等待自身登记的异步作业。"""
+    job_id = f"test-owned-task-{uuid4()}"
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def task():
+        """等待关停信号，验证任务确实由 Scheduler 持有。"""
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    scheduler = _build_scheduler(job_id, task)
+
+    async def run_task():
+        """在当前事件循环启动并收口异步作业。"""
+        scheduler.start(job_id)
+        await started.wait()
+        assert len(scheduler._async_tasks) == 1
+        await scheduler.async_stop(timeout_seconds=1)
+
+    asyncio.run(run_task())
+
+    assert cancelled.is_set()
+    assert scheduler._async_tasks == set()
+
+
 def test_scheduler_returns_none_for_unknown_job():
     """未注册且无历史进度的定时服务应返回空。"""
     job_id = f"test-unknown-{uuid4()}"