fix(scheduler): read the lease owner under SQLite's writer lock (#5777)

* fix(scheduler): read the lease owner under SQLite's writer lock

release_dispatch_lease guards on the lease owner and then writes the
status its caller asked for — "enabled" for a scheduled dispatch. It read
the row with session.get(..., with_for_update=True), and FOR UPDATE is
ignored on SQLite, so that guard ran against whatever snapshot the plain
SELECT saw.

A user pausing a task while its dispatch was in flight therefore lost the
pause: pause clears lease_owner, which is exactly what the guard checks,
but the release had already read the pre-pause row, passed the check, and
overwrote status with "enabled". The API had answered "paused", the task
row went back to enabled with next_run_at untouched, and the scheduler
kept firing it.

The read now goes through _lock_task, the helper every other mutating
path here uses, which takes the writer up front on SQLite (FOR UPDATE
keeps doing the work on Postgres, where a pause that commits first is
seen and the release declines). The reconciliation paths in this file
keep their documented "no SQLite writer emulation" trade-off.

* docs(changelog): note the scheduler lease-release lock fix (#5777)

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -941,6 +941,14 @@ This release closes that milestone with **765 merged pull requests**.
 
 ### Fixed
 
+- **scheduler:** Pausing a scheduled task no longer loses the pause when a
+  dispatch is in flight on SQLite. `release_dispatch_lease` guards on the lease
+  owner — which pausing clears — but read the row without taking SQLite's
+  writer, so a stale read passed the guard and wrote the task back to
+  `enabled` with `next_run_at` untouched, leaving the scheduler firing a task
+  the API had reported as paused. The read now takes the writer first, as every
+  other mutating path in that repository does. PostgreSQL was unaffected.
+  ([#5777])
 - **uploads:** Deleting an uploaded document no longer deletes the converted
   Markdown beside it. Conversion names a companion after the document's stem
   and falls back to a `_N` suffix when that name is taken, so the `.md` next to
@@ -4379,3 +4387,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#5673]: https://github.com/bytedance/deer-flow/pull/5673
 [#5734]: https://github.com/bytedance/deer-flow/pull/5734
 [#5776]: https://github.com/bytedance/deer-flow/pull/5776
+[#5777]: https://github.com/bytedance/deer-flow/pull/5777
+

CHANGELOG_zh.md

@@ -727,6 +727,11 @@
 
 ### 修复
 
+- **调度器：** 在 SQLite 上，调度分发进行中暂停计划任务时不再丢失暂停状态。
+  `release_dispatch_lease` 依据租约持有者做校验（暂停会清除该字段），但读取任务行时没有先获取
+  SQLite 的写锁，因此过期的读取会通过校验，并把任务状态写回 `enabled` 且不改动 `next_run_at`，
+  导致接口已回复“已暂停”的任务仍被继续触发。现在该读取会像该仓储中其他写入路径一样先获取写锁。
+  PostgreSQL 不受影响。([#5777])
 - **上传：** 删除已上传的文档时，不再连带删除其旁边转换生成的 Markdown。转换以文档主干名
   命名配套文件，名称被占用时回退为 `_N` 后缀，因此文档旁的 `.md` 可能属于主干名相同的另一个
   文档，或属于用户自己：上传 `a.docx` 与 `a.pdf` 会生成 `a.md` 与 `a_1.md`，删除 `a.pdf`
@@ -3555,3 +3560,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#5673]: https://github.com/bytedance/deer-flow/pull/5673
 [#5734]: https://github.com/bytedance/deer-flow/pull/5734
 [#5776]: https://github.com/bytedance/deer-flow/pull/5776
+[#5777]: https://github.com/bytedance/deer-flow/pull/5777

backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py

@@ -348,9 +348,16 @@ class ScheduledTaskRepository:
         expected_lease_owner: str | None,
         status: str,
     ) -> bool:
-        """Release the short due-task claim after its occurrence is queued."""
+        """Release the short due-task claim after its occurrence is queued.
+
+        The lease-owner guard is only as fresh as the row it reads, so the
+        read takes the writer first (``_lock_task``): on SQLite a plain
+        ``SELECT`` sees a pre-pause snapshot, and the unconditional
+        ``row.status = status`` below would then write the caller's
+        ``"enabled"`` over a pause that had already committed.
+        """
         async with self._sf() as session:
-            row = await session.get(ScheduledTaskRow, task_id, with_for_update=True)
+            row = await self._lock_task(session, task_id)
             if row is None:
                 return False
             if expected_lease_owner is not None and row.lease_owner != expected_lease_owner:

backend/tests/test_scheduled_task_dispatch_race.py

@@ -69,7 +69,31 @@ def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService:
     )
 
 
-async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str) -> dict:
+class _PausingTaskRepo(ScheduledTaskRepository):
+    """Real repository that parks the first task-row lock it takes.
+
+    The pause transaction stops after acquiring the row (holding SQLite's
+    writer) and before committing, so a dispatch release genuinely runs while
+    a pause is in flight — the window in which the release used to read a
+    stale ``lease_owner`` and write the task back to ``enabled``.
+    """
+
+    def __init__(self, session_factory, *, holding: asyncio.Event, resume: asyncio.Event) -> None:
+        super().__init__(session_factory)
+        self._holding = holding
+        self._resume = resume
+        self._parked = False
+
+    async def _lock_task(self, session, task_id):  # type: ignore[override]
+        row = await ScheduledTaskRepository._lock_task(session, task_id)
+        if not self._parked:
+            self._parked = True
+            self._holding.set()
+            await self._resume.wait()
+        return row
+
+
+async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str, *, next_run_at: datetime | None = None) -> dict:
     # fresh_thread_per_run: every dispatch gets a NEW thread_id, so #4003's
     # per-thread uq_runs_thread_active can never fire for two dispatches of the
     # same task — this is precisely the gap the per-task index closes.
@@ -84,7 +108,7 @@ async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str) -> dict:
         schedule_type="cron",
         schedule_spec={"cron": "*/5 * * * *"},
         timezone="UTC",
-        next_run_at=None,
+        next_run_at=next_run_at,
     )
     task = await task_repo.get(task_id, user_id="user-1")
     assert task is not None
@@ -217,3 +241,40 @@ async def test_partial_unique_index_enforces_one_active_run_per_task(tmp_path):
         assert await run_repo.has_active_runs("t1") is True
     finally:
         await close_engine()
+
+
+async def test_release_dispatch_lease_does_not_revive_a_paused_task(tmp_path):
+    """A pause landing mid-dispatch must survive the scheduler's lease release.
+
+    ``release_dispatch_lease`` writes the status its caller asked for
+    (``"enabled"`` for scheduled dispatch) and only guards on the lease owner.
+    Reading that owner outside SQLite's writer lock let the release pass its
+    guard against a pre-pause snapshot and then overwrite the user's pause.
+    """
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        sf = get_session_factory()
+        assert sf is not None
+        holding, resume = asyncio.Event(), asyncio.Event()
+        task_repo = _PausingTaskRepo(sf, holding=holding, resume=resume)
+        now = datetime.now(UTC)
+        await _seed_task(task_repo, "task-pause-race", next_run_at=now)
+        claimed = await task_repo.claim_due_tasks(now=now, lease_owner="scheduler-A", lease_seconds=120, limit=10)
+        assert [row["id"] for row in claimed] == ["task-pause-race"]
+
+        pause = asyncio.create_task(task_repo.pause_with_queue_cancellation("task-pause-race", user_id="user-1", error="paused by user", now=now))
+        await asyncio.wait_for(holding.wait(), timeout=5)
+
+        release = asyncio.create_task(task_repo.release_dispatch_lease("task-pause-race", expected_lease_owner="scheduler-A", status="enabled"))
+        await asyncio.sleep(0.2)  # the release is in flight while the pause holds the row
+        resume.set()
+
+        assert await asyncio.wait_for(pause, timeout=10) == "paused"
+        assert await asyncio.wait_for(release, timeout=10) is False
+
+        task = await task_repo.get("task-pause-race", user_id="user-1")
+        assert task is not None
+        assert task["status"] == "paused", "the user's pause must not be reverted by the lease release"
+        assert task["lease_owner"] is None
+    finally:
+        await close_engine()