fix(schedule): stop the completion hook rolling back the launch write

A cron task could become permanently unschedulable after a run that failed
fast. Reproduced against the real SQL adapters, and against the legacy path
for contrast -- the legacy code does not have this bug, so the hexagonal
slice introduced it.

The window
----------
`dispatch_task` already documents it: "a fast-failing run can reach
handle_run_completion before this write lands". The hook therefore reads the
task while the dispatch path has not yet written its bookkeeping, and the
snapshot it holds still carries the elapsed `next_run_at` the claim was made
on. `handle_run_completion` then wrote that whole snapshot back through
`save()`, so whichever landed second undid the other.

For a cron task the result is terminal in the worst way. `record_launch`
writes the fresh fire time, bumps `run_count` and clears the claim; the
completion's whole-aggregate write restores the elapsed fire time, rolls
`run_count` back and leaves `status='running'` with `lease_expires_at IS
NULL`. Neither `claim_due` branch matches that shape (one needs `enabled`,
the other needs an expired claim), and `cancel_stuck_once_tasks` only sweeps
`once` rows -- so nothing can ever reach the task again.

The legacy `app/scheduler/service.py` wrote the same outcome field by field
(`update(..., updates={"last_error": ...})`) and never touched scheduling
state, which is why it survives the same interleaving.

The fix
-------
`record_completion` joins `record_launch` as a second deliberately narrow
port method, for the same stated reason `record_launch` is not expressed as
`save(task)`: the two race, so neither may write through the aggregate. They
now own disjoint fields -- the launch owns the schedule, the completion owns
the verdict (terminal status plus `last_error`, with `None` meaning "do not
move the status", i.e. every cron task).

`save()` keeps its documented purpose, the user-initiated whole-aggregate
updates (`update_task` / `pause` / `resume`). `handle_run_completion` still
reads the task first, but only to ask `status_after_completion`, which reads
nothing but the schedule type -- immutable, so that read carries no
time-of-check risk.

Tests
-----
The contract suite missed this because it groups cases by port method:
`record_launch` appears only among its own, never interleaved with `save`,
and `test_protect_terminal_keeps_a_concurrently_finalized_verdict` pins the
mirror-image direction only. The in-memory double replaces the whole row too,
so both implementations were faithfully wrong -- the defect was in the
contract, not either adapter.

Added: `TestRecordCompletion` (5 cases, both implementations), including the
interleaving itself; and `test_a_cron_task_survives_a_launch_write_landing_mid_completion`,
which drives a real `ScheduleService` over a repo double that commits
`record_launch` between the hook's read and its write. Both were watched
failing first -- the service case failing with the actual symptom, a rolled
back `next_run_at`, not a missing method.

Verified end to end against file-backed sqlite with the real service: the
task keeps `enabled`, the concurrent fire time and run count survive, the
verdict is recorded, and the next poll still claims it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/AGENTS.md

@@ -18,6 +18,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu
 - With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|<ns>`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events.
 - Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack.
 - Scheduled-task dispatch enforces "at most one active run per task when `overlap_policy=skip`" at the DB layer via the partial unique index `uq_scheduled_task_run_active` (`scheduled_task_runs.task_id WHERE status IN ('queued','running')`). `ScheduledTaskService.dispatch_task`'s `has_active_runs` check is a non-atomic fast path (its own session, separated from the `create()` insert by `await` points), so two concurrent dispatches — a manual `POST /scheduled-tasks/{id}/trigger` racing the poller, a double-click, or a client retry — can both pass it; the index is the atomic arbiter, and the losing `create` surfaces as `ActiveScheduledRunConflict` (translated from `IntegrityError` in the repository) and collapses to the same outcome as the fast path (manual → 409 conflict, scheduled → a `"skipped"` tombstone). The scheduled-skip tombstone is created directly as terminal `"skipped"` (not a transient `"queued"`) so it never occupies the active slot the pre-existing run still holds. Sibling of the `runs` table's `uq_runs_thread_active` (PR #4003), which keys on `thread_id` and so does not cover the default `fresh_thread_per_run` context where every dispatch gets a new thread. Index is status-only, not `overlap_policy`-conditional (the policy is fixed to `"skip"` in the MVP).
+- The dispatch path and the run-completion hook write to the same `scheduled_tasks` row concurrently, so **the two writes own disjoint field sets and neither may write through the whole aggregate**. A run that fails fast reaches its completion hook before the dispatch path's own bookkeeping write lands, so the hook is holding a snapshot taken *before* it. The launch write (`record_launch`) owns the schedule — `next_run_at`, `last_run_at`, `last_run_id`, `last_thread_id`, `run_count`, the claim — and the completion write (`record_completion`) owns only the verdict: the terminal status and `last_error`. Expressing the completion as a read-modify-write through the aggregate replays the stale snapshot and rolls the launch write back, restoring an elapsed `next_run_at` and leaving the task in `running` with no live claim — a shape neither `claim_due` branch nor `cancel_stuck_once_tasks` can reach, i.e. a cron task that is permanently unschedulable with no recovery path. `record_launch`'s `protect_terminal` flag closes the mirror-image race in the other direction (the hook's terminal status must survive a launch write landing after it). Pinned by the `TestRecordCompletion` contract cases (both the in-memory double and real sqlite) and by `test_a_cron_task_survives_a_launch_write_landing_mid_completion`.
 
 **Project Structure**:
 ```

backend/app/adapters/schedule/scheduled_task_repository.py

@@ -259,6 +259,28 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository):
             row.updated_at = datetime.now(UTC)
             await session.commit()
 
+    async def record_completion(
+        self,
+        task_id: str,
+        *,
+        user_id: str,
+        status: TaskStatus | None,
+        error: str | None,
+    ) -> None:
+        async with self._sf() as session:
+            row = await session.get(ScheduledTaskRow, task_id)
+            if row is None or row.user_id != user_id:
+                return
+            # Field-level on purpose: `record_launch` may commit on either side
+            # of this write, and the two must not undo each other. Assigning
+            # anything below `last_error` here is what reintroduces the race --
+            # see the port docstring.
+            row.last_error = error
+            if status is not None:
+                row.status = str(status)
+            row.updated_at = datetime.now(UTC)
+            await session.commit()
+
     async def cancel_stuck_once_tasks(self, *, error: str) -> int:
         """Reconcile `once` tasks orphaned in `running` by a process crash.
 

backend/packages/harness/deerflow/domain/schedule/ports.py

@@ -155,6 +155,40 @@ class ScheduledTaskRepository(Protocol):
         """
         ...
 
+    async def record_completion(
+        self,
+        task_id: str,
+        *,
+        user_id: str,
+        status: TaskStatus | None,
+        error: str | None,
+    ) -> None:
+        """Write a finished run's verdict onto the task.
+
+        Deliberately NOT expressed as `save(task)`, for the same reason
+        `record_launch` is not: the two race. A run that fails fast reaches its
+        completion hook while the dispatch path is still writing its
+        bookkeeping, so a read-modify-write through the aggregate would replay
+        a snapshot taken before that write and roll it back -- restoring an
+        elapsed `next_run_at` and an out-of-date `run_count`, and leaving the
+        task in `running` with no live claim, which neither `claim_due` branch
+        nor `cancel_stuck_once_tasks` can reach.
+
+        The two writes therefore own disjoint fields: the launch owns the
+        schedule, this owns the verdict. Nothing here touches `next_run_at`,
+        `last_run_at`, `last_run_id`, `last_thread_id`, `run_count` or the
+        claim.
+
+        `status` is `None` when the verdict must not move the status -- every
+        cron task, whose schedule outlives any single run. `error` is written
+        either way, so a cron task still reports what went wrong last time.
+
+        Scoped by `user_id` like every other read: a task belonging to someone
+        else is left alone rather than reported. An unknown task is ignored --
+        one deleted mid-flight simply has nothing to update.
+        """
+        ...
+
     async def cancel_stuck_once_tasks(self, *, error: str) -> int:
         """Reconcile `once` tasks orphaned mid-flight by a process crash.
 

backend/packages/harness/deerflow/domain/schedule/service.py

@@ -350,16 +350,22 @@ class ScheduleService:
             error=outcome.error,
             finished_at=now,
         )
+        # Read to ask the aggregate what this outcome means, then write only
+        # that. `status_after_completion` reads nothing but the schedule type,
+        # which no concurrent write can change, so this read carries no
+        # time-of-check risk -- and `record_completion` deliberately does not
+        # write the fields that a concurrent `record_launch` owns.
         task = await self._tasks.get(outcome.task_id, user_id=outcome.user_id)
         if task is None:
             return
         # The error is recorded whether or not the status moves: a cron task
         # keeps its schedule but still reports what went wrong last time.
-        updated = replace(task, last_error=outcome.error)
-        new_status = task.status_after_completion(outcome.status)
-        if new_status is not None:
-            updated = replace(updated, status=new_status)
-        await self._tasks.save(updated)
+        await self._tasks.record_completion(
+            outcome.task_id,
+            user_id=outcome.user_id,
+            status=task.status_after_completion(outcome.status),
+            error=outcome.error,
+        )
 
     async def reconcile_on_startup(self, *, error: str) -> tuple[int, int]:
         """Clean up what a process crash left behind, returning what was fixed.

backend/tests/schedule_fakes.py

@@ -154,6 +154,22 @@ class InMemoryScheduledTaskRepository:
         row.lease_owner = None
         row.lease_expires_at = None
 
+    async def record_completion(
+        self,
+        task_id: str,
+        *,
+        user_id: str,
+        status: TaskStatus | None,
+        error: str | None,
+    ) -> None:
+        row = self._rows.get(task_id)
+        if row is None or row.task.user_id != user_id:
+            return
+        # Only the verdict; every scheduling field belongs to record_launch.
+        row.task = replace(row.task, last_error=error)
+        if status is not None:
+            row.task = replace(row.task, status=status)
+
     async def cancel_stuck_once_tasks(self, *, error: str) -> int:
         cancelled = 0
         for row in self._rows.values():

backend/tests/test_schedule_fakes.py

@@ -352,6 +352,84 @@ class TestRecordLaunch:
         )
 
 
+class TestRecordCompletion:
+    """The completion hook's write.
+
+    Deliberately as narrow as `record_launch` is, and for the same reason: the
+    two race, so neither may write through the whole aggregate. This one owns
+    the terminal verdict and nothing else -- every scheduling field belongs to
+    the launch path, which may commit at any point around it.
+    """
+
+    async def test_records_the_terminal_status_and_error(self, tasks):
+        await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=ONCE))
+
+        await tasks.record_completion("task-1", user_id="user-1", status=TaskStatus.FAILED, error="boom")
+
+        task = await tasks.get("task-1", user_id="user-1")
+        assert task.status is TaskStatus.FAILED
+        assert task.last_error == "boom"
+
+    async def test_a_none_status_records_the_error_and_leaves_the_status(self, tasks):
+        """A cron task's schedule outlives any single run, so only what went
+        wrong is recorded."""
+        await tasks.add(make_task(status=TaskStatus.ENABLED, schedule=CRON))
+
+        await tasks.record_completion("task-1", user_id="user-1", status=None, error="boom")
+
+        task = await tasks.get("task-1", user_id="user-1")
+        assert task.status is TaskStatus.ENABLED
+        assert task.last_error == "boom"
+
+    async def test_never_rolls_back_a_concurrent_launch_write(self, tasks):
+        """The regression this method exists to prevent.
+
+        A fast-failing run reaches the completion hook while the dispatch path
+        is still writing its bookkeeping. Whichever lands second must not undo
+        the other: the launch owns the schedule, the completion owns the
+        verdict.
+        """
+        await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
+        await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)
+
+        next_at = NOW + timedelta(days=1)
+        await tasks.record_launch(
+            "task-1",
+            status=TaskStatus.ENABLED,
+            next_run_at=next_at,
+            last_run_at=NOW,
+            last_run_id="run-1",
+            last_thread_id="thread-1",
+            last_error=None,
+            increment_run_count=True,
+            protect_terminal=True,
+        )
+
+        await tasks.record_completion("task-1", user_id="user-1", status=None, error="boom")
+
+        task = await tasks.get("task-1", user_id="user-1")
+        assert task.next_run_at == next_at, "the launch path's next fire time must survive"
+        assert task.run_count == 1, "the launch path's run count must survive"
+        assert task.last_run_id == "run-1"
+        assert task.last_thread_id == "thread-1"
+        assert task.last_error == "boom", "the completion still records its verdict"
+        # The whole point: the task is still reachable by the next poll.
+        claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10)
+        assert [t.task_id for t in claimed] == ["task-1"]
+
+    async def test_another_users_task_is_untouched(self, tasks):
+        await tasks.add(make_task(status=TaskStatus.ENABLED))
+
+        await tasks.record_completion("task-1", user_id="someone-else", status=TaskStatus.FAILED, error="boom")
+
+        task = await tasks.get("task-1", user_id="user-1")
+        assert task.status is TaskStatus.ENABLED
+        assert task.last_error is None
+
+    async def test_unknown_task_is_ignored(self, tasks):
+        await tasks.record_completion("nope", user_id="user-1", status=TaskStatus.FAILED, error="boom")
+
+
 class TestCancelStuckOnceTasks:
     async def test_cancels_a_launched_once_task_with_no_claim(self, tasks):
         """Launched, so the claim was released; the completion hook then died

backend/tests/test_schedule_service.py

@@ -64,6 +64,42 @@ class _BlindRunRepo(InMemoryScheduledRunRepository):
         return False
 
 
+class _LaunchWritesMidReadTaskRepo(InMemoryScheduledTaskRepository):
+    """`record_launch` commits between the completion hook's read and its write.
+
+    That ordering is not exotic -- `dispatch_task` performs its two bookkeeping
+    writes after the launch returns, and a run that fails fast reaches the hook
+    in between. Modelling it here rather than with real concurrency keeps the
+    window deterministic.
+    """
+
+    def __init__(self, *, launch_next_run_at: datetime) -> None:
+        super().__init__()
+        self._launch_next_run_at = launch_next_run_at
+        self._armed = False
+
+    def arm(self) -> None:
+        """Fire the interleaving on the next read, once."""
+        self._armed = True
+
+    async def get(self, task_id: str, *, user_id: str):
+        task = await super().get(task_id, user_id=user_id)
+        if self._armed:
+            self._armed = False
+            await self.record_launch(
+                task_id,
+                status=TaskStatus.ENABLED,
+                next_run_at=self._launch_next_run_at,
+                last_run_at=NOW,
+                last_run_id="run-1",
+                last_thread_id="thread-1",
+                last_error=None,
+                increment_run_count=True,
+                protect_terminal=True,
+            )
+        return task
+
+
 def make_service(
     *,
     tasks: InMemoryScheduledTaskRepository | None = None,
@@ -473,6 +509,42 @@ class TestRunCompletion:
             now=NOW,
         )
 
+    async def test_a_cron_task_survives_a_launch_write_landing_mid_completion(self):
+        """The interleaving `dispatch_task` already admits is possible.
+
+        A run that fails fast reaches this hook while the dispatch path has not
+        yet written its bookkeeping, so the hook reads a task still carrying the
+        elapsed `next_run_at`. If the write-back replays that whole snapshot,
+        the launch path's fresh fire time is rolled back and the task lands in
+        `running` with no live claim -- a shape neither `claim_due` branch nor
+        `cancel_stuck_once_tasks` can reach, i.e. permanently unschedulable.
+        """
+        tasks = _LaunchWritesMidReadTaskRepo(launch_next_run_at=NOW + timedelta(days=1))
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(tasks=tasks, runs=runs)
+        task = await create_cron_task(service)
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        record = runs.all_runs()[0]
+        tasks.arm()
+
+        await service.handle_run_completion(
+            RunOutcome(
+                task_id=task.task_id,
+                record_id=record.record_id,
+                run_id="run-1",
+                user_id="user-1",
+                status=RunStatus.FAILED,
+                error="boom",
+            ),
+            now=NOW,
+        )
+
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.last_error == "boom", "the completion still records its verdict"
+        assert after.next_run_at == NOW + timedelta(days=1), "the launch path's fire time must survive"
+        claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10)
+        assert [t.task_id for t in claimed] == [task.task_id], "the task must remain schedulable"
+
 
 class TestReconcileOnStartup:
     async def test_sweeps_orphaned_runs_and_stuck_once_tasks(self):