mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
d856ae8573
The inner ring of the schedule slice, added on its own so it can be read as domain modelling rather than as a diff against the old implementation: two aggregates with their state machines, the policy value object, the output ports the service depends on, and the errors it raises. Nothing wires it up yet -- no existing code path changes. The service is exercised end to end against in-memory fakes, which is what makes the rules (overlap policy, lease handling, which write owns which timestamp) assertable without a database at all. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
906 lines
38 KiB
Python
906 lines
38 KiB
Python
"""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]
|