fix(schedule): forbid launcher thread redirection instead of recording it

LaunchedRun's docstring permitted the adapter to return a different
thread than requested, but the execution record is created with the
requested thread before the launch and update_status cannot correct
it -- a redirecting launcher left history on a thread nothing ran on
while the task pointed at the actual one.

No real adapter redirects (the Gateway launch path runs on exactly the
thread it is given), so the contract now requires launching on the
requested thread; the echoed thread_id is demoted to a verification
field. The service checks the echo: on a mismatch a run is still live
somewhere, so retention applies, the bookkeeping stays on the requested
thread the record row was created with, and the violation is surfaced
on the dispatch result and logged as an adapter bug.

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

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

@@ -26,9 +26,12 @@ from deerflow.domain.schedule.model import (
 class LaunchedRun:
     """What the launcher reports back once a run is admitted.
 
-    `thread_id` is echoed rather than assumed: the launcher is free to return a
-    different thread than the one requested, and the task's bookkeeping records
-    what actually ran.
+    `thread_id` is an echo of the requested thread, for verification only:
+    the contract requires the adapter to launch on exactly the thread it was
+    given, because the execution record is created with that thread before
+    the launch and cannot be corrected afterwards. The service treats a
+    mismatch as an adapter bug -- it keeps the bookkeeping on the requested
+    thread and surfaces the violation on the dispatch result.
     """
 
     run_id: str
@@ -325,7 +328,11 @@ class RunLauncher(Protocol):
         owner_user_id: str | None,
         metadata: dict[str, str],
     ) -> LaunchedRun:
-        """Start one execution and return its identity.
+        """Start one execution on `thread_id` and return its identity.
+
+        The run MUST be launched on the given thread -- redirection is a
+        contract violation (see LaunchedRun) -- and the echoed thread exists
+        so the service can verify that.
 
         `metadata` is opaque correlation data the domain attaches so the
         eventual outcome can be traced back to this task and record; the

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

@@ -314,7 +314,24 @@ class ScheduleService:
             # 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))
 
-        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)
+        if launched.thread_id != execution_thread_id:
+            # Contract violation: the launcher redirected the run. A run is
+            # live somewhere, so retention still applies -- but the bookkeeping
+            # stays on the requested thread the record row was created with,
+            # and the violation is surfaced instead of letting history and
+            # task silently diverge.
+            logger.error("RunLauncher violated its contract: requested thread %s, launched on %s (task %s)", execution_thread_id, launched.thread_id, task.task_id)
+            return await self._retain_launched(
+                task,
+                record_id=record.record_id,
+                run_id=launched.run_id,
+                thread_id=execution_thread_id,
+                now=now,
+                trigger=trigger,
+                error=f"launcher redirected the run to thread {launched.thread_id!r}; the contract requires the requested thread",
+            )
+
+        return await self._retain_launched(task, record_id=record.record_id, run_id=launched.run_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=None)
 
     # ------------------------------------------------------------------ lifecycle
 

backend/tests/test_schedule_service.py

@@ -53,7 +53,7 @@ from deerflow.domain.schedule.model import (
     TaskStatus,
     TriggerKind,
 )
-from deerflow.domain.schedule.ports import RunOutcome
+from deerflow.domain.schedule.ports import LaunchedRun, RunOutcome
 from deerflow.domain.schedule.service import ScheduleService
 
 pytestmark = pytest.mark.asyncio
@@ -497,6 +497,63 @@ class TestPostLaunchRetention:
         assert len(launcher.calls) == 2
 
 
+class _RedirectingLauncher(FakeRunLauncher):
+    """Violates the port contract by launching on a thread of its own choosing."""
+
+    async def launch(self, **kwargs):
+        launched = await super().launch(**kwargs)
+        return LaunchedRun(run_id=launched.run_id, thread_id="somewhere-else")
+
+
+class TestLauncherThreadEcho:
+    """The launcher must launch on the requested thread and echo it back.
+
+    The run record is created with the requested thread before the launch and
+    `update_status` cannot correct it, so a redirecting launcher would leave
+    history pointing at a thread nothing ever ran on while the task points at
+    the actual one. The contract therefore forbids redirection; the echo
+    exists to catch an adapter that breaks it.
+    """
+
+    async def test_the_echoed_thread_matches_the_record_and_the_task(self):
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs)
+        task = await create_cron_task(service)
+
+        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        after = await service.get_task(task.task_id, user_id="user-1")
+        stored = runs.all_runs()[0]
+        assert result.thread_id == stored.thread_id == after.last_thread_id
+
+    async def test_a_redirecting_launcher_is_flagged_without_diverging_the_bookkeeping(self):
+        """A run is live somewhere, so retention still applies -- but the
+        bookkeeping must stay on the requested thread the record row was
+        created with, and the violation must be surfaced, not absorbed."""
+        runs = InMemoryScheduledRunRepository()
+        launcher = _RedirectingLauncher()
+        service = make_service(runs=runs, launcher=launcher)
+        task = await create_cron_task(service)
+
+        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        assert result.outcome is DispatchOutcome.LAUNCHED
+        assert result.error is not None and "somewhere-else" in result.error
+
+        requested = launcher.calls[0]["thread_id"]
+        after = await service.get_task(task.task_id, user_id="user-1")
+        stored = runs.all_runs()[0]
+        assert result.thread_id == requested
+        assert stored.thread_id == requested
+        assert after.last_thread_id == requested, "history and task must not diverge"
+
+        # A live run exists somewhere: the slot stays held.
+        assert stored.is_active
+        second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        assert second.outcome is DispatchOutcome.SKIPPED
+        assert len(launcher.calls) == 1
+
+
 class TestConflictCollapse:
     """The fast path and the active-slot rejection must be indistinguishable.