diff --git a/backend/app/adapters/schedule/run_launcher.py b/backend/app/adapters/schedule/run_launcher.py
new file mode 100644
index 000000000..995ffaf64
--- /dev/null
+++ b/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)
diff --git a/backend/app/adapters/schedule/run_outcome_mapping.py b/backend/app/adapters/schedule/run_outcome_mapping.py
new file mode 100644
index 000000000..40f0e1b44
--- /dev/null
+++ b/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,
+    )
diff --git a/backend/app/adapters/schedule/thread_lookup.py b/backend/app/adapters/schedule/thread_lookup.py
new file mode 100644
index 000000000..9192f6f40
--- /dev/null
+++ b/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))
diff --git a/backend/app/scheduler/poller.py b/backend/app/scheduler/poller.py
new file mode 100644
index 000000000..0a1998029
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_schedule_poller.py b/backend/tests/test_schedule_poller.py
new file mode 100644
index 000000000..2db99354d
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_schedule_run_launcher.py b/backend/tests/test_schedule_run_launcher.py
new file mode 100644
index 000000000..16cac930a
--- /dev/null
+++ b/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)
diff --git a/backend/tests/test_schedule_run_outcome_mapping.py b/backend/tests/test_schedule_run_outcome_mapping.py
new file mode 100644
index 000000000..65e8152da
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_schedule_thread_lookup.py b/backend/tests/test_schedule_thread_lookup.py
new file mode 100644
index 000000000..acb71e607
--- /dev/null
+++ b/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