fix(scheduler):修复单个定时任务派发异常中断整批执行 (#8047)

* fix(scheduler): isolate queued run drain failures

* fix(scheduler): isolate queue timeout rows

* fix(scheduler): enforce queue timeout at launch claim

Keep expired queued occurrences ineligible to start when the timeout sweep fails. Cover per-row and whole-scan cleanup failures with SQLite service tests that also verify healthy queued and due work still launches.

* fix(scheduler): exclude expired rows from queue selection

Filter rows past the queue deadline before owner ranking and the fair batch limit so failed timeout cleanup cannot starve healthy waiting work. Keep the atomic claim cutoff as the final launch guard.

---------

Co-authored-by: AochenShen99 <sac1063467113@hotmail.com>
Co-authored-by: Nan Gao <thunderggnn@gmail.com>

CHANGELOG.md

@@ -237,6 +237,14 @@ This section accumulates work toward the **2.3.0** milestone
   future per-record raise costs one tick instead of every other claimed task.
   ([#7998])
 
+- **scheduler:** Exceptions escaping the queue-timeout sweep or the
+  queued-occurrence drain are now contained to that occurrence. The
+  scheduler logs the failed row and continues with the remaining queue
+  and the tasks claimed in the same poll. Expired rows are excluded before
+  owner ranking and the batch limit, and a launch claim enforces the same
+  cutoff, so a row left queued by a failed sweep neither starts nor crowds
+  healthy work out of the queue while timeout recovery retries it. ([#7998])
+
 - **scheduler:** One scheduled task whose saved schedule can no longer be
   computed no longer stops every other task. When a stored timezone or cron
   expression stopped working (a zone missing from the time zone database

CHANGELOG_zh.md

@@ -77,6 +77,7 @@
   用户消息，因此一次 100 KB 的粘贴会让每次模型调用阻塞所有线程约 6 秒。现在扫描为线性耗时，
   且脱敏的地址与此前完全相同。([#8006])
 
+- **调度器：** 队列超时扫描或队列排空中的单条记录出现异常时，调度器会记录该失败并继续处理剩余队列，以及同一轮启动的到期任务。过期记录会在所有者排名和批次限额之前被排除，启动领取还会再次检查同一截止时间，因此超时清理失败时遗留的排队记录既不会启动，也不会挤占正常排队工作的候选位置，只会继续等待超时恢复流程处理。([#7998])
 - **调度器：** 一个已保存的计划无法再计算的定时任务，不再让其他所有任务停止运行。此前
   已保存的时区或 cron 表达式失效时（例如升级后时区数据库中缺少该时区，或新版 croniter
   不再接受该表达式），在调度器轮询中计算其下一次运行时间会抛出异常；其启动租约过期后，
@@ -12899,6 +12900,7 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7981]: https://github.com/bytedance/deer-flow/pull/7981
 [#7992]: https://github.com/bytedance/deer-flow/pull/7992
 [#7993]: https://github.com/bytedance/deer-flow/pull/7993
+[#7998]: https://github.com/bytedance/deer-flow/issues/7998
 [#8006]: https://github.com/bytedance/deer-flow/pull/8006
 [#8019]: https://github.com/bytedance/deer-flow/pull/8019
 [#8043]: https://github.com/bytedance/deer-flow/pull/8043

backend/AGENTS.md

@@ -14,6 +14,7 @@ Read the guide for an area before you change it.
 | --- | --- |
 | Run streaming, checkpoints, run ownership | [runtime](packages/harness/deerflow/runtime/AGENTS.md) |
 | Repositories, leases, scheduled tasks, schema changes | [persistence](packages/harness/deerflow/persistence/AGENTS.md), [migrations](packages/harness/deerflow/persistence/migrations/AGENTS.md) |
+| Scheduler failure isolation | [design conventions](app/scheduler/guide/failure-isolation.md) |
 | `config.yaml`, `extensions_config.json` | [config](packages/harness/deerflow/config/AGENTS.md) |
 | Lead agent, prompt, thread state, middleware chain | [agents](packages/harness/deerflow/agents/AGENTS.md), [middlewares](packages/harness/deerflow/agents/middlewares/AGENTS.md) |
 | Memory, subagents, skills, tools | [memory](packages/harness/deerflow/agents/memory/AGENTS.md), [subagents](packages/harness/deerflow/subagents/AGENTS.md), [skills](packages/harness/deerflow/skills/AGENTS.md), [tools](packages/harness/deerflow/tools/AGENTS.md) |

backend/app/scheduler/guide/failure-isolation.md

@@ -0,0 +1,26 @@
+# Scheduler failure isolation
+
+Scheduled work shares one polling cycle. Treat an individual occurrence as the
+failure boundary so one damaged record cannot repeatedly prevent unrelated
+owners' work from progressing. Apply this convention to maintenance passes as
+well as launches: failing to clean up an older occurrence is not a reason to
+withhold admission from a healthy task.
+
+An exception during dispatch does not establish whether Gateway accepted the
+run. Leave that uncertainty to durable queue and lease recovery. Eagerly
+clearing a lease or starting a replacement from the exception handler can
+duplicate work that already started. Cleanup is appropriate only when the
+operation's outcome and ownership are known.
+
+Containment preserves progress, not success. Keep failures observable and let
+normal recovery retry the affected occurrence. Apply queue age before building
+the fair drain batch: expired rows must not consume owner ranks or batch slots,
+or a failed timeout sweep can starve healthy work indefinitely. Queue timeout is
+enforced again when a waiting row is claimed for launch. If timeout cleanup
+fails, the row remains queued for the next recovery pass but is no longer
+eligible to start; cleanup failure must never turn expired work into a live run.
+Failure to read the queue or claim due work still prevents that stage from
+progressing; the next poll retries it.
+
+For user-visible behavior and deployment requirements, see the
+[scheduled tasks guide](../../../../docs/guide/scheduled-tasks.md).

backend/app/scheduler/service.py

@@ -468,7 +468,13 @@ class ScheduledTaskService:
                 # failure (e.g. the candidate query) must not skip admission or
                 # queue draining for this poll.
                 logger.exception("Failed to reconcile terminal scheduled runs")
-        await self._expire_waiting_runs(now=now)
+        try:
+            await self._expire_waiting_runs(now=now)
+        except Exception:
+            # A failed timeout sweep must not skip queue draining or this
+            # poll's due-task admission. Individual rows are also isolated by
+            # expire_queued_runs; this guard covers the candidate query/session.
+            logger.exception("Failed to expire queued scheduled runs")
         await self._drain_queue(now=now)
         # Admission and execution capacity are separate. Due occurrences are
         # persisted even when all execution slots are busy; claim_queued_run()
@@ -704,6 +710,7 @@ class ScheduledTaskService:
         queued: dict[str, Any],
         *,
         now: datetime,
+        created_after: datetime | None = None,
     ) -> dict[str, Any]:
         """Turn one queued occurrence into a live run under its own trace scope.
 
@@ -712,9 +719,17 @@ class ScheduledTaskService:
         sharing one id across a whole poll cycle. A manual trigger arrives
         inside a Gateway request and keeps that request's trace instead, so
         the launched run stays correlated with the call that asked for it.
+
+        ``created_after`` applies the queue-age cutoff when retrying a row
+        from the waiting queue. A freshly admitted row has not waited.
         """
         with ensure_trace_context():
-            return await self._launch_queued_occurrence(task, queued, now=now)
+            return await self._launch_queued_occurrence(
+                task,
+                queued,
+                now=now,
+                created_after=created_after,
+            )
 
     async def _launch_queued_occurrence(
         self,
@@ -722,6 +737,7 @@ class ScheduledTaskService:
         queued: dict[str, Any],
         *,
         now: datetime,
+        created_after: datetime | None = None,
     ) -> dict[str, Any]:
         task_run_id = queued["id"]
         execution_thread_id = queued["thread_id"]
@@ -733,6 +749,7 @@ class ScheduledTaskService:
             lease_seconds=self._lease_seconds,
             global_max_concurrent_runs=self._max_concurrent_runs,
             per_user_max_concurrent_runs=self._max_concurrent_runs_per_user,
+            created_after=created_after,
         )
         if claimed is None:
             return self._queued_result(task_run_id, execution_thread_id)
@@ -1026,31 +1043,52 @@ class ScheduledTaskService:
         queued_rows = await self._task_run_repo.list_queued_runs(
             limit=max(16, self._max_concurrent_runs * 4),
             per_user_max_concurrent_runs=self._max_concurrent_runs_per_user,
+            created_after=now - timedelta(seconds=self._queue_timeout_seconds),
         )
         for queued in queued_rows:
-            await self._task_repo.release_queued_admission_lease(queued["task_id"])
-            task = await self._task_repo.get_internal(queued["task_id"])
-            if task is None:
-                await self._task_run_repo.update_status(
-                    queued["id"],
-                    status="interrupted",
-                    error=RUN_ERROR_DELETED_WHILE_QUEUED,
-                    finished_at=now,
+            try:
+                await self._drain_queued_run(queued, now=now)
+            except Exception:
+                # A storage failure on one occurrence must not leave every
+                # later queued row and this poll's due tasks undispatched. The
+                # failed row keeps its queue/lease state for the next recovery
+                # pass, and queue_timeout still bounds how long it can wait.
+                logger.exception(
+                    "Failed to drain scheduled occurrence %s; continuing with the remaining queue",
+                    queued.get("id"),
                 )
-                continue
-            # Pausing suppresses automatic occurrences, but a manual trigger is
-            # an explicit request and has always been allowed to run without
-            # resuming the schedule. A later pause still cancels an already
-            # queued manual row atomically in pause_with_queue_cancellation().
-            if task.get("status") == "paused" and queued["trigger"] != "manual":
-                await self._task_run_repo.update_status(
-                    queued["id"],
-                    status="interrupted",
-                    error=RUN_ERROR_PAUSED_WHILE_QUEUED,
-                    finished_at=now,
-                )
-                continue
-            await self._attempt_queued_run(task, queued, now=now)
+
+    async def _drain_queued_run(self, queued: dict[str, Any], *, now: datetime) -> None:
+        await self._task_repo.release_queued_admission_lease(queued["task_id"])
+        task = await self._task_repo.get_internal(queued["task_id"])
+        if task is None:
+            await self._task_run_repo.update_status(
+                queued["id"],
+                status="interrupted",
+                error=RUN_ERROR_DELETED_WHILE_QUEUED,
+                finished_at=now,
+            )
+            return
+        # Pausing suppresses automatic occurrences, but a manual trigger is
+        # an explicit request and has always been allowed to run without
+        # resuming the schedule. A later pause still cancels an already
+        # queued manual row atomically in pause_with_queue_cancellation().
+        if task.get("status") == "paused" and queued["trigger"] != "manual":
+            await self._task_run_repo.update_status(
+                queued["id"],
+                status="interrupted",
+                error=RUN_ERROR_PAUSED_WHILE_QUEUED,
+                finished_at=now,
+            )
+            return
+        # The timeout sweep may have failed for this row. Enforce the same
+        # cutoff in the atomic claim so cleanup retries cannot start it.
+        await self._attempt_queued_run(
+            task,
+            queued,
+            now=now,
+            created_after=now - timedelta(seconds=self._queue_timeout_seconds),
+        )
 
     async def _expire_waiting_runs(self, *, now: datetime) -> None:
         await self._task_run_repo.expire_queued_runs(

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

@@ -454,26 +454,44 @@ class ScheduledTaskRunRepository:
             result = await session.execute(stmt)
             return int(result.scalar() or 0)
 
-    async def list_queued_runs(self, *, limit: int, per_user_max_concurrent_runs: int = 0) -> list[dict[str, Any]]:
+    async def list_queued_runs(
+        self,
+        *,
+        limit: int,
+        per_user_max_concurrent_runs: int = 0,
+        created_after: datetime | None = None,
+    ) -> list[dict[str, Any]]:
         """The next bounded drain batch, fair to task owners (see ``_fair_queue_statement``)."""
-        stmt = self._fair_queue_statement(limit=limit, per_user_max_concurrent_runs=per_user_max_concurrent_runs)
+        stmt = self._fair_queue_statement(
+            limit=limit,
+            per_user_max_concurrent_runs=per_user_max_concurrent_runs,
+            created_after=created_after,
+        )
         async with self._sf() as session:
             result = await session.execute(stmt)
             return [self._row_to_dict(row) for row in result.scalars()]
 
     @staticmethod
-    def _fair_queue_statement(*, limit: int, per_user_max_concurrent_runs: int = 0):
+    def _fair_queue_statement(
+        *,
+        limit: int,
+        per_user_max_concurrent_runs: int = 0,
+        created_after: datetime | None = None,
+    ):
         """SELECT for the next bounded drain batch, fair to task owners.
 
         Only same-thread FIFO heads are eligible (an older active row on the
-        same thread hides every newer one). Eligible rows are ranked within
-        their owner by ``attempt_count, created_at, id`` and the batch is
-        ordered by that rank first, so every owner's head comes before any
-        owner's second row: one owner's backlog cannot fill the batch. With a
-        per-owner cap, an owner only contributes as many rows as it has free
-        slots (``cap - executing``), and owners already at the cap drop out
-        before the limit is applied. With a single owner the order equals the
-        plain ``attempt_count, created_at, id`` order.
+        same thread hides every newer one). Rows at or before ``created_after``
+        are already due for timeout and are removed before FIFO checks, owner
+        ranking, and the batch limit; they must not occupy another row's slot.
+        Eligible rows are ranked within their owner by
+        ``attempt_count, created_at, id`` and the batch is ordered by that rank
+        first, so every owner's head comes before any owner's second row: one
+        owner's backlog cannot fill the batch. With a per-owner cap, an owner
+        only contributes as many rows as it has free slots
+        (``cap - executing``), and owners already at the cap drop out before
+        the limit is applied. With a single owner the order equals the plain
+        ``attempt_count, created_at, id`` order.
 
         Rows whose task is gone (deleted while queued) have no owner. They are
         never launched (the drain marks them interrupted), so the cap does not
@@ -481,19 +499,20 @@ class ScheduledTaskRunRepository:
         """
         candidate = aliased(ScheduledTaskRunRow)
         older = aliased(ScheduledTaskRunRow)
-        older_same_thread = exists(
-            select(older.id).where(
-                older.thread_id == candidate.thread_id,
-                older.status.in_(ACTIVE_RUN_STATUSES),
-                or_(
-                    older.created_at < candidate.created_at,
-                    and_(
-                        older.created_at == candidate.created_at,
-                        older.id < candidate.id,
-                    ),
+        older_conditions = [
+            older.thread_id == candidate.thread_id,
+            older.status.in_(ACTIVE_RUN_STATUSES),
+            or_(
+                older.created_at < candidate.created_at,
+                and_(
+                    older.created_at == candidate.created_at,
+                    older.id < candidate.id,
                 ),
-            )
-        )
+            ),
+        ]
+        if created_after is not None:
+            older_conditions.append(older.created_at > created_after)
+        older_same_thread = exists(select(older.id).where(*older_conditions))
         executing_run = aliased(ScheduledTaskRunRow)
         executing_task = aliased(ScheduledTaskRow)
         executing = (
@@ -505,6 +524,9 @@ class ScheduledTaskRunRepository:
             .subquery("executing")
         )
         owner_task = aliased(ScheduledTaskRow)
+        eligible_conditions = [candidate.status == "queued", ~older_same_thread]
+        if created_after is not None:
+            eligible_conditions.append(candidate.created_at > created_after)
         eligible = (
             select(
                 candidate.id.label("id"),
@@ -522,7 +544,7 @@ class ScheduledTaskRunRepository:
             .select_from(candidate)
             .outerjoin(owner_task, owner_task.id == candidate.task_id)
             .outerjoin(executing, executing.c.owner == owner_task.user_id)
-            .where(candidate.status == "queued", ~older_same_thread)
+            .where(*eligible_conditions)
             .subquery("eligible")
         )
         stmt = select(ScheduledTaskRunRow).join(eligible, eligible.c.id == ScheduledTaskRunRow.id)
@@ -566,12 +588,16 @@ class ScheduledTaskRunRepository:
         lease_seconds: int,
         global_max_concurrent_runs: int,
         per_user_max_concurrent_runs: int = 0,
+        created_after: datetime | None = None,
     ) -> dict[str, Any] | None:
         """Atomically move one waiting row into the lease-fenced launch phase.
 
         Both budgets are checked under the same database-wide budget lock, so
         the per-owner cap is as atomic across workers and instances as the
         global one. A row rejected by either budget stays ``queued``.
+
+        ``created_after`` is the queue-age cutoff. Rows at or before it are
+        due for timeout and must stay waiting even if the timeout sweep failed.
         """
         async with self._sf() as session:
             dialect = session.get_bind().dialect.name
@@ -637,13 +663,16 @@ class ScheduledTaskRunRepository:
                     ),
                 )
             )
+            claim_conditions = [
+                ScheduledTaskRunRow.id == run_record_id,
+                ScheduledTaskRunRow.status == "queued",
+                ~older_same_thread,
+            ]
+            if created_after is not None:
+                claim_conditions.append(ScheduledTaskRunRow.created_at > created_after)
             result = await session.execute(
                 update(ScheduledTaskRunRow)
-                .where(
-                    ScheduledTaskRunRow.id == run_record_id,
-                    ScheduledTaskRunRow.status == "queued",
-                    ~older_same_thread,
-                )
+                .where(*claim_conditions)
                 .values(
                     status="launching",
                     lease_owner=lease_owner,
@@ -719,48 +748,57 @@ class ScheduledTaskRunRepository:
 
         expired: list[dict[str, Any]] = []
         for row_id, task_id in candidate_keys:
-            async with self._sf() as session:
-                task = await self._lock_task(session, task_id)
-                row = await session.get(ScheduledTaskRunRow, row_id, with_for_update=True)
-                if row is None or row.status != "queued":
-                    await session.rollback()
-                    continue
+            try:
+                async with self._sf() as session:
+                    task = await self._lock_task(session, task_id)
+                    row = await session.get(ScheduledTaskRunRow, row_id, with_for_update=True)
+                    if row is None or row.status != "queued":
+                        await session.rollback()
+                        continue
 
-                row.status = "failed"
-                row.error = error
-                row.finished_at = now
-                row.lease_owner = None
-                row.lease_expires_at = None
+                    row.status = "failed"
+                    row.error = error
+                    row.finished_at = now
+                    row.lease_owner = None
+                    row.lease_expires_at = None
 
-                if task is not None and can_project(task, row):
-                    if task.status == "paused":
-                        # A concurrent/later pause owns the parent presentation.
-                        task.lease_owner = None
-                        task.lease_expires_at = None
-                    else:
-                        schedule_error = None
-                        if row.trigger == "manual":
-                            next_at = task.next_run_at
-                            task_status = task.status or "enabled"
+                    if task is not None and can_project(task, row):
+                        if task.status == "paused":
+                            # A concurrent/later pause owns the parent presentation.
+                            task.lease_owner = None
+                            task.lease_expires_at = None
                         else:
-                            next_at, schedule_error = stored_next_run_at(
-                                task.schedule_type,
-                                task.schedule_spec,
-                                task.timezone,
-                                now=now,
-                            )
-                            task_status = "failed" if task.schedule_type == "once" else "enabled"
-                        task.status = task_status
-                        task.next_run_at = next_at
-                        task.last_thread_id = row.thread_id
-                        task.last_error = error
-                        task.lease_owner = None
-                        task.lease_expires_at = None
-                        pause_unschedulable(task, schedule_error)
-                    task.updated_at = datetime.now(UTC)
+                            schedule_error = None
+                            if row.trigger == "manual":
+                                next_at = task.next_run_at
+                                task_status = task.status or "enabled"
+                            else:
+                                next_at, schedule_error = stored_next_run_at(
+                                    task.schedule_type,
+                                    task.schedule_spec,
+                                    task.timezone,
+                                    now=now,
+                                )
+                                task_status = "failed" if task.schedule_type == "once" else "enabled"
+                            task.status = task_status
+                            task.next_run_at = next_at
+                            task.last_thread_id = row.thread_id
+                            task.last_error = error
+                            task.lease_owner = None
+                            task.lease_expires_at = None
+                            pause_unschedulable(task, schedule_error)
+                        task.updated_at = datetime.now(UTC)
 
-                await session.commit()
-                expired.append(self._row_to_dict(row))
+                    await session.commit()
+                    expired.append(self._row_to_dict(row))
+            except Exception:
+                # The session above rolls back this row's uncommitted changes.
+                # Keep the remaining candidates eligible for this pass instead
+                # of letting one damaged parent block every queue-timeout write.
+                logger.exception(
+                    "Failed to expire queued scheduled occurrence %s; continuing with the remaining queue",
+                    row_id,
+                )
         return expired
 
     async def fail_launching_run(

backend/tests/test_scheduled_task_queue.py

@@ -10,7 +10,7 @@ import httpx
 import pytest
 import pytest_asyncio
 from _router_auth_helpers import make_authed_test_app
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 
 from app.gateway.auth.models import User
 from app.gateway.auth_disabled import AUTH_SOURCE_SESSION
@@ -922,6 +922,168 @@ async def test_scheduled_queue_timeout_advances_cron_without_immediate_requeue(t
         await close_engine()
 
 
+@pytest.mark.parametrize("failure_mode", ["row", "scan"])
+async def test_queue_expiry_failure_does_not_launch_expired_work(
+    sqlite_repos,
+    monkeypatch,
+    caplog,
+    failure_mode,
+):
+    """A failed timeout sweep may delay cleanup, but never admits old work."""
+    sf, task_repo, run_repo = sqlite_repos
+    poll_at = datetime.now(UTC).replace(microsecond=0)
+    await _owner_task(
+        task_repo,
+        "task-expired",
+        user_id=OWNER_A,
+        next_run_at=poll_at - timedelta(minutes=5),
+    )
+    await _owner_task(
+        task_repo,
+        "task-healthy-queued",
+        user_id=OWNER_A,
+        next_run_at=poll_at + timedelta(hours=1),
+    )
+    await _owner_task(
+        task_repo,
+        "task-due",
+        user_id=OWNER_A,
+        next_run_at=poll_at - timedelta(minutes=1),
+    )
+    await _owned_occurrence(task_repo, run_repo, "expired", user_id=OWNER_A, task_id="task-expired")
+    await _owned_occurrence(
+        task_repo,
+        run_repo,
+        "healthy-queued",
+        user_id=OWNER_A,
+        task_id="task-healthy-queued",
+    )
+    async with sf() as session:
+        await session.execute(update(ScheduledTaskRunRow).where(ScheduledTaskRunRow.id == "expired").values(created_at=poll_at - timedelta(hours=2)))
+        await session.execute(update(ScheduledTaskRunRow).where(ScheduledTaskRunRow.id == "healthy-queued").values(created_at=poll_at - timedelta(minutes=1)))
+        await session.commit()
+
+    if failure_mode == "scan":
+
+        async def fail_expiry_scan(**_kwargs):
+            raise RuntimeError("queue expiry scan failed")
+
+        monkeypatch.setattr(run_repo, "expire_queued_runs", fail_expiry_scan)
+    else:
+        original_lock_task = run_repo._lock_task
+        failed = False
+
+        async def fail_expired_row_once(session, task_id):
+            nonlocal failed
+            if task_id == "task-expired" and not failed:
+                failed = True
+                raise RuntimeError("queue expiry row failed")
+            return await original_lock_task(session, task_id)
+
+        monkeypatch.setattr(run_repo, "_lock_task", fail_expired_row_once)
+
+    launches: list[dict] = []
+    service = _capped_service(task_repo, run_repo, launches, max_concurrent_runs=3, queue_timeout_seconds=3600)
+
+    await service.run_once(now=poll_at)
+
+    launched_tasks = {item["metadata"]["scheduled_task_id"] for item in launches}
+    assert launched_tasks == {"task-healthy-queued", "task-due"}
+    expired = (await run_repo.list_by_task("task-expired"))[0]
+    assert (expired["status"], expired["error"], expired["attempt_count"]) == (
+        "queued",
+        None,
+        0,
+    )
+    assert "Failed to expire queued scheduled" in caplog.text
+
+
+async def test_expired_queue_rows_do_not_consume_capped_candidates(
+    sqlite_repos,
+    monkeypatch,
+):
+    """Timeout failures must not let expired rows crowd healthy work out of the batch."""
+    sf, task_repo, run_repo = sqlite_repos
+    poll_at = datetime.now(UTC).replace(microsecond=0)
+    for task_id, row_id, created_at in (
+        ("task-expired-1", "expired-1", poll_at - timedelta(hours=2)),
+        ("task-expired-2", "expired-2", poll_at - timedelta(hours=2)),
+        ("task-healthy", "healthy-queued", poll_at - timedelta(minutes=1)),
+    ):
+        await _owner_task(
+            task_repo,
+            task_id,
+            user_id=OWNER_A,
+            next_run_at=poll_at + timedelta(hours=1),
+        )
+        await _owned_occurrence(
+            task_repo,
+            run_repo,
+            row_id,
+            user_id=OWNER_A,
+            task_id=task_id,
+        )
+        async with sf() as session:
+            await session.execute(update(ScheduledTaskRunRow).where(ScheduledTaskRunRow.id == row_id).values(created_at=created_at))
+            await session.commit()
+
+    async def fail_expiry_scan(**_kwargs):
+        raise RuntimeError("queue expiry scan failed")
+
+    monkeypatch.setattr(run_repo, "expire_queued_runs", fail_expiry_scan)
+
+    launches: list[dict] = []
+    service = _capped_service(
+        task_repo,
+        run_repo,
+        launches,
+        max_concurrent_runs=3,
+        max_concurrent_runs_per_user=2,
+        queue_timeout_seconds=3600,
+    )
+
+    for offset in range(3):
+        await service.run_once(now=poll_at + timedelta(seconds=offset))
+
+    assert [item["metadata"]["scheduled_task_id"] for item in launches] == ["task-healthy"]
+    for row_id in ("expired-1", "expired-2"):
+        row = (await run_repo.list_by_task(f"task-{row_id}"))[0]
+        assert (row["status"], row["attempt_count"]) == ("queued", 0)
+    healthy = (await run_repo.list_by_task("task-healthy"))[0]
+    assert healthy["status"] == "running"
+
+
+async def test_expired_same_thread_head_does_not_hide_unexpired_candidate(sqlite_repos):
+    """An expired FIFO head cannot block a newer row once it is due for timeout."""
+    sf, task_repo, run_repo = sqlite_repos
+    poll_at = datetime.now(UTC).replace(microsecond=0)
+    await _owned_occurrence(
+        task_repo,
+        run_repo,
+        "expired-head",
+        user_id=OWNER_A,
+        thread_id="shared-thread",
+    )
+    await _owned_occurrence(
+        task_repo,
+        run_repo,
+        "healthy-next",
+        user_id=OWNER_A,
+        thread_id="shared-thread",
+    )
+    async with sf() as session:
+        await session.execute(update(ScheduledTaskRunRow).where(ScheduledTaskRunRow.id == "expired-head").values(created_at=poll_at - timedelta(hours=2)))
+        await session.execute(update(ScheduledTaskRunRow).where(ScheduledTaskRunRow.id == "healthy-next").values(created_at=poll_at - timedelta(minutes=1)))
+        await session.commit()
+
+    batch = await run_repo.list_queued_runs(
+        limit=16,
+        created_after=poll_at - timedelta(hours=1),
+    )
+
+    assert [row["id"] for row in batch] == ["healthy-next"]
+
+
 async def test_manual_queue_timeout_preserves_serialized_next_run_at(tmp_path):
     await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
     try:

backend/tests/test_scheduled_task_service.py

@@ -540,6 +540,147 @@ async def test_run_once_continues_batch_after_invalid_thread_id():
     assert launched[0]["thread_id"] == "thread-ok"
 
 
+@pytest.mark.asyncio
+async def test_run_once_continues_after_queued_run_drain_error(monkeypatch, caplog):
+    """A poison queued row must not block the rest of the drain or due claims."""
+    now = datetime.now(UTC)
+    launched = []
+
+    async def fake_launch(**kwargs):
+        launched.append(kwargs)
+        return {"run_id": f"run-{len(launched)}", "thread_id": kwargs["thread_id"]}
+
+    poison_task = {
+        "id": "task-poison-drain",
+        "user_id": "user-1",
+        "thread_id": "thread-poison-drain",
+        "context_mode": "reuse_thread",
+        "assistant_id": "lead_agent",
+        "prompt": "Summarize thread",
+        "schedule_type": "cron",
+        "schedule_spec": {"cron": "0 9 * * *"},
+        "timezone": "UTC",
+        "status": "enabled",
+    }
+    healthy_task = {
+        "id": "task-healthy-drain",
+        "user_id": "user-1",
+        "thread_id": "thread-healthy-drain",
+        "context_mode": "reuse_thread",
+        "assistant_id": "lead_agent",
+        "prompt": "Summarize thread",
+        "schedule_type": "cron",
+        "schedule_spec": {"cron": "0 9 * * *"},
+        "timezone": "UTC",
+        "status": "enabled",
+    }
+    due_task = {
+        **healthy_task,
+        "id": "task-due-after-drain",
+        "thread_id": "thread-due-after-drain",
+    }
+
+    class DrainRunRepo(DummyRunRepo):
+        async def list_queued_runs(self, *, limit, **_kwargs):
+            return [
+                {"id": "run-poison-drain", "task_id": poison_task["id"], "thread_id": poison_task["thread_id"], "trigger": "manual", "scheduled_for": now},
+                {"id": "run-healthy-drain", "task_id": healthy_task["id"], "thread_id": healthy_task["thread_id"], "trigger": "manual", "scheduled_for": now},
+            ][:limit]
+
+    class DrainTaskRepo(DummyTaskRepo):
+        async def release_queued_admission_lease(self, task_id):
+            if task_id == poison_task["id"]:
+                raise RuntimeError("queue admission lease release failed")
+            return False
+
+    task_repo = DrainTaskRepo([poison_task, healthy_task])
+    run_repo = DrainRunRepo()
+    service = ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=3,
+    )
+
+    async def claim_due_tasks(**_kwargs):
+        return [due_task]
+
+    monkeypatch.setattr(task_repo, "claim_due_tasks", claim_due_tasks)
+
+    await service.run_once(now=now)
+
+    assert [item["thread_id"] for item in launched] == ["thread-healthy-drain", "thread-due-after-drain"]
+    assert "Failed to drain scheduled occurrence run-poison-drain" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_run_once_continues_after_queue_expiry_error(monkeypatch, caplog):
+    """A timeout-sweep failure must not skip draining or due-task admission."""
+    launched = []
+
+    async def fake_launch(**kwargs):
+        launched.append(kwargs)
+        return {"run_id": "run-ok", "thread_id": kwargs["thread_id"]}
+
+    healthy_task = {
+        "id": "task-healthy-expiry",
+        "user_id": "user-1",
+        "thread_id": "thread-healthy-expiry",
+        "context_mode": "reuse_thread",
+        "assistant_id": "lead_agent",
+        "prompt": "Summarize thread",
+        "schedule_type": "cron",
+        "schedule_spec": {"cron": "0 9 * * *"},
+        "timezone": "UTC",
+        "status": "enabled",
+    }
+    queued_task = {
+        **healthy_task,
+        "id": "task-healthy-drain",
+        "thread_id": "thread-healthy-drain",
+    }
+
+    class DrainRunRepo(DummyRunRepo):
+        async def list_queued_runs(self, *, limit, **_kwargs):
+            return [
+                {
+                    "id": "run-healthy-drain",
+                    "task_id": queued_task["id"],
+                    "thread_id": queued_task["thread_id"],
+                    "trigger": "manual",
+                    "scheduled_for": datetime.now(UTC),
+                }
+            ][:limit]
+
+    task_repo = DummyTaskRepo([queued_task, healthy_task])
+    run_repo = DrainRunRepo()
+    service = ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=3,
+    )
+
+    async def broken_expiry(**_kwargs):
+        raise RuntimeError("queue expiry pass failed")
+
+    monkeypatch.setattr(run_repo, "expire_queued_runs", broken_expiry)
+
+    async def claim_due_tasks(**_kwargs):
+        return [healthy_task]
+
+    monkeypatch.setattr(task_repo, "claim_due_tasks", claim_due_tasks)
+
+    await service.run_once(now=datetime.now(UTC))
+
+    assert [item["thread_id"] for item in launched] == ["thread-healthy-drain", "thread-healthy-expiry"]
+    assert "Failed to expire queued scheduled runs" in caplog.text
+
+
 @pytest.mark.asyncio
 async def test_handle_run_completion_uses_atomic_repository_boundary():
     task_repo = DummyTaskRepo(
@@ -1914,9 +2055,9 @@ async def test_manual_trial_retry_reuses_the_admitted_origin():
     attempted: list[dict] = []
     original_attempt = service._attempt_queued_run
 
-    async def capture_attempt(task, queued, *, now):
+    async def capture_attempt(task, queued, *, now, **kwargs):
         attempted.append(dict(queued))
-        return await original_attempt(task, queued, now=now)
+        return await original_attempt(task, queued, now=now, **kwargs)
 
     service._attempt_queued_run = capture_attempt
     admitted = datetime(2026, 10, 7, 1, 0, tzinfo=UTC)

backend/tests/test_scheduler_unschedulable_task.py

@@ -164,6 +164,47 @@ async def test_queue_expiry_pauses_an_unschedulable_task(repos):
     await _assert_paused(tasks, "broken")
 
 
+async def test_queue_expiry_continues_after_one_storage_error(repos, monkeypatch, caplog):
+    tasks, runs, _run_store, factory = repos
+    now = datetime.now(UTC)
+    await _create_task(tasks, "a-poison", next_run_at=None)
+    await _create_task(tasks, "z-healthy", next_run_at=None)
+    await runs.create(
+        run_record_id="poison-occurrence",
+        task_id="a-poison",
+        thread_id="thread-a-poison",
+        scheduled_for=now,
+        trigger="scheduled",
+        status="queued",
+    )
+    await runs.create(
+        run_record_id="healthy-occurrence",
+        task_id="z-healthy",
+        thread_id="thread-z-healthy",
+        scheduled_for=now,
+        trigger="scheduled",
+        status="queued",
+    )
+    original_lock_task = ScheduledTaskRunRepository._lock_task
+
+    async def flaky_lock_task(session, task_id):
+        if task_id == "a-poison":
+            raise RuntimeError("parent row lock failed")
+        return await original_lock_task(session, task_id)
+
+    monkeypatch.setattr(ScheduledTaskRunRepository, "_lock_task", staticmethod(flaky_lock_task))
+    expired = await runs.expire_queued_runs(
+        created_before=now + timedelta(seconds=1),
+        error="queue timeout",
+        now=now,
+    )
+
+    assert [row["id"] for row in expired] == ["healthy-occurrence"]
+    assert [row["status"] for row in await runs.list_by_task("a-poison")] == ["queued"]
+    assert [row["status"] for row in await runs.list_by_task("z-healthy")] == ["failed"]
+    assert "Failed to expire queued scheduled occurrence poison-occurrence" in caplog.text
+
+
 async def test_failed_recovery_pass_does_not_block_admission(repos, monkeypatch):
     tasks, runs, run_store, _factory = repos
     now = datetime.now(UTC)

docs/guide/scheduled-tasks.md

@@ -21,6 +21,7 @@ Current MVP capabilities:
 - Persist a due execution as `queued` when its reused thread or the global execution budget is busy, then launch it when capacity is available; queued occurrences survive Gateway restarts and fail after `scheduler.queue_timeout_seconds`
 - Retrying a busy scheduled task keeps its execution slot while an earlier recovery pass is still in progress.
 - A scheduled run whose completion hook fails is finalized from its durable terminal run on the next scheduler poll instead of staying `running` until a restart.
+- A storage or bookkeeping error on one occurrence during queue cleanup or dispatch is logged without stopping the remaining occurrences and healthy due tasks in that poll. The affected occurrence remains subject to normal queue and lease recovery; persistent storage failures can delay its recovery. A run that has exceeded `scheduler.queue_timeout_seconds` never starts while cleanup is retried, and expired rows do not consume capacity or fair-queue slots from healthy waiting work; they stay queued until a later timeout sweep records the skip.
 - A recurring task whose saved schedule can no longer be computed (for example, its timezone is missing from the time zone database after an upgrade) is paused with the reason as its last error, and every other task keeps running. Resuming it reports the problem until you fix the schedule.
 - Share execution slots fairly between task owners: one owner may have at most `scheduler.max_concurrent_runs_per_user` scheduled runs starting or running at a time (default 2, never more than `max_concurrent_runs`; `0` turns the per-owner cap off), and the waiting queue is drained owner by owner, so one owner's backlog never holds back another owner's run. A run that waits longer than `scheduler.queue_timeout_seconds` is skipped, and its history reads "Skipped: it waited too long for a free slot"
 - Freeze a task's definition while an occurrence is `queued`, `launching`, or `running`, so a durable occurrence cannot silently pick up a different prompt, thread, or schedule; transitioning a task to paused or deleting it cancels an existing waiting occurrence, while `launching`/`running` work must finish before those mutations are retried and an explicit manual trigger may still wait and run without resuming a paused schedule