fix(scheduler): pause a task whose schedule no longer computes (#7993)

* fix(scheduler): pause a task whose schedule no longer computes

Admission validates every schedule, but a stored one can stop computing
later: a zone the installed tz database no longer has, or a cron
expression a newer croniter rejects. The scheduler recomputes next_run_at
inside the shared poll, and that raise escaped:

- dispatch_task recomputed it in its except block after a successful
  launch, so run_once aborted the claimed batch and left the occurrence
  in "launching" with no last_error;
- once the launch lease expired, recover_expired_launch_claims found the
  durable run and raised again for every poll, before claim_due_tasks,
  so no other scheduled task was admitted for the rest of the process.

Runtime cursor writes (launch, launch repair, queue timeout, launch
failure, finalization) now use stored_next_run_at, which returns the
reason instead of raising. A live recurring task is paused with
"schedule can no longer be computed: ..." as its last_error and no next
run; a once task keeps its normal outcome. Resume re-validates the
schedule, so it reports the problem until the schedule is fixed.

run_once also isolates the lease-recovery pass, like the terminal-run
reconciliation next to it, so a failed pass no longer skips admission.

* docs(changelog): reference #7993 for the unschedulable task fix

CHANGELOG.md

@@ -28,6 +28,14 @@ This section accumulates work toward the **2.3.0** milestone
 
 ### Fixed
 
+- **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
+  after an upgrade, a cron expression a newer croniter rejects), recomputing
+  its next run raised inside the scheduler poll; once its launch lease
+  expired, every later poll failed before admitting any due task. That task
+  is now paused with the reason as its last error, and resuming it reports
+  the problem until the schedule is fixed. ([#7993])
 - **community:** A search or fetch API key that ends in a newline no longer
   leaks into the conversation. GroundRoute (`GROUNDROUTE_API_KEY`), InfoQuest,
   Tavily, Exa, Jina, Firecrawl, fastCRW and Crawl4AI put the key into an HTTP
@@ -16119,3 +16127,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7953]: https://github.com/bytedance/deer-flow/issues/7953
 [#7958]: https://github.com/bytedance/deer-flow/pull/7958
 [#7981]: https://github.com/bytedance/deer-flow/pull/7981
+[#7993]: https://github.com/bytedance/deer-flow/pull/7993

CHANGELOG_zh.md

@@ -26,6 +26,11 @@
 
 ### 修复
 
+- **调度器：** 一个已保存的计划无法再计算的定时任务，不再让其他所有任务停止运行。此前
+  已保存的时区或 cron 表达式失效时（例如升级后时区数据库中缺少该时区，或新版 croniter
+  不再接受该表达式），在调度器轮询中计算其下一次运行时间会抛出异常；其启动租约过期后，
+  之后每次轮询都会在接纳到期任务之前失败。现在该任务会被暂停，并把原因记录为最近错误；
+  在修复计划之前，恢复该任务会提示具体问题。([#7993])
 - **社区工具：** 以换行结尾的搜索或抓取 API 密钥不再泄露到对话中。GroundRoute
   （`GROUNDROUTE_API_KEY`）、InfoQuest、Tavily、Exa、Jina、Firecrawl、fastCRW 和
   Crawl4AI 此前把读取到的密钥原样写入 HTTP 请求头；从文件读取的密钥（例如用
@@ -12811,3 +12816,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7953]: https://github.com/bytedance/deer-flow/issues/7953
 [#7958]: https://github.com/bytedance/deer-flow/pull/7958
 [#7981]: https://github.com/bytedance/deer-flow/pull/7981
+[#7993]: https://github.com/bytedance/deer-flow/pull/7993

README.md

@@ -3935,6 +3935,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 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
 - Pause, resume, trigger, inspect history, and delete tasks

backend/app/scheduler/service.py

@@ -29,7 +29,7 @@ from deerflow.persistence.user.model import UserPreferenceRow
 from deerflow.runtime import ConflictError, RunRecord
 from deerflow.runtime.run_origin import origin_kind_of
 from deerflow.scheduler.host_notes import RUN_ERROR_DELETED_WHILE_QUEUED, RUN_ERROR_INTERRUPTED, RUN_ERROR_LEASE_LOST, RUN_ERROR_PAUSED_WHILE_QUEUED, RUN_ERROR_QUEUE_TIMEOUT, RUN_ERROR_RESTARTED
-from deerflow.scheduler.schedules import next_run_at, should_advance_schedule_cursor
+from deerflow.scheduler.schedules import pauses_unschedulable, should_advance_schedule_cursor, stored_next_run_at
 from deerflow.scheduler.stop_rule import launch_prompt
 from deerflow.trace_context import ensure_trace_context
 from deerflow.utils.thread_id import validate_thread_id
@@ -450,10 +450,15 @@ class ScheduledTaskService:
             else:
                 await self._reconcile_active_state(now=now)
         else:
-            await self._task_run_repo.recover_expired_launch_claims(
-                error=_LEASE_RECOVERY_ERROR,
-                now=now,
-            )
+            try:
+                await self._task_run_repo.recover_expired_launch_claims(
+                    error=_LEASE_RECOVERY_ERROR,
+                    now=now,
+                )
+            except Exception:
+                # Like the reconciliation below: a failed recovery pass must
+                # not keep every other due task from being admitted.
+                logger.exception("Failed to recover expired scheduled launch claims")
             try:
                 finalized = await self._task_repo.reconcile_terminal_runs(now=now)
                 if finalized:
@@ -530,22 +535,39 @@ class ScheduledTaskService:
         *,
         trigger: str,
         now: datetime,
-    ) -> datetime | str | None:
+    ) -> tuple[datetime | str | None, str | None]:
         """Keep a recurring task's cadence when a manual trial launches.
 
         A manual run is an extra trial and must not consume the task's next
         scheduled occurrence. One-time tasks retain their existing boundary
         semantics, so only recurring manual launches keep the stored value.
+        The second value is why a stored schedule no longer computes; the
+        launch bookkeeping then pauses the task instead of raising out of the
+        poll (see ``_unschedulable_outcome``).
         """
         if not should_advance_schedule_cursor(task["schedule_type"], trigger=trigger):
-            return task.get("next_run_at")
-        return next_run_at(
+            return task.get("next_run_at"), None
+        return stored_next_run_at(
             task["schedule_type"],
             task["schedule_spec"],
             task["timezone"],
             now=now,
         )
 
+    @staticmethod
+    def _unschedulable_outcome(
+        task: dict[str, Any],
+        *,
+        status: str,
+        last_error: str | None,
+        schedule_error: str | None,
+    ) -> tuple[str, str | None]:
+        """The status and ``last_error`` to write, pausing an unschedulable task."""
+        if not pauses_unschedulable(task["schedule_type"], status, schedule_error):
+            return status, last_error
+        logger.warning("Paused scheduled task %s: %s", task["id"], schedule_error)
+        return "paused", schedule_error
+
     async def dispatch_task(
         self,
         task: dict[str, Any],
@@ -570,15 +592,21 @@ class ScheduledTaskService:
             # surface as HTTP 500 on manual trigger and, in the poller, abort
             # the rest of the claimed batch every cycle while the task itself
             # is never marked with last_error.
-            task_status = self._task_status_for_failure(task, trigger=trigger)
+            next_at, schedule_error = self._next_run_at_after_launch(task, trigger=trigger, now=now)
+            task_status, last_error = self._unschedulable_outcome(
+                task,
+                status=self._task_status_for_failure(task, trigger=trigger),
+                last_error=str(exc),
+                schedule_error=schedule_error,
+            )
             await self._task_repo.update_after_launch(
                 task["id"],
                 status=task_status,
-                next_run_at=self._next_run_at_after_launch(task, trigger=trigger, now=now),
+                next_run_at=next_at,
                 last_run_at=now,
                 last_run_id=None,
                 last_thread_id=execution_thread_id,
-                last_error=str(exc),
+                last_error=last_error,
                 increment_run_count=False,
                 expected_lease_owner=expected_lease_owner,
             )
@@ -736,8 +764,13 @@ class ScheduledTaskService:
             launch_succeeded = True
             launched_run_id = result["run_id"]
             launched_thread_id = result["thread_id"]
-            next_at = self._next_run_at_after_launch(task, trigger=trigger, now=now)
-            task_status = self._task_status_for_launch(task, trigger=trigger, next_at=next_at)
+            next_at, schedule_error = self._next_run_at_after_launch(task, trigger=trigger, now=now)
+            task_status, last_error = self._unschedulable_outcome(
+                task,
+                status=self._task_status_for_launch(task, trigger=trigger, next_at=next_at),
+                last_error=None,
+                schedule_error=schedule_error,
+            )
             await self._record_launched_run(
                 task_run_id=task_run_id,
                 task_id=task["id"],
@@ -751,7 +784,7 @@ class ScheduledTaskService:
                 last_run_at=now,
                 last_run_id=launched_run_id,
                 last_thread_id=launched_thread_id,
-                last_error=None,
+                last_error=last_error,
                 increment_run_count=True,
                 task_run_id=task_run_id,
                 # Same race as the run-row write above: a fast-failing run's
@@ -774,8 +807,6 @@ class ScheduledTaskService:
                 )
                 return self._queued_result(task_run_id, execution_thread_id, error=str(exc))
 
-            next_at = self._next_run_at_after_launch(task, trigger=trigger, now=now)
-
             if launch_succeeded:
                 # _launch_run succeeded, so a run is live even though
                 # post-launch bookkeeping raised. Keep the task-run row
@@ -786,7 +817,13 @@ class ScheduledTaskService:
                 # if the DB is still down the row stays "queued" -- still
                 # active, still holding the slot -- so we log and still report
                 # the run as launched so callers know a run is in flight.
-                task_status = self._task_status_for_launch(task, trigger=trigger, next_at=next_at)
+                next_at, schedule_error = self._next_run_at_after_launch(task, trigger=trigger, now=now)
+                task_status, last_error = self._unschedulable_outcome(
+                    task,
+                    status=self._task_status_for_launch(task, trigger=trigger, next_at=next_at),
+                    last_error=None,
+                    schedule_error=schedule_error,
+                )
                 try:
                     await self._record_launched_run(
                         task_run_id=task_run_id,
@@ -816,7 +853,7 @@ class ScheduledTaskService:
                         # on a task whose run is actively running; the real
                         # terminal outcome is written by handle_run_completion.
                         # The transient itself is logged above.
-                        last_error=None,
+                        last_error=last_error,
                         increment_run_count=True,
                         task_run_id=task_run_id,
                         protect_terminal=True,

backend/packages/harness/deerflow/persistence/AGENTS.md

@@ -47,7 +47,7 @@ Scheduled-task lifecycle rules live in one place each. `ScheduledTaskRepository.
 
 The per-owner run cap `scheduler.max_concurrent_runs_per_user` is effective `min(per_user, max_concurrent_runs)` (0 = off) and is checked in `claim_queued_run` right after the global count, under the same budget lock. `list_queued_runs` ranks same-thread FIFO heads per owner (`ROW_NUMBER() OVER (PARTITION BY owner)`) and orders the batch by that rank, so the drain is owner-fair even with the cap off; owners at their cap drop out before the batch limit, and orphan rows (task deleted while queued) are never held back by the cap. Scheduled-task repository boundaries coerce serialized timestamps before SQL `DateTime` binding.
 
-Recovery locks task/run pairs by task/run ID; restore `run_id`/`started_at`/errors before releasing claims. Bind takeover grants to candidate incarnation/revision; log unconfirmed publication without reauthorizing. Launch/failure/timeout writes lock parent first; queue timeout fails the occurrence and advances scheduled work.
+Recovery locks task/run pairs by task/run ID; restore `run_id`/`started_at`/errors before releasing claims. Bind takeover grants to candidate incarnation/revision; log unconfirmed publication without reauthorizing. Launch/failure/timeout writes lock parent first; queue timeout fails the occurrence and advances scheduled work. Runtime cursor writes use `stored_next_run_at`: a schedule that stops computing pauses its task, never raises.
 
 Single-instance polling also performs a non-destructive terminal-run reconciliation: a `running` occurrence whose durable `runs` row is already terminal is finalized from that row's status, error and goal verdict. Live runs are never taken over or interrupted; multi-instance reconciliation already covers terminal runs through lease ownership. This repairs a completion hook that failed after the run committed its terminal status.
 

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

@@ -11,8 +11,8 @@ from sqlalchemy import and_, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
-from deerflow.persistence.scheduled_task_runs.projection import account_launch, can_project, once_run_still_scheduled
-from deerflow.scheduler.schedules import next_run_at, should_advance_schedule_cursor
+from deerflow.persistence.scheduled_task_runs.projection import account_launch, can_project, once_run_still_scheduled, pause_unschedulable
+from deerflow.scheduler.schedules import should_advance_schedule_cursor, stored_next_run_at
 from deerflow.utils.time import coerce_iso
 
 if TYPE_CHECKING:
@@ -150,6 +150,7 @@ async def finalize_occurrence(
         account_launch(task, occurrence, run_id)
     if not can_project(task, occurrence):
         return True
+    schedule_error = None
     if run_id is not None and task.last_run_id != run_id:
         launched_at = utc(occurrence.started_at or occurrence.scheduled_for)
         task.last_run_at = launched_at
@@ -158,7 +159,7 @@ async def finalize_occurrence(
         # A manual recurring run is an extra trial. Keep the task's existing
         # cadence; only scheduled runs and once-task boundary logic advance it.
         if should_advance_schedule_cursor(task.schedule_type, trigger=occurrence.trigger):
-            task.next_run_at = next_run_at(task.schedule_type, task.schedule_spec, task.timezone, now=launched_at)
+            task.next_run_at, schedule_error = stored_next_run_at(task.schedule_type, task.schedule_spec, task.timezone, now=launched_at)
         task.lease_owner = None
         task.lease_expires_at = None
         if task.schedule_type != "once" and task.status == "running":
@@ -195,6 +196,7 @@ async def finalize_occurrence(
         task.last_error = AUTO_PAUSE_LAST_ERROR
         if not was_paused:
             events += ("task_paused",)
+    pause_unschedulable(task, schedule_error)
     task.updated_at = finished_at
     if observer is not None:
         await observer(session, task, occurrence, events=events)

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

@@ -2,14 +2,19 @@
 
 from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING
 
 from sqlalchemy.orm.attributes import flag_modified
 
+from deerflow.scheduler.schedules import pauses_unschedulable
+
 if TYPE_CHECKING:
     from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
     from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
 
+logger = logging.getLogger(__name__)
+
 
 def can_project(task: ScheduledTaskRow, occurrence: ScheduledTaskRunRow) -> bool:
     """Keep unsequenced history best-effort until a sequenced run is admitted."""
@@ -50,3 +55,16 @@ def account_launch(task: ScheduledTaskRow, occurrence: ScheduledTaskRunRow, run_
     # timestamp. Callers that also project a result explicitly set updated_at.
     flag_modified(task, "updated_at")
     return True
+
+
+def pause_unschedulable(task: ScheduledTaskRow, schedule_error: str | None) -> None:
+    """Pause a recurring task whose stored schedule no longer computes.
+
+    Its cursor is already cleared, so nothing would claim it again; the pause
+    and ``last_error`` say why. See ``stored_next_run_at``.
+    """
+    if not pauses_unschedulable(task.schedule_type, task.status, schedule_error):
+        return
+    logger.warning("Paused scheduled task %s: %s", task.id, schedule_error)
+    task.status = "paused"
+    task.last_error = schedule_error

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

@@ -15,7 +15,7 @@ from deerflow.persistence.run import RunRepository
 from deerflow.persistence.run.model import RunRow
 from deerflow.persistence.scheduled_task_runs.finalization import FinalizationObserver, end_condition_reached, finalize_occurrence, finish_task_at_end_condition, is_host_pause_marker
 from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
-from deerflow.persistence.scheduled_task_runs.projection import account_launch, can_project, once_run_still_scheduled
+from deerflow.persistence.scheduled_task_runs.projection import account_launch, can_project, once_run_still_scheduled, pause_unschedulable
 from deerflow.persistence.scheduled_tasks.model import (
     ACTIVE_RUN_STATUSES,
     TERMINAL_RUN_STATUSES,
@@ -24,8 +24,7 @@ from deerflow.persistence.scheduled_tasks.model import (
 )
 from deerflow.runtime.events.terminal import RecoveryTerminalPublisher, RecoveryTerminalTarget
 from deerflow.scheduler.host_notes import RUN_ERROR_END_REACHED
-from deerflow.scheduler.schedules import next_run_at as compute_next_run_at
-from deerflow.scheduler.schedules import should_advance_schedule_cursor
+from deerflow.scheduler.schedules import should_advance_schedule_cursor, stored_next_run_at
 from deerflow.utils.time import coerce_iso
 
 logger = logging.getLogger(__name__)
@@ -220,8 +219,9 @@ class ScheduledTaskRunRepository:
         task.last_thread_id = row.thread_id
         # A manual recurring run is an extra trial. Keep the task's existing
         # cadence while repairing launch bookkeeping after a crash.
+        schedule_error = None
         if should_advance_schedule_cursor(task.schedule_type, trigger=row.trigger):
-            task.next_run_at = compute_next_run_at(
+            task.next_run_at, schedule_error = stored_next_run_at(
                 task.schedule_type,
                 task.schedule_spec,
                 task.timezone,
@@ -248,6 +248,7 @@ class ScheduledTaskRunRepository:
         elif not (row.trigger == "manual" and task.status == "paused"):
             task.status = "enabled"
             task.last_error = candidate.error if candidate.status in {"error", "timeout", "interrupted"} else None
+        pause_unschedulable(task, schedule_error)
 
     async def create(
         self,
@@ -737,11 +738,12 @@ class ScheduledTaskRunRepository:
                         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"
                         else:
-                            next_at = compute_next_run_at(
+                            next_at, schedule_error = stored_next_run_at(
                                 task.schedule_type,
                                 task.schedule_spec,
                                 task.timezone,
@@ -754,6 +756,7 @@ class ScheduledTaskRunRepository:
                         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()
@@ -785,12 +788,13 @@ class ScheduledTaskRunRepository:
             row.lease_expires_at = None
 
             if task is not None and can_project(task, row):
+                schedule_error = None
                 if row.trigger == "manual":
                     task_status = task.status or "enabled"
                     next_at = task.next_run_at
                 else:
                     task_status = "failed" if task.schedule_type == "once" else "enabled"
-                    next_at = compute_next_run_at(
+                    next_at, schedule_error = stored_next_run_at(
                         task.schedule_type,
                         task.schedule_spec,
                         task.timezone,
@@ -807,6 +811,7 @@ class ScheduledTaskRunRepository:
                 task.lease_owner = None
                 task.lease_expires_at = None
                 task.updated_at = datetime.now(UTC)
+                pause_unschedulable(task, schedule_error)
 
             await session.commit()
             return True

backend/packages/harness/deerflow/scheduler/schedules.py

@@ -6,6 +6,7 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 from croniter import croniter
 
 MAX_INTERVAL_SECONDS = 30 * 24 * 60 * 60
+UNSCHEDULABLE_ERROR_PREFIX = "schedule can no longer be computed: "
 
 
 def should_advance_schedule_cursor(schedule_type: str, *, trigger: str) -> bool:
@@ -110,3 +111,34 @@ def next_run_at(
         return now.astimezone(UTC) + timedelta(seconds=every_seconds)
 
     raise ValueError(f"Unsupported schedule_type: {schedule_type}")
+
+
+def stored_next_run_at(
+    schedule_type: str,
+    schedule_spec: dict[str, object],
+    timezone_name: str,
+    *,
+    now: datetime,
+) -> tuple[datetime | None, str | None]:
+    """``next_run_at`` for an admitted schedule, or why it no longer computes.
+
+    Admission validates every schedule, but a stored one can stop computing
+    later: a zone the installed tz database no longer has, or a cron expression
+    a newer croniter rejects. Runtime bookkeeping runs inside the shared
+    scheduler poll, so it must not raise for one task; callers take the task
+    off the schedule with :func:`pauses_unschedulable` instead.
+    """
+    try:
+        return next_run_at(schedule_type, schedule_spec, timezone_name, now=now), None
+    except (ValueError, OverflowError) as exc:
+        return None, f"{UNSCHEDULABLE_ERROR_PREFIX}{exc}"
+
+
+def pauses_unschedulable(schedule_type: str, status: str | None, schedule_error: str | None) -> bool:
+    """Whether a task whose schedule no longer computes is paused.
+
+    Only a live recurring task changes: a once task has no later run to lose,
+    and a paused or finished task keeps its status. Resuming re-validates the
+    schedule, so it reports the problem until the schedule is fixed.
+    """
+    return schedule_error is not None and schedule_type != "once" and status in {"enabled", "running"}

backend/tests/test_scheduled_task_schedules.py

@@ -3,9 +3,12 @@ from datetime import UTC, datetime, timedelta
 import pytest
 
 from deerflow.scheduler.schedules import (
+    UNSCHEDULABLE_ERROR_PREFIX,
     next_run_at,
     normalize_cron_expression,
     parse_interval_seconds,
+    pauses_unschedulable,
+    stored_next_run_at,
     validate_timezone,
 )
 
@@ -208,3 +211,28 @@ def test_next_run_at_for_interval_does_not_catch_up_from_a_stale_now():
 def test_parse_interval_seconds_rejects_invalid_spec(spec):
     with pytest.raises(ValueError, match="every_seconds"):
         parse_interval_seconds(spec)
+
+
+def test_stored_next_run_at_reports_instead_of_raising():
+    now = datetime(2026, 11, 8, tzinfo=UTC)
+
+    assert stored_next_run_at("interval", {"every_seconds": 60}, "UTC", now=now) == (now + timedelta(seconds=60), None)
+    assert stored_next_run_at("interval", {"every_seconds": 60}, "Mars/Base", now=now) == (None, f"{UNSCHEDULABLE_ERROR_PREFIX}Unknown timezone: Mars/Base")
+    next_at, error = stored_next_run_at("cron", {"cron": "61 * * * *"}, "UTC", now=now)
+    assert next_at is None
+    assert error.startswith(UNSCHEDULABLE_ERROR_PREFIX)
+
+
+@pytest.mark.parametrize(
+    ("schedule_type", "status", "schedule_error", "expected"),
+    [
+        ("interval", "enabled", "broken", True),
+        ("cron", "running", "broken", True),
+        ("cron", "enabled", None, False),
+        ("once", "running", "broken", False),
+        ("cron", "paused", "broken", False),
+        ("cron", "completed", "broken", False),
+    ],
+)
+def test_only_a_live_recurring_task_is_paused(schedule_type, status, schedule_error, expected):
+    assert pauses_unschedulable(schedule_type, status, schedule_error) is expected

backend/tests/test_scheduler_unschedulable_task.py

@@ -0,0 +1,211 @@
+"""A task whose stored schedule no longer computes must not stop the scheduler.
+
+Admission validates every schedule, but a stored one can stop computing later:
+a zone the installed tz database no longer has, or a cron expression a newer
+croniter rejects. The scheduler recomputes the next run inside the shared poll
+(launch bookkeeping, lease recovery, queue expiry, launch failure and
+finalization), so a raise there used to abort ``run_once`` on every poll and no
+other task was admitted again. The task is now paused with the reason instead.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+import pytest_asyncio
+from fastapi import HTTPException
+from sqlalchemy import update
+
+from app.gateway.scheduled_task_validation import prepare_resume_updates
+from app.scheduler.service import ScheduledTaskService
+from deerflow.config.database_config import DatabaseConfig
+from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
+from deerflow.persistence.run import RunRepository
+from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
+from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
+from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
+from deerflow.scheduler.schedules import UNSCHEDULABLE_ERROR_PREFIX
+
+pytestmark = pytest.mark.asyncio
+
+UNKNOWN_ZONE = "Mars/Olympus_Mons"
+UNSCHEDULABLE_ERROR = f"{UNSCHEDULABLE_ERROR_PREFIX}Unknown timezone: {UNKNOWN_ZONE}"
+
+
+@pytest_asyncio.fixture
+async def repos(tmp_path):
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        factory = get_session_factory()
+        assert factory is not None
+        yield ScheduledTaskRepository(factory), ScheduledTaskRunRepository(factory), RunRepository(factory), factory
+    finally:
+        await close_engine()
+
+
+async def _create_task(tasks, task_id, *, next_run_at, context_mode="reuse_thread", schedule_type="interval", schedule_spec=None):
+    await tasks.create(
+        task_id=task_id,
+        user_id="user-1",
+        thread_id=f"thread-{task_id}" if context_mode == "reuse_thread" else None,
+        context_mode=context_mode,
+        assistant_id="lead_agent",
+        title=task_id,
+        prompt="Work",
+        schedule_type=schedule_type,
+        schedule_spec=schedule_spec or {"every_seconds": 600},
+        timezone="UTC",
+        next_run_at=next_run_at,
+    )
+
+
+async def _break_timezone(factory, task_id):
+    """Simulate a zone the tz database dropped after the task was admitted."""
+    async with factory() as session:
+        await session.execute(update(ScheduledTaskRow).where(ScheduledTaskRow.id == task_id).values(timezone=UNKNOWN_ZONE))
+        await session.commit()
+
+
+def _service(tasks, runs, launch_run):
+    return ScheduledTaskService(task_repo=tasks, task_run_repo=runs, launch_run=launch_run, poll_interval_seconds=5, lease_seconds=120, max_concurrent_runs=10, queue_timeout_seconds=3600)
+
+
+def _gateway_launcher(run_store, launched):
+    """Record a durable run per occurrence, as Gateway admission does."""
+    run_ids: dict[str, str] = {}
+
+    async def launch_run(**kwargs):
+        occurrence = kwargs["metadata"]["scheduled_task_run_id"]
+        launched.append(kwargs["metadata"]["scheduled_task_id"])
+        if occurrence not in run_ids:
+            run_ids[occurrence] = f"run-{len(run_ids) + 1}"
+            await run_store.put(run_ids[occurrence], thread_id=kwargs["thread_id"], user_id="user-1", status="running", metadata=dict(kwargs["metadata"]))
+        return {"run_id": run_ids[occurrence], "thread_id": kwargs["thread_id"]}
+
+    return launch_run
+
+
+async def _assert_paused(tasks, task_id):
+    task = await tasks.get(task_id, user_id="user-1")
+    assert task["status"] == "paused"
+    assert task["next_run_at"] is None
+    assert task["last_error"] == UNSCHEDULABLE_ERROR
+
+
+@pytest.mark.parametrize("context_mode", ["reuse_thread", "fresh_thread_per_run"])
+async def test_unschedulable_task_is_paused_and_every_other_task_keeps_running(repos, context_mode):
+    tasks, runs, run_store, factory = repos
+    now = datetime.now(UTC)
+    await _create_task(tasks, "broken", next_run_at=now - timedelta(seconds=30), context_mode=context_mode)
+    await _create_task(tasks, "healthy", next_run_at=now - timedelta(seconds=10), context_mode=context_mode)
+    await _create_task(tasks, "later", next_run_at=now + timedelta(seconds=300), context_mode=context_mode)
+    await _break_timezone(factory, "broken")
+    launched: list[str] = []
+    service = _service(tasks, runs, _gateway_launcher(run_store, launched))
+
+    # Past the 120 s launch lease, so lease recovery sees every claim.
+    for poll in range(8):
+        await service.run_once(now=now + timedelta(seconds=70 * poll))
+
+    assert launched.count("broken") == (1 if context_mode == "reuse_thread" else 0)
+    assert launched.count("healthy") == 1
+    assert launched.count("later") == 1
+    await _assert_paused(tasks, "broken")
+    assert (await tasks.get("healthy", user_id="user-1"))["status"] == "enabled"
+
+
+async def _launching_occurrence(tasks, runs, factory, *, now, run_status=None, run_store=None):
+    await _create_task(tasks, "broken", next_run_at=None)
+    await runs.create(run_record_id="occurrence-1", task_id="broken", thread_id="thread-broken", scheduled_for=now, trigger="scheduled", status="queued")
+    assert await runs.claim_queued_run("occurrence-1", lease_owner="worker", now=now, lease_seconds=5, global_max_concurrent_runs=1)
+    if run_status is not None:
+        await run_store.put("run-1", thread_id="thread-broken", user_id="user-1", status=run_status, metadata={"scheduled_task_id": "broken", "scheduled_task_run_id": "occurrence-1", "scheduled_trigger": "scheduled"})
+    await _break_timezone(factory, "broken")
+
+
+@pytest.mark.parametrize("run_status", ["running", "success"])
+async def test_lease_recovery_pauses_an_unschedulable_task_instead_of_raising(repos, run_status):
+    """``running`` repairs the launch; ``success`` finalizes the occurrence."""
+    tasks, runs, run_store, factory = repos
+    now = datetime.now(UTC)
+    await _launching_occurrence(tasks, runs, factory, now=now, run_status=run_status, run_store=run_store)
+
+    assert await runs.recover_expired_launch_claims(error="expired", now=now + timedelta(seconds=6)) == 1
+
+    await _assert_paused(tasks, "broken")
+    [occurrence] = await runs.list_by_task("broken")
+    assert occurrence["status"] == run_status
+    assert occurrence["run_id"] == "run-1"
+
+
+async def test_failed_launch_pauses_an_unschedulable_task(repos):
+    tasks, runs, _run_store, factory = repos
+    now = datetime.now(UTC)
+    await _launching_occurrence(tasks, runs, factory, now=now)
+
+    assert await runs.fail_launching_run("occurrence-1", task_id="broken", lease_owner="worker", error="launch failed", now=now)
+
+    await _assert_paused(tasks, "broken")
+    assert [row["status"] for row in await runs.list_by_task("broken")] == ["failed"]
+
+
+async def test_queue_expiry_pauses_an_unschedulable_task(repos):
+    tasks, runs, _run_store, factory = repos
+    now = datetime.now(UTC)
+    await _create_task(tasks, "broken", next_run_at=None)
+    await runs.create(run_record_id="occurrence-1", task_id="broken", thread_id="thread-broken", scheduled_for=now, trigger="scheduled", status="queued")
+    await _break_timezone(factory, "broken")
+
+    expired = await runs.expire_queued_runs(created_before=now + timedelta(seconds=1), error="queue timeout", now=now)
+
+    assert [row["id"] for row in expired] == ["occurrence-1"]
+    await _assert_paused(tasks, "broken")
+
+
+async def test_failed_recovery_pass_does_not_block_admission(repos, monkeypatch):
+    tasks, runs, run_store, _factory = repos
+    now = datetime.now(UTC)
+    await _create_task(tasks, "healthy", next_run_at=now - timedelta(seconds=10))
+    launched: list[str] = []
+    service = _service(tasks, runs, _gateway_launcher(run_store, launched))
+
+    async def broken_recovery(**_kwargs):
+        raise RuntimeError("recovery pass failed")
+
+    monkeypatch.setattr(runs, "recover_expired_launch_claims", broken_recovery)
+    await service.run_once(now=now)
+
+    assert launched == ["healthy"]
+
+
+async def test_resuming_reports_the_schedule_until_it_is_fixed(repos):
+    """The pause is not a dead end: resume re-validates and names the problem."""
+    tasks, runs, run_store, factory = repos
+    now = datetime.now(UTC)
+    await _create_task(tasks, "broken", next_run_at=now - timedelta(seconds=30))
+    await _break_timezone(factory, "broken")
+    await _service(tasks, runs, _gateway_launcher(run_store, [])).run_once(now=now)
+    paused = await tasks.get("broken", user_id="user-1")
+
+    with pytest.raises(HTTPException) as exc_info:
+        prepare_resume_updates(paused, {}, used=0, now=now)
+
+    assert exc_info.value.status_code == 422
+    assert exc_info.value.detail == {"code": "invalid_schedule", "message": f"Unknown timezone: {UNKNOWN_ZONE}"}
+
+
+async def test_once_task_without_a_computable_schedule_keeps_its_once_outcome(repos):
+    tasks, runs, run_store, factory = repos
+    now = datetime.now(UTC)
+    await _create_task(tasks, "once", next_run_at=now - timedelta(seconds=30), schedule_type="once", schedule_spec={"run_at": (now - timedelta(seconds=30)).isoformat()})
+    await _break_timezone(factory, "once")
+    launched: list[str] = []
+
+    await _service(tasks, runs, _gateway_launcher(run_store, launched)).run_once(now=now)
+
+    task = await tasks.get("once", user_id="user-1")
+    assert launched == ["once"]
+    assert task["status"] == "running"
+    assert task["next_run_at"] is None
+    assert task["last_error"] is None