fix(scheduler): preserve completed task outcome on stop (#6418)

app/scheduler.py

@@ -917,9 +917,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         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)
         try:
+            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={
@@ -1198,17 +1198,21 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                 and not target_loop.is_closed()
             )
             if running_loop and (not target_loop_available or running_loop is target_loop):
+                started = threading.Event()
+
+                async def run_owned_job() -> None:
+                    started.set()
+                    await self.__run_coro_job(
+                        coro_factory=coro_factory,
+                        job_id=job_id,
+                        job=job,
+                        generation=generation,
+                    )
+
                 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,
-                        ),
-                    )
+                    handle = running_loop.create_task(run_owned_job())
                     registered = self._register_handle(
                         job_id=job_id,
                         generation=generation,
@@ -1219,7 +1223,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
                     def _finish_cancelled_before_start(
                             submitted: asyncio.Future[Any],
                     ) -> None:
-                        if submitted.cancelled():
+                        if submitted.cancelled() and not started.is_set():
                             self._finish_unsubmitted_job(
                                 job_id=job_id,
                                 job=job,

tests/test_scheduler_lifecycle.py

@@ -99,6 +99,56 @@ async def test_stop_async_cancels_and_awaits_scheduler_owned_job(monkeypatch) ->
     assert scheduler._lifecycle_state == "stopped"
 
 
+@pytest.mark.anyio
+async def test_stop_during_final_progress_does_not_mark_completed_job_unsubmitted(
+        monkeypatch,
+) -> None:
+    """业务协程已完成后，取消最终进度写入不得改写任务执行结果。"""
+    finish_started = asyncio.Event()
+
+    class BlockingFinishProgress(_AsyncProgressStub):
+        """把任务停在最终进度读取阶段。"""
+
+        async def get(self):
+            finish_started.set()
+            await asyncio.Event().wait()
+
+    async def job() -> None:
+        return None
+
+    monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub)
+    monkeypatch.setattr(
+        scheduler_module,
+        "AsyncProgressHelper",
+        BlockingFinishProgress,
+    )
+    scheduler = _scheduler("final-progress-stop", job)
+
+    assert scheduler.start("final-progress-stop") is True
+    await asyncio.wait_for(finish_started.wait(), timeout=1)
+
+    await scheduler.stop_async()
+
+    assert scheduler._jobs["final-progress-stop"]["running"] is False
+    assert scheduler._jobs["final-progress-stop"]["last_error"] is None
+    assert scheduler._handles == {}
+    assert scheduler._active_job_generations == {}
+
+    monkeypatch.setattr(
+        scheduler_module,
+        "AsyncProgressHelper",
+        _AsyncProgressStub,
+    )
+    scheduler._lifecycle_state = "running"
+    assert scheduler.start("final-progress-stop") is True
+
+    async def wait_until_finished() -> None:
+        while scheduler._handles or scheduler._active_job_generations:
+            await asyncio.sleep(0)
+
+    await asyncio.wait_for(wait_until_finished(), timeout=1)
+
+
 @pytest.mark.anyio
 async def test_foreign_loop_submission_runs_on_main_loop_and_finishes_before_stop(
         monkeypatch,