Remove the legacy stack the hexagonal slice replaced, now that nothing
assembles it: app/scheduler/service.py (the old orchestration),
app/gateway/routers/scheduled_tasks.py (the old dict-returning router,
no longer mounted), the deerflow/scheduler package (its cron/timezone
rules live in ScheduleSpec), the dict-returning repositories in
persistence/scheduled_task*/sql.py, and the deps.py providers and
app.state wiring that served them. The ORM rows and the
uq_scheduled_task_run_active partial unique index stay -- the table
definitions live with the shared engine/alembic infrastructure and the
schedule adapters are their only readers and writers.
The legacy test suites go with the code they pinned; every scenario has
a counterpart in the new suites. The one suite that guarded semantics
rather than the old implementation -- the real-database dispatch-race
TOCTOU tests -- is migrated to the new stack as
test_schedule_dispatch_race.py, driving ScheduleService over the real
SQL adapters with the same barrier, natural-timing, and index-semantics
cases.
Docs and comments that named the old classes as the current wiring
(backend/AGENTS.md, reload_boundary.py, channel/service comments) now
name the composition-root wiring instead.
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.
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>
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.
Four Protocols the domain declares and the outer ring will implement,
plus the two DTOs that keep infrastructure types out of the inner ring:
- ScheduledTaskRepository / ScheduledRunRepository, exchanging domain
objects rather than the bare dicts the current repositories return
- RunLauncher, whose contract is that only ThreadBusyError or
LaunchFailedError may escape -- that translation is what keeps the run
runtime and the web framework out of the domain
- ThreadLookup, one method rather than the whole thread store
- LaunchedRun and RunOutcome, so the completion path stops taking a
runtime record the purity test would reject
Two deliberate departures from the earlier sketch. There is no Clock
port: `now` is already an explicit parameter throughout, so the domain
never reads a clock and the tests are already deterministic -- adding
one would only create a second source of truth for the same value. And
`record_launch` is not expressed as `save(task)`, because
`protect_terminal` makes it a compare-and-set against a concurrently
finalizing run; a read-modify-write through the aggregate would
reintroduce the race the flag exists to close.
The in-memory doubles model the active-slot rule rather than skipping
it: a double that never refuses a second active run would let the
service's conflict collapse go untested. Their semantics are pinned by
test_schedule_fakes.py, which becomes the contract suite once the SQL
adapters land and both tiers run the same cases.
Concurrency is out of scope for the doubles and says so in their module
docstring -- it stays covered against a real database in
test_scheduled_task_dispatch_race.py.
