diff --git a/backend/tests/test_schedule_domain.py b/backend/tests/test_schedule_domain.py index 2a86bf894..a20def22f 100644 --- a/backend/tests/test_schedule_domain.py +++ b/backend/tests/test_schedule_domain.py @@ -13,12 +13,13 @@ 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.model import ( - ACTIVE_RUN_STATUSES, ContextMode, InvalidContextModeError, InvalidScheduleError, @@ -101,16 +102,13 @@ class TestScheduleSpecInvariants: 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 - def test_zulu_iso_input_is_accepted_as_the_same_instant(self): - """The shape the frontend actually submits: `zonedLocalToUtcIso` output, - which carries a trailing Z. It is already aware, so localization must - leave it alone rather than reinterpreting it in the task's timezone.""" - spec = ScheduleSpec.once_at(datetime.fromisoformat("2026-08-01T01:00:00Z"), "Asia/Shanghai") - assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) - # ---------------------------------------------------------------- A2. from_primitives @@ -249,9 +247,26 @@ class TestEnsureLaunchable: spec = once_spec(after_seconds=3600) assert spec.ensure_launchable(NOW.replace(tzinfo=None), MIN_60) == spec.ensure_launchable(NOW, MIN_60) - def test_matches_next_after_for_a_valid_once_schedule(self): - spec = once_spec(after_seconds=3600) - assert spec.ensure_launchable(NOW, MIN_60) == spec.next_after(NOW) + +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 @@ -412,6 +427,8 @@ class TestStatusAfterCompletion: ("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), ], @@ -419,11 +436,6 @@ class TestStatusAfterCompletion: def test_once_maps_each_terminal_outcome(self, outcome, expected): assert make_task(schedule=once_spec()).status_after_completion(outcome) is expected - def test_interrupted_is_cancelled_not_failed(self): - """A user cancel or same-thread takeover carries no execution failure.""" - task = make_task(schedule=once_spec()) - assert task.status_after_completion(RunStatus.INTERRUPTED) is TaskStatus.CANCELLED - # ---------------------------------------------------------------- G. mutability gate @@ -454,6 +466,16 @@ class TestMutabilityGate: 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 @@ -525,12 +547,20 @@ class TestResolveExecutionThread: 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.""" - task = make_task() + 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) - assert ScheduledTask.resolve_execution_thread(legacy) != task.thread_id + + 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 @@ -555,5 +585,27 @@ class TestScheduledRun: second = ScheduledRun.queued(task_id="task-1", thread_id="t", scheduled_for=NOW, trigger=TriggerKind.MANUAL) assert first.record_id != second.record_id - def test_active_statuses_are_exactly_queued_and_running(self): - assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING) + @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_fakes.py b/backend/tests/test_schedule_fakes.py index 3142d560d..ec1b248e9 100644 --- a/backend/tests/test_schedule_fakes.py +++ b/backend/tests/test_schedule_fakes.py @@ -32,7 +32,6 @@ from app.adapters.schedule.scheduled_run_repository import SqlScheduledRunReposi from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository from deerflow.config.database_config import DatabaseConfig from deerflow.domain.schedule.model import ( - ACTIVE_RUN_STATUSES, ActiveRunConflictError, ContextMode, RunStatus, @@ -532,9 +531,6 @@ class TestRunStatusWrites: assert by_id[active.record_id].error == "gateway restarted" assert by_id[done.record_id].status is RunStatus.SKIPPED - async def test_active_statuses_are_exactly_queued_and_running(self): - assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING) - class TestLauncherAndThreadLookup: """Fake-only: these two ports have no SQL implementation -- one starts a diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py index 84b3d26e0..7e6b8c098 100644 --- a/backend/tests/test_schedule_service.py +++ b/backend/tests/test_schedule_service.py @@ -176,6 +176,11 @@ class TestFullLifecycle: 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) @@ -271,6 +276,47 @@ class TestDispatchOutcomes: 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) @@ -412,16 +458,34 @@ class TestRunOnceBudget: assert len(results) == 2 assert all(r.outcome is DispatchOutcome.LAUNCHED for r in results) - async def test_a_claimed_task_is_marked_running_before_dispatch(self): + 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 must land before the launch.""" + 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() - service = make_service(tasks=tasks) + 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 tasks.lease_of(task.task_id) == (None, None), "the claim is released after dispatch" + 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 @@ -493,8 +557,28 @@ class TestRunCompletion: 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() + 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_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] await service.delete_task(task.task_id, user_id="user-1") await service.handle_run_completion( @@ -509,6 +593,11 @@ class TestRunCompletion: 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. diff --git a/backend/tests/test_scheduled_task_models.py b/backend/tests/test_scheduled_task_models.py index c4f7b1f4b..47cfbc94f 100644 --- a/backend/tests/test_scheduled_task_models.py +++ b/backend/tests/test_scheduled_task_models.py @@ -1,4 +1,9 @@ +import re + +from sqlalchemy import Index + from deerflow.config.app_config import AppConfig +from deerflow.domain.schedule.model import ACTIVE_RUN_STATUSES from deerflow.persistence.models import ScheduledTaskRow, ScheduledTaskRunRow @@ -17,3 +22,37 @@ def test_app_config_exposes_scheduler_section(): def test_scheduled_task_models_registered(): assert ScheduledTaskRow.__tablename__ == "scheduled_tasks" assert ScheduledTaskRunRow.__tablename__ == "scheduled_task_runs" + + +def _active_run_index() -> Index: + return next(arg for arg in ScheduledTaskRunRow.__table_args__ if isinstance(arg, Index) and arg.name == "uq_scheduled_task_run_active") + + +def test_active_run_index_arbitrates_one_active_run_per_task(): + """The index is the atomic arbiter of the overlap rule, so its shape is a + contract, not a detail: unique, keyed on `task_id` alone.""" + index = _active_run_index() + assert index.unique is True + assert [column.name for column in index.expressions] == ["task_id"] + + +def test_active_run_index_predicate_matches_the_domain_constant(): + """`ACTIVE_RUN_STATUSES` and this predicate must stay in lockstep. + + The domain's fast path (`has_active`) and the index disagree the moment + they drift, which silently decouples the overlap check from its arbiter. + The domain tests cannot assert this -- they are deliberately + dependency-free and cannot import an ORM model -- so the assertion the + `ACTIVE_RUN_STATUSES` docstring promises lives here. + + Both dialect predicates are checked: `create_all` renders the SQLite one + and production renders the Postgres one, so a drift in either is real. + """ + index = _active_run_index() + expected = {str(status) for status in ACTIVE_RUN_STATUSES} + assert expected == {"queued", "running"}, "domain constant changed -- update the ORM predicates below" + + predicates = {key: str(value) for key, value in index.dialect_kwargs.items() if key.endswith("_where")} + assert set(predicates) == {"sqlite_where", "postgresql_where"}, "a dialect lost its partial-index predicate" + for dialect, predicate in predicates.items(): + assert set(re.findall(r"'([^']+)'", predicate)) == expected, f"{dialect} predicate drifted from ACTIVE_RUN_STATUSES"