feat(schedule): complete the outer ring with the launch adapters

Adds the three remaining adapters plus the poller. Nothing is wired yet --
the composition root is the next commit -- so this is additive and the
legacy `app/scheduler/service.py` still serves production.

`run_launcher.py` is the pivot of the whole slice. The Gateway signals a
busy thread two ways -- `ConflictError` from the run manager, or an
`HTTPException(409)` from the route-level path -- which is why the legacy
scheduler service imported fastapi to tell them apart. Both are one domain
fact, and saying so here is what lets that import disappear without the
busy/failed distinction disappearing with it. Everything else becomes
`LaunchFailedError`, because the port promises the domain that nothing but
its two errors escapes. `CancelledError` is deliberately not caught:
shutdown is control flow, not a launch outcome.

`thread_lookup.py` narrows `ThreadMetaStore` to the one question this
context asks. `require_existing=True` is load-bearing -- the store's
default treats an absent row as accessible, which is right for a thread
not yet written and wrong for binding a task to it.

Both inherit their port explicitly, matching every other adapter in the
codebase including feedback's own anti-corruption layer, and both carry
the TODO naming the published contract that would replace them once the
upstream context has been through a slice of its own.

`run_outcome_mapping.py` implements no port: it is the inbound translation
the composition root will install on the completion hook, and it owns the
filtering the legacy hook did inline. Returning None means "this run is
none of the schedule context's business", so the service is simply never
called and needs no guard clauses.

`poller.py` keeps the two behaviours the legacy loop got right: a failing
poll must not end the loop (one transient "database is locked" used to
stop scheduling for the rest of the process life), and reconciliation must
not block startup.

One deliberate behaviour change: the legacy `start()` swept stale runs and
stuck once-tasks under separate try/excepts, so the first failing did not
stop the second. `reconcile_on_startup` is one call that lets failures
propagate -- the domain's position is that fatality is the caller's policy
-- so the poller's single except means a failed first sweep now skips the
second. Both end up logged and non-fatal, as before.

Tests: 50 new cases across the four modules, each port method called and
asserted on its return value. That is not decoration: inheriting a
Protocol means a misspelled method silently inherits its `...` body and
returns None, so the suite was verified by mutation -- renaming `launch`
and `exists_for_user` turns 16 and 6 cases red respectively.

backend/app/adapters/schedule/run_launcher.py

@@ -0,0 +1,89 @@
+"""Secondary adapter (anti-corruption layer) -- starting a run through Gateway.
+
+Implements ``RunLauncher`` from ``deerflow.domain.schedule.ports``. This context
+owns no part of the run lifecycle: it asks the Gateway to start one and
+translates whatever comes back into the two outcomes the domain distinguishes.
+
+That translation is the reason this file exists. The Gateway signals a busy
+thread two different ways -- ``ConflictError`` from the run manager, or an
+``HTTPException(409)`` from the route-level path -- and the legacy scheduler
+service therefore imported ``fastapi`` to tell them apart. Both are the same
+domain fact, and saying so here is what keeps the web framework and the run
+runtime out of the inner ring.
+
+TODO(hexagonal): this depends on ``launch_scheduled_thread_run``, a Gateway
+service function returning an untyped dict, rather than on a contract published
+by the run context -- that context has not been through a hexagonal slice yet.
+When it publishes one (a DTO, not its aggregate and not its repository),
+replace the body of this class. The ``RunLauncher`` port does not move.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Any
+
+from fastapi import HTTPException
+
+from deerflow.domain.schedule.model import LaunchFailedError, ThreadBusyError
+from deerflow.domain.schedule.ports import LaunchedRun, RunLauncher
+from deerflow.runtime import ConflictError
+
+LaunchRun = Callable[..., Awaitable[Mapping[str, Any]]]
+
+
+class GatewayRunLauncher(RunLauncher):
+    """Adapts the Gateway's scheduled-run launch path to the ``RunLauncher`` port.
+
+    Takes the launch callable rather than importing it, because the production
+    one is bound to the FastAPI app (``launch_scheduled_thread_run(app=app,
+    ...)``) and that binding belongs to the composition root.
+
+    Explicit inheritance is a readability aid only: a misspelled method would
+    still instantiate fine and silently inherit the Protocol's ``...`` body,
+    so the contract tests must call every port method and assert on what it
+    returns.
+    """
+
+    def __init__(self, launch_run: LaunchRun) -> None:
+        self._launch_run = launch_run
+
+    async def launch(
+        self,
+        *,
+        thread_id: str,
+        assistant_id: str | None,
+        prompt: str,
+        owner_user_id: str | None,
+        metadata: dict[str, str],
+    ) -> LaunchedRun:
+        try:
+            result = await self._launch_run(
+                thread_id=thread_id,
+                assistant_id=assistant_id,
+                prompt=prompt,
+                owner_user_id=owner_user_id,
+                metadata=metadata,
+            )
+        except ConflictError as exc:
+            raise ThreadBusyError(str(exc)) from exc
+        except HTTPException as exc:
+            if exc.status_code == 409:
+                raise ThreadBusyError(str(exc.detail)) from exc
+            raise LaunchFailedError(str(exc.detail)) from exc
+        except Exception as exc:
+            # Deliberately broad: the port promises the domain that nothing but
+            # its two errors escapes, so an unclassifiable failure has to become
+            # the "genuine failure" branch rather than unwinding the poll loop.
+            # `CancelledError` derives from BaseException and is not caught --
+            # shutdown is control flow, not a launch outcome.
+            raise LaunchFailedError(str(exc)) from exc
+
+        run_id = result.get("run_id")
+        launched_thread_id = result.get("thread_id")
+        if not isinstance(run_id, str) or not isinstance(launched_thread_id, str):
+            # The run path broke its own contract. Reporting it as a failure
+            # keeps the task's bookkeeping honest instead of recording a launch
+            # whose run can never be traced.
+            raise LaunchFailedError(f"run launch returned no usable identity: {result!r}")
+        return LaunchedRun(run_id=run_id, thread_id=launched_thread_id)

backend/app/adapters/schedule/run_outcome_mapping.py

@@ -0,0 +1,74 @@
+"""Boundary mapping (not a port implementation) -- RunRecord -> RunOutcome.
+
+Unlike its siblings in this package, this module implements no port: it is the
+inbound translation the composition root installs on the run runtime's
+completion hook, so the domain never imports ``RunRecord``.
+
+It also owns the filtering the legacy completion hook did inline. Every run in
+the process reaches that hook, so most of them are none of this context's
+business, and returning ``None`` says exactly that -- not an error, just
+nothing to write back. The service is then never called at all, which is why it
+has no guard clauses of its own.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deerflow.domain.schedule.model import RunStatus
+from deerflow.domain.schedule.ports import RunOutcome
+
+if TYPE_CHECKING:
+    from deerflow.runtime import RunRecord
+
+# The runtime reports four terminal states; the domain has three, because
+# `timeout` and `error` are the same fact to a scheduled task while
+# `interrupted` is deliberately not -- a cancel or same-thread takeover ends
+# the task CANCELLED, not FAILED.
+_TERMINAL_STATUSES = {
+    "success": RunStatus.SUCCESS,
+    "error": RunStatus.FAILED,
+    "timeout": RunStatus.FAILED,
+    "interrupted": RunStatus.INTERRUPTED,
+}
+
+_INTERRUPTED_WITHOUT_ERROR = "run was interrupted before completion"
+
+
+def run_outcome_from_record(record: RunRecord) -> RunOutcome | None:
+    """Translate a finished run into domain vocabulary, or `None` to ignore it.
+
+    `None` is returned when the run is not a scheduled execution (no usable
+    task metadata, no owner) or has not reached a terminal state yet.
+    """
+    metadata = record.metadata or {}
+    task_id = metadata.get("scheduled_task_id")
+    record_id = metadata.get("scheduled_task_run_id")
+    user_id = record.user_id
+    # `metadata` is a free-form dict a caller can influence, so the ids are
+    # type-checked rather than assumed; `user_id` is required because every
+    # task read is scoped by it.
+    if not isinstance(task_id, str) or not isinstance(record_id, str) or not user_id:
+        return None
+
+    status = _TERMINAL_STATUSES.get(str(record.status.value))
+    if status is None:
+        return None
+
+    if status is RunStatus.SUCCESS:
+        # A stale error left on a successful record must not be written back as
+        # the task's last_error.
+        error = None
+    elif status is RunStatus.INTERRUPTED:
+        error = record.error or _INTERRUPTED_WITHOUT_ERROR
+    else:
+        error = record.error
+
+    return RunOutcome(
+        task_id=task_id,
+        record_id=record_id,
+        run_id=record.run_id,
+        user_id=user_id,
+        status=status,
+        error=error,
+    )

backend/app/adapters/schedule/thread_lookup.py

@@ -0,0 +1,46 @@
+"""Secondary adapter (anti-corruption layer) -- narrowing ThreadMetaStore to ThreadLookup.
+
+Implements ``ThreadLookup`` from ``deerflow.domain.schedule.ports``. This
+context does not own the ``threads_meta`` table and writes no SQL against it: it
+asks its one question through the store the thread context already provides.
+
+``require_existing=True`` is the load-bearing argument. The store's default
+treats an absent row as accessible -- reasonable for a thread that has not been
+written yet, wrong for binding a task to it, since the task would then reference
+a thread that never existed.
+
+TODO(hexagonal): this depends on ``ThreadMetaStore``, an infrastructure
+component, rather than on a contract published by the thread context -- that
+context has not been through a hexagonal slice yet. When it publishes one (a
+DTO, not its aggregate and not its repository), replace the body of this class.
+The ``ThreadLookup`` port does not move.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deerflow.domain.schedule.ports import ThreadLookup
+
+if TYPE_CHECKING:
+    from deerflow.persistence.thread_meta.base import ThreadMetaStore
+
+
+class ThreadStoreThreadLookup(ThreadLookup):
+    """Adapts the wide ``ThreadMetaStore`` to the one question this context asks.
+
+    Both halves of that question -- does the thread exist, and may this user use
+    it -- collapse into a single bool on purpose: reporting them separately
+    would let a caller probe for the existence of threads they cannot see.
+
+    Explicit inheritance is a readability aid only: a misspelled method would
+    still instantiate fine and silently inherit the Protocol's ``...`` body,
+    so the contract tests must call every port method and assert on what it
+    returns.
+    """
+
+    def __init__(self, thread_store: ThreadMetaStore) -> None:
+        self._thread_store = thread_store
+
+    async def exists_for_user(self, thread_id: str, user_id: str) -> bool:
+        return bool(await self._thread_store.check_access(thread_id, user_id, require_existing=True))

backend/app/scheduler/poller.py

@@ -0,0 +1,84 @@
+"""Driving adapter -- the clock that asks the schedule service to work.
+
+This is the only part of the scheduler that knows about time passing. It owns
+*when* `ScheduleService.run_once` is called and what happens when a poll fails;
+it owns nothing about what a poll means. Everything the old
+`app/scheduler/service.py` mixed into its loop -- overlap policy, lease
+semantics, budget accounting -- now lives in the domain and reaches this file
+only as one awaited call.
+
+Two behaviours here are load-bearing:
+
+  - **A failing poll must not end the loop.** A transient error (SQLite's
+    "database is locked" is the realistic one) would otherwise stop every
+    scheduled task for the rest of the process life, silently.
+  - **Startup reconciliation must not block startup.** The service lets
+    reconcile failures propagate on purpose -- whether they are fatal is the
+    caller's policy -- and this caller's policy is to log and keep scheduling.
+    A gateway refusing to start over leftover rows is worse than one running
+    with them.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from deerflow.domain.schedule.service import ScheduleService
+
+logger = logging.getLogger(__name__)
+
+RESTART_ERROR = "interrupted: gateway restarted before the run reached a terminal state"
+
+
+class SchedulePoller:
+    """Runs `ScheduleService.run_once` on an interval until stopped."""
+
+    def __init__(self, service: ScheduleService, *, poll_interval_seconds: float) -> None:
+        self._service = service
+        self._poll_interval_seconds = poll_interval_seconds
+        self._task: asyncio.Task | None = None
+        self._stop = asyncio.Event()
+
+    async def start(self) -> None:
+        """Reconcile what a crash left behind, then begin polling.
+
+        Idempotent: a second call while running is a no-op, so a caller cannot
+        end up with two loops claiming the same tasks.
+        """
+        if self._task is not None:
+            return
+        try:
+            stale_runs, stuck_tasks = await self._service.reconcile_on_startup(error=RESTART_ERROR)
+            if stale_runs:
+                logger.warning("Marked %d stale scheduled task run(s) as interrupted after restart", stale_runs)
+            if stuck_tasks:
+                logger.warning("Cancelled %d stuck once task(s) after restart", stuck_tasks)
+        except Exception:
+            logger.exception("Failed to reconcile scheduled tasks at startup; scheduling anyway")
+        self._stop.clear()
+        self._task = asyncio.create_task(self._run_loop())
+
+    async def stop(self) -> None:
+        """Signal the loop and wait for the in-flight poll to finish."""
+        if self._task is None:
+            return
+        self._stop.set()
+        await self._task
+        self._task = None
+
+    async def _run_loop(self) -> None:
+        while not self._stop.is_set():
+            try:
+                await self._service.run_once(now=datetime.now(UTC))
+            except Exception:
+                logger.exception("Scheduled task poll failed; retrying next interval")
+            try:
+                # Waiting on the stop event rather than sleeping keeps shutdown
+                # prompt at a production-sized interval.
+                await asyncio.wait_for(self._stop.wait(), timeout=self._poll_interval_seconds)
+            except TimeoutError:
+                continue

backend/tests/test_schedule_poller.py

@@ -0,0 +1,149 @@
+"""Tests for the scheduler poller.
+
+The poller owns everything about *when* the service is asked to work, and
+nothing about what the work means. Two behaviours here are load-bearing and
+were carried over deliberately from `app/scheduler/service.py`:
+
+  - a failing poll must not kill the loop for the rest of the process life
+    (a transient "database is locked" would otherwise silently stop every
+    scheduled task until the next restart), and
+  - startup reconciliation must not block startup.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime
+
+import pytest
+
+from app.scheduler.poller import SchedulePoller
+
+
+class _SpyService:
+    """Stands in for `ScheduleService`, recording how the poller drove it."""
+
+    def __init__(self, *, run_once_error: Exception | None = None, reconcile_error: Exception | None = None) -> None:
+        self.run_once_error = run_once_error
+        self.reconcile_error = reconcile_error
+        self.run_once_calls: list[datetime] = []
+        self.reconcile_calls: list[str] = []
+        self.polled = asyncio.Event()
+
+    async def run_once(self, *, now: datetime) -> list:
+        self.run_once_calls.append(now)
+        self.polled.set()
+        if self.run_once_error is not None:
+            raise self.run_once_error
+        return []
+
+    async def reconcile_on_startup(self, *, error: str) -> tuple[int, int]:
+        self.reconcile_calls.append(error)
+        if self.reconcile_error is not None:
+            raise self.reconcile_error
+        return 2, 1
+
+
+async def _drain(poller: SchedulePoller, service: _SpyService, *, polls: int = 1) -> None:
+    """Start the poller, wait for it to poll, then stop it."""
+    await poller.start()
+    try:
+        for _ in range(polls):
+            service.polled.clear()
+            await asyncio.wait_for(service.polled.wait(), timeout=2)
+    finally:
+        await poller.stop()
+
+
+class TestStartupReconciliation:
+    @pytest.mark.asyncio
+    async def test_start_reconciles_before_polling(self):
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await _drain(poller, service)
+        assert len(service.reconcile_calls) == 1
+        assert "restart" in service.reconcile_calls[0]
+
+    @pytest.mark.asyncio
+    async def test_a_failed_reconcile_does_not_block_the_loop(self):
+        """Whether a partial reconcile blocks startup is the caller's policy,
+        and this caller's policy is: log it and keep scheduling. A gateway that
+        refuses to start because of leftover rows is worse than one that runs
+        with them."""
+        service = _SpyService(reconcile_error=RuntimeError("db down"))
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await _drain(poller, service)
+        assert service.run_once_calls
+
+
+class TestPolling:
+    @pytest.mark.asyncio
+    async def test_polls_repeatedly(self):
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await _drain(poller, service, polls=3)
+        assert len(service.run_once_calls) >= 3
+
+    @pytest.mark.asyncio
+    async def test_each_poll_passes_a_tz_aware_now(self):
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await _drain(poller, service)
+        assert all(now.tzinfo is not None for now in service.run_once_calls)
+
+    @pytest.mark.asyncio
+    async def test_a_failing_poll_does_not_kill_the_loop(self):
+        """The regression this guards: one transient DB error used to end
+        scheduling for the rest of the process life."""
+        service = _SpyService(run_once_error=RuntimeError("database is locked"))
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await _drain(poller, service, polls=3)
+        assert len(service.run_once_calls) >= 3
+
+
+class TestLifecycle:
+    @pytest.mark.asyncio
+    async def test_stop_is_awaited_to_completion(self):
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await poller.start()
+        await asyncio.wait_for(service.polled.wait(), timeout=2)
+        await poller.stop()
+        before = len(service.run_once_calls)
+        await asyncio.sleep(0.05)
+        assert len(service.run_once_calls) == before
+
+    @pytest.mark.asyncio
+    async def test_stop_does_not_wait_out_the_poll_interval(self):
+        """The loop waits on a stop event rather than sleeping, so shutdown is
+        prompt even with a production-sized interval."""
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=300)
+        await poller.start()
+        await asyncio.wait_for(service.polled.wait(), timeout=2)
+        await asyncio.wait_for(poller.stop(), timeout=2)
+
+    @pytest.mark.asyncio
+    async def test_start_is_idempotent(self):
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await poller.start()
+        await poller.start()
+        try:
+            await asyncio.wait_for(service.polled.wait(), timeout=2)
+        finally:
+            await poller.stop()
+        assert len(service.reconcile_calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_stop_without_start_is_a_no_op(self):
+        poller = SchedulePoller(_SpyService(), poll_interval_seconds=0.01)
+        await poller.stop()
+
+    @pytest.mark.asyncio
+    async def test_it_can_be_restarted(self):
+        service = _SpyService()
+        poller = SchedulePoller(service, poll_interval_seconds=0.01)
+        await _drain(poller, service)
+        await _drain(poller, service)
+        assert len(service.reconcile_calls) == 2

backend/tests/test_schedule_run_launcher.py

@@ -0,0 +1,149 @@
+"""Contract tests for the run-launcher anti-corruption layer.
+
+The `RunLauncher` port allows exactly two exceptions to escape -- `ThreadBusyError`
+and `LaunchFailedError` -- and the domain branches on the difference: a busy
+thread on a scheduled dispatch is a *skipped* occurrence, a genuine failure is a
+*recorded* one. Everything the Gateway can raise is therefore classified here,
+and this file is what pins that classification.
+
+That translation is the whole point of the adapter: it is what lets
+`app/scheduler/service.py`'s `from fastapi import HTTPException` disappear
+without the busy/failed distinction disappearing with it.
+
+There is deliberately no `isinstance(launcher, RunLauncher)` assertion. The
+adapter inherits the port explicitly, which makes that check trivially true --
+and worse than useless: inheritance is exactly what turns a misspelled method
+into a silent inherited `...` body returning `None`. Calling every port method
+and asserting on what it returns, as this file does, is what actually catches
+that.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+
+from app.adapters.schedule.run_launcher import GatewayRunLauncher
+from deerflow.domain.schedule.model import LaunchFailedError, ThreadBusyError
+from deerflow.domain.schedule.ports import LaunchedRun
+from deerflow.runtime import ConflictError
+
+LAUNCH_KWARGS = {
+    "thread_id": "thread-1",
+    "assistant_id": "assistant-1",
+    "prompt": "do the thing",
+    "owner_user_id": "user-1",
+    "metadata": {"scheduled_task_id": "task-1", "scheduled_task_run_id": "rec-1", "scheduled_trigger": "scheduled"},
+}
+
+
+def _launcher_returning(payload):
+    async def launch_run(**kwargs):
+        launch_run.calls.append(kwargs)
+        return payload
+
+    launch_run.calls = []
+    return GatewayRunLauncher(launch_run), launch_run
+
+
+def _launcher_raising(exc):
+    async def launch_run(**kwargs):
+        raise exc
+
+    return GatewayRunLauncher(launch_run)
+
+
+class TestSuccessfulLaunch:
+    @pytest.mark.asyncio
+    async def test_returns_what_the_gateway_reported(self):
+        launcher, _ = _launcher_returning({"run_id": "run-9", "thread_id": "thread-other"})
+        result = await launcher.launch(**LAUNCH_KWARGS)
+        assert result == LaunchedRun(run_id="run-9", thread_id="thread-other")
+
+    @pytest.mark.asyncio
+    async def test_echoes_the_gateways_thread_not_the_requested_one(self):
+        """`LaunchedRun.thread_id` is documented as what actually ran, so the
+        adapter must not substitute the thread it asked for."""
+        launcher, _ = _launcher_returning({"run_id": "run-9", "thread_id": "thread-substituted"})
+        result = await launcher.launch(**LAUNCH_KWARGS)
+        assert result.thread_id == "thread-substituted"
+
+    @pytest.mark.asyncio
+    async def test_carries_every_argument_through_untouched(self):
+        launcher, spy = _launcher_returning({"run_id": "r", "thread_id": "t"})
+        await launcher.launch(**LAUNCH_KWARGS)
+        assert spy.calls == [LAUNCH_KWARGS]
+
+    @pytest.mark.asyncio
+    async def test_a_malformed_gateway_payload_is_a_launch_failure(self):
+        """A missing id is not a busy thread -- it is the run path breaking its
+        own contract, which the domain records as a failure."""
+        launcher, _ = _launcher_returning({"thread_id": "t"})
+        with pytest.raises(LaunchFailedError):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+
+class TestBusyThreadTranslation:
+    @pytest.mark.asyncio
+    async def test_conflict_error_becomes_thread_busy(self):
+        launcher = _launcher_raising(ConflictError("thread already has an active run"))
+        with pytest.raises(ThreadBusyError):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+    @pytest.mark.asyncio
+    async def test_http_409_becomes_thread_busy(self):
+        """`start_run` rejects a busy thread as an HTTP 409 rather than a
+        ConflictError on some paths; both mean the same thing here."""
+        launcher = _launcher_raising(HTTPException(status_code=409, detail="thread is busy"))
+        with pytest.raises(ThreadBusyError):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+    @pytest.mark.asyncio
+    async def test_the_cause_survives_in_the_message(self):
+        launcher = _launcher_raising(ConflictError("thread already has an active run"))
+        with pytest.raises(ThreadBusyError, match="thread already has an active run"):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+    @pytest.mark.asyncio
+    async def test_http_409_message_is_the_detail_not_the_repr(self):
+        launcher = _launcher_raising(HTTPException(status_code=409, detail="thread is busy"))
+        with pytest.raises(ThreadBusyError, match="^thread is busy$"):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+
+class TestFailureTranslation:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("status_code", [400, 404, 422, 500, 502])
+    async def test_any_other_http_error_is_a_launch_failure(self, status_code):
+        launcher = _launcher_raising(HTTPException(status_code=status_code, detail="nope"))
+        with pytest.raises(LaunchFailedError):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+    @pytest.mark.asyncio
+    async def test_an_arbitrary_exception_is_a_launch_failure(self):
+        launcher = _launcher_raising(RuntimeError("database is on fire"))
+        with pytest.raises(LaunchFailedError, match="database is on fire"):
+            await launcher.launch(**LAUNCH_KWARGS)
+
+    @pytest.mark.asyncio
+    async def test_the_original_exception_is_chained(self):
+        """The domain only needs the two categories, but an operator reading a
+        log needs the real traceback."""
+        original = RuntimeError("database is on fire")
+        launcher = _launcher_raising(original)
+        with pytest.raises(LaunchFailedError) as caught:
+            await launcher.launch(**LAUNCH_KWARGS)
+        assert caught.value.__cause__ is original
+
+
+class TestCancellationIsNotSwallowed:
+    @pytest.mark.asyncio
+    async def test_cancelled_error_propagates(self):
+        """`CancelledError` is shutdown control flow, not a launch outcome.
+        Translating it to LaunchFailedError would record a spurious failure and
+        break cooperative cancellation of the poll loop."""
+        launcher = _launcher_raising(asyncio.CancelledError())
+        with pytest.raises(asyncio.CancelledError):
+            await launcher.launch(**LAUNCH_KWARGS)

backend/tests/test_schedule_run_outcome_mapping.py

@@ -0,0 +1,122 @@
+"""Boundary tests for the RunRecord -> RunOutcome conversion.
+
+This converter is where the completion hook's inline filtering moved to. The
+old `handle_run_completion` opened with four guard clauses and a status
+`if/elif` chain over runtime strings; the service now receives a `RunOutcome`
+or is never called at all, so all of that lives here and the domain never
+imports the run runtime.
+
+`None` therefore has one meaning: *this run is none of the schedule context's
+business*. It is not an error and must never reach the service.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.adapters.schedule.run_outcome_mapping import run_outcome_from_record
+from deerflow.domain.schedule.model import RunStatus
+from deerflow.domain.schedule.ports import RunOutcome
+from deerflow.runtime.runs.manager import RunRecord
+from deerflow.runtime.runs.schemas import DisconnectMode
+from deerflow.runtime.runs.schemas import RunStatus as RuntimeRunStatus
+
+TASK_METADATA = {
+    "scheduled_task_id": "task-1",
+    "scheduled_task_run_id": "rec-1",
+    "scheduled_trigger": "scheduled",
+}
+
+
+_DEFAULT = object()
+
+
+def _record(*, status, metadata=_DEFAULT, user_id="user-1", error=None, run_id="run-1"):
+    return RunRecord(
+        run_id=run_id,
+        thread_id="thread-1",
+        assistant_id=None,
+        status=status,
+        on_disconnect=DisconnectMode.continue_,
+        metadata=TASK_METADATA if metadata is _DEFAULT else metadata,
+        user_id=user_id,
+        error=error,
+    )
+
+
+class TestTerminalStatusMapping:
+    def test_success(self):
+        outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.success))
+        assert outcome == RunOutcome(
+            task_id="task-1",
+            record_id="rec-1",
+            run_id="run-1",
+            user_id="user-1",
+            status=RunStatus.SUCCESS,
+            error=None,
+        )
+
+    def test_a_successful_run_carries_no_error_even_if_the_record_has_one(self):
+        """A stale `error` on a successful record must not be written back as
+        the task's `last_error`; success is success."""
+        outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.success, error="stale"))
+        assert outcome is not None and outcome.error is None
+
+    @pytest.mark.parametrize("status", [RuntimeRunStatus.error, RuntimeRunStatus.timeout])
+    def test_error_and_timeout_both_become_failed(self, status):
+        outcome = run_outcome_from_record(_record(status=status, error="boom"))
+        assert outcome is not None
+        assert outcome.status is RunStatus.FAILED
+        assert outcome.error == "boom"
+
+    def test_interrupted_is_distinct_from_failed(self):
+        """Red line: a cancel or same-thread takeover is not an execution
+        failure -- the task must end CANCELLED, and only INTERRUPTED gets it
+        there."""
+        outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.interrupted))
+        assert outcome is not None and outcome.status is RunStatus.INTERRUPTED
+
+    def test_an_interrupt_without_an_error_gets_an_explanatory_one(self):
+        outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.interrupted, error=None))
+        assert outcome is not None and outcome.error == "run was interrupted before completion"
+
+    def test_an_interrupts_own_error_is_preserved(self):
+        outcome = run_outcome_from_record(_record(status=RuntimeRunStatus.interrupted, error="cancelled by user"))
+        assert outcome is not None and outcome.error == "cancelled by user"
+
+
+class TestFilteredOut:
+    @pytest.mark.parametrize("status", [RuntimeRunStatus.pending, RuntimeRunStatus.running])
+    def test_a_non_terminal_run_produces_nothing(self, status):
+        assert run_outcome_from_record(_record(status=status)) is None
+
+    def test_a_run_with_no_metadata_produces_nothing(self):
+        """An ordinary chat run reaches the same completion hook; it is not a
+        scheduled execution and must not be written back as one."""
+        assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, metadata={})) is None
+
+    @pytest.mark.parametrize(
+        "metadata",
+        [
+            {"scheduled_task_run_id": "rec-1"},
+            {"scheduled_task_id": "task-1"},
+            {"scheduled_task_id": "task-1", "scheduled_task_run_id": None},
+            {"scheduled_task_id": 7, "scheduled_task_run_id": "rec-1"},
+            {"scheduled_task_id": "task-1", "scheduled_task_run_id": 7},
+        ],
+    )
+    def test_half_or_mistyped_metadata_produces_nothing(self, metadata):
+        """Both ids are required and both must be strings -- `metadata` is a
+        free-form dict a client can influence, so its shape is not assumed."""
+        assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, metadata=metadata)) is None
+
+    @pytest.mark.parametrize("user_id", [None, ""])
+    def test_a_run_without_an_owner_produces_nothing(self, user_id):
+        """Every task read is scoped by user_id; without one there is no
+        authorized way to look the task up."""
+        assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, user_id=user_id)) is None
+
+    def test_a_none_metadata_produces_nothing(self):
+        """`metadata` is typed as a dict but the legacy hook defended against
+        `None`; keep that defence rather than rediscovering it in production."""
+        assert run_outcome_from_record(_record(status=RuntimeRunStatus.success, metadata=None)) is None

backend/tests/test_schedule_thread_lookup.py

@@ -0,0 +1,71 @@
+"""Contract tests for the thread-lookup anti-corruption layer.
+
+The port asks one question -- "does this thread exist AND may this user use
+it?" -- and deliberately answers both halves with a single bool, so a caller
+cannot probe for the existence of threads they cannot see. These tests pin that
+the adapter does not accidentally widen it back into two answers.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.adapters.schedule.thread_lookup import ThreadStoreThreadLookup
+
+
+class _RecordingThreadStore:
+    """Stands in for `ThreadMetaStore`, recording how it was asked."""
+
+    def __init__(self, owners: dict[str, str] | None = None) -> None:
+        self._owners = dict(owners or {})
+        self.calls: list[tuple[str, str, bool]] = []
+
+    async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
+        self.calls.append((thread_id, user_id, require_existing))
+        owner = self._owners.get(thread_id)
+        if owner is None:
+            # Mirrors the real store: absent rows pass a non-strict check.
+            return not require_existing
+        return owner == user_id
+
+
+class TestTheOneQuestion:
+    @pytest.mark.asyncio
+    async def test_an_owned_thread_exists_for_its_user(self):
+        lookup = ThreadStoreThreadLookup(_RecordingThreadStore({"thread-1": "user-1"}))
+        assert await lookup.exists_for_user("thread-1", "user-1") is True
+
+    @pytest.mark.asyncio
+    async def test_someone_elses_thread_does_not(self):
+        lookup = ThreadStoreThreadLookup(_RecordingThreadStore({"thread-1": "user-1"}))
+        assert await lookup.exists_for_user("thread-1", "user-2") is False
+
+    @pytest.mark.asyncio
+    async def test_a_missing_thread_does_not(self):
+        """This is the half `require_existing` buys: without it the store
+        treats an absent row as accessible, and a task could be bound to a
+        thread that does not exist."""
+        lookup = ThreadStoreThreadLookup(_RecordingThreadStore())
+        assert await lookup.exists_for_user("thread-nope", "user-1") is False
+
+    @pytest.mark.asyncio
+    async def test_missing_and_forbidden_are_indistinguishable(self):
+        lookup = ThreadStoreThreadLookup(_RecordingThreadStore({"thread-1": "user-1"}))
+        missing = await lookup.exists_for_user("thread-nope", "user-2")
+        forbidden = await lookup.exists_for_user("thread-1", "user-2")
+        assert missing == forbidden is False
+
+
+class TestHowTheStoreIsAsked:
+    @pytest.mark.asyncio
+    async def test_require_existing_is_always_set(self):
+        store = _RecordingThreadStore({"thread-1": "user-1"})
+        await ThreadStoreThreadLookup(store).exists_for_user("thread-1", "user-1")
+        assert store.calls == [("thread-1", "user-1", True)]
+
+    @pytest.mark.asyncio
+    async def test_the_result_is_a_real_bool(self):
+        """The port is typed `bool`; a truthy row object leaking through would
+        satisfy the domain's `if` and still be the wrong contract."""
+        store = _RecordingThreadStore({"thread-1": "user-1"})
+        assert await ThreadStoreThreadLookup(store).exists_for_user("thread-1", "user-1") is True