fix(schedule): retain the active slot across post-launch failures

Port the #4504 retention semantics (#4452 duplicate-execution fix) into
the hexagonal dispatch path. Once launch() returns -- or raises without
being able to say whether a run started -- a live run may exist, so
bookkeeping failures must not release the task's single active slot:

- The two post-launch writes are best-effort: failures are logged and
  surfaced on DispatchResult.error while the outcome stays LAUNCHED.
- New LaunchIndeterminateError expresses main's launch_succeeded-before-
  unpack semantics at the port boundary: the adapter raises it when the
  side effect may have happened but the identity is unknown, and the
  service retains the slot with run_id=None.
- LaunchFailedError is narrowed to "the adapter is CERTAIN no run
  started", since that path releases the slot.

Regression tests ported from tests/test_scheduled_task_service.py.

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

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

@@ -64,4 +64,22 @@ class ThreadBusyError(ScheduleError):
 
 
 class LaunchFailedError(ScheduleError):
-    """The run could not be launched for any non-conflict reason."""
+    """The launch definitely did not start a run.
+
+    The adapter may raise this only when it is CERTAIN no run exists -- the
+    service releases the task's active slot on this path, so raising it after
+    the launch side effect may have happened reopens the #4452 duplicate
+    execution. When in doubt, raise LaunchIndeterminateError instead.
+    """
+
+
+class LaunchIndeterminateError(ScheduleError):
+    """The launch side effect may have happened, but its identity is unknown.
+
+    Raised by the RunLauncher adapter when the launch call did not fail
+    cleanly -- the response could not be decoded, the connection dropped after
+    the request was sent, and so on. The service treats this as launched with
+    an unknown run id: the execution record stays active so the task's single
+    active slot remains held (#4452 / #4504), and reconciliation later settles
+    what actually happened.
+    """

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

@@ -288,12 +288,16 @@ class RunLauncher(Protocol):
     The contract the adapter MUST honour, because it is what keeps the run
     runtime and the web framework out of the inner ring:
 
-      - the execution thread is already busy  -> ThreadBusyError
+      - the execution thread is already busy       -> ThreadBusyError
-      - anything else goes wrong              -> LaunchFailedError
+      - certainly failed, no run exists            -> LaunchFailedError
+      - a run may exist but its identity is lost   -> LaunchIndeterminateError
 
-    Nothing else may escape. The domain distinguishes those two because they
+    Nothing else may escape. The distinction is load-bearing: ThreadBusyError
-    lead to different outcomes -- a busy thread on a scheduled dispatch is a
+    on a scheduled dispatch is a skipped occurrence; LaunchFailedError
-    skipped occurrence, while a genuine failure is recorded as one.
+    releases the task's single active slot, so the adapter may raise it only
+    when it is CERTAIN no run started; LaunchIndeterminateError covers every
+    doubt after the launch request was sent -- an undecodable response, a
+    dropped connection -- and keeps the slot held (#4452 / #4504).
     """
 
     async def launch(

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

@@ -6,14 +6,16 @@ to the aggregate -- and knows nothing about HTTP, SQL, or the run runtime.
 `user_id` is always passed in explicitly; resolving the current user is the
 primary adapter's job.
 
-The dispatch path deliberately mirrors the structure of the pre-migration
+The dispatch path deliberately mirrors the structure of the legacy
-`app/scheduler/service.py` (since deleted), including its ordering and its
+`app/scheduler/service.py` (still present until the adapter slice replaces
-comments, because its concurrency and idempotency semantics are load-bearing
+it), including its ordering, its comments, and its post-launch retention
-and are pinned by tests that must keep passing unchanged.
+semantics (#4452 / #4504), because its concurrency and idempotency semantics
+are load-bearing and are pinned by tests that must keep passing unchanged.
 """
 
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass, replace
 from datetime import datetime
 
@@ -29,6 +31,7 @@ from deerflow.domain.schedule.commands import (
 from deerflow.domain.schedule.exceptions import (
     ActiveRunConflictError,
     LaunchFailedError,
+    LaunchIndeterminateError,
     TaskNotFoundError,
     ThreadBusyError,
     ThreadNotFoundError,
@@ -51,6 +54,8 @@ from deerflow.domain.schedule.ports import (
     ThreadLookup,
 )
 
+logger = logging.getLogger(__name__)
+
 # Shared so the has_active fast path and the active-slot race path produce
 # byte-identical outcomes for the same "task already has an active run"
 # condition. Two callers must not be able to tell which one rejected them.
@@ -260,10 +265,13 @@ class ScheduleService:
                 return self._conflict(execution_thread_id)
             return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
 
-        # Only the launch is guarded. The port contract admits exactly two
+        # Only the launch is guarded, and only its two certain outcomes may
-        # escapes, so a failure of the bookkeeping writes below is a genuine
+        # release the slot. Past this point a live run may exist, so every
-        # fault and propagates instead of being recorded as a failed launch --
+        # failure -- an indeterminate launch, a bookkeeping write dying -- must
-        # which would mark an already-running execution as failed.
+        # take the retention path below rather than mark the record failed:
+        # "failed" is terminal, terminal records do not hold the task's single
+        # active slot, and a released slot lets the next dispatch launch a
+        # duplicate of a run that is still alive (#4452, fixed by #4504).
         try:
             launched = await self._launcher.launch(
                 thread_id=execution_thread_id,
@@ -284,30 +292,16 @@ class ScheduleService:
                 return await self._finalize_skip(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, error=str(exc))
             return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.CONFLICT)
         except LaunchFailedError as exc:
+            # The adapter certifies no run started (see the port contract), so
+            # releasing the slot is safe.
             return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.FAILED)
+        except LaunchIndeterminateError as exc:
+            # The launch side effect may have happened but its identity could
+            # not be decoded. Retain the slot with the identity unknown;
+            # reconciliation settles what actually happened.
+            return await self._retain_launched(task, record_id=record.record_id, run_id=None, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc))
 
-        await self._runs.update_status(
+        return await self._retain_launched(task, record_id=record.record_id, run_id=launched.run_id, thread_id=launched.thread_id, now=now, trigger=trigger, error=None)
-            record.record_id,
-            status=RunStatus.RUNNING,
-            run_id=launched.run_id,
-            started_at=now,
-            # A fast-failing run can reach handle_run_completion before this
-            # write lands; never clobber its terminal verdict.
-            protect_terminal=True,
-        )
-        await self._tasks.record_launch(
-            task.task_id,
-            status=task.status_after_launch(trigger=trigger),
-            next_run_at=task.schedule.next_after(now),
-            last_run_at=now,
-            last_run_id=launched.run_id,
-            last_thread_id=launched.thread_id,
-            last_error=None,
-            increment_run_count=True,
-            # Same race as the record write above.
-            protect_terminal=True,
-        )
-        return DispatchResult(DispatchOutcome.LAUNCHED, record.record_id, launched.run_id, launched.thread_id, None)
 
     # ------------------------------------------------------------------ lifecycle
 
@@ -381,6 +375,64 @@ class ScheduleService:
             raise TaskNotFoundError("Scheduled task not found")
         return saved
 
+    async def _retain_launched(
+        self,
+        task: ScheduledTask,
+        *,
+        record_id: str,
+        run_id: str | None,
+        thread_id: str,
+        now: datetime,
+        trigger: TriggerKind,
+        error: str | None,
+    ) -> DispatchResult:
+        """Record a launch whose run is (or may be) live. Best-effort by design.
+
+        Both writes are attempted independently and their failures logged
+        rather than raised: the run is already in flight, so there is no
+        outcome these writes could change -- only bookkeeping they could lose.
+        A failed record write leaves the row queued, a failed task write
+        leaves the claim held; both states still hold the task's single
+        active slot, which is the invariant that matters (#4452). The first
+        error is surfaced on the result so the caller knows the bookkeeping
+        is behind, while the outcome stays LAUNCHED.
+        """
+        try:
+            await self._runs.update_status(
+                record_id,
+                status=RunStatus.RUNNING,
+                run_id=run_id,
+                started_at=now,
+                # A fast-failing run can reach handle_run_completion before
+                # this write lands; never clobber its terminal verdict.
+                protect_terminal=True,
+            )
+        except Exception:
+            logger.exception("Scheduled run record %s: post-launch bookkeeping failed; run %s is still live (task %s)", record_id, run_id, task.task_id)
+            if error is None:
+                error = f"post-launch bookkeeping failed for record {record_id}"
+        try:
+            await self._tasks.record_launch(
+                task.task_id,
+                status=task.status_after_launch(trigger=trigger),
+                next_run_at=task.schedule.next_after(now),
+                last_run_at=now,
+                last_run_id=run_id,
+                last_thread_id=thread_id,
+                # A bookkeeping transient is not the run's verdict: the launch
+                # succeeded, so the task list must not show an error while the
+                # run is in flight. handle_run_completion writes the real one.
+                last_error=None,
+                increment_run_count=True,
+                # Same race as the record write above.
+                protect_terminal=True,
+            )
+        except Exception:
+            logger.exception("Scheduled task %s: post-launch update failed; run %s is still live", task.task_id, run_id)
+            if error is None:
+                error = f"post-launch bookkeeping failed for task {task.task_id}"
+        return DispatchResult(DispatchOutcome.LAUNCHED, record_id, run_id, thread_id, error)
+
     def _conflict(self, thread_id: str) -> DispatchResult:
         """A manual trigger against an active run.
 

backend/tests/test_schedule_service.py

@@ -37,6 +37,7 @@ from deerflow.domain.schedule.commands import (
 from deerflow.domain.schedule.exceptions import (
     InvalidScheduleError,
     LaunchFailedError,
+    LaunchIndeterminateError,
     TaskNotFoundError,
     TaskNotMutableError,
     ThreadBusyError,
@@ -354,6 +355,147 @@ class TestDispatchOutcomes:
         assert result.record_id is not None, "the attempt itself is still recorded"
 
 
+class _RecordWriteFailsOnce(InMemoryScheduledRunRepository):
+    """`update_status` raises once -- the queued->running write dies mid-flight."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.fail_next_update = False
+
+    async def update_status(self, record_id, **kwargs):
+        if self.fail_next_update:
+            self.fail_next_update = False
+            raise RuntimeError("db down: update_status")
+        return await super().update_status(record_id, **kwargs)
+
+
+class _TaskWriteFailsOnce(InMemoryScheduledTaskRepository):
+    """`record_launch` raises once -- the parent-task write dies mid-flight."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.fail_next_record_launch = False
+
+    async def record_launch(self, task_id, **kwargs):
+        if self.fail_next_record_launch:
+            self.fail_next_record_launch = False
+            raise RuntimeError("db down: record_launch")
+        return await super().record_launch(task_id, **kwargs)
+
+
+class TestPostLaunchRetention:
+    """Regression for issue #4452 (fixed on main by #4504), ported to the
+    domain service.
+
+    Once `launch()` has returned -- or has raised without being able to say
+    whether a run started -- a live run may exist. A bookkeeping failure after
+    that point must NOT release the task's single active slot or reclassify
+    the run: the record stays active, the dispatch still reports LAUNCHED,
+    and the next dispatch must observe the held slot instead of launching a
+    duplicate.
+    """
+
+    async def test_post_launch_record_write_failure_does_not_release_active_slot(self):
+        runs = _RecordWriteFailsOnce()
+        launcher = FakeRunLauncher()
+        service = make_service(runs=runs, launcher=launcher)
+        task = await create_cron_task(service)
+        runs.fail_next_update = True
+
+        first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        # The run launched despite the bookkeeping error; the result says so
+        # and surfaces the error instead of hiding it.
+        assert first.outcome is DispatchOutcome.LAUNCHED
+        assert first.run_id == "run-1"
+        assert first.error is not None
+
+        # The row never reached RUNNING, but queued is still active: the slot
+        # is held and the next dispatch is an overlap, not a duplicate.
+        assert runs.all_runs()[0].is_active
+        second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        assert len(launcher.calls) == 1, "a duplicate launch is the #4452 bug"
+        assert second.outcome is DispatchOutcome.SKIPPED
+
+        # The bookkeeping transient is not the run's verdict: record_launch
+        # still ran, cleared last_error, and counted the execution.
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.last_error is None
+        assert after.run_count == 1
+
+    async def test_post_launch_task_write_failure_does_not_release_active_slot(self):
+        tasks = _TaskWriteFailsOnce()
+        launcher = FakeRunLauncher()
+        service = make_service(tasks=tasks, launcher=launcher)
+        task = await create_cron_task(service)
+        tasks.fail_next_record_launch = True
+
+        first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        assert first.outcome is DispatchOutcome.LAUNCHED
+        assert first.run_id == "run-1"
+        assert first.error is not None
+
+        # The record write landed before the task write failed, so the row is
+        # RUNNING with the launched id retained for recovery and cancellation.
+        stored = (await service.list_task_runs(task.task_id, user_id="user-1"))[0]
+        assert stored.status is RunStatus.RUNNING
+        assert stored.run_id == "run-1"
+
+        second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        assert len(launcher.calls) == 1, "a duplicate launch is the #4452 bug"
+        assert second.outcome is DispatchOutcome.SKIPPED
+
+    async def test_indeterminate_launch_retains_active_slot(self):
+        """The port-contract form of main's malformed-launch-result case: the
+        adapter knows the launch side effect may have happened but cannot
+        decode the run's identity. The slot must be retained with the identity
+        unknown, never released via the LaunchFailedError path."""
+        runs = InMemoryScheduledRunRepository()
+        launcher = FakeRunLauncher(fail_with=LaunchIndeterminateError("launch result undecodable"))
+        service = make_service(runs=runs, launcher=launcher)
+        task = await create_cron_task(service)
+
+        first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        assert first.outcome is DispatchOutcome.LAUNCHED
+        assert first.run_id is None, "the identity is unknown, not invented"
+        assert "launch result undecodable" in first.error
+
+        stored = runs.all_runs()[0]
+        assert stored.status is RunStatus.RUNNING
+        assert stored.run_id is None
+        assert stored.is_active
+
+        launcher.fail_with = None
+        second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        assert len(launcher.calls) == 1, "a duplicate launch is the #4452 bug"
+        assert second.outcome is DispatchOutcome.SKIPPED
+
+        # The execution is counted and the task keeps no stale identity.
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.last_run_id is None
+        assert after.run_count == 1
+
+    async def test_pre_launch_failure_still_releases_active_slot(self):
+        """Complement pinning the boundary: LaunchFailedError means the
+        adapter is CERTAIN no run started, so the failure path stays -- the
+        row goes terminal and the slot is released for the next dispatch."""
+        runs = InMemoryScheduledRunRepository()
+        launcher = FakeRunLauncher(fail_with=LaunchFailedError("provider exploded"))
+        service = make_service(runs=runs, launcher=launcher)
+        task = await create_cron_task(service)
+
+        first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        assert first.outcome is DispatchOutcome.FAILED
+        assert not runs.all_runs()[0].is_active, "a certain failure must not hold the slot"
+
+        launcher.fail_with = None
+        second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        assert second.outcome is DispatchOutcome.LAUNCHED
+        assert len(launcher.calls) == 2
+
+
 class TestConflictCollapse:
     """The fast path and the active-slot rejection must be indistinguishable.