diff --git a/CHANGELOG.md b/CHANGELOG.md index 5edfa00447..818b5c191c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -804,6 +804,17 @@ This release closes that milestone with **439 merged pull requests**. ### Fixed +- **scheduler:** "Run once now" on a one-time task before its run time no longer + cancels the scheduled run. The trial launched as the task's own run: the task + was marked `running`, and the trial's outcome then finished it (`completed`, + `failed` or `cancelled`), so the poller, which claims only `enabled` tasks, + never ran it at `run_at`, although `next_run_at` still showed that time. A + trial launched before the run time now leaves the task's status and + `next_run_at` unchanged, like a trial on a recurring task; its outcome is kept + on the trial's run row. A trial after the run time has passed still counts as + the task's run. A trial whose launch bookkeeping lands after the poller has + claimed the now-due task no longer clears that claim's lease, which made the + claim's admission fail and left the task `running` with nothing scheduled. ([#6512]) - **auth:** Login lockouts are now counted once per client IP across every Gateway replica. `POST /api/v1/auth/login/local` kept its failed-login counter in a per-process dict, so with N replicas behind one load balancer an @@ -9284,3 +9295,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6495]: https://github.com/bytedance/deer-flow/pull/6495 [#6501]: https://github.com/bytedance/deer-flow/pull/6501 [#6506]: https://github.com/bytedance/deer-flow/pull/6506 +[#6512]: https://github.com/bytedance/deer-flow/pull/6512 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 66a513855d..6fca015b23 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -694,6 +694,12 @@ ### 修复 +- **调度器:** 在一次性任务的执行时间之前点击“立即运行一次”,不会再取消原定的执行。此前这次试运行被当作任务本身的 + 执行:任务被标记为 `running`,试运行结束后又按其结果把任务终结为 `completed`、`failed` 或 `cancelled`。轮询器只认领 + `enabled` 的任务,因此到了 `run_at` 也不会执行,尽管 `next_run_at` 仍显示该时间。现在在执行时间之前启动的试运行 + 不会改变任务的状态和 `next_run_at`,与周期任务的试运行一致,其结果只记录在试运行自己的运行记录上。执行时间已过之后 + 的试运行仍算作任务本身的执行。若试运行的启动记录在轮询器认领已到期的任务之后才写入,也不会再清除该认领的租约;此前这会让 + 该认领的入队失败,任务停留在 `running` 且没有任何待执行的运行。([#6512]) - **技能:** `/技能名` 激活现在在模型调用重试时不再丢失。此前激活在调用模型之前就被标记为已完成,因此调用失败 (限流、过载、超时)或返回空响应而重试时,重试请求不再包含 `SKILL.md` 正文,而该技能的工具限制仍然生效。现在重试 会携带与首次尝试相同的激活内容,不会重新读取技能或重复记录激活,重试得到的响应也保留技能使用记录。([#6506]) @@ -7537,3 +7543,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#6481]: https://github.com/bytedance/deer-flow/pull/6481 [#6484]: https://github.com/bytedance/deer-flow/pull/6484 [#6506]: https://github.com/bytedance/deer-flow/pull/6506 +[#6512]: https://github.com/bytedance/deer-flow/pull/6512 diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 82f3c57d92..a320a6dd57 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -442,13 +442,18 @@ class ScheduledTaskService: return "enabled" @staticmethod - def _task_status_for_launch(task: dict[str, Any], *, trigger: str) -> str: + def _task_status_for_launch(task: dict[str, Any], *, trigger: str, next_at: datetime | None) -> str: # The task-level status to write once _launch_run has produced a live # run. A `once` task stays "running" until handle_run_completion # observes the real terminal outcome; declaring "completed" at launch # would stick if the run fails or the process dies (startup - # reconciliation is cancel_stuck_once_tasks). - if task["schedule_type"] == "once": + # reconciliation is cancel_stuck_once_tasks). A manual trial before + # the run time (``next_at`` still set) is not that run: like a failed + # trial, it must not consume the task's scheduled future. Keep this in + # step with projection.once_run_still_scheduled, which applies the same + # rule when the trial finalizes; if they disagree, a trial parks the + # task in "running" and the stuck-once recovery ends up owning it. + if task["schedule_type"] == "once" and (trigger != "manual" or next_at is None): return "running" if trigger == "manual" and task.get("status") == "paused": return "paused" @@ -651,7 +656,7 @@ class ScheduledTaskService: task["timezone"], now=now, ) - task_status = self._task_status_for_launch(task, trigger=trigger) + 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, task_id=task["id"], @@ -705,7 +710,7 @@ 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) + task_status = self._task_status_for_launch(task, trigger=trigger, next_at=next_at) try: await self._record_launched_run( task_run_id=task_run_id, diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index 13aec4f279..1908c69666 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -21,7 +21,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 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). `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 5aa16b1a59..41e157a609 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/finalization.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/finalization.py @@ -11,7 +11,7 @@ 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 +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.utils.time import coerce_iso @@ -168,10 +168,12 @@ async def finalize_occurrence( # A once task's result remains meaningful: an unmet once task failed. # Limits and agent stop apply to the recurring schedule lifecycle. if task.schedule_type == "once": - task.status = ONCE_TASK_STATUS_BY_RUN_STATUS[status] - # A cancel or a skip (``cancelled``) stays silent, like interrupted runs. - if status in _ONCE_OUTCOME_STATUSES: - events += ("task_finished",) + # A trial ahead of the run time leaves the task as it was. + if not once_run_still_scheduled(task, occurrence): + task.status = ONCE_TASK_STATUS_BY_RUN_STATUS[status] + # A cancel or a skip (``cancelled``) stays silent, like interrupted runs. + if status in _ONCE_OUTCOME_STATUSES: + events += ("task_finished",) elif await end_condition_reached(session, task, now=finished_at): task.status = "completed" task.next_run_at = None diff --git a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/projection.py b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/projection.py index 13fff52342..fa9fc1da56 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/projection.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/projection.py @@ -18,6 +18,21 @@ def can_project(task: ScheduledTaskRow, occurrence: ScheduledTaskRunRow) -> bool return occurrence.occurrence_seq == task.last_occurrence_seq +def once_run_still_scheduled(task: ScheduledTaskRow, occurrence: ScheduledTaskRunRow) -> bool: + """Whether a ``once`` task's own run is still ahead of this trial. + + A manual trigger is a trial ("Run once now"): launched before the run + time it must not consume that run, so the parent keeps its status rather + than taking the once outcome. ``next_run_at`` holds the run time only + while it was still ahead at the trial's launch; the launch write (or the + projection repairing it) clears it otherwise. Always False for a recurring + task, whose ``next_run_at`` is set between runs. The launch path, which + has no occurrence row yet, encodes the same rule in + ``ScheduledTaskService._task_status_for_launch``. + """ + return task.schedule_type == "once" and occurrence.trigger == "manual" and task.next_run_at is not None + + def account_launch(task: ScheduledTaskRow, occurrence: ScheduledTaskRunRow, run_id: str) -> bool: """Count a proven launch once, in the same transaction as its marker. 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 901c68327c..edce9b61b9 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py @@ -14,7 +14,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 +from deerflow.persistence.scheduled_task_runs.projection import account_launch, can_project, once_run_still_scheduled from deerflow.persistence.scheduled_tasks.model import ( ACTIVE_RUN_STATUSES, TERMINAL_RUN_STATUSES, @@ -220,7 +220,9 @@ class ScheduledTaskRunRepository: task.lease_owner = None task.lease_expires_at = None task.updated_at = datetime.now(UTC) - if task.schedule_type == "once": + # A trial ahead of a once task's run time is projected like a + # recurring task's run, so the once run stays scheduled. + if task.schedule_type == "once" and not once_run_still_scheduled(task, row): if candidate.status == "success": task.status = "completed" task.last_error = None diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py index e93285c338..dc94487de3 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py @@ -611,8 +611,15 @@ class ScheduledTaskRepository: row.last_thread_id = last_thread_id if should_increment_run_count: row.run_count += 1 - row.lease_owner = None - row.lease_expires_at = None + if occurrence is None: + row.lease_owner = None + row.lease_expires_at = None + # else: a launch write never owns the parent lease. Scheduled + # admission released the due-task claim with the queue insert and a + # manual trial never took one, so a lease here is a newer claim + # (e.g. the poller claiming a once task an early trial left due, + # before this late write); clearing it would make that claim's + # admission fail as stale and strand the task in "running". row.updated_at = datetime.now(UTC) await session.commit() return True diff --git a/backend/tests/test_scheduler_manual_once_trial.py b/backend/tests/test_scheduler_manual_once_trial.py new file mode 100644 index 0000000000..8560dac150 --- /dev/null +++ b/backend/tests/test_scheduler_manual_once_trial.py @@ -0,0 +1,320 @@ +"""A trial run ("Run once now") must not consume a one-time task's own run. + +The UI offers the trial on every task and says trial runs don't count. For a +``once`` task launched before its run time, the trial's outcome belongs to +the trial's run row only: the task keeps its status and its ``next_run_at``, +and the poller still claims it at the run time. A trial after the run time +has passed is the task's run, so it finalizes the task as before. +""" + +from datetime import UTC, datetime, timedelta + +import pytest +import pytest_asyncio + +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.model import RunRow +from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository +from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow +from deerflow.persistence.scheduled_task_runs.projection import once_run_still_scheduled +from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository +from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow +from deerflow.runtime import RunStatus +from deerflow.runtime.runs.manager import RunRecord +from deerflow.runtime.runs.schemas import DisconnectMode +from deerflow.utils.time import coerce_iso + +OWNER = "user-1" +TASK_ID = "task-once" + + +@pytest_asyncio.fixture +async def repos(tmp_path): + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + sf = get_session_factory() + try: + yield ScheduledTaskRepository(sf), ScheduledTaskRunRepository(sf) + finally: + await close_engine() + + +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=3, + ) + + +async def _create_once_task(tasks, *, run_at: datetime, status: str = "enabled") -> 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="once", + schedule_spec={"run_at": run_at.isoformat()}, + timezone="UTC", + next_run_at=run_at, + ) + if status != "enabled": + await tasks.update(TASK_ID, user_id=OWNER, updates={"status": status}) + 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, + thread_id=launch["thread_id"], + assistant_id=None, + status=status, + on_disconnect=DisconnectMode.continue_, + metadata=launch["metadata"], + user_id=OWNER, + error=error, + ) + + +async def _claimed_ids(tasks, *, now: datetime) -> list[str]: + claimed = await tasks.claim_due_tasks(now=now, lease_owner="poller", lease_seconds=30, limit=10) + return [task["id"] for task in claimed] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("run_status", "error", "trial_status"), + [ + (RunStatus.success, None, "success"), + (RunStatus.error, "boom", "failed"), + (RunStatus.interrupted, "cancelled by user", "interrupted"), + ], +) +async def test_trial_before_run_time_keeps_the_once_run_scheduled(repos, run_status, error, trial_status): + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + run_at = now + timedelta(days=1) + launches = [] + + async def launch_run(**kwargs): + launches.append(kwargs) + return {"run_id": "run-trial", "thread_id": kwargs["thread_id"]} + + service = _service(tasks, runs, launch_run) + task = await _create_once_task(tasks, run_at=run_at) + + result = await service.dispatch_task(task, now=now, trigger="manual") + assert result["outcome"] == "launched" + during = await tasks.get(TASK_ID, user_id=OWNER) + assert (during["status"], during["next_run_at"]) == ("enabled", coerce_iso(run_at)) + + await service.handle_run_completion(_record(launches[0], "run-trial", run_status, error)) + + after = await tasks.get(TASK_ID, user_id=OWNER) + assert (after["status"], after["next_run_at"]) == ("enabled", coerce_iso(run_at)) + # The trial's own outcome is still recorded on its run row. + (occurrence,) = await runs.list_by_task(TASK_ID) + assert (occurrence["trigger"], occurrence["status"]) == ("manual", trial_status) + assert await _claimed_ids(tasks, now=now + timedelta(minutes=1)) == [] + assert await _claimed_ids(tasks, now=run_at + timedelta(minutes=1)) == [TASK_ID] + + +@pytest.mark.asyncio +async def test_trial_on_a_paused_once_task_keeps_it_paused_and_scheduled(repos): + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + run_at = now + timedelta(days=1) + launches = [] + + async def launch_run(**kwargs): + launches.append(kwargs) + return {"run_id": "run-trial", "thread_id": kwargs["thread_id"]} + + service = _service(tasks, runs, launch_run) + task = await _create_once_task(tasks, run_at=run_at, status="paused") + + await service.dispatch_task(task, now=now, trigger="manual") + assert (await tasks.get(TASK_ID, user_id=OWNER))["status"] == "paused" + await service.handle_run_completion(_record(launches[0], "run-trial", RunStatus.success)) + + after = await tasks.get(TASK_ID, user_id=OWNER) + assert (after["status"], after["next_run_at"]) == ("paused", coerce_iso(run_at)) + + +@pytest.mark.asyncio +async def test_completion_before_launch_bookkeeping_keeps_the_once_run_scheduled(repos): + """A fast trial can finish before the launch path writes the parent.""" + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + run_at = now + timedelta(days=1) + service = None + + async def launch_run(**kwargs): + await service.handle_run_completion(_record(kwargs, "run-trial", RunStatus.success)) + return {"run_id": "run-trial", "thread_id": kwargs["thread_id"]} + + service = _service(tasks, runs, launch_run) + task = await _create_once_task(tasks, run_at=run_at) + + assert (await service.dispatch_task(task, now=now, trigger="manual"))["outcome"] == "launched" + + after = await tasks.get(TASK_ID, user_id=OWNER) + assert (after["status"], after["next_run_at"]) == ("enabled", coerce_iso(run_at)) + assert await _claimed_ids(tasks, now=run_at + timedelta(minutes=1)) == [TASK_ID] + + +@pytest.mark.asyncio +async def test_late_trial_bookkeeping_keeps_a_newer_due_task_claim(repos): + """The trial finishes, the poller claims the now-due task, then the trial's launch write lands. + + The poller's claim holds the parent's dispatch lease until its admission + inserts the scheduled occurrence. The late trial write passes + ``can_project`` (no new occurrence sequence yet), and used to clear that + lease: admission then rejected the poller as stale, leaving the task + ``running`` with no lease and no occurrence, which only stuck-once + recovery would later finalize from the trial's outcome. + """ + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + run_at = now + timedelta(days=1) + poll_at = run_at + timedelta(minutes=1) + service = None + claimed = [] + scheduled_launches = [] + + async def launch_run(**kwargs): + if kwargs["metadata"]["scheduled_trigger"] == "scheduled": + scheduled_launches.append(kwargs) + return {"run_id": "run-scheduled", "thread_id": kwargs["thread_id"]} + await service.handle_run_completion(_record(kwargs, "run-trial", RunStatus.success)) + claimed.extend(await tasks.claim_due_tasks(now=poll_at, lease_owner=service._lease_owner, lease_seconds=30, limit=10)) + return {"run_id": "run-trial", "thread_id": kwargs["thread_id"]} + + service = _service(tasks, runs, launch_run) + task = await _create_once_task(tasks, run_at=run_at) + + assert (await service.dispatch_task(task, now=now, trigger="manual"))["outcome"] == "launched" + assert [row["id"] for row in claimed] == [TASK_ID] + parent = await tasks.get(TASK_ID, user_id=OWNER) + assert (parent["status"], parent["lease_owner"]) == ("running", service._lease_owner) + + result = await service.dispatch_task(claimed[0], now=poll_at, trigger="scheduled") + + assert (result["outcome"], result["run_id"]) == ("launched", "run-scheduled") + assert len(scheduled_launches) == 1 + assert (await tasks.get(TASK_ID, user_id=OWNER))["status"] == "running" # the scheduled run, not a stuck claim + + +@pytest.mark.asyncio +async def test_failure_before_admission_still_releases_the_pollers_own_claim(repos): + """The other half of the lease rule: a write with no occurrence row is the claimer's own. + + A legacy thread id that fails validation is recorded before any occurrence + exists, by the poller that holds the claim; that write must release it. + """ + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + + async def launch_run(**_kwargs): + raise AssertionError("an invalid thread id must not launch") + + service = _service(tasks, runs, launch_run) + await tasks.create( + task_id=TASK_ID, + user_id=OWNER, + thread_id="thread.with.dot", + context_mode="reuse_thread", + assistant_id=None, + title="Release check", + prompt="Check the release checklist.", + schedule_type="once", + schedule_spec={"run_at": (now - timedelta(minutes=1)).isoformat()}, + timezone="UTC", + next_run_at=now - timedelta(minutes=1), + ) + (claimed,) = await tasks.claim_due_tasks(now=now, lease_owner=service._lease_owner, lease_seconds=30, limit=10) + + assert (await service.dispatch_task(claimed, now=now, trigger="scheduled"))["outcome"] == "failed" + + after = await tasks.get(TASK_ID, user_id=OWNER) + assert (after["status"], after["lease_owner"], after["lease_expires_at"]) == ("failed", None, None) + + +@pytest.mark.asyncio +async def test_recovered_trial_launch_keeps_the_once_task_enabled(repos): + """Launch-claim recovery that finds the trial's live run must not mark the task running.""" + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + run_at = now + timedelta(days=1) + await _create_once_task(tasks, run_at=run_at) + await runs.create( + run_record_id="occurrence-trial", + task_id=TASK_ID, + thread_id="thread-trial", + scheduled_for=now, + trigger="manual", + status="queued", + coordinate_with_task=True, + expected_task_user_id=OWNER, + expected_task_status="enabled", + ) + assert await runs.claim_queued_run("occurrence-trial", lease_owner="crashed", now=now, lease_seconds=120, global_max_concurrent_runs=3) is not None + # The launcher committed the durable run, then died before its bookkeeping. + async with get_session_factory()() as session: + session.add( + RunRow( + run_id="run-trial", + thread_id="thread-trial", + user_id=OWNER, + status="running", + metadata_json={"scheduled_task_id": TASK_ID, "scheduled_task_run_id": "occurrence-trial"}, + created_at=now, + ) + ) + await session.commit() + + assert await runs.recover_expired_launch_claims(error="recovered", now=now + 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", "run-trial", coerce_iso(run_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).""" + tasks, runs = repos + now = datetime.now(UTC).replace(microsecond=0) + run_at = now - timedelta(minutes=5) + launches = [] + + async def launch_run(**kwargs): + launches.append(kwargs) + return {"run_id": "run-trial", "thread_id": kwargs["thread_id"]} + + service = _service(tasks, runs, launch_run) + task = await _create_once_task(tasks, run_at=run_at) + + await service.dispatch_task(task, now=now, trigger="manual") + assert (await tasks.get(TASK_ID, user_id=OWNER))["status"] == "running" + await service.handle_run_completion(_record(launches[0], "run-trial", RunStatus.success)) + + after = await tasks.get(TASK_ID, user_id=OWNER) + assert (after["status"], after["next_run_at"]) == ("completed", None) + + +@pytest.mark.parametrize( + ("schedule_type", "trigger", "expected"), + [("once", "manual", True), ("once", "scheduled", False), ("cron", "manual", False), ("interval", "manual", False)], +) +def test_once_run_still_scheduled_applies_only_to_once_trials(schedule_type, trigger, expected): + """A recurring task always has a next run, so the rule must not read that as a pending once run.""" + task = ScheduledTaskRow(schedule_type=schedule_type, next_run_at=datetime.now(UTC) + timedelta(days=1)) + occurrence = ScheduledTaskRunRow(trigger=trigger) + assert once_run_still_scheduled(task, occurrence) is expected diff --git a/backend/tests/test_scheduler_occurrence_ordering.py b/backend/tests/test_scheduler_occurrence_ordering.py index 7ae66b561c..4768655023 100644 --- a/backend/tests/test_scheduler_occurrence_ordering.py +++ b/backend/tests/test_scheduler_occurrence_ordering.py @@ -182,6 +182,10 @@ async def test_late_launch_preserves_newer_result_and_counts_once(occurrence_dat service_b = _service(tasks_b, occurrences_b, launch_b, multi) if multi else service_a peer = _service(tasks_recovery, occurrences_recovery, unexpected_launch, multi) if multi else service_a scheduled_at = now - timedelta(seconds=1) if multi else now + timedelta(days=1) + # SQLite drives manual trials ahead of the run time so its poller cannot + # claim the task; such trials leave the once task enabled. Postgres + # launches the due run itself, so its outcome finalizes the task. + success_status, failure_status = ("completed", "failed") if multi else ("enabled", "enabled") await tasks_a.create( task_id=task_id, user_id="user-order", @@ -219,7 +223,7 @@ async def test_late_launch_preserves_newer_result_and_counts_once(occurrence_dat await durable_b.update_status(run_b, "error", error="newer run failed") await service_b.handle_run_completion(_completion(run_b, launches["b"], RunStatus.error, "newer run failed")) expected = await _parent_projection(factories[1], task_id) - assert expected["status"] == "failed" + assert expected["status"] == failure_status assert expected["last_run_id"] == run_b assert expected["run_count"] == 1 expected["run_count"] = 2 @@ -245,7 +249,7 @@ async def test_late_launch_preserves_newer_result_and_counts_once(occurrence_dat if newer_occurrence: assert after == expected else: - assert after["status"] == "completed" + assert after["status"] == success_status assert after["last_run_id"] == run_a assert after["run_count"] == 1 async with factories[1]() as session: