fix(scheduler): preserve recurring cadence after manual trials (#7209)

* fix(scheduler): preserve cadence after manual trials

* test(scheduler): cover overdue trials and share cursor rule

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
+from deerflow.scheduler.schedules import next_run_at, should_advance_schedule_cursor
 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
@@ -515,6 +515,28 @@ class ScheduledTaskService:
             return "paused"
         return "enabled"
 
+    @staticmethod
+    def _next_run_at_after_launch(
+        task: dict[str, Any],
+        *,
+        trigger: str,
+        now: datetime,
+    ) -> datetime | 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.
+        """
+        if not should_advance_schedule_cursor(task["schedule_type"], trigger=trigger):
+            return task.get("next_run_at")
+        return next_run_at(
+            task["schedule_type"],
+            task["schedule_spec"],
+            task["timezone"],
+            now=now,
+        )
+
     async def dispatch_task(
         self,
         task: dict[str, Any],
@@ -543,12 +565,7 @@ class ScheduledTaskService:
             await self._task_repo.update_after_launch(
                 task["id"],
                 status=task_status,
-                next_run_at=next_run_at(
-                    task["schedule_type"],
-                    task["schedule_spec"],
-                    task["timezone"],
-                    now=now,
-                ),
+                next_run_at=self._next_run_at_after_launch(task, trigger=trigger, now=now),
                 last_run_at=now,
                 last_run_id=None,
                 last_thread_id=execution_thread_id,
@@ -706,12 +723,7 @@ class ScheduledTaskService:
             launch_succeeded = True
             launched_run_id = result["run_id"]
             launched_thread_id = result["thread_id"]
-            next_at = next_run_at(
-                task["schedule_type"],
-                task["schedule_spec"],
-                task["timezone"],
-                now=now,
-            )
+            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)
             await self._record_launched_run(
                 task_run_id=task_run_id,
@@ -749,12 +761,7 @@ class ScheduledTaskService:
                 )
                 return self._queued_result(task_run_id, execution_thread_id, error=str(exc))
 
-            next_at = next_run_at(
-                task["schedule_type"],
-                task["schedule_spec"],
-                task["timezone"],
-                now=now,
-            )
+            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

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

@@ -25,7 +25,7 @@ numbers to BIGINT or NUMERIC. Preserve integer/float/boolean/string distinctions
 with `DEERFLOW_TEST_POSTGRES_URL` and uses connection-local temporary tables.
 ## Scheduled-task lifecycle
 
-Scheduled-task lifecycle rules live in one place each. `ScheduledTaskRepository.update` is the single reactivation choke point: any `status -> enabled` from a non-enabled row re-checks `end_condition_reached` after applying the same request's renewal fields and raises `ScheduledTaskLimitsExhausted` (HTTP/tool `limits_exhausted`). The same method moves `unmet_streak_after_seq` when `goal_objective`, `prompt` or `stop_condition` actually changes (`append_standing_note` too; Resume never), and `_unmet_streak` skips both rows at/below that boundary and goal-check failures (`finalization.CHECK_FAILURE_CODES`). A host pause marker (`is_host_pause_marker`) survives a manual trial on a paused task. A trial launched before a `once` task's run time is not that run: the parent keeps its status and `next_run_at` instead of the once outcome (`projection.once_run_still_scheduled`, applied at launch, in `finalize_occurrence` and in the launch repair). `update_after_launch` clears the parent's dispatch lease only for a write with no occurrence row: an occurrence's launch write never owns it, so a lease there is a newer due-task claim. A task is finished (`completed` by `max_runs`/`end_at`) only by `finalize_occurrence` or `finish_task_at_end_condition` (both in `scheduled_task_runs/finalization.py`), which report `task_stopped`/`task_paused`/`task_finished` to the finalization observer on a live -> paused/finished transition only, inside the caller's transaction; `test_no_inline_end_condition_completion` fails on any other `status = "completed"` in the two repositories. The scheduler's observer (`ScheduledTaskService._on_finalization`) is always installed and writes `scheduled_task_events` rows for the originating chat in that transaction; its lookups fall back on bad stored data and only database errors propagate, because it runs inside batch recovery. Pausing a terminal task returns `"finished"`. `stop_condition` has its own column; only `ScheduledTaskService` composes it into the launched message (`scheduler/stop_rule.launch_prompt`), so never write it into `prompt`. REST and capability validation share `app/gateway/scheduled_task_validation.py`, and every error raised there, in `routers/scheduled_tasks.py` and in `scheduled_task_access.py` uses `scheduler_error(status, code, message, **params)` with a code listed in `contracts/scheduled_task_errors_contract.json` (request models declare types only, so value rules never become FastAPI list details).
+Scheduled-task lifecycle rules live in one place each. `ScheduledTaskRepository.update` is the single reactivation choke point: any `status -> enabled` from a non-enabled row re-checks `end_condition_reached` after applying the same request's renewal fields and raises `ScheduledTaskLimitsExhausted` (HTTP/tool `limits_exhausted`). The same method moves `unmet_streak_after_seq` when `goal_objective`, `prompt` or `stop_condition` actually changes (`append_standing_note` too; Resume never), and `_unmet_streak` skips both rows at/below that boundary and goal-check failures (`finalization.CHECK_FAILURE_CODES`). A host pause marker (`is_host_pause_marker`) survives a manual trial on a paused task. A trial launched before a `once` task's run time is not that run: the parent keeps its status and `next_run_at` instead of the once outcome (`projection.once_run_still_scheduled`, applied at launch, in `finalize_occurrence` and in the launch repair). A manual `interval`/`cron` trial never advances `next_run_at`, even when it is already overdue (`scheduler.schedules.should_advance_schedule_cursor`, shared by the launch, finalization and launch-repair paths). The single-active-occurrence slot blocks the due scheduled run while the trial is active; after the trial finishes, the overdue occurrence remains claimable once, and its catch-up dispatch advances the cursor from the dispatch time. `tests/test_scheduler_manual_once_trial.py` pins both early trials and missed-slot catch-up for interval and cron tasks. `update_after_launch` clears the parent's dispatch lease only for a write with no occurrence row: an occurrence's launch write never owns it, so a lease there is a newer due-task claim. A task is finished (`completed` by `max_runs`/`end_at`) only by `finalize_occurrence` or `finish_task_at_end_condition` (both in `scheduled_task_runs/finalization.py`), which report `task_stopped`/`task_paused`/`task_finished` to the finalization observer on a live -> paused/finished transition only, inside the caller's transaction; `test_no_inline_end_condition_completion` fails on any other `status = "completed"` in the two repositories. The scheduler's observer (`ScheduledTaskService._on_finalization`) is always installed and writes `scheduled_task_events` rows for the originating chat in that transaction; its lookups fall back on bad stored data and only database errors propagate, because it runs inside batch recovery. Pausing a terminal task returns `"finished"`. `stop_condition` has its own column; only `ScheduledTaskService` composes it into the launched message (`scheduler/stop_rule.launch_prompt`), so never write it into `prompt`. REST and capability validation share `app/gateway/scheduled_task_validation.py`, and every error raised there, in `routers/scheduled_tasks.py` and in `scheduled_task_access.py` uses `scheduler_error(status, code, message, **params)` with a code listed in `contracts/scheduled_task_errors_contract.json` (request models declare types only, so value rules never become FastAPI list details).
 
 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.
 

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

@@ -12,7 +12,7 @@ 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
+from deerflow.scheduler.schedules import next_run_at, should_advance_schedule_cursor
 from deerflow.utils.time import coerce_iso
 
 if TYPE_CHECKING:
@@ -155,7 +155,10 @@ async def finalize_occurrence(
         task.last_run_at = launched_at
         task.last_run_id = run_id
         task.last_thread_id = occurrence.thread_id
-        task.next_run_at = next_run_at(task.schedule_type, task.schedule_spec, task.timezone, now=launched_at)
+        # 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.lease_owner = None
         task.lease_expires_at = None
         if task.schedule_type != "once" and task.status == "running":

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

@@ -23,6 +23,7 @@ from deerflow.persistence.scheduled_tasks.model import (
 )
 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.utils.time import coerce_iso
 
 EXECUTING_RUN_STATUSES: tuple[str, ...] = ("launching", "running")
@@ -211,12 +212,15 @@ class ScheduledTaskRunRepository:
         task.last_run_at = launched_at
         task.last_run_id = candidate.run_id
         task.last_thread_id = row.thread_id
-        task.next_run_at = compute_next_run_at(
-            task.schedule_type,
-            task.schedule_spec,
-            task.timezone,
-            now=launched_at,
-        )
+        # A manual recurring run is an extra trial. Keep the task's existing
+        # cadence while repairing launch bookkeeping after a crash.
+        if should_advance_schedule_cursor(task.schedule_type, trigger=row.trigger):
+            task.next_run_at = compute_next_run_at(
+                task.schedule_type,
+                task.schedule_spec,
+                task.timezone,
+                now=launched_at,
+            )
         task.lease_owner = None
         task.lease_expires_at = None
         task.updated_at = datetime.now(UTC)

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

@@ -8,6 +8,11 @@ from croniter import croniter
 MAX_INTERVAL_SECONDS = 30 * 24 * 60 * 60
 
 
+def should_advance_schedule_cursor(schedule_type: str, *, trigger: str) -> bool:
+    """Recurring manual trials keep the scheduled cursor, even when overdue."""
+    return trigger != "manual" or schedule_type == "once"
+
+
 def validate_timezone(timezone_name: str) -> str:
     try:
         ZoneInfo(timezone_name)

backend/tests/test_scheduler_manual_once_trial.py

@@ -70,6 +70,23 @@ async def _create_once_task(tasks, *, run_at: datetime, status: str = "enabled")
     return await tasks.get(TASK_ID, user_id=OWNER)
 
 
+async def _create_recurring_task(tasks, *, task_id: str, schedule_type: str, schedule_spec: dict, next_run_at: datetime) -> dict:
+    await tasks.create(
+        task_id=task_id,
+        user_id=OWNER,
+        thread_id=None,
+        context_mode="fresh_thread_per_run",
+        assistant_id=None,
+        title="Release check",
+        prompt="Check the release checklist.",
+        schedule_type=schedule_type,
+        schedule_spec=schedule_spec,
+        timezone="UTC",
+        next_run_at=next_run_at,
+    )
+    return await tasks.get(task_id, user_id=OWNER)
+
+
 def _record(launch: dict, run_id: str, status: RunStatus, error: str | None = None) -> RunRecord:
     return RunRecord(
         run_id=run_id,
@@ -286,6 +303,155 @@ async def test_recovered_trial_launch_keeps_the_once_task_enabled(repos):
     assert (after["status"], after["last_run_id"], after["next_run_at"]) == ("enabled", "run-trial", coerce_iso(run_at))
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("schedule_type", "schedule_spec"),
+    [("interval", {"every_seconds": 3600}), ("cron", {"cron": "0 10 * * *"})],
+)
+async def test_manual_trial_does_not_push_recurring_next_run(repos, schedule_type, schedule_spec):
+    """A trial before the next occurrence leaves the recurring cadence claimable."""
+    tasks, runs = repos
+    scheduled_at = datetime(2030, 1, 1, 10, tzinfo=UTC)
+    trial_at = scheduled_at - timedelta(minutes=1)
+    task_id = f"task-{schedule_type}"
+    launches = []
+
+    async def launch_run(**kwargs):
+        launches.append(kwargs)
+        return {"run_id": f"run-{schedule_type}-trial", "thread_id": kwargs["thread_id"]}
+
+    service = _service(tasks, runs, launch_run)
+    task = await _create_recurring_task(
+        tasks,
+        task_id=task_id,
+        schedule_type=schedule_type,
+        schedule_spec=schedule_spec,
+        next_run_at=scheduled_at,
+    )
+
+    result = await service.dispatch_task(task, now=trial_at, trigger="manual")
+    assert result["outcome"] == "launched"
+    assert (await tasks.get(task_id, user_id=OWNER))["next_run_at"] == coerce_iso(scheduled_at)
+
+    await service.handle_run_completion(_record(launches[0], result["run_id"], RunStatus.success))
+    after = await tasks.get(task_id, user_id=OWNER)
+    assert (after["status"], after["next_run_at"]) == ("enabled", coerce_iso(scheduled_at))
+
+    claimed = await tasks.claim_due_tasks(now=scheduled_at, lease_owner="poller", lease_seconds=30, limit=10)
+    assert [claimed_task["id"] for claimed_task in claimed] == [task_id]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("schedule_type", "schedule_spec", "expected_next_run_at"),
+    [
+        ("interval", {"every_seconds": 3600}, datetime(2030, 1, 1, 11, 6, tzinfo=UTC)),
+        ("cron", {"cron": "0 10 * * *"}, datetime(2030, 1, 2, 10, tzinfo=UTC)),
+    ],
+)
+async def test_trial_after_missed_recurring_slot_allows_one_catch_up(repos, schedule_type, schedule_spec, expected_next_run_at):
+    """An overdue cursor survives the trial and advances only at catch-up dispatch."""
+    tasks, runs = repos
+    scheduled_at = datetime(2030, 1, 1, 10, tzinfo=UTC)
+    trial_at = scheduled_at + timedelta(minutes=5)
+    catch_up_at = scheduled_at + timedelta(minutes=6)
+    task_id = f"overdue-{schedule_type}"
+    launches = []
+
+    async def launch_run(**kwargs):
+        launches.append(kwargs)
+        return {"run_id": f"run-{len(launches)}", "thread_id": kwargs["thread_id"]}
+
+    service = _service(tasks, runs, launch_run)
+    task = await _create_recurring_task(
+        tasks,
+        task_id=task_id,
+        schedule_type=schedule_type,
+        schedule_spec=schedule_spec,
+        next_run_at=scheduled_at,
+    )
+    assert (await service.dispatch_task(task, now=trial_at, trigger="manual"))["outcome"] == "launched"
+    assert (await tasks.get(task_id, user_id=OWNER))["next_run_at"] == coerce_iso(scheduled_at)
+
+    # The due cursor must not bypass the trial's single active-occurrence slot.
+    await service.run_once(now=trial_at + timedelta(seconds=30))
+    assert len(launches) == 1
+    assert len(await runs.list_by_task(task_id)) == 1
+
+    await service.handle_run_completion(_record(launches[0], "run-1", RunStatus.success))
+    assert (await tasks.get(task_id, user_id=OWNER))["next_run_at"] == coerce_iso(scheduled_at)
+    await service.run_once(now=catch_up_at)
+
+    assert [launch["metadata"]["scheduled_trigger"] for launch in launches] == ["manual", "scheduled"]
+    assert (await tasks.get(task_id, user_id=OWNER))["next_run_at"] == coerce_iso(expected_next_run_at)
+    # Neither another poll while active nor one after completion repeats the
+    # overdue occurrence; the catch-up dispatch has moved the cursor forward.
+    await service.run_once(now=catch_up_at + timedelta(seconds=1))
+    await service.handle_run_completion(_record(launches[1], "run-2", RunStatus.success))
+    await service.run_once(now=catch_up_at + timedelta(minutes=1))
+    assert len(launches) == 2
+    occurrences = await runs.list_by_task(task_id)
+    assert sorted((row["trigger"], row["status"]) for row in occurrences) == [("manual", "success"), ("scheduled", "success")]
+    assert (await tasks.get(task_id, user_id=OWNER))["next_run_at"] == coerce_iso(expected_next_run_at)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("schedule_type", "schedule_spec"),
+    [("interval", {"every_seconds": 3600}), ("cron", {"cron": "0 10 * * *"})],
+)
+async def test_recovered_recurring_trial_keeps_next_run(repos, schedule_type, schedule_spec):
+    """Crash recovery must preserve cadence for a manual recurring trial too."""
+    tasks, runs = repos
+    scheduled_at = datetime(2030, 1, 1, 10, tzinfo=UTC)
+    trial_at = scheduled_at - timedelta(minutes=1)
+    task_id = f"recovered-{schedule_type}"
+    await _create_recurring_task(
+        tasks,
+        task_id=task_id,
+        schedule_type=schedule_type,
+        schedule_spec=schedule_spec,
+        next_run_at=scheduled_at,
+    )
+    await runs.create(
+        run_record_id=f"occurrence-{schedule_type}",
+        task_id=task_id,
+        thread_id=f"thread-{schedule_type}",
+        scheduled_for=trial_at,
+        trigger="manual",
+        status="queued",
+        coordinate_with_task=True,
+        expected_task_user_id=OWNER,
+        expected_task_status="enabled",
+    )
+    assert (
+        await runs.claim_queued_run(
+            f"occurrence-{schedule_type}",
+            lease_owner="crashed",
+            now=trial_at,
+            lease_seconds=120,
+            global_max_concurrent_runs=3,
+        )
+        is not None
+    )
+    async with get_session_factory()() as session:
+        session.add(
+            RunRow(
+                run_id=f"run-{schedule_type}-trial",
+                thread_id=f"thread-{schedule_type}",
+                user_id=OWNER,
+                status="running",
+                metadata_json={"scheduled_task_id": task_id, "scheduled_task_run_id": f"occurrence-{schedule_type}"},
+                created_at=trial_at,
+            )
+        )
+        await session.commit()
+
+    assert await runs.recover_expired_launch_claims(error="recovered", now=trial_at + timedelta(minutes=5)) == 1
+    after = await tasks.get(task_id, user_id=OWNER)
+    assert (after["status"], after["last_run_id"], after["next_run_at"]) == ("enabled", f"run-{schedule_type}-trial", coerce_iso(scheduled_at))
+
+
 @pytest.mark.asyncio
 async def test_trial_after_run_time_still_finalizes_the_once_task(repos):
     """Once the run time has passed, the trial is the task's run (scheduler off or not yet polled)."""