fix(scheduler): isolate a malformed stored schedule from the poll batch (#7999)

remain from its review:

- `stored_next_run_at` caught the ValueErrors a dropped zone or a newer cron
  parser raises, but a stored `schedule_spec` that is JSON null, a list or a
  string fails on `.get` with AttributeError. The launch-failure handler
  recomputes the schedule, so that escaped `run_once`'s claimed-task loop and
  skipped the rest of the batch. Catch it too; the task is paused with the
  reason like the other unusable schedules.
- `run_once` dispatched its claimed batch with a bare loop, so any per-record
  raise in `dispatch_task` re-aborted the batch. Isolate each task's dispatch
  and keep claim-level store failures loud above the loop.

Tests: `test_scheduler_poll_batch_isolation.py` (non-object specs, end-to-end
pause, per-task isolation). Red on `91f20a7`, green at head.

Refs #7998

CHANGELOG.md

@@ -70,6 +70,16 @@ This section accumulates work toward the **2.3.0** milestone
   preventing `OSError` failures on Windows hosts while keeping existing
   read and delete operations permissive.
 
+- **scheduler:** A stored `schedule_spec` that is JSON null, a list or a string
+  no longer aborts the scheduler poll. `stored_next_run_at` caught the value
+  errors from a dropped zone or a newer cron parser, but a non-object spec
+  fails on `.get`; the launch-failure handler recomputed it again, so the
+  `AttributeError` escaped `run_once`'s claimed-task loop. Such a task is now
+  paused with the reason, like the other unusable schedules. The poll loop also
+  isolates one task's dispatch failure from the rest of its claimed batch, so a
+  future per-record raise costs one tick instead of every other claimed task.
+  ([#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
@@ -16186,3 +16196,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7992]: https://github.com/bytedance/deer-flow/pull/7992
 [#7993]: https://github.com/bytedance/deer-flow/pull/7993
 [#8006]: https://github.com/bytedance/deer-flow/pull/8006
+
+[#7998]: https://github.com/bytedance/deer-flow/issues/7998

backend/app/scheduler/service.py

@@ -480,7 +480,17 @@ class ScheduledTaskService:
             limit=self._max_concurrent_runs,
         )
         for task in claimed:
-            await self.dispatch_task(task, now=now, trigger="scheduled")
+            try:
+                await self.dispatch_task(task, now=now, trigger="scheduled")
+            except Exception:
+                # A per-record failure inside one task's dispatch must not abort
+                # the rest of this poll's claimed batch: every claimed task
+                # already holds a lease, and the next tick would skip them the
+                # same way. Claim-level store failures stay loud above the loop.
+                logger.exception(
+                    "Scheduled task %s: dispatch failed; continuing with the claimed batch",
+                    task.get("id"),
+                )
 
     @staticmethod
     def _is_overlap_conflict(exc: Exception) -> bool:

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. Runtime cursor writes use `stored_next_run_at`: a schedule that stops computing pauses its task, never raises.
+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. `stored_next_run_at` never raises (non-object specs included) and pauses the task; `run_once` isolates per-task failures.
 
 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/scheduler/schedules.py

@@ -123,14 +123,16 @@ def stored_next_run_at(
     """``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.
+    later: a zone the installed tz database no longer has, a cron expression
+    a newer croniter rejects, or a ``schedule_spec`` an operator edit or an old
+    release left as JSON null, a list or a string. 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:
+    except (ValueError, OverflowError, AttributeError) as exc:
         return None, f"{UNSCHEDULABLE_ERROR_PREFIX}{exc}"
 
 

backend/tests/test_scheduler_poll_batch_isolation.py

@@ -0,0 +1,128 @@
+"""One unusable record must not abort the scheduler's claimed-task batch.
+
+``stored_next_run_at`` tolerates a zone the installed tz database dropped and a
+cron a newer croniter rejects, but a stored ``schedule_spec`` that is JSON
+null, a list or a string raises ``AttributeError`` from ``.get``. Because the
+launch-failure handler recomputes the schedule again, that escape aborted the
+rest of ``run_once``'s claimed batch every poll. The loop also relied on every
+dispatch path staying raise-free; it now isolates a single task's failure.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+
+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.scheduled_task_runs import ScheduledTaskRunRepository
+from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
+from deerflow.scheduler.schedules import UNSCHEDULABLE_ERROR_PREFIX, next_run_at, stored_next_run_at
+
+pytestmark = pytest.mark.asyncio
+
+DUE = datetime(2026, 1, 1, tzinfo=UTC)
+
+
+def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService:
+    async def fake_launch(**kwargs):
+        launched.append(kwargs)
+        return {"run_id": f"run-{len(launched)}", "thread_id": kwargs["thread_id"]}
+
+    return ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=10,
+    )
+
+
+async def _seed_task(task_repo, task_id: str, *, schedule_spec) -> None:
+    await task_repo.create(
+        task_id=task_id,
+        user_id="user-1",
+        thread_id=None,
+        context_mode="fresh_thread_per_run",
+        assistant_id="lead_agent",
+        title=task_id,
+        prompt="do the thing",
+        schedule_type="cron",
+        schedule_spec=schedule_spec,
+        timezone="UTC",
+        next_run_at=DUE,
+    )
+
+
+@pytest.mark.parametrize("spec", [None, [], "*/5 * * * *", 5])
+async def test_stored_next_run_at_rejects_a_non_object_spec_without_raising(spec):
+    # The raw helper still raises: only runtime bookkeeping tolerates it.
+    with pytest.raises(AttributeError):
+        next_run_at("cron", spec, "UTC", now=datetime.now(UTC))
+
+    next_at, reason = stored_next_run_at("cron", spec, "UTC", now=datetime.now(UTC))
+
+    assert next_at is None
+    assert reason is not None
+    assert reason.startswith(UNSCHEDULABLE_ERROR_PREFIX)
+
+
+async def test_non_object_stored_spec_is_paused_instead_of_aborting_the_batch(tmp_path):
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        sf = get_session_factory()
+        task_repo = ScheduledTaskRepository(sf)
+        run_repo = ScheduledTaskRunRepository(sf)
+        launched: list = []
+        service = _make_service(task_repo, run_repo, launched)
+
+        await _seed_task(task_repo, "poisoned-spec", schedule_spec=[])
+        await _seed_task(task_repo, "healthy-cron", schedule_spec={"cron": "*/5 * * * *"})
+
+        # Before the fix the AttributeError escaped the claimed-task loop.
+        await service.run_once(now=datetime.now(UTC))
+
+        # The run itself still launches; only the schedule bookkeeping fails.
+        assert len(launched) == 2
+        poisoned = await task_repo.get("poisoned-spec", user_id="user-1")
+        assert poisoned is not None
+        assert poisoned["next_run_at"] is None
+        assert poisoned["status"] == "paused"
+        assert UNSCHEDULABLE_ERROR_PREFIX in (poisoned["last_error"] or "")
+        healthy = await task_repo.get("healthy-cron", user_id="user-1")
+        assert healthy is not None
+        assert healthy["next_run_at"] is not None
+    finally:
+        await close_engine()
+
+
+async def test_one_raising_dispatch_does_not_stop_the_rest_of_the_batch(tmp_path, monkeypatch):
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        sf = get_session_factory()
+        task_repo = ScheduledTaskRepository(sf)
+        run_repo = ScheduledTaskRunRepository(sf)
+        launched: list = []
+        service = _make_service(task_repo, run_repo, launched)
+
+        await _seed_task(task_repo, "raising-task", schedule_spec={"cron": "*/5 * * * *"})
+        await _seed_task(task_repo, "healthy-cron", schedule_spec={"cron": "*/5 * * * *"})
+
+        original_dispatch = service.dispatch_task
+
+        async def flaky_dispatch(task, *, now, trigger):
+            if task["id"] == "raising-task":
+                raise RuntimeError("per-record failure")
+            return await original_dispatch(task, now=now, trigger=trigger)
+
+        monkeypatch.setattr(service, "dispatch_task", flaky_dispatch)
+
+        # The loop-level guard keeps the batch going and logs the failing task.
+        await service.run_once(now=datetime.now(UTC))
+
+        assert len(launched) == 1
+    finally:
+        await close_engine()