From 9e08568455177977384587a174c18827fb3851ae Mon Sep 17 00:00:00 2001 From: Lucien <164857476+dakjdakd@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:06:09 +0800 Subject: [PATCH] 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 | 45 +++-- .../harness/deerflow/persistence/AGENTS.md | 2 +- .../scheduled_task_runs/finalization.py | 7 +- .../persistence/scheduled_task_runs/sql.py | 16 +- .../harness/deerflow/scheduler/schedules.py | 5 + .../tests/test_scheduler_manual_once_trial.py | 166 ++++++++++++++++++ 6 files changed, 213 insertions(+), 28 deletions(-) diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 5ede6915a9..bebe7c4d78 100644 --- a/backend/app/scheduler/service.py +++ b/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 diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index 03c04c0031..d80a0790e8 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/finalization.py b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/finalization.py index 41e157a609..ff68322b17 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/finalization.py +++ b/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": diff --git a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py index 68984d8790..98287fc459 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py +++ b/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) diff --git a/backend/packages/harness/deerflow/scheduler/schedules.py b/backend/packages/harness/deerflow/scheduler/schedules.py index 460aa033e5..753efa3259 100644 --- a/backend/packages/harness/deerflow/scheduler/schedules.py +++ b/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) diff --git a/backend/tests/test_scheduler_manual_once_trial.py b/backend/tests/test_scheduler_manual_once_trial.py index 8560dac150..d1a06d2eb1 100644 --- a/backend/tests/test_scheduler_manual_once_trial.py +++ b/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)."""