From ff9c56d2cd653112a234e95bc4afaa6e1da2a5ae Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Tue, 28 Jul 2026 11:38:04 +0800 Subject: [PATCH] feat(schedule): add the schedule application service The input port of the context: every scheduled-task use case, orchestrated over the four output ports. It holds no business rules -- each decision is delegated to the aggregate -- and `test_schedule_service.py` runs the complete lifecycle with no HTTP, no database and no run runtime, which is the acceptance criterion this migration was for. `dispatch_task` mirrors the pre-migration structure line for line, including its comments, and makes exactly three substitutions: bare dicts become domain objects, the HTTPException-409 sniffing becomes `except ThreadBusyError`, and the status-derivation static methods become aggregate methods. Its four exits and their two conflict paths are unchanged, and the tests pin the collapse: the fast-path rejection and the active-slot rejection must produce identical results, asserted over every field of DispatchResult. One behavioural narrowing, deliberate. The old code wrapped the launch *and* its follow-up writes in `except Exception`, so a failing bookkeeping write was recorded as a failed launch -- marking an execution that had actually started as failed. The port contract admits exactly two escapes from `launch`, so only the launch is guarded now and a genuine write fault propagates instead of being misreported. SchedulePolicy gains max_concurrent_runs and lease_seconds. Both are operator-tunable thresholds the domain needs but must not read, which is what that value object is for; the claiming process's identity stays a constructor argument since it is an identity, not a threshold. --- .../deerflow/domain/schedule/model/spec.py | 21 +- .../deerflow/domain/schedule/service.py | 456 +++++++++++++ backend/tests/schedule_fakes.py | 3 +- backend/tests/test_schedule_service.py | 630 ++++++++++++++++++ 4 files changed, 1108 insertions(+), 2 deletions(-) create mode 100644 backend/packages/harness/deerflow/domain/schedule/service.py create mode 100644 backend/tests/test_schedule_service.py diff --git a/backend/packages/harness/deerflow/domain/schedule/model/spec.py b/backend/packages/harness/deerflow/domain/schedule/model/spec.py index 3176d2dcb..3c966b81d 100644 --- a/backend/packages/harness/deerflow/domain/schedule/model/spec.py +++ b/backend/packages/harness/deerflow/domain/schedule/model/spec.py @@ -14,9 +14,28 @@ CRON_FIELD_COUNT = 5 @dataclass(frozen=True) class SchedulePolicy: - """Operator-tunable thresholds the domain needs but must not read itself.""" + """Operator-tunable thresholds the domain needs but must not read itself. + + Built by the composition root from the scheduler configuration and passed + in. Deliberately not held by any aggregate: a task whose meaning changes + with deployment config is not a domain object. + + The defaults are the permissive ones on purpose -- "nobody configured a + policy" must not invent a business constraint. Real values only ever + arrive from the outer ring. + """ min_once_delay_seconds: int = 0 + """How far ahead a one-shot schedule must be at submission time. Read by + `ScheduleSpec.ensure_launchable`; a cron schedule is never subject to it.""" + + max_concurrent_runs: int = 1 + """Ceiling on active scheduled executions across ALL tasks. Long runs + accumulate across polls, so each poll may only claim into what is left.""" + + lease_seconds: int = 60 + """How long a claim on a task stays valid. Bounds how quickly a task + orphaned between claim and dispatch becomes reachable again.""" @dataclass(frozen=True) 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..52fb3f98d --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -0,0 +1,456 @@ +"""Application service of the schedule context (its input port). + +Orchestrates use cases only: fetch, apply domain rules, persist through output +ports. It holds no business rules itself -- every decision below is delegated +to the aggregate -- and knows nothing about HTTP, SQL, or the run runtime. +`user_id` is always passed in explicitly; resolving the current user is the +primary adapter's job. + +The dispatch path deliberately mirrors the structure of the pre-migration +`app/scheduler/service.py`, including its ordering and its comments, because +its concurrency and idempotency semantics are load-bearing and are pinned by +tests that must keep passing unchanged. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, replace +from datetime import datetime +from typing import Any, Literal + +from deerflow.domain.schedule.model import ( + ActiveRunConflictError, + ContextMode, + LaunchFailedError, + RunStatus, + ScheduledRun, + ScheduledTask, + SchedulePolicy, + ScheduleSpec, + ScheduleType, + TaskNotFoundError, + ThreadBusyError, + ThreadNotFoundError, + TriggerKind, +) +from deerflow.domain.schedule.ports import ( + RunLauncher, + RunOutcome, + ScheduledRunRepository, + ScheduledTaskRepository, + ThreadLookup, +) + +# Shared so the has_active fast path and the active-slot race path produce +# byte-identical outcomes for the same "task already has an active run" +# condition. Two callers must not be able to tell which one rejected them. +_ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run" +_SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active" + +_UNSET: Any = object() +"""Distinguishes "field omitted" from "field set to None" in update_task.""" + + +@dataclass(frozen=True) +class DispatchResult: + """What one dispatch attempt produced. + + `outcome` drives the caller's protocol mapping -- a manual trigger turns + `conflict` into 409 and `failed` into 502 -- so the four values are part of + the contract, not an implementation detail. + """ + + outcome: Literal["launched", "skipped", "conflict", "failed"] + record_id: str | None + run_id: str | None + thread_id: str + error: str | None + + +class ScheduleService: + def __init__( + self, + *, + tasks: ScheduledTaskRepository, + runs: ScheduledRunRepository, + launcher: RunLauncher, + threads: ThreadLookup, + policy: SchedulePolicy, + lease_owner: str, + ) -> None: + self._tasks = tasks + self._runs = runs + self._launcher = launcher + self._threads = threads + self._policy = policy + # Identity of the claiming process. Recorded for diagnostics only -- + # the repository never reads it back. + self._lease_owner = lease_owner + + # ------------------------------------------------------------------ reads + + async def list_tasks(self, user_id: str) -> list[ScheduledTask]: + return await self._tasks.list_by_user(user_id) + + async def list_tasks_by_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]: + return await self._tasks.list_by_user_and_thread(user_id, thread_id) + + async def get_task(self, task_id: str, *, user_id: str) -> ScheduledTask: + """Raises TaskNotFoundError when absent or owned by someone else -- the + caller must not be able to tell those apart.""" + task = await self._tasks.get(task_id, user_id=user_id) + if task is None: + raise TaskNotFoundError("Scheduled task not found") + return task + + async def list_task_runs(self, task_id: str, *, user_id: str, limit: int = 50, offset: int = 0) -> list[ScheduledRun]: + """Execution history, gated on ownership of the parent task.""" + await self.get_task(task_id, user_id=user_id) + return await self._runs.list_by_task(task_id, limit=limit, offset=offset) + + # ------------------------------------------------------------------ writes + + async def create_task( + self, + *, + user_id: str, + title: str, + prompt: str, + schedule: ScheduleSpec, + context_mode: str | ContextMode, + thread_id: str | None, + now: datetime, + ) -> ScheduledTask: + """Register a new standing instruction. + + The aggregate is built first so a malformed schedule or context mode is + reported as such before any IO happens; only then is the thread + binding verified. + + Raises: + InvalidScheduleError / InvalidContextModeError: from the aggregate. + ThreadNotFoundError: reuse_thread pointing at a thread the user + cannot use. + """ + task = ScheduledTask.create( + user_id=user_id, + title=title, + prompt=prompt, + schedule=schedule, + context_mode=context_mode, + thread_id=thread_id, + now=now, + policy=self._policy, + ) + await self._require_thread(task) + return await self._tasks.add(task) + + async def update_task( + self, + task_id: str, + *, + user_id: str, + now: datetime, + title: str = _UNSET, + prompt: str = _UNSET, + schedule: ScheduleSpec = _UNSET, + context_mode: str | ContextMode = _UNSET, + thread_id: str | None = _UNSET, + ) -> ScheduledTask: + """Partially update a task; omitted fields are left alone. + + Context and schedule are applied through the aggregate's own + transitions, so the re-arm rule and the running-task gate cannot be + bypassed by patching fields directly. + """ + task = await self.get_task(task_id, user_id=user_id) + task.ensure_mutable() + + if context_mode is not _UNSET or thread_id is not _UNSET: + effective_mode = task.context_mode if context_mode is _UNSET else context_mode + effective_thread = task.thread_id if thread_id is _UNSET else thread_id + task = task.with_context(effective_mode, effective_thread) + await self._require_thread(task) + + if schedule is not _UNSET: + task = task.with_schedule(schedule, now=now, policy=self._policy) + + if title is not _UNSET: + task = replace(task, title=title) + if prompt is not _UNSET: + task = replace(task, prompt=prompt) + + saved = await self._tasks.save(task) + if saved is None: + raise TaskNotFoundError("Scheduled task not found") + return saved + + async def pause_task(self, task_id: str, *, user_id: str) -> ScheduledTask: + return await self._transition(task_id, user_id=user_id, transition=ScheduledTask.paused) + + async def resume_task(self, task_id: str, *, user_id: str) -> ScheduledTask: + return await self._transition(task_id, user_id=user_id, transition=ScheduledTask.resumed) + + async def delete_task(self, task_id: str, *, user_id: str) -> None: + """Deleting is deliberately not gated on the task being idle: the + pre-migration router applied that gate to update/pause/resume only.""" + if not await self._tasks.delete(task_id, user_id=user_id): + raise TaskNotFoundError("Scheduled task not found") + + # ------------------------------------------------------------------ dispatch + + async def trigger_task(self, task_id: str, *, user_id: str, now: datetime) -> DispatchResult: + """Dispatch a task on demand. Unlike the scheduled path this is allowed + while the task is paused, and leaves it paused.""" + task = await self.get_task(task_id, user_id=user_id) + return await self.dispatch_task(task, now=now, trigger=TriggerKind.MANUAL) + + async def run_once(self, *, now: datetime) -> list[DispatchResult]: + """Claim whatever is due and dispatch it. + + `max_concurrent_runs` is a global cap on active scheduled runs, not a + per-poll batch size: long runs accumulate across poll cycles, so each + cycle only claims into the remaining budget. + """ + active = await self._runs.count_active() + budget = self._policy.max_concurrent_runs - active + if budget <= 0: + return [] + claimed = await self._tasks.claim_due( + now=now, + lease_owner=self._lease_owner, + lease_seconds=self._policy.lease_seconds, + limit=budget, + ) + return [await self.dispatch_task(task, now=now, trigger=TriggerKind.SCHEDULED) for task in claimed] + + async def dispatch_task(self, task: ScheduledTask, *, now: datetime, trigger: TriggerKind) -> DispatchResult: + """Turn one due task into one execution. + + Called once per dispatch, so `resolve_execution_thread` is called once + and its value reused for the record, the launch, and the result. + """ + execution_thread_id = task.resolve_execution_thread() + + # "skip" must hold for fresh-thread runs too, where every run gets a + # new thread and the same-thread busy signal below can never fire. + # Checked before creating this dispatch's own record so the record does + # not count itself as the active run. A manual trigger against an + # active run is rejected outright instead of being recorded as a + # skipped occurrence -- nothing was scheduled to happen. + # + # This check is a NON-ATOMIC fast path: two concurrent dispatches (a + # manual trigger racing the poller, a double-click, a client retry) can + # both observe no active run. The repository is the atomic arbiter -- + # it rejects the second active record with ActiveRunConflictError, + # which collapses to the SAME outcome as this fast path just below. + if task.skips_on_overlap and await self._runs.has_active(task.task_id): + if trigger is TriggerKind.MANUAL: + return self._conflict(execution_thread_id) + return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger) + + record = ScheduledRun.queued( + task_id=task.task_id, + thread_id=execution_thread_id, + scheduled_for=now, + trigger=trigger, + ) + try: + await self._runs.add(record) + except ActiveRunConflictError: + # Lost the race for the task's single active slot: a concurrent + # dispatch passed the same fast-path check and inserted first. + # Identical outcome to the fast path above -- that equality is what + # the dispatch-race regression tests pin. + if trigger is TriggerKind.MANUAL: + return self._conflict(execution_thread_id) + return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger) + + # Only the launch is guarded. The port contract admits exactly two + # escapes, so a failure of the bookkeeping writes below is a genuine + # fault and propagates instead of being recorded as a failed launch -- + # which would mark an already-running execution as failed. + try: + launched = await self._launcher.launch( + thread_id=execution_thread_id, + assistant_id=task.assistant_id, + prompt=task.prompt, + owner_user_id=task.user_id, + metadata={ + "scheduled_task_id": task.task_id, + "scheduled_task_run_id": record.record_id, + "scheduled_trigger": str(trigger), + }, + ) + except ThreadBusyError as exc: + # The execution thread is already busy. On the scheduled path under + # a skip policy this is an overlap like any other; anything else is + # reported as a conflict the caller has to deal with. + if trigger is TriggerKind.SCHEDULED and task.skips_on_overlap: + return await self._finalize_skip(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, error=str(exc)) + return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome="conflict") + except LaunchFailedError as exc: + return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome="failed") + + await self._runs.update_status( + record.record_id, + status=RunStatus.RUNNING, + run_id=launched.run_id, + started_at=now, + # A fast-failing run can reach handle_run_completion before this + # write lands; never clobber its terminal verdict. + protect_terminal=True, + ) + await self._tasks.record_launch( + task.task_id, + status=task.status_after_launch(trigger=trigger), + next_run_at=task.schedule.next_after(now), + last_run_at=now, + last_run_id=launched.run_id, + last_thread_id=launched.thread_id, + last_error=None, + increment_run_count=True, + # Same race as the record write above. + protect_terminal=True, + ) + return DispatchResult("launched", record.record_id, launched.run_id, launched.thread_id, None) + + # ------------------------------------------------------------------ lifecycle + + async def handle_run_completion(self, outcome: RunOutcome, *, now: datetime) -> None: + """Write back a launched run's terminal verdict. + + A task deleted while its run was in flight simply has nothing to + update, and that is not an error. + """ + await self._runs.update_status( + outcome.record_id, + status=outcome.status, + run_id=outcome.run_id, + error=outcome.error, + finished_at=now, + ) + task = await self._tasks.get(outcome.task_id, user_id=outcome.user_id) + if task is None: + return + # The error is recorded whether or not the status moves: a cron task + # keeps its schedule but still reports what went wrong last time. + updated = replace(task, last_error=outcome.error) + new_status = task.status_after_completion(outcome.status) + if new_status is not None: + updated = replace(updated, status=new_status) + await self._tasks.save(updated) + + async def reconcile_on_startup(self, *, error: str) -> tuple[int, int]: + """Clean up what a process crash left behind, returning what was fixed. + + Two sweeps, because a crash strands two different things: execution + records that can never finish, and `once` tasks parked waiting for a + completion hook that died with the process. The second is not covered + by expired-claim reclaim -- a launched task released its claim, so the + claim query can never see it again. + + Failures propagate: whether a partial reconcile should block startup is + the caller's policy, not the domain's. + """ + stale_runs = await self._runs.mark_stale_active(error=error) + stuck_tasks = await self._tasks.cancel_stuck_once_tasks(error=error) + return stale_runs, stuck_tasks + + # ------------------------------------------------------------------ internals + + async def _require_thread(self, task: ScheduledTask) -> None: + if task.context_mode is not ContextMode.REUSE_THREAD: + return + # The aggregate guarantees a thread here; the check keeps that + # guarantee from being silently dropped under `python -O`. + if not task.thread_id or not await self._threads.exists_for_user(task.thread_id, task.user_id): + raise ThreadNotFoundError("Thread not found") + + async def _transition(self, task_id: str, *, user_id: str, transition: Callable[[ScheduledTask], ScheduledTask]) -> ScheduledTask: + task = await self.get_task(task_id, user_id=user_id) + saved = await self._tasks.save(transition(task)) + if saved is None: + raise TaskNotFoundError("Scheduled task not found") + return saved + + def _conflict(self, thread_id: str) -> DispatchResult: + """A manual trigger against an active run. + + No history record is written: nothing was scheduled to happen, so there + is no occurrence to account for. + """ + return DispatchResult("conflict", None, None, thread_id, _ACTIVE_RUN_CONFLICT_ERROR) + + async def _record_scheduled_skip(self, task: ScheduledTask, *, thread_id: str, now: datetime, trigger: TriggerKind) -> DispatchResult: + """Account for a scheduled occurrence dropped because of an overlap. + + The tombstone is created directly terminal rather than as the transient + queued record the launch path uses: a queued record is active and would + itself be refused against the pre-existing run that is still holding + the task's single active slot. + """ + record = ScheduledRun.skipped_tombstone( + task_id=task.task_id, + thread_id=thread_id, + scheduled_for=now, + trigger=trigger, + ) + await self._runs.add(record) + return await self._finalize_skip(task, record_id=record.record_id, thread_id=thread_id, now=now, error=_SKIP_ACTIVE_RUN_ERROR) + + async def _finalize_skip(self, task: ScheduledTask, *, record_id: str, thread_id: str, now: datetime, error: str) -> DispatchResult: + await self._runs.update_status( + record_id, + status=RunStatus.SKIPPED, + error=error, + started_at=now, + finished_at=now, + ) + await self._tasks.record_launch( + task.task_id, + status=task.status_after_skip(), + next_run_at=task.schedule.next_after(now), + # A skip is not an execution, so the launch bookkeeping carries + # over unchanged; record_launch assigns unconditionally, so + # "unchanged" has to be spelled out. + last_run_at=task.last_run_at, + last_run_id=task.last_run_id, + last_thread_id=task.last_thread_id, + # Only a lost one-shot occurrence is worth surfacing; a cron task + # simply waits for its next turn. + last_error=error if task.schedule.schedule_type is ScheduleType.ONCE else None, + increment_run_count=False, + ) + return DispatchResult("skipped", record_id, None, thread_id, error) + + async def _fail( + self, + task: ScheduledTask, + *, + record_id: str, + thread_id: str, + now: datetime, + trigger: TriggerKind, + error: str, + outcome: Literal["conflict", "failed"], + ) -> DispatchResult: + await self._runs.update_status( + record_id, + status=RunStatus.FAILED, + error=error, + started_at=now, + finished_at=now, + ) + await self._tasks.record_launch( + task.task_id, + status=task.status_after_failure(trigger=trigger), + next_run_at=task.schedule.next_after(now), + last_run_at=now, + last_run_id=None, + last_thread_id=thread_id, + last_error=error, + increment_run_count=False, + ) + return DispatchResult(outcome, record_id, None, thread_id, error) diff --git a/backend/tests/schedule_fakes.py b/backend/tests/schedule_fakes.py index 3d1ffdbb9..268f7b29e 100644 --- a/backend/tests/schedule_fakes.py +++ b/backend/tests/schedule_fakes.py @@ -25,6 +25,7 @@ from deerflow.domain.schedule.model import ( RunStatus, ScheduledRun, ScheduledTask, + ScheduleType, TaskStatus, ) from deerflow.domain.schedule.ports import LaunchedRun @@ -155,7 +156,7 @@ class InMemoryScheduledTaskRepository: async def cancel_stuck_once_tasks(self, *, error: str) -> int: cancelled = 0 for row in self._rows.values(): - stuck = row.task.status is TaskStatus.RUNNING and row.task.schedule.schedule_type.value == "once" and row.lease_expires_at is None + stuck = row.task.status is TaskStatus.RUNNING and row.task.schedule.schedule_type is ScheduleType.ONCE and row.lease_expires_at is None if stuck: row.task = replace(row.task, status=TaskStatus.CANCELLED, last_error=error) cancelled += 1 diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py new file mode 100644 index 000000000..e833390ea --- /dev/null +++ b/backend/tests/test_schedule_service.py @@ -0,0 +1,630 @@ +"""Use-case tests for ScheduleService, run entirely on in-memory doubles. + +This file is the acceptance criterion of the hexagonal migration: the complete +scheduled-task lifecycle runs here with no HTTP, no database, and no run +runtime -- if any of that were still reachable from the domain, these tests +could not exist. + +The dispatch cases mirror the pre-migration behaviour deliberately. Where a +comment says two paths must be indistinguishable, that is a regression the +tests are here to catch, not a description of the implementation. +""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta + +import pytest +from schedule_fakes import ( + FakeRunLauncher, + FakeThreadLookup, + InMemoryScheduledRunRepository, + InMemoryScheduledTaskRepository, +) + +from deerflow.domain.schedule.model import ( + ContextMode, + InvalidScheduleError, + LaunchFailedError, + RunStatus, + SchedulePolicy, + ScheduleSpec, + TaskNotFoundError, + TaskNotMutableError, + TaskStatus, + ThreadBusyError, + ThreadNotFoundError, + TriggerKind, +) +from deerflow.domain.schedule.ports import RunOutcome +from deerflow.domain.schedule.service import ScheduleService + +pytestmark = pytest.mark.asyncio + +NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC) +CRON = ScheduleSpec.cron_schedule("0 9 * * *", "UTC") +POLICY = SchedulePolicy(min_once_delay_seconds=60, max_concurrent_runs=3, lease_seconds=120) + + +def once_spec(*, after_seconds: int = 3600) -> ScheduleSpec: + return ScheduleSpec.once_at(NOW + timedelta(seconds=after_seconds), "UTC") + + +class _BlindRunRepo(InMemoryScheduledRunRepository): + """`has_active` always misses. + + Reproduces the TOCTOU window the fast path cannot close: a concurrent + dispatch inserted its record after this one looked. The active-slot + rejection on `add` is then the only thing standing in the way. + """ + + async def has_active(self, task_id: str) -> bool: + return False + + +def make_service( + *, + tasks: InMemoryScheduledTaskRepository | None = None, + runs: InMemoryScheduledRunRepository | None = None, + launcher: FakeRunLauncher | None = None, + threads: FakeThreadLookup | None = None, + policy: SchedulePolicy = POLICY, +) -> ScheduleService: + return ScheduleService( + tasks=tasks if tasks is not None else InMemoryScheduledTaskRepository(), + runs=runs if runs is not None else InMemoryScheduledRunRepository(), + launcher=launcher if launcher is not None else FakeRunLauncher(), + threads=threads if threads is not None else FakeThreadLookup(), + policy=policy, + lease_owner="test-worker", + ) + + +async def create_cron_task(service: ScheduleService, *, schedule: ScheduleSpec = CRON): + return await service.create_task( + user_id="user-1", + title="Daily summary", + prompt="summarize", + schedule=schedule, + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + ) + + +# ==================================================================== lifecycle + + +class TestFullLifecycle: + async def test_create_claim_dispatch_overlap_complete_pause_delete(self): + """The acceptance criterion: the whole feature, zero IO. + + Every step below is a real use case going through real domain rules -- + only the four ports are doubles. + """ + tasks = InMemoryScheduledTaskRepository() + runs = InMemoryScheduledRunRepository() + launcher = FakeRunLauncher() + service = make_service(tasks=tasks, runs=runs, launcher=launcher) + + # -- create --------------------------------------------------------- + task = await create_cron_task(service) + assert task.status is TaskStatus.ENABLED + assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC) + + # -- become due, get claimed and dispatched -------------------------- + due_at = task.next_run_at + timedelta(seconds=1) + results = await service.run_once(now=due_at) + assert [r.outcome for r in results] == ["launched"] + assert len(launcher.calls) == 1 + launch = launcher.calls[0] + assert launch["prompt"] == "summarize" + assert launch["owner_user_id"] == "user-1" + assert launch["metadata"]["scheduled_task_id"] == task.task_id + assert launch["metadata"]["scheduled_trigger"] == "scheduled" + + after_launch = await service.get_task(task.task_id, user_id="user-1") + assert after_launch.status is TaskStatus.ENABLED, "a cron task stays claimable" + assert after_launch.run_count == 1 + assert after_launch.last_run_id == "run-1" + assert after_launch.next_run_at > due_at + + # -- next occurrence overlaps the still-running one ------------------ + overlap_at = after_launch.next_run_at + timedelta(seconds=1) + overlapped = await service.run_once(now=overlap_at) + assert [r.outcome for r in overlapped] == ["skipped"] + assert len(launcher.calls) == 1, "the overlapping occurrence must not launch" + + after_skip = await service.get_task(task.task_id, user_id="user-1") + assert after_skip.run_count == 1, "a skip is not an execution" + assert after_skip.last_run_id == after_launch.last_run_id, "bookkeeping carried over" + assert after_skip.status is TaskStatus.ENABLED + + # -- the first run finally finishes ---------------------------------- + record = next(r for r in runs.all_runs() if r.status is RunStatus.RUNNING) + await service.handle_run_completion( + RunOutcome( + task_id=task.task_id, + record_id=record.record_id, + run_id="run-1", + user_id="user-1", + status=RunStatus.SUCCESS, + error=None, + ), + now=overlap_at, + ) + assert await runs.count_active() == 0, "the active slot is free again" + + # -- pause / resume --------------------------------------------------- + paused = await service.pause_task(task.task_id, user_id="user-1") + assert paused.status is TaskStatus.PAUSED + assert await service.run_once(now=overlap_at + timedelta(days=2)) == [], "a paused task is not claimed" + + resumed = await service.resume_task(task.task_id, user_id="user-1") + assert resumed.status is TaskStatus.ENABLED + + # -- history and delete ---------------------------------------------- + history = await service.list_task_runs(task.task_id, user_id="user-1") + assert sorted(run.status for run in history) == [RunStatus.SKIPPED, RunStatus.SUCCESS] + + await service.delete_task(task.task_id, user_id="user-1") + with pytest.raises(TaskNotFoundError): + await service.get_task(task.task_id, user_id="user-1") + + +# ==================================================================== dispatch + + +class TestDispatchOutcomes: + async def test_launch_records_the_run_and_the_bookkeeping(self): + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs) + task = await create_cron_task(service) + + result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + assert result.outcome == "launched" + assert result.run_id == "run-1" + assert result.error is None + stored = runs.all_runs()[0] + assert stored.status is RunStatus.RUNNING + assert stored.run_id == "run-1" + assert stored.started_at == NOW + + async def test_manual_trigger_against_an_active_run_is_a_conflict_with_no_record(self): + """Nothing was scheduled to happen, so there is no occurrence to + account for -- the caller gets a conflict and the history stays clean.""" + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs) + task = await create_cron_task(service) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + before = len(runs.all_runs()) + + result = await service.trigger_task(task.task_id, user_id="user-1", now=NOW) + + assert result.outcome == "conflict" + assert result.record_id is None + assert result.error == "task already has an active run" + assert len(runs.all_runs()) == before, "no history row for a rejected manual trigger" + + async def test_scheduled_overlap_records_a_terminal_tombstone(self): + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs) + task = await create_cron_task(service) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + assert result.outcome == "skipped" + assert result.error == "skipped: a previous run of this task is still active" + tombstone = next(r for r in runs.all_runs() if r.status is RunStatus.SKIPPED) + assert tombstone.is_active is False, "a queued tombstone would collide with the live run" + assert tombstone.started_at == tombstone.finished_at == NOW + + async def test_launch_failure_is_recorded_as_failed(self): + runs = InMemoryScheduledRunRepository() + launcher = FakeRunLauncher(fail_with=LaunchFailedError("provider exploded")) + service = make_service(runs=runs, launcher=launcher) + task = await create_cron_task(service) + + result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + assert result.outcome == "failed" + assert result.run_id is None + assert "provider exploded" in result.error + assert runs.all_runs()[0].status is RunStatus.FAILED + + async def test_busy_thread_on_the_scheduled_path_degrades_to_a_skip(self): + launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy")) + service = make_service(launcher=launcher) + task = await create_cron_task(service) + + result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + assert result.outcome == "skipped" + + async def test_busy_thread_on_a_manual_trigger_stays_a_conflict(self): + """Not a skip: the user asked for this one, so it is reported rather + than quietly accounted for. The router maps it to 409, not 502.""" + launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy")) + service = make_service(launcher=launcher) + task = await create_cron_task(service) + + result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL) + + assert result.outcome == "conflict" + assert result.record_id is not None, "the attempt itself is still recorded" + + +class TestConflictCollapse: + """The fast path and the active-slot rejection must be indistinguishable. + + Two concurrent dispatches can both pass `has_active`; whichever loses is + rejected by the repository instead. A caller must not be able to tell which + of the two mechanisms stopped it, or retry behaviour diverges. + """ + + async def _dispatch_second(self, run_repo, trigger: TriggerKind): + # reuse_thread so the execution thread is fixed: a fresh-thread task + # mints a new uuid per dispatch, which would make the two runs differ + # for a reason that has nothing to do with the collapse. + service = make_service(runs=run_repo, threads=FakeThreadLookup({"thread-1": "user-1"})) + task = await service.create_task( + user_id="user-1", + title="t", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + now=NOW, + ) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + return await service.dispatch_task(task, now=NOW, trigger=trigger) + + @pytest.mark.parametrize("trigger", [TriggerKind.SCHEDULED, TriggerKind.MANUAL]) + async def test_both_paths_produce_the_same_result(self, trigger): + via_fast_path = await self._dispatch_second(InMemoryScheduledRunRepository(), trigger) + via_slot_rejection = await self._dispatch_second(_BlindRunRepo(), trigger) + + # record_id is freshly generated; every other field must match exactly. + assert replace(via_fast_path, record_id=None) == replace(via_slot_rejection, record_id=None) + + async def test_the_losing_scheduled_dispatch_still_leaves_a_tombstone(self): + runs = _BlindRunRepo() + result = await self._dispatch_second(runs, TriggerKind.SCHEDULED) + assert result.outcome == "skipped" + assert any(r.status is RunStatus.SKIPPED for r in runs.all_runs()) + + async def test_the_losing_manual_dispatch_leaves_nothing(self): + runs = _BlindRunRepo() + result = await self._dispatch_second(runs, TriggerKind.MANUAL) + assert result.outcome == "conflict" + assert not any(r.status is RunStatus.SKIPPED for r in runs.all_runs()) + + +class TestOnceTaskDispatch: + async def test_a_once_task_waits_in_running_for_its_completion(self): + """Declaring it complete at launch would stick if the run failed or the + process died before the hook could correct it.""" + service = make_service() + task = await service.create_task( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + ) + + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + after = await service.get_task(task.task_id, user_id="user-1") + assert after.status is TaskStatus.RUNNING + + async def test_a_skipped_once_task_is_failed_not_completed(self): + """The single occurrence was lost; `completed` would claim an execution + that never happened.""" + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs) + task = await service.create_task( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + ) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + reloaded = await service.get_task(task.task_id, user_id="user-1") + + await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED) + + after = await service.get_task(task.task_id, user_id="user-1") + assert after.status is TaskStatus.FAILED + assert after.last_error == "skipped: a previous run of this task is still active" + + +# ==================================================================== run_once + + +class TestRunOnceBudget: + async def test_claims_nothing_when_the_global_budget_is_exhausted(self): + """The cap is on active runs across all tasks, not on one poll's batch: + long runs accumulate across cycles.""" + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs, policy=replace(POLICY, max_concurrent_runs=1)) + first = await create_cron_task(service) + await service.dispatch_task(first, now=NOW, trigger=TriggerKind.SCHEDULED) + + second = await create_cron_task(service) + results = await service.run_once(now=second.next_run_at + timedelta(seconds=1)) + + assert results == [] + + async def test_claims_only_into_the_remaining_budget(self): + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks, policy=replace(POLICY, max_concurrent_runs=2)) + for _ in range(3): + await create_cron_task(service) + + due_at = datetime(2026, 7, 28, 9, 0, 1, tzinfo=UTC) + results = await service.run_once(now=due_at) + + assert len(results) == 2 + assert all(r.outcome == "launched" for r in results) + + async def test_a_claimed_task_is_marked_running_before_dispatch(self): + """Claiming is what makes the task uneditable while it is being + dispatched, so the claim must land before the launch.""" + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks) + task = await create_cron_task(service) + + await service.run_once(now=task.next_run_at + timedelta(seconds=1)) + + assert tasks.lease_of(task.task_id) == (None, None), "the claim is released after dispatch" + + +# ==================================================================== completion + + +class TestRunCompletion: + async def _launched_once_task(self): + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs) + task = await service.create_task( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + ) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + record = runs.all_runs()[0] + return service, task, record + + @pytest.mark.parametrize( + ("outcome_status", "expected"), + [ + (RunStatus.SUCCESS, TaskStatus.COMPLETED), + (RunStatus.FAILED, TaskStatus.FAILED), + (RunStatus.INTERRUPTED, TaskStatus.CANCELLED), + ], + ) + async def test_once_task_terminal_mapping(self, outcome_status, expected): + service, task, record = await self._launched_once_task() + + await service.handle_run_completion( + RunOutcome( + task_id=task.task_id, + record_id=record.record_id, + run_id="run-1", + user_id="user-1", + status=outcome_status, + error=None if outcome_status is RunStatus.SUCCESS else "boom", + ), + now=NOW, + ) + + after = await service.get_task(task.task_id, user_id="user-1") + assert after.status is expected + + async def test_a_cron_task_keeps_its_status_but_records_the_error(self): + runs = InMemoryScheduledRunRepository() + service = make_service(runs=runs) + task = await create_cron_task(service) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + record = runs.all_runs()[0] + + await service.handle_run_completion( + RunOutcome( + task_id=task.task_id, + record_id=record.record_id, + run_id="run-1", + user_id="user-1", + status=RunStatus.FAILED, + error="boom", + ), + now=NOW, + ) + + after = await service.get_task(task.task_id, user_id="user-1") + assert after.status is TaskStatus.ENABLED, "the schedule outlives any single run" + assert after.last_error == "boom" + + async def test_a_task_deleted_mid_flight_is_not_an_error(self): + service, task, record = await self._launched_once_task() + await service.delete_task(task.task_id, user_id="user-1") + + await service.handle_run_completion( + RunOutcome( + task_id=task.task_id, + record_id=record.record_id, + run_id="run-1", + user_id="user-1", + status=RunStatus.SUCCESS, + error=None, + ), + now=NOW, + ) + + +class TestReconcileOnStartup: + async def test_sweeps_orphaned_runs_and_stuck_once_tasks(self): + runs = InMemoryScheduledRunRepository() + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks, runs=runs) + task = await service.create_task( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + ) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + # The process dies here: the run is active and the once task is parked + # in running with its claim already released. + + stale_runs, stuck_tasks = await service.reconcile_on_startup(error="gateway restarted") + + assert (stale_runs, stuck_tasks) == (1, 1) + assert runs.all_runs()[0].status is RunStatus.INTERRUPTED + after = await service.get_task(task.task_id, user_id="user-1") + assert after.status is TaskStatus.CANCELLED + + +# ==================================================================== CRUD + + +class TestTaskManagement: + async def test_reuse_thread_requires_an_accessible_thread(self): + service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"})) + + created = await service.create_task( + user_id="user-1", + title="t", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + now=NOW, + ) + assert created.thread_id == "thread-1" + + with pytest.raises(ThreadNotFoundError): + await service.create_task( + user_id="user-2", + title="t", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + now=NOW, + ) + + async def test_an_invalid_schedule_is_rejected_before_any_write(self): + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks) + + with pytest.raises(InvalidScheduleError): + await service.create_task( + user_id="user-1", + title="t", + prompt="p", + schedule=once_spec(after_seconds=10), # inside min_once_delay + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + ) + assert await service.list_tasks("user-1") == [] + + async def test_update_leaves_omitted_fields_alone(self): + service = make_service() + task = await create_cron_task(service) + + updated = await service.update_task(task.task_id, user_id="user-1", now=NOW, title="renamed") + + assert updated.title == "renamed" + assert updated.prompt == task.prompt + assert updated.schedule == task.schedule + + async def test_rescheduling_a_terminal_task_re_arms_it(self): + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks) + task = await create_cron_task(service) + tasks.seed(replace(task, status=TaskStatus.FAILED)) + + updated = await service.update_task( + task.task_id, + user_id="user-1", + now=NOW, + schedule=ScheduleSpec.cron_schedule("0 10 * * *", "UTC"), + ) + + assert updated.status is TaskStatus.ENABLED, "otherwise it would never be claimed again" + + async def test_a_running_task_cannot_be_edited(self): + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks) + task = await create_cron_task(service) + tasks.seed(replace(task, status=TaskStatus.RUNNING)) + + with pytest.raises(TaskNotMutableError): + await service.update_task(task.task_id, user_id="user-1", now=NOW, title="nope") + with pytest.raises(TaskNotMutableError): + await service.pause_task(task.task_id, user_id="user-1") + + async def test_a_running_task_can_still_be_deleted(self): + """The pre-migration router gated update/pause/resume on this, but not + delete -- that asymmetry is preserved.""" + tasks = InMemoryScheduledTaskRepository() + service = make_service(tasks=tasks) + task = await create_cron_task(service) + tasks.seed(replace(task, status=TaskStatus.RUNNING)) + + await service.delete_task(task.task_id, user_id="user-1") + + @pytest.mark.parametrize("call", ["get", "update", "pause", "resume", "delete", "runs"]) + async def test_another_users_task_is_reported_as_missing(self, call): + service = make_service() + task = await create_cron_task(service) + kwargs = {"user_id": "intruder"} + + with pytest.raises(TaskNotFoundError): + if call == "get": + await service.get_task(task.task_id, **kwargs) + elif call == "update": + await service.update_task(task.task_id, now=NOW, title="x", **kwargs) + elif call == "pause": + await service.pause_task(task.task_id, **kwargs) + elif call == "resume": + await service.resume_task(task.task_id, **kwargs) + elif call == "delete": + await service.delete_task(task.task_id, **kwargs) + else: + await service.list_task_runs(task.task_id, **kwargs) + + async def test_tasks_are_listed_per_user_and_per_thread(self): + service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"})) + bound = await service.create_task( + user_id="user-1", + title="bound", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + now=NOW, + ) + await create_cron_task(service) + + assert len(await service.list_tasks("user-1")) == 2 + assert await service.list_tasks("someone-else") == [] + by_thread = await service.list_tasks_by_thread("user-1", "thread-1") + assert [t.task_id for t in by_thread] == [bound.task_id]