feat(schedule): add the schedule application service

The input port of the context: every scheduled-task use case, orchestrated
over the four output ports. It holds no business rules -- each decision is
delegated to the aggregate -- and `test_schedule_service.py` runs the
complete lifecycle with no HTTP, no database and no run runtime, which is
the acceptance criterion this migration was for.

`dispatch_task` mirrors the pre-migration structure line for line,
including its comments, and makes exactly three substitutions: bare dicts
become domain objects, the HTTPException-409 sniffing becomes `except
ThreadBusyError`, and the status-derivation static methods become
aggregate methods. Its four exits and their two conflict paths are
unchanged, and the tests pin the collapse: the fast-path rejection and
the active-slot rejection must produce identical results, asserted over
every field of DispatchResult.

One behavioural narrowing, deliberate. The old code wrapped the launch
*and* its follow-up writes in `except Exception`, so a failing bookkeeping
write was recorded as a failed launch -- marking an execution that had
actually started as failed. The port contract admits exactly two escapes
from `launch`, so only the launch is guarded now and a genuine write fault
propagates instead of being misreported.

SchedulePolicy gains max_concurrent_runs and lease_seconds. Both are
operator-tunable thresholds the domain needs but must not read, which is
what that value object is for; the claiming process's identity stays a
constructor argument since it is an identity, not a threshold.

backend/packages/harness/deerflow/domain/schedule/model/spec.py

@@ -14,9 +14,28 @@ CRON_FIELD_COUNT = 5
 
 @dataclass(frozen=True)
 class SchedulePolicy:
-    """Operator-tunable thresholds the domain needs but must not read itself."""
+    """Operator-tunable thresholds the domain needs but must not read itself.
+
+    Built by the composition root from the scheduler configuration and passed
+    in. Deliberately not held by any aggregate: a task whose meaning changes
+    with deployment config is not a domain object.
+
+    The defaults are the permissive ones on purpose -- "nobody configured a
+    policy" must not invent a business constraint. Real values only ever
+    arrive from the outer ring.
+    """
 
     min_once_delay_seconds: int = 0
+    """How far ahead a one-shot schedule must be at submission time. Read by
+    `ScheduleSpec.ensure_launchable`; a cron schedule is never subject to it."""
+
+    max_concurrent_runs: int = 1
+    """Ceiling on active scheduled executions across ALL tasks. Long runs
+    accumulate across polls, so each poll may only claim into what is left."""
+
+    lease_seconds: int = 60
+    """How long a claim on a task stays valid. Bounds how quickly a task
+    orphaned between claim and dispatch becomes reachable again."""
 
 
 @dataclass(frozen=True)

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

@@ -0,0 +1,456 @@
+"""Application service of the schedule context (its input port).
+
+Orchestrates use cases only: fetch, apply domain rules, persist through output
+ports. It holds no business rules itself -- every decision below is delegated
+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
+`app/scheduler/service.py`, including its ordering and its comments, because
+its concurrency and idempotency semantics are load-bearing and are pinned by
+tests that must keep passing unchanged.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass, replace
+from datetime import datetime
+from typing import Any, Literal
+
+from deerflow.domain.schedule.model import (
+    ActiveRunConflictError,
+    ContextMode,
+    LaunchFailedError,
+    RunStatus,
+    ScheduledRun,
+    ScheduledTask,
+    SchedulePolicy,
+    ScheduleSpec,
+    ScheduleType,
+    TaskNotFoundError,
+    ThreadBusyError,
+    ThreadNotFoundError,
+    TriggerKind,
+)
+from deerflow.domain.schedule.ports import (
+    RunLauncher,
+    RunOutcome,
+    ScheduledRunRepository,
+    ScheduledTaskRepository,
+    ThreadLookup,
+)
+
+# 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.
+_ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run"
+_SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active"
+
+_UNSET: Any = object()
+"""Distinguishes "field omitted" from "field set to None" in update_task."""
+
+
+@dataclass(frozen=True)
+class DispatchResult:
+    """What one dispatch attempt produced.
+
+    `outcome` drives the caller's protocol mapping -- a manual trigger turns
+    `conflict` into 409 and `failed` into 502 -- so the four values are part of
+    the contract, not an implementation detail.
+    """
+
+    outcome: Literal["launched", "skipped", "conflict", "failed"]
+    record_id: str | None
+    run_id: str | None
+    thread_id: str
+    error: str | None
+
+
+class ScheduleService:
+    def __init__(
+        self,
+        *,
+        tasks: ScheduledTaskRepository,
+        runs: ScheduledRunRepository,
+        launcher: RunLauncher,
+        threads: ThreadLookup,
+        policy: SchedulePolicy,
+        lease_owner: str,
+    ) -> None:
+        self._tasks = tasks
+        self._runs = runs
+        self._launcher = launcher
+        self._threads = threads
+        self._policy = policy
+        # Identity of the claiming process. Recorded for diagnostics only --
+        # the repository never reads it back.
+        self._lease_owner = lease_owner
+
+    # ------------------------------------------------------------------ reads
+
+    async def list_tasks(self, user_id: str) -> list[ScheduledTask]:
+        return await self._tasks.list_by_user(user_id)
+
+    async def list_tasks_by_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]:
+        return await self._tasks.list_by_user_and_thread(user_id, thread_id)
+
+    async def get_task(self, task_id: str, *, user_id: str) -> ScheduledTask:
+        """Raises TaskNotFoundError when absent or owned by someone else -- the
+        caller must not be able to tell those apart."""
+        task = await self._tasks.get(task_id, user_id=user_id)
+        if task is None:
+            raise TaskNotFoundError("Scheduled task not found")
+        return task
+
+    async def list_task_runs(self, task_id: str, *, user_id: str, limit: int = 50, offset: int = 0) -> list[ScheduledRun]:
+        """Execution history, gated on ownership of the parent task."""
+        await self.get_task(task_id, user_id=user_id)
+        return await self._runs.list_by_task(task_id, limit=limit, offset=offset)
+
+    # ------------------------------------------------------------------ writes
+
+    async def create_task(
+        self,
+        *,
+        user_id: str,
+        title: str,
+        prompt: str,
+        schedule: ScheduleSpec,
+        context_mode: str | ContextMode,
+        thread_id: str | None,
+        now: datetime,
+    ) -> ScheduledTask:
+        """Register a new standing instruction.
+
+        The aggregate is built first so a malformed schedule or context mode is
+        reported as such before any IO happens; only then is the thread
+        binding verified.
+
+        Raises:
+            InvalidScheduleError / InvalidContextModeError: from the aggregate.
+            ThreadNotFoundError: reuse_thread pointing at a thread the user
+                cannot use.
+        """
+        task = ScheduledTask.create(
+            user_id=user_id,
+            title=title,
+            prompt=prompt,
+            schedule=schedule,
+            context_mode=context_mode,
+            thread_id=thread_id,
+            now=now,
+            policy=self._policy,
+        )
+        await self._require_thread(task)
+        return await self._tasks.add(task)
+
+    async def update_task(
+        self,
+        task_id: str,
+        *,
+        user_id: str,
+        now: datetime,
+        title: str = _UNSET,
+        prompt: str = _UNSET,
+        schedule: ScheduleSpec = _UNSET,
+        context_mode: str | ContextMode = _UNSET,
+        thread_id: str | None = _UNSET,
+    ) -> ScheduledTask:
+        """Partially update a task; omitted fields are left alone.
+
+        Context and schedule are applied through the aggregate's own
+        transitions, so the re-arm rule and the running-task gate cannot be
+        bypassed by patching fields directly.
+        """
+        task = await self.get_task(task_id, user_id=user_id)
+        task.ensure_mutable()
+
+        if context_mode is not _UNSET or thread_id is not _UNSET:
+            effective_mode = task.context_mode if context_mode is _UNSET else context_mode
+            effective_thread = task.thread_id if thread_id is _UNSET else thread_id
+            task = task.with_context(effective_mode, effective_thread)
+            await self._require_thread(task)
+
+        if schedule is not _UNSET:
+            task = task.with_schedule(schedule, now=now, policy=self._policy)
+
+        if title is not _UNSET:
+            task = replace(task, title=title)
+        if prompt is not _UNSET:
+            task = replace(task, prompt=prompt)
+
+        saved = await self._tasks.save(task)
+        if saved is None:
+            raise TaskNotFoundError("Scheduled task not found")
+        return saved
+
+    async def pause_task(self, task_id: str, *, user_id: str) -> ScheduledTask:
+        return await self._transition(task_id, user_id=user_id, transition=ScheduledTask.paused)
+
+    async def resume_task(self, task_id: str, *, user_id: str) -> ScheduledTask:
+        return await self._transition(task_id, user_id=user_id, transition=ScheduledTask.resumed)
+
+    async def delete_task(self, task_id: str, *, user_id: str) -> None:
+        """Deleting is deliberately not gated on the task being idle: the
+        pre-migration router applied that gate to update/pause/resume only."""
+        if not await self._tasks.delete(task_id, user_id=user_id):
+            raise TaskNotFoundError("Scheduled task not found")
+
+    # ------------------------------------------------------------------ dispatch
+
+    async def trigger_task(self, task_id: str, *, user_id: str, now: datetime) -> DispatchResult:
+        """Dispatch a task on demand. Unlike the scheduled path this is allowed
+        while the task is paused, and leaves it paused."""
+        task = await self.get_task(task_id, user_id=user_id)
+        return await self.dispatch_task(task, now=now, trigger=TriggerKind.MANUAL)
+
+    async def run_once(self, *, now: datetime) -> list[DispatchResult]:
+        """Claim whatever is due and dispatch it.
+
+        `max_concurrent_runs` is a global cap on active scheduled runs, not a
+        per-poll batch size: long runs accumulate across poll cycles, so each
+        cycle only claims into the remaining budget.
+        """
+        active = await self._runs.count_active()
+        budget = self._policy.max_concurrent_runs - active
+        if budget <= 0:
+            return []
+        claimed = await self._tasks.claim_due(
+            now=now,
+            lease_owner=self._lease_owner,
+            lease_seconds=self._policy.lease_seconds,
+            limit=budget,
+        )
+        return [await self.dispatch_task(task, now=now, trigger=TriggerKind.SCHEDULED) for task in claimed]
+
+    async def dispatch_task(self, task: ScheduledTask, *, now: datetime, trigger: TriggerKind) -> DispatchResult:
+        """Turn one due task into one execution.
+
+        Called once per dispatch, so `resolve_execution_thread` is called once
+        and its value reused for the record, the launch, and the result.
+        """
+        execution_thread_id = task.resolve_execution_thread()
+
+        # "skip" must hold for fresh-thread runs too, where every run gets a
+        # new thread and the same-thread busy signal below can never fire.
+        # Checked before creating this dispatch's own record so the record does
+        # not count itself as the active run. A manual trigger against an
+        # active run is rejected outright instead of being recorded as a
+        # skipped occurrence -- nothing was scheduled to happen.
+        #
+        # This check is a NON-ATOMIC fast path: two concurrent dispatches (a
+        # manual trigger racing the poller, a double-click, a client retry) can
+        # both observe no active run. The repository is the atomic arbiter --
+        # it rejects the second active record with ActiveRunConflictError,
+        # which collapses to the SAME outcome as this fast path just below.
+        if task.skips_on_overlap and await self._runs.has_active(task.task_id):
+            if trigger is TriggerKind.MANUAL:
+                return self._conflict(execution_thread_id)
+            return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
+
+        record = ScheduledRun.queued(
+            task_id=task.task_id,
+            thread_id=execution_thread_id,
+            scheduled_for=now,
+            trigger=trigger,
+        )
+        try:
+            await self._runs.add(record)
+        except ActiveRunConflictError:
+            # Lost the race for the task's single active slot: a concurrent
+            # dispatch passed the same fast-path check and inserted first.
+            # Identical outcome to the fast path above -- that equality is what
+            # the dispatch-race regression tests pin.
+            if trigger is TriggerKind.MANUAL:
+                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
+        # escapes, so a failure of the bookkeeping writes below is a genuine
+        # fault and propagates instead of being recorded as a failed launch --
+        # which would mark an already-running execution as failed.
+        try:
+            launched = await self._launcher.launch(
+                thread_id=execution_thread_id,
+                assistant_id=task.assistant_id,
+                prompt=task.prompt,
+                owner_user_id=task.user_id,
+                metadata={
+                    "scheduled_task_id": task.task_id,
+                    "scheduled_task_run_id": record.record_id,
+                    "scheduled_trigger": str(trigger),
+                },
+            )
+        except ThreadBusyError as exc:
+            # The execution thread is already busy. On the scheduled path under
+            # a skip policy this is an overlap like any other; anything else is
+            # reported as a conflict the caller has to deal with.
+            if trigger is TriggerKind.SCHEDULED and task.skips_on_overlap:
+                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="conflict")
+        except LaunchFailedError as exc:
+            return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome="failed")
+
+        await self._runs.update_status(
+            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("launched", record.record_id, launched.run_id, launched.thread_id, None)
+
+    # ------------------------------------------------------------------ lifecycle
+
+    async def handle_run_completion(self, outcome: RunOutcome, *, now: datetime) -> None:
+        """Write back a launched run's terminal verdict.
+
+        A task deleted while its run was in flight simply has nothing to
+        update, and that is not an error.
+        """
+        await self._runs.update_status(
+            outcome.record_id,
+            status=outcome.status,
+            run_id=outcome.run_id,
+            error=outcome.error,
+            finished_at=now,
+        )
+        task = await self._tasks.get(outcome.task_id, user_id=outcome.user_id)
+        if task is None:
+            return
+        # The error is recorded whether or not the status moves: a cron task
+        # keeps its schedule but still reports what went wrong last time.
+        updated = replace(task, last_error=outcome.error)
+        new_status = task.status_after_completion(outcome.status)
+        if new_status is not None:
+            updated = replace(updated, status=new_status)
+        await self._tasks.save(updated)
+
+    async def reconcile_on_startup(self, *, error: str) -> tuple[int, int]:
+        """Clean up what a process crash left behind, returning what was fixed.
+
+        Two sweeps, because a crash strands two different things: execution
+        records that can never finish, and `once` tasks parked waiting for a
+        completion hook that died with the process. The second is not covered
+        by expired-claim reclaim -- a launched task released its claim, so the
+        claim query can never see it again.
+
+        Failures propagate: whether a partial reconcile should block startup is
+        the caller's policy, not the domain's.
+        """
+        stale_runs = await self._runs.mark_stale_active(error=error)
+        stuck_tasks = await self._tasks.cancel_stuck_once_tasks(error=error)
+        return stale_runs, stuck_tasks
+
+    # ------------------------------------------------------------------ internals
+
+    async def _require_thread(self, task: ScheduledTask) -> None:
+        if task.context_mode is not ContextMode.REUSE_THREAD:
+            return
+        # The aggregate guarantees a thread here; the check keeps that
+        # guarantee from being silently dropped under `python -O`.
+        if not task.thread_id or not await self._threads.exists_for_user(task.thread_id, task.user_id):
+            raise ThreadNotFoundError("Thread not found")
+
+    async def _transition(self, task_id: str, *, user_id: str, transition: Callable[[ScheduledTask], ScheduledTask]) -> ScheduledTask:
+        task = await self.get_task(task_id, user_id=user_id)
+        saved = await self._tasks.save(transition(task))
+        if saved is None:
+            raise TaskNotFoundError("Scheduled task not found")
+        return saved
+
+    def _conflict(self, thread_id: str) -> DispatchResult:
+        """A manual trigger against an active run.
+
+        No history record is written: nothing was scheduled to happen, so there
+        is no occurrence to account for.
+        """
+        return DispatchResult("conflict", None, None, thread_id, _ACTIVE_RUN_CONFLICT_ERROR)
+
+    async def _record_scheduled_skip(self, task: ScheduledTask, *, thread_id: str, now: datetime, trigger: TriggerKind) -> DispatchResult:
+        """Account for a scheduled occurrence dropped because of an overlap.
+
+        The tombstone is created directly terminal rather than as the transient
+        queued record the launch path uses: a queued record is active and would
+        itself be refused against the pre-existing run that is still holding
+        the task's single active slot.
+        """
+        record = ScheduledRun.skipped_tombstone(
+            task_id=task.task_id,
+            thread_id=thread_id,
+            scheduled_for=now,
+            trigger=trigger,
+        )
+        await self._runs.add(record)
+        return await self._finalize_skip(task, record_id=record.record_id, thread_id=thread_id, now=now, error=_SKIP_ACTIVE_RUN_ERROR)
+
+    async def _finalize_skip(self, task: ScheduledTask, *, record_id: str, thread_id: str, now: datetime, error: str) -> DispatchResult:
+        await self._runs.update_status(
+            record_id,
+            status=RunStatus.SKIPPED,
+            error=error,
+            started_at=now,
+            finished_at=now,
+        )
+        await self._tasks.record_launch(
+            task.task_id,
+            status=task.status_after_skip(),
+            next_run_at=task.schedule.next_after(now),
+            # A skip is not an execution, so the launch bookkeeping carries
+            # over unchanged; record_launch assigns unconditionally, so
+            # "unchanged" has to be spelled out.
+            last_run_at=task.last_run_at,
+            last_run_id=task.last_run_id,
+            last_thread_id=task.last_thread_id,
+            # Only a lost one-shot occurrence is worth surfacing; a cron task
+            # simply waits for its next turn.
+            last_error=error if task.schedule.schedule_type is ScheduleType.ONCE else None,
+            increment_run_count=False,
+        )
+        return DispatchResult("skipped", record_id, None, thread_id, error)
+
+    async def _fail(
+        self,
+        task: ScheduledTask,
+        *,
+        record_id: str,
+        thread_id: str,
+        now: datetime,
+        trigger: TriggerKind,
+        error: str,
+        outcome: Literal["conflict", "failed"],
+    ) -> DispatchResult:
+        await self._runs.update_status(
+            record_id,
+            status=RunStatus.FAILED,
+            error=error,
+            started_at=now,
+            finished_at=now,
+        )
+        await self._tasks.record_launch(
+            task.task_id,
+            status=task.status_after_failure(trigger=trigger),
+            next_run_at=task.schedule.next_after(now),
+            last_run_at=now,
+            last_run_id=None,
+            last_thread_id=thread_id,
+            last_error=error,
+            increment_run_count=False,
+        )
+        return DispatchResult(outcome, record_id, None, thread_id, error)

backend/tests/schedule_fakes.py

@@ -25,6 +25,7 @@ from deerflow.domain.schedule.model import (
     RunStatus,
     ScheduledRun,
     ScheduledTask,
+    ScheduleType,
     TaskStatus,
 )
 from deerflow.domain.schedule.ports import LaunchedRun
@@ -155,7 +156,7 @@ class InMemoryScheduledTaskRepository:
     async def cancel_stuck_once_tasks(self, *, error: str) -> int:
         cancelled = 0
         for row in self._rows.values():
-            stuck = row.task.status is TaskStatus.RUNNING and row.task.schedule.schedule_type.value == "once" and row.lease_expires_at is None
+            stuck = row.task.status is TaskStatus.RUNNING and row.task.schedule.schedule_type is ScheduleType.ONCE and row.lease_expires_at is None
             if stuck:
                 row.task = replace(row.task, status=TaskStatus.CANCELLED, last_error=error)
                 cancelled += 1

backend/tests/test_schedule_service.py

@@ -0,0 +1,630 @@
+"""Use-case tests for ScheduleService, run entirely on in-memory doubles.
+
+This file is the acceptance criterion of the hexagonal migration: the complete
+scheduled-task lifecycle runs here with no HTTP, no database, and no run
+runtime -- if any of that were still reachable from the domain, these tests
+could not exist.
+
+The dispatch cases mirror the pre-migration behaviour deliberately. Where a
+comment says two paths must be indistinguishable, that is a regression the
+tests are here to catch, not a description of the implementation.
+"""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from schedule_fakes import (
+    FakeRunLauncher,
+    FakeThreadLookup,
+    InMemoryScheduledRunRepository,
+    InMemoryScheduledTaskRepository,
+)
+
+from deerflow.domain.schedule.model import (
+    ContextMode,
+    InvalidScheduleError,
+    LaunchFailedError,
+    RunStatus,
+    SchedulePolicy,
+    ScheduleSpec,
+    TaskNotFoundError,
+    TaskNotMutableError,
+    TaskStatus,
+    ThreadBusyError,
+    ThreadNotFoundError,
+    TriggerKind,
+)
+from deerflow.domain.schedule.ports import RunOutcome
+from deerflow.domain.schedule.service import ScheduleService
+
+pytestmark = pytest.mark.asyncio
+
+NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
+CRON = ScheduleSpec.cron_schedule("0 9 * * *", "UTC")
+POLICY = SchedulePolicy(min_once_delay_seconds=60, max_concurrent_runs=3, lease_seconds=120)
+
+
+def once_spec(*, after_seconds: int = 3600) -> ScheduleSpec:
+    return ScheduleSpec.once_at(NOW + timedelta(seconds=after_seconds), "UTC")
+
+
+class _BlindRunRepo(InMemoryScheduledRunRepository):
+    """`has_active` always misses.
+
+    Reproduces the TOCTOU window the fast path cannot close: a concurrent
+    dispatch inserted its record after this one looked. The active-slot
+    rejection on `add` is then the only thing standing in the way.
+    """
+
+    async def has_active(self, task_id: str) -> bool:
+        return False
+
+
+def make_service(
+    *,
+    tasks: InMemoryScheduledTaskRepository | None = None,
+    runs: InMemoryScheduledRunRepository | None = None,
+    launcher: FakeRunLauncher | None = None,
+    threads: FakeThreadLookup | None = None,
+    policy: SchedulePolicy = POLICY,
+) -> ScheduleService:
+    return ScheduleService(
+        tasks=tasks if tasks is not None else InMemoryScheduledTaskRepository(),
+        runs=runs if runs is not None else InMemoryScheduledRunRepository(),
+        launcher=launcher if launcher is not None else FakeRunLauncher(),
+        threads=threads if threads is not None else FakeThreadLookup(),
+        policy=policy,
+        lease_owner="test-worker",
+    )
+
+
+async def create_cron_task(service: ScheduleService, *, schedule: ScheduleSpec = CRON):
+    return await service.create_task(
+        user_id="user-1",
+        title="Daily summary",
+        prompt="summarize",
+        schedule=schedule,
+        context_mode=ContextMode.FRESH_THREAD_PER_RUN,
+        thread_id=None,
+        now=NOW,
+    )
+
+
+# ==================================================================== lifecycle
+
+
+class TestFullLifecycle:
+    async def test_create_claim_dispatch_overlap_complete_pause_delete(self):
+        """The acceptance criterion: the whole feature, zero IO.
+
+        Every step below is a real use case going through real domain rules --
+        only the four ports are doubles.
+        """
+        tasks = InMemoryScheduledTaskRepository()
+        runs = InMemoryScheduledRunRepository()
+        launcher = FakeRunLauncher()
+        service = make_service(tasks=tasks, runs=runs, launcher=launcher)
+
+        # -- create ---------------------------------------------------------
+        task = await create_cron_task(service)
+        assert task.status is TaskStatus.ENABLED
+        assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)
+
+        # -- become due, get claimed and dispatched --------------------------
+        due_at = task.next_run_at + timedelta(seconds=1)
+        results = await service.run_once(now=due_at)
+        assert [r.outcome for r in results] == ["launched"]
+        assert len(launcher.calls) == 1
+        launch = launcher.calls[0]
+        assert launch["prompt"] == "summarize"
+        assert launch["owner_user_id"] == "user-1"
+        assert launch["metadata"]["scheduled_task_id"] == task.task_id
+        assert launch["metadata"]["scheduled_trigger"] == "scheduled"
+
+        after_launch = await service.get_task(task.task_id, user_id="user-1")
+        assert after_launch.status is TaskStatus.ENABLED, "a cron task stays claimable"
+        assert after_launch.run_count == 1
+        assert after_launch.last_run_id == "run-1"
+        assert after_launch.next_run_at > due_at
+
+        # -- next occurrence overlaps the still-running one ------------------
+        overlap_at = after_launch.next_run_at + timedelta(seconds=1)
+        overlapped = await service.run_once(now=overlap_at)
+        assert [r.outcome for r in overlapped] == ["skipped"]
+        assert len(launcher.calls) == 1, "the overlapping occurrence must not launch"
+
+        after_skip = await service.get_task(task.task_id, user_id="user-1")
+        assert after_skip.run_count == 1, "a skip is not an execution"
+        assert after_skip.last_run_id == after_launch.last_run_id, "bookkeeping carried over"
+        assert after_skip.status is TaskStatus.ENABLED
+
+        # -- the first run finally finishes ----------------------------------
+        record = next(r for r in runs.all_runs() if r.status is RunStatus.RUNNING)
+        await service.handle_run_completion(
+            RunOutcome(
+                task_id=task.task_id,
+                record_id=record.record_id,
+                run_id="run-1",
+                user_id="user-1",
+                status=RunStatus.SUCCESS,
+                error=None,
+            ),
+            now=overlap_at,
+        )
+        assert await runs.count_active() == 0, "the active slot is free again"
+
+        # -- pause / resume ---------------------------------------------------
+        paused = await service.pause_task(task.task_id, user_id="user-1")
+        assert paused.status is TaskStatus.PAUSED
+        assert await service.run_once(now=overlap_at + timedelta(days=2)) == [], "a paused task is not claimed"
+
+        resumed = await service.resume_task(task.task_id, user_id="user-1")
+        assert resumed.status is TaskStatus.ENABLED
+
+        # -- history and delete ----------------------------------------------
+        history = await service.list_task_runs(task.task_id, user_id="user-1")
+        assert sorted(run.status for run in history) == [RunStatus.SKIPPED, RunStatus.SUCCESS]
+
+        await service.delete_task(task.task_id, user_id="user-1")
+        with pytest.raises(TaskNotFoundError):
+            await service.get_task(task.task_id, user_id="user-1")
+
+
+# ==================================================================== dispatch
+
+
+class TestDispatchOutcomes:
+    async def test_launch_records_the_run_and_the_bookkeeping(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)
+
+        assert result.outcome == "launched"
+        assert result.run_id == "run-1"
+        assert result.error is None
+        stored = runs.all_runs()[0]
+        assert stored.status is RunStatus.RUNNING
+        assert stored.run_id == "run-1"
+        assert stored.started_at == NOW
+
+    async def test_manual_trigger_against_an_active_run_is_a_conflict_with_no_record(self):
+        """Nothing was scheduled to happen, so there is no occurrence to
+        account for -- the caller gets a conflict and the history stays clean."""
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs)
+        task = await create_cron_task(service)
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        before = len(runs.all_runs())
+
+        result = await service.trigger_task(task.task_id, user_id="user-1", now=NOW)
+
+        assert result.outcome == "conflict"
+        assert result.record_id is None
+        assert result.error == "task already has an active run"
+        assert len(runs.all_runs()) == before, "no history row for a rejected manual trigger"
+
+    async def test_scheduled_overlap_records_a_terminal_tombstone(self):
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs)
+        task = await create_cron_task(service)
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        assert result.outcome == "skipped"
+        assert result.error == "skipped: a previous run of this task is still active"
+        tombstone = next(r for r in runs.all_runs() if r.status is RunStatus.SKIPPED)
+        assert tombstone.is_active is False, "a queued tombstone would collide with the live run"
+        assert tombstone.started_at == tombstone.finished_at == NOW
+
+    async def test_launch_failure_is_recorded_as_failed(self):
+        runs = InMemoryScheduledRunRepository()
+        launcher = FakeRunLauncher(fail_with=LaunchFailedError("provider exploded"))
+        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 == "failed"
+        assert result.run_id is None
+        assert "provider exploded" in result.error
+        assert runs.all_runs()[0].status is RunStatus.FAILED
+
+    async def test_busy_thread_on_the_scheduled_path_degrades_to_a_skip(self):
+        launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
+        service = make_service(launcher=launcher)
+        task = await create_cron_task(service)
+
+        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        assert result.outcome == "skipped"
+
+    async def test_busy_thread_on_a_manual_trigger_stays_a_conflict(self):
+        """Not a skip: the user asked for this one, so it is reported rather
+        than quietly accounted for. The router maps it to 409, not 502."""
+        launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
+        service = make_service(launcher=launcher)
+        task = await create_cron_task(service)
+
+        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL)
+
+        assert result.outcome == "conflict"
+        assert result.record_id is not None, "the attempt itself is still recorded"
+
+
+class TestConflictCollapse:
+    """The fast path and the active-slot rejection must be indistinguishable.
+
+    Two concurrent dispatches can both pass `has_active`; whichever loses is
+    rejected by the repository instead. A caller must not be able to tell which
+    of the two mechanisms stopped it, or retry behaviour diverges.
+    """
+
+    async def _dispatch_second(self, run_repo, trigger: TriggerKind):
+        # reuse_thread so the execution thread is fixed: a fresh-thread task
+        # mints a new uuid per dispatch, which would make the two runs differ
+        # for a reason that has nothing to do with the collapse.
+        service = make_service(runs=run_repo, threads=FakeThreadLookup({"thread-1": "user-1"}))
+        task = await service.create_task(
+            user_id="user-1",
+            title="t",
+            prompt="p",
+            schedule=CRON,
+            context_mode=ContextMode.REUSE_THREAD,
+            thread_id="thread-1",
+            now=NOW,
+        )
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        return await service.dispatch_task(task, now=NOW, trigger=trigger)
+
+    @pytest.mark.parametrize("trigger", [TriggerKind.SCHEDULED, TriggerKind.MANUAL])
+    async def test_both_paths_produce_the_same_result(self, trigger):
+        via_fast_path = await self._dispatch_second(InMemoryScheduledRunRepository(), trigger)
+        via_slot_rejection = await self._dispatch_second(_BlindRunRepo(), trigger)
+
+        # record_id is freshly generated; every other field must match exactly.
+        assert replace(via_fast_path, record_id=None) == replace(via_slot_rejection, record_id=None)
+
+    async def test_the_losing_scheduled_dispatch_still_leaves_a_tombstone(self):
+        runs = _BlindRunRepo()
+        result = await self._dispatch_second(runs, TriggerKind.SCHEDULED)
+        assert result.outcome == "skipped"
+        assert any(r.status is RunStatus.SKIPPED for r in runs.all_runs())
+
+    async def test_the_losing_manual_dispatch_leaves_nothing(self):
+        runs = _BlindRunRepo()
+        result = await self._dispatch_second(runs, TriggerKind.MANUAL)
+        assert result.outcome == "conflict"
+        assert not any(r.status is RunStatus.SKIPPED for r in runs.all_runs())
+
+
+class TestOnceTaskDispatch:
+    async def test_a_once_task_waits_in_running_for_its_completion(self):
+        """Declaring it complete at launch would stick if the run failed or the
+        process died before the hook could correct it."""
+        service = make_service()
+        task = await service.create_task(
+            user_id="user-1",
+            title="one shot",
+            prompt="go",
+            schedule=once_spec(),
+            context_mode=ContextMode.FRESH_THREAD_PER_RUN,
+            thread_id=None,
+            now=NOW,
+        )
+
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.status is TaskStatus.RUNNING
+
+    async def test_a_skipped_once_task_is_failed_not_completed(self):
+        """The single occurrence was lost; `completed` would claim an execution
+        that never happened."""
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs)
+        task = await service.create_task(
+            user_id="user-1",
+            title="one shot",
+            prompt="go",
+            schedule=once_spec(),
+            context_mode=ContextMode.FRESH_THREAD_PER_RUN,
+            thread_id=None,
+            now=NOW,
+        )
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        reloaded = await service.get_task(task.task_id, user_id="user-1")
+
+        await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.status is TaskStatus.FAILED
+        assert after.last_error == "skipped: a previous run of this task is still active"
+
+
+# ==================================================================== run_once
+
+
+class TestRunOnceBudget:
+    async def test_claims_nothing_when_the_global_budget_is_exhausted(self):
+        """The cap is on active runs across all tasks, not on one poll's batch:
+        long runs accumulate across cycles."""
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs, policy=replace(POLICY, max_concurrent_runs=1))
+        first = await create_cron_task(service)
+        await service.dispatch_task(first, now=NOW, trigger=TriggerKind.SCHEDULED)
+
+        second = await create_cron_task(service)
+        results = await service.run_once(now=second.next_run_at + timedelta(seconds=1))
+
+        assert results == []
+
+    async def test_claims_only_into_the_remaining_budget(self):
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks, policy=replace(POLICY, max_concurrent_runs=2))
+        for _ in range(3):
+            await create_cron_task(service)
+
+        due_at = datetime(2026, 7, 28, 9, 0, 1, tzinfo=UTC)
+        results = await service.run_once(now=due_at)
+
+        assert len(results) == 2
+        assert all(r.outcome == "launched" for r in results)
+
+    async def test_a_claimed_task_is_marked_running_before_dispatch(self):
+        """Claiming is what makes the task uneditable while it is being
+        dispatched, so the claim must land before the launch."""
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks)
+        task = await create_cron_task(service)
+
+        await service.run_once(now=task.next_run_at + timedelta(seconds=1))
+
+        assert tasks.lease_of(task.task_id) == (None, None), "the claim is released after dispatch"
+
+
+# ==================================================================== completion
+
+
+class TestRunCompletion:
+    async def _launched_once_task(self):
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs)
+        task = await service.create_task(
+            user_id="user-1",
+            title="one shot",
+            prompt="go",
+            schedule=once_spec(),
+            context_mode=ContextMode.FRESH_THREAD_PER_RUN,
+            thread_id=None,
+            now=NOW,
+        )
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        record = runs.all_runs()[0]
+        return service, task, record
+
+    @pytest.mark.parametrize(
+        ("outcome_status", "expected"),
+        [
+            (RunStatus.SUCCESS, TaskStatus.COMPLETED),
+            (RunStatus.FAILED, TaskStatus.FAILED),
+            (RunStatus.INTERRUPTED, TaskStatus.CANCELLED),
+        ],
+    )
+    async def test_once_task_terminal_mapping(self, outcome_status, expected):
+        service, task, record = await self._launched_once_task()
+
+        await service.handle_run_completion(
+            RunOutcome(
+                task_id=task.task_id,
+                record_id=record.record_id,
+                run_id="run-1",
+                user_id="user-1",
+                status=outcome_status,
+                error=None if outcome_status is RunStatus.SUCCESS else "boom",
+            ),
+            now=NOW,
+        )
+
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.status is expected
+
+    async def test_a_cron_task_keeps_its_status_but_records_the_error(self):
+        runs = InMemoryScheduledRunRepository()
+        service = make_service(runs=runs)
+        task = await create_cron_task(service)
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        record = runs.all_runs()[0]
+
+        await service.handle_run_completion(
+            RunOutcome(
+                task_id=task.task_id,
+                record_id=record.record_id,
+                run_id="run-1",
+                user_id="user-1",
+                status=RunStatus.FAILED,
+                error="boom",
+            ),
+            now=NOW,
+        )
+
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.status is TaskStatus.ENABLED, "the schedule outlives any single run"
+        assert after.last_error == "boom"
+
+    async def test_a_task_deleted_mid_flight_is_not_an_error(self):
+        service, task, record = await self._launched_once_task()
+        await service.delete_task(task.task_id, user_id="user-1")
+
+        await service.handle_run_completion(
+            RunOutcome(
+                task_id=task.task_id,
+                record_id=record.record_id,
+                run_id="run-1",
+                user_id="user-1",
+                status=RunStatus.SUCCESS,
+                error=None,
+            ),
+            now=NOW,
+        )
+
+
+class TestReconcileOnStartup:
+    async def test_sweeps_orphaned_runs_and_stuck_once_tasks(self):
+        runs = InMemoryScheduledRunRepository()
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks, runs=runs)
+        task = await service.create_task(
+            user_id="user-1",
+            title="one shot",
+            prompt="go",
+            schedule=once_spec(),
+            context_mode=ContextMode.FRESH_THREAD_PER_RUN,
+            thread_id=None,
+            now=NOW,
+        )
+        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
+        # The process dies here: the run is active and the once task is parked
+        # in running with its claim already released.
+
+        stale_runs, stuck_tasks = await service.reconcile_on_startup(error="gateway restarted")
+
+        assert (stale_runs, stuck_tasks) == (1, 1)
+        assert runs.all_runs()[0].status is RunStatus.INTERRUPTED
+        after = await service.get_task(task.task_id, user_id="user-1")
+        assert after.status is TaskStatus.CANCELLED
+
+
+# ==================================================================== CRUD
+
+
+class TestTaskManagement:
+    async def test_reuse_thread_requires_an_accessible_thread(self):
+        service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"}))
+
+        created = await service.create_task(
+            user_id="user-1",
+            title="t",
+            prompt="p",
+            schedule=CRON,
+            context_mode=ContextMode.REUSE_THREAD,
+            thread_id="thread-1",
+            now=NOW,
+        )
+        assert created.thread_id == "thread-1"
+
+        with pytest.raises(ThreadNotFoundError):
+            await service.create_task(
+                user_id="user-2",
+                title="t",
+                prompt="p",
+                schedule=CRON,
+                context_mode=ContextMode.REUSE_THREAD,
+                thread_id="thread-1",
+                now=NOW,
+            )
+
+    async def test_an_invalid_schedule_is_rejected_before_any_write(self):
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks)
+
+        with pytest.raises(InvalidScheduleError):
+            await service.create_task(
+                user_id="user-1",
+                title="t",
+                prompt="p",
+                schedule=once_spec(after_seconds=10),  # inside min_once_delay
+                context_mode=ContextMode.FRESH_THREAD_PER_RUN,
+                thread_id=None,
+                now=NOW,
+            )
+        assert await service.list_tasks("user-1") == []
+
+    async def test_update_leaves_omitted_fields_alone(self):
+        service = make_service()
+        task = await create_cron_task(service)
+
+        updated = await service.update_task(task.task_id, user_id="user-1", now=NOW, title="renamed")
+
+        assert updated.title == "renamed"
+        assert updated.prompt == task.prompt
+        assert updated.schedule == task.schedule
+
+    async def test_rescheduling_a_terminal_task_re_arms_it(self):
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks)
+        task = await create_cron_task(service)
+        tasks.seed(replace(task, status=TaskStatus.FAILED))
+
+        updated = await service.update_task(
+            task.task_id,
+            user_id="user-1",
+            now=NOW,
+            schedule=ScheduleSpec.cron_schedule("0 10 * * *", "UTC"),
+        )
+
+        assert updated.status is TaskStatus.ENABLED, "otherwise it would never be claimed again"
+
+    async def test_a_running_task_cannot_be_edited(self):
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks)
+        task = await create_cron_task(service)
+        tasks.seed(replace(task, status=TaskStatus.RUNNING))
+
+        with pytest.raises(TaskNotMutableError):
+            await service.update_task(task.task_id, user_id="user-1", now=NOW, title="nope")
+        with pytest.raises(TaskNotMutableError):
+            await service.pause_task(task.task_id, user_id="user-1")
+
+    async def test_a_running_task_can_still_be_deleted(self):
+        """The pre-migration router gated update/pause/resume on this, but not
+        delete -- that asymmetry is preserved."""
+        tasks = InMemoryScheduledTaskRepository()
+        service = make_service(tasks=tasks)
+        task = await create_cron_task(service)
+        tasks.seed(replace(task, status=TaskStatus.RUNNING))
+
+        await service.delete_task(task.task_id, user_id="user-1")
+
+    @pytest.mark.parametrize("call", ["get", "update", "pause", "resume", "delete", "runs"])
+    async def test_another_users_task_is_reported_as_missing(self, call):
+        service = make_service()
+        task = await create_cron_task(service)
+        kwargs = {"user_id": "intruder"}
+
+        with pytest.raises(TaskNotFoundError):
+            if call == "get":
+                await service.get_task(task.task_id, **kwargs)
+            elif call == "update":
+                await service.update_task(task.task_id, now=NOW, title="x", **kwargs)
+            elif call == "pause":
+                await service.pause_task(task.task_id, **kwargs)
+            elif call == "resume":
+                await service.resume_task(task.task_id, **kwargs)
+            elif call == "delete":
+                await service.delete_task(task.task_id, **kwargs)
+            else:
+                await service.list_task_runs(task.task_id, **kwargs)
+
+    async def test_tasks_are_listed_per_user_and_per_thread(self):
+        service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"}))
+        bound = await service.create_task(
+            user_id="user-1",
+            title="bound",
+            prompt="p",
+            schedule=CRON,
+            context_mode=ContextMode.REUSE_THREAD,
+            thread_id="thread-1",
+            now=NOW,
+        )
+        await create_cron_task(service)
+
+        assert len(await service.list_tasks("user-1")) == 2
+        assert await service.list_tasks("someone-else") == []
+        by_thread = await service.list_tasks_by_thread("user-1", "thread-1")
+        assert [t.task_id for t in by_thread] == [bound.task_id]