refactor(schedule): wire the slice through a composition root

Switches production onto the hexagonal path. The legacy modules still
compile and still have tests, but nothing assembles them any more; deleting
them is the next commit, kept separate so it stays reviewable.

Composition root
----------------
`app/composition.py::build_domain_services()` is now the only place an
adapter is instantiated. It was extracted from `deps.py::langgraph_runtime`
rather than added to it: wiring there was tangled with engine startup,
orphan recovery and shutdown, so the one rule that governs it -- no SQL
backend means no service and the routes answer 503 -- could not be tested
without booting the whole application, and was held up by a single comment.
It is a pure function of already-built infrastructure, so that rule is now
an assertion. Feedback moved with it; doing this while adding schedule's
five objects costs one change instead of two.

Primary adapter
---------------
The router is protocol translation only. What is gone is the giveaway: cron
normalisation, `next_run_at` arithmetic, the re-arm rule and hand-written
ownership checks all now live in the aggregate. Domain errors map to status
codes through one table, so a new error surfaces as a 500 to be classified
rather than being swallowed by whichever `except` was nearest.

`spec_mapping` split in two (AWS's own layout puts the wire model under the
entrypoint that owns it, and a primary adapter must not import a secondary
one): `adapters/schedule/spec_column.py` for the JSON column,
`routers/schedule/spec_wire.py` for the HTTP body. The two shapes are equal
only by coincidence, so `test_schedule_spec_parity.py` runs every case
against both and compares their outputs and messages directly. Function
names differ per side so an import from the wrong one is visible.

Explicit responses
------------------
Routes returned the ORM row's `to_dict()`, leaking `user_id`,
`assistant_id`, `overlap_policy` and the two lease columns. The response
models publish exactly the field set the frontend declares -- asserted in
both directions, since an extra field is a leak and a missing one breaks a
client.

One wire detail was nearly changed by accident: Pydantic v2 serializes a UTC
datetime as `...Z`, while the legacy `coerce_iso` path emitted `+00:00`.
`UtcTimestamp` pins `isoformat()` so adopting a model does not silently
alter the wire format for every client parsing these.

Tests
-----
73 new cases: router behaviour driven through a real `ScheduleService` over
in-memory fakes (a mocked service would let the error mapping pass without
a domain error ever being raised), response shape, and the composition root.
Router mappings verified by mutation -- a wrong status code or a dropped
timezone fallback turns 9, 2 and 1 cases red respectively.

Two lifespan tests carried a `SimpleNamespace` config that predates this
change; `langgraph_runtime` now reads `config.scheduler`, so they were given
one. Tolerating the gap with `getattr` was rejected: `AppConfig.scheduler`
always exists, so the fallback would be unreachable in production and exist
purely to excuse an incomplete test double.

Full suite is back to its 24 pre-existing failures.

backend/app/adapters/schedule/scheduled_task_repository.py

@@ -5,8 +5,9 @@ This context owns the `scheduled_tasks` table and writes its own queries, so
 SQL/ORM vocabulary stops at this file: methods exchange domain objects and
 normalize SQLite's tz-naive reads.
 
-Sibling of `scheduled_run_repository.py`; both consume `spec_mapping` for the
+Sibling of `scheduled_run_repository.py`. The stored `schedule_spec` JSON
-stored JSON spec.
+column is translated by `spec_column.py`, which belongs to this side of the
+boundary only -- the HTTP shape has its own translation next to the router.
 
 **The queries are migrated unchanged from the legacy repository.** The claim
 statement's `FOR UPDATE SKIP LOCKED` and the `protect_terminal` conditional
@@ -24,7 +25,7 @@ from datetime import UTC, datetime, timedelta
 from sqlalchemy import and_, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from app.adapters.schedule.spec_mapping import spec_to_domain, spec_to_wire
+from app.adapters.schedule.spec_column import column_to_spec, spec_to_column
 from deerflow.domain.schedule.model import (
     TERMINAL_TASK_STATUSES,
     ContextMode,
@@ -77,7 +78,7 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository):
             user_id=row.user_id,
             title=row.title,
             prompt=row.prompt,
-            schedule=spec_to_domain(row.schedule_type, row.schedule_spec, row.timezone),
+            schedule=column_to_spec(row.schedule_type, row.schedule_spec, row.timezone),
             context_mode=ContextMode(row.context_mode),
             thread_id=row.thread_id,
             assistant_id=row.assistant_id,
@@ -109,7 +110,7 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository):
         row.title = task.title
         row.prompt = task.prompt
         row.schedule_type = str(task.schedule.schedule_type)
-        row.schedule_spec = spec_to_wire(task.schedule)
+        row.schedule_spec = spec_to_column(task.schedule)
         row.timezone = task.schedule.timezone
         row.context_mode = str(task.context_mode)
         row.thread_id = task.thread_id

backend/app/adapters/schedule/spec_column.py

@@ -1,17 +1,25 @@
-"""Boundary mapping (not a port implementation) -- wire/storage <-> ScheduleSpec.
+"""Boundary mapping (not a port implementation) -- the schedule_spec JSON column.
 
 Unlike its siblings in this package, this module implements no port: it is the
-shared translation both of them need. `schedule_spec` is both an HTTP request
+translation `scheduled_task_repository` needs between the stored column and the
-field and a JSON column -- one shape, two boundaries -- so the mapping lives
+value object, so the domain never grows a `Mapping[str, Any]` in its
-here once and both callers import it (`scheduled_task_repository` for the
+signatures.
-stored column, the scheduled-task router for the request body), rather than the
-domain growing a `Mapping[str, Any]` in its signatures.
 
-The split is deliberate: **structural** checks (is the key present? is it a
+Its counterpart on the other side of the application is
-str?) belong to this boundary, **value** rules (5-field cron, resolvable
+`app/gateway/routers/schedule/spec_wire.py`, which does the same job for the
-timezone, run_at present) belong to `ScheduleSpec.__post_init__`. That is why
+HTTP request/response body. The two are near-identical today and are still kept
-this module can look thin -- most of what could go wrong is caught one layer
+apart on purpose: a primary adapter must not import a secondary one, and the
-in, and reported with the same domain error.
+two shapes are only equal by coincidence -- the day the API grows a field the
+column does not have, they diverge without either side having to be untangled
+first. The function names differ (`column_to_spec` here, `wire_to_spec` there)
+so an import from the wrong side is visible rather than silently working.
+
+The duplication is bounded because the split inside each is deliberate:
+**structural** checks (is the key present? is it a str?) belong to the
+boundary, **value** rules (5-field cron, resolvable timezone, run_at present)
+belong to `ScheduleSpec.__post_init__`. Only the structural half is repeated,
+and `tests/test_schedule_spec_parity.py` feeds both the same malformed inputs
+so a drift between them fails a test rather than reaching production.
 """
 
 from __future__ import annotations
@@ -23,7 +31,7 @@ from typing import Any
 from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType
 
 
-def spec_to_domain(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec:
+def column_to_spec(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec:
     """Parse the stored/submitted triple into the value object.
 
     Raises:
@@ -55,7 +63,7 @@ def spec_to_domain(schedule_type: str, spec: Mapping[str, Any] | None, timezone:
     return ScheduleSpec.once_at(run_at, timezone)
 
 
-def spec_to_wire(spec: ScheduleSpec) -> dict[str, str]:
+def spec_to_column(spec: ScheduleSpec) -> dict[str, str]:
     """Rebuild the persisted/wire JSON shape.
 
     Note this normalizes the stored string rather than echoing the caller's

backend/app/composition.py

@@ -0,0 +1,136 @@
+"""Composition root -- the one place adapters are instantiated and wired.
+
+Every application service is assembled here and nowhere else. Ports are
+declared by the domain, implemented under ``app/adapters/``, and the two are
+introduced to each other in this file; no router, middleware, or lifespan hook
+constructs an adapter of its own.
+
+**Why this is a pure function rather than part of the lifespan.** Wiring used
+to live inside ``deps.py::langgraph_runtime``, tangled with engine startup,
+orphan recovery, and graceful shutdown -- so the one rule that actually
+governs it ("no SQL backend means no service, and the routes answer 503")
+could not be tested without driving a full application startup, and was held
+up by a single comment. ``build_domain_services`` takes what it needs and
+returns what it built, so that rule is an assertion in
+``tests/test_composition.py`` instead.
+
+**What "no SQL backend" means.** ``session_factory is None`` is how a
+``database.backend: memory`` deployment presents itself. Both contexts own
+tables, so neither can run on it; each service is ``None`` and the dependency
+providers translate that into 503. This is deliberately not a silent
+degradation to an in-memory implementation -- scheduled work that vanishes on
+restart is worse than scheduled work that is refused.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+from deerflow.domain.feedback.service import FeedbackService
+from deerflow.domain.schedule.model import SchedulePolicy
+from deerflow.domain.schedule.service import ScheduleService
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+    from deerflow.config.scheduler_config import SchedulerConfig
+    from deerflow.persistence.thread_meta.base import ThreadMetaStore
+    from deerflow.runtime.runs.store import RunStore
+
+
+@dataclass(frozen=True)
+class DomainServices:
+    """Every application service the Gateway serves, or ``None`` where the
+    configured backend cannot support one."""
+
+    feedback: FeedbackService | None
+    schedule: ScheduleService | None
+
+
+def build_domain_services(
+    *,
+    session_factory: async_sessionmaker[AsyncSession] | None,
+    run_store: RunStore,
+    thread_store: ThreadMetaStore,
+    launch_run: Callable[..., Awaitable[Mapping[str, Any]]],
+    scheduler_config: SchedulerConfig,
+) -> DomainServices:
+    """Wire the domain services from already-built infrastructure.
+
+    Takes infrastructure rather than building it: engines, stores and the run
+    launcher have lifecycles (startup, recovery, shutdown) that belong to the
+    lifespan, while deciding *what is assembled from them* is this function's
+    only job. That split is what makes it callable from a test.
+    """
+    if session_factory is None:
+        return DomainServices(feedback=None, schedule=None)
+
+    # Imported here, not at module scope: these pull in SQLAlchemy and the
+    # Gateway run path, and the composition root is imported by tests that
+    # only want the memory-backend branch.
+    from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
+    from app.adapters.feedback.run_lookup import RunStoreRunLookup
+    from app.adapters.schedule.run_launcher import GatewayRunLauncher
+    from app.adapters.schedule.scheduled_run_repository import SqlScheduledRunRepository
+    from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
+    from app.adapters.schedule.thread_lookup import ThreadStoreThreadLookup
+
+    return DomainServices(
+        feedback=FeedbackService(
+            repository=SqlFeedbackRepository(session_factory),
+            runs=RunStoreRunLookup(run_store),
+        ),
+        schedule=ScheduleService(
+            tasks=SqlScheduledTaskRepository(session_factory),
+            runs=SqlScheduledRunRepository(session_factory),
+            launcher=GatewayRunLauncher(launch_run),
+            threads=ThreadStoreThreadLookup(thread_store),
+            policy=build_schedule_policy(scheduler_config),
+        ),
+    )
+
+
+def build_run_completion_hook(
+    schedule_service: ScheduleService | None,
+) -> Callable[[Any], Awaitable[None]] | None:
+    """Install the schedule context on the run runtime's completion callback.
+
+    This is the inbound half of the wiring: the run runtime hands every
+    finished run to one callback, and `run_outcome_from_record` decides which
+    of them the schedule context has any business with. Runs that are not
+    scheduled executions produce no outcome and the service is never called,
+    which is why it carries no guard clauses of its own.
+
+    Returns ``None`` when there is no service, so the runtime installs no hook
+    at all rather than one that always declines.
+    """
+    if schedule_service is None:
+        return None
+
+    from app.adapters.schedule.run_outcome_mapping import run_outcome_from_record
+
+    async def on_run_completed(record: Any) -> None:
+        outcome = run_outcome_from_record(record)
+        if outcome is not None:
+            await schedule_service.handle_run_completion(outcome, now=datetime.now(UTC))
+
+    return on_run_completed
+
+
+def build_schedule_policy(scheduler_config: SchedulerConfig) -> SchedulePolicy:
+    """Project the operator's scheduler config onto the domain's policy.
+
+    Separate from the wiring above because it is the whole of the
+    configuration-to-domain translation: the domain declares which thresholds
+    it needs, and this names where each one comes from. `poll_interval_seconds`
+    is deliberately absent -- how often to look is the poller's business, not a
+    rule any task is subject to.
+    """
+    return SchedulePolicy(
+        min_once_delay_seconds=scheduler_config.min_once_delay_seconds,
+        max_concurrent_runs=scheduler_config.max_concurrent_runs,
+        lease_seconds=scheduler_config.lease_seconds,
+    )

backend/app/gateway/app.py

@@ -30,13 +30,13 @@ from app.gateway.routers import (
     memory,
     models,
     runs,
-    scheduled_tasks,
     skills,
     suggestions,
     thread_runs,
     threads,
     uploads,
 )
+from app.gateway.routers.schedule import router as schedule_router
 from app.gateway.trace_middleware import TraceMiddleware, resolve_trace_enabled
 from deerflow.config import app_config as deerflow_app_config
 from deerflow.logging_config import DEFAULT_LOG_DATE_FORMAT, DEFAULT_LOG_FORMAT, configure_logging
@@ -305,23 +305,23 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             logger.exception("No IM channels configured or channel service failed to start")
 
         try:
-            from app.gateway.services import launch_scheduled_thread_run
+            # The service itself was assembled by the composition root inside
-            from app.scheduler import ScheduledTaskService
+            # `langgraph_runtime`; all that is left here is the clock that
+            # drives it. It is None when the configured backend cannot support
+            # scheduling, in which case there is nothing to poll.
+            from app.scheduler.poller import SchedulePoller
 
-            if getattr(app.state, "scheduled_task_repo", None) is not None and getattr(app.state, "scheduled_task_run_repo", None) is not None:
+            schedule_service = getattr(app.state, "schedule_service", None)
-                scheduled_task_service = ScheduledTaskService(
+            if schedule_service is not None:
-                    task_repo=app.state.scheduled_task_repo,
+                schedule_poller = SchedulePoller(
-                    task_run_repo=app.state.scheduled_task_run_repo,
+                    schedule_service,
-                    launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs),
                     poll_interval_seconds=startup_config.scheduler.poll_interval_seconds,
-                    lease_seconds=startup_config.scheduler.lease_seconds,
-                    max_concurrent_runs=startup_config.scheduler.max_concurrent_runs,
                 )
-                app.state.scheduled_task_service = scheduled_task_service
+                app.state.schedule_poller = schedule_poller
                 if startup_config.scheduler.enabled:
-                    await scheduled_task_service.start()
+                    await schedule_poller.start()
         except Exception:
-            logger.exception("Failed to initialize scheduled task service")
+            logger.exception("Failed to start the scheduled task poller")
 
         yield
 
@@ -346,11 +346,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Failed to stop channel service")
 
-        if getattr(app.state, "scheduled_task_service", None) is not None:
+        if getattr(app.state, "schedule_poller", None) is not None:
             try:
-                await app.state.scheduled_task_service.stop()
+                await app.state.schedule_poller.stop()
             except Exception:
-                logger.exception("Failed to stop scheduled task service")
+                logger.exception("Failed to stop the scheduled task poller")
 
         try:
             from deerflow.community.browser_automation import get_browser_session_manager
@@ -593,7 +593,7 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for
     app.include_router(threads.router)
 
     # Scheduled tasks API is mounted at /api/scheduled-tasks
-    app.include_router(scheduled_tasks.router)
+    app.include_router(schedule_router.router)
 
     # Agents API is mounted at /api/agents
     app.include_router(agents.router)

backend/app/gateway/deps.py

@@ -22,14 +22,15 @@ import logging
 import os
 from collections.abc import AsyncGenerator, Callable
 from contextlib import AsyncExitStack, asynccontextmanager
-from typing import TYPE_CHECKING, TypeVar, cast
+from typing import TYPE_CHECKING, Annotated, TypeVar, cast
 
-from fastapi import FastAPI, HTTPException, Request
+from fastapi import Depends, FastAPI, HTTPException, Request
 from langgraph.types import Checkpointer
 
 from deerflow.community.browser_automation.session import browser_multi_worker_error
 from deerflow.config.app_config import AppConfig, get_app_config
 from deerflow.domain.feedback import FeedbackService
+from deerflow.domain.schedule.service import ScheduleService
 from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, RunContext, RunManager, StreamBridge
 from deerflow.runtime.events.store.base import RunEventStore
 from deerflow.runtime.runs.store.base import RunStore
@@ -400,27 +401,18 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
         # Initialize repositories — one get_session_factory() call for all.
         sf = get_session_factory()
         if sf is not None:
-            from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
-            from app.adapters.feedback.run_lookup import RunStoreRunLookup
             from deerflow.persistence.run import RunRepository
 
             app.state.run_store = RunRepository(sf)
-            # Hexagonal feedback slice: the service (input port) is wired with
-            # its SQL adapter here — the composition root is the only place
-            # adapters are instantiated.
-            app.state.feedback_service = FeedbackService(
-                repository=SqlFeedbackRepository(sf),
-                runs=RunStoreRunLookup(app.state.run_store),
-            )
         else:
             from deerflow.runtime.runs.store.memory import MemoryRunStore
 
             app.state.run_store = MemoryRunStore()
-            app.state.feedback_service = None  # memory backend → 503, as before
 
         from deerflow.persistence.thread_meta import make_thread_store
 
         app.state.thread_store = make_thread_store(sf, app.state.store)
+
         if sf is not None:
             from deerflow.persistence.scheduled_task_runs import (
                 ScheduledTaskRunRepository,
@@ -433,6 +425,27 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
             app.state.scheduled_task_repo = None
             app.state.scheduled_task_run_repo = None
 
+        # Hexagonal slices: every adapter is instantiated by the composition
+        # root and nowhere else. It is a pure function of the infrastructure
+        # built above, so what it decides — including "no SQL backend means no
+        # service, and the routes answer 503" — is covered by
+        # tests/test_composition.py rather than only by this call site.
+        from app.composition import build_domain_services, build_run_completion_hook
+        from app.gateway.services import launch_scheduled_thread_run
+
+        domain_services = build_domain_services(
+            session_factory=sf,
+            run_store=app.state.run_store,
+            thread_store=app.state.thread_store,
+            launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs),
+            scheduler_config=config.scheduler,
+        )
+        app.state.feedback_service = domain_services.feedback
+        app.state.schedule_service = domain_services.schedule
+        # Built once here rather than per request: it closes over the service,
+        # and every RunContext needs the same one.
+        app.state.run_completion_hook = build_run_completion_hook(domain_services.schedule)
+
         # Run event store. The store and the matching ``run_events_config`` are
         # both frozen at startup so ``get_run_context`` does not combine a
         # freshly-reloaded ``AppConfig.run_events`` with a store still bound to
@@ -583,6 +596,25 @@ def get_scheduled_task_service(request: Request):
     return val
 
 
+def get_schedule_service(request: Request) -> ScheduleService:
+    """The schedule context's input port.
+
+    ``None`` here means the composition root refused to build it, which today
+    means a non-SQL backend -- see ``app/composition.py`` for why that is a
+    503 rather than a degraded in-memory implementation.
+    """
+    val = getattr(request.app.state, "schedule_service", None)
+    if val is None:
+        raise HTTPException(status_code=503, detail="Scheduled tasks require a SQL database backend")
+    return val
+
+
+# Declared as an alias so routes take the service without a default argument
+# (leaving parameter order unconstrained) and so swapping the provider is a
+# one-line change here rather than an edit to every handler signature.
+ScheduleServiceDep = Annotated[ScheduleService, Depends(get_schedule_service)]
+
+
 def get_run_context(request: Request) -> RunContext:
     """Build a :class:`RunContext` from ``app.state`` singletons.
 
@@ -601,7 +633,7 @@ def get_run_context(request: Request) -> RunContext:
         checkpoint_channel_mode=getattr(request.app.state, "checkpoint_channel_mode", "full"),
         thread_store=get_thread_store(request),
         app_config=get_config(),
-        on_run_completed=getattr(request.app.state, "scheduled_task_service", None).handle_run_completion if getattr(request.app.state, "scheduled_task_service", None) is not None else None,
+        on_run_completed=getattr(request.app.state, "run_completion_hook", None),
     )
 
 

backend/app/gateway/routers/schedule/models.py

@@ -0,0 +1,150 @@
+"""The HTTP shapes of the scheduled-task API.
+
+The primary adapter's own model of what a client sends and receives -- the
+counterpart to the domain aggregates, not a view of them. Two things follow
+from that:
+
+**Responses are an allowlist, not a dump.** The pre-migration router returned
+the ORM row's ``to_dict()``, which leaked ``user_id``, ``lease_owner``,
+``lease_expires_at``, ``overlap_policy`` and ``assistant_id`` -- lease fields
+are scheduler-internal bookkeeping, and the other three are server-owned. None
+appear in the frontend's ``ScheduledTask`` type or anywhere in its code, so
+naming the fields explicitly here closes the leak without a client change. A
+field added to the aggregate from now on stays invisible until it is
+deliberately published.
+
+**Timestamps keep the legacy spelling.** The legacy path emitted
+``coerce_iso`` -> ``astimezone(UTC).isoformat()``, i.e.
+``2026-08-01T09:00:00+00:00``. Pydantic v2 would serialize the same instant as
+``...T09:00:00Z``, which is a silent wire change for every client parsing
+these, so ``UtcTimestamp`` pins ``isoformat()`` explicitly. Both spellings are
+valid ISO 8601 and JS ``Date`` accepts either -- the point is that changing it
+is a decision, not a side effect of adopting a model.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Annotated, Any
+
+from pydantic import BaseModel, Field, PlainSerializer
+
+from app.gateway.routers.schedule.spec_wire import spec_to_wire
+from deerflow.domain.schedule.model import ScheduledRun, ScheduledTask
+
+UtcTimestamp = Annotated[datetime, PlainSerializer(lambda value: value.isoformat(), return_type=str)]
+
+
+def _utc(value: datetime | None) -> datetime | None:
+    """Match the legacy `coerce_iso` normalisation exactly."""
+    if value is None:
+        return None
+    return value.astimezone(UTC) if value.tzinfo is not None else value.replace(tzinfo=UTC)
+
+
+class ScheduledTaskCreateRequest(BaseModel):
+    thread_id: str | None = None
+    context_mode: str = "fresh_thread_per_run"
+    title: str = Field(min_length=1)
+    prompt: str = Field(min_length=1)
+    schedule_type: str
+    schedule_spec: dict[str, Any]
+    timezone: str
+
+
+class ScheduledTaskUpdateRequest(BaseModel):
+    context_mode: str | None = None
+    thread_id: str | None = None
+    title: str | None = Field(default=None, min_length=1)
+    prompt: str | None = Field(default=None, min_length=1)
+    schedule_spec: dict[str, Any] | None = None
+    timezone: str | None = None
+
+
+class ScheduledTaskResponse(BaseModel):
+    """One scheduled task as the client sees it.
+
+    Mirrors the frontend's `ScheduledTask` type field for field.
+    """
+
+    id: str
+    thread_id: str | None
+    context_mode: str
+    title: str
+    prompt: str
+    schedule_type: str
+    schedule_spec: dict[str, str]
+    timezone: str
+    status: str
+    next_run_at: UtcTimestamp | None
+    last_run_at: UtcTimestamp | None
+    last_run_id: str | None
+    last_thread_id: str | None
+    last_error: str | None
+    run_count: int
+    created_at: UtcTimestamp
+    updated_at: UtcTimestamp
+
+    @classmethod
+    def from_domain(cls, task: ScheduledTask) -> ScheduledTaskResponse:
+        return cls(
+            id=task.task_id,
+            thread_id=task.thread_id,
+            context_mode=str(task.context_mode),
+            title=task.title,
+            prompt=task.prompt,
+            schedule_type=str(task.schedule.schedule_type),
+            schedule_spec=spec_to_wire(task.schedule),
+            timezone=task.schedule.timezone,
+            status=str(task.status),
+            next_run_at=_utc(task.next_run_at),
+            last_run_at=_utc(task.last_run_at),
+            last_run_id=task.last_run_id,
+            last_thread_id=task.last_thread_id,
+            last_error=task.last_error,
+            run_count=task.run_count,
+            created_at=_utc(task.created_at),
+            updated_at=_utc(task.updated_at),
+        )
+
+
+class ScheduledRunResponse(BaseModel):
+    """One execution record. Mirrors the frontend's `ScheduledTaskRun` type."""
+
+    id: str
+    task_id: str
+    thread_id: str
+    run_id: str | None
+    scheduled_for: UtcTimestamp
+    trigger: str
+    status: str
+    error: str | None
+    started_at: UtcTimestamp | None
+    finished_at: UtcTimestamp | None
+    created_at: UtcTimestamp
+
+    @classmethod
+    def from_domain(cls, run: ScheduledRun) -> ScheduledRunResponse:
+        return cls(
+            id=run.record_id,
+            task_id=run.task_id,
+            thread_id=run.thread_id,
+            run_id=run.run_id,
+            scheduled_for=_utc(run.scheduled_for),
+            trigger=str(run.trigger),
+            status=str(run.status),
+            error=run.error,
+            started_at=_utc(run.started_at),
+            finished_at=_utc(run.finished_at),
+            created_at=_utc(run.created_at),
+        )
+
+
+class TriggerResponse(BaseModel):
+    id: str
+    triggered: bool
+
+
+class DeleteResponse(BaseModel):
+    id: str
+    deleted: bool

backend/app/gateway/routers/schedule/router.py

@@ -0,0 +1,237 @@
+"""Primary adapter -- the scheduled-task HTTP API.
+
+Everything here is protocol translation. Parse the body into domain values,
+call one service method, render the result, and map domain errors onto status
+codes. What is deliberately *absent* is the giveaway: no cron normalisation,
+no `next_run_at` arithmetic, no re-arm rule, no ownership checks written out
+by hand. Those moved into the aggregate, where they are covered by domain
+tests that never construct an HTTP request.
+
+The error mapping is one table rather than a `raise HTTPException` per branch,
+so a new domain error surfaces as a 500 that has to be classified, instead of
+being silently swallowed by whichever `except` happened to be nearest.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from functools import wraps
+from typing import Annotated
+
+from fastapi import APIRouter, HTTPException, Query, Request
+
+from app.gateway.authz import require_permission
+from app.gateway.deps import ScheduleServiceDep, get_optional_user_from_request
+from app.gateway.routers.schedule.models import (
+    DeleteResponse,
+    ScheduledRunResponse,
+    ScheduledTaskCreateRequest,
+    ScheduledTaskResponse,
+    ScheduledTaskUpdateRequest,
+    TriggerResponse,
+)
+from app.gateway.routers.schedule.spec_wire import spec_to_wire, wire_to_spec
+from deerflow.domain.schedule.model import (
+    DispatchOutcome,
+    InvalidContextModeError,
+    InvalidScheduleError,
+    ScheduleError,
+    TaskNotFoundError,
+    TaskNotMutableError,
+    ThreadNotFoundError,
+)
+from deerflow.domain.schedule.service import ContextChange
+
+router = APIRouter(prefix="/api", tags=["scheduled-tasks"])
+
+# One family of domain errors, one place they become a protocol.
+_STATUS_BY_ERROR: dict[type[ScheduleError], int] = {
+    # "Not yours" and "does not exist" are both 404 by design -- the service
+    # already refuses to distinguish them, and so must the status code.
+    TaskNotFoundError: 404,
+    ThreadNotFoundError: 404,
+    InvalidScheduleError: 422,
+    InvalidContextModeError: 422,
+    TaskNotMutableError: 409,
+}
+
+
+def _map_domain_errors(handler):
+    """Translate domain errors raised by the service into HTTP responses.
+
+    An unclassified `ScheduleError` becomes a 500 on purpose: a new domain
+    error is a new protocol decision, and defaulting it to 4xx would let it
+    ship as a client error nobody chose.
+    """
+
+    @wraps(handler)
+    async def wrapper(*args, **kwargs):
+        try:
+            return await handler(*args, **kwargs)
+        except ScheduleError as exc:
+            status = _STATUS_BY_ERROR.get(type(exc))
+            if status is None:
+                raise
+            raise HTTPException(status_code=status, detail=str(exc)) from exc
+
+    return wrapper
+
+
+async def _require_user_id(request: Request) -> str:
+    user = await get_optional_user_from_request(request)
+    if user is None:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    return str(user.id)
+
+
+@router.get("/scheduled-tasks", response_model=list[ScheduledTaskResponse])
+@require_permission("threads", "read")
+@_map_domain_errors
+async def list_scheduled_tasks(request: Request, service: ScheduleServiceDep):
+    user = await get_optional_user_from_request(request)
+    if user is None:
+        # Not 401: an unauthenticated listing has always been an empty one.
+        return []
+    return [ScheduledTaskResponse.from_domain(task) for task in await service.list_tasks(str(user.id))]
+
+
+@router.post("/scheduled-tasks", response_model=ScheduledTaskResponse)
+@require_permission("threads", "write")
+@_map_domain_errors
+async def create_scheduled_task(request: Request, body: ScheduledTaskCreateRequest, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    task = await service.create_task(
+        user_id=user_id,
+        title=body.title,
+        prompt=body.prompt,
+        schedule=wire_to_spec(body.schedule_type, body.schedule_spec, body.timezone),
+        context_mode=body.context_mode,
+        thread_id=body.thread_id,
+        now=datetime.now(UTC),
+    )
+    return ScheduledTaskResponse.from_domain(task)
+
+
+@router.get("/scheduled-tasks/{task_id}", response_model=ScheduledTaskResponse)
+@require_permission("threads", "read")
+@_map_domain_errors
+async def get_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    return ScheduledTaskResponse.from_domain(await service.get_task(task_id, user_id=user_id))
+
+
+@router.patch("/scheduled-tasks/{task_id}", response_model=ScheduledTaskResponse)
+@require_permission("threads", "write")
+@_map_domain_errors
+async def update_scheduled_task(
+    task_id: str,
+    request: Request,
+    body: ScheduledTaskUpdateRequest,
+    service: ScheduleServiceDep,
+):
+    user_id = await _require_user_id(request)
+    # `exclude_none` rather than `exclude_unset`, matching the pre-migration
+    # router: an explicit `null` has always meant "not supplied" on this
+    # endpoint, and unbinding a thread is expressed by switching
+    # `context_mode`, not by nulling `thread_id`.
+    supplied = body.model_dump(exclude_none=True)
+
+    changes_schedule = "schedule_spec" in supplied or "timezone" in supplied
+    changes_context = "context_mode" in supplied or "thread_id" in supplied
+
+    # Both a schedule and a context change need the parts the client omitted,
+    # so the current task is read once and supplies the defaults. That one
+    # extra fetch is what lets the service receive whole value objects instead
+    # of a patch of loose fields.
+    current = await service.get_task(task_id, user_id=user_id) if changes_schedule or changes_context else None
+
+    schedule = None
+    if current is not None and changes_schedule:
+        schedule = wire_to_spec(
+            # The schedule *type* is not patchable; only its spec and zone are.
+            str(current.schedule.schedule_type),
+            supplied.get("schedule_spec", spec_to_wire(current.schedule)),
+            supplied.get("timezone", current.schedule.timezone),
+        )
+
+    context = None
+    if current is not None and changes_context:
+        context = ContextChange(
+            context_mode=supplied.get("context_mode", str(current.context_mode)),
+            thread_id=supplied.get("thread_id", current.thread_id),
+        )
+
+    task = await service.update_task(
+        task_id,
+        user_id=user_id,
+        now=datetime.now(UTC),
+        title=supplied.get("title"),
+        prompt=supplied.get("prompt"),
+        schedule=schedule,
+        context=context,
+    )
+    return ScheduledTaskResponse.from_domain(task)
+
+
+@router.post("/scheduled-tasks/{task_id}/pause", response_model=ScheduledTaskResponse)
+@require_permission("threads", "write")
+@_map_domain_errors
+async def pause_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    return ScheduledTaskResponse.from_domain(await service.pause_task(task_id, user_id=user_id))
+
+
+@router.post("/scheduled-tasks/{task_id}/resume", response_model=ScheduledTaskResponse)
+@require_permission("threads", "write")
+@_map_domain_errors
+async def resume_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    return ScheduledTaskResponse.from_domain(await service.resume_task(task_id, user_id=user_id))
+
+
+@router.post("/scheduled-tasks/{task_id}/trigger", response_model=TriggerResponse)
+@require_permission("threads", "write")
+@_map_domain_errors
+async def trigger_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    result = await service.trigger_task(task_id, user_id=user_id, now=datetime.now(UTC))
+    if result.outcome is DispatchOutcome.CONFLICT:
+        raise HTTPException(status_code=409, detail=result.error or "Scheduled task trigger conflicted with an active run")
+    if result.outcome is DispatchOutcome.FAILED:
+        # 502, not 500: the failure is downstream of this API -- the run could
+        # not be started -- and the task itself is intact.
+        raise HTTPException(status_code=502, detail=result.error or "Scheduled task trigger failed")
+    return TriggerResponse(id=task_id, triggered=True)
+
+
+@router.delete("/scheduled-tasks/{task_id}", response_model=DeleteResponse)
+@require_permission("threads", "write")
+@_map_domain_errors
+async def delete_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    await service.delete_task(task_id, user_id=user_id)
+    return DeleteResponse(id=task_id, deleted=True)
+
+
+@router.get("/scheduled-tasks/{task_id}/runs", response_model=list[ScheduledRunResponse])
+@require_permission("threads", "read")
+@_map_domain_errors
+async def list_scheduled_task_runs(
+    task_id: str,
+    request: Request,
+    service: ScheduleServiceDep,
+    limit: Annotated[int, Query(ge=1, le=200)] = 50,
+    offset: Annotated[int, Query(ge=0)] = 0,
+):
+    user_id = await _require_user_id(request)
+    runs = await service.list_task_runs(task_id, user_id=user_id, limit=limit, offset=offset)
+    return [ScheduledRunResponse.from_domain(run) for run in runs]
+
+
+@router.get("/threads/{thread_id}/scheduled-tasks", response_model=list[ScheduledTaskResponse])
+@require_permission("threads", "read", owner_check=True)
+@_map_domain_errors
+async def list_thread_scheduled_tasks(thread_id: str, request: Request, service: ScheduleServiceDep):
+    user_id = await _require_user_id(request)
+    tasks = await service.list_tasks_by_thread(user_id, thread_id)
+    return [ScheduledTaskResponse.from_domain(task) for task in tasks]

backend/app/gateway/routers/schedule/spec_wire.py

@@ -0,0 +1,68 @@
+"""Boundary mapping -- the schedule_spec HTTP body field.
+
+The primary adapter's own translation between the request/response shape and
+`ScheduleSpec`. Its counterpart is `app/adapters/schedule/spec_column.py`, which
+does the same job for the stored JSON column; see that module for why the two
+are kept apart rather than shared.
+
+The split is deliberate: **structural** checks (is the key present? is it a
+str?) belong here, **value** rules (5-field cron, resolvable timezone, run_at
+present) belong to `ScheduleSpec.__post_init__`. That is why this module is
+thin -- most of what could go wrong is caught one layer in, and reported with
+the same domain error, so the router maps one family onto 422.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Any
+
+from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType
+
+
+def wire_to_spec(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec:
+    """Parse the submitted triple into the value object.
+
+    Raises:
+        InvalidScheduleError: unknown schedule type, or the type's required key
+            is missing or not a string. Raising a *domain* error from a primary
+            adapter is intentional -- domain errors are the vocabulary the outer
+            ring uses to say "this violates a domain rule", and the router maps
+            that one family uniformly onto 422.
+    """
+    try:
+        kind = ScheduleType(schedule_type)
+    except ValueError as exc:
+        raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc
+
+    fields = spec or {}
+    if kind is ScheduleType.CRON:
+        raw_cron = fields.get("cron")
+        if not isinstance(raw_cron, str):
+            raise InvalidScheduleError("cron schedule requires schedule_spec.cron")
+        return ScheduleSpec.cron_schedule(raw_cron, timezone)
+
+    raw_run_at = fields.get("run_at")
+    if not isinstance(raw_run_at, str):
+        raise InvalidScheduleError("once schedule requires run_at")
+    try:
+        run_at = datetime.fromisoformat(raw_run_at)
+    except ValueError as exc:
+        raise InvalidScheduleError(f"once schedule has an unparseable run_at: {raw_run_at!r}") from exc
+    return ScheduleSpec.once_at(run_at, timezone)
+
+
+def spec_to_wire(spec: ScheduleSpec) -> dict[str, str]:
+    """Rebuild the response body shape.
+
+    Note this emits the normalized value rather than echoing the caller's
+    bytes: the frontend submits an already-UTC-aware ISO value
+    (`zonedLocalToUtcIso`), so a trailing-Z input comes back as "+00:00". Both
+    forms parse on either side, so the normalization is deliberate --
+    preferable to carrying the raw dict on the value object just to preserve
+    the exact input spelling.
+    """
+    if spec.schedule_type is ScheduleType.CRON:
+        return {"cron": spec.cron or ""}
+    return {"run_at": spec.run_at.isoformat() if spec.run_at else ""}

backend/docs/SCHEDULE_DESIGN_zh.md

@@ -38,15 +38,15 @@ run 结束后回写执行记录，并算出下一次的时间
 
 ---
 
-## 2. 当前状态：新旧并存
+## 2. 当前状态：生产已切换，旧代码待删
 
 **这一节请先读，否则你会在代码库里迷路。**
 
-模块正处于六边形重构的中途。**整个内圈已经落地**，但生产代码路径还没有切过来：
+生产路径**已经走新架构**。旧代码还留在仓库里，但已经没有任何东西装配它，下一个提交负责删除：
 
 ```mermaid
 flowchart LR
-    subgraph NEW["✅ 已落地 · 内圈"]
+    subgraph IN["✅ 内圈 · harness"]
         direction TB
         SV["service.py<br/>ScheduleService（用例编排）"]
         PO["ports.py<br/>4 个 Protocol + 2 个 DTO"]
@@ -54,29 +54,37 @@ flowchart LR
         SV --> PO
         SV --> M
     end
-    subgraph OLD["⏳ 仍是旧形态 · 生产路径"]
+    subgraph PRIM["✅ 主适配器 · app"]
-        R["gateway/routers/scheduled_tasks.py<br/>入参校验 + 业务判断混在一起"]
+        R["gateway/routers/schedule/<br/>router · models · spec_wire"]
-        S["app/scheduler/service.py<br/>轮询 + 派发编排 + 状态推导"]
+        PL["scheduler/poller.py<br/>轮询时钟"]
-        P["persistence/scheduled_task*/sql.py<br/>仓储返回裸 dict"]
-        C["deerflow/scheduler/schedules.py<br/>时区 / cron 计算"]
     end
-    subgraph TODO["🚧 待建 · 外圈"]
+    subgraph SEC["✅ 从适配器 · app"]
-        AD["app/adapters/schedule/<br/>app/scheduler/poller.py"]
+        AD["adapters/schedule/<br/>两个仓储 · run_launcher · thread_lookup<br/>spec_column · run_outcome_mapping"]
+    end
+    subgraph CR["✅ 组合根"]
+        CO["composition.py<br/>build_domain_services()"]
+    end
+    subgraph DEAD["🗑️ 已停用 · 待删除"]
+        S["app/scheduler/service.py"]
+        OR["gateway/routers/scheduled_tasks.py"]
+        P["persistence/scheduled_task*/sql.py"]
+        C["deerflow/scheduler/"]
     end
 
-    R -.->|尚未调用| SV
+    R --> SV
-    S -.->|尚未调用| SV
+    PL --> SV
-    AD -.->|将实现| PO
+    AD -.->|实现| PO
+    CO -->|装配| SV
 
-    style NEW fill:#eef6ff
+    style IN fill:#eef6ff
-    style OLD fill:#f5f5f5
+    style DEAD fill:#f5f5f5
 ```
 
 含义很具体：
 
-- **`domain/schedule/` 是唯一的真相声明处**，但目前只有测试在用它。运行中的定时任务走的仍是 `app/scheduler/service.py` 那套。
+- **`domain/schedule/` 是唯一的真相声明处**，运行中的定时任务已经走它。改一条业务规则只改这一处。
-- 两边**规则内容一致**（内圈是逐条从旧代码搬迁的，每个方法的 docstring 都标了来源行号），但**代码是重复的**。这是迁移中间态的正常代价。
+- `DEAD` 里的文件仍能编译、仍有测试，但**不再被任何装配路径引用**——`app.py` 启动的是 `SchedulePoller`，注册的是 `routers/schedule/`，完成回调走 `composition.py` 装的那个。留着只是为了让删除单独成为一个可审查的提交。
-- 你现在改一条业务规则，要**两边都改**，直到迁移完成。旧位置见 §12 的索引。
+- 旧的重复规则因此不再需要"两边都改"。如果你在 `DEAD` 里发现和内圈不一致的逻辑，以内圈为准。
 - 新增业务规则请**只写在内圈**，然后在旧位置调用它——不要再往 router / 旧 service 里加新的判断。
 
 **还差什么**：四个端口都没有真实实现。适配器落地后，`app/scheduler/service.py`、`deerflow/scheduler/` 整包、两个 `sql.py` 都会消失，router 瘦身成协议转换。
@@ -612,19 +620,44 @@ await service.update_task(
 | [`tests/test_schedule_domain.py`](../tests/test_schedule_domain.py) | 域测试，全同步零 IO；四张真值表逐格覆盖 |
 | [`tests/test_schedule_service.py`](../tests/test_schedule_service.py) | 用例测试；完整生命周期跑在 fake 上，是迁移的验收标准 |
 | [`tests/schedule_fakes.py`](../tests/schedule_fakes.py) | 四个端口的内存实现 |
-| [`tests/test_schedule_fakes.py`](../tests/test_schedule_fakes.py) | 端口语义；适配器落地后升级成契约测试 |
+| [`tests/test_schedule_fakes.py`](../tests/test_schedule_fakes.py) | 契约测试：31 用例 × 内存 fake + 真 sqlite 两套实现 |
 
-**尚未迁移（生产路径，见 §2）**
+**主适配器（入口）**
 
 | 文件 | 内容 |
 |---|---|
-| [`app/gateway/routers/scheduled_tasks.py`](../app/gateway/routers/scheduled_tasks.py) | 10 个 HTTP 端点，含入参校验与业务判断 |
+| [`gateway/routers/schedule/router.py`](../app/gateway/routers/schedule/router.py) | 10 个 HTTP 端点，只做协议转换 + 领域错误→状态码 |
-| [`app/scheduler/service.py`](../app/scheduler/service.py) | 轮询循环、派发编排、状态推导、完成回调、启动清扫 |
+| [`gateway/routers/schedule/models.py`](../app/gateway/routers/schedule/models.py) | 请求/响应模型；响应是白名单，不是 ORM 转储 |
-| [`deerflow/scheduler/schedules.py`](../packages/harness/deerflow/scheduler/schedules.py) | 时区 / cron / 下次时间计算 |
+| [`gateway/routers/schedule/spec_wire.py`](../app/gateway/routers/schedule/spec_wire.py) | HTTP body ↔ `ScheduleSpec` |
-| [`persistence/scheduled_tasks/`](../packages/harness/deerflow/persistence/scheduled_tasks/) | 任务表 ORM + 仓储 |
+| [`app/scheduler/poller.py`](../app/scheduler/poller.py) | 轮询时钟 + 启动恢复 |
-| [`persistence/scheduled_task_runs/`](../packages/harness/deerflow/persistence/scheduled_task_runs/) | 执行表 ORM + 仓储，含唯一索引定义 |
+
+**从适配器**
+
+| 文件 | 内容 |
+|---|---|
+| [`adapters/schedule/scheduled_task_repository.py`](../app/adapters/schedule/scheduled_task_repository.py) | 自有持久化；`FOR UPDATE SKIP LOCKED` 与 `protect_terminal` CAS |
+| [`adapters/schedule/scheduled_run_repository.py`](../app/adapters/schedule/scheduled_run_repository.py) | 自有持久化；`IntegrityError → ActiveRunConflictError` 的翻译点 |
+| [`adapters/schedule/run_launcher.py`](../app/adapters/schedule/run_launcher.py) | 防腐层；`ConflictError` / `HTTPException(409)` → `ThreadBusyError` |
+| [`adapters/schedule/thread_lookup.py`](../app/adapters/schedule/thread_lookup.py) | 防腐层；`check_access(require_existing=True)` |
+| [`adapters/schedule/spec_column.py`](../app/adapters/schedule/spec_column.py) | JSON 列 ↔ `ScheduleSpec`（与 `spec_wire` 成对，见其 docstring） |
+| [`adapters/schedule/run_outcome_mapping.py`](../app/adapters/schedule/run_outcome_mapping.py) | `RunRecord → RunOutcome \| None`，承接旧完成钩子的内联过滤 |
+
+**组合根**
+
+| 文件 | 内容 |
+|---|---|
+| [`app/composition.py`](../app/composition.py) | `build_domain_services()` · `build_run_completion_hook()` · `build_schedule_policy()` |
 | [`config/scheduler_config.py`](../packages/harness/deerflow/config/scheduler_config.py) | `enabled` `poll_interval_seconds` `lease_seconds` `max_concurrent_runs` `min_once_delay_seconds` |
 
+**已停用，待删除（见 §2）**
+
+| 文件 | 内容 |
+|---|---|
+| [`app/gateway/routers/scheduled_tasks.py`](../app/gateway/routers/scheduled_tasks.py) | 旧端点，已不注册 |
+| [`app/scheduler/service.py`](../app/scheduler/service.py) | 旧轮询 + 派发编排，已不装配 |
+| [`deerflow/scheduler/schedules.py`](../packages/harness/deerflow/scheduler/schedules.py) | 时区 / cron 计算，规则已进 `ScheduleSpec` |
+| [`persistence/scheduled_task*/sql.py`](../packages/harness/deerflow/persistence/scheduled_tasks/) | 旧仓储（返回裸 dict）。**ORM 行与唯一索引定义仍在用，不要一起删** |
+
 **前端**
 
 `frontend/src/core/scheduled-tasks/`（类型、API、cron 解析、预设配方）与 `frontend/src/app/workspace/scheduled-tasks/page.tsx`。

backend/tests/test_composition.py

@@ -0,0 +1,141 @@
+"""Tests for the composition root.
+
+The point of extracting `build_domain_services` from the lifespan is that the
+rules below become assertions. Before, "a memory backend means no service,
+and the routes answer 503" was a comment inside a 180-line startup function
+that no test could reach without booting the whole application.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.composition import DomainServices, build_domain_services, build_schedule_policy
+from deerflow.config.scheduler_config import SchedulerConfig
+from deerflow.domain.schedule.model import SchedulePolicy
+
+
+class _StubRunStore:
+    async def get(self, run_id: str):
+        return None
+
+
+class _StubThreadStore:
+    async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
+        return True
+
+
+async def _launch_run(**kwargs):
+    return {"run_id": "run-1", "thread_id": kwargs.get("thread_id", "thread-1")}
+
+
+def _build(session_factory, scheduler_config=None) -> DomainServices:
+    return build_domain_services(
+        session_factory=session_factory,
+        run_store=_StubRunStore(),
+        thread_store=_StubThreadStore(),
+        launch_run=_launch_run,
+        scheduler_config=scheduler_config or SchedulerConfig(),
+    )
+
+
+class TestMemoryBackend:
+    """`session_factory is None` is how `database.backend: memory` presents."""
+
+    def test_no_session_factory_yields_no_services(self):
+        services = _build(None)
+        assert services.feedback is None
+        assert services.schedule is None
+
+    def test_it_does_not_degrade_to_an_in_memory_implementation(self):
+        """Refusing is the intended behaviour: a scheduled task that silently
+        vanishes on restart is worse than one the API declines to accept."""
+        assert _build(None).schedule is None
+
+
+class TestSqlBackend:
+    def test_a_session_factory_yields_both_services(self):
+        """A fake sessionmaker is enough -- wiring must not touch the database,
+        which is what makes this assertable without a live engine."""
+        services = _build(object())
+        assert services.feedback is not None
+        assert services.schedule is not None
+
+    def test_the_two_contexts_are_wired_independently(self):
+        services = _build(object())
+        assert services.feedback is not services.schedule
+
+
+class TestSchedulePolicy:
+    def test_every_threshold_comes_from_the_operator_config(self):
+        policy = build_schedule_policy(
+            SchedulerConfig(
+                min_once_delay_seconds=30,
+                max_concurrent_runs=7,
+                lease_seconds=90,
+            )
+        )
+        assert policy == SchedulePolicy(
+            min_once_delay_seconds=30,
+            max_concurrent_runs=7,
+            lease_seconds=90,
+        )
+
+    def test_the_domain_defaults_are_not_what_production_gets(self):
+        """The domain's defaults are permissive so that "nobody configured a
+        policy" invents no business constraint. Production must not inherit
+        them by accident -- the config's own defaults are the real values."""
+        from_config = build_schedule_policy(SchedulerConfig())
+        assert from_config != SchedulePolicy()
+        assert from_config.min_once_delay_seconds == SchedulerConfig().min_once_delay_seconds
+
+    @pytest.mark.parametrize(
+        ("field", "value"),
+        [
+            ("min_once_delay_seconds", 45),
+            ("max_concurrent_runs", 5),
+            ("lease_seconds", 300),
+        ],
+    )
+    def test_each_field_is_mapped_from_its_own_config_key(self, field, value):
+        """Guards against two thresholds being wired from one key -- a
+        transposition the type checker cannot see, since all three are ints."""
+        policy = build_schedule_policy(SchedulerConfig(**{field: value}))
+        assert getattr(policy, field) == value
+
+    def test_poll_interval_is_not_part_of_the_policy(self):
+        """How often to look is the poller's business, not a rule any task is
+        subject to."""
+        assert not hasattr(SchedulePolicy(), "poll_interval_seconds")
+
+
+class TestEveryPortGetsTheRightAdapter:
+    """Deliberately white-box: verifying which adapter landed in which slot is
+    the entire job of a composition root, and all four are keyword arguments
+    of compatible shape, so a transposition type-checks cleanly and would only
+    surface as a production error.
+    """
+
+    def test_each_schedule_port_is_filled_with_its_own_adapter(self):
+        from app.adapters.schedule.run_launcher import GatewayRunLauncher
+        from app.adapters.schedule.scheduled_run_repository import SqlScheduledRunRepository
+        from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
+        from app.adapters.schedule.thread_lookup import ThreadStoreThreadLookup
+
+        service = _build(object()).schedule
+        assert isinstance(service._tasks, SqlScheduledTaskRepository)
+        assert isinstance(service._runs, SqlScheduledRunRepository)
+        assert isinstance(service._launcher, GatewayRunLauncher)
+        assert isinstance(service._threads, ThreadStoreThreadLookup)
+
+    def test_each_feedback_port_is_filled_with_its_own_adapter(self):
+        from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
+        from app.adapters.feedback.run_lookup import RunStoreRunLookup
+
+        service = _build(object()).feedback
+        assert isinstance(service._repository, SqlFeedbackRepository)
+        assert isinstance(service._runs, RunStoreRunLookup)
+
+    def test_the_policy_reaches_the_service(self):
+        service = _build(object(), SchedulerConfig(max_concurrent_runs=9)).schedule
+        assert service._policy.max_concurrent_runs == 9

backend/tests/test_gateway_run_drain_shutdown.py

@@ -30,6 +30,7 @@ from typing import Annotated, TypedDict
 import pytest
 from langgraph.checkpoint.memory import InMemorySaver
 
+from deerflow.config.scheduler_config import SchedulerConfig
 from deerflow.runtime import RunManager, RunStatus
 
 
@@ -196,7 +197,7 @@ async def test_langgraph_runtime_drains_runs_before_closing_checkpointer(monkeyp
     monkeypatch.setattr(RunManager, "shutdown", spy_shutdown, raising=False)
 
     app = FastAPI()
-    startup_config = SimpleNamespace(database=SimpleNamespace(backend="memory", checkpoint_channel_mode="full"), run_events=None)
+    startup_config = SimpleNamespace(database=SimpleNamespace(backend="memory", checkpoint_channel_mode="full"), run_events=None, scheduler=SchedulerConfig())
 
     async with langgraph_runtime(app, startup_config):
         pass

backend/tests/test_gateway_run_recovery.py

@@ -14,6 +14,7 @@ from fastapi import FastAPI
 import deerflow.runtime as runtime_module
 from app.gateway import deps as gateway_deps
 from deerflow.config.run_ownership_config import RunOwnershipConfig
+from deerflow.config.scheduler_config import SchedulerConfig
 from deerflow.persistence import engine as engine_module
 from deerflow.persistence import thread_meta as thread_meta_module
 from deerflow.runtime import END_SENTINEL, MemoryStreamBridge, RunManager
@@ -224,6 +225,7 @@ async def test_sqlite_runtime_reconciles_orphaned_runs_on_startup(monkeypatch):
         database=SimpleNamespace(backend="sqlite", checkpoint_channel_mode="full"),
         run_events=SimpleNamespace(backend="memory"),
         stream_bridge=SimpleNamespace(recovered_stream_cleanup_delay_seconds=60.0),
+        scheduler=SchedulerConfig(),
     )
     thread_store = _FakeThreadStore()
     stream_bridge = _FakeStreamBridge(existing_streams={"run-1"})
@@ -269,6 +271,7 @@ async def test_sqlite_runtime_does_not_mark_thread_error_when_newer_run_is_succe
         database=SimpleNamespace(backend="sqlite", checkpoint_channel_mode="full"),
         run_events=SimpleNamespace(backend="memory"),
         stream_bridge=SimpleNamespace(recovered_stream_cleanup_delay_seconds=60.0),
+        scheduler=SchedulerConfig(),
     )
     thread_store = _FakeThreadStore()
     stream_bridge = _FakeStreamBridge(existing_streams={"old-running"})

backend/tests/test_schedule_response_models.py

@@ -0,0 +1,184 @@
+"""Tests for the scheduled-task HTTP response models.
+
+Two things are being pinned, and they pull in opposite directions:
+
+  - **the leak is closed** -- server-owned and scheduler-internal fields that
+    the pre-migration router dumped from the ORM row must not appear, and
+  - **nothing the client already reads changed** -- the field set and the
+    timestamp spelling have to stay byte-compatible with what the legacy
+    `to_dict()` + `coerce_iso` path emitted, or the frontend breaks.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta, timezone
+
+import pytest
+
+from app.gateway.routers.schedule.models import ScheduledRunResponse, ScheduledTaskResponse
+from deerflow.domain.schedule.model import (
+    ContextMode,
+    RunStatus,
+    ScheduledRun,
+    ScheduledTask,
+    ScheduleSpec,
+    TaskStatus,
+    TriggerKind,
+)
+
+# Exactly the frontend's `ScheduledTask` type (frontend/src/core/scheduled-tasks/types.ts).
+FRONTEND_TASK_FIELDS = {
+    "id",
+    "thread_id",
+    "context_mode",
+    "title",
+    "prompt",
+    "schedule_type",
+    "schedule_spec",
+    "timezone",
+    "status",
+    "next_run_at",
+    "last_run_at",
+    "last_run_id",
+    "last_thread_id",
+    "last_error",
+    "run_count",
+    "created_at",
+    "updated_at",
+}
+
+# Exactly the frontend's `ScheduledTaskRun` type.
+FRONTEND_RUN_FIELDS = {
+    "id",
+    "task_id",
+    "thread_id",
+    "run_id",
+    "scheduled_for",
+    "trigger",
+    "status",
+    "error",
+    "started_at",
+    "finished_at",
+    "created_at",
+}
+
+# What the ORM dump used to expose. Lease fields never reached the domain, so
+# they cannot leak now; the other three can, and must not.
+LEAKED_FIELDS = {"user_id", "assistant_id", "overlap_policy", "lease_owner", "lease_expires_at"}
+
+
+def _task(**overrides) -> ScheduledTask:
+    defaults = dict(
+        task_id="task-1",
+        user_id="user-1",
+        title="Morning digest",
+        prompt="summarize",
+        schedule=ScheduleSpec.cron_schedule("0 9 * * *", "Asia/Shanghai"),
+        context_mode=ContextMode.REUSE_THREAD,
+        thread_id="thread-1",
+        status=TaskStatus.ENABLED,
+        next_run_at=datetime(2026, 8, 1, 1, 0, tzinfo=UTC),
+        last_run_at=datetime(2026, 7, 31, 1, 0, tzinfo=UTC),
+        last_run_id="run-1",
+        last_thread_id="thread-1",
+        last_error=None,
+        run_count=3,
+        created_at=datetime(2026, 7, 1, 0, 0, tzinfo=UTC),
+        updated_at=datetime(2026, 7, 31, 1, 0, tzinfo=UTC),
+    )
+    return ScheduledTask(**{**defaults, **overrides})
+
+
+def _run(**overrides) -> ScheduledRun:
+    defaults = dict(
+        record_id="task-run-1",
+        task_id="task-1",
+        thread_id="thread-1",
+        scheduled_for=datetime(2026, 8, 1, 1, 0, tzinfo=UTC),
+        trigger=TriggerKind.SCHEDULED,
+        status=RunStatus.SUCCESS,
+        run_id="run-1",
+        error=None,
+        started_at=datetime(2026, 8, 1, 1, 0, 1, tzinfo=UTC),
+        finished_at=datetime(2026, 8, 1, 1, 0, 9, tzinfo=UTC),
+        created_at=datetime(2026, 8, 1, 1, 0, tzinfo=UTC),
+    )
+    return ScheduledRun(**{**defaults, **overrides})
+
+
+class TestTheLeakIsClosed:
+    def test_no_server_owned_field_is_published(self):
+        emitted = set(ScheduledTaskResponse.from_domain(_task()).model_dump())
+        assert emitted & LEAKED_FIELDS == set()
+
+    def test_the_field_set_is_exactly_what_the_frontend_declares(self):
+        """Both directions matter: an extra field is a leak, a missing one
+        breaks a client that reads it."""
+        assert set(ScheduledTaskResponse.from_domain(_task()).model_dump()) == FRONTEND_TASK_FIELDS
+
+    def test_run_records_publish_exactly_the_declared_fields(self):
+        assert set(ScheduledRunResponse.from_domain(_run()).model_dump()) == FRONTEND_RUN_FIELDS
+
+
+class TestFieldMapping:
+    def test_the_aggregate_id_is_published_as_id(self):
+        assert ScheduledTaskResponse.from_domain(_task()).id == "task-1"
+
+    def test_the_record_id_is_published_as_id(self):
+        assert ScheduledRunResponse.from_domain(_run()).id == "task-run-1"
+
+    def test_the_schedule_value_object_is_flattened_into_three_fields(self):
+        """The client's shape predates the value object and keeps the three
+        columns separate; the flattening is this model's job, not the
+        domain's."""
+        response = ScheduledTaskResponse.from_domain(_task())
+        assert response.schedule_type == "cron"
+        assert response.schedule_spec == {"cron": "0 9 * * *"}
+        assert response.timezone == "Asia/Shanghai"
+
+    def test_a_once_schedule_flattens_to_run_at(self):
+        task = _task(schedule=ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC"))
+        response = ScheduledTaskResponse.from_domain(task)
+        assert response.schedule_type == "once"
+        assert response.schedule_spec == {"run_at": "2026-08-01T09:00:00+00:00"}
+
+    def test_enums_are_emitted_as_their_string_values(self):
+        """`StrEnum` would serialize acceptably either way, but the client
+        compares against string literals, so the model declares `str`."""
+        payload = json.loads(ScheduledTaskResponse.from_domain(_task()).model_dump_json())
+        assert payload["status"] == "enabled"
+        assert payload["context_mode"] == "reuse_thread"
+
+
+class TestTimestampCompatibility:
+    def test_utc_timestamps_keep_the_legacy_spelling(self):
+        """`coerce_iso` emitted `astimezone(UTC).isoformat()`; anything else
+        is a silent wire change for every client parsing these."""
+        payload = json.loads(ScheduledTaskResponse.from_domain(_task()).model_dump_json())
+        assert payload["next_run_at"] == "2026-08-01T01:00:00+00:00"
+        assert payload["created_at"] == "2026-07-01T00:00:00+00:00"
+
+    def test_a_non_utc_timestamp_is_converted_not_echoed(self):
+        """The legacy path normalised the offset away. A value that reached
+        the aggregate in another zone must not start emitting `+08:00`."""
+        shanghai = timezone(timedelta(hours=8))
+        task = _task(next_run_at=datetime(2026, 8, 1, 9, 0, tzinfo=shanghai))
+        payload = json.loads(ScheduledTaskResponse.from_domain(task).model_dump_json())
+        assert payload["next_run_at"] == "2026-08-01T01:00:00+00:00"
+
+    def test_a_naive_timestamp_is_assumed_utc(self):
+        """Same assumption the repository's `_tz_aware` makes on read."""
+        task = _task(next_run_at=datetime(2026, 8, 1, 1, 0))
+        payload = json.loads(ScheduledTaskResponse.from_domain(task).model_dump_json())
+        assert payload["next_run_at"] == "2026-08-01T01:00:00+00:00"
+
+    @pytest.mark.parametrize("field", ["next_run_at", "last_run_at"])
+    def test_absent_timestamps_stay_null(self, field):
+        payload = json.loads(ScheduledTaskResponse.from_domain(_task(**{field: None})).model_dump_json())
+        assert payload[field] is None
+
+    def test_run_timestamps_use_the_same_spelling(self):
+        payload = json.loads(ScheduledRunResponse.from_domain(_run()).model_dump_json())
+        assert payload["scheduled_for"] == "2026-08-01T01:00:00+00:00"
+        assert payload["finished_at"] == "2026-08-01T01:00:09+00:00"

backend/tests/test_schedule_router.py

@@ -0,0 +1,538 @@
+"""Behaviour tests for the scheduled-task HTTP adapter.
+
+Driven through the handlers with a **real** `ScheduleService` over in-memory
+port fakes, not a mocked service. The router's whole remaining job is protocol
+translation, and half of that is turning domain errors into status codes -- a
+mocked service would let those assertions pass without a domain error ever
+being raised.
+
+`__wrapped__` unwraps `@require_permission` (authorization is covered by its
+own suite) while keeping `_map_domain_errors` in the call path, which is the
+layer under test here.
+
+The legacy `test_scheduled_task_router_behavior.py` remains the reference for
+what this endpoint must do; every scenario it pins has a counterpart below,
+plus the response-shape and error-mapping cases the old dict-returning router
+could not express.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from fastapi import HTTPException
+from schedule_fakes import (
+    FakeRunLauncher,
+    FakeThreadLookup,
+    InMemoryScheduledRunRepository,
+    InMemoryScheduledTaskRepository,
+)
+
+from app.gateway.routers.schedule import router as router_module
+from deerflow.domain.schedule.model import (
+    LaunchFailedError,
+    RunStatus,
+    ScheduledRun,
+    SchedulePolicy,
+    TaskStatus,
+    ThreadBusyError,
+    TriggerKind,
+)
+from deerflow.domain.schedule.service import ScheduleService
+
+USER = "user-1"
+OTHER_USER = "user-2"
+THREAD = "thread-1"
+
+
+class _User:
+    def __init__(self, user_id: str | None) -> None:
+        self.id = user_id
+
+
+@pytest.fixture
+def tasks():
+    return InMemoryScheduledTaskRepository()
+
+
+@pytest.fixture
+def runs():
+    return InMemoryScheduledRunRepository()
+
+
+@pytest.fixture
+def launcher():
+    return FakeRunLauncher()
+
+
+@pytest.fixture
+def service(tasks, runs, launcher):
+    return ScheduleService(
+        tasks=tasks,
+        runs=runs,
+        launcher=launcher,
+        threads=FakeThreadLookup({THREAD: USER}),
+        policy=SchedulePolicy(min_once_delay_seconds=60, max_concurrent_runs=3, lease_seconds=120),
+    )
+
+
+@pytest.fixture
+def as_user(monkeypatch):
+    """Authenticate the handlers as a given user id (None = anonymous)."""
+
+    def _apply(user_id: str | None = USER):
+        async def _resolve(_request):
+            return None if user_id is None else _User(user_id)
+
+        monkeypatch.setattr(router_module, "get_optional_user_from_request", _resolve)
+
+    _apply()
+    return _apply
+
+
+def _call(handler, **kwargs):
+    """Invoke a route handler with authorization unwrapped."""
+    return handler.__wrapped__(request=object(), **kwargs)
+
+
+def _create_body(**overrides):
+    defaults = dict(
+        thread_id=None,
+        context_mode="fresh_thread_per_run",
+        title="Daily summary",
+        prompt="Summarize the thread",
+        schedule_type="cron",
+        schedule_spec={"cron": "0 9 * * *"},
+        timezone="UTC",
+    )
+    return router_module.ScheduledTaskCreateRequest(**{**defaults, **overrides})
+
+
+def _update_body(**fields):
+    return router_module.ScheduledTaskUpdateRequest(**fields)
+
+
+async def _create(service, **overrides):
+    return await _call(router_module.create_scheduled_task, body=_create_body(**overrides), service=service)
+
+
+class TestCreate:
+    @pytest.mark.asyncio
+    async def test_a_cron_task_is_created_and_rendered(self, service, as_user):
+        created = await _create(service)
+        assert created.title == "Daily summary"
+        assert created.schedule_type == "cron"
+        assert created.schedule_spec == {"cron": "0 9 * * *"}
+        assert created.status == "enabled"
+        assert created.next_run_at is not None
+
+    @pytest.mark.asyncio
+    async def test_the_response_carries_no_server_owned_fields(self, service, as_user):
+        """The pre-migration router returned the ORM row, leaking `user_id`,
+        `overlap_policy` and `assistant_id` to every client."""
+        rendered = (await _create(service)).model_dump()
+        assert {"user_id", "assistant_id", "overlap_policy", "lease_owner", "lease_expires_at"} & set(rendered) == set()
+
+    @pytest.mark.asyncio
+    async def test_a_fresh_thread_task_needs_no_thread_id(self, service, as_user):
+        created = await _create(service, context_mode="fresh_thread_per_run", thread_id=None)
+        assert created.thread_id is None
+
+    @pytest.mark.asyncio
+    async def test_reuse_thread_binds_the_thread(self, service, as_user):
+        created = await _create(service, context_mode="reuse_thread", thread_id=THREAD)
+        assert created.thread_id == THREAD
+
+    @pytest.mark.asyncio
+    async def test_reuse_thread_without_a_thread_id_is_422(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, context_mode="reuse_thread", thread_id=None)
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_reuse_of_an_unknown_thread_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, context_mode="reuse_thread", thread_id="thread-nope")
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_reuse_of_someone_elses_thread_is_also_404(self, service, as_user):
+        """Indistinguishable from "does not exist" on purpose: telling them
+        apart would let a caller probe for threads they cannot see."""
+        as_user(OTHER_USER)
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, context_mode="reuse_thread", thread_id=THREAD)
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_an_unknown_schedule_type_is_422(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, schedule_type="teleport")
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_a_cron_without_its_key_is_422(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, schedule_type="cron", schedule_spec={})
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_a_malformed_cron_is_422(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, schedule_type="cron", schedule_spec={"cron": "0 9 * *"})
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_an_unknown_timezone_is_422(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, timezone="Mars/Olympus_Mons")
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_a_once_schedule_too_close_to_now_is_422(self, service, as_user):
+        """`min_once_delay_seconds` is operator policy, and it reaches the
+        aggregate through SchedulePolicy rather than being re-checked here."""
+        soon = (datetime.now(UTC) + timedelta(seconds=5)).isoformat()
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, schedule_type="once", schedule_spec={"run_at": soon})
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_a_once_schedule_in_the_past_is_422(self, service, as_user):
+        past = (datetime.now(UTC) - timedelta(days=1)).isoformat()
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, schedule_type="once", schedule_spec={"run_at": past})
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_an_unknown_context_mode_is_422(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _create(service, context_mode="telepathy")
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_an_anonymous_caller_is_401(self, service, as_user):
+        as_user(None)
+        with pytest.raises(HTTPException) as caught:
+            await _create(service)
+        assert caught.value.status_code == 401
+
+
+class TestRead:
+    @pytest.mark.asyncio
+    async def test_a_task_is_fetched_by_id(self, service, as_user):
+        created = await _create(service)
+        fetched = await _call(router_module.get_scheduled_task, task_id=created.id, service=service)
+        assert fetched.id == created.id
+
+    @pytest.mark.asyncio
+    async def test_an_unknown_task_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.get_scheduled_task, task_id="task-nope", service=service)
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_another_users_task_is_404(self, service, as_user):
+        created = await _create(service)
+        as_user(OTHER_USER)
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.get_scheduled_task, task_id=created.id, service=service)
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_listing_returns_the_users_tasks(self, service, as_user):
+        await _create(service, title="One")
+        await _create(service, title="Two")
+        listed = await _call(router_module.list_scheduled_tasks, service=service)
+        assert {task.title for task in listed} == {"One", "Two"}
+
+    @pytest.mark.asyncio
+    async def test_an_anonymous_listing_is_empty_not_401(self, service, as_user):
+        """Unchanged from the pre-migration router: this endpoint has always
+        answered an unauthenticated GET with an empty list."""
+        as_user(None)
+        assert await _call(router_module.list_scheduled_tasks, service=service) == []
+
+    @pytest.mark.asyncio
+    async def test_thread_listing_filters_by_thread(self, service, as_user):
+        await _create(service, context_mode="reuse_thread", thread_id=THREAD, title="Bound")
+        await _create(service, title="Unbound")
+        listed = await _call(router_module.list_thread_scheduled_tasks, thread_id=THREAD, service=service)
+        assert [task.title for task in listed] == ["Bound"]
+
+
+class TestUpdate:
+    @pytest.mark.asyncio
+    async def test_a_title_is_updated_in_place(self, service, as_user):
+        created = await _create(service)
+        updated = await _call(
+            router_module.update_scheduled_task,
+            task_id=created.id,
+            body=_update_body(title="Renamed"),
+            service=service,
+        )
+        assert updated.title == "Renamed"
+        assert updated.schedule_spec == created.schedule_spec
+
+    @pytest.mark.asyncio
+    async def test_a_schedule_spec_change_keeps_the_existing_timezone(self, service, as_user):
+        """Only one of the three schedule parts was supplied; the router reads
+        the task to complete the value object rather than the service being
+        handed a partial one."""
+        created = await _create(service, timezone="Asia/Shanghai")
+        updated = await _call(
+            router_module.update_scheduled_task,
+            task_id=created.id,
+            body=_update_body(schedule_spec={"cron": "30 2 * * *"}),
+            service=service,
+        )
+        assert updated.schedule_spec == {"cron": "30 2 * * *"}
+        assert updated.timezone == "Asia/Shanghai"
+
+    @pytest.mark.asyncio
+    async def test_a_timezone_change_keeps_the_existing_spec(self, service, as_user):
+        created = await _create(service)
+        updated = await _call(
+            router_module.update_scheduled_task,
+            task_id=created.id,
+            body=_update_body(timezone="Asia/Shanghai"),
+            service=service,
+        )
+        assert updated.schedule_spec == created.schedule_spec
+        assert updated.timezone == "Asia/Shanghai"
+
+    @pytest.mark.asyncio
+    async def test_a_running_task_cannot_be_updated(self, service, tasks, as_user):
+        """Red line: the mutability gate lives in the aggregate now, and the
+        router only maps it onto 409."""
+        created = await _create(service)
+        stored = await tasks.get(created.id, user_id=USER)
+        await tasks.save(stored.__class__(**{**stored.__dict__, "status": TaskStatus.RUNNING}))
+        with pytest.raises(HTTPException) as caught:
+            await _call(
+                router_module.update_scheduled_task,
+                task_id=created.id,
+                body=_update_body(title="Renamed"),
+                service=service,
+            )
+        assert caught.value.status_code == 409
+
+    @pytest.mark.asyncio
+    async def test_updating_an_unknown_task_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _call(
+                router_module.update_scheduled_task,
+                task_id="task-nope",
+                body=_update_body(title="Renamed"),
+                service=service,
+            )
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_a_malformed_new_schedule_is_422(self, service, as_user):
+        created = await _create(service)
+        with pytest.raises(HTTPException) as caught:
+            await _call(
+                router_module.update_scheduled_task,
+                task_id=created.id,
+                body=_update_body(schedule_spec={"cron": "0 9 * *"}),
+                service=service,
+            )
+        assert caught.value.status_code == 422
+
+    @pytest.mark.asyncio
+    async def test_a_terminal_once_task_pushed_into_the_future_is_rearmed(self, service, tasks, as_user):
+        """Red line: without re-arming, the API answers 200 with a
+        `next_run_at` that can never fire, because claiming only admits
+        `enabled` rows."""
+        future = (datetime.now(UTC) + timedelta(days=1)).isoformat()
+        created = await _create(service, schedule_type="once", schedule_spec={"run_at": future})
+        stored = await tasks.get(created.id, user_id=USER)
+        await tasks.save(stored.__class__(**{**stored.__dict__, "status": TaskStatus.FAILED}))
+
+        later = (datetime.now(UTC) + timedelta(days=2)).isoformat()
+        updated = await _call(
+            router_module.update_scheduled_task,
+            task_id=created.id,
+            body=_update_body(schedule_spec={"run_at": later}),
+            service=service,
+        )
+        assert updated.status == "enabled"
+        assert updated.next_run_at is not None
+
+
+class TestPauseResume:
+    @pytest.mark.asyncio
+    async def test_pause_then_resume(self, service, as_user):
+        created = await _create(service)
+        paused = await _call(router_module.pause_scheduled_task, task_id=created.id, service=service)
+        assert paused.status == "paused"
+        resumed = await _call(router_module.resume_scheduled_task, task_id=created.id, service=service)
+        assert resumed.status == "enabled"
+
+    @pytest.mark.asyncio
+    async def test_pausing_a_running_task_is_409(self, service, tasks, as_user):
+        created = await _create(service)
+        stored = await tasks.get(created.id, user_id=USER)
+        await tasks.save(stored.__class__(**{**stored.__dict__, "status": TaskStatus.RUNNING}))
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.pause_scheduled_task, task_id=created.id, service=service)
+        assert caught.value.status_code == 409
+
+    @pytest.mark.asyncio
+    async def test_pausing_an_unknown_task_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.pause_scheduled_task, task_id="task-nope", service=service)
+        assert caught.value.status_code == 404
+
+
+class TestTrigger:
+    """Red line: this endpoint answers exactly 409 / 502 / 200, and the success
+    body is `{"id": ..., "triggered": true}`."""
+
+    @pytest.mark.asyncio
+    async def test_a_launched_trigger_is_200_with_the_documented_body(self, service, as_user):
+        created = await _create(service)
+        response = await _call(router_module.trigger_scheduled_task, task_id=created.id, service=service)
+        assert response.model_dump() == {"id": created.id, "triggered": True}
+
+    @pytest.mark.asyncio
+    async def test_a_busy_thread_is_409(self, service, launcher, as_user):
+        created = await _create(service)
+        launcher.fail_with = ThreadBusyError("thread already has an active run")
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.trigger_scheduled_task, task_id=created.id, service=service)
+        assert caught.value.status_code == 409
+
+    @pytest.mark.asyncio
+    async def test_an_active_run_for_the_same_task_is_409(self, service, runs, as_user):
+        """The other route to a conflict: the task's single active slot is
+        already taken, which the service reports without calling the launcher
+        at all."""
+        created = await _create(service)
+        await runs.add(
+            ScheduledRun(
+                record_id="task-run-existing",
+                task_id=created.id,
+                thread_id="thread-x",
+                scheduled_for=datetime.now(UTC),
+                trigger=TriggerKind.SCHEDULED,
+                status=RunStatus.RUNNING,
+            )
+        )
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.trigger_scheduled_task, task_id=created.id, service=service)
+        assert caught.value.status_code == 409
+
+    @pytest.mark.asyncio
+    async def test_a_launch_failure_is_502(self, service, launcher, as_user):
+        """502, not 500: the failure is downstream of this API and the task
+        itself is intact."""
+        created = await _create(service)
+        launcher.fail_with = LaunchFailedError("run backend unavailable")
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.trigger_scheduled_task, task_id=created.id, service=service)
+        assert caught.value.status_code == 502
+
+    @pytest.mark.asyncio
+    async def test_triggering_an_unknown_task_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.trigger_scheduled_task, task_id="task-nope", service=service)
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_a_paused_task_can_still_be_triggered(self, service, as_user):
+        """Manual dispatch is deliberately allowed while paused, and leaves the
+        task paused."""
+        created = await _create(service)
+        await _call(router_module.pause_scheduled_task, task_id=created.id, service=service)
+        response = await _call(router_module.trigger_scheduled_task, task_id=created.id, service=service)
+        assert response.triggered is True
+        after = await _call(router_module.get_scheduled_task, task_id=created.id, service=service)
+        assert after.status == "paused"
+
+
+class TestDelete:
+    @pytest.mark.asyncio
+    async def test_a_task_is_deleted(self, service, as_user):
+        created = await _create(service)
+        response = await _call(router_module.delete_scheduled_task, task_id=created.id, service=service)
+        assert response.model_dump() == {"id": created.id, "deleted": True}
+
+    @pytest.mark.asyncio
+    async def test_deleting_an_unknown_task_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.delete_scheduled_task, task_id="task-nope", service=service)
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_deleting_another_users_task_is_404(self, service, as_user):
+        created = await _create(service)
+        as_user(OTHER_USER)
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.delete_scheduled_task, task_id=created.id, service=service)
+        assert caught.value.status_code == 404
+
+
+class TestRunHistory:
+    @pytest.mark.asyncio
+    async def test_runs_are_listed_for_an_owned_task(self, service, as_user):
+        created = await _create(service)
+        await _call(router_module.trigger_scheduled_task, task_id=created.id, service=service)
+        listed = await _call(router_module.list_scheduled_task_runs, task_id=created.id, service=service)
+        assert len(listed) == 1
+        assert listed[0].task_id == created.id
+        assert listed[0].trigger == "manual"
+
+    @pytest.mark.asyncio
+    async def test_history_of_an_unknown_task_is_404(self, service, as_user):
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.list_scheduled_task_runs, task_id="task-nope", service=service)
+        assert caught.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_history_of_another_users_task_is_404(self, service, as_user):
+        """Ownership is checked on the parent task, so run history cannot be
+        read sideways."""
+        created = await _create(service)
+        as_user(OTHER_USER)
+        with pytest.raises(HTTPException) as caught:
+            await _call(router_module.list_scheduled_task_runs, task_id=created.id, service=service)
+        assert caught.value.status_code == 404
+
+
+class TestErrorMapping:
+    def test_every_mapped_error_is_a_schedule_error(self):
+        from deerflow.domain.schedule.model import ScheduleError
+
+        assert all(issubclass(error, ScheduleError) for error in router_module._STATUS_BY_ERROR)
+
+    @pytest.mark.asyncio
+    async def test_an_unclassified_domain_error_is_not_swallowed(self):
+        """A new domain error is a new protocol decision. Letting it through
+        surfaces as a 500 that has to be classified, rather than shipping as
+        whatever 4xx happened to be nearest."""
+        from deerflow.domain.schedule.model import ScheduleError
+
+        class NewDomainError(ScheduleError):
+            pass
+
+        @router_module._map_domain_errors
+        async def handler():
+            raise NewDomainError("something new")
+
+        with pytest.raises(NewDomainError):
+            await handler()
+
+    @pytest.mark.asyncio
+    async def test_a_mapped_error_keeps_its_message_as_the_detail(self):
+        from deerflow.domain.schedule.model import TaskNotFoundError
+
+        @router_module._map_domain_errors
+        async def handler():
+            raise TaskNotFoundError("Scheduled task not found")
+
+        with pytest.raises(HTTPException) as caught:
+            await handler()
+        assert caught.value.detail == "Scheduled task not found"

backend/tests/test_schedule_spec_mapping.py

@@ -1,78 +0,0 @@
-"""Boundary tests for the schedule_spec mapping.
-
-This is where a `Mapping[str, Any]` is allowed to exist. The split it enforces
-is the point: **structural** problems (key missing, wrong type, unknown
-schedule type) are caught here, **value** problems (5-field cron, resolvable
-timezone) are left to `ScheduleSpec.__post_init__` -- and both surface as the
-same domain error, so the router maps one family.
-"""
-
-from __future__ import annotations
-
-from datetime import UTC, datetime
-
-import pytest
-
-from app.adapters.schedule.spec_mapping import spec_to_domain, spec_to_wire
-from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType
-
-
-class TestStructuralChecks:
-    def test_unknown_schedule_type_is_rejected(self):
-        """The one rule the domain cannot state: ScheduleSpec only accepts the
-        enum, so a bad string has to be caught at the boundary."""
-        with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"):
-            spec_to_domain("teleport", {"cron": "0 9 * * *"}, "UTC")
-
-    @pytest.mark.parametrize("spec", [{}, None, {"cron": 5}, {"run_at": "..."}])
-    def test_cron_without_a_string_cron_is_rejected(self, spec):
-        with pytest.raises(InvalidScheduleError, match="requires schedule_spec"):
-            spec_to_domain("cron", spec, "UTC")
-
-    @pytest.mark.parametrize("spec", [{}, None, {"run_at": 5}, {"cron": "0 9 * * *"}])
-    def test_once_without_a_string_run_at_is_rejected(self, spec):
-        with pytest.raises(InvalidScheduleError, match="requires run_at"):
-            spec_to_domain("once", spec, "UTC")
-
-    def test_an_unparseable_run_at_is_rejected(self):
-        with pytest.raises(InvalidScheduleError, match="unparseable run_at"):
-            spec_to_domain("once", {"run_at": "next tuesday"}, "UTC")
-
-
-class TestValueChecksStayInTheDomain:
-    """These are not re-implemented here -- they arrive from __post_init__,
-    which is why the boundary can stay thin."""
-
-    def test_a_bad_cron_expression_still_raises(self):
-        with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
-            spec_to_domain("cron", {"cron": "0 9 * *"}, "UTC")
-
-    def test_a_bad_timezone_still_raises(self):
-        with pytest.raises(InvalidScheduleError, match="Unknown timezone"):
-            spec_to_domain("cron", {"cron": "0 9 * * *"}, "Mars/Olympus_Mons")
-
-
-class TestRoundTrip:
-    def test_cron_round_trips_normalized(self):
-        spec = spec_to_domain("cron", {"cron": "  0  9 * * * "}, "Asia/Shanghai")
-        assert spec.schedule_type is ScheduleType.CRON
-        assert spec_to_wire(spec) == {"cron": "0 9 * * *"}
-        assert spec_to_domain("cron", spec_to_wire(spec), "Asia/Shanghai") == spec
-
-    def test_once_round_trips_to_the_same_instant(self):
-        spec = spec_to_domain("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai")
-        assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
-        assert spec_to_domain("once", spec_to_wire(spec), "Asia/Shanghai") == spec
-
-    def test_a_trailing_z_input_parses_and_re_emits_as_an_offset(self):
-        """The shape the frontend actually submits (`zonedLocalToUtcIso`).
-        Both spellings parse on either side, so normalizing is deliberate."""
-        spec = spec_to_domain("once", {"run_at": "2026-08-01T01:00:00Z"}, "Asia/Shanghai")
-        emitted = spec_to_wire(spec)["run_at"]
-        assert emitted.endswith("+00:00")
-        assert spec_to_domain("once", {"run_at": emitted}, "Asia/Shanghai") == spec
-
-    def test_wire_output_is_idempotent(self):
-        spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC")
-        once = spec_to_wire(spec)
-        assert spec_to_wire(spec_to_domain("once", once, "UTC")) == once

backend/tests/test_schedule_spec_parity.py

@@ -0,0 +1,162 @@
+"""Boundary tests for the two schedule_spec mappings, run against both.
+
+`schedule_spec` crosses two boundaries -- an HTTP body field and a JSON column
+-- and each side owns its own translation, because a primary adapter must not
+import a secondary one. The shapes are equal today only by coincidence.
+
+Coincidence is exactly what needs a test. Every case here runs against both
+implementations, so the day one side is changed without the other, this file
+fails instead of production accepting a spec the repository cannot read back.
+That is the same N-cases-x-2-implementations shape `test_schedule_fakes.py`
+uses for the repository ports.
+
+The split each side enforces is the point: **structural** problems (key
+missing, wrong type, unknown schedule type) are caught at the boundary,
+**value** problems (5-field cron, resolvable timezone) are left to
+`ScheduleSpec.__post_init__` -- and both surface as the same domain error, so
+the router maps one family.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+
+from app.adapters.schedule.spec_column import column_to_spec, spec_to_column
+from app.gateway.routers.schedule.spec_wire import spec_to_wire, wire_to_spec
+from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType
+
+
+@pytest.fixture(
+    params=[
+        pytest.param((column_to_spec, spec_to_column), id="column"),
+        pytest.param((wire_to_spec, spec_to_wire), id="wire"),
+    ]
+)
+def mapping(request):
+    """The (parse, emit) pair under test, once per boundary."""
+    return request.param
+
+
+@pytest.fixture
+def parse(mapping):
+    return mapping[0]
+
+
+@pytest.fixture
+def emit(mapping):
+    return mapping[1]
+
+
+class TestStructuralChecks:
+    def test_unknown_schedule_type_is_rejected(self, parse):
+        """The one rule the domain cannot state: ScheduleSpec only accepts the
+        enum, so a bad string has to be caught at the boundary."""
+        with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"):
+            parse("teleport", {"cron": "0 9 * * *"}, "UTC")
+
+    @pytest.mark.parametrize("spec", [{}, None, {"cron": 5}, {"run_at": "..."}])
+    def test_cron_without_a_string_cron_is_rejected(self, parse, spec):
+        with pytest.raises(InvalidScheduleError, match="requires schedule_spec"):
+            parse("cron", spec, "UTC")
+
+    @pytest.mark.parametrize("spec", [{}, None, {"run_at": 5}, {"cron": "0 9 * * *"}])
+    def test_once_without_a_string_run_at_is_rejected(self, parse, spec):
+        with pytest.raises(InvalidScheduleError, match="requires run_at"):
+            parse("once", spec, "UTC")
+
+    def test_an_unparseable_run_at_is_rejected(self, parse):
+        with pytest.raises(InvalidScheduleError, match="unparseable run_at"):
+            parse("once", {"run_at": "next tuesday"}, "UTC")
+
+
+class TestValueChecksStayInTheDomain:
+    """These are not re-implemented on either side -- they arrive from
+    __post_init__, which is why each boundary can stay thin and why the
+    duplication between them is bounded."""
+
+    def test_a_bad_cron_expression_still_raises(self, parse):
+        with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
+            parse("cron", {"cron": "0 9 * *"}, "UTC")
+
+    def test_a_bad_timezone_still_raises(self, parse):
+        with pytest.raises(InvalidScheduleError, match="Unknown timezone"):
+            parse("cron", {"cron": "0 9 * * *"}, "Mars/Olympus_Mons")
+
+
+class TestRoundTrip:
+    def test_cron_round_trips_normalized(self, parse, emit):
+        spec = parse("cron", {"cron": "  0  9 * * * "}, "Asia/Shanghai")
+        assert spec.schedule_type is ScheduleType.CRON
+        assert emit(spec) == {"cron": "0 9 * * *"}
+        assert parse("cron", emit(spec), "Asia/Shanghai") == spec
+
+    def test_once_round_trips_to_the_same_instant(self, parse, emit):
+        spec = parse("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai")
+        assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
+        assert parse("once", emit(spec), "Asia/Shanghai") == spec
+
+    def test_a_trailing_z_input_parses_and_re_emits_as_an_offset(self, parse, emit):
+        """The shape the frontend actually submits (`zonedLocalToUtcIso`).
+        Both spellings parse on either side, so normalizing is deliberate."""
+        spec = parse("once", {"run_at": "2026-08-01T01:00:00Z"}, "Asia/Shanghai")
+        emitted = emit(spec)["run_at"]
+        assert emitted.endswith("+00:00")
+        assert parse("once", {"run_at": emitted}, "Asia/Shanghai") == spec
+
+    def test_emitted_output_is_idempotent(self, parse, emit):
+        spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC")
+        once = emit(spec)
+        assert emit(parse("once", once, "UTC")) == once
+
+
+class TestCrossBoundaryParity:
+    """The two sides must agree, not merely each be self-consistent.
+
+    The cases above run against both and would catch a behavioural drift;
+    these compare the outputs directly, which is what catches a *silent* one --
+    a side that starts emitting a different but still-self-consistent shape.
+    """
+
+    @pytest.mark.parametrize(
+        ("schedule_type", "spec", "timezone"),
+        [
+            ("cron", {"cron": "0 9 * * *"}, "UTC"),
+            ("cron", {"cron": "  30 2 * * 1 "}, "Asia/Shanghai"),
+            ("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai"),
+            ("once", {"run_at": "2026-08-01T01:00:00Z"}, "UTC"),
+        ],
+    )
+    def test_both_sides_parse_to_the_same_value_object(self, schedule_type, spec, timezone):
+        assert column_to_spec(schedule_type, spec, timezone) == wire_to_spec(schedule_type, spec, timezone)
+
+    @pytest.mark.parametrize(
+        "spec",
+        [
+            ScheduleSpec.cron_schedule("0 9 * * *", "UTC"),
+            ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "Asia/Shanghai"),
+        ],
+    )
+    def test_both_sides_emit_the_same_shape(self, spec):
+        assert spec_to_column(spec) == spec_to_wire(spec)
+
+    @pytest.mark.parametrize(
+        ("schedule_type", "spec"),
+        [
+            ("teleport", {"cron": "0 9 * * *"}),
+            ("cron", {}),
+            ("cron", {"cron": 5}),
+            ("once", {}),
+            ("once", {"run_at": 5}),
+            ("once", {"run_at": "next tuesday"}),
+        ],
+    )
+    def test_both_sides_reject_the_same_inputs_with_the_same_message(self, schedule_type, spec):
+        """Same message, not just same type: the router turns this text into a
+        422 detail, so a divergence here is user-visible."""
+        with pytest.raises(InvalidScheduleError) as from_column:
+            column_to_spec(schedule_type, spec, "UTC")
+        with pytest.raises(InvalidScheduleError) as from_wire:
+            wire_to_spec(schedule_type, spec, "UTC")
+        assert str(from_column.value) == str(from_wire.value)