diff --git a/backend/app/adapters/schedule/scheduled_task_repository.py b/backend/app/adapters/schedule/scheduled_task_repository.py
index c0e277e58..a7e297387 100644
--- a/backend/app/adapters/schedule/scheduled_task_repository.py
+++ b/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
-stored JSON spec.
+Sibling of `scheduled_run_repository.py`. The stored `schedule_spec` JSON
+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
diff --git a/backend/app/adapters/schedule/spec_mapping.py b/backend/app/adapters/schedule/spec_column.py
similarity index 61%
rename from backend/app/adapters/schedule/spec_mapping.py
rename to backend/app/adapters/schedule/spec_column.py
index 81eb7a36e..d0fc32c74 100644
--- a/backend/app/adapters/schedule/spec_mapping.py
+++ b/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
-field and a JSON column -- one shape, two boundaries -- so the mapping lives
-here once and both callers import it (`scheduled_task_repository` for the
-stored column, the scheduled-task router for the request body), rather than the
-domain growing a `Mapping[str, Any]` in its signatures.
+translation `scheduled_task_repository` needs between the stored column and the
+value object, so the domain never grows a `Mapping[str, Any]` in its
+signatures.
-The split is deliberate: **structural** checks (is the key present? is it a
-str?) belong to this boundary, **value** rules (5-field cron, resolvable
-timezone, run_at present) belong to `ScheduleSpec.__post_init__`. That is why
-this module can look thin -- most of what could go wrong is caught one layer
-in, and reported with the same domain error.
+Its counterpart on the other side of the application is
+`app/gateway/routers/schedule/spec_wire.py`, which does the same job for the
+HTTP request/response body. The two are near-identical today and are still kept
+apart on purpose: a primary adapter must not import a secondary one, and the
+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
diff --git a/backend/app/composition.py b/backend/app/composition.py
new file mode 100644
index 000000000..c2a95cf74
--- /dev/null
+++ b/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,
+ )
diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py
index 637bdbbe6..41ca43228 100644
--- a/backend/app/gateway/app.py
+++ b/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
- from app.scheduler import ScheduledTaskService
+ # The service itself was assembled by the composition root inside
+ # `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:
- scheduled_task_service = ScheduledTaskService(
- task_repo=app.state.scheduled_task_repo,
- task_run_repo=app.state.scheduled_task_run_repo,
- launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs),
+ schedule_service = getattr(app.state, "schedule_service", None)
+ if schedule_service is not None:
+ schedule_poller = SchedulePoller(
+ schedule_service,
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)
diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py
index 3f758203a..067554a08 100644
--- a/backend/app/gateway/deps.py
+++ b/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),
)
diff --git a/backend/app/gateway/routers/schedule/__init__.py b/backend/app/gateway/routers/schedule/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/backend/app/gateway/routers/schedule/models.py b/backend/app/gateway/routers/schedule/models.py
new file mode 100644
index 000000000..51da2a13d
--- /dev/null
+++ b/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
diff --git a/backend/app/gateway/routers/schedule/router.py b/backend/app/gateway/routers/schedule/router.py
new file mode 100644
index 000000000..badf99401
--- /dev/null
+++ b/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]
diff --git a/backend/app/gateway/routers/schedule/spec_wire.py b/backend/app/gateway/routers/schedule/spec_wire.py
new file mode 100644
index 000000000..ca891b483
--- /dev/null
+++ b/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 ""}
diff --git a/backend/docs/SCHEDULE_DESIGN_zh.md b/backend/docs/SCHEDULE_DESIGN_zh.md
index eb2e5e32d..1a3918a7f 100644
--- a/backend/docs/SCHEDULE_DESIGN_zh.md
+++ b/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
ScheduleService（用例编排）"]
PO["ports.py
4 个 Protocol + 2 个 DTO"]
@@ -54,29 +54,37 @@ flowchart LR
SV --> PO
SV --> M
end
- subgraph OLD["⏳ 仍是旧形态 · 生产路径"]
- R["gateway/routers/scheduled_tasks.py
入参校验 + 业务判断混在一起"]
- S["app/scheduler/service.py
轮询 + 派发编排 + 状态推导"]
- P["persistence/scheduled_task*/sql.py
仓储返回裸 dict"]
- C["deerflow/scheduler/schedules.py
时区 / cron 计算"]
+ subgraph PRIM["✅ 主适配器 · app"]
+ R["gateway/routers/schedule/
router · models · spec_wire"]
+ PL["scheduler/poller.py
轮询时钟"]
end
- subgraph TODO["🚧 待建 · 外圈"]
- AD["app/adapters/schedule/
app/scheduler/poller.py"]
+ subgraph SEC["✅ 从适配器 · app"]
+ AD["adapters/schedule/
两个仓储 · run_launcher · thread_lookup
spec_column · run_outcome_mapping"]
+ end
+ subgraph CR["✅ 组合根"]
+ CO["composition.py
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
- S -.->|尚未调用| SV
- AD -.->|将实现| PO
+ R --> SV
+ PL --> SV
+ AD -.->|实现| PO
+ CO -->|装配| SV
- style NEW fill:#eef6ff
- style OLD fill:#f5f5f5
+ style IN fill:#eef6ff
+ style DEAD fill:#f5f5f5
```
含义很具体：
-- **`domain/schedule/` 是唯一的真相声明处**，但目前只有测试在用它。运行中的定时任务走的仍是 `app/scheduler/service.py` 那套。
-- 两边**规则内容一致**（内圈是逐条从旧代码搬迁的，每个方法的 docstring 都标了来源行号），但**代码是重复的**。这是迁移中间态的正常代价。
-- 你现在改一条业务规则，要**两边都改**，直到迁移完成。旧位置见 §12 的索引。
+- **`domain/schedule/` 是唯一的真相声明处**，运行中的定时任务已经走它。改一条业务规则只改这一处。
+- `DEAD` 里的文件仍能编译、仍有测试，但**不再被任何装配路径引用**——`app.py` 启动的是 `SchedulePoller`，注册的是 `routers/schedule/`，完成回调走 `composition.py` 装的那个。留着只是为了让删除单独成为一个可审查的提交。
+- 旧的重复规则因此不再需要"两边都改"。如果你在 `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 端点，含入参校验与业务判断 |
-| [`app/scheduler/service.py`](../app/scheduler/service.py) | 轮询循环、派发编排、状态推导、完成回调、启动清扫 |
-| [`deerflow/scheduler/schedules.py`](../packages/harness/deerflow/scheduler/schedules.py) | 时区 / cron / 下次时间计算 |
-| [`persistence/scheduled_tasks/`](../packages/harness/deerflow/persistence/scheduled_tasks/) | 任务表 ORM + 仓储 |
-| [`persistence/scheduled_task_runs/`](../packages/harness/deerflow/persistence/scheduled_task_runs/) | 执行表 ORM + 仓储，含唯一索引定义 |
+| [`gateway/routers/schedule/router.py`](../app/gateway/routers/schedule/router.py) | 10 个 HTTP 端点，只做协议转换 + 领域错误→状态码 |
+| [`gateway/routers/schedule/models.py`](../app/gateway/routers/schedule/models.py) | 请求/响应模型；响应是白名单，不是 ORM 转储 |
+| [`gateway/routers/schedule/spec_wire.py`](../app/gateway/routers/schedule/spec_wire.py) | HTTP body ↔ `ScheduleSpec` |
+| [`app/scheduler/poller.py`](../app/scheduler/poller.py) | 轮询时钟 + 启动恢复 |
+
+**从适配器**
+
+| 文件 | 内容 |
+|---|---|
+| [`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`。
diff --git a/backend/tests/test_composition.py b/backend/tests/test_composition.py
new file mode 100644
index 000000000..554e11e49
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_gateway_run_drain_shutdown.py b/backend/tests/test_gateway_run_drain_shutdown.py
index fd5aa975d..206e72945 100644
--- a/backend/tests/test_gateway_run_drain_shutdown.py
+++ b/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
diff --git a/backend/tests/test_gateway_run_recovery.py b/backend/tests/test_gateway_run_recovery.py
index f7e17b3d6..01fa269fb 100644
--- a/backend/tests/test_gateway_run_recovery.py
+++ b/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"})
diff --git a/backend/tests/test_schedule_response_models.py b/backend/tests/test_schedule_response_models.py
new file mode 100644
index 000000000..a900a5f33
--- /dev/null
+++ b/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"
diff --git a/backend/tests/test_schedule_router.py b/backend/tests/test_schedule_router.py
new file mode 100644
index 000000000..36611c86a
--- /dev/null
+++ b/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"
diff --git a/backend/tests/test_schedule_spec_mapping.py b/backend/tests/test_schedule_spec_mapping.py
deleted file mode 100644
index 11015ed18..000000000
--- a/backend/tests/test_schedule_spec_mapping.py
+++ /dev/null
@@ -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
diff --git a/backend/tests/test_schedule_spec_parity.py b/backend/tests/test_schedule_spec_parity.py
new file mode 100644
index 000000000..bcc9b269f
--- /dev/null
+++ b/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)