github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-01 19:06:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(schedule): add the domain model, ports, and application service

Browse Source 
The inner ring of the schedule slice, added on its own so it can be read
as domain modelling rather than as a diff against the old implementation:
two aggregates with their state machines, the policy value object, the
output ports the service depends on, and the errors it raises.

Nothing wires it up yet -- no existing code path changes. The service is
exercised end to end against in-memory fakes, which is what makes the
rules (overlap policy, lease handling, which write owns which timestamp)
assertable without a database at all.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
rayhpeng 2026-07-31 11:43:35 +08:00
parent d9c7d1ccab
commit d856ae8573
13 changed files with 3621 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
backend/packages/harness/deerflow/domain/schedule/__init__.py Normal file
View File

@ -0,0 +1,83 @@
"""Schedule bounded context: standing instructions to run a prompt on time.
Public API of the context. Import domain objects, commands, errors, and the
service from here; import ports from `deerflow.domain.schedule.ports` -- they
are contracts consumed by adapters and tests, not everyday call-site symbols.
"""
from deerflow.domain.schedule.commands import (
UNSET,
ContextChange,
CreateScheduledTask,
DeleteTask,
PauseTask,
ResumeTask,
TriggerTask,
UnsetType,
UpdateScheduledTask,
)
from deerflow.domain.schedule.exceptions import (
ActiveRunConflictError,
CorruptStoredScheduleError,
InvalidContextModeError,
InvalidScheduleError,
LaunchFailedError,
ScheduleError,
TaskNotFoundError,
TaskNotMutableError,
ThreadBusyError,
ThreadNotFoundError,
)
from deerflow.domain.schedule.model import (
ACTIVE_RUN_STATUSES,
TERMINAL_RUN_STATUSES,
TERMINAL_TASK_STATUSES,
ContextMode,
DispatchOutcome,
RunStatus,
ScheduledRun,
ScheduledTask,
SchedulePolicy,
ScheduleSpec,
ScheduleType,
TaskStatus,
TriggerKind,
)
from deerflow.domain.schedule.service import DispatchResult, ScheduleService
__all__ = [
"ACTIVE_RUN_STATUSES",
"TERMINAL_RUN_STATUSES",
"TERMINAL_TASK_STATUSES",
"UNSET",
"ActiveRunConflictError",
"ContextChange",
"ContextMode",
"CorruptStoredScheduleError",
"CreateScheduledTask",
"DeleteTask",
"DispatchOutcome",
"DispatchResult",
"InvalidContextModeError",
"InvalidScheduleError",
"LaunchFailedError",
"PauseTask",
"ResumeTask",
"RunStatus",
"ScheduleError",
"SchedulePolicy",
"ScheduleService",
"ScheduleSpec",
"ScheduleType",
"ScheduledRun",
"ScheduledTask",
"TaskNotFoundError",
"TaskNotMutableError",
"TaskStatus",
"ThreadBusyError",
"ThreadNotFoundError",
"TriggerKind",
"TriggerTask",
"UnsetType",
"UpdateScheduledTask",
]

130
backend/packages/harness/deerflow/domain/schedule/commands.py Normal file
View File

@ -0,0 +1,130 @@
"""Commands of the schedule context.
One frozen dataclass per HTTP-driven state-changing use case -- the named
carrier of "the information required to perform an operation on the domain"
(AWS hexagonal guidance). Commands are dumb data on purpose: business
validation stays on the aggregate (``ScheduledTask``'s factory and
transitions), and structural validation stays on the primary adapter's api
model, so error attribution (a malformed schedule reported before an unknown
thread) is owned by the handler's construction order.
Two groups of use cases are deliberately NOT commands:
- **Queries** (``list_tasks``, ``get_task``, ``list_task_runs``, ...) keep
plain parameters -- a command expresses an intent to change state, and
wrapping reads would be pure boilerplate.
- **Clock- and callback-driven writes** (``run_once``, ``dispatch_task``,
``handle_run_completion``, ``reconcile_on_startup``). Those drivers have no
wire shape to translate (spec §5.3): the poller hands the service a claimed
aggregate and a clock reading, and the completion hook hands it an already
domain-typed ``RunOutcome``. A command would re-wrap domain vocabulary in
more domain vocabulary.
``now`` is not a command field: it is the server's clock reading, passed
explicitly to the handler (``now=``) like every other rule input, not part of
the client's intent.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Final, final
if TYPE_CHECKING:
from deerflow.domain.schedule.model import ContextMode, ScheduleSpec
@final
class UnsetType:
"""The type of ``UNSET`` -- "the client did not supply this field".
Partial updates need three states (absent, null, value). ``None`` cannot
carry two of them, so absence gets its own singleton; a field that is
``UNSET`` is left untouched by the handler.
"""
_instance: UnsetType | None = None
def __new__(cls) -> UnsetType:
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __repr__(self) -> str:
return "UNSET"
def __bool__(self) -> bool:
return False
UNSET: Final[UnsetType] = UnsetType()
@dataclass(frozen=True)
class ContextChange:
"""A requested change of execution context.
The mode and the thread always move together -- ``with_context`` takes
both, and clearing the thread is what switching to a fresh-thread mode
means. Packaging them keeps ``None`` unambiguous everywhere else: the one
field for which ``None`` is a meaningful value travels inside this object.
"""
context_mode: str | ContextMode
thread_id: str | None = None
@dataclass(frozen=True)
class CreateScheduledTask:
"""Register a new standing instruction to run a prompt on time."""
user_id: str
title: str
prompt: str
schedule: ScheduleSpec
context_mode: str | ContextMode
thread_id: str | None = None
@dataclass(frozen=True)
class UpdateScheduledTask:
"""Partially update a task; ``UNSET`` means "not supplied"."""
task_id: str
user_id: str
title: str | UnsetType = UNSET
prompt: str | UnsetType = UNSET
schedule: ScheduleSpec | UnsetType = UNSET
context: ContextChange | UnsetType = UNSET
@dataclass(frozen=True)
class PauseTask:
"""Stop claiming this task until it is resumed."""
task_id: str
user_id: str
@dataclass(frozen=True)
class ResumeTask:
"""Re-admit this task to claiming."""
task_id: str
user_id: str
@dataclass(frozen=True)
class DeleteTask:
"""Remove the task; its execution history rows go with it."""
task_id: str
user_id: str
@dataclass(frozen=True)
class TriggerTask:
"""Dispatch a task on demand, even while it is paused."""
task_id: str
user_id: str

67
backend/packages/harness/deerflow/domain/schedule/exceptions.py Normal file
View File

@ -0,0 +1,67 @@
"""The known errors of the schedule context.
One family under one base class, so the primary adapter can map the whole
family onto protocol codes in a single table. Class names keep the PEP 8
``Error`` suffix; the module is named ``exceptions`` after the AWS
hexagonal guidance's domain folder of the same name.
"""
from __future__ import annotations
class ScheduleError(Exception):
"""Base error for the schedule domain."""
class InvalidScheduleError(ScheduleError):
"""Timezone, cron expression, or run_at is not usable."""
class InvalidContextModeError(ScheduleError):
"""context_mode is unknown, or reuse_thread is missing its thread_id."""
class TaskNotFoundError(ScheduleError):
"""The task does not exist or does not belong to the user."""
class TaskNotMutableError(ScheduleError):
"""The task is currently running and cannot be edited."""
class ThreadNotFoundError(ScheduleError):
"""reuse_thread points at a thread the user cannot access."""
class CorruptStoredScheduleError(ScheduleError):
"""A stored task row can no longer be rebuilt into a valid aggregate.
Raised by the persistence adapter, never by the aggregate: it means the
*storage* is damaged, not that a client submitted something invalid --
which is why it is deliberately absent from the router's status table and
falls through to the unclassified-500 branch instead of riding
``InvalidScheduleError``'s 422.
"""
class ActiveRunConflictError(ScheduleError):
"""The task already holds its single active run slot.
Raised by the run repository when the partial unique index
``uq_scheduled_task_run_active`` rejects a second active row. Moved here
from ``persistence/scheduled_task_runs/sql.py`` (was
``ActiveScheduledRunConflict``) so the domain owns its own vocabulary.
"""
class ThreadBusyError(ScheduleError):
"""The execution thread already has an in-flight run.
Translated by the RunLauncher adapter from ConflictError / HTTP 409.
This is what removes `from fastapi import HTTPException` from the
orchestration layer.
"""
class LaunchFailedError(ScheduleError):
"""The run could not be launched for any non-conflict reason."""

51
backend/packages/harness/deerflow/domain/schedule/model/__init__.py Normal file
View File

@ -0,0 +1,51 @@
"""Domain model of the schedule context.
A package rather than a single module because this context has two
aggregates plus a value object, which a single `model.py` could not hold
without becoming the file nobody reads. Re-exports below make the split
invisible to callers: `deerflow.domain.schedule.model` exposes exactly
what a single `model.py` would have.
The re-exports below are ordered alphabetically because ruff's isort rule
owns that block; alphabetical happens to satisfy the real dependency order
too (enums depend on nothing, run depends on enums, spec depends on enums,
task depends on both), so it needs no override. The domain errors are not
re-exported here: they live one level up in
`deerflow.domain.schedule.exceptions`, a sibling of the model per the AWS
domain layout, and are imported from there.
What actually keeps the package acyclic is a rule isort cannot express:
**a submodule imports its siblings directly, never this package.** Reaching
back through `deerflow.domain.schedule.model` makes a submodule depend on a
partially initialized package — it only works while the symbol happens to
sit above that submodule's own line here, and reordering this block silently
breaks it. Keep it that way.
"""
from deerflow.domain.schedule.model.enums import (
ContextMode,
DispatchOutcome,
RunStatus,
ScheduleType,
TaskStatus,
TriggerKind,
)
from deerflow.domain.schedule.model.run import ACTIVE_RUN_STATUSES, TERMINAL_RUN_STATUSES, ScheduledRun
from deerflow.domain.schedule.model.spec import SchedulePolicy, ScheduleSpec
from deerflow.domain.schedule.model.task import TERMINAL_TASK_STATUSES, ScheduledTask
__all__ = [
"ACTIVE_RUN_STATUSES",
"TERMINAL_RUN_STATUSES",
"TERMINAL_TASK_STATUSES",
"ContextMode",
"DispatchOutcome",
"RunStatus",
"SchedulePolicy",
"ScheduleSpec",
"ScheduleType",
"ScheduledRun",
"ScheduledTask",
"TaskStatus",
"TriggerKind",
]

94
backend/packages/harness/deerflow/domain/schedule/model/enums.py Normal file
View File

@ -0,0 +1,94 @@
from __future__ import annotations
from enum import StrEnum
class TaskStatus(StrEnum):
"""Lifecycle of a standing instruction.
RUNNING does not mean "the agent is executing" -- it means this dispatch
round's scheduling ownership is held. That is why `ensure_mutable` refuses
edits in that state, and why a crash can strand a task there.
The last three are terminal for the *current* schedule, not forever: moving
the schedule into the future re-arms them (see TERMINAL_TASK_STATUSES).
"""
ENABLED = "enabled"
PAUSED = "paused"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class ContextMode(StrEnum):
"""Which thread a dispatch executes in.
REUSE_THREAD lets consecutive executions see each other's history and
requires a thread up front; the default gives each one a fresh thread and
therefore no shared context.
"""
FRESH_THREAD_PER_RUN = "fresh_thread_per_run"
REUSE_THREAD = "reuse_thread"
class ScheduleType(StrEnum):
"""How the next fire time is derived.
The two branch almost every rule in this context: CRON always has a next
occurrence, ONCE is consumed by its first one.
"""
ONCE = "once"
CRON = "cron"
class RunStatus(StrEnum):
"""Lifecycle of one execution record.
QUEUED and RUNNING are the two that occupy a task's single active slot;
everything else is terminal.
SKIPPED is created directly terminal and never passes through QUEUED -- a
queued tombstone would collide with the very run it is recording the
overlap with (see ScheduledRun.skipped_tombstone). INTERRUPTED is kept
distinct from FAILED because a cancel or takeover is not an execution
failure, and the two lead a `once` task to different terminal states.
"""
QUEUED = "queued"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
INTERRUPTED = "interrupted"
SKIPPED = "skipped"
class DispatchOutcome(StrEnum):
"""What one dispatch attempt produced.
A domain vocabulary rather than a set of strings: the caller branches on
all four, and the distinction between SKIPPED and CONFLICT is itself a
business rule -- a dropped scheduled occurrence is accounted for, while a
rejected on-demand trigger is reported and leaves no trace.
"""
LAUNCHED = "launched"
SKIPPED = "skipped"
CONFLICT = "conflict"
FAILED = "failed"
class TriggerKind(StrEnum):
"""What caused a dispatch.
The two kinds diverge in almost every decision the domain makes — how an
overlap is handled, what status survives a failed launch, whether a paused
task stays paused — so this is a first-class concept rather than the raw
string the old service compared in four places.
"""
SCHEDULED = "scheduled"
MANUAL = "manual"

97
backend/packages/harness/deerflow/domain/schedule/model/run.py Normal file
View File

@ -0,0 +1,97 @@
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from deerflow.domain.schedule.model.enums import RunStatus
ACTIVE_RUN_STATUSES: tuple[RunStatus, ...] = (RunStatus.QUEUED, RunStatus.RUNNING)
"""The statuses that occupy a task's single active-run slot.
Must stay in lockstep with the predicate of the partial unique index
``uq_scheduled_task_run_active`` (``status IN ('queued','running')``, declared
in ``persistence/scheduled_task_runs/model.py``). Drift here silently
decouples the overlap fast path from its atomic arbiter — the consistency
assertion lives in a separate test module rather than the domain tests, which
stay dependency-free.
"""
TERMINAL_RUN_STATUSES: frozenset[RunStatus] = frozenset(
{
RunStatus.SUCCESS,
RunStatus.FAILED,
RunStatus.SKIPPED,
RunStatus.INTERRUPTED,
}
)
"""Statuses a run row can no longer leave.
Used by the repository's ``protect_terminal`` compare-and-set: a fast-failing
run can reach the completion hook before the launch path's own write lands.
"""
@dataclass(frozen=True)
class ScheduledRun:
"""One execution record of a scheduled task — the history row.
A separate aggregate from ``ScheduledTask``: the two are written in
independent transactions and reference each other only by ``task_id``.
"""
record_id: str
task_id: str
thread_id: str
scheduled_for: datetime
trigger: str
status: RunStatus
run_id: str | None = None
error: str | None = None
started_at: datetime | None = None
finished_at: datetime | None = None
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
@classmethod
def queued(cls, *, task_id: str, thread_id: str, scheduled_for: datetime, trigger: str) -> ScheduledRun:
"""The active row of a normal dispatch.
Inserting this is what the unique index arbitrates: the loser of a
concurrent insert surfaces as ``ActiveRunConflictError``.
The ``task-run-{hex}`` id shape is depended on by existing rows and by
the run metadata that links a Gateway run back to this record — do not
change it.
"""
return cls(
record_id=f"task-run-{uuid.uuid4().hex}",
task_id=task_id,
thread_id=thread_id,
scheduled_for=scheduled_for,
trigger=trigger,
status=RunStatus.QUEUED,
)
@classmethod
def skipped_tombstone(cls, *, task_id: str, thread_id: str, scheduled_for: datetime, trigger: str) -> ScheduledRun:
"""A dropped occurrence, created directly as terminal ``SKIPPED``.
Deliberately a second factory rather than ``queued()`` followed by a
status change: ``QUEUED`` falls inside ``uq_scheduled_task_run_active``'s
predicate and would collide with the pre-existing run that still holds
the task's single active slot. ``SKIPPED`` is outside the predicate and
can never conflict (service.py:249-256).
"""
return cls(
record_id=f"task-run-{uuid.uuid4().hex}",
task_id=task_id,
thread_id=thread_id,
scheduled_for=scheduled_for,
trigger=trigger,
status=RunStatus.SKIPPED,
)
@property
def is_active(self) -> bool:
"""Whether this row occupies the task's single active-run slot."""
return self.status in ACTIVE_RUN_STATUSES

191
backend/packages/harness/deerflow/domain/schedule/model/spec.py Normal file
View File

@ -0,0 +1,191 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from croniter import croniter
from deerflow.domain.schedule.exceptions import InvalidScheduleError
from deerflow.domain.schedule.model.enums import ScheduleType
CRON_FIELD_COUNT = 5
@dataclass(frozen=True)
class SchedulePolicy:
"""Operator-tunable thresholds the domain needs but must not read itself.
Built by the composition root from the scheduler configuration and passed
in. Deliberately not held by any aggregate: a task whose meaning changes
with deployment config is not a domain object.
The defaults are the permissive ones on purpose -- "nobody configured a
policy" must not invent a business constraint. Real values only ever
arrive from the outer ring.
"""
min_once_delay_seconds: int = 0
"""How far ahead a one-shot schedule must be at submission time. Read by
`ScheduleSpec.ensure_launchable`; a cron schedule is never subject to it."""
max_concurrent_runs: int = 1
"""Ceiling on active scheduled executions across ALL tasks. Long runs
accumulate across polls, so each poll may only claim into what is left."""
lease_seconds: int = 60
"""How long a claim on a task stays valid. Bounds how quickly a task
orphaned between claim and dispatch becomes reachable again."""
@dataclass(frozen=True)
class ScheduleSpec:
"""Parsed, validated view of (schedule_type, schedule_spec, timezone).
The stored JSON spec is mapped in and out by the adapter layer, never here:
a `Mapping[str, Any]` in a domain signature would mean the domain is
handling a persistence/transport format. The two halves of that parsing
split cleanly — 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 __post_init__ below. Storage keeps the same raw JSON, so
this needs no migration.
Normalization happens in __post_init__ rather than in the factories below,
so direct construction cannot bypass it: a frozen dataclass is still
constructible field-by-field, and "valid on construction" has to hold for
that path too.
"""
schedule_type: ScheduleType
timezone: str
cron: str | None = None
run_at: datetime | None = None
def __post_init__(self) -> None:
# The timezone is checked first because normalizing a naive run_at
# below needs it to already be known-good.
try:
zone = ZoneInfo(self.timezone)
except ZoneInfoNotFoundError as exc:
raise InvalidScheduleError(f"Unknown timezone: {self.timezone}") from exc
if self.schedule_type is ScheduleType.CRON:
if not self.cron:
raise InvalidScheduleError("cron schedule requires schedule_spec.cron")
fields = [part for part in self.cron.split() if part]
if len(fields) != CRON_FIELD_COUNT:
raise InvalidScheduleError(f"Cron expression must contain exactly {CRON_FIELD_COUNT} fields")
object.__setattr__(self, "cron", " ".join(fields))
if self.schedule_type is ScheduleType.ONCE:
if self.run_at is None:
raise InvalidScheduleError("once schedule requires run_at")
if self.run_at.tzinfo is None:
# A naive run_at means wall-clock time in this schedule's own
# timezone (schedules.py:40-43). Localizing here rather than at
# every read site keeps the rest of this class tz-aware only.
object.__setattr__(self, "run_at", self.run_at.replace(tzinfo=zone))
@classmethod
def cron_schedule(cls, expr: str, timezone: str) -> ScheduleSpec:
"""Readability sugar — all validation lives in __post_init__."""
return cls(ScheduleType.CRON, timezone, cron=expr)
@classmethod
def once_at(cls, run_at: datetime, timezone: str) -> ScheduleSpec:
"""Readability sugar — all validation lives in __post_init__."""
return cls(ScheduleType.ONCE, timezone, run_at=run_at)
@classmethod
def from_primitives(cls, schedule_type: str, *, cron: str | None, run_at: str | None, timezone: str) -> ScheduleSpec:
"""Build from the loose strings both boundaries arrive as.
The HTTP body and the stored JSON column both carry a schedule type
plus a mapping, so the rule for turning that into a value object lives
here instead of being written out once per adapter. Each adapter keeps
only what is genuinely its own — which keys its own format uses — and
the errors below stay one family the router maps uniformly.
Four strings rather than a `Mapping[str, Any]` on purpose: a mapping in
this signature would mean the domain is handling a transport or storage
format, which is the thing this class exists to avoid.
`cron` and `run_at` are annotated as the contract expects them but
checked rather than trusted — both callers read from data a client can
influence, so a non-string must be rejected here instead of reaching
`fromisoformat` or the cron parser. The field the given type does not
use is ignored: a boundary that carries both keys is not an error.
Raises:
InvalidScheduleError: unknown schedule type, the type's field
missing or not a string, or an unparseable `run_at`. Value
rules — 5-field cron, resolvable timezone — are left to
__post_init__ and surface as the same error.
"""
try:
kind = ScheduleType(schedule_type)
except ValueError as exc:
raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc
if kind is ScheduleType.CRON:
if not isinstance(cron, str):
raise InvalidScheduleError("cron schedule requires schedule_spec.cron")
return cls.cron_schedule(cron, timezone)
if not isinstance(run_at, str):
raise InvalidScheduleError("once schedule requires run_at")
try:
parsed = datetime.fromisoformat(run_at)
except ValueError as exc:
raise InvalidScheduleError(f"once schedule has an unparseable run_at: {run_at!r}") from exc
return cls.once_at(parsed, timezone)
def next_after(self, now: datetime) -> datetime | None:
"""Next fire time in UTC, or None when there is no future occurrence.
The dispatch-path calculation (was `next_run_at` in schedules.py:24-55).
It applies no submission-time policy — see ensure_launchable for that,
and do not swap the two: re-arming a cron task through the stricter one
would reject it right after a perfectly normal launch.
ONCE returns run_at while it is still ahead of `now`, else None (the
single occurrence is in the past). CRON is evaluated in this schedule's
timezone and returned as UTC. A naive `now` is read as UTC
(schedules.py:32-33).
"""
if now.tzinfo is None:
now = now.replace(tzinfo=UTC)
if self.schedule_type is ScheduleType.ONCE:
return self.run_at if self.run_at > now else None
zone = ZoneInfo(self.timezone)
next_local = croniter(self.cron, now.astimezone(zone)).get_next(datetime)
if next_local.tzinfo is None:
# Unreachable with an aware input on today's croniter, and kept
# verbatim from schedules.py:51-52 rather than dropped: it costs
# one branch and guards a library detail we do not control.
next_local = next_local.replace(tzinfo=zone)
return next_local.astimezone(UTC)
def ensure_launchable(self, now: datetime, policy: SchedulePolicy) -> datetime | None:
"""Next fire time, with the constraints that only apply at submission.
Used by create/update; the dispatch path must use next_after instead.
Raises:
InvalidScheduleError: a ONCE schedule with no future occurrence, or
one closer than policy.min_once_delay_seconds. CRON is never
subject to the delay floor (router:105, router:196).
"""
if now.tzinfo is None:
now = now.replace(tzinfo=UTC)
next_at = self.next_after(now)
if self.schedule_type is not ScheduleType.ONCE:
return next_at
if next_at is None:
raise InvalidScheduleError("once schedule must be in the future")
if (next_at - now).total_seconds() < policy.min_once_delay_seconds:
raise InvalidScheduleError(f"once schedule must be at least {policy.min_once_delay_seconds} seconds in the future")
return next_at

289
backend/packages/harness/deerflow/domain/schedule/model/task.py Normal file
View File

@ -0,0 +1,289 @@
from __future__ import annotations
import uuid
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime
from deerflow.domain.schedule.exceptions import InvalidContextModeError, TaskNotMutableError
from deerflow.domain.schedule.model.enums import ContextMode, RunStatus, ScheduleType, TaskStatus, TriggerKind
from deerflow.domain.schedule.model.spec import SchedulePolicy, ScheduleSpec
TERMINAL_TASK_STATUSES = frozenset({TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED})
"""The statuses in which a task's current schedule is done.
Lives here rather than in enums.py: it is a business subset of TaskStatus, so
it belongs beside the aggregate that reasons about it.
Two rules read it, and they share one constant rather than drifting into two
same-valued ones:
- `with_schedule` re-arms them to ENABLED once the schedule moves into the
future — claiming only admits ENABLED rows, so leaving a terminal status
there would hand back a next_run_at that silently never fires.
- The repository's `protect_terminal` compare-and-set refuses to overwrite
them, because a fast-failing run's completion hook can land before the launch
path's own write.
The two coincide because terminal *is* the definition of re-armable. The name
states what the statuses are; what each rule does with them belongs to that
rule.
"""
def _parse_context_mode(value: str | ContextMode) -> ContextMode:
"""Coerce a wire-level context_mode into the enum.
Reported as InvalidContextModeError rather than letting ValueError
escape, so the router can map the whole ScheduleError family uniformly
— an unknown mode was a 422 at router:76-77.
"""
try:
return ContextMode(value)
except ValueError as exc:
raise InvalidContextModeError(f"Unsupported context_mode: {value}") from exc
@dataclass(frozen=True)
class ScheduledTask:
"""A user's standing instruction to run one prompt on a schedule.
Aggregate root of the schedule context. Holds every invariant the old
router/service pair scattered across three files; nothing here knows about
HTTP, SQL, or the run runtime.
The two bookkeeping timestamps have one owner each. `created_at` is the
aggregate's construction instant -- `create` stamps it from its explicit
`now` input and the adapter persists it verbatim, never minting its own.
`updated_at` belongs to the *storage* write path: the CAS methods
(`record_launch` / `record_completion`) update the row without ever
holding an aggregate, so only the adapter can stamp "last written" -- the
`with_*` / `paused` / `resumed` transitions therefore deliberately leave
`updated_at` alone, and the field's value on a read-back is the adapter's
stamp, seeded at construction time. Neither timestamp is a rule input;
the clock the rules see is always an explicit `now=` parameter.
"""
task_id: str
user_id: str
title: str
prompt: str
schedule: ScheduleSpec
context_mode: ContextMode = ContextMode.FRESH_THREAD_PER_RUN
thread_id: str | None = None
assistant_id: str | None = "lead_agent"
status: TaskStatus = TaskStatus.ENABLED
# The MVP fixes this to "skip"; a single-valued enum would be noise. The
# comparison is confined to `skips_on_overlap` so a second policy only has
# to change one place.
overlap_policy: str = "skip"
next_run_at: datetime | None = None
last_run_at: datetime | None = None
last_run_id: str | None = None
last_thread_id: str | None = None
last_error: str | None = None
run_count: int = 0
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
def __post_init__(self) -> None:
if self.context_mode is ContextMode.REUSE_THREAD and not self.thread_id:
raise InvalidContextModeError("reuse_thread requires thread_id")
@classmethod
def create(
cls,
*,
user_id: str,
title: str,
prompt: str,
schedule: ScheduleSpec,
context_mode: str | ContextMode,
thread_id: str | None,
now: datetime,
policy: SchedulePolicy,
) -> ScheduledTask:
"""Factory: generate identity, normalize context, validate invariants.
FRESH_THREAD_PER_RUN drops any supplied thread_id (router:164-165 does
the same on update) — carrying a thread the task will never use would
make `resolve_execution_thread` ambiguous.
Thread existence/ownership is NOT checked here: that needs the
ThreadLookup port and belongs to the service.
"""
mode = _parse_context_mode(context_mode)
effective_thread = thread_id if mode is ContextMode.REUSE_THREAD else None
return cls(
task_id=f"task-{uuid.uuid4().hex}",
user_id=user_id,
title=title,
prompt=prompt,
schedule=schedule,
context_mode=mode,
thread_id=effective_thread,
next_run_at=schedule.ensure_launchable(now, policy),
# The clock is already an explicit rule input here; reading it a
# second time through the field defaults would give one
# construction two instants.
created_at=now,
updated_at=now,
)
@property
def skips_on_overlap(self) -> bool:
"""Whether an overlapping dispatch is dropped rather than queued."""
return self.overlap_policy == "skip"
def resolve_execution_thread(self) -> str:
"""Pick the thread this dispatch executes in (service.py:94-96).
NOT idempotent: FRESH_THREAD_PER_RUN mints a new uuid4 on every call,
so a dispatch must call this exactly once and reuse the value for the
run row, the launch, and the result.
The empty-thread_id fallback is kept even though __post_init__ makes it
unreachable for new aggregates — rows predating that invariant can
still carry REUSE_THREAD with no thread.
"""
if self.context_mode is ContextMode.FRESH_THREAD_PER_RUN or not self.thread_id:
return str(uuid.uuid4())
return self.thread_id
def ensure_mutable(self) -> None:
"""Reject edits while this dispatch round's lease is held.
Was `_ensure_task_mutable` in the router (409). Called by every
state-changing method here rather than by the caller, so the rule
cannot be forgotten at a new call site. The router historically applied
it to update/pause/resume but not trigger/delete; that asymmetry is
intentional and is preserved by keeping those two off this path.
Raises:
TaskNotMutableError: the task is currently RUNNING.
"""
if self.status is TaskStatus.RUNNING:
raise TaskNotMutableError("Scheduled task is currently running; retry after the active execution finishes")
def status_after_launch(self, *, trigger: TriggerKind) -> TaskStatus:
"""Status to persist after a successful launch (service.py:152-161).
Precondition on `self.status`: for a SCHEDULED trigger it is already
RUNNING (written by claim_due); for a MANUAL trigger it is the
user-facing status. The PAUSED rule below depends on that.
Decision order is load-bearing — ONCE is tested first, so manually
triggering a paused ONCE task yields RUNNING, not PAUSED:
ONCE -> RUNNING (await the completion
hook; declaring COMPLETED at
launch would stick if the run
fails or the process dies)
MANUAL and self.status PAUSED -> PAUSED (a manual run must not
silently resume the schedule)
otherwise -> ENABLED
"""
if self.schedule.schedule_type is ScheduleType.ONCE:
return TaskStatus.RUNNING
if trigger is TriggerKind.MANUAL and self.status is TaskStatus.PAUSED:
return TaskStatus.PAUSED
return TaskStatus.ENABLED
def status_after_failure(self, *, trigger: TriggerKind) -> TaskStatus:
"""Status to persist after a failed launch (was _task_status_for_failure).
Decision order is load-bearing — MANUAL is tested first, so a failed
manual trigger of a ONCE task keeps its status instead of burning the
occurrence:
MANUAL -> self.status (unchanged; the old dict-based code
needed an `or "enabled"` fallback, which the
field default makes unnecessary here)
SCHEDULED, ONCE -> FAILED
SCHEDULED, CRON -> ENABLED (the next occurrence still stands)
"""
if trigger is TriggerKind.MANUAL:
return self.status
if self.schedule.schedule_type is ScheduleType.ONCE:
return TaskStatus.FAILED
return TaskStatus.ENABLED
def status_after_skip(self) -> TaskStatus:
"""Status to persist after an overlap skip (was _task_status_for_skip).
No `trigger` parameter: a skip only happens on the SCHEDULED path — a
manual trigger that overlaps is rejected outright and records no run
row at all, so a parameter here would imply a branch that cannot exist.
ONCE -> FAILED (the single occurrence was lost; COMPLETED would
claim an execution that never happened)
CRON -> ENABLED
"""
if self.schedule.schedule_type is ScheduleType.ONCE:
return TaskStatus.FAILED
return TaskStatus.ENABLED
def status_after_completion(self, outcome: RunStatus) -> TaskStatus | None:
"""Status to persist when a launched run reaches a terminal state.
Returns None when the status must not change — every CRON task, whose
schedule outlives any single run.
CRON -> None
ONCE, SUCCESS -> COMPLETED
ONCE, INTERRUPTED -> CANCELLED (a cancel or same-thread takeover is
not an execution failure)
ONCE, otherwise -> FAILED
The occurrence is consumed either way: the run did launch, so re-arming
would risk duplicate side effects. Note the caller writes `last_error`
unconditionally, whether or not this returns None (service.py:346).
"""
if self.schedule.schedule_type is not ScheduleType.ONCE:
return None
if outcome is RunStatus.SUCCESS:
return TaskStatus.COMPLETED
if outcome is RunStatus.INTERRUPTED:
return TaskStatus.CANCELLED
return TaskStatus.FAILED
def with_schedule(self, schedule: ScheduleSpec, *, now: datetime, policy: SchedulePolicy) -> ScheduledTask:
"""Replace the schedule and recompute next_run_at (router:172-208).
A terminal task whose schedule just moved into the future must be
re-armed: claim_due only admits ENABLED rows, so leaving the terminal
status would return 200 with a next_run_at that silently never fires.
Raises:
TaskNotMutableError: via ensure_mutable.
InvalidScheduleError: via ensure_launchable.
"""
self.ensure_mutable()
next_at = schedule.ensure_launchable(now, policy)
status = TaskStatus.ENABLED if next_at is not None and self.status in TERMINAL_TASK_STATUSES else self.status
return replace(self, schedule=schedule, next_run_at=next_at, status=status)
def with_context(self, context_mode: str | ContextMode, thread_id: str | None) -> ScheduledTask:
"""Replace execution context (router:153-165).
FRESH_THREAD_PER_RUN forces thread_id to None. Thread existence is the
service's job (ThreadLookup port), not this aggregate's.
Raises:
TaskNotMutableError: via ensure_mutable.
InvalidContextModeError: unknown mode, or REUSE_THREAD with no
thread (raised by __post_init__ on the replacement).
"""
self.ensure_mutable()
mode = _parse_context_mode(context_mode)
effective_thread = thread_id if mode is ContextMode.REUSE_THREAD else None
return replace(self, context_mode=mode, thread_id=effective_thread)
def paused(self) -> ScheduledTask:
"""Stop claiming this task until resumed."""
self.ensure_mutable()
return replace(self, status=TaskStatus.PAUSED)
def resumed(self) -> ScheduledTask:
"""Re-admit this task to claim_due."""
self.ensure_mutable()
return replace(self, status=TaskStatus.ENABLED)

332
backend/packages/harness/deerflow/domain/schedule/ports.py Normal file
View File

@ -0,0 +1,332 @@
"""Output ports of the schedule context.
Contracts the domain declares and the outer ring implements. Signatures are
technology-neutral: no SQL, table names, HTTP status codes, or run-runtime
types appear here, and every method exchanges domain objects.
The docstrings are longer than the code on purpose -- they are the semantic
contract the adapter must honour and the contract tests are written from.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol, runtime_checkable
from deerflow.domain.schedule.model import (
RunStatus,
ScheduledRun,
ScheduledTask,
TaskStatus,
)
@dataclass(frozen=True)
class LaunchedRun:
"""What the launcher reports back once a run is admitted.
`thread_id` is echoed rather than assumed: the launcher is free to return a
different thread than the one requested, and the task's bookkeeping records
what actually ran.
"""
run_id: str
thread_id: str
@dataclass(frozen=True)
class RunOutcome:
"""A launched run reaching a terminal state, in domain vocabulary.
The app layer converts its own run record into this before calling the
service, so the domain never imports the run runtime. That converter also
owns the filtering the completion hook used to do inline -- a run that
carries no scheduled-task metadata, or has not reached a terminal state,
simply produces no RunOutcome and the service is never called.
`status` is narrowed to the three terminal outcomes a scheduled run can
report: SUCCESS, FAILED, INTERRUPTED. INTERRUPTED is deliberately distinct
from FAILED -- a cancel or same-thread takeover is not an execution
failure, and the task ends CANCELLED rather than FAILED.
"""
task_id: str
record_id: str
run_id: str
user_id: str
status: RunStatus
error: str | None
@runtime_checkable
class ScheduledTaskRepository(Protocol):
"""Persistence port for the task aggregate.
Every read is scoped by `user_id`: a task belonging to someone else is
reported as absent (None / False / omitted from a list), never as a
permission error -- the caller must not be able to distinguish "not yours"
from "does not exist".
"""
async def add(self, task: ScheduledTask) -> ScheduledTask:
"""Insert a new task and return the stored state."""
...
async def get(self, task_id: str, *, user_id: str) -> ScheduledTask | None:
"""Return the task, or None when it is absent or owned by someone else."""
...
async def list_by_user(self, user_id: str) -> list[ScheduledTask]:
"""Every task owned by the user, newest first."""
...
async def list_by_user_and_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]:
"""The user's tasks bound to one thread, newest first.
Only `reuse_thread` tasks can match: a `fresh_thread_per_run` task
carries no thread.
"""
...
async def save(self, task: ScheduledTask) -> ScheduledTask | None:
"""Persist a whole aggregate, keyed by its own id and owner.
Whole-aggregate replacement rather than a field patch: the aggregate is
immutable, so a caller that changed anything is holding a complete new
value. Returns None when the row is absent or owned by someone else.
Not to be used for the post-dispatch write -- see `record_launch`.
"""
...
async def delete(self, task_id: str, *, user_id: str) -> bool:
"""Remove the task. False when it was absent or owned by someone else."""
...
async def claim_due(self, *, now: datetime, lease_seconds: int, limit: int) -> list[ScheduledTask]:
"""Atomically take ownership of up to `limit` tasks that are due.
A task is due when its next fire time has passed AND either it is
claimable (enabled, with no live claim), or it is stuck mid-dispatch
with an expired claim -- the process that took it died between claiming
and dispatching, and it must not stay unreachable forever.
Claiming marks the tasks as running and stamps the claim, so a
concurrent claimer cannot take the same ones. Returns them in the state
they were left in *after* the claim.
Which process is claiming is deliberately absent: it is an identity,
not a rule. The implementation may record one for diagnostics, but
nothing reads it back -- expiry alone decides whether a claim can be
taken over -- so the domain has no reason to carry it.
Atomicity is the implementation's responsibility. An in-memory double
can satisfy every rule above under single-threaded use while providing
no concurrency guarantee at all; that difference is out of scope for
the contract tests and is covered separately against a real database.
"""
...
async def record_launch(
self,
task_id: str,
*,
status: TaskStatus,
next_run_at: datetime | None,
last_run_at: datetime | None,
last_run_id: str | None,
last_thread_id: str | None,
last_error: str | None,
increment_run_count: bool,
protect_terminal: bool = False,
) -> None:
"""Write the outcome of a dispatch and release the claim.
Deliberately NOT expressed as `save(task)`: `protect_terminal` makes
this a compare-and-set against a run that may be finalizing
concurrently. When it is set and the stored task has already reached a
terminal status, the status and error are left alone and only the
scheduling bookkeeping is written -- a read-modify-write through the
aggregate would reintroduce the very race the flag exists to close.
Every field is assigned unconditionally, so a caller preserving a value
must pass the current one back. The claim is always released.
"""
...
async def record_completion(
self,
task_id: str,
*,
user_id: str,
status: TaskStatus | None,
error: str | None,
) -> None:
"""Write a finished run's verdict onto the task.
Deliberately NOT expressed as `save(task)`, for the same reason
`record_launch` is not: the two race. A run that fails fast reaches its
completion hook while the dispatch path is still writing its
bookkeeping, so a read-modify-write through the aggregate would replay
a snapshot taken before that write and roll it back -- restoring an
elapsed `next_run_at` and an out-of-date `run_count`, and leaving the
task in `running` with no live claim, which neither `claim_due` branch
nor `cancel_stuck_once_tasks` can reach.
The two writes therefore own disjoint fields: the launch owns the
schedule, this owns the verdict. Nothing here touches `next_run_at`,
`last_run_at`, `last_run_id`, `last_thread_id`, `run_count` or the
claim.
`status` is `None` when the verdict must not move the status -- every
cron task, whose schedule outlives any single run. `error` is written
either way, so a cron task still reports what went wrong last time.
Scoped by `user_id` like every other read: a task belonging to someone
else is left alone rather than reported. An unknown task is ignored --
one deleted mid-flight simply has nothing to update.
"""
...
async def cancel_stuck_once_tasks(self, *, error: str) -> int:
"""Reconcile `once` tasks orphaned mid-flight by a process crash.
A launched `once` task waits in running for its completion hook, and
its claim was released at launch -- so the expired-claim branch of
`claim_due` can never see it, and after a crash the hook is gone. This
marks those tasks cancelled and returns how many were reconciled.
Tasks still holding a claim are left alone: they were claimed but not
launched, and expired-claim reclaim recovers them safely.
"""
...
@runtime_checkable
class ScheduledRunRepository(Protocol):
"""Persistence port for execution records.
Unlike tasks, runs are not read by owner: they are only ever reached
through a task the caller already proved it owns.
"""
async def add(self, run: ScheduledRun) -> ScheduledRun:
"""Insert an execution record.
Raises:
ActiveRunConflictError: the task already holds its single active
slot. Translating the storage-level rejection into this domain
error is the adapter's job and is load-bearing -- the service
collapses it to exactly the same outcome as the `has_active`
fast path, and those two paths must stay indistinguishable.
A terminal record (a skip tombstone) is outside the active-slot rule
and must never raise it.
"""
...
async def list_by_task(self, task_id: str, *, limit: int, offset: int) -> list[ScheduledRun]:
"""One task's execution history, newest first."""
...
async def count_active(self) -> int:
"""How many executions are active across ALL tasks.
Global on purpose: it bounds total scheduled concurrency, not per-task
overlap. Long runs accumulate across polls, so each poll may only claim
into whatever budget is left.
"""
...
async def has_active(self, task_id: str) -> bool:
"""Whether this task currently holds its active slot.
A non-atomic fast path by nature -- it cannot be relied on to exclude a
concurrent dispatch. `add` is the arbiter; this exists to avoid the
common case of doing pointless work.
"""
...
async def update_status(
self,
record_id: str,
*,
status: RunStatus,
run_id: str | None = None,
error: str | None = None,
started_at: datetime | None = None,
finished_at: datetime | None = None,
protect_terminal: bool = False,
) -> None:
"""Advance an execution record.
With `protect_terminal`, a record that already reached a terminal state
keeps its status and error; only bookkeeping the terminal write could
not have known (the run id, the start time) is backfilled. A run that
fails fast can reach its completion hook before the launch path's own
write lands, and the completion is the authoritative one.
Unknown record ids are ignored rather than raising: the caller is
writing bookkeeping, not asserting existence.
"""
...
async def mark_stale_active(self, *, error: str) -> int:
"""Terminalize executions orphaned by a process crash, returning the count.
Runs execute in-process, so any record still active at startup belongs
to a process that is gone. This is only sound while a single scheduler
instance owns the table.
"""
...
class RunLauncher(Protocol):
"""Output port for actually starting the work.
The contract the adapter MUST honour, because it is what keeps the run
runtime and the web framework out of the inner ring:
- the execution thread is already busy -> ThreadBusyError
- anything else goes wrong -> LaunchFailedError
Nothing else may escape. The domain distinguishes those two because they
lead to different outcomes -- a busy thread on a scheduled dispatch is a
skipped occurrence, while a genuine failure is recorded as one.
"""
async def launch(
self,
*,
thread_id: str,
assistant_id: str | None,
prompt: str,
owner_user_id: str | None,
metadata: dict[str, str],
) -> LaunchedRun:
"""Start one execution and return its identity.
`metadata` is opaque correlation data the domain attaches so the
eventual outcome can be traced back to this task and record; the
adapter must carry it through untouched.
"""
...
class ThreadLookup(Protocol):
"""Narrow port: the only question the schedule context asks about threads.
Deliberately not the full thread store -- depending on this one method
keeps the context decoupled from the wider conversation model.
"""
async def exists_for_user(self, thread_id: str, user_id: str) -> bool:
"""Whether this thread exists AND the user may use it.
Both halves matter: binding a task to a thread that does not exist yet
is as invalid as binding it to someone else's. The two are deliberately
not distinguished in the result -- reporting them differently would let
a caller probe for the existence of threads they cannot see.
"""
...

462
backend/packages/harness/deerflow/domain/schedule/service.py Normal file
View File

@ -0,0 +1,462 @@
"""Application service of the schedule context (its input port).
Orchestrates use cases only: fetch, apply domain rules, persist through output
ports. It holds no business rules itself -- every decision below is delegated
to the aggregate -- and knows nothing about HTTP, SQL, or the run runtime.
`user_id` is always passed in explicitly; resolving the current user is the
primary adapter's job.
The dispatch path deliberately mirrors the structure of the pre-migration
`app/scheduler/service.py` (since deleted), including its ordering and its
comments, because its concurrency and idempotency semantics are load-bearing
and are pinned by tests that must keep passing unchanged.
"""
from __future__ import annotations
from dataclasses import dataclass, replace
from datetime import datetime
from deerflow.domain.schedule.commands import (
UNSET,
CreateScheduledTask,
DeleteTask,
PauseTask,
ResumeTask,
TriggerTask,
UpdateScheduledTask,
)
from deerflow.domain.schedule.exceptions import (
ActiveRunConflictError,
LaunchFailedError,
TaskNotFoundError,
ThreadBusyError,
ThreadNotFoundError,
)
from deerflow.domain.schedule.model import (
ContextMode,
DispatchOutcome,
RunStatus,
ScheduledRun,
ScheduledTask,
SchedulePolicy,
ScheduleType,
TriggerKind,
)
from deerflow.domain.schedule.ports import (
RunLauncher,
RunOutcome,
ScheduledRunRepository,
ScheduledTaskRepository,
ThreadLookup,
)
# Shared so the has_active fast path and the active-slot race path produce
# byte-identical outcomes for the same "task already has an active run"
# condition. Two callers must not be able to tell which one rejected them.
_ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run"
_SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active"
@dataclass(frozen=True)
class DispatchResult:
"""What one dispatch attempt produced.
`outcome` drives the caller's protocol mapping -- a manual trigger turns
CONFLICT into 409 and FAILED into 502 -- so it is domain vocabulary
rather than a bare string.
"""
outcome: DispatchOutcome
record_id: str | None
run_id: str | None
thread_id: str
error: str | None
class ScheduleService:
"""Every scheduled-task use case, orchestrated over the four output ports.
The input port of this context: primary adapters (an HTTP router, the
poller, the run-completion hook) call these methods and translate what
comes back. Domain errors are allowed to propagate -- mapping them onto a
protocol is the adapter's job, not this class's.
"""
def __init__(
self,
*,
tasks: ScheduledTaskRepository,
runs: ScheduledRunRepository,
launcher: RunLauncher,
threads: ThreadLookup,
policy: SchedulePolicy,
) -> None:
self._tasks = tasks
self._runs = runs
self._launcher = launcher
self._threads = threads
self._policy = policy
# ------------------------------------------------------------------ reads
async def list_tasks(self, user_id: str) -> list[ScheduledTask]:
"""Every task the user owns, newest first."""
return await self._tasks.list_by_user(user_id)
async def list_tasks_by_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]:
"""The user's tasks bound to one thread.
Only reuse_thread tasks can appear: a fresh-thread task carries no
binding to show.
"""
return await self._tasks.list_by_user_and_thread(user_id, thread_id)
async def get_task(self, task_id: str, *, user_id: str) -> ScheduledTask:
"""Raises TaskNotFoundError when absent or owned by someone else -- the
caller must not be able to tell those apart."""
task = await self._tasks.get(task_id, user_id=user_id)
if task is None:
raise TaskNotFoundError("Scheduled task not found")
return task
async def list_task_runs(self, task_id: str, *, user_id: str, limit: int = 50, offset: int = 0) -> list[ScheduledRun]:
"""Execution history, gated on ownership of the parent task."""
await self.get_task(task_id, user_id=user_id)
return await self._runs.list_by_task(task_id, limit=limit, offset=offset)
# ------------------------------------------------------------------ writes
async def create_scheduled_task(self, cmd: CreateScheduledTask, *, now: datetime) -> ScheduledTask:
"""Register a new standing instruction.
The aggregate is built first so a malformed schedule or context mode is
reported as such before any IO happens; only then is the thread
binding verified.
Raises:
InvalidScheduleError / InvalidContextModeError: from the aggregate.
ThreadNotFoundError: reuse_thread pointing at a thread the user
cannot use.
"""
task = ScheduledTask.create(
user_id=cmd.user_id,
title=cmd.title,
prompt=cmd.prompt,
schedule=cmd.schedule,
context_mode=cmd.context_mode,
thread_id=cmd.thread_id,
now=now,
policy=self._policy,
)
await self._require_thread(task)
return await self._tasks.add(task)
async def update_scheduled_task(self, cmd: UpdateScheduledTask, *, now: datetime) -> ScheduledTask:
"""Partially update a task; ``UNSET`` means "not supplied".
The one field for which ``None`` is a meaningful value, `thread_id`,
travels inside `ContextChange` where it is unambiguous.
Context and schedule are applied through the aggregate's own
transitions, so the re-arm rule and the running-task gate cannot be
bypassed by patching fields directly.
"""
task = await self.get_task(cmd.task_id, user_id=cmd.user_id)
task.ensure_mutable()
if cmd.context is not UNSET:
task = task.with_context(cmd.context.context_mode, cmd.context.thread_id)
await self._require_thread(task)
if cmd.schedule is not UNSET:
task = task.with_schedule(cmd.schedule, now=now, policy=self._policy)
if cmd.title is not UNSET:
task = replace(task, title=cmd.title)
if cmd.prompt is not UNSET:
task = replace(task, prompt=cmd.prompt)
return await self._save(task)
async def pause_task(self, cmd: PauseTask) -> ScheduledTask:
"""Stop claiming this task until it is resumed.
Refused while the task is being dispatched -- see `ensure_mutable`.
"""
task = await self.get_task(cmd.task_id, user_id=cmd.user_id)
return await self._save(task.paused())
async def resume_task(self, cmd: ResumeTask) -> ScheduledTask:
"""Re-admit this task to claiming. Same gate as `pause_task`."""
task = await self.get_task(cmd.task_id, user_id=cmd.user_id)
return await self._save(task.resumed())
async def delete_task(self, cmd: DeleteTask) -> None:
"""Deleting is deliberately not gated on the task being idle: the
pre-migration router applied that gate to update/pause/resume only."""
if not await self._tasks.delete(cmd.task_id, user_id=cmd.user_id):
raise TaskNotFoundError("Scheduled task not found")
# ------------------------------------------------------------------ dispatch
async def trigger_task(self, cmd: TriggerTask, *, now: datetime) -> DispatchResult:
"""Dispatch a task on demand. Unlike the scheduled path this is allowed
while the task is paused, and leaves it paused."""
task = await self.get_task(cmd.task_id, user_id=cmd.user_id)
return await self.dispatch_task(task, now=now, trigger=TriggerKind.MANUAL)
async def run_once(self, *, now: datetime) -> list[DispatchResult]:
"""Claim whatever is due and dispatch it.
`max_concurrent_runs` is a global cap on active scheduled runs, not a
per-poll batch size: long runs accumulate across poll cycles, so each
cycle only claims into the remaining budget.
"""
active = await self._runs.count_active()
budget = self._policy.max_concurrent_runs - active
if budget <= 0:
return []
claimed = await self._tasks.claim_due(now=now, lease_seconds=self._policy.lease_seconds, limit=budget)
return [await self.dispatch_task(task, now=now, trigger=TriggerKind.SCHEDULED) for task in claimed]
async def dispatch_task(self, task: ScheduledTask, *, now: datetime, trigger: TriggerKind) -> DispatchResult:
"""Turn one due task into one execution.
Called once per dispatch, so `resolve_execution_thread` is called once
and its value reused for the record, the launch, and the result.
"""
execution_thread_id = task.resolve_execution_thread()
# "skip" must hold for fresh-thread runs too, where every run gets a
# new thread and the same-thread busy signal below can never fire.
# Checked before creating this dispatch's own record so the record does
# not count itself as the active run. A manual trigger against an
# active run is rejected outright instead of being recorded as a
# skipped occurrence -- nothing was scheduled to happen.
#
# This check is a NON-ATOMIC fast path: two concurrent dispatches (a
# manual trigger racing the poller, a double-click, a client retry) can
# both observe no active run. The repository is the atomic arbiter --
# it rejects the second active record with ActiveRunConflictError,
# which collapses to the SAME outcome as this fast path just below.
if task.skips_on_overlap and await self._runs.has_active(task.task_id):
if trigger is TriggerKind.MANUAL:
return self._conflict(execution_thread_id)
return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
record = ScheduledRun.queued(
task_id=task.task_id,
thread_id=execution_thread_id,
scheduled_for=now,
trigger=trigger,
)
try:
await self._runs.add(record)
except ActiveRunConflictError:
# Lost the race for the task's single active slot: a concurrent
# dispatch passed the same fast-path check and inserted first.
# Identical outcome to the fast path above -- that equality is what
# the dispatch-race regression tests pin.
if trigger is TriggerKind.MANUAL:
return self._conflict(execution_thread_id)
return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
# Only the launch is guarded. The port contract admits exactly two
# escapes, so a failure of the bookkeeping writes below is a genuine
# fault and propagates instead of being recorded as a failed launch --
# which would mark an already-running execution as failed.
try:
launched = await self._launcher.launch(
thread_id=execution_thread_id,
assistant_id=task.assistant_id,
prompt=task.prompt,
owner_user_id=task.user_id,
metadata={
"scheduled_task_id": task.task_id,
"scheduled_task_run_id": record.record_id,
"scheduled_trigger": str(trigger),
},
)
except ThreadBusyError as exc:
# The execution thread is already busy. On the scheduled path under
# a skip policy this is an overlap like any other; anything else is
# reported as a conflict the caller has to deal with.
if trigger is TriggerKind.SCHEDULED and task.skips_on_overlap:
return await self._finalize_skip(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, error=str(exc))
return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.CONFLICT)
except LaunchFailedError as exc:
return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.FAILED)
await self._runs.update_status(
record.record_id,
status=RunStatus.RUNNING,
run_id=launched.run_id,
started_at=now,
# A fast-failing run can reach handle_run_completion before this
# write lands; never clobber its terminal verdict.
protect_terminal=True,
)
await self._tasks.record_launch(
task.task_id,
status=task.status_after_launch(trigger=trigger),
next_run_at=task.schedule.next_after(now),
last_run_at=now,
last_run_id=launched.run_id,
last_thread_id=launched.thread_id,
last_error=None,
increment_run_count=True,
# Same race as the record write above.
protect_terminal=True,
)
return DispatchResult(DispatchOutcome.LAUNCHED, record.record_id, launched.run_id, launched.thread_id, None)
# ------------------------------------------------------------------ lifecycle
async def handle_run_completion(self, outcome: RunOutcome, *, now: datetime) -> None:
"""Write back a launched run's terminal verdict.
A task deleted while its run was in flight simply has nothing to
update, and that is not an error.
"""
await self._runs.update_status(
outcome.record_id,
status=outcome.status,
run_id=outcome.run_id,
error=outcome.error,
finished_at=now,
)
# Read to ask the aggregate what this outcome means, then write only
# that. `status_after_completion` reads nothing but the schedule type,
# which no concurrent write can change, so this read carries no
# time-of-check risk -- and `record_completion` deliberately does not
# write the fields that a concurrent `record_launch` owns.
task = await self._tasks.get(outcome.task_id, user_id=outcome.user_id)
if task is None:
return
# The error is recorded whether or not the status moves: a cron task
# keeps its schedule but still reports what went wrong last time.
await self._tasks.record_completion(
outcome.task_id,
user_id=outcome.user_id,
status=task.status_after_completion(outcome.status),
error=outcome.error,
)
async def reconcile_on_startup(self, *, error: str) -> tuple[int, int]:
"""Clean up what a process crash left behind, returning what was fixed.
Two sweeps, because a crash strands two different things: execution
records that can never finish, and `once` tasks parked waiting for a
completion hook that died with the process. The second is not covered
by expired-claim reclaim -- a launched task released its claim, so the
claim query can never see it again.
Failures propagate: whether a partial reconcile should block startup is
the caller's policy, not the domain's.
"""
stale_runs = await self._runs.mark_stale_active(error=error)
stuck_tasks = await self._tasks.cancel_stuck_once_tasks(error=error)
return stale_runs, stuck_tasks
# ------------------------------------------------------------------ internals
async def _require_thread(self, task: ScheduledTask) -> None:
if task.context_mode is not ContextMode.REUSE_THREAD:
return
# The aggregate guarantees a thread here; the check keeps that
# guarantee from being silently dropped under `python -O`.
if not task.thread_id or not await self._threads.exists_for_user(task.thread_id, task.user_id):
raise ThreadNotFoundError("Thread not found")
async def _save(self, task: ScheduledTask) -> ScheduledTask:
"""Persist an already-validated aggregate.
Read-modify-write rather than a conditional UPDATE: pushing the
"not while running" rule into a storage predicate would put it beyond
the reach of a zero-IO test and give it a second home. The pre-existing
code had the same shape, and closing the window properly needs
optimistic locking, which needs a schema change.
"""
saved = await self._tasks.save(task)
if saved is None:
raise TaskNotFoundError("Scheduled task not found")
return saved
def _conflict(self, thread_id: str) -> DispatchResult:
"""A manual trigger against an active run.
No history record is written: nothing was scheduled to happen, so there
is no occurrence to account for.
"""
return DispatchResult(DispatchOutcome.CONFLICT, None, None, thread_id, _ACTIVE_RUN_CONFLICT_ERROR)
async def _record_scheduled_skip(self, task: ScheduledTask, *, thread_id: str, now: datetime, trigger: TriggerKind) -> DispatchResult:
"""Account for a scheduled occurrence dropped because of an overlap.
The tombstone is created directly terminal rather than as the transient
queued record the launch path uses: a queued record is active and would
itself be refused against the pre-existing run that is still holding
the task's single active slot.
"""
record = ScheduledRun.skipped_tombstone(
task_id=task.task_id,
thread_id=thread_id,
scheduled_for=now,
trigger=trigger,
)
await self._runs.add(record)
return await self._finalize_skip(task, record_id=record.record_id, thread_id=thread_id, now=now, error=_SKIP_ACTIVE_RUN_ERROR)
async def _finalize_skip(self, task: ScheduledTask, *, record_id: str, thread_id: str, now: datetime, error: str) -> DispatchResult:
await self._runs.update_status(
record_id,
status=RunStatus.SKIPPED,
error=error,
started_at=now,
finished_at=now,
)
await self._tasks.record_launch(
task.task_id,
status=task.status_after_skip(),
next_run_at=task.schedule.next_after(now),
# A skip is not an execution, so the launch bookkeeping carries
# over unchanged; record_launch assigns unconditionally, so
# "unchanged" has to be spelled out.
last_run_at=task.last_run_at,
last_run_id=task.last_run_id,
last_thread_id=task.last_thread_id,
# Only a lost one-shot occurrence is worth surfacing; a cron task
# simply waits for its next turn.
last_error=error if task.schedule.schedule_type is ScheduleType.ONCE else None,
increment_run_count=False,
)
return DispatchResult(DispatchOutcome.SKIPPED, record_id, None, thread_id, error)
async def _fail(
self,
task: ScheduledTask,
*,
record_id: str,
thread_id: str,
now: datetime,
trigger: TriggerKind,
error: str,
outcome: DispatchOutcome,
) -> DispatchResult:
await self._runs.update_status(
record_id,
status=RunStatus.FAILED,
error=error,
started_at=now,
finished_at=now,
)
await self._tasks.record_launch(
task.task_id,
status=task.status_after_failure(trigger=trigger),
next_run_at=task.schedule.next_after(now),
last_run_at=now,
last_run_id=None,
last_thread_id=thread_id,
last_error=error,
increment_run_count=False,
)
return DispatchResult(outcome, record_id, None, thread_id, error)

293
backend/tests/schedule_fakes.py Normal file
View File

@ -0,0 +1,293 @@
"""Shared zero-IO doubles for the schedule context.
Lives in its own module rather than inside a test file so the port contract
suite and the service tests can both import it as an ordinary module, without
relying on the tests directory happening to be on ``sys.path``.
**Concurrency is explicitly out of scope.** These doubles model the
single-threaded semantics of each port -- which rows `claim_due` selects, that
`add` refuses a second active run -- and provide no atomicity whatsoever. A
green run here says nothing about two dispatchers racing; that is covered
against a real database in ``test_schedule_dispatch_race.py``. Do not
read a passing contract suite as licence to run more than one scheduler.
"""
from __future__ import annotations
from dataclasses import dataclass, field, replace
from datetime import datetime, timedelta
from deerflow.domain.schedule.exceptions import ActiveRunConflictError
from deerflow.domain.schedule.model import (
ACTIVE_RUN_STATUSES,
TERMINAL_RUN_STATUSES,
TERMINAL_TASK_STATUSES,
RunStatus,
ScheduledRun,
ScheduledTask,
ScheduleType,
TaskStatus,
)
from deerflow.domain.schedule.ports import LaunchedRun
@dataclass
class _TaskRow:
"""One stored task: the aggregate plus the columns the domain has no
business knowing about.
The lease lives here rather than on ``ScheduledTask`` for the same reason
it lives in the table and not in the domain -- it is a storage-level
concurrency control, not a business fact about the task.
"""
task: ScheduledTask
lease_owner: str | None = None
lease_expires_at: datetime | None = None
class InMemoryScheduledTaskRepository:
"""ScheduledTaskRepository double backed by a dict."""
def __init__(self) -> None:
self._rows: dict[str, _TaskRow] = {}
# -- inspection helpers (tests only, not part of the port) -----------
def lease_of(self, task_id: str) -> tuple[str | None, datetime | None]:
row = self._rows[task_id]
return row.lease_owner, row.lease_expires_at
def seed(self, task: ScheduledTask, *, lease_owner: str | None = None, lease_expires_at: datetime | None = None) -> ScheduledTask:
"""Install a task directly, optionally already claimed."""
self._rows[task.task_id] = _TaskRow(task=task, lease_owner=lease_owner, lease_expires_at=lease_expires_at)
return task
# -- port ------------------------------------------------------------
async def add(self, task: ScheduledTask) -> ScheduledTask:
self._rows[task.task_id] = _TaskRow(task=task)
return task
async def get(self, task_id: str, *, user_id: str) -> ScheduledTask | None:
row = self._rows.get(task_id)
if row is None or row.task.user_id != user_id:
return None
return row.task
async def list_by_user(self, user_id: str) -> list[ScheduledTask]:
rows = [row.task for row in self._rows.values() if row.task.user_id == user_id]
return sorted(rows, key=lambda t: (t.created_at, t.task_id), reverse=True)
async def list_by_user_and_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]:
tasks = await self.list_by_user(user_id)
return [task for task in tasks if task.thread_id == thread_id]
async def save(self, task: ScheduledTask) -> ScheduledTask | None:
row = self._rows.get(task.task_id)
if row is None or row.task.user_id != task.user_id:
return None
row.task = task
return task
async def delete(self, task_id: str, *, user_id: str) -> bool:
row = self._rows.get(task_id)
if row is None or row.task.user_id != user_id:
return False
del self._rows[task_id]
return True
async def claim_due(self, *, now: datetime, lease_seconds: int, limit: int) -> list[ScheduledTask]:
def claimable(row: _TaskRow) -> bool:
task = row.task
if task.next_run_at is None or task.next_run_at > now:
return False
expired = row.lease_expires_at is not None and row.lease_expires_at < now
if task.status is TaskStatus.ENABLED:
return row.lease_expires_at is None or expired
# Stuck mid-dispatch: the claimer died between claim and launch.
return task.status is TaskStatus.RUNNING and expired
due = sorted(
(row for row in self._rows.values() if claimable(row)),
key=lambda row: (row.task.next_run_at, row.task.task_id),
)[:limit]
claimed = []
for row in due:
# Stands in for whatever identity a real adapter records.
row.lease_owner = "fake-worker"
row.lease_expires_at = now + timedelta(seconds=lease_seconds)
row.task = replace(row.task, status=TaskStatus.RUNNING)
claimed.append(row.task)
return claimed
async def record_launch(
self,
task_id: str,
*,
status: TaskStatus,
next_run_at: datetime | None,
last_run_at: datetime | None,
last_run_id: str | None,
last_thread_id: str | None,
last_error: str | None,
increment_run_count: bool,
protect_terminal: bool = False,
) -> None:
row = self._rows.get(task_id)
if row is None:
return
task = row.task
if not (protect_terminal and task.status in TERMINAL_TASK_STATUSES):
task = replace(task, status=status, last_error=last_error)
task = replace(
task,
next_run_at=next_run_at,
last_run_at=last_run_at,
last_run_id=last_run_id,
last_thread_id=last_thread_id,
)
if increment_run_count:
task = replace(task, run_count=task.run_count + 1)
row.task = task
row.lease_owner = None
row.lease_expires_at = None
async def record_completion(
self,
task_id: str,
*,
user_id: str,
status: TaskStatus | None,
error: str | None,
) -> None:
row = self._rows.get(task_id)
if row is None or row.task.user_id != user_id:
return
# Only the verdict; every scheduling field belongs to record_launch.
row.task = replace(row.task, last_error=error)
if status is not None:
row.task = replace(row.task, status=status)
async def cancel_stuck_once_tasks(self, *, error: str) -> int:
cancelled = 0
for row in self._rows.values():
stuck = row.task.status is TaskStatus.RUNNING and row.task.schedule.schedule_type is ScheduleType.ONCE and row.lease_expires_at is None
if stuck:
row.task = replace(row.task, status=TaskStatus.CANCELLED, last_error=error)
cancelled += 1
return cancelled
class InMemoryScheduledRunRepository:
"""ScheduledRunRepository double that also models the active-slot rule.
``add`` refusing a second active record for one task is not an
implementation detail to be skipped here: the service collapses that
rejection to the same outcome as its own fast path, and a double that never
raises would let that collapse go untested.
"""
def __init__(self) -> None:
self._rows: dict[str, ScheduledRun] = {}
def all_runs(self) -> list[ScheduledRun]:
"""Inspection helper (tests only, not part of the port)."""
return list(self._rows.values())
async def add(self, run: ScheduledRun) -> ScheduledRun:
if run.is_active and any(other.task_id == run.task_id and other.is_active for other in self._rows.values()):
raise ActiveRunConflictError(f"scheduled task {run.task_id!r} already has an active run")
self._rows[run.record_id] = run
return run
async def list_by_task(self, task_id: str, *, limit: int = 50, offset: int = 0) -> list[ScheduledRun]:
rows = [run for run in self._rows.values() if run.task_id == task_id]
rows.sort(key=lambda run: (run.created_at, run.record_id), reverse=True)
return rows[offset : offset + limit]
async def count_active(self) -> int:
return sum(1 for run in self._rows.values() if run.is_active)
async def has_active(self, task_id: str) -> bool:
return any(run.task_id == task_id and run.is_active for run in self._rows.values())
async def update_status(
self,
record_id: str,
*,
status: RunStatus,
run_id: str | None = None,
error: str | None = None,
started_at: datetime | None = None,
finished_at: datetime | None = None,
protect_terminal: bool = False,
) -> None:
run = self._rows.get(record_id)
if run is None:
return
if protect_terminal and run.status in TERMINAL_RUN_STATUSES:
# Keep the terminal verdict; only backfill what it could not know.
if run.run_id is None and run_id is not None:
run = replace(run, run_id=run_id)
if run.started_at is None and started_at is not None:
run = replace(run, started_at=started_at)
self._rows[record_id] = run
return
run = replace(run, status=status, run_id=run_id, error=error)
if started_at is not None:
run = replace(run, started_at=started_at)
if finished_at is not None:
run = replace(run, finished_at=finished_at)
self._rows[record_id] = run
async def mark_stale_active(self, *, error: str) -> int:
stale = [run for run in self._rows.values() if run.status in ACTIVE_RUN_STATUSES]
for run in stale:
self._rows[run.record_id] = replace(run, status=RunStatus.INTERRUPTED, error=error)
return len(stale)
@dataclass
class FakeRunLauncher:
"""RunLauncher double.
`fail_with` makes every launch raise instead -- set it to a ThreadBusyError
or a LaunchFailedError to drive the two branches the domain distinguishes.
"""
fail_with: Exception | None = None
calls: list[dict] = field(default_factory=list)
async def launch(
self,
*,
thread_id: str,
assistant_id: str | None,
prompt: str,
owner_user_id: str | None,
metadata: dict[str, str],
) -> LaunchedRun:
self.calls.append(
{
"thread_id": thread_id,
"assistant_id": assistant_id,
"prompt": prompt,
"owner_user_id": owner_user_id,
"metadata": metadata,
}
)
if self.fail_with is not None:
raise self.fail_with
return LaunchedRun(run_id=f"run-{len(self.calls)}", thread_id=thread_id)
class FakeThreadLookup:
"""ThreadLookup double backed by a thread_id -> owner mapping."""
def __init__(self, threads: dict[str, str] | None = None) -> None:
self._threads = dict(threads or {})
async def exists_for_user(self, thread_id: str, user_id: str) -> bool:
return self._threads.get(thread_id) == user_id

627
backend/tests/test_schedule_domain.py Normal file
View File

@ -0,0 +1,627 @@
"""Domain tests for the schedule bounded context.
Deliberately synchronous and dependency-free: no pytest-asyncio, no fakes, no
IO. Everything under `deerflow.domain.schedule.model` is pure, and this file is
the proof — if a rule here ever needs a stub, the rule has leaked out of the
inner ring.
The three `status_after_*` truth tables (TestStatusAfter*) are the reason this
commit exists: those rules used to be static methods on
`app/scheduler/service.py` with no direct coverage at all, reachable only
through the service's integration tests.
"""
from __future__ import annotations
import uuid
from dataclasses import replace
from datetime import UTC, datetime, timedelta
import pytest
from deerflow.domain.schedule.exceptions import InvalidContextModeError, InvalidScheduleError, TaskNotMutableError
from deerflow.domain.schedule.model import (
ContextMode,
RunStatus,
ScheduledRun,
ScheduledTask,
SchedulePolicy,
ScheduleSpec,
ScheduleType,
TaskStatus,
TriggerKind,
)
NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
NO_DELAY = SchedulePolicy(min_once_delay_seconds=0)
MIN_60 = SchedulePolicy(min_once_delay_seconds=60)
def cron_spec(expr: str = "0 9 * * *", timezone: str = "UTC") -> ScheduleSpec:
return ScheduleSpec.cron_schedule(expr, timezone)
def once_spec(*, after_seconds: int = 3600, timezone: str = "UTC") -> ScheduleSpec:
return ScheduleSpec.once_at(NOW + timedelta(seconds=after_seconds), timezone)
def make_task(
*,
schedule: ScheduleSpec | None = None,
status: TaskStatus = TaskStatus.ENABLED,
context_mode: ContextMode = ContextMode.FRESH_THREAD_PER_RUN,
thread_id: str | None = None,
) -> ScheduledTask:
return ScheduledTask(
task_id="task-1",
user_id="user-1",
title="Daily summary",
prompt="Summarize",
schedule=schedule if schedule is not None else cron_spec(),
status=status,
context_mode=context_mode,
thread_id=thread_id,
)
# ---------------------------------------------------------------- A. ScheduleSpec invariants
class TestScheduleSpecInvariants:
def test_unknown_timezone_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="Unknown timezone"):
ScheduleSpec.cron_schedule("0 9 * * *", "Mars/Olympus_Mons")
def test_cron_without_expression_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="requires schedule_spec"):
ScheduleSpec(ScheduleType.CRON, "UTC")
@pytest.mark.parametrize("expr", ["0 9 * *", "0 9 * * * *", "0"])
def test_cron_must_have_exactly_five_fields(self, expr):
with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
ScheduleSpec.cron_schedule(expr, "UTC")
def test_once_without_run_at_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="requires run_at"):
ScheduleSpec(ScheduleType.ONCE, "UTC")
def test_direct_construction_cannot_bypass_validation(self):
"""A frozen dataclass is still constructible field-by-field, so the
rules have to live in __post_init__ rather than in the factories."""
with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
ScheduleSpec(ScheduleType.CRON, "UTC", cron="not-a-cron")
def test_cron_whitespace_is_normalized_on_construction(self):
assert ScheduleSpec(ScheduleType.CRON, "UTC", cron=" 0 9 * * * ").cron == "0 9 * * *"
def test_naive_run_at_is_localized_to_the_schedule_timezone(self):
spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai") # noqa: DTZ001 -- naive input is the subject
assert spec.run_at.utcoffset() == timedelta(hours=8)
assert spec.run_at.astimezone(UTC) == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
def test_aware_run_at_is_left_alone(self):
"""Covers the shape the frontend submits (`zonedLocalToUtcIso`, trailing
Z): already aware, so localization must leave it alone rather than
reinterpreting it in the task's timezone. Parsing that spelling is
`from_primitives`' job and is tested there."""
run_at = datetime(2026, 8, 1, 9, 0, tzinfo=UTC)
assert ScheduleSpec.once_at(run_at, "Asia/Shanghai").run_at == run_at
# ---------------------------------------------------------------- A2. from_primitives
class TestFromPrimitives:
"""The one construction path that starts from untrusted strings.
Both boundaries -- the HTTP body and the stored JSON column -- arrive as a
schedule type plus a loose mapping, so the rule for turning that into a
value object lives here rather than being written out once per adapter.
Each adapter is left with the part that is genuinely its own: which keys
its format uses.
Values are type-checked rather than assumed. The signature says `str |
None` because that is the contract, but both callers read from data a
client can influence, so a non-string has to be rejected here rather than
reaching `datetime.fromisoformat` or the cron parser.
"""
def test_builds_a_cron_schedule(self):
spec = ScheduleSpec.from_primitives("cron", cron="0 9 * * *", run_at=None, timezone="Asia/Shanghai")
assert spec.schedule_type is ScheduleType.CRON
assert spec.cron == "0 9 * * *"
assert spec.timezone == "Asia/Shanghai"
def test_builds_a_once_schedule_from_an_iso_string(self):
spec = ScheduleSpec.from_primitives("once", cron=None, run_at="2026-08-01T09:00:00", timezone="Asia/Shanghai")
assert spec.schedule_type is ScheduleType.ONCE
assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
def test_a_trailing_z_run_at_is_the_same_instant(self):
"""The shape the frontend submits (`zonedLocalToUtcIso`)."""
spec = ScheduleSpec.from_primitives("once", cron=None, run_at="2026-08-01T01:00:00Z", timezone="Asia/Shanghai")
assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
def test_unknown_schedule_type_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"):
ScheduleSpec.from_primitives("teleport", cron="0 9 * * *", run_at=None, timezone="UTC")
@pytest.mark.parametrize("cron", [None, 5, ""])
def test_cron_without_a_usable_expression_is_rejected(self, cron):
with pytest.raises(InvalidScheduleError, match="requires schedule_spec"):
ScheduleSpec.from_primitives("cron", cron=cron, run_at=None, timezone="UTC")
@pytest.mark.parametrize("run_at", [None, 5])
def test_once_without_a_string_run_at_is_rejected(self, run_at):
with pytest.raises(InvalidScheduleError, match="requires run_at"):
ScheduleSpec.from_primitives("once", cron=None, run_at=run_at, timezone="UTC")
def test_an_unparseable_run_at_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="unparseable run_at"):
ScheduleSpec.from_primitives("once", cron=None, run_at="next tuesday", timezone="UTC")
def test_the_irrelevant_field_is_ignored(self):
"""A boundary that carries both keys must not be rejected for it."""
spec = ScheduleSpec.from_primitives("cron", cron="0 9 * * *", run_at="2026-08-01T09:00:00", timezone="UTC")
assert spec.run_at is None
def test_value_rules_still_come_from_post_init(self):
"""Not re-implemented here -- this is why each adapter stays thin."""
with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
ScheduleSpec.from_primitives("cron", cron="0 9 * *", run_at=None, timezone="UTC")
with pytest.raises(InvalidScheduleError, match="Unknown timezone"):
ScheduleSpec.from_primitives("cron", cron="0 9 * * *", run_at=None, timezone="Mars/Olympus_Mons")
def test_whitespace_in_a_cron_expression_is_normalized(self):
spec = ScheduleSpec.from_primitives("cron", cron=" 0 9 * * * ", run_at=None, timezone="UTC")
assert spec.cron == "0 9 * * *"
# ---------------------------------------------------------------- B. next_after
class TestNextAfter:
def test_once_in_the_future_returns_run_at(self):
spec = once_spec(after_seconds=3600)
assert spec.next_after(NOW) == NOW + timedelta(seconds=3600)
def test_once_in_the_past_returns_none(self):
spec = ScheduleSpec.once_at(NOW - timedelta(seconds=1), "UTC")
assert spec.next_after(NOW) is None
def test_once_exactly_now_returns_none(self):
"""`run_at > now`, not `>=` (schedules.py:44)."""
assert ScheduleSpec.once_at(NOW, "UTC").next_after(NOW) is None
def test_cron_returns_the_next_occurrence_in_utc(self):
# 09:00 UTC daily; NOW is 12:00 UTC, so the next one is tomorrow.
assert cron_spec("0 9 * * *", "UTC").next_after(NOW) == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)
def test_cron_is_evaluated_in_the_schedule_timezone(self):
# 09:00 Shanghai == 01:00 UTC. NOW (12:00 UTC) is past today's, so the
# answer is tomorrow 01:00 UTC -- not 09:00 UTC.
assert cron_spec("0 9 * * *", "Asia/Shanghai").next_after(NOW) == datetime(2026, 7, 28, 1, 0, tzinfo=UTC)
def test_cron_absorbs_a_dst_transition(self):
"""US DST starts 2026-03-08, so the same wall-clock cron maps to a
different UTC instant on either side of it. This is what evaluating in
the schedule's timezone buys."""
spec = cron_spec("0 9 * * *", "America/New_York")
before = spec.next_after(datetime(2026, 3, 7, 0, 0, tzinfo=UTC))
after = spec.next_after(datetime(2026, 3, 9, 0, 0, tzinfo=UTC))
assert before == datetime(2026, 3, 7, 14, 0, tzinfo=UTC) # EST, UTC-5
assert after == datetime(2026, 3, 9, 13, 0, tzinfo=UTC) # EDT, UTC-4
def test_naive_now_is_read_as_utc(self):
spec = cron_spec("0 9 * * *", "UTC")
assert spec.next_after(NOW.replace(tzinfo=None)) == spec.next_after(NOW)
# ---------------------------------------------------------------- C. ensure_launchable
class TestEnsureLaunchable:
def test_once_in_the_past_is_rejected(self):
spec = ScheduleSpec.once_at(NOW - timedelta(hours=1), "UTC")
with pytest.raises(InvalidScheduleError, match="must be in the future"):
spec.ensure_launchable(NOW, NO_DELAY)
def test_once_closer_than_the_minimum_delay_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="at least 60 seconds"):
once_spec(after_seconds=59).ensure_launchable(NOW, MIN_60)
def test_once_exactly_at_the_minimum_delay_is_accepted(self):
assert once_spec(after_seconds=60).ensure_launchable(NOW, MIN_60) == NOW + timedelta(seconds=60)
def test_cron_is_never_subject_to_the_delay_floor(self):
"""router:105 / router:196 apply the floor only to `once`; a cron
schedule whose next fire is seconds away is perfectly legal."""
spec = cron_spec("* * * * *", "UTC")
assert spec.ensure_launchable(NOW, MIN_60) is not None
def test_naive_now_is_read_as_utc(self):
"""Same tolerance as next_after: a caller handing over a naive clock
reading must not silently shift the delay floor by the local offset."""
spec = once_spec(after_seconds=3600)
assert spec.ensure_launchable(NOW.replace(tzinfo=None), MIN_60) == spec.ensure_launchable(NOW, MIN_60)
class TestSchedulePolicyDefaults:
""" "Nobody configured a policy" must not invent a business constraint.
Real values only ever arrive from the composition root (spec.py:22-25). The
defaults being the permissive ones is what keeps an unconfigured deployment
from silently rejecting schedules a configured one would accept -- so they
are pinned here rather than left to whatever the dataclass happens to say.
"""
def test_defaults_are_permissive(self):
policy = SchedulePolicy()
assert policy.min_once_delay_seconds == 0
assert policy.max_concurrent_runs == 1
assert policy.lease_seconds == 60
def test_the_default_delay_floor_admits_an_imminent_once_schedule(self):
"""The consequence of the constant above, stated as behaviour: with no
configured policy a one-second-away one-shot is legal."""
assert once_spec(after_seconds=1).ensure_launchable(NOW, SchedulePolicy()) == NOW + timedelta(seconds=1)
# ---------------------------------------------------------------- D. value semantics
class TestScheduleSpecEquality:
"""A value object is its values -- two specs built from equivalent inputs
must compare equal, because the repository relies on that to decide whether
a schedule actually changed.
Serialization shape (the JSON written to `scheduled_tasks.schedule_spec`)
is deliberately NOT tested here: that mapping lives in the adapter layer,
and so do its tests.
"""
def test_equivalent_cron_inputs_converge(self):
assert ScheduleSpec.cron_schedule(" 0 9 * * * ", "UTC") == ScheduleSpec.cron_schedule("0 9 * * *", "UTC")
def test_equivalent_once_inputs_converge(self):
naive = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai") # noqa: DTZ001 -- naive input is the subject
aware = ScheduleSpec.once_at(datetime.fromisoformat("2026-08-01T01:00:00Z"), "Asia/Shanghai")
assert naive == aware
def test_timezone_is_part_of_identity(self):
at = datetime(2026, 8, 1, 9, 0, tzinfo=UTC)
assert ScheduleSpec.once_at(at, "UTC") != ScheduleSpec.once_at(at, "Asia/Shanghai")
# ---------------------------------------------------------------- E. ScheduledTask invariants
class TestScheduledTaskInvariants:
def test_reuse_thread_requires_a_thread(self):
with pytest.raises(InvalidContextModeError, match="reuse_thread requires thread_id"):
make_task(context_mode=ContextMode.REUSE_THREAD, thread_id=None)
def test_unknown_context_mode_is_a_domain_error(self):
"""Not a bare ValueError -- the router maps the ScheduleError family
uniformly (router:76-77 was a 422)."""
with pytest.raises(InvalidContextModeError, match="Unsupported context_mode"):
ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec(),
context_mode="teleport",
thread_id=None,
now=NOW,
policy=NO_DELAY,
)
def test_create_drops_thread_id_for_fresh_thread_mode(self):
task = ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id="thread-9",
now=NOW,
policy=NO_DELAY,
)
assert task.thread_id is None
def test_create_computes_next_run_at_and_defaults(self):
task = ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec("0 9 * * *", "UTC"),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
now=NOW,
policy=NO_DELAY,
)
assert task.task_id.startswith("task-")
assert task.status is TaskStatus.ENABLED
assert task.run_count == 0
assert task.assistant_id == "lead_agent"
assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)
def test_create_stamps_the_bookkeeping_timestamps_from_now(self):
"""The factory already receives the clock as a rule input; reading it
a second time through the field defaults would give the aggregate two
slightly different construction instants -- and the stored row a third
if the adapter minted its own. One explicit `now`, one truth."""
task = ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec("0 9 * * *", "UTC"),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
now=NOW,
policy=NO_DELAY,
)
assert task.created_at == NOW
assert task.updated_at == NOW
def test_create_rejects_a_once_schedule_inside_the_delay_floor(self):
with pytest.raises(InvalidScheduleError, match="at least 60 seconds"):
ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=once_spec(after_seconds=10),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
now=NOW,
policy=MIN_60,
)
def test_skips_on_overlap_reflects_the_policy(self):
assert make_task().skips_on_overlap is True
# ---------------------------------------------------------------- F. the three truth tables
class TestStatusAfterLaunch:
@pytest.mark.parametrize(
("schedule", "trigger", "status", "expected"),
[
(once_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.RUNNING),
(once_spec(), TriggerKind.MANUAL, TaskStatus.ENABLED, TaskStatus.RUNNING),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.PAUSED, TaskStatus.PAUSED),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.ENABLED, TaskStatus.ENABLED),
(cron_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.ENABLED),
],
)
def test_truth_table(self, schedule, trigger, status, expected):
assert make_task(schedule=schedule, status=status).status_after_launch(trigger=trigger) is expected
def test_manual_trigger_of_paused_once_task_yields_running(self):
"""Decision-order trap: ONCE is tested before the MANUAL+PAUSED rule,
so the paused branch never sees a once task."""
task = make_task(schedule=once_spec(), status=TaskStatus.PAUSED)
assert task.status_after_launch(trigger=TriggerKind.MANUAL) is TaskStatus.RUNNING
class TestStatusAfterFailure:
@pytest.mark.parametrize(
("schedule", "trigger", "status", "expected"),
[
(once_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.FAILED),
(cron_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.ENABLED),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.ENABLED, TaskStatus.ENABLED),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.PAUSED, TaskStatus.PAUSED),
],
)
def test_truth_table(self, schedule, trigger, status, expected):
assert make_task(schedule=schedule, status=status).status_after_failure(trigger=trigger) is expected
@pytest.mark.parametrize("status", [TaskStatus.ENABLED, TaskStatus.PAUSED])
def test_failed_manual_trigger_of_once_task_keeps_status(self, status):
"""Decision-order trap: MANUAL is tested before ONCE, so a failed
manual trigger cannot burn the task's single scheduled occurrence."""
task = make_task(schedule=once_spec(), status=status)
assert task.status_after_failure(trigger=TriggerKind.MANUAL) is status
class TestStatusAfterSkip:
def test_once_is_failed_not_completed(self):
assert make_task(schedule=once_spec()).status_after_skip() is TaskStatus.FAILED
def test_cron_stays_enabled(self):
assert make_task(schedule=cron_spec()).status_after_skip() is TaskStatus.ENABLED
class TestStatusAfterCompletion:
@pytest.mark.parametrize("outcome", list(RunStatus))
def test_cron_never_changes_status(self, outcome):
assert make_task(schedule=cron_spec()).status_after_completion(outcome) is None
@pytest.mark.parametrize(
("outcome", "expected"),
[
(RunStatus.SUCCESS, TaskStatus.COMPLETED),
# INTERRUPTED is CANCELLED rather than FAILED: a user cancel or a
# same-thread takeover carries no execution failure.
(RunStatus.INTERRUPTED, TaskStatus.CANCELLED),
(RunStatus.FAILED, TaskStatus.FAILED),
],
)
def test_once_maps_each_terminal_outcome(self, outcome, expected):
assert make_task(schedule=once_spec()).status_after_completion(outcome) is expected
# ---------------------------------------------------------------- G. mutability gate
class TestMutabilityGate:
def test_ensure_mutable_rejects_a_running_task(self):
with pytest.raises(TaskNotMutableError, match="currently running"):
make_task(status=TaskStatus.RUNNING).ensure_mutable()
@pytest.mark.parametrize("status", [s for s in TaskStatus if s is not TaskStatus.RUNNING])
def test_ensure_mutable_allows_every_other_status(self, status):
make_task(status=status).ensure_mutable()
def test_pause_and_resume_are_gated(self):
running = make_task(status=TaskStatus.RUNNING)
with pytest.raises(TaskNotMutableError):
running.paused()
with pytest.raises(TaskNotMutableError):
running.resumed()
def test_pause_then_resume_round_trips(self):
task = make_task(status=TaskStatus.ENABLED)
assert task.paused().status is TaskStatus.PAUSED
assert task.paused().resumed().status is TaskStatus.ENABLED
def test_transitions_do_not_mutate_the_original(self):
task = make_task(status=TaskStatus.ENABLED)
task.paused()
assert task.status is TaskStatus.ENABLED
def test_transitions_leave_updated_at_to_the_repository(self):
"""The repository stamps `updated_at` on write (task.py:54-57). A second
clock read here would make the domain impure and a competing source of
truth for the same column."""
task = make_task(status=TaskStatus.ENABLED)
assert task.paused().updated_at == task.updated_at
assert task.paused().resumed().updated_at == task.updated_at
assert task.with_schedule(cron_spec("0 10 * * *"), now=NOW, policy=NO_DELAY).updated_at == task.updated_at
assert task.with_context(ContextMode.FRESH_THREAD_PER_RUN, None).updated_at == task.updated_at
# ---------------------------------------------------------------- H. with_schedule
class TestWithSchedule:
@pytest.mark.parametrize("status", [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED])
def test_terminal_task_is_rearmed_by_a_future_schedule(self, status):
"""claim_due only admits ENABLED rows, so leaving the terminal status
would return a next_run_at that silently never fires (router:203-208)."""
task = make_task(schedule=once_spec(), status=status)
updated = task.with_schedule(once_spec(after_seconds=7200), now=NOW, policy=NO_DELAY)
assert updated.status is TaskStatus.ENABLED
assert updated.next_run_at == NOW + timedelta(seconds=7200)
@pytest.mark.parametrize("status", [TaskStatus.ENABLED, TaskStatus.PAUSED])
def test_non_terminal_status_is_preserved(self, status):
updated = make_task(status=status).with_schedule(cron_spec("0 10 * * *"), now=NOW, policy=NO_DELAY)
assert updated.status is status
def test_running_task_cannot_be_rescheduled(self):
task = make_task(status=TaskStatus.RUNNING)
with pytest.raises(TaskNotMutableError):
task.with_schedule(cron_spec("0 10 * * *"), now=NOW, policy=NO_DELAY)
def test_invalid_once_schedule_propagates(self):
task = make_task(schedule=once_spec(), status=TaskStatus.COMPLETED)
with pytest.raises(InvalidScheduleError, match="must be in the future"):
task.with_schedule(ScheduleSpec.once_at(NOW - timedelta(hours=1), "UTC"), now=NOW, policy=NO_DELAY)
# ---------------------------------------------------------------- I. with_context
class TestWithContext:
def test_switching_to_fresh_thread_clears_the_thread(self):
task = make_task(context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1")
updated = task.with_context(ContextMode.FRESH_THREAD_PER_RUN, "thread-1")
assert updated.context_mode is ContextMode.FRESH_THREAD_PER_RUN
assert updated.thread_id is None
def test_switching_to_reuse_thread_without_a_thread_is_rejected(self):
with pytest.raises(InvalidContextModeError, match="reuse_thread requires thread_id"):
make_task().with_context(ContextMode.REUSE_THREAD, None)
def test_switching_to_reuse_thread_keeps_the_thread(self):
updated = make_task().with_context("reuse_thread", "thread-7")
assert updated.context_mode is ContextMode.REUSE_THREAD
assert updated.thread_id == "thread-7"
def test_running_task_cannot_change_context(self):
with pytest.raises(TaskNotMutableError):
make_task(status=TaskStatus.RUNNING).with_context(ContextMode.FRESH_THREAD_PER_RUN, None)
# ---------------------------------------------------------------- J. execution thread
class TestResolveExecutionThread:
def test_fresh_thread_mode_mints_a_new_thread_every_call(self):
"""Non-idempotent by design -- a dispatch must call this once and reuse
the value for the run row, the launch, and the result."""
task = make_task(context_mode=ContextMode.FRESH_THREAD_PER_RUN)
assert task.resolve_execution_thread() != task.resolve_execution_thread()
def test_reuse_thread_mode_returns_the_bound_thread(self):
task = make_task(context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1")
assert task.resolve_execution_thread() == "thread-1"
assert task.resolve_execution_thread() == "thread-1"
def test_reuse_thread_with_an_empty_thread_falls_back_to_a_fresh_one(self):
"""Unreachable for new aggregates (__post_init__ forbids it) but rows
predating that invariant can still carry this shape.
Asserting against the fresh-thread semantics rather than against `None`:
the fallback has to mint a real, distinct thread per call, which is what
the fresh-thread branch means.
"""
legacy = ScheduledTask.__new__(ScheduledTask)
object.__setattr__(legacy, "context_mode", ContextMode.REUSE_THREAD)
object.__setattr__(legacy, "thread_id", None)
first = ScheduledTask.resolve_execution_thread(legacy)
second = ScheduledTask.resolve_execution_thread(legacy)
assert uuid.UUID(first), "a real thread id, not a placeholder"
assert first != second, "a fresh thread per call, like FRESH_THREAD_PER_RUN"
# ---------------------------------------------------------------- K. ScheduledRun
class TestScheduledRun:
def test_queued_run_occupies_the_active_slot(self):
run = ScheduledRun.queued(task_id="task-1", thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)
assert run.status is RunStatus.QUEUED
assert run.is_active is True
assert run.record_id.startswith("task-run-")
def test_skipped_tombstone_is_terminal_and_never_queued(self):
"""QUEUED falls inside uq_scheduled_task_run_active's predicate and
would collide with the pre-existing run still holding the slot."""
run = ScheduledRun.skipped_tombstone(task_id="task-1", thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)
assert run.status is RunStatus.SKIPPED
assert run.is_active is False
def test_each_run_gets_a_distinct_identity(self):
first = ScheduledRun.queued(task_id="task-1", thread_id="t", scheduled_for=NOW, trigger=TriggerKind.MANUAL)
second = ScheduledRun.queued(task_id="task-1", thread_id="t", scheduled_for=NOW, trigger=TriggerKind.MANUAL)
assert first.record_id != second.record_id
@pytest.mark.parametrize(
("status", "active"),
[
(RunStatus.QUEUED, True),
(RunStatus.RUNNING, True),
(RunStatus.SUCCESS, False),
(RunStatus.FAILED, False),
(RunStatus.SKIPPED, False),
(RunStatus.INTERRUPTED, False),
],
)
def test_only_queued_and_running_occupy_the_active_slot(self, status, active):
"""Stated as behaviour over every status rather than as an equality
against ACTIVE_RUN_STATUSES -- restating the constant proves nothing,
since editing it would edit the assertion with it. The other half of
this rule, that the set matches the partial unique index's predicate,
cannot be checked from a dependency-free domain test and lives in
test_scheduled_task_models.py.
"""
run = replace(
ScheduledRun.queued(task_id="task-1", thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED),
status=status,
)
assert run.is_active is active

905
backend/tests/test_schedule_service.py Normal file
View File

@ -0,0 +1,905 @@
"""Use-case tests for ScheduleService, run entirely on in-memory doubles.
This file is the acceptance criterion of the hexagonal migration: the complete
scheduled-task lifecycle runs here with no HTTP, no database, and no run
runtime -- if any of that were still reachable from the domain, these tests
could not exist.
The dispatch cases mirror the pre-migration behaviour deliberately. Where a
comment says two paths must be indistinguishable, that is a regression the
tests are here to catch, not a description of the implementation.
"""
from __future__ import annotations
from dataclasses import replace
from datetime import UTC, datetime, timedelta
import pytest
from schedule_fakes import (
FakeRunLauncher,
FakeThreadLookup,
InMemoryScheduledRunRepository,
InMemoryScheduledTaskRepository,
)
from deerflow.domain.schedule.commands import (
UNSET,
ContextChange,
CreateScheduledTask,
DeleteTask,
PauseTask,
ResumeTask,
TriggerTask,
UnsetType,
UpdateScheduledTask,
)
from deerflow.domain.schedule.exceptions import (
InvalidScheduleError,
LaunchFailedError,
TaskNotFoundError,
TaskNotMutableError,
ThreadBusyError,
ThreadNotFoundError,
)
from deerflow.domain.schedule.model import (
ContextMode,
DispatchOutcome,
RunStatus,
SchedulePolicy,
ScheduleSpec,
TaskStatus,
TriggerKind,
)
from deerflow.domain.schedule.ports import RunOutcome
from deerflow.domain.schedule.service import ScheduleService
pytestmark = pytest.mark.asyncio
NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
CRON = ScheduleSpec.cron_schedule("0 9 * * *", "UTC")
POLICY = SchedulePolicy(min_once_delay_seconds=60, max_concurrent_runs=3, lease_seconds=120)
def once_spec(*, after_seconds: int = 3600) -> ScheduleSpec:
return ScheduleSpec.once_at(NOW + timedelta(seconds=after_seconds), "UTC")
class _BlindRunRepo(InMemoryScheduledRunRepository):
"""`has_active` always misses.
Reproduces the TOCTOU window the fast path cannot close: a concurrent
dispatch inserted its record after this one looked. The active-slot
rejection on `add` is then the only thing standing in the way.
"""
async def has_active(self, _task_id: str) -> bool:
return False
class _LaunchWritesMidReadTaskRepo(InMemoryScheduledTaskRepository):
"""`record_launch` commits between the completion hook's read and its write.
That ordering is not exotic -- `dispatch_task` performs its two bookkeeping
writes after the launch returns, and a run that fails fast reaches the hook
in between. Modelling it here rather than with real concurrency keeps the
window deterministic.
"""
def __init__(self, *, launch_next_run_at: datetime) -> None:
super().__init__()
self._launch_next_run_at = launch_next_run_at
self._armed = False
def arm(self) -> None:
"""Fire the interleaving on the next read, once."""
self._armed = True
async def get(self, task_id: str, *, user_id: str):
task = await super().get(task_id, user_id=user_id)
if self._armed:
self._armed = False
await self.record_launch(
task_id,
status=TaskStatus.ENABLED,
next_run_at=self._launch_next_run_at,
last_run_at=NOW,
last_run_id="run-1",
last_thread_id="thread-1",
last_error=None,
increment_run_count=True,
protect_terminal=True,
)
return task
def make_service(
*,
tasks: InMemoryScheduledTaskRepository | None = None,
runs: InMemoryScheduledRunRepository | None = None,
launcher: FakeRunLauncher | None = None,
threads: FakeThreadLookup | None = None,
policy: SchedulePolicy = POLICY,
) -> ScheduleService:
return ScheduleService(
tasks=tasks if tasks is not None else InMemoryScheduledTaskRepository(),
runs=runs if runs is not None else InMemoryScheduledRunRepository(),
launcher=launcher if launcher is not None else FakeRunLauncher(),
threads=threads if threads is not None else FakeThreadLookup(),
policy=policy,
)
async def create_cron_task(service: ScheduleService, *, schedule: ScheduleSpec = CRON):
return await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="Daily summary",
prompt="summarize",
schedule=schedule,
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
# ==================================================================== lifecycle
class TestFullLifecycle:
async def test_create_claim_dispatch_overlap_complete_pause_delete(self):
"""The acceptance criterion: the whole feature, zero IO.
Every step below is a real use case going through real domain rules --
only the four ports are doubles.
"""
tasks = InMemoryScheduledTaskRepository()
runs = InMemoryScheduledRunRepository()
launcher = FakeRunLauncher()
service = make_service(tasks=tasks, runs=runs, launcher=launcher)
# -- create ---------------------------------------------------------
task = await create_cron_task(service)
assert task.status is TaskStatus.ENABLED
assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)
# -- become due, get claimed and dispatched --------------------------
due_at = task.next_run_at + timedelta(seconds=1)
results = await service.run_once(now=due_at)
assert [r.outcome for r in results] == [DispatchOutcome.LAUNCHED]
assert len(launcher.calls) == 1
launch = launcher.calls[0]
assert launch["prompt"] == "summarize"
assert launch["owner_user_id"] == "user-1"
assert launch["metadata"]["scheduled_task_id"] == task.task_id
assert launch["metadata"]["scheduled_trigger"] == "scheduled"
after_launch = await service.get_task(task.task_id, user_id="user-1")
assert after_launch.status is TaskStatus.ENABLED, "a cron task stays claimable"
assert after_launch.run_count == 1
assert after_launch.last_run_id == "run-1"
assert after_launch.next_run_at > due_at
# -- next occurrence overlaps the still-running one ------------------
overlap_at = after_launch.next_run_at + timedelta(seconds=1)
overlapped = await service.run_once(now=overlap_at)
assert [r.outcome for r in overlapped] == [DispatchOutcome.SKIPPED]
assert len(launcher.calls) == 1, "the overlapping occurrence must not launch"
after_skip = await service.get_task(task.task_id, user_id="user-1")
assert after_skip.run_count == 1, "a skip is not an execution"
assert after_skip.last_run_id == after_launch.last_run_id, "bookkeeping carried over"
assert after_skip.status is TaskStatus.ENABLED
# Only a lost one-shot occurrence is worth surfacing; a cron task simply
# waits for its next turn, so the skip must not leave a user-visible
# error behind (service.py:455). The `once` half is covered by
# TestOnceTaskDispatch.test_a_skipped_once_task_is_failed_not_completed.
assert after_skip.last_error is None, "a routine cron overlap is not an error to report"
# -- the first run finally finishes ----------------------------------
record = next(r for r in runs.all_runs() if r.status is RunStatus.RUNNING)
await service.handle_run_completion(
RunOutcome(
task_id=task.task_id,
record_id=record.record_id,
run_id="run-1",
user_id="user-1",
status=RunStatus.SUCCESS,
error=None,
),
now=overlap_at,
)
assert await runs.count_active() == 0, "the active slot is free again"
# -- pause / resume ---------------------------------------------------
paused = await service.pause_task(PauseTask(task_id=task.task_id, user_id="user-1"))
assert paused.status is TaskStatus.PAUSED
assert await service.run_once(now=overlap_at + timedelta(days=2)) == [], "a paused task is not claimed"
resumed = await service.resume_task(ResumeTask(task_id=task.task_id, user_id="user-1"))
assert resumed.status is TaskStatus.ENABLED
# -- history and delete ----------------------------------------------
history = await service.list_task_runs(task.task_id, user_id="user-1")
assert sorted(run.status for run in history) == [RunStatus.SKIPPED, RunStatus.SUCCESS]
await service.delete_task(DeleteTask(task_id=task.task_id, user_id="user-1"))
with pytest.raises(TaskNotFoundError):
await service.get_task(task.task_id, user_id="user-1")
# ==================================================================== dispatch
class TestDispatchOutcomes:
async def test_launch_records_the_run_and_the_bookkeeping(self):
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await create_cron_task(service)
result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
assert result.outcome is DispatchOutcome.LAUNCHED
assert result.run_id == "run-1"
assert result.error is None
stored = runs.all_runs()[0]
assert stored.status is RunStatus.RUNNING
assert stored.run_id == "run-1"
assert stored.started_at == NOW
async def test_manual_trigger_against_an_active_run_is_a_conflict_with_no_record(self):
"""Nothing was scheduled to happen, so there is no occurrence to
account for -- the caller gets a conflict and the history stays clean."""
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await create_cron_task(service)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
before = len(runs.all_runs())
result = await service.trigger_task(TriggerTask(task_id=task.task_id, user_id="user-1"), now=NOW)
assert result.outcome is DispatchOutcome.CONFLICT
assert result.record_id is None
assert result.error == "task already has an active run"
assert len(runs.all_runs()) == before, "no history row for a rejected manual trigger"
async def test_scheduled_overlap_records_a_terminal_tombstone(self):
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await create_cron_task(service)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
assert result.outcome is DispatchOutcome.SKIPPED
assert result.error == "skipped: a previous run of this task is still active"
tombstone = next(r for r in runs.all_runs() if r.status is RunStatus.SKIPPED)
assert tombstone.is_active is False, "a queued tombstone would collide with the live run"
assert tombstone.started_at == tombstone.finished_at == NOW
async def test_launch_failure_is_recorded_as_failed(self):
runs = InMemoryScheduledRunRepository()
launcher = FakeRunLauncher(fail_with=LaunchFailedError("provider exploded"))
service = make_service(runs=runs, launcher=launcher)
task = await create_cron_task(service)
result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
assert result.outcome is DispatchOutcome.FAILED
assert result.run_id is None
assert "provider exploded" in result.error
assert runs.all_runs()[0].status is RunStatus.FAILED
async def test_a_failed_launch_replaces_the_previous_run_bookkeeping(self):
"""A failed launch is still an execution *attempt*, so it overwrites the
launch bookkeeping rather than carrying it over the way a skip does
(contrast `_finalize_skip`, which passes the current values back).
`record_launch` assigns every field unconditionally, so this is what
`last_run_id=None` in `_fail` actually means for a task that had already
run successfully: the id of the previous, unrelated run must not be left
pointing at a launch that never happened.
"""
launcher = FakeRunLauncher()
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs, launcher=launcher)
task = await create_cron_task(service)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
succeeded = await service.get_task(task.task_id, user_id="user-1")
assert succeeded.last_run_id == "run-1", "precondition: a successful launch was recorded"
# The slot has to be free, or the next dispatch is an overlap instead.
await service.handle_run_completion(
RunOutcome(
task_id=task.task_id,
record_id=runs.all_runs()[0].record_id,
run_id="run-1",
user_id="user-1",
status=RunStatus.SUCCESS,
error=None,
),
now=NOW,
)
launcher.fail_with = LaunchFailedError("provider exploded")
reloaded = await service.get_task(task.task_id, user_id="user-1")
await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED)
after = await service.get_task(task.task_id, user_id="user-1")
assert after.last_run_id is None, "no run started, so no run id to point at"
assert after.last_error == "provider exploded"
assert after.last_run_at == NOW, "the attempt itself is timestamped"
assert after.run_count == 1, "a failed launch is not a completed execution"
async def test_busy_thread_on_the_scheduled_path_degrades_to_a_skip(self):
launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
service = make_service(launcher=launcher)
task = await create_cron_task(service)
result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
assert result.outcome is DispatchOutcome.SKIPPED
async def test_busy_thread_on_a_manual_trigger_stays_a_conflict(self):
"""Not a skip: the user asked for this one, so it is reported rather
than quietly accounted for. The router maps it to 409, not 502."""
launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
service = make_service(launcher=launcher)
task = await create_cron_task(service)
result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL)
assert result.outcome is DispatchOutcome.CONFLICT
assert result.record_id is not None, "the attempt itself is still recorded"
class TestConflictCollapse:
"""The fast path and the active-slot rejection must be indistinguishable.
Two concurrent dispatches can both pass `has_active`; whichever loses is
rejected by the repository instead. A caller must not be able to tell which
of the two mechanisms stopped it, or retry behaviour diverges.
"""
async def _dispatch_second(self, run_repo, trigger: TriggerKind):
# reuse_thread so the execution thread is fixed: a fresh-thread task
# mints a new uuid per dispatch, which would make the two runs differ
# for a reason that has nothing to do with the collapse.
service = make_service(runs=run_repo, threads=FakeThreadLookup({"thread-1": "user-1"}))
task = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="t",
prompt="p",
schedule=CRON,
context_mode=ContextMode.REUSE_THREAD,
thread_id="thread-1",
),
now=NOW,
)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
return await service.dispatch_task(task, now=NOW, trigger=trigger)
@pytest.mark.parametrize("trigger", [TriggerKind.SCHEDULED, TriggerKind.MANUAL])
async def test_both_paths_produce_the_same_result(self, trigger):
via_fast_path = await self._dispatch_second(InMemoryScheduledRunRepository(), trigger)
via_slot_rejection = await self._dispatch_second(_BlindRunRepo(), trigger)
# record_id is freshly generated; every other field must match exactly.
assert replace(via_fast_path, record_id=None) == replace(via_slot_rejection, record_id=None)
async def test_the_losing_scheduled_dispatch_still_leaves_a_tombstone(self):
runs = _BlindRunRepo()
result = await self._dispatch_second(runs, TriggerKind.SCHEDULED)
assert result.outcome is DispatchOutcome.SKIPPED
assert any(r.status is RunStatus.SKIPPED for r in runs.all_runs())
async def test_the_losing_manual_dispatch_leaves_nothing(self):
runs = _BlindRunRepo()
result = await self._dispatch_second(runs, TriggerKind.MANUAL)
assert result.outcome is DispatchOutcome.CONFLICT
assert not any(r.status is RunStatus.SKIPPED for r in runs.all_runs())
class TestOnceTaskDispatch:
async def test_a_once_task_waits_in_running_for_its_completion(self):
"""Declaring it complete at launch would stick if the run failed or the
process died before the hook could correct it."""
service = make_service()
task = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="one shot",
prompt="go",
schedule=once_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
after = await service.get_task(task.task_id, user_id="user-1")
assert after.status is TaskStatus.RUNNING
async def test_a_skipped_once_task_is_failed_not_completed(self):
"""The single occurrence was lost; `completed` would claim an execution
that never happened."""
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="one shot",
prompt="go",
schedule=once_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
reloaded = await service.get_task(task.task_id, user_id="user-1")
await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED)
after = await service.get_task(task.task_id, user_id="user-1")
assert after.status is TaskStatus.FAILED
assert after.last_error == "skipped: a previous run of this task is still active"
# ==================================================================== run_once
class TestRunOnceBudget:
async def test_claims_nothing_when_the_global_budget_is_exhausted(self):
"""The cap is on active runs across all tasks, not on one poll's batch:
long runs accumulate across cycles."""
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs, policy=replace(POLICY, max_concurrent_runs=1))
first = await create_cron_task(service)
await service.dispatch_task(first, now=NOW, trigger=TriggerKind.SCHEDULED)
second = await create_cron_task(service)
results = await service.run_once(now=second.next_run_at + timedelta(seconds=1))
assert results == []
async def test_claims_only_into_the_remaining_budget(self):
tasks = InMemoryScheduledTaskRepository()
service = make_service(tasks=tasks, policy=replace(POLICY, max_concurrent_runs=2))
for _ in range(3):
await create_cron_task(service)
due_at = datetime(2026, 7, 28, 9, 0, 1, tzinfo=UTC)
results = await service.run_once(now=due_at)
assert len(results) == 2
assert all(r.outcome is DispatchOutcome.LAUNCHED for r in results)
async def test_the_claim_is_held_across_the_launch_and_released_after(self):
"""Claiming is what makes the task uneditable while it is being
dispatched, so the claim has to be live *at the moment of the launch* --
not merely taken at some point and released by the end.
The launch is the only place that ordering is observable, so the
launcher double reads the repository from inside `launch`. Asserting
only on the end state would pass even if the claim were taken after the
run had already started.
"""
tasks = InMemoryScheduledTaskRepository()
observed: dict = {}
class _ObservingLauncher(FakeRunLauncher):
async def launch(self, **kwargs):
task_id = kwargs["metadata"]["scheduled_task_id"]
observed["lease"] = tasks.lease_of(task_id)
observed["status"] = (await tasks.get(task_id, user_id="user-1")).status
return await super().launch(**kwargs)
service = make_service(tasks=tasks, launcher=_ObservingLauncher())
task = await create_cron_task(service)
await service.run_once(now=task.next_run_at + timedelta(seconds=1))
assert observed["status"] is TaskStatus.RUNNING, "the claim marks the task running before the launch"
assert observed["lease"][1] is not None, "and the lease is still held while the run starts"
assert tasks.lease_of(task.task_id) == (None, None), "released only once the dispatch is done"
# ==================================================================== completion
class TestRunCompletion:
async def _launched_once_task(self):
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="one shot",
prompt="go",
schedule=once_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
record = runs.all_runs()[0]
return service, task, record
@pytest.mark.parametrize(
("outcome_status", "expected"),
[
(RunStatus.SUCCESS, TaskStatus.COMPLETED),
(RunStatus.FAILED, TaskStatus.FAILED),
(RunStatus.INTERRUPTED, TaskStatus.CANCELLED),
],
)
async def test_once_task_terminal_mapping(self, outcome_status, expected):
service, task, record = await self._launched_once_task()
await service.handle_run_completion(
RunOutcome(
task_id=task.task_id,
record_id=record.record_id,
run_id="run-1",
user_id="user-1",
status=outcome_status,
error=None if outcome_status is RunStatus.SUCCESS else "boom",
),
now=NOW,
)
after = await service.get_task(task.task_id, user_id="user-1")
assert after.status is expected
async def test_a_cron_task_keeps_its_status_but_records_the_error(self):
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await create_cron_task(service)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
record = runs.all_runs()[0]
await service.handle_run_completion(
RunOutcome(
task_id=task.task_id,
record_id=record.record_id,
run_id="run-1",
user_id="user-1",
status=RunStatus.FAILED,
error="boom",
),
now=NOW,
)
after = await service.get_task(task.task_id, user_id="user-1")
assert after.status is TaskStatus.ENABLED, "the schedule outlives any single run"
assert after.last_error == "boom"
async def test_a_task_deleted_mid_flight_still_finalizes_its_run(self):
"""A task deleted while its run was in flight has nothing to write back,
and that is not an error -- but the *run* record must still be closed.
The hook writes the record first and only then reads the task, so
asserting on the record is what keeps that ordering honest; a test that
only checked "no exception" would stay green if the record write were
dropped entirely.
"""
runs = InMemoryScheduledRunRepository()
service = make_service(runs=runs)
task = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="one shot",
prompt="go",
schedule=once_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
record = runs.all_runs()[0]
await service.delete_task(DeleteTask(task_id=task.task_id, user_id="user-1"))
await service.handle_run_completion(
RunOutcome(
task_id=task.task_id,
record_id=record.record_id,
run_id="run-1",
user_id="user-1",
status=RunStatus.SUCCESS,
error=None,
),
now=NOW,
)
stored = runs.all_runs()[0]
assert stored.status is RunStatus.SUCCESS, "the execution record is finalized regardless"
assert stored.finished_at == NOW
assert await runs.count_active() == 0, "the active slot is freed"
async def test_a_cron_task_survives_a_launch_write_landing_mid_completion(self):
"""The interleaving `dispatch_task` already admits is possible.
A run that fails fast reaches this hook while the dispatch path has not
yet written its bookkeeping, so the hook reads a task still carrying the
elapsed `next_run_at`. If the write-back replays that whole snapshot,
the launch path's fresh fire time is rolled back and the task lands in
`running` with no live claim -- a shape neither `claim_due` branch nor
`cancel_stuck_once_tasks` can reach, i.e. permanently unschedulable.
"""
tasks = _LaunchWritesMidReadTaskRepo(launch_next_run_at=NOW + timedelta(days=1))
runs = InMemoryScheduledRunRepository()
service = make_service(tasks=tasks, runs=runs)
task = await create_cron_task(service)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
record = runs.all_runs()[0]
tasks.arm()
await service.handle_run_completion(
RunOutcome(
task_id=task.task_id,
record_id=record.record_id,
run_id="run-1",
user_id="user-1",
status=RunStatus.FAILED,
error="boom",
),
now=NOW,
)
after = await service.get_task(task.task_id, user_id="user-1")
assert after.last_error == "boom", "the completion still records its verdict"
assert after.next_run_at == NOW + timedelta(days=1), "the launch path's fire time must survive"
claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10)
assert [t.task_id for t in claimed] == [task.task_id], "the task must remain schedulable"
class TestReconcileOnStartup:
async def test_sweeps_orphaned_runs_and_stuck_once_tasks(self):
runs = InMemoryScheduledRunRepository()
tasks = InMemoryScheduledTaskRepository()
service = make_service(tasks=tasks, runs=runs)
task = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="one shot",
prompt="go",
schedule=once_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
# The process dies here: the run is active and the once task is parked
# in running with its claim already released.
stale_runs, stuck_tasks = await service.reconcile_on_startup(error="gateway restarted")
assert (stale_runs, stuck_tasks) == (1, 1)
assert runs.all_runs()[0].status is RunStatus.INTERRUPTED
after = await service.get_task(task.task_id, user_id="user-1")
assert after.status is TaskStatus.CANCELLED
# ==================================================================== CRUD
class TestTaskManagement:
async def test_reuse_thread_requires_an_accessible_thread(self):
service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"}))
created = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="t",
prompt="p",
schedule=CRON,
context_mode=ContextMode.REUSE_THREAD,
thread_id="thread-1",
),
now=NOW,
)
assert created.thread_id == "thread-1"
with pytest.raises(ThreadNotFoundError):
await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-2",
title="t",
prompt="p",
schedule=CRON,
context_mode=ContextMode.REUSE_THREAD,
thread_id="thread-1",
),
now=NOW,
)
async def test_an_invalid_schedule_is_rejected_before_any_write(self):
tasks = InMemoryScheduledTaskRepository()
service = make_service(tasks=tasks)
with pytest.raises(InvalidScheduleError):
await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="t",
prompt="p",
schedule=once_spec(after_seconds=10), # inside min_once_delay
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
),
now=NOW,
)
assert await service.list_tasks("user-1") == []
async def test_commands_are_dumb_data(self):
# A command carries intent without validating it: business rules stay
# on the aggregate, so error attribution (a malformed schedule before
# an unknown thread) is owned by the handler's construction order,
# not by the command's own constructor.
cmd = CreateScheduledTask(user_id="u", title="", prompt="", schedule=CRON, context_mode="not-a-mode", thread_id=None)
assert cmd.context_mode == "not-a-mode"
async def test_unset_is_a_singleton_distinct_from_none(self):
# Three states, not two: an update field is UNSET (leave it alone),
# None can stay a meaningful value elsewhere, and UNSET is falsy so
# it cannot masquerade as a supplied value.
assert UnsetType() is UNSET
cmd = UpdateScheduledTask(task_id="t", user_id="u")
assert cmd.title is UNSET
assert cmd.title is not None
assert not UNSET
async def test_update_leaves_omitted_fields_alone(self):
service = make_service()
task = await create_cron_task(service)
updated = await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", title="renamed"), now=NOW)
assert updated.title == "renamed"
assert updated.prompt == task.prompt
assert updated.schedule == task.schedule
async def test_update_with_everything_unset_changes_nothing(self):
service = make_service()
task = await create_cron_task(service)
updated = await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1"), now=NOW)
assert updated == task
async def test_update_can_change_the_prompt(self):
service = make_service()
task = await create_cron_task(service)
updated = await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", prompt="new instructions"), now=NOW)
assert updated.prompt == "new instructions"
assert updated.title == task.title
async def test_a_task_deleted_mid_update_reports_as_missing(self):
"""get_task saw it, save no longer does -- a concurrent delete landed
in between. The caller gets the same not-found it would have got a
moment earlier, rather than a None leaking out."""
class _VanishingRepo(InMemoryScheduledTaskRepository):
async def save(self, _task):
return None
tasks = _VanishingRepo()
service = make_service(tasks=tasks)
task = await create_cron_task(service)
with pytest.raises(TaskNotFoundError):
await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", title="x"), now=NOW)
with pytest.raises(TaskNotFoundError):
await service.pause_task(PauseTask(task_id=task.task_id, user_id="user-1"))
async def test_context_is_changed_through_the_packaged_value(self):
"""context_mode and thread_id move together, so they are supplied
together -- which is what lets every other update field use plain
`None` for "not supplied"."""
service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"}))
task = await create_cron_task(service)
bound = await service.update_scheduled_task(
UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.REUSE_THREAD, "thread-1")),
now=NOW,
)
assert bound.context_mode is ContextMode.REUSE_THREAD
assert bound.thread_id == "thread-1"
unbound = await service.update_scheduled_task(
UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.FRESH_THREAD_PER_RUN)),
now=NOW,
)
assert unbound.thread_id is None, "switching to a fresh thread clears the binding"
async def test_changing_context_to_an_inaccessible_thread_is_rejected(self):
service = make_service(threads=FakeThreadLookup({"thread-1": "someone-else"}))
task = await create_cron_task(service)
with pytest.raises(ThreadNotFoundError):
await service.update_scheduled_task(
UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.REUSE_THREAD, "thread-1")),
now=NOW,
)
async def test_rescheduling_a_terminal_task_re_arms_it(self):
tasks = InMemoryScheduledTaskRepository()
service = make_service(tasks=tasks)
task = await create_cron_task(service)
tasks.seed(replace(task, status=TaskStatus.FAILED))
updated = await service.update_scheduled_task(
UpdateScheduledTask(task_id=task.task_id, user_id="user-1", schedule=ScheduleSpec.cron_schedule("0 10 * * *", "UTC")),
now=NOW,
)
assert updated.status is TaskStatus.ENABLED, "otherwise it would never be claimed again"
async def test_a_running_task_cannot_be_edited(self):
tasks = InMemoryScheduledTaskRepository()
service = make_service(tasks=tasks)
task = await create_cron_task(service)
tasks.seed(replace(task, status=TaskStatus.RUNNING))
with pytest.raises(TaskNotMutableError):
await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", title="nope"), now=NOW)
with pytest.raises(TaskNotMutableError):
await service.pause_task(PauseTask(task_id=task.task_id, user_id="user-1"))
async def test_a_running_task_can_still_be_deleted(self):
"""The pre-migration router gated update/pause/resume on this, but not
delete -- that asymmetry is preserved."""
tasks = InMemoryScheduledTaskRepository()
service = make_service(tasks=tasks)
task = await create_cron_task(service)
tasks.seed(replace(task, status=TaskStatus.RUNNING))
await service.delete_task(DeleteTask(task_id=task.task_id, user_id="user-1"))
@pytest.mark.parametrize("call", ["get", "update", "pause", "resume", "delete", "runs"])
async def test_another_users_task_is_reported_as_missing(self, call):
service = make_service()
task = await create_cron_task(service)
with pytest.raises(TaskNotFoundError):
if call == "get":
await service.get_task(task.task_id, user_id="intruder")
elif call == "update":
await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="intruder", title="x"), now=NOW)
elif call == "pause":
await service.pause_task(PauseTask(task_id=task.task_id, user_id="intruder"))
elif call == "resume":
await service.resume_task(ResumeTask(task_id=task.task_id, user_id="intruder"))
elif call == "delete":
await service.delete_task(DeleteTask(task_id=task.task_id, user_id="intruder"))
else:
await service.list_task_runs(task.task_id, user_id="intruder")
async def test_tasks_are_listed_per_user_and_per_thread(self):
service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"}))
bound = await service.create_scheduled_task(
CreateScheduledTask(
user_id="user-1",
title="bound",
prompt="p",
schedule=CRON,
context_mode=ContextMode.REUSE_THREAD,
thread_id="thread-1",
),
now=NOW,
)
await create_cron_task(service)
assert len(await service.list_tasks("user-1")) == 2
assert await service.list_tasks("someone-else") == []
by_thread = await service.list_tasks_by_thread("user-1", "thread-1")
assert [t.task_id for t in by_thread] == [bound.task_id]