fix(scheduler): keep a one-time task's scheduled run after an early trial (#6512)

* fix(scheduler): keep a one-time task's scheduled run after an early trial

"Run once now" on a once task before its run time launched the trial as
the task's own run: the launch marked the task running and
finalize_occurrence then mapped the trial's outcome to completed, failed
or cancelled. claim_due_tasks only claims enabled tasks, so the run never
fired at run_at although next_run_at still showed it.

A manual trial launched while next_run_at is still ahead now leaves the
parent status and next_run_at unchanged, like a trial on a recurring
task, at launch, in finalize_occurrence and in the launch repair
(_associate_task_with_run). A trial after the run time has passed is
still the task's run.

* docs(changelog): reference #6512

* refactor(scheduler): scope the once-trial rule to once tasks and link its launch twin

* fix(scheduler): keep a newer due-task claim through late launch bookkeeping

With early trials leaving a once task enabled, the poller can claim the
now-due task after the trial finished but before the trial's launch
bookkeeping commits. That late update_after_launch passed can_project (the
claim had not allocated an occurrence yet), kept the claim's "running"
status under protect_terminal, and cleared the claim's lease, so the
poller's admission was rejected as stale and the task was left running
with no lease and no occurrence: unclaimable, and finalized from the
trial by stuck-once recovery.

A launch write identified by its occurrence never owns the parent lease
(scheduled admission releases the claim with the queue insert, and a
manual trial takes none), so update_after_launch now clears the lease only
for a write with no occurrence row, which is the claimer's own pre-admission
failure.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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

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

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,

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.
 

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

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.
 

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

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

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

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: