From d856ae8573aaf640a168c4b1991b2ae4daf07dc9 Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Fri, 31 Jul 2026 11:43:35 +0800 Subject: [PATCH] feat(schedule): add the domain model, ports, and application service 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 --- .../deerflow/domain/schedule/__init__.py | 83 ++ .../deerflow/domain/schedule/commands.py | 130 +++ .../deerflow/domain/schedule/exceptions.py | 67 ++ .../domain/schedule/model/__init__.py | 51 + .../deerflow/domain/schedule/model/enums.py | 94 ++ .../deerflow/domain/schedule/model/run.py | 97 ++ .../deerflow/domain/schedule/model/spec.py | 191 ++++ .../deerflow/domain/schedule/model/task.py | 289 ++++++ .../harness/deerflow/domain/schedule/ports.py | 332 +++++++ .../deerflow/domain/schedule/service.py | 462 +++++++++ backend/tests/schedule_fakes.py | 293 ++++++ backend/tests/test_schedule_domain.py | 627 ++++++++++++ backend/tests/test_schedule_service.py | 905 ++++++++++++++++++ 13 files changed, 3621 insertions(+) create mode 100644 backend/packages/harness/deerflow/domain/schedule/__init__.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/commands.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/exceptions.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/model/__init__.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/model/enums.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/model/run.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/model/spec.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/model/task.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/ports.py create mode 100644 backend/packages/harness/deerflow/domain/schedule/service.py create mode 100644 backend/tests/schedule_fakes.py create mode 100644 backend/tests/test_schedule_domain.py create mode 100644 backend/tests/test_schedule_service.py diff --git a/backend/packages/harness/deerflow/domain/schedule/__init__.py b/backend/packages/harness/deerflow/domain/schedule/__init__.py new file mode 100644 index 000000000..f673e972e --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/__init__.py @@ -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", +] diff --git a/backend/packages/harness/deerflow/domain/schedule/commands.py b/backend/packages/harness/deerflow/domain/schedule/commands.py new file mode 100644 index 000000000..75cf4f09b --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/commands.py @@ -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 diff --git a/backend/packages/harness/deerflow/domain/schedule/exceptions.py b/backend/packages/harness/deerflow/domain/schedule/exceptions.py new file mode 100644 index 000000000..d07c3eb36 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/exceptions.py @@ -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.""" diff --git a/backend/packages/harness/deerflow/domain/schedule/model/__init__.py b/backend/packages/harness/deerflow/domain/schedule/model/__init__.py new file mode 100644 index 000000000..c8f3e2d44 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/__init__.py @@ -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", +] diff --git a/backend/packages/harness/deerflow/domain/schedule/model/enums.py b/backend/packages/harness/deerflow/domain/schedule/model/enums.py new file mode 100644 index 000000000..d54846fd2 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/enums.py @@ -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" diff --git a/backend/packages/harness/deerflow/domain/schedule/model/run.py b/backend/packages/harness/deerflow/domain/schedule/model/run.py new file mode 100644 index 000000000..373387f6e --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/run.py @@ -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 diff --git a/backend/packages/harness/deerflow/domain/schedule/model/spec.py b/backend/packages/harness/deerflow/domain/schedule/model/spec.py new file mode 100644 index 000000000..4fe54ff93 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/spec.py @@ -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 diff --git a/backend/packages/harness/deerflow/domain/schedule/model/task.py b/backend/packages/harness/deerflow/domain/schedule/model/task.py new file mode 100644 index 000000000..bf5b6e848 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/task.py @@ -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) diff --git a/backend/packages/harness/deerflow/domain/schedule/ports.py b/backend/packages/harness/deerflow/domain/schedule/ports.py new file mode 100644 index 000000000..931485144 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/ports.py @@ -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. + """ + ... diff --git a/backend/packages/harness/deerflow/domain/schedule/service.py b/backend/packages/harness/deerflow/domain/schedule/service.py new file mode 100644 index 000000000..7f31d37be --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -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) diff --git a/backend/tests/schedule_fakes.py b/backend/tests/schedule_fakes.py new file mode 100644 index 000000000..4e7fae5f3 --- /dev/null +++ b/backend/tests/schedule_fakes.py @@ -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 diff --git a/backend/tests/test_schedule_domain.py b/backend/tests/test_schedule_domain.py new file mode 100644 index 000000000..302b6eed3 --- /dev/null +++ b/backend/tests/test_schedule_domain.py @@ -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 diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py new file mode 100644 index 000000000..1890b0d97 --- /dev/null +++ b/backend/tests/test_schedule_service.py @@ -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]