The two command-shape tests added with the command refactor are plain
functions in a module whose global pytestmark is asyncio, which pytest
reports as a warning per run. Making them coroutines matches the module
convention.
Add domain/schedule/commands.py with one frozen dataclass per HTTP-driven
write use case -- CreateScheduledTask, UpdateScheduledTask, PauseTask,
ResumeTask, DeleteTask, TriggerTask -- and make the service methods their
handlers, keeping the naming chain aligned across all three spellings
(command / handler / <Command>Request). The clock stays an explicit now=
handler parameter: it is a rule input, not part of the client's intent.
UpdateScheduledTask expresses partial updates with an UNSET sentinel, so
absence is unambiguous; the wire keeps its historical None-means-omitted
convention and the request model's to_command owns the translation. The
former ContextChange moves from service.py into commands.py unchanged.
The request models are renamed to <Command>Request and own transformation
① (to_command): identity is injected server-side and pinned by a test to
never appear on the wire models. Clock- and callback-driven writes
(run_once, dispatch_task, handle_run_completion, reconcile_on_startup)
deliberately stay plain methods -- those drivers have no wire shape to
translate, their inputs are already domain vocabulary.
The context package now also exports the commands and the service, which
completes its public API and retires the stale 'service not landed' note.
Move model/errors.py up one level to domain/schedule/exceptions.py, a
sibling of the model, matching the AWS domain layout the spec mandates
(exceptions/ is its own member of the domain folder, not part of the
model) and the feedback reference implementation. Class names keep the
PEP 8 Error suffix. Pure move -- the nine classes are AST-identical to
the originals; imports across the domain, adapters, router, and tests
now take errors from deerflow.domain.schedule.exceptions.
All 102 domain cases were green, and four of them would have stayed green
through the exact regression they were named after.
Assertions that could not fail
------------------------------
`test_a_claimed_task_is_marked_running_before_dispatch` asserted the lease was
released *after* dispatch -- the opposite of the ordering its name and
docstring describe. The claim is what makes a task uneditable while it is
being dispatched, so the only place that ordering is observable is inside the
launch; the launcher double now reads the repository from there.
`test_active_statuses_are_exactly_queued_and_running` restated the constant it
was checking, so editing the constant edits the assertion with it. Replaced by
`is_active` over all six statuses, which also covers `RUNNING` and the two
terminal statuses that had none.
`test_reuse_thread_with_an_empty_thread_falls_back_to_a_fresh_one` compared
its result against `task.thread_id`, which is `None` on the default task -- it
asserted "not None" against a method whose body is `str(uuid.uuid4())`. Now
asserts the fresh-thread semantics it is named for: a real uuid, distinct per
call.
`test_a_task_deleted_mid_flight_is_not_an_error` had no assert at all. That
path does have observable behaviour: the hook writes the run record before it
reads the task, so a task deleted mid-flight must still leave a finalized
record and a freed active slot.
Contracts stated in a docstring and nowhere else
------------------------------------------------
- a cron overlap must not leave `last_error` behind (service.py:455 branches
on it; only the `once` half was covered, so dropping the branch was free)
- a failed launch replaces the launch bookkeeping instead of carrying it over
the way a skip does -- which is what `last_run_id=None` in `_fail` means for
a task that had already run successfully
- `SchedulePolicy`'s defaults are the permissive ones, so a deployment that
configured no policy cannot have a business constraint invented for it
- transitions leave `updated_at` to the repository, rather than becoming a
second source of truth for the same column
ACTIVE_RUN_STATUSES' promised assertion
---------------------------------------
Its docstring says the check that it stays in lockstep with the partial unique
index's predicate "lives in a separate test module rather than the domain
tests". It did not exist, in that module or any other. `test_scheduled_task_
models.py` now reads both dialect predicates off `__table_args__` and compares
the values it extracts against the constant. The domain suite cannot do this
-- it is deliberately dependency-free and cannot import an ORM model -- so the
new domain case names where the other half of the rule lives.
Removed
-------
Three duplicates: an `INTERRUPTED -> CANCELLED` case the parametrize directly
above it already made, a trailing-Z case identical in path to the aware-run_at
case beside it (the `from_primitives` one is the real one, because it parses a
string), and an `ensure_launchable == next_after` case whose value another
case already asserts outright. Their reasoning moved into comments where it
still applies. The second copy of the tautology, in `test_schedule_fakes.py`,
goes with it.
Verification
------------
A green run is not evidence for this kind of change, so every new or rewritten
assertion was checked by mutation -- break the production rule, confirm the
guarding test fails. All nine caught, re-run after `ruff format` to confirm
the reformat did not soften any of them.
Net +207/-31 across four test files; 425 passed, 3 skipped for the schedule
and composition suites.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
A cron task could become permanently unschedulable after a run that failed
fast. Reproduced against the real SQL adapters, and against the legacy path
for contrast -- the legacy code does not have this bug, so the hexagonal
slice introduced it.
The window
----------
`dispatch_task` already documents it: "a fast-failing run can reach
handle_run_completion before this write lands". The hook therefore reads the
task while the dispatch path has not yet written its bookkeeping, and the
snapshot it holds still carries the elapsed `next_run_at` the claim was made
on. `handle_run_completion` then wrote that whole snapshot back through
`save()`, so whichever landed second undid the other.
For a cron task the result is terminal in the worst way. `record_launch`
writes the fresh fire time, bumps `run_count` and clears the claim; the
completion's whole-aggregate write restores the elapsed fire time, rolls
`run_count` back and leaves `status='running'` with `lease_expires_at IS
NULL`. Neither `claim_due` branch matches that shape (one needs `enabled`,
the other needs an expired claim), and `cancel_stuck_once_tasks` only sweeps
`once` rows -- so nothing can ever reach the task again.
The legacy `app/scheduler/service.py` wrote the same outcome field by field
(`update(..., updates={"last_error": ...})`) and never touched scheduling
state, which is why it survives the same interleaving.
The fix
-------
`record_completion` joins `record_launch` as a second deliberately narrow
port method, for the same stated reason `record_launch` is not expressed as
`save(task)`: the two race, so neither may write through the aggregate. They
now own disjoint fields -- the launch owns the schedule, the completion owns
the verdict (terminal status plus `last_error`, with `None` meaning "do not
move the status", i.e. every cron task).
`save()` keeps its documented purpose, the user-initiated whole-aggregate
updates (`update_task` / `pause` / `resume`). `handle_run_completion` still
reads the task first, but only to ask `status_after_completion`, which reads
nothing but the schedule type -- immutable, so that read carries no
time-of-check risk.
Tests
-----
The contract suite missed this because it groups cases by port method:
`record_launch` appears only among its own, never interleaved with `save`,
and `test_protect_terminal_keeps_a_concurrently_finalized_verdict` pins the
mirror-image direction only. The in-memory double replaces the whole row too,
so both implementations were faithfully wrong -- the defect was in the
contract, not either adapter.
Added: `TestRecordCompletion` (5 cases, both implementations), including the
interleaving itself; and `test_a_cron_task_survives_a_launch_write_landing_mid_completion`,
which drives a real `ScheduleService` over a repo double that commits
`record_launch` between the hook's read and its write. Both were watched
failing first -- the service case failing with the actual symptom, a rolled
back `next_run_at`, not a missing method.
Verified end to end against file-backed sqlite with the real service: the
task keeps `enabled`, the concurrent fire time and run count survive, the
verdict is recorded, and the next poll still claims it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
A review pass over domain/schedule found three genuinely untested
branches, all of which now have cases:
- ensure_launchable with a naive `now`. next_after already had this
covered; the delay floor did not, so a caller handing over a naive
clock reading could have had it shifted by the local offset unnoticed.
- update_task changing the prompt. Only the title path was exercised.
- _save finding the row gone. get_task saw it and save no longer does,
which is a concurrent delete; the caller must get the same not-found it
would have got a moment earlier rather than a None leaking out.
That takes service.py and every model module to 100%. The one remaining
uncovered line is croniter's naive-return guard, carried over verbatim
from schedules.py and unreachable with an aware input -- it now says so
instead of looking like an untested branch.
Also fills in the documentation the migration skipped: TaskStatus,
ContextMode and RunStatus arrived from the original draft without
docstrings while their newer siblings had them, and ScheduleService plus
four of its use cases were undocumented. Each now records the reasoning a
reader would otherwise have to reconstruct -- why RUNNING is not "the
agent is executing", why SKIPPED never passes through QUEUED, why
INTERRUPTED is not FAILED.
CRON_FIELD_COUNT stops being exported: it has no consumer outside the
module that defines it.
Four changes, all from the same observation -- the service was expressing
domain concepts in language the domain does not use.
DispatchOutcome replaces the Literal. The caller branches on all four
values and the SKIPPED/CONFLICT distinction is itself a business rule, so
it is domain vocabulary like every other enum in this context; a bare
string was the odd one out.
lease_owner leaves the domain entirely. It was a pure pass-through: the
service held it only to hand it to claim_due, which never reads it back.
Which process claimed a task is an identity, not a rule -- an adapter can
record one for diagnostics without the domain carrying it. lease_seconds
stays, because how long a claim survives genuinely changes recovery
behaviour.
The _UNSET sentinel is gone. It existed for one field: thread_id, the only
update parameter with a meaningful None. But thread_id and context_mode
always move together, so packaging them as ContextChange removes the
ambiguity and lets every other field use plain None for "not supplied" --
which also matches what the HTTP layer already does with exclude_none.
The previous `title: str = _UNSET` annotation was simply untrue.
pause/resume no longer route through a _transition helper taking an
unbound method. Two direct bodies plus a _save that raises on a missing
row read better, and _save documents why this stays a read-modify-write:
pushing "not while running" into a storage predicate would put the rule
beyond a zero-IO test and give it a second home. Closing that window
properly needs optimistic locking, which needs a schema change.
Also cleans three lint findings surfaced by a broader rule set than the
project enables: a regex metacharacter in a pytest match=, an unused
override parameter, and two deliberately-naive datetimes that now say so
with a noqa rather than looking accidental.
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.
