From d5fa1809845b01def313f08839cf5835702b5f00 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 09:45:49 +0700 Subject: [PATCH] docs: specify deployment scheduling and verify implementation plan --- ...ployment-scheduling-implementation-plan.md | 594 ++++++ ...2026-09-08-deployment-scheduling-design.md | 330 +++ probes/deployment_scheduling_verify/README.md | 40 + .../test_calendar_probe.py | 413 ++++ .../test_expression_contract_probe.py | 138 ++ .../test_schedule_state_model.py | 1803 +++++++++++++++++ 6 files changed, 3318 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-09-deployment-scheduling-implementation-plan.md create mode 100644 docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md create mode 100644 probes/deployment_scheduling_verify/README.md create mode 100644 probes/deployment_scheduling_verify/test_calendar_probe.py create mode 100644 probes/deployment_scheduling_verify/test_expression_contract_probe.py create mode 100644 probes/deployment_scheduling_verify/test_schedule_state_model.py diff --git a/docs/superpowers/plans/2026-09-09-deployment-scheduling-implementation-plan.md b/docs/superpowers/plans/2026-09-09-deployment-scheduling-implementation-plan.md new file mode 100644 index 00000000..318c0e16 --- /dev/null +++ b/docs/superpowers/plans/2026-09-09-deployment-scheduling-implementation-plan.md @@ -0,0 +1,594 @@ +# Deployment Scheduling: Verification Report And Sequenced Implementation Plan + +Status: verification complete; planning only. No production scheduling code +exists on this branch. No blocking product decisions remain (the calendar +policy is decided: croniter owns DST resolution — see Gate 2; the former +custom skip/filter approach is retired, not pending). + +- Branch: `opencode/sched-verify-plan` +- Worktree: `C:\Users\Admin\Documents\lda.chat\lda-workflow-as-struct-sched-verify` +- Spec under review: `docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md` + (updated in this worktree; the original untracked main-worktree copy is + unchanged and is no longer identical) +- Probes (disposable, not production): `probes/deployment_scheduling_verify/` + +Scheduling policies remain settled. Calendar behavior follows the subsequent +approved simplification: croniter owns DST resolution. Findings below record +the missing implementation seams and the completed verification work. + +## Gate 1 — Spec audit against actual code + +Three parallel audit sweeps (expression bindings, deployment invocation, +runs/resume/server/stores) agree. Headline: the spec's description of the +current code is accurate, and every scheduling seam it names is genuinely +missing. In particular: **the existing run API supports stopped-run +persistence only; durable admission does not exist.** + +### 1.1 Input authoring and serialization (spec lines 164-207) + +- Reuse literal/object/array/target-path/strict-JSON/budget semantics: + `src/wf_core/models/input_bindings.py:11-12` (budget consts), + `:15-35` `InputPathBinding`, `:38-56` `InputValueBinding` + strict + JSON, `:59-97` literal/path/array/object expressions, + `src/wf_core/models/json_values.py:9-31`, + `src/wf_core/local_paths.py:38-74` overlap checks. + Supported as vocabulary. +- Closed 4-kind expression union; no date/format/arithmetic machinery: + `input_bindings.py:100-103` discriminator union; only graph-path + resolution in `src/wf_core/runtime/input_bindings.py:23-114`. + Supported. +- Shared composition traversal behind a typed source-resolver seam: + `resolve_input_expression()` takes concrete + `state/workflow_input/context` mappings; `grep SourceResolver` in + `src/` has no hits; two more hardcoded recursions exist + (`src/wf_core/validation/steps.py:242-284`, + `src/wf_api/input_expressions.py:99-253`). **Missing seam** — + schedule evaluation can only copy the evaluator or fake `context`. +- Typed occurrence reference; graph refs invalid in schedules: no + `occurrence`/`scheduled_at`/`schedule_id` model anywhere in `src/`. + Missing (expected; this is the work). +- Do not extend `GraphSourcePath` with schedule-only roots: + `src/wf_core/paths.py:17` (`GraphRoot` = input/state/context), + `:232-265` closed roots/parse/factories, `:423-446` resolution. + Supported as constraint; any occurrence-as-root edit contradicts it. +- Expression budget enforced on the schedule path: budget checked once + at `InputExpressionBinding.check_limits` + (`input_bindings.py:176-183`); the runtime resolver has no limit + check. Missing on any path that bypasses `StepInputBinding`. + +### 1.2 Deployment invocation and pinned environments (spec 22-24, 198-206) + +- Deployment-run API receives resolved input data: + `WorkflowRunApi.run_deployment(deployment_id, workflow_input: dict, + ...)` (`src/wf_api/runs.py:77-84`); same shape in `service.py:1045`, + `surface.py:514`, `protocols.py:118`, `transport/models.py:434`, + `wf_client/deployments.py:182`. Supported — the scheduler needs only + a binding→resolved composer, no API change. +- Pinned artifact tree captured at run start: + `PinnedRunEnvironment{deployment, root_artifact, child_artifacts}` + (`src/wf_artifacts/runs/models.py:51`), + `resolve_saved_subgraph_tree` (`src/wf_api/saved_subgraphs.py:65`), + resume reuses `record.environment` without re-reading the deployment + (`runs.py:192-201`). Supported for resume; missing for admission + (frozen only in memory, persisted only after stop). +- Deployment revision for edit-race detection: `WorkflowDeployment` + (`src/wf_artifacts/models.py:215`) has NO revision; + `save_deployment` overwrites (`src/wf_artifacts/store.py:113`); only + `DraftWorkspace` has revisions. **Missing** — "captured invocation + stays unchanged" needs a deployment revision or content hash. +- Durable admission (preassigned run id, persist-before-dispatch, + idempotent reconcile): `runs.py:101-110` runs + `raw_plan_from_artifact` → full in-memory `run_workflow_from_plan` → + `persist_stopped_run`; id allocated post-execution in + `run_lifecycle.py:63`. **Missing entirely** (see 1.3). + +### 1.3 Durable admission, resume, server, stores (spec 184-260) + +- `wf_api/runs.py` executes before persisting: `runs.py:102-110` + (execute, then `persist_stopped_run`). Supported. +- Run models permit only stopped summaries + required checkpoint id: + `StoredRunStatus` = interrupted/completed/failed + (`runs/models.py:27-32`); `WorkflowRunRecord.latest_checkpoint_id` + required (`:70`); `persist_stopped_run` rejects active states + (`run_lifecycle.py:46-61`); `restore/load` assume a checkpoint exists + (`:101-134`). Supported — inspection of an admitted/in-flight run is + impossible today. +- Resume marks the active attempt durably before re-executing: + `resume_run` (`src/wf_api/runs.py:194-208`): load → validate → + `resume_workflow_from_plan` → persist. No store write between load and + execute (the RUNNING flip in + `src/wf_core/runtime/preparation.py:87` is in-memory only). + **Missing** — crash-during-resume re-presents the old interrupted + checkpoint as safe to retry. +- `persist_stopped_run` is a transaction: two separate writes, + `save_checkpoint` then `save_run` (`run_lifecycle.py:96-97`); each + file atomic via tmp+rename (`runs/store.py:90-94`). Supported as + single-file atomicity; cross-file atomicity missing (recovery + authority needed). +- Server has an enable flag, poll loop, capacity, drain: + `WorkflowServer` is a frozen dataclass with no lifecycle methods + (`wf_server/context.py:279-341`); `cli.py:112-118` bare + `uvicorn.run`; no lifespan/startup/shutdown/background task in + `wf_server` or `wf_transport_rpc_http/app.py:27-68`. All missing + (expected; this is the work). +- File-store single-process limits: per-process `RLock` around + individual writes (`runs/store.py:42-54`); per-run async lock is + process-local (`runs.py:147`, `run_locks.py:15-55`); zero + `flock/fcntl/msvcrt/portalocker/FileLock` hits in `src/`; no + PID/lease/ownership concept. Supported — matches + `2026-06-09-store-transaction-boundary.md`. Scheduler ownership needs + a Windows-tested held-lock design (that spec forbids ad-hoc lock + files without one). +- Stopped-run persistence implies durable admission: **rejected**. + `run_deployment` proves execute-then-persist; there is no + pre-dispatch record, no preassigned id, no admission lock, and no + reconcile. + +## Gate 2 — Calendar-library probe + +Isolated env (NOT the repo env): Python 3.14.7, `apscheduler==3.11.0`, +`croniter==6.2.4`, `tzdata==2026.3`, `pytest==9.1.1`, no `pytz` +(APScheduler used its zoneinfo path). + +Reproduce: + +```powershell +$probe = "C:\Users\Admin\AppData\Local\Temp\opencode\sched-cal-probe" +$file = "\probes\deployment_scheduling_verify\test_calendar_probe.py" +uv run --project $probe python -m pytest $file -q -p no:cacheprovider ` + -o addopts="" +``` + +Result: **16 passed, 2 xfailed** (Part A pins the croniter contract; +Part B's 2 xfails are APScheduler rejected-candidate evidence — gap +phantom and fold replay — marked strict so re-opening that candidacy +fails loudly). + +Policy: croniter owns calendar calculation, including DST resolution. +Part A tests describe observed 6.2.4 behavior the thin adapter consumes +(convert now into the schedule zone, ask for next/previous, convert +back to UTC). No custom skip/filter machinery exists anymore. + +- UTC daily chain (`30 9 * * *`): strictly increasing, unique UTC + instants over 10 occurrences. Pass. +- `Asia/Ho_Chi_Minh`: 09:30 local == 02:30 UTC; 48/48 hourly hits + across March and November windows (no DST). Pass. +- Expression forms on an ordinary day: `*` → next minute, `5/15` → + :05, `0,30 1-2` → 01:00, `*/20 1-3` from 01:50 → 02:00. Pass. +- Weekday dialect: `0` AND `7` mean Sunday (`weekday()=6`), + `1`/`mon` mean Monday (`0 12 * * 0 == 0 12 * * sun`). Pass. +- DOM/DOW: `day_or=True` selected (Unix OR — `0 12 13 * fri` hits + 09-04, 09-11, 09-13, 09-18); `day_or=False` pinned as the + non-selected AND reference. Pass. +- Zone handling: a UTC start yields UTC results with no conversion — + the caller supplies the schedule-zone instant. Pass (adapter rule). +- DST gap forward (America/New_York 2026-03-08): daily 02:30 resolves + to 03:00-04:00 the same day; the following occurrence is 03-09 + 02:30. Per-minute streams jump 01:59 EST straight to 03:00 EDT with + no 02:xx wall times, strictly increasing unique UTC. Pass (observed; + the resolution IS the occurrence). +- DST gap backward: `get_prev` on the gap day returns the resolved + 03:00-04:00; from 03-09 00:00 the same; pre-gap queries return the + prior valid occurrence. Pass (observed; latest-missed on a gap day + is the resolved instant). +- DST fold (2026-11-01 01:30): both repeats occur as distinct UTC + instants (`05:30Z` then `06:30Z`); per-minute iteration across the + fold is 560/560 strictly increasing unique. Pass. +- Long downtime, per-minute, ~3 years: `get_prev` answers the latest + missed occurrence in ~0.0001s, tz-aware, within a minute of now + (UTC and named-zone variants). Pass. +- Boundary exclusivity: `get_next`/`get_prev` from exactly a due + instant return the neighboring occurrences; a tick just before due + admits it. Pass (drives the T01 watermark rule below). +- Impossible date (Feb 30): raises documented `CroniterBadDateError` + in ~0.001s. Pass (adapter maps it to exhausted). +- Malformed expression: `CroniterBadCronError` at construction. Naive + datetimes pass straight through unrejected — the adapter rejects + naive itself (spec requires rejection; no DST logic involved). Pass. + +Rejected-candidate evidence (APScheduler 3.11.0; NOT acceptance): + +- DST gap: APScheduler fabricates a phantom `02:30-05:00` + (= `07:30Z`, actually 03:30 EDT) — a wall time that never existed. + **FAIL (strict xfail)**. +- DST fold per-minute: after `06:00Z`, APScheduler flips back to + `-04:00` and replays `05:01Z`–`06:00Z` (~59 duplicate UTC + identities, UTC goes backward). **FAIL (strict xfail)**. +- No bounded latest-missed seam exists in its documented trigger API + surface (forward-only `get_next_fire_time`); ~1.6M iterations would + be needed for 3 years of per-minute misses. + +### Thin-adapter contract (replaces the retired skip rule) + +T01 implements exactly this and nothing more: + +- Convert the query instant into the schedule's named zone; ask + croniter for the next (`next_after`) or previous (`prev_before`) + occurrence; convert the result to UTC. +- Iteration is exclusive both directions (pinned above), so forward + queries start from the last-consumed instant and catch-up results + are compared against the same watermark: a due occurrence is + admitted exactly once, never missed by exclusivity nor doubled. +- Use the library's bounded-search controls and documented + exceptions: `CroniterBadDateError` maps to exhausted; + `CroniterBadCronError` at construction is a definition rejection. + Search exhaustion or a backward/non-progressing result fails + visibly — it must never look like successful progress, and the + adapter must not "repair" results with its own calendar engine. +- Admitted identity stays `(schedule_id, resolved UTC)`, which always + exists. The definition's cron expression, time-zone name, and + admission snapshots are retained; no finer intended-wall-time + provenance is claimed. Minute precision (this slice). + +### Recommendation + +1. APScheduler has no bounded latest-missed seam in its documented trigger + API surface (forward-only `get_next_fire_time`). The spec's "never + enumerate years of missed occurrences" gate cannot be met with it; + `croniter.get_prev` answers in ~0.0004s. +2. APScheduler violates the occurrence-identity gate: it replays ~59 past + UTC minutes after every fall fold (duplicates + backward UTC), which + would double-admit per-minute schedules. croniter is clean (pinned). +3. Across DST gaps APScheduler fabricates a nonexistent wall time; + croniter resolves gap days forward to an existing wall time and + emits both fold hours as distinct UTC occurrences (all pinned). +4. The spec needs only iteration, not APScheduler's job store/executor + (which it already excludes). croniter is the smaller dependency with the + needed `get_prev` seam and standard DOM/DOW `day_or=True` (pinned). +5. Pin `croniter==6.2.4` + `tzdata` floor (`python-dateutil` stays + transitive-only); the probe pins gap/fold/dialect/zone behavior so + upgrades re-probe (both strict xfails fail loudly if a future + APScheduler fix tempts re-opening that candidacy, and the croniter + pins — including the exact version assertion — catch drift). + +## Gate 3 — Scheduling state-model probe + +`probes/deployment_scheduling_verify/test_schedule_state_model.py`: +pure-stdlib reference model (injected clock, scripted executor outcomes, +fault injection, ownership lock) + **31 tests, all passing** in the repo +env: + +```powershell +uv run pytest -q ` + probes/deployment_scheduling_verify/test_schedule_state_model.py ` + -p no:cacheprovider -o addopts="" +``` + +The calendar is abstract (`next_after`/`prev_before` only — deliberately no +enumeration seam), so this pressures the state rules, not date math. Each +test is a short timeline, no sleeps. Coverage maps to the assignment bullets: + +- `overlap=skip|parallel × misfire=skip|latest` — all four combinations + (incl. parallel limits + limit lowering + parallel×latest catch-up) +- latest-means-ONE-candidate + supersession + no-double-admit at the + boundary tick + newer-due-never-touches-admitted-runs + timely admission + supersedes older held candidates +- manual runs and other schedules' runs excluded from per-schedule overlap +- run identities are store-backed and survive restart (no `_seq`-style + in-memory counter reuse — this was a real reference-model bug, fixed + with a regression test) +- terminal overlap skips never reappear after restart/catch-up +- terminal overlap skips never reappear after restart/catch-up +- per-schedule limits count interrupted runs; waiting interruptions hold no + task slot; resume requires a task slot +- explicit pause is not downtime (pause excludes the interval; enabled + downtime catches up) via `resume_schedule`/`edit_schedule` semantics +- edit discards candidates without backfill; delete clears pending, keeps + history, never cancels active runs, ids not reusable +- restart persists candidate + consumed progress; rollback never re-admits +- 3-year per-minute downtime: ≤ `SCAN_CAP + 2` source calls, one + interval-summary row, ≤1 admission (capacity) or exactly 1 held candidate +- round-robin fairness across schedules; capacity-deadline expiry for skip + vs hold-one for latest; capacity-wait within allowance (undecided, not + consumed) then admit-or-expire +- fault before AND after admission (no dispatch; recovery materializes + the view and flags pending-dispatch, the next poll dispatches through + capacity checks exactly once — recovery NEVER executes), fault after + materialize, fault after complete and after interrupt (recovery + reconciles the missing terminal record), fault after resume-mark / + resume-complete / resume-attempt-clear / resume-interrupt before AND + after (attempt identity distinguishes fresh results from stale + checkpoints), crash after dispatch (abandoned FAILED with + external-effects disclosure, no replay), resumed runs re-interrupting + durably +- crash during resume: ACTIVE-attempt marker distinguishes ambiguous + (FAILED, never retried) from merely waiting (resumable) +- preflight rejection invents no run; admitted runs freeze invocation +- second-owner rejection; no-expiry held lock; unsupported locking rejects + startup; corrupt view-without-admission fails closed and blocks the + schedule + +Key state findings for implementation (all encoded in the reference model): + +- **F1 — scan cap + jump rule.** Per poll/schedule: at most `SCAN_CAP` + (100) `next_after` calls, then jump via one `prev_before` (latest) or an + interval-summary + `consumed = now` (skip). This is the executable form of + "never enumerate"; only `prev_before` makes latest bounded (Gate 2). +- **F2 — consumed + candidate are the whole restart story.** + `consumed_through` (latest decided instant) plus at most one pending + candidate reconstruct everything; terminal skips/admissions are never + rebuilt because their instants are ≤ consumed. +- **F3 — ordering inside admission:** recheck → overlap → capacity → + allocate/freeze → persist admission → materialize view → dispatch; the + admitted-history entry belongs to the admission persist, not the view. +- **F4 — resume marker first.** `resume_attempts[run] = ACTIVE` is durable + before re-execution; recovery reconciles a matching stopped result and + fails unmatched ACTIVE attempts closed. Pre-existing + waiting interruptions (no marker) stay resumable — no migration problem. +- **F5 — dispatched RUNNING work at recovery is abandoned** (in-memory + tasks die with the process). Proven undispatched admissions remain pending; + a run without an admission is corrupt and must block its schedule. +- **F6 — one-shot exhaustion is a flag**, set on admission or on + skip-expiry; without it the expiry branch refires every poll. +- **F7 — pause/edit/delete/disable are baseline operations** + (`consumed = max(...)` to now, clear candidates), not filters inside the + poll loop. Resolved edge: administrative disable behaves like pause for + catch-up (the spec defines no separate disable semantics). +- **F8 — timely admission supersedes older held candidates.** A poll that + admits instant N while a candidate for older instant C is held records + C superseded and clears it; the admitted-history entry belongs to the + admission persist (before the view materialization), so an + admission-after crash still reconciles exactly once. +- **F9 — recovery reconciles missing terminal records.** A COMPLETED or + waiting-INTERRUPTED run with no terminal history entry (crash between + state persist and record append) gets exactly one reconciled entry; + re-polling never re-admits because consumed already advanced. A + COMPLETED run with a still-ACTIVE attempt crashed between completion + and attempt-clearing: reconcile to DONE, never re-execute. +- **F10 — run identities are store-backed.** The id counter lives in the + store, not the scheduler instance; restart allocates fresh identities + and both occurrences keep their runs. +- **F11 — recovery never executes.** Runs materialized by recovery are + flagged pending-dispatch: they hold a schedule slot but no task slot, + and the next poll dispatches them through capacity checks (waiting + while full). Pending runs of blocked schedules stay pending. +- **F12 — resume completion is three persists, re-interruption is + durable, and stopped results carry the attempt identity.** Completion + persist, attempt-clear, and history append each have a fault boundary; + a resumed run may interrupt again (own persist, stays resumable). + Every resume attempt takes a store-backed identity at mark time, and + every stopped result it produces echoes that identity back: recovery + matches result to ACTIVE attempt (fresh → DONE + resumable) versus + stale (ambiguous → failed, never retried). Crashes inside completion + windows still fail closed, but a persisted re-interruption is never + mistaken for the previous checkpoint. + +## Sequenced implementation plan + +Conventions: each task lists goal, exact seams/files, test-first source +(port the named probe tests into `tests/scheduling/` — do not copy probe +logic into `src/`), and done criteria. Dependencies noted per phase. +Assumes the Gate 2 calendar policy (croniter owns DST resolution; thin +adapter, no custom skip/filter). New production code lives in a focused +package +`src/wf_scheduling/` (new area; per AGENTS.md prefer packages over flat +files) plus the listed seam edits; no second evaluator, no fake workflow +context, no GraphSourcePath extension. + +### R0 — Review checkpoint (before any production code) + +Reviewers confirm the thin-adapter contract, the croniter pin, and the +`src/wf_scheduling/` package boundary. Gate: this document + green probes. + +### Phase 0 — Calendar adapter (no scheduler yet) + +- **T01 — Occurrence-source seam + croniter adapter.** New + `src/wf_scheduling/calendar.py`: `OccurrenceSource` Protocol + (`next_after`, `prev_before` over aware datetimes), `CronSource` + (5-field cron + IANA zone; converts now→zone, queries, returns UTC) + and `OneShotSource` (offset-required, naive rejected). Deliberate + dialect: croniter Unix convention (numeric `0` AND `7` = Sunday; + `day_or=True` selected — NOT APScheduler's Monday-first). Thin by + construction: no calendar correction of its own — forward queries + start from the last-consumed instant, catch-up results compare + against the same watermark (iteration is exclusive both directions, + pinned), `CroniterBadDateError` maps to exhausted, + `CroniterBadCronError` at construction is a definition rejection, + and any backward/non-progressing result fails visibly. Add + `croniter==6.2.4` and a `tzdata` floor to dependencies + (`python-dateutil` remains transitive-only). Port: all Part A + `test_calendar_probe.py` tests into adapter-level tests (the two + Part B strict xfails stay as APScheduler rejected-candidate pins). + Done: adapter suite green; impossible schedules map to exhausted + (no `CroniterBadDateError` leaks). +- **T02 — Occurrence identity + UTC persistence helpers.** + `src/wf_scheduling/occurrences.py`: identity = `(schedule_id, resolved_utc)`; + `occurrence_id` derived deterministically from that pair (document the + derivation; spec exposes schedule_id/occurrence_id/scheduled_at); + monotonic-clock sleep vs wall-clock eligibility split; rollback guard + (`resolved <= consumed` never re-admitted). Port: calendar uniqueness/ + increasing tests plus the reference-model rollback test. Depends: T01. + +### Phase 1 — Expression seam (review checkpoint R1) + +- **T03 — Extract shared traversal behind a typed source resolver.** + New `src/wf_core/runtime/input_sources.py` hosting the `SourceResolver` + Protocol; refactor `src/wf_core/runtime/input_bindings.py:23-114` so path + resolution goes through it (graph root stays a resolver argument — no + `GraphSourcePath` change); unify the sibling recursions + (`src/wf_core/validation/steps.py:242-284`, + `src/wf_api/input_expressions.py:99-253`) onto one traversal + budget + check. Existing node/subgraph/interrupt call sites + (`src/wf_core/runtime/ops/nodes.py:48-66`, + `src/wf_core/runtime/subgraphs.py:139-145`, + `src/wf_core/runtime/ops/interrupts.py:26-43`) pass the graph resolver. + Done: no behavior change; full existing suite green. New tests: resolver + unit tests (graph vs occurrence resolvers over one traversal). +- **T04 — Typed occurrence expression kind.** + New `OccurrenceExpression{field: schedule_id|occurrence_id|scheduled_at}` + in `src/wf_core/models/input_bindings.py` WITHOUT touching + `src/wf_core/paths.py` roots; schedule-side binding list type + (`ScheduleInputBindings`) reusing `validate_input_expression_limits`; + resolution via an occurrence resolver (never via faked graph `context`). + Strict-JSON/target-conflict/schema validation unchanged. Tests: JSON + round-trip, missing-field, graph-only-path, over-budget, conflicting + target, invalid-resolved-input — port the contract pins from + `probes/.../test_expression_contract_probe.py` (8 tests documenting the + current union/budget/strict-JSON/overlap/roots behavior) and extend them + to the new kind. Depends: T03. +- **R1 — review:** resolver Protocol, occurrence kind, budget parity. + +### Phase 2 — Durable admission representation (review checkpoint R2) + +- **T05 — Admission record + in-flight run view.** + Extend `src/wf_artifacts/runs/models.py`: admitted/in-flight status + alongside stopped statuses (never fabricate checkpoint/trace/output for + unknown outcomes); preassigned run identity from a store-backed counter + (never an instance counter — restart must not reuse identities, per + F10); admission persists the pinned environment + (`PinnedRunEnvironment`: deployment + root/child artifacts), resolved + input, limits (`max_steps`), deployment revision, and the occurrence's + resolved UTC instant; run + inspection distinguishes admitted vs stopped-with-checkpoint. + Atomic-per-file writes stay; multi-file authority order per F3 (the + occurrence is decided at the admission persist: history entry, + candidate clearing, consumed progress, and one-shot exhaustion all + belong to it). Port: fault-before/after-admission and + admission-vs-view recovery tests (reference model, minus executor). + Depends: T02. +- **T06 — Admission path through the run API.** + Rework `WorkflowRunApi.run_deployment` (`src/wf_api/runs.py:77-134`) + into recheck → allocate/freeze → persist admission → materialize → + dispatch-captured → persist stopped → reconcile, reusing the resolved-input + signature (no second API). Manual runs keep the current call shape; + existing manual run/resume tests must pass unchanged (regression gate). + Depends: T05. **R2 — review:** admission ordering + inspection contract. + +### Phase 3 — Scheduler core (review checkpoint R3) + +- **T07 — Schedule/deployment-revision models + file store.** + `src/wf_scheduling/models.py` (schedule, revision, overlap/misfire, + `max_active_runs`, lateness allowance, zone, pause/delete/disable flags, + `exhausted` flag per F6, no id reuse) and `src/wf_scheduling/store.py` + (`FileScheduleStore`: schedules, at-most-one candidate, consumed + progress, interval summaries, occurrence history with cursor pagination + over `(resolved_utc, occurrence_id)` + limit). Add deployment revision + (or content hash) to `WorkflowDeployment` + (`src/wf_artifacts/models.py:215`, `store.py:113`) for edit-race + rechecks. Port: edit/delete/pause/disable/no-backfill/id-reuse tests. + Inspection payload carries resolved-UTC/admission/actual-start times, + revision, run id, and failure/skip reason (spec lines 263-272). +- **T08 — Poll loop: overlap/misfire/candidates/fairness/capacity.** + `src/wf_scheduling/poll.py`: F1 scan-cap rule, latest-only coalescing + + supersession records, overlap-before-capacity precedence, terminal skips, + capacity-deadline expiry (skip) vs hold-one (latest), round-robin + fairness, per-poll batch bounds, task-slot accounting (interrupted and + pending-dispatch runs hold schedule slots only), plus the + pending-dispatch sweep (dispatch recovery-materialized runs through + capacity checks; never inside recovery — F11). Port: + matrix/coalescing/fairness/capacity/pending-dispatch reference tests. + Depends: T02, T07. + **R3 — review:** poll semantics vs reference model. + +### Phase 4 — Resume safety, recovery, ownership (review checkpoint R4) + +Safety and recovery land BEFORE anything enables dispatch. + +- **T09 — Durable resume-attempt marker + granular completion.** + `resume_run` (`src/wf_api/runs.py:136-224`) persists ACTIVE with a + store-backed attempt identity before re-executing; completion, + attempt-clearing, and history recording are separate persists with a + fault boundary between each pair; every stopped result the attempt + produces echoes the attempt identity back; a resumed run may + interrupt again (durable re-interruption, stays resumable); waiting + interruptions (no marker) stay resumable across restart. Port: + crash-during-resume, resume-complete/attempt-clear/interrupt + before+after fault, attempt-identity match/mismatch, and + re-interruption tests. Depends: T06. +- **T10 — Startup recovery + reconciliation.** + Recovery under exclusive ownership, and recovery NEVER executes: it + materializes missing views as pending-dispatch (dispatched later only + via the poll sweep), fails abandoned/ambiguous runs with + external-effects disclosure (no replay), matches stopped results to + the ACTIVE attempt by identity (fresh → DONE + resumable; stale → + ambiguous FAILED, never retried), reconciles missing terminal + records and COMPLETED-with-ACTIVE attempts (to DONE, never re-run), + fails corrupt views closed + blocks the schedule, and preserves + stopped interruptions in their slots. Port: all recovery reference + tests. Depends: T06, T08, T09. +- **T11 — Exclusive file-store ownership.** + Held cross-process lock (no PID file, no expiring lease), second-owner + rejection, release on death, startup rejection where locking is + unsupported — consistent with + `2026-06-09-store-transaction-boundary.md` (no ad-hoc lock files without + a tested design). Start with a locking-design spike: candidates are a + dedicated lock file held via `msvcrt.locking` (Windows) / + `fcntl.flock` (POSIX) in one new module, vs a `portalocker`-style + dependency; pick after a Windows crash-hold-release test. Port: + ownership reference tests. Depends: T07. + **R4 — review:** fault-injection suite (every boundary before/after). + +### Phase 5 — Lifecycle enablement and administration surface + +- **T12 — Shutdown drain + lifecycle (GATED: only after R4 passes).** + Opt-in scheduler composition on `WorkflowServer` + (`src/wf_server/context.py:279-341`; startup in `src/wf_server/cli.py`; + transport hooks in `src/wf_transport_rpc_http/app.py:27-68`): + explicitly-enabled flag, stop-admission-first + grace-period drain + (`scheduler_drain_grace_s` config) with cancellation/failure recording. + Dispatch must not be enabled before resume safety (T09), recovery + (T10), and ownership (T11) land. Port: paused/deleted-schedule + completion/resume reference tests; drain and bounded-concurrency tests + are NEW (no probe source — the reference model has no server). + Depends: T08, R4. + +- **T13 — API + Python client.** + New `WorkflowApi` schedule methods (`create/get/list/update/pause/resume/ + delete_schedule` — public names finalized in this task, not assumed to + exist) + paginated occurrence inspection (pending, coalesced/superseded, + skipped-overlap, skipped-misfire, preflight-rejection, + admitted/running, interrupted, completed, failed) through + `wf_api/service.py` + `wf_api/surface.py`, JSON-RPC transport + (`wf_transport_rpc_http`), and `wf_client` (`protocols.py`, + `deployments.py`, `app.py`, `runs.py`); revision-checked edits; run + limits incl. schedule `max_steps` reusing manual-run validation; + inspection without a checkpoint must work (admitted runs have none). + Client round-trip/pagination tests are NEW (no probe source — the + reference model has no transport). Depends: T08, T10. +- **T14 — Docs + seam comments + probe retirement.** + Update live docs per `docs/AGENTS.md` (roadmap pointers, no narrative + bloat), add code-seam comments where docs describe behavior, delete + `probes/deployment_scheduling_verify/` once production tests subsume it. + Markdown lint on changed files only. Depends: everything. + +Non-goals (unchanged): wait nodes, distributed workers, automatic +failed-run retries, replay-all bursts, generic `Runtime[UserContext]`, +exactly-once external effects, multi-writer file stores beyond scheduler +ownership. + +## Remaining concerns + +1. No product decisions block Phase 0. The 60-second default lateness + allowance is approved (per-schedule configurable). +2. `CroniterBadDateError` from impossible-schedule search and naive-time + passthrough are adapter-level (map to exhausted / reject; no spec + change needed). +3. Locking mechanism (T11) needs a Windows-tested design before code, per + the store transaction boundary — flagged as a task, not a decision. +4. Probes use scripted/abstract occurrence sources; production cron parsing + arrives via T01's adapter, already pinned by the calendar probe. + +## Planning bundle files + +- `docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md` — + updated worktree specification incorporating the approved calendar policy +- `probes/deployment_scheduling_verify/README.md` — new (disposable label; + isolated-env + repo-env run commands) +- `probes/deployment_scheduling_verify/test_calendar_probe.py` — new + (16 passed, 2 strict xfailed in the isolated env: Part A pins the + croniter contract, Part B keeps APScheduler rejected-candidate + evidence; `importorskip` keeps default repo collection green) +- `probes/deployment_scheduling_verify/test_schedule_state_model.py` — new + (31 passed in the repo env) +- `probes/deployment_scheduling_verify/test_expression_contract_probe.py` — + new (8 passed in the repo env; pins the current expression contract for + Phase 1) +- `docs/superpowers/plans/2026-09-09-deployment-scheduling-implementation-plan.md` + — this file (new; reviewed three times — first pass B1–B8/A1–A3, + second pass (identity/ordering fixes), third pass: skip rule corrected + to expanded sets with work-bounded forward/backward forms, attempt + identity for stopped results, lifecycle enablement gated after + safety/recovery/ownership; fourth pass: custom skip/filter machinery + retired — croniter owns DST resolution, thin-adapter contract with + last-consumed watermark, `day_or=True` selected, resolved-UTC + identity throughout) diff --git a/docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md b/docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md new file mode 100644 index 00000000..2f08c5d2 --- /dev/null +++ b/docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md @@ -0,0 +1,330 @@ +# Deployment Scheduling + +Status: draft for review; not implemented. + +## Purpose and scope + +A schedule starts ordinary runs of a deployment without a connected client. +The explicitly enabled scheduling service lives in the workflow server. It is +not the core runtime's frame scheduler and introduces no workflow node type. + +First slice: one-shot and recurring cron schedules, durable admission, +coalesced missed-start recovery, bounded parallel runs, occurrence inspection, +and API/Python-client administration. Wait nodes, +distributed workers, automatic execution retries, and arbitrary in-flight +checkpoint recovery are outside this slice. + +The existing file-store contract remains single-process. See +[store transactions](2026-06-09-store-transaction-boundary.md). + +## Decided behavior + +- A schedule follows its deployment. At admission, capture its configuration + revision, resolved input, limits, deployment, and pinned artifact tree. + Later edits cannot change that occurrence or its run. +- Overlap is per schedule, not per deployment. With default `overlap="skip"`, + an unfinished scheduled run, including an interrupted run awaiting input, + blocks another occurrence from that schedule. Manual runs and other + schedules do not participate in that overlap check. +- `overlap="parallel"` admits independent runs up to a required positive + `max_active_runs` limit per schedule, subject to server capacity. Admitted, + running, and interrupted runs all count toward this limit. Do not store + only one active run ID per schedule. Queue and replace-running policies + are not included. +- Default `misfire="skip"` drops missed starts. Optional `misfire="latest"` + means "run as soon as possible after a missed start": retain at most one + latest unadmitted candidate per schedule. It is not replay-all/burst mode. +- Pausing stops future admission, not an active run. Resuming selects the next + future occurrence; paused times are not replayed. +- Deleting a schedule stops future admission. Existing runs and occurrence + history survive, including their schedule identity. Active runs continue. + Schedule identifiers must not be reused to annex old history. +- A known completed or failed run releases overlap. After exclusive startup, + an abandoned in-flight execution becomes failed without retrying its + occurrence. Future occurrences may proceed. The failure must disclose that + external effects may already have occurred; it does not assert rollback or + remote cancellation. +- A durably interrupted execution is not abandoned. It remains resumable and + continues to occupy its schedule's overlap/concurrency slot after restart. + +## Time and triggers + +Cron uses an explicit IANA time zone, default UTC. One-shot timestamps must +include an offset. Persist occurrence instants as UTC timestamps; retain the +cron time-zone name in the definition. Reject invalid zones and naive times. + +Proposed default lateness allowance: 60 seconds, configurable per schedule as +a non-negative finite duration. During uninterrupted operation, a due instant +within the allowance can be admitted. Older instants and enabled times missed +while the server was unavailable follow the configured misfire policy. +With `skip`, startup selects the next future instant regardless of lateness +allowance, and an expired one-shot becomes exhausted. With `latest`, startup +retains the latest missed instant, including an expired one-shot, for prompt +admission subject to overlap and capacity. Never enumerate years of missed +occurrences on startup; find the latest eligible instant with bounded library +queries and record a skipped/coalesced interval summary instead. + +### Missed starts versus overlap + +`latest` has no age expiry while the schedule remains enabled. Its pending +candidate survives restart without becoming a run until admission. A newer +due instant replaces an older unadmitted candidate, with an inspectable +superseded/coalesced reason; admission always uses that candidate's +resolved occurrence instant, not the current clock time. Once admitted, +an occurrence is immutable +and can never be superseded or replayed by this policy. + +For example, an hourly schedule returning at 12:20 after missing 10:00, +11:00, and 12:00 offers one 12:00 occurrence, not three runs. If global +capacity remains unavailable until 13:00, the pending candidate becomes +13:00. A tick at exactly 13:00 must not independently admit both candidates. + +Overlap decisions take precedence over waiting for server capacity. With +`overlap="skip"`, a candidate examined while that schedule has an unfinished +run is terminally skipped-overlap, not held until the run finishes. With +`parallel`, reaching `max_active_runs` similarly produces skipped-overlap. +Neither policy resurrects that skipped instant as a later catch-up candidate. +Global execution-capacity shortage instead leaves a `latest` candidate pending; +with `skip`, it expires when its lateness allowance is exceeded. + +Explicit pause is not downtime. Pause clears unadmitted candidates and excludes +the paused interval from catch-up under both policies. Resume starts from the +next future instant. Definition edits discard unadmitted candidates from the +old revision and begin the new revision at the edit time; creation and edits +do not backfill time before that revision. Deletion clears pending candidates +without touching admitted runs or retained history. An enabled one-shot missed +during downtime can catch up with `latest`; one missed while paused cannot. + +Persist consumed/superseded interval progress with candidate selection so a +restart cannot reconstruct a terminally skipped or already-admitted candidate. +Pending selection and immutable admission are different states: only admission +freezes input and the deployment/artifact snapshot. + +Task Scheduler is inspiration, not a compatibility target. Microsoft's +[StartWhenAvailable documentation](https://learn.microsoft.com/en-us/windows/win32/taskschd/tasksettings-startwhenavailable) +describes delayed starts, while its +[instance policy documentation](https://learn.microsoft.com/en-us/windows/win32/api/taskschd/ne-taskschd-task_instances_policy) +separately defines parallel, queue, ignore-new, and stop-existing. Those pages +do not specify our latest-candidate supersession rule. We choose that rule +explicitly and do not copy Windows' documented default ten-minute delay. + +Occurrence identity is the schedule identity plus the resolved UTC instant. +Clock rollback cannot admit an already-consumed instant again. Use a monotonic +clock for sleeping, and a wall clock for calendar eligibility. A forward jump +applies lateness policy, not unconditional replay. + +Use a library for calendar calculation. Decided: croniter behind a thin +next/previous-occurrence adapter (`next_after` / `prev_before` over +schedule-zone instants, UTC at the boundary), without adopting any job +store or executor. Probes (see the implementation plan) disqualified +APScheduler 3.x triggers: no bounded latest-missed seam, phantom wall +times across DST gaps, and replayed UTC minutes after fall folds. The +adapter converts the query instant into the schedule's named zone, asks +croniter for the next/previous occurrence, and converts the result to +UTC. It applies no calendar correction of its own: croniter owns DST +resolution, including nonexistent and repeated local times. + +Probed and pinned on Python 3.14 (`croniter==6.2.4`, `tzdata` floor; +`python-dateutil` only transitive): some nonexistent scheduled wall +times resolve forward — daily 02:30 on the spring-gap day resolves to +03:00-04:00 the same day — and repeated times surface as distinct UTC +occurrences (both 01:30s on the fall-fold day). Occurrence identity is +`(schedule_id, resolved UTC instant)`. The definition's cron +expression, time-zone name, and admission snapshots are retained, but +the platform claims no finer intended-wall-time provenance than the +library supplies. Iteration is exclusive in both directions (a query +from exactly a due instant returns the neighboring occurrence), so the +adapter queries forward from the last-consumed instant and compares +catch-up results against the same watermark: a due occurrence is +admitted exactly once. Impossible schedules surface promptly as a +documented library exhaustion error, never as silent no-progress; +search exhaustion or backward/non-progressing results fail visibly. +Five-field cron scope; the adapter itself rejects invalid zones and +naive timestamps (the library does not reject naive). Re-probe on any +calendar-dependency upgrade; the strict failure pins guard the +APScheduler behaviors we rejected. + +The adapter uses Unix cron dialect explicitly: numeric `0` and `7` +both mean Sunday (not Monday-first), and day-of-month/day-of-week +matching uses croniter's standard `day_or=True` (Unix OR), exposed +explicitly rather than as an undocumented promise. No jitter or +extended trigger combinations in this slice. + +Alternatives considered: APScheduler 3.x triggers were probed and rejected +(bounded latest-missed lookup impossible, DST-gap phantoms, fold replay); +a full scheduling framework owns useful job machinery but would create a +second persistence/execution lifecycle alongside our run API. Prefer +croniter iteration plus our existing run lifecycle. + +Sources checked on 2026-09-08: + +- [APScheduler 3.x cron trigger](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/cron.html) +- [APScheduler date trigger](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/date.html) +- [croniter project documentation](https://pypi.org/project/croniter/) + +## Input authoring and serialization + +A schedule stores instructions for building a future workflow input object. +The deployment-run API still receives resolved data, not expression objects. +Do not add parallel raw-input and occurrence-binding mechanisms. + +Reuse literal, object, array, target-path, strict-JSON, and expression-budget +semantics from the existing input-binding system. Add a typed occurrence +reference for the schedule environment; graph references to input/state/context +are invalid here. Do not extend GraphSourcePath with schedule-only roots. +Existing expressions compose data; they do not implement date formatting, +arithmetic, template evaluation, Python execution, or arbitrary transforms. + +Proposed persisted binding example, pending concrete model names: + +```json +{ + "input_bindings": [ + {"target": "team", "value": "engineering"}, + { + "target": "report_time", + "expression": {"kind": "occurrence", "field": "scheduled_at"} + } + ] +} +``` + +Occurrence references initially expose schedule_id, occurrence_id, and +scheduled_at. Date-time values serialize as UTC RFC 3339 strings. The admitted +run's resolved input is persisted once and never re-evaluated on restart. +Validate target conflicts, expression bounds, source fields, and the resulting +workflow input schema. Recheck the current deployment contract at admission; +an edit may have changed the expected input since schedule creation. + +Extract the shared composition traversal and limits behind a typed source +resolver seam. Keep graph and schedule source models distinct. Do not copy +the recursive evaluator into a second package or use fake graph context to +smuggle occurrence values into graph-path evaluation. + +Generic host Runtime[ContextT] remains a separate future feature. Schedule +provenance is persisted platform metadata, not an arbitrary host object and +not a new graph-visible context namespace. Child workflows receive business +values through their declared input bindings as before. + +## Durable admission and recovery + +Current seams needing change: + +- wf_api/runs.py executes before persisting a stopped run. +- wf_artifacts/runs/models.py permits only stopped summaries, with a required + checkpoint identifier. +- wf_api/run_lifecycle.py assumes an existing checkpoint when updating a run. + +Introduce a durable admission representation with a preassigned run identity. +Run inspection must distinguish an admitted/in-flight run from a stopped run +with a checkpoint. Never fabricate a completed checkpoint, trace, output, or +successful step count for work whose outcome is unknown. + +Required ordering under the single-owner admission lock: + +1. Recheck schedule revision, enabled status, due time, capacity, and overlap. +2. Allocate occurrence/run identities and freeze the invocation data. +3. Atomically persist the authoritative admission record before dispatch. +4. Materialize the run admission view using that same identity. +5. Dispatch the captured invocation without resolving the deployment again. +6. Persist a stopped checkpoint and summary, then reconcile occurrence status. + +The admission record is the recovery authority for partial multi-file writes; +atomic rename is not a transaction across files. A failed durable admission +must never dispatch. Reconciliation is idempotent: it completes missing views, +recognizes durable stopped results, or marks abandoned work failed without +redispatch. A persisted interruption must win over stale in-flight metadata. +Corrupt or contradictory records fail closed with diagnostics; do not silently +discard them to clear overlap. + +Resume of a scheduled interrupted run must mark its active attempt durably +before executing again. Otherwise a crash during resume could leave an old +interrupted checkpoint looking safe to retry. The mark carries a +store-backed attempt identity, and every stopped result the attempt +produces echoes that identity back. Recovery matches result to active +attempt: a result belonging to the active attempt is fresh and resumable, +anything else under an active attempt is stale and fails closed without +retry. Recovery must distinguish both cases from an interruption that was +merely waiting across server restart. + +The process must own the store exclusively before recovery. Enforce scheduler +ownership with a held cross-process lock, not a stale PID file or a lease that +can expire while the old owner still runs. Unsupported locking must reject +scheduler startup. Other processes mutating/resuming the same store remain +unsupported. This does not upgrade the rest of the file stores to multi-writer +safety or claim exactly-once external effects. + +## Lifecycle, administration, and resource bounds + +Expose create/get/list/update/pause/resume/delete and paginated occurrence +inspection through the workflow API and Python client. Public client names +are finalized in the implementation plan, not treated as existing methods. +Reject stale schedule edits using revisions within the owning process. + +Occurrence inspection distinguishes pending, coalesced/superseded, +skipped-overlap, skipped-misfire, +preflight rejection, admitted/running, interrupted, completed, and failed. +These are platform occurrence states, not business outcomes or core node +outcomes. Preflight rejection does not invent a run that never started. + +Store resolved input and pinned environment with admitted runs. Inspection +includes resolved occurrence instant (UTC), admission time, actual start +when known, schedule +revision, run identity, and failure/skip reason. Preserve existing run limits; +allow a schedule-specific max_steps using the same validation as manual runs. + +Bound active scheduled execution tasks and work per polling batch. Do not +block calendar polling on a long run or create unbounded pending tasks. +Capacity-delayed `skip` occurrences expire at their lateness deadline; `latest` +retains at most one candidate as specified above. Interrupted runs consume +per-schedule active-run slots but no executing-task slot while waiting. +Resumption must acquire a server execution slot before dispatch. Lowering a +schedule's active limit never cancels existing runs; block new admission until +the count drops below the new limit. Polling must be fair across schedules so +a frequently due schedule cannot monopolize available capacity. The concrete +server capacity default is deployment configuration, with deterministic tests +using a small injected limit. + +On shutdown stop admission first and drain active tasks within a configured +grace period. Record cancellation/failure when possible; abrupt termination +uses startup recovery. Paused/deleted schedule definitions must not prevent +run completion or resume from updating retained occurrence history. + +## Verification gates + +Use injected clocks and controlled executors, not real-time sleeps: + +- Cron parsing, time zones, invalid syntax, leap/calendar boundaries, DST gaps + and folds, UTC uniqueness, impossible schedules, and bounded next-time search. +- One-shot success/exhaustion/catch-up; lateness boundary; startup behavior + under both policies; explicit pause/resume exclusion; long downtime without + unbounded enumeration; clock rollback/forward jumps. +- Latest-only coalescing across multiple missed times, candidate persistence, + supersession at the next due instant, no double admission at that boundary, + and no resurrection after overlap skips or admission. +- Literal and nested occurrence expressions round-trip through JSON; missing + fields, graph-only paths, excessive trees, conflicting targets, and invalid + resolved workflow input fail before dispatch. +- Same-schedule overlap across running and interrupted states; manual/other + schedule independence; release on failure/completion and on resumed completion. +- Parallel admission limits, interrupted slot accounting, capacity-delayed + candidates, fair polling, limit edits, and bounded task allocation. Exercise + all four overlap/misfire combinations with controlled executors. +- Edit/repoint/delete races at admission; captured invocation stays unchanged. +- Fault injection before/after every persistence boundary, including resume: + no dispatch before durable admission and no replay after ambiguous execution. +- Recovery preserves stopped interruptions, reconciles terminal results, marks + abandoned attempts failed, and leaves corrupt records visibly blocked. +- Second-owner rejection, lock release on process death, bounded concurrency, + shutdown drain, and full occurrence history after schedule deletion. +- Public API/client round trips, pagination, inspection without a checkpoint, + and existing manual run/resume behavior remain valid. + +## Review before implementation planning + +User policy decisions above are settled. Review the proposed 60-second +allowance, expression seam, and admission/recovery representation together. +The calendar-library probe is an explicit gate, not a claimed passing test. +After approval, create a sequenced implementation plan with fault-injection +tests before enabling scheduling in the server. WaitNode is a later contract +that may reuse timed admission but must persist its own suspended execution. diff --git a/probes/deployment_scheduling_verify/README.md b/probes/deployment_scheduling_verify/README.md new file mode 100644 index 00000000..bf185a8d --- /dev/null +++ b/probes/deployment_scheduling_verify/README.md @@ -0,0 +1,40 @@ +# DISPOSABLE VERIFICATION PROBE — NOT PRODUCTION CODE + +This directory holds throwaway verification probes for the +deployment-scheduling slice +(`docs/superpowers/specs/2026-09-08-deployment-scheduling-design.md`). +They are evidence-gathering scripts, not a production implementation: + +- `test_calendar_probe.py` — calendar-library probe (16 passed, 2 strict + xfailed: Part A pins the croniter contract, Part B keeps APScheduler + rejected-candidate evidence). Requires `croniter==6.2.4`, `tzdata` + (`apscheduler==3.11.0` for Part B only) on Python 3.14. + Run it from an isolated project (NOT the repo env, which does not + depend on croniter; the file `importorskip`s itself elsewhere): + + ```powershell + $probe = "C:\Users\Admin\AppData\Local\Temp\opencode\sched-cal-probe" + $file = "\probes\deployment_scheduling_verify\test_calendar_probe.py" + uv run --project $probe python -m pytest $file -q -p no:cacheprovider ` + -o addopts="" + ``` + +- `test_schedule_state_model.py` — pure-stdlib reference model of the + scheduling state machine (25 tests: overlap x misfire, coalescing, + slots, pause, faults, ownership). Runs in the repo env: + + ```powershell + uv run pytest -q probes/deployment_scheduling_verify/test_schedule_state_model.py + ``` + +- `test_expression_contract_probe.py` — pins the current input-expression + contract for the Phase 1 implementer (8 tests). Runs in the repo env: + + ```powershell + uv run pytest -q probes/deployment_scheduling_verify/test_expression_contract_probe.py + ``` + +Do not import these probes from `src/`. Do not copy their logic into +production without going through the sequenced implementation plan at +`docs/superpowers/plans/2026-09-09-deployment-scheduling-implementation-plan.md`. +Delete this directory once the slice is implemented. diff --git a/probes/deployment_scheduling_verify/test_calendar_probe.py b/probes/deployment_scheduling_verify/test_calendar_probe.py new file mode 100644 index 00000000..b1919708 --- /dev/null +++ b/probes/deployment_scheduling_verify/test_calendar_probe.py @@ -0,0 +1,413 @@ +# DISPOSABLE CALENDAR-LIBRARY PROBE — NOT PRODUCTION CODE. +# See README.md in this directory. Requires croniter==6.2.4 and tzdata +# (plus apscheduler==3.11.0 for the Part B rejected-candidate evidence) +# on Python 3.14 (isolated env, not repo env). +"""Pin the croniter 6.2.4 calendar contract for deployment scheduling. + +Policy: croniter owns calendar calculation, including DST resolution. +The adapter consumes it thinly — convert the query instant into the +schedule's named zone, ask for the next/previous occurrence, convert +the result to UTC — and applies no correction of its own. Each test +prints OBSERVED lines pinning 6.2.4 behavior. After any +calendar-dependency upgrade a failure IS the finding: re-probe, do not +hand-roll around it. + +Part A pins the chosen-library contract (shipping acceptance). +Part B preserves rejected-candidate evidence (NOT acceptance). + +Run: + uv run --project python -m pytest + -q -p no:cacheprovider -o addopts="" +""" + +from __future__ import annotations + +import time +from datetime import datetime, timedelta, timezone + +import pytest + +# This probe runs ONLY in the isolated calendar env (see README.md). +# Skip — do not error — when collected by a default repo test run. +pytest.importorskip("apscheduler", reason="isolated calendar-probe env only") +pytest.importorskip("croniter", reason="isolated calendar-probe env only") + +from apscheduler.triggers.cron import CronTrigger # noqa: E402 (Part B only) +from croniter import ( # noqa: E402 + CroniterBadCronError, + CroniterBadDateError, +) +from croniter import ( + croniter as Croniter, +) + +try: # noqa: E402 + from zoneinfo import ZoneInfo +except ImportError: # pragma: no cover + from backports.zoneinfo import ZoneInfo # type: ignore[no-redef] + +from importlib.metadata import version as _pkg_version # noqa: E402 + +import apscheduler # noqa: E402 + +CRONITER_VERSION = _pkg_version("croniter") + +print(f"croniter=={CRONITER_VERSION}") +print(f"apscheduler=={apscheduler.__version__} (rejected candidate, Part B only)") + +UTC = timezone.utc +HCMC = ZoneInfo("Asia/Ho_Chi_Minh") +NYC = ZoneInfo("America/New_York") + +# 2026 DST transitions (US): spring forward 2026-03-08 02:00 -> 03:00, +# fall back 2026-11-01 02:00 -> 01:00. + + +def utc(*args) -> datetime: + return datetime(*args, tzinfo=UTC) + + +# --------------------------------------------------------------------------- +# Part A — chosen-library contract (croniter; shipping acceptance) +# --------------------------------------------------------------------------- + + +def test_versions_pinned(): + print(f"OBSERVED croniter version={CRONITER_VERSION}") + print(f"OBSERVED apscheduler.__version__={apscheduler.__version__}") + # Exact pin: a calendar-dependency upgrade must fail here deliberately, + # forcing a re-probe before any pin update. + assert CRONITER_VERSION == "6.2.4" + assert apscheduler.__version__.startswith("3.") + + +def test_utc_daily_next_is_strictly_increasing_and_unique(): + it = Croniter("30 9 * * *", utc(2026, 9, 1, 0, 0, 0)) + seen: set[datetime] = set() + prev = None + for _ in range(10): + nxt = it.get_next(datetime) + assert nxt.tzinfo is not None + as_utc = nxt.astimezone(UTC) + assert as_utc not in seen, f"duplicate UTC instant {as_utc}" + seen.add(as_utc) + if prev is not None: + assert as_utc > prev, "not strictly increasing" + print(f"OBSERVED utc-daily next={as_utc.isoformat()}") + prev = as_utc + assert len(seen) == 10 + + +def test_hcmc_daily_converts_and_hourly_counts(): + # Asia/Ho_Chi_Minh is UTC+7 with no DST. 09:30 local == 02:30 UTC. + nxt = Croniter("30 9 * * *", datetime(2026, 9, 1, 0, 0, tzinfo=HCMC)).get_next( + datetime + ) + as_utc = nxt.astimezone(UTC) + print(f"OBSERVED hcmc next local={nxt.isoformat()} utc={as_utc.isoformat()}") + assert (as_utc.hour, as_utc.minute) == (2, 30) + assert as_utc.date().isoformat() == "2026-09-01" + # No-DST zone: hourly occurrence count over two 48-hour windows + # (March and November, starting off-tick) must be exactly 48 each. + for label, start in ( + ("march", utc(2026, 3, 7, 17, 30, 0)), + ("november", utc(2026, 10, 31, 17, 30, 0)), + ): + it = Croniter("0 * * * *", start.astimezone(HCMC)) + count = 0 + while True: + hit = it.get_next(datetime).astimezone(UTC) + if hit >= start + timedelta(hours=48): + break + count += 1 + assert count < 60 + print(f"OBSERVED hcmc hourly count window={label} count={count}") + assert count == 48, f"{label}: no-DST zone must yield exactly 48, got {count}" + + +def test_expression_forms_wildcard_step_list_range(): + """Wildcards, steps, lists, and ranges resolve through plain + get_next on an ordinary day (no DST involved).""" + cases = { + "* * * * *": (datetime(2026, 9, 8, 0, 0, tzinfo=UTC), "2026-09-08T00:01:00"), + "5/15 * * * *": (datetime(2026, 9, 8, 0, 0, tzinfo=UTC), "2026-09-08T00:05:00"), + "0,30 1-2 * * *": ( + datetime(2026, 9, 8, 0, 0, tzinfo=UTC), + "2026-09-08T01:00:00", + ), + "*/20 1-3 * * *": ( + datetime(2026, 9, 8, 1, 50, tzinfo=UTC), + "2026-09-08T02:00:00", + ), + } + for expr, (start, want) in cases.items(): + nxt = Croniter(expr, start).get_next(datetime) + print( + f"OBSERVED croniter {expr!r} from {start.isoformat()} -> {nxt.isoformat()}" + ) + assert nxt.astimezone(UTC).isoformat() == want + "+00:00" + + +def test_weekday_names_and_numbers_unix(): + """Pin croniter 6.2.4 weekday dialect (re-probe on upgrade). Unix + convention: numeric 0 AND 7 mean Sunday; 1/mon mean Monday.""" + monday = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) # a Monday + cases = { + "0 12 * * 0": 6, # Sunday + "0 12 * * 7": 6, # Sunday (alias) + "0 12 * * 1": 0, # Monday + "0 12 * * mon": 0, + "0 12 * * sun": 6, + } + for expr, want_wd in cases.items(): + nxt = Croniter(expr, monday).get_next(datetime) + print( + f"OBSERVED croniter {expr!r} -> {nxt.date().isoformat()} weekday={nxt.weekday()}" + ) + assert nxt.weekday() == want_wd, f"{expr}: want weekday={want_wd}" + assert Croniter("0 12 * * 0", monday).get_next(datetime) == Croniter( + "0 12 * * sun", monday + ).get_next(datetime) + + +def test_dom_dow_day_or_true_is_selected(): + """Day-of-month/day-of-week uses croniter's standard day_or=True + (Unix OR). The alternative (AND) is pinned for reference only.""" + start = datetime(2026, 9, 1, 0, 0, tzinfo=UTC) + it_or = Croniter("0 12 13 * fri", start, day_or=True) + hits_or = [it_or.get_next(datetime).date().isoformat() for _ in range(4)] + print(f"OBSERVED croniter day_or=True hits={hits_or}") + # Fridays plus the 13th (a Sunday): classic Unix OR. + assert hits_or == ["2026-09-04", "2026-09-11", "2026-09-13", "2026-09-18"] + it_and = Croniter("0 12 13 * fri", start, day_or=False) + first_and = it_and.get_next(datetime).date() + print( + f"OBSERVED croniter day_or=False first={first_and.isoformat()} (not selected)" + ) + assert (first_and.day, first_and.weekday()) == (13, 4) + + +def test_zone_conversion_is_the_callers_job(): + """croniter iterates in the tz of the supplied datetime and performs + no conversion: a UTC start yields UTC results. The adapter converts + now into the schedule zone before querying and back to UTC after.""" + utc_start = Croniter( + "30 2 * * *", datetime(2026, 3, 7, 12, 0, tzinfo=UTC) + ).get_next(datetime) + print( + f"OBSERVED croniter utc-start gap next={utc_start.isoformat()} (no zone conversion)" + ) + assert utc_start.tzinfo is UTC + hcmc_start = Croniter( + "30 9 * * *", datetime(2026, 9, 1, 0, 0, tzinfo=HCMC) + ).get_next(datetime) + print(f"OBSERVED croniter hcmc-start next={hcmc_start.isoformat()}") + assert hcmc_start.utcoffset() == timedelta(hours=7) + + +def test_gap_nonexistent_wall_times_resolve_forward(): + """Observed 6.2.4 gap behavior: daily 02:30 on 2026-03-08 (a wall + time that never existed in America/New_York) resolves to 03:00-04:00 + the same day. That resolution IS the occurrence — there is no + separate validity concept and no day is skipped over.""" + first = Croniter("30 2 * * *", datetime(2026, 3, 7, 12, 0, tzinfo=NYC)).get_next( + datetime + ) + print(f"OBSERVED croniter gap-day daily-0230 resolves={first.isoformat()}") + assert first.isoformat() == "2026-03-08T03:00:00-04:00" + following = Croniter("30 2 * * *", first).get_next(datetime) + print(f"OBSERVED croniter gap following={following.isoformat()}") + assert following.isoformat() == "2026-03-09T02:30:00-04:00" + + +def test_gap_per_minute_stream_jumps_forward(): + """Per-minute iteration across the spring gap jumps 01:59 EST + straight to 03:00 EDT: no 02:xx wall times are emitted, UTC is + strictly increasing and unique.""" + it = Croniter("* * * * *", datetime(2026, 3, 8, 0, 0, tzinfo=NYC)) + seq = [it.get_next(datetime) for _ in range(300)] + us = [d.astimezone(UTC) for d in seq] + assert all(b > a for a, b in zip(us, us[1:])) + assert len(set(us)) == len(us) + gap_day = [d for d in seq if d.date().isoformat() == "2026-03-08"] + assert all(d.hour != 2 for d in gap_day), "no 02:xx wall times emitted" + assert "2026-03-08T01:59:00-05:00" in (d.isoformat() for d in gap_day) + assert "2026-03-08T03:00:00-04:00" in (d.isoformat() for d in gap_day) + print( + "OBSERVED per-minute gap jump 01:59-05:00 -> 03:00-04:00, " + f"{len(gap_day)} gap-day hits" + ) + + +def test_gap_backward_queries_return_library_resolution(): + """Backward queries resolve the same way: get_prev on the gap day + returns the forward-resolved 03:00-04:00, not a skipped-over day. + Latest-missed on a gap day is that resolved instant.""" + cases = { + "2026-03-08T12:00": "2026-03-08T03:00:00-04:00", + "2026-03-09T00:00": "2026-03-08T03:00:00-04:00", + "2026-03-07T12:00": "2026-03-07T02:30:00-05:00", + "2026-03-09T12:00": "2026-03-09T02:30:00-04:00", + } + for start_iso, want in cases.items(): + start = datetime.fromisoformat(start_iso).replace(tzinfo=NYC) + got = Croniter("30 2 * * *", start).get_prev(datetime) + print(f"OBSERVED croniter gap get_prev from {start_iso} -> {got.isoformat()}") + assert got.isoformat() == want + + +def test_fold_repeated_times_are_distinct_utc(): + """Both 01:30s on 2026-11-01 occur as distinct UTC instants + (05:30Z EDT, then 06:30Z EST).""" + it = Croniter("30 1 * * *", datetime(2026, 10, 31, 12, 0, tzinfo=NYC)) + first = it.get_next(datetime) + second = it.get_next(datetime) + print( + f"OBSERVED croniter fold first={first.isoformat()} second={second.isoformat()}" + ) + assert first.astimezone(UTC) == utc(2026, 11, 1, 5, 30) + assert second.astimezone(UTC) == utc(2026, 11, 1, 6, 30) + + +def test_fold_per_minute_unique_increasing(): + """Per-minute iteration across the fall fold emits both fold hours + with strictly increasing unique UTC instants.""" + it = Croniter("* * * * *", datetime(2026, 10, 31, 20, 0, tzinfo=NYC)) + seq = [it.get_next(datetime) for _ in range(560)] + us = [d.astimezone(UTC) for d in seq] + print(f"OBSERVED croniter fold-span count={len(us)} last={us[-1].isoformat()}") + assert all(b > a for a, b in zip(us, us[1:])), "must be strictly increasing" + assert len(set(us)) == len(us), "UTC identities must be unique" + iso = {d.isoformat() for d in us} + assert "2026-11-01T05:30:00+00:00" in iso, "first 01:30 (EDT) must occur" + assert "2026-11-01T06:30:00+00:00" in iso, "second 01:30 (EST) must occur" + + +def test_latest_missed_after_years_of_downtime(): + """Per-minute schedule, ~3 years of downtime: get_prev answers the + latest missed occurrence directly — no enumeration of missed years.""" + now = utc(2026, 9, 8, 12, 0, 0) + t0 = time.perf_counter() + latest = Croniter("* * * * *", now).get_prev(datetime) + elapsed = time.perf_counter() - t0 + print( + f"OBSERVED croniter get_prev({now.isoformat()})={latest.isoformat()} " + f"elapsed={elapsed:.4f}s" + ) + assert latest.tzinfo is not None, "get_prev must preserve tz-awareness" + assert elapsed < 5, "latest-missed lookup must be bounded" + assert latest <= now + assert (now - latest) < timedelta(minutes=2) + now_nyc = datetime(2026, 9, 8, 12, 0, tzinfo=NYC) + latest_nyc = Croniter("* * * * *", now_nyc).get_prev(datetime) + print( + f"OBSERVED croniter nyc get_prev now={now_nyc.isoformat()} " + f"latest={latest_nyc.isoformat()}" + ) + assert latest_nyc.tzinfo is not None + assert latest_nyc <= now_nyc + assert (now_nyc - latest_nyc) < timedelta(minutes=2) + assert latest_nyc.astimezone(UTC).isoformat() == "2026-09-08T15:59:00+00:00" + + +def test_iteration_is_exclusive_both_directions(): + """Both get_next and get_prev are exclusive of their start: a query + from exactly a due instant returns the neighboring occurrence, not + the instant itself. Adapter consequence (T01, not enforced here): + query forward from the last-consumed instant and compare catch-up + results against the same watermark, so a due occurrence is admitted + exactly once — neither missed by exclusivity nor admitted twice.""" + due = utc(2026, 9, 8, 13, 0, 0) + fwd = Croniter("0 13 * * *", due).get_next(datetime) + back = Croniter("0 13 * * *", due).get_prev(datetime) + print(f"OBSERVED exclusive get_next(due)={fwd.isoformat()}") + print(f"OBSERVED exclusive get_prev(due)={back.isoformat()}") + assert fwd.astimezone(UTC) == utc(2026, 9, 9, 13, 0, 0) + assert (back.year, back.month, back.day, back.hour, back.minute) == ( + 2026, + 9, + 7, + 13, + 0, + ) + just_before = Croniter("0 13 * * *", due - timedelta(seconds=1)).get_next(datetime) + assert just_before.astimezone(UTC) == due, "a tick just before due admits it" + + +def test_impossible_schedule_raises_promptly(): + """February 30th never occurs: the library raises its documented + exhaustion error promptly instead of searching forever. The adapter + maps this to exhausted — exhaustion must never look like progress.""" + start = time.perf_counter() + with pytest.raises(CroniterBadDateError): + Croniter("0 12 30 2 *", utc(2026, 1, 1)).get_next(datetime) + elapsed = time.perf_counter() - start + print( + f"OBSERVED impossible-schedule raised CroniterBadDateError elapsed={elapsed:.3f}s" + ) + assert elapsed < 5, f"search must be bounded, took {elapsed:.1f}s" + + +def test_bad_expressions_rejected_naive_passes_through(): + """Malformed expressions fail at construction with a documented + error. Naive datetimes are NOT rejected by the library — they pass + straight through — so the adapter rejects naive itself per spec.""" + with pytest.raises(CroniterBadCronError): + Croniter("nonsense", utc(2026, 1, 1)) + print("OBSERVED malformed expression rejected with CroniterBadCronError") + naive = Croniter("* * * * *", datetime(2026, 9, 8, 12, 0)).get_next(datetime) + print(f"OBSERVED naive start passes through tzinfo={naive.tzinfo}") + assert naive.tzinfo is None + + +# --------------------------------------------------------------------------- +# Part B — rejected-candidate evidence (APScheduler; NOT acceptance) +# --------------------------------------------------------------------------- + + +@pytest.mark.xfail( + strict=True, + reason="REJECTED CANDIDATE: APScheduler 3.11.0 fabricates a phantom " + "02:30-05:00 for the DST gap. Kept as evidence for the rejection; " + "the chosen library's gap behavior is pinned in Part A. Strict so " + "re-opening the candidacy fails loudly. See implementation plan.", +) +def test_rejected_candidate_gap_phantom(): + # 02:30 does not exist in America/New_York on 2026-03-08. The + # disqualifier is the phantom itself: whatever the library returns + # must not be a nonexistent wall time. + trig = CronTrigger(hour=2, minute=30, second=0, timezone="America/New_York") + nxt = trig.get_next_fire_time(None, utc(2026, 3, 7, 12, 0, 0)) + assert nxt is not None + local = nxt.astimezone(NYC) + print(f"OBSERVED apscheduler dst-gap next local={local.isoformat()}") + assert not ( + (local.year, local.month, local.day) == (2026, 3, 8) and local.hour == 2 + ), f"library must not fabricate a nonexistent wall time, got {local.isoformat()}" + + +@pytest.mark.xfail( + strict=True, + reason="REJECTED CANDIDATE: APScheduler 3.11.0 replays 05:01Z-06:00Z " + "(~59 past minutes) after the fall fold, duplicating UTC occurrence " + "identities. Kept as evidence for the rejection; the chosen " + "library's fold behavior is pinned in Part A. See implementation plan.", +) +def test_rejected_candidate_fold_replay(): + trig = CronTrigger(minute="*", second=0, timezone="America/New_York") + now = utc(2026, 11, 1, 0, 0, 0) + prev = None + seen: set[str] = set() + last: datetime | None = None + for _ in range(450): # spans past 07:00Z: covers BOTH fold hours + nxt = trig.get_next_fire_time(prev, now) + assert nxt is not None + as_utc = nxt.astimezone(UTC) + ident = f"sched-1|{as_utc.isoformat()}" + assert ident not in seen, f"duplicate occurrence identity {ident}" + seen.add(ident) + if last is not None: + assert as_utc > last, "occurrences must be strictly increasing" + last = as_utc + prev, now = nxt, nxt + print(f"OBSERVED fold-span count={len(seen)} unique, last={last.isoformat()}") diff --git a/probes/deployment_scheduling_verify/test_expression_contract_probe.py b/probes/deployment_scheduling_verify/test_expression_contract_probe.py new file mode 100644 index 00000000..7f41baa9 --- /dev/null +++ b/probes/deployment_scheduling_verify/test_expression_contract_probe.py @@ -0,0 +1,138 @@ +# DISPOSABLE EXPRESSION-CONTRACT PROBE — NOT PRODUCTION CODE. +# See README.md in this directory. Runs in the repo env: +# uv run pytest -q probes/deployment_scheduling_verify/test_expression_contract_probe.py +"""Pin the CURRENT input-expression contract that the scheduling slice must +reuse (spec: Input authoring and serialization). + +These tests document what exists today for the Phase 1 implementer: the +closed 4-kind union, budget enforcement point, strict-JSON literals, +target-conflict detection, closed GraphSourcePath roots, and the single +hardcoded graph-context evaluator. A schedule occurrence source must plug +into this traversal (T03/T04 of the implementation plan), not copy it. +""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from wf_core.local_paths import has_overlapping_paths +from wf_core.models.input_bindings import ( + MAX_INPUT_EXPRESSION_DEPTH, + MAX_INPUT_EXPRESSION_NODES, + ArrayExpression, + InputExpressionBinding, + LiteralExpression, + ObjectExpression, + PathExpression, + validate_input_expression_limits, +) +from wf_core.models.json_values import validate_strict_json_value +from wf_core.paths import GraphSourcePath +from wf_core.runtime.input_bindings import ( + resolve_input_expression, + resolve_step_input_bindings, +) + + +def test_expression_union_is_closed_to_four_kinds(): + assert LiteralExpression(kind="literal", value=1).kind == "literal" + assert PathExpression(kind="path", path="input.a").kind == "path" + assert ArrayExpression(kind="array", items=[]).kind == "array" + assert ObjectExpression(kind="object", fields={}).kind == "object" + with pytest.raises(ValidationError): + InputExpressionBinding( + target="x", expression={"kind": "occurrence", "field": "scheduled_at"} + ) # type: ignore[dict-item] + print("OBSERVED occurrence kind rejected: union closed to 4 kinds") + + +def test_budget_constants_and_validator_entry_point(): + assert (MAX_INPUT_EXPRESSION_DEPTH, MAX_INPUT_EXPRESSION_NODES) == (64, 1024) + deep: dict = {"kind": "literal", "value": 0} + for _ in range(MAX_INPUT_EXPRESSION_DEPTH + 5): + deep = {"kind": "array", "items": [deep]} + with pytest.raises(ValueError, match="limit exceeded"): + validate_input_expression_limits(deep) # type: ignore[arg-type] + print("OBSERVED depth budget enforced by validate_input_expression_limits") + + +def test_strict_json_rejects_non_finite_and_non_string_keys(): + with pytest.raises(ValueError): + validate_strict_json_value(float("inf")) + with pytest.raises(ValueError): + validate_strict_json_value({1: "x"}) + assert validate_strict_json_value({"a": [1, None, "x"]}) == {"a": [1, None, "x"]} + print("OBSERVED strict-JSON validator rejects inf and non-string keys") + + +def test_target_conflicts_detected_on_local_paths(): + assert has_overlapping_paths(["a.b", "a.b.c"]) + assert not has_overlapping_paths(["a.b", "a.c"]) + print("OBSERVED overlapping local-path targets detected") + + +def test_graph_source_roots_closed_to_input_state_context(): + assert GraphSourcePath.parse("input.a").root == "input" + assert GraphSourcePath.parse("state.a").root == "state" + assert GraphSourcePath.parse("context.a").root == "context" + with pytest.raises(ValueError): + GraphSourcePath.parse("occurrence.scheduled_at") + print("OBSERVED occurrence root rejected: GraphSourcePath closed") + + +def test_runtime_resolver_composes_literal_object_array_and_paths(): + expr = ObjectExpression( + kind="object", + fields={ + "team": LiteralExpression(kind="literal", value="eng"), + "tags": ArrayExpression( + kind="array", + items=[LiteralExpression(kind="literal", value="a")], + ), + "req": PathExpression(kind="path", path="input.request_id"), + }, + ) + resolved = resolve_input_expression( + expr, + state={}, + workflow_input={"request_id": "r1"}, + context={}, + label="probe", + location="$", + ) + assert resolved == {"team": "eng", "tags": ["a"], "req": "r1"} + print(f"OBSERVED composed resolution -> {resolved}") + + +def test_runtime_resolver_takes_only_graph_context_mappings(): + # There is no source-resolver seam: the only injection point is the + # concrete state/workflow_input/context mappings (faking occurrence + # values through context is exactly what the spec forbids). + import inspect + + sig = inspect.signature(resolve_input_expression) + assert list(sig.parameters) == [ + "expression", + "state", + "workflow_input", + "context", + "label", + "location", + ], f"no resolver parameter exists: {list(sig.parameters)}" + assert "resolver" not in inspect.signature(resolve_step_input_bindings).parameters + print("OBSERVED resolver signatures are concrete graph mappings; no seam") + + +def test_node_input_binding_rejects_expression_kind_at_top_level(): + # StepInputBinding allows expressions, but plain InputBinding + # (deployment-level inputs) does not carry them — resolved data only. + from pydantic import TypeAdapter + + from wf_core.models.input_bindings import InputBinding + + with pytest.raises(ValidationError): + TypeAdapter(InputBinding).validate_python( + {"target": "x", "expression": {"kind": "literal", "value": 1}} + ) + print("OBSERVED top-level InputBinding carries no expressions") diff --git a/probes/deployment_scheduling_verify/test_schedule_state_model.py b/probes/deployment_scheduling_verify/test_schedule_state_model.py new file mode 100644 index 00000000..f69f68ff --- /dev/null +++ b/probes/deployment_scheduling_verify/test_schedule_state_model.py @@ -0,0 +1,1803 @@ +# DISPOSABLE SCHEDULING STATE-MODEL PROBE — NOT PRODUCTION CODE. +# See README.md in this directory. Pure stdlib; runs in the repo env: +# uv run pytest -q probes/deployment_scheduling_verify/test_schedule_state_model.py +"""Executable reference model of the deployment-scheduling state machine. + +Covers the spec's state gates with injected clocks and controlled +execution (no sleeps, no threads): +overlap=skip|parallel x misfire=skip|latest, latest-means-ONE-candidate, +supersession, terminal skips, slot accounting, pause-vs-downtime, +edit/delete/restart, capacity/fairness, fault boundaries, crash-during- +resume, exclusive ownership. + +The calendar is an abstract due-instant source here (next_after / +prev_before only — deliberately NO iter_between, to forbid unbounded +enumeration). Calendar math itself is probed in test_calendar_probe.py. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone + +import pytest + +UTC = timezone.utc + + +def ts(y, mo, d, h=0, mi=0, s=0) -> datetime: + return datetime(y, mo, d, h, mi, s, tzinfo=UTC) + + +# -------------------------------------------------------------------------- +# Errors +# -------------------------------------------------------------------------- + + +class InjectedFault(Exception): + pass + + +class SecondOwnerError(Exception): + pass + + +class StartupRejected(Exception): + pass + + +class BlockedSchedule(Exception): + pass + + +class ExecutorCrashed(Exception): + pass + + +# -------------------------------------------------------------------------- +# Records +# -------------------------------------------------------------------------- + +RUNNING = "running" +INTERRUPTED = "interrupted" # durably waiting, resumable +COMPLETED = "completed" +FAILED = "failed" +ACTIVE_STATES = (RUNNING, INTERRUPTED) + + +@dataclass +class Schedule: + id: str + rev: int = 1 + enabled: bool = True + paused: bool = False + deleted: bool = False + overlap: str = "skip" # skip | parallel + misfire: str = "skip" # skip | latest + max_active: int = 1 + allowance_s: float = 60.0 + deployment_id: str = "dep-1" + created_dep_rev: int = 1 + exhausted: bool = False + blocked_reason: str | None = None + + +@dataclass +class Candidate: + sched_id: str + intended: datetime + rev: int + + +@dataclass +class Run: + id: str + sched_id: str + intended: datetime + rev: int + frozen_input: dict + state: str = RUNNING + dispatched_unknown: bool = False # dispatched, no stopped result yet + needs_dispatch: bool = False # admitted + view exists, never dispatched + attempt_id: int = 0 # resume attempt the run currently belongs to + result_attempt: int | None = None # attempt that produced the stopped result + fail_reason: str = "" + + +@dataclass +class Record: + kind: str # admitted|completed|failed|skipped-overlap|skipped-misfire| + # superseded|preflight-rejected|exhausted|interval-summary|interrupted + sched_id: str + intended: datetime | None = None + run_id: str | None = None + reason: str = "" + interval: tuple[datetime, datetime] | None = None + count: int = 0 + + +# -------------------------------------------------------------------------- +# Occurrence sources: next_after / prev_before ONLY (no enumeration seam) +# -------------------------------------------------------------------------- + + +class CountingMixin: + def __init__(self) -> None: + self.next_calls = 0 + self.prev_calls = 0 + + +class PeriodicSource(CountingMixin): + def __init__(self, period: timedelta, start: datetime) -> None: + super().__init__() + self.period = period + self.start = start + + def next_after(self, instant: datetime) -> datetime | None: + self.next_calls += 1 + if instant < self.start: + return self.start + n = (instant - self.start) // self.period + 1 + return self.start + n * self.period + + def prev_before(self, instant: datetime) -> datetime | None: + self.prev_calls += 1 + if instant <= self.start: + return None + n = (instant - self.start - timedelta(microseconds=1)) // self.period + return self.start + n * self.period + + +class OneShotSource(CountingMixin): + def __init__(self, at: datetime) -> None: + super().__init__() + self.at = at + + def next_after(self, instant: datetime) -> datetime | None: + self.next_calls += 1 + return self.at if instant < self.at else None + + def prev_before(self, instant: datetime) -> datetime | None: + self.prev_calls += 1 + return self.at if instant > self.at else None + + +# -------------------------------------------------------------------------- +# Store with fault injection +# -------------------------------------------------------------------------- + + +@dataclass +class MemStore: + lockable: bool = True + lock_holder: str | None = None + id_seq: int = 0 # store-backed run counter: survives scheduler restart + attempt_seq: int = 0 # store-backed resume-attempt counter: same reason + schedules: dict[str, Schedule] = field(default_factory=dict) + candidates: dict[str, Candidate | None] = field(default_factory=dict) + consumed: dict[str, datetime] = field(default_factory=dict) + admissions: dict[str, dict] = field(default_factory=dict) # run_id -> record + runs: dict[str, Run] = field(default_factory=dict) + resume_attempts: dict[str, str] = field(default_factory=dict) # run -> ACTIVE|DONE + history: list[Record] = field(default_factory=list) + deployments: dict[str, dict] = field(default_factory=dict) # id -> {rev, required} + faults: dict[str, str] = field(default_factory=dict) # op -> before|after + op_log: list[str] = field(default_factory=list) + poll_cursor: int = 0 + + def check_fault(self, op: str) -> None: + mode = self.faults.pop(op, None) + self.op_log.append(op) + if mode == "before": + raise InjectedFault(f"{op}:before") + if mode == "after": + self.op_log.append(f"{op}:after-pending") + + def after_ok(self, op: str) -> None: + # A test driver calls this after the op's effect to confirm the + # injected 'after' fault (crash between effect and next step). + if f"{op}:after-pending" in self.op_log: + self.op_log.remove(f"{op}:after-pending") + raise InjectedFault(f"{op}:after") + + +EPOCH = ts(2020, 1, 1) + + +# -------------------------------------------------------------------------- +# Scheduler reference model +# -------------------------------------------------------------------------- + +SCAN_CAP = 100 # max next_after calls per schedule per poll before jumping + + +class Scheduler: + def __init__( + self, + store: MemStore, + owner: str, + capacity: int, + sources: dict[str, CountingMixin], + outcomes: dict[str, str] | None = None, + ) -> None: + if not store.lockable: + raise StartupRejected("unsupported locking rejects scheduler startup") + if store.lock_holder is not None and store.lock_holder != owner: + raise SecondOwnerError(f"store owned by {store.lock_holder}") + store.lock_holder = owner + self.store = store + self.owner = owner + self.capacity = capacity + self.sources = sources + self.outcomes = outcomes or {} + + def close(self) -> None: + if self.store.lock_holder == self.owner: + self.store.lock_holder = None + + def __enter__(self) -> Scheduler: + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + # -- helpers --------------------------------------------------------- + def _active(self, sched_id: str) -> list[Run]: + return [ + r + for r in self.store.runs.values() + if r.sched_id == sched_id and r.state in ACTIVE_STATES + ] + + def _task_load(self) -> int: + # Admitted-but-never-dispatched runs hold a schedule slot but no + # executing-task slot: nothing is running on their behalf. + return sum( + 1 + for r in self.store.runs.values() + if r.state == RUNNING and not r.needs_dispatch + ) + + def _record(self, **kw: object) -> None: + self.store.history.append(Record(**kw)) # type: ignore[arg-type] + + def _admit(self, sched: Schedule, intended: datetime, now: datetime) -> str | None: + """Ordered admission. Returns run_id, 'held', or None (terminal).""" + st = self.store + if sched.blocked_reason: + raise BlockedSchedule(sched.blocked_reason) + if not sched.enabled or sched.deleted or sched.paused: + return None + # 0. recheck + preflight against CURRENT deployment contract + dep = st.deployments.get(sched.deployment_id) + if dep is None: + self._record( + kind="preflight-rejected", + sched_id=sched.id, + intended=intended, + reason="deployment-deleted", + ) + st.consumed[sched.id] = intended + return None + frozen = { + "team": "eng", + "report_time": intended.isoformat(), + "sched": sched.id, + "dep_rev": dep["rev"], + } + missing = [k for k in dep["required"] if k not in frozen] + if missing: + self._record( + kind="preflight-rejected", + sched_id=sched.id, + intended=intended, + reason=f"missing-input:{missing}", + ) + if ( + st.candidates.get(sched.id) is not None + and st.candidates[sched.id].intended == intended + ): # type: ignore[union-attr] + st.candidates[sched.id] = None + st.consumed[sched.id] = intended + return None + # 1. overlap first (takes precedence over capacity) + active = self._active(sched.id) + if sched.overlap == "skip" and active: + self._record( + kind="skipped-overlap", + sched_id=sched.id, + intended=intended, + reason=f"active={[r.id for r in active]}", + ) + if ( + st.candidates.get(sched.id) is not None + and st.candidates[sched.id].intended == intended + ): # type: ignore[union-attr] + st.candidates[sched.id] = None + st.consumed[sched.id] = intended + return None + if sched.overlap == "parallel" and len(active) >= sched.max_active: + self._record( + kind="skipped-overlap", + sched_id=sched.id, + intended=intended, + reason=f"max_active={sched.max_active}", + ) + if ( + st.candidates.get(sched.id) is not None + and st.candidates[sched.id].intended == intended + ): # type: ignore[union-attr] + st.candidates[sched.id] = None + st.consumed[sched.id] = intended + return None + # 2. capacity: skip expires at deadline, latest holds one candidate + if self._task_load() >= self.capacity: + if sched.misfire == "latest": + st.candidates[sched.id] = Candidate(sched.id, intended, sched.rev) + st.consumed[sched.id] = intended + return "held" + if (now - intended).total_seconds() > sched.allowance_s: + self._record( + kind="skipped-misfire", + sched_id=sched.id, + intended=intended, + reason="capacity-deadline", + ) + st.consumed[sched.id] = intended + return None + return "held-undecided" # retry next poll, consumed NOT advanced + # 3. allocate + freeze (store-backed counter: no reuse on restart) + st.id_seq += 1 + run_id = f"run-{sched.id}-{st.id_seq}" + # 4. persist admission record BEFORE dispatch (fault boundary). + # The occurrence is decided here: history entry, candidate + # clearing, consumed progress, and one-shot exhaustion all belong + # to this persist, so a later crash can never re-decide it. + st.check_fault("admission") + st.admissions[run_id] = { + "sched": sched.id, + "intended": intended, + "rev": sched.rev, + "input": dict(frozen), + } + # The occurrence-status entry is part of the admission persist itself. + self._record( + kind="admitted", + sched_id=sched.id, + intended=intended, + run_id=run_id, + reason=f"rev={sched.rev}", + ) + # One-shot admission exhausts the schedule (admit exactly once). + if isinstance(self.sources.get(sched.id), OneShotSource): + sched.exhausted = True + if ( + st.candidates.get(sched.id) is not None + and st.candidates[sched.id].intended == intended + ): # type: ignore[union-attr] + st.candidates[sched.id] = None + st.consumed[sched.id] = intended + st.after_ok("admission") + # 5. materialize run view with same identity (fault boundary) + st.check_fault("materialize") + st.runs[run_id] = Run(run_id, sched.id, intended, sched.rev, dict(frozen)) + st.after_ok("materialize") + # 6. dispatch the captured invocation without re-resolving (faults) + st.check_fault("dispatch") + try: + self._dispatch(run_id, now) + finally: + st.after_ok("dispatch") + return run_id + + def _dispatch(self, run_id: str, now: datetime) -> None: + st = self.store + run = st.runs[run_id] + outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete")) + if outcome == "hang": + run.dispatched_unknown = True # task alive in-process; restart abandons it + return # stays RUNNING, occupies schedule + task slots + if outcome == "crash": + run.dispatched_unknown = True + raise ExecutorCrashed(run_id) + if outcome == "interrupt": + st.check_fault("interrupt-persist") + run.state = INTERRUPTED + st.after_ok("interrupt-persist") + self._record( + kind="interrupted", + sched_id=run.sched_id, + intended=run.intended, + run_id=run_id, + ) + return + assert outcome == "complete" + st.check_fault("complete") + run.state = COMPLETED + st.after_ok("complete") + self._record( + kind="completed", + sched_id=run.sched_id, + intended=run.intended, + run_id=run_id, + ) + + # -- polling ---------------------------------------------------------- + def poll(self, now: datetime) -> dict[str, str]: + st = self.store + if st.lock_holder != self.owner: + raise SecondOwnerError("lost ownership") + self._dispatch_pending(now) + ids = sorted(s.id for s in st.schedules.values()) + if not ids: + return {} + start = st.poll_cursor % len(ids) + order = ids[start:] + ids[:start] + st.poll_cursor += 1 + results: dict[str, str] = {} + for sid in order: + results[sid] = self._poll_one(st.schedules[sid], now) + return results + + def _dispatch_pending(self, now: datetime) -> None: + """Dispatch runs that recovery materialized but never executed. + + Recovery NEVER executes: it only completes missing views and flags + them pending. Execution happens here, through the same capacity + checks as normal admission — never inside recovery. Pending runs + of blocked schedules stay pending (fail closed). + """ + st = self.store + for run in sorted(st.runs.values(), key=lambda r: r.id): + if not run.needs_dispatch: + continue + sched = st.schedules.get(run.sched_id) + if sched is None or sched.blocked_reason: + continue + if self._task_load() >= self.capacity: + continue # stays pending until a slot frees + run.needs_dispatch = False + try: + self._dispatch(run.id, now) + except ExecutorCrashed: + # Dispatched with unknown outcome: next recovery fails it + # closed as abandoned (no replay), like any dispatch crash. + run.dispatched_unknown = True + + def _poll_one(self, sched: Schedule, now: datetime) -> str: + st = self.store + if sched.blocked_reason: + raise BlockedSchedule(sched.blocked_reason) + if sched.deleted: + # Deletion clears pending candidates; admitted runs/history stay. + if st.candidates.get(sched.id) is not None: + st.candidates[sched.id] = None + return "deleted" + if not sched.enabled: + # Resolved edge (no spec disable concept): administrative disable + # behaves like pause for catch-up — excluded, never backfilled. + if st.candidates.get(sched.id) is not None: + st.candidates[sched.id] = None + st.consumed[sched.id] = max(st.consumed.get(sched.id, EPOCH), now) + return "disabled" + if sched.exhausted: + return "exhausted" + if sched.paused: + # Explicit pause: clear unadmitted candidates, exclude interval. + if st.candidates.get(sched.id) is not None: + st.candidates[sched.id] = None + st.consumed[sched.id] = max(st.consumed.get(sched.id, EPOCH), now) + return "paused" + src = self.sources[sched.id] + consumed = st.consumed.get(sched.id, EPOCH) + if consumed > now: + return "clock-rollback-held" # never re-admit consumed instants + # Bounded scan: at most SCAN_CAP next_after calls, then jump. + due: list[datetime] = [] + cursor = consumed + jumped = False + while True: + nxt = src.next_after(cursor) # type: ignore[attr-defined] + if nxt is None or nxt > now: + if ( + not due + and isinstance(src, OneShotSource) + and nxt is None + and not sched.exhausted + and sched.misfire == "skip" + and self._is_oneshot_expired(src, now) + ): + self._record( + kind="exhausted", + sched_id=sched.id, + intended=src.at, + reason="oneshot-expired-skip", + ) + sched.exhausted = True + st.consumed[sched.id] = now + st.candidates[sched.id] = None + return "exhausted" + break + due.append(nxt) + cursor = nxt + if len(due) >= SCAN_CAP: + jumped = True + break + if jumped: + # Never enumerate further: one bounded query + interval summary. + if sched.misfire == "latest": + latest = src.prev_before(now) # type: ignore[attr-defined] + assert latest is not None + old = st.candidates.get(sched.id) + if old is not None and old.intended != latest: + self._record( + kind="superseded", + sched_id=sched.id, + intended=old.intended, + reason=f"coalesced-into:{latest.isoformat()}", + ) + st.candidates[sched.id] = Candidate(sched.id, latest, sched.rev) + self._record( + kind="interval-summary", + sched_id=sched.id, + interval=(consumed, now), + count=-1, + reason="coalesced-missed-span", + ) + st.consumed[sched.id] = now + return self._admit_held_candidate(sched, now) or "candidate-held" + self._record( + kind="interval-summary", + sched_id=sched.id, + interval=(consumed, now), + count=-1, + reason="skipped-missed-span", + ) + st.consumed[sched.id] = now + return "span-skipped" + # Normal path: decide each due instant in order. + last_result = "idle" + for instant in due: + if instant <= st.consumed.get(sched.id, EPOCH): + continue + age = (now - instant).total_seconds() + if age <= sched.allowance_s: + # A timely admission consumes any older held candidate: the + # newer due instant supersedes it (recorded, never admitted). + old = st.candidates.get(sched.id) + if old is not None and old.intended != instant: + self._record( + kind="superseded", + sched_id=sched.id, + intended=old.intended, + reason=f"admitted-newer:{instant.isoformat()}", + ) + st.candidates[sched.id] = None + r = self._admit(sched, instant, now) + last_result = f"admit:{r}" + elif sched.misfire == "latest": + old = st.candidates.get(sched.id) + if old is not None and old.intended != instant: + self._record( + kind="superseded", + sched_id=sched.id, + intended=old.intended, + reason=f"coalesced-into:{instant.isoformat()}", + ) + # A newer due time supersedes the unadmitted candidate; an + # admitted run is never touched (candidates only). + st.candidates[sched.id] = Candidate(sched.id, instant, sched.rev) + st.consumed[sched.id] = instant + last_result = self._admit_held_candidate(sched, now) or "candidate-held" + else: + if isinstance(src, OneShotSource) and not sched.exhausted: + self._record( + kind="exhausted", + sched_id=sched.id, + intended=instant, + reason="oneshot-expired-skip", + ) + sched.exhausted = True + st.consumed[sched.id] = instant + last_result = "exhausted" + continue + self._record( + kind="skipped-misfire", + sched_id=sched.id, + intended=instant, + reason=f"age={age:.0f}s", + ) + st.consumed[sched.id] = instant + last_result = "skipped-misfire" + # Also try a held candidate whose slot may have freed. + if last_result in ("idle",) and st.candidates.get(sched.id) is not None: + last_result = self._admit_held_candidate(sched, now) or "candidate-held" + return last_result + + def _is_oneshot_expired(self, src: OneShotSource, now: datetime) -> bool: + return src.at <= now + + def _admit_held_candidate(self, sched: Schedule, now: datetime) -> str | None: + st = self.store + cand = st.candidates.get(sched.id) + if cand is None or cand.rev != sched.rev: + if cand is not None and cand.rev != sched.rev: + self._record( + kind="superseded", + sched_id=sched.id, + intended=cand.intended, + reason="schedule-edit", + ) + st.candidates[sched.id] = None + return None + r = self._admit(sched, cand.intended, now) + return r if r != "held-undecided" else None + + # -- administration ---------------------------------------------------- + def resume_schedule(self, sid: str, now: datetime) -> None: + """Unpause: resume selects the next future occurrence; the paused + interval is excluded from catch-up under both policies.""" + sched = self.store.schedules[sid] + sched.paused = False + self.store.consumed[sid] = max(self.store.consumed.get(sid, EPOCH), now) + self.store.candidates[sid] = None + + def edit_schedule(self, sid: str, now: datetime) -> None: + """Definition edit: new revision, discard old candidates, begin at + the edit time (creation/edits never backfill).""" + st = self.store + sched = st.schedules[sid] + sched.rev += 1 + old = st.candidates.get(sid) + if old is not None: + self._record( + kind="superseded", + sched_id=sid, + intended=old.intended, + reason="schedule-edit", + ) + st.candidates[sid] = None + st.consumed[sid] = max(st.consumed.get(sid, EPOCH), now) + + # -- resume ------------------------------------------------------------ + def resume_run(self, run_id: str, now: datetime) -> str: + st = self.store + run = st.runs[run_id] + assert run.state == INTERRUPTED, "only waiting interruptions resume" + if st.resume_attempts.get(run_id) == "ACTIVE": + raise BlockedSchedule("ambiguous attempt already active") + if self._task_load() >= self.capacity: + return "blocked-capacity" # stays waiting; slot needed + # Durably mark the active attempt BEFORE executing again. The mark + # carries a fresh attempt identity that later stopped results echo + # back, so recovery can tell a new result from the old checkpoint. + st.check_fault("resume-mark") + st.attempt_seq += 1 + run.attempt_id = st.attempt_seq + st.resume_attempts[run_id] = "ACTIVE" + st.after_ok("resume-mark") + run.state = RUNNING + run.dispatched_unknown = True + st.check_fault("resume-execute") + try: + outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete")) + if outcome == "crash": + raise ExecutorCrashed(run_id) + if outcome == "interrupt": + return self._finish_resume_interrupted(run) + return self._finish_resume_completed(run) + finally: + st.after_ok("resume-execute") + + def _finish_resume_completed(self, run: Run) -> str: + # Completion, attempt-clearing, and history recording are separate + # persists with a fault boundary between each pair; a resumed run + # may also interrupt again instead of completing. + st = self.store + st.check_fault("resume-complete") + run.state = COMPLETED + run.dispatched_unknown = False + run.result_attempt = run.attempt_id + st.after_ok("resume-complete") + st.check_fault("resume-attempt-clear") + st.resume_attempts[run.id] = "DONE" + st.after_ok("resume-attempt-clear") + self._record( + kind="completed", + sched_id=run.sched_id, + intended=run.intended, + run_id=run.id, + reason="resumed-complete", + ) + return "resumed-complete" + + def _finish_resume_interrupted(self, run: Run) -> str: + st = self.store + st.check_fault("resume-interrupt") + run.state = INTERRUPTED + run.dispatched_unknown = False + run.result_attempt = run.attempt_id + st.after_ok("resume-interrupt") + st.check_fault("resume-attempt-clear") + st.resume_attempts[run.id] = "DONE" + st.after_ok("resume-attempt-clear") + self._record( + kind="interrupted", + sched_id=run.sched_id, + intended=run.intended, + run_id=run.id, + reason="resumed-reinterrupted", + ) + return "resumed-interrupted" + + # -- recovery ------------------------------------------------------------ + def recover(self, now: datetime) -> list[str]: + """Startup recovery under exclusive ownership. Returns diagnostics.""" + st = self.store + if st.lock_holder != self.owner: + raise SecondOwnerError("lost ownership") + diags: list[str] = [] + # Admission record is the recovery authority. Recovery NEVER + # executes: it only materializes missing views (flagged pending for + # the capacity-checked dispatch sweep in poll()), fails abandoned / + # ambiguous runs closed, reconciles terminal records, and blocks + # corrupt schedules. Any RUNNING run lost its in-memory task with + # the old process: with an admission record it is abandoned + # (failed, no replay); without one it is corrupt (fail closed). + # Interrupted runs carry the identity of the attempt that produced + # them: a result matching the ACTIVE attempt is fresh (resumable); + # anything else under an ACTIVE attempt is stale (ambiguous, failed). + for run_id, rec in st.admissions.items(): + if run_id not in st.runs: + # Admitted but never materialized/dispatched: complete the + # missing view and flag it pending. The next poll dispatches + # the captured invocation through capacity checks — exactly + # once, because the occurrence is already consumed. + st.runs[run_id] = Run( + run_id, + rec["sched"], + rec["intended"], + rec["rev"], + dict(rec["input"]), + ) + st.runs[run_id].needs_dispatch = True + diags.append(f"{run_id}:view-completed-pending-dispatch") + for run in st.runs.values(): + if run.state == INTERRUPTED: + if st.resume_attempts.get(run.id) == "ACTIVE": + if ( + run.result_attempt is not None + and run.result_attempt == run.attempt_id + ): + # The stopped result belongs to the active attempt: + # the re-interruption landed before the crash. Fresh, + # resumable; the attempt is done executing. + st.resume_attempts[run.id] = "DONE" + diags.append(f"{run.id}:fresh-result-resumable") + else: + run.state = FAILED + run.fail_reason = ( + "ambiguous resume attempt: may have executed; " + "external effects may already have occurred; no retry" + ) + self._record( + kind="failed", + sched_id=run.sched_id, + intended=run.intended, + run_id=run.id, + reason=run.fail_reason, + ) + diags.append(f"{run.id}:failed-closed") + else: + diags.append(f"{run.id}:waiting-resumable") + elif run.state == RUNNING: + if run.needs_dispatch: + # Admitted and materialized but provably never + # dispatched: stays pending for the poll sweep, never + # failed as abandoned. + diags.append(f"{run.id}:pending-dispatch-kept") + continue + if run.id not in st.admissions: + sched = st.schedules[run.sched_id] + sched.blocked_reason = ( + f"corrupt run view without admission: {run.id}" + ) + diags.append(f"{run.id}:corrupt-blocked") + continue + if st.resume_attempts.get(run.id) == "ACTIVE": + run.state = FAILED + run.fail_reason = ( + "ambiguous resume attempt: may have executed; " + "external effects may already have occurred; no retry" + ) + else: + run.state = FAILED + run.fail_reason = ( + "abandoned execution: outcome unknown; external " + "effects may already have occurred; no replay" + ) + self._record( + kind="failed", + sched_id=run.sched_id, + intended=run.intended, + run_id=run.id, + reason=run.fail_reason, + ) + diags.append(f"{run.id}:failed-closed") + # Reconcile terminal runs whose history entry was lost to a crash + # between the state persist and the record append (idempotent: only + # appends when no terminal record exists for the run). A COMPLETED + # run with a still-ACTIVE attempt crashed between completion and + # attempt-clearing: mark DONE, reconcile the record, never re-run. + for run in st.runs.values(): + if run.state == COMPLETED and st.resume_attempts.get(run.id) == "ACTIVE": + st.resume_attempts[run.id] = "DONE" + diags.append(f"{run.id}:attempt-reconciled") + if run.state == COMPLETED and not any( + r.kind == "completed" and r.run_id == run.id for r in st.history + ): + self._record( + kind="completed", + sched_id=run.sched_id, + intended=run.intended, + run_id=run.id, + reason="reconciled-on-recovery", + ) + diags.append(f"{run.id}:terminal-reconciled") + elif ( + run.state == INTERRUPTED + and st.resume_attempts.get(run.id) != "ACTIVE" + and not any( + r.kind == "interrupted" and r.run_id == run.id for r in st.history + ) + ): + self._record( + kind="interrupted", + sched_id=run.sched_id, + intended=run.intended, + run_id=run.id, + reason="reconciled-on-recovery", + ) + diags.append(f"{run.id}:terminal-reconciled") + return diags + + +def make_store() -> MemStore: + st = MemStore() + st.deployments["dep-1"] = {"rev": 1, "required": ["team", "report_time"]} + return st + + +def add_sched( + st: MemStore, + sid: str, + src: CountingMixin, + sources: dict, + start: datetime, + **kw: object, +) -> Schedule: + if sid in st.schedules: + raise ValueError(f"schedule id reused: {sid}") + sched = Schedule(id=sid, **kw) # type: ignore[arg-type] + st.schedules[sid] = sched + st.candidates[sid] = None + st.consumed[sid] = start + sources[sid] = src + return sched + + +# -------------------------------------------------------------------------- +# Tests +# -------------------------------------------------------------------------- + + +def test_defaults_skip_skip(): + s = Schedule(id="s") + assert s.overlap == "skip" and s.misfire == "skip" + + +def test_overlap_skip_x_misfire_skip_running_blocks_and_late_drops(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(minutes=10), t0), + sources, + t0 - timedelta(minutes=10), + ) + with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch: + sch.poll(t0) # admits 12:00, hangs (RUNNING) + (run_id,) = list(st.runs) + sch.poll(t0 + timedelta(minutes=10)) # 12:10 due while active + kinds = [(r.kind, r.intended) for r in st.history if r.sched_id == "a"] + assert ("skipped-overlap", ts(2026, 9, 8, 12, 10)) in kinds + # Late instant beyond allowance with skip: skipped-misfire. + sch.poll(t0 + timedelta(minutes=30)) # 12:20,12:30 missed (>60s) + kinds = [(r.kind, r.intended) for r in st.history if r.sched_id == "a"] + assert ("skipped-misfire", ts(2026, 9, 8, 12, 20)) in kinds + assert st.runs[run_id].state == RUNNING + + +def test_overlap_skip_x_misfire_latest_single_catchup_and_supersession(): + st = make_store() + sources: dict = {} + add_sched( + st, + "h", + PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)), + sources, + ts(2026, 9, 8, 9, 0), + misfire="latest", + ) + with Scheduler(st, "p1", 0, sources) as sch: # no capacity: hold candidates + sch.poll(ts(2026, 9, 8, 12, 20)) # missed 10:00,11:00,12:00 + cand = st.candidates["h"] + assert cand is not None and cand.intended == ts(2026, 9, 8, 12, 0) + runs_for_h = [r for r in st.history if r.kind == "admitted"] + assert runs_for_h == [], "ONE pending candidate, not replay-all" + sup = [r for r in st.history if r.kind == "superseded"] + assert {r.intended for r in sup} == { + ts(2026, 9, 8, 10, 0), + ts(2026, 9, 8, 11, 0), + } + # Capacity returns at 13:00 while 13:00 is also due: exactly one admission, + # for the 13:00 instant (newer supersedes the unadmitted 12:00). + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch2: + sch2.poll(ts(2026, 9, 8, 13, 0)) + admitted = [r for r in st.history if r.kind == "admitted"] + assert len(admitted) == 1 + assert admitted[0].intended == ts(2026, 9, 8, 13, 0) + assert st.candidates["h"] is None, "no stale candidate survives admission" + assert any( + r.kind == "superseded" and r.intended == ts(2026, 9, 8, 12, 0) + for r in st.history + ) + assert admitted[0].run_id is not None + assert ( + st.runs[admitted[0].run_id].frozen_input["report_time"] + == "2026-09-08T13:00:00+00:00" + ) + + +def test_newer_due_never_supersedes_admitted_run(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(minutes=10), t0), + sources, + t0 - timedelta(minutes=10), + misfire="latest", + ) + with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch: + first = sch.poll(t0)["a"] + run_id = first.split(":", 1)[1] + assert st.runs[run_id].intended == t0 + sch.poll(t0 + timedelta(minutes=10)) + # 12:10 is skipped-overlap (terminal); the admitted 12:00 run is + # untouched and no candidate resurrects 12:10. + assert st.runs[run_id].state == RUNNING + assert st.candidates["a"] is None + assert any( + r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=10) + for r in st.history + ) + + +def test_terminal_overlap_skip_never_reappears_through_catchup(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(minutes=10), t0), + sources, + t0 - timedelta(minutes=10), + misfire="latest", + ) + with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch: + sch.poll(t0) + sch.poll(t0 + timedelta(minutes=10)) # terminal skipped-overlap @12:10 + # Restart abandons the hanging in-flight task (failed, no replay), but + # the terminal 12:10 skip is never reconstructed as a candidate. + with Scheduler(st, "p1", 0, sources) as sch2: + diags = sch2.recover(t0 + timedelta(minutes=11)) + assert any("failed-closed" in d for d in diags) + sch2.poll(t0 + timedelta(minutes=25)) # 12:20 missed -> held candidate + cand = st.candidates["a"] + assert cand is not None and cand.intended == t0 + timedelta(minutes=20) + intents = [ + r.intended + for r in st.history + if r.kind == "admitted" and r.intended == t0 + timedelta(minutes=10) + ] + assert intents == [], "terminally skipped 12:10 must never be admitted" + + +def test_parallel_limits_count_interrupted_and_waiting_frees_task_slot(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "p", + PeriodicSource(timedelta(minutes=5), t0), + sources, + t0 - timedelta(minutes=5), + overlap="parallel", + max_active=2, + ) + with Scheduler(st, "owner", 1, sources, {"*": "hang"}) as sch: + sch.poll(t0) # run1 RUNNING (task slot taken) + assert sch._task_load() == 1 + st.runs["run-p-1"].state = INTERRUPTED # scripted durable interrupt + st.history.append( + Record(kind="interrupted", sched_id="p", intended=t0, run_id="run-p-1") + ) + assert sch._task_load() == 0, "waiting interruptions hold no task slot" + sch.poll(t0 + timedelta(minutes=5)) # run2 admitted (1 task slot free) + assert st.runs["run-p-2"].state == RUNNING + # max_active=2 reached (1 waiting + 1 running): 12:10 skipped-overlap. + sch.poll(t0 + timedelta(minutes=10)) + assert any( + r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=10) + for r in st.history + ) + # Lowering the limit never cancels; blocks new admission instead. + st.schedules["p"].max_active = 1 + st.runs["run-p-2"].state = INTERRUPTED + sch.poll(t0 + timedelta(minutes=15)) + assert any( + r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=15) + for r in st.history + ) + assert st.runs["run-p-1"].state == INTERRUPTED # not cancelled + # Resume needs a task slot: none free while... free one by completing. + st.runs["run-p-2"].state = COMPLETED + assert ( + sch.resume_run("run-p-1", t0 + timedelta(minutes=16)) == "resumed-complete" + ) + + +def test_explicit_pause_is_not_downtime(): + st = make_store() + sources: dict = {} + add_sched( + st, + "a", + PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)), + sources, + ts(2026, 9, 8, 9, 0), + misfire="latest", + ) + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + sch.poll(ts(2026, 9, 8, 9, 0)) + st.schedules["a"].paused = True + sch.poll(ts(2026, 9, 8, 10, 30)) # paused polls exclude the interval + assert st.candidates["a"] is None + sch.poll(ts(2026, 9, 8, 11, 30)) + sch.resume_schedule("a", ts(2026, 9, 8, 12, 30)) # next future only + sch.poll(ts(2026, 9, 8, 12, 30)) + assert st.candidates["a"] is None, "paused times are not replayed" + admitted_intents = [r.intended for r in st.history if r.kind == "admitted"] + assert ts(2026, 9, 8, 10, 0) not in admitted_intents + assert ts(2026, 9, 8, 11, 0) not in admitted_intents + assert ts(2026, 9, 8, 12, 0) not in admitted_intents + # Contrast: enabled downtime DOES catch up under latest. + st2 = make_store() + sources2: dict = {} + add_sched( + st2, + "b", + PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)), + sources2, + ts(2026, 9, 8, 9, 0), + misfire="latest", + ) + sch2 = Scheduler(st2, "p1", 0, sources2) + sch2.poll(ts(2026, 9, 8, 12, 20)) # 3 missed while "down", enabled + assert st2.candidates["b"] is not None + assert st2.candidates["b"].intended == ts(2026, 9, 8, 12, 0) # type: ignore[union-attr] + sch2.close() + + +def test_edit_discards_candidates_no_backfill_delete_keeps_history(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)), + sources, + ts(2026, 9, 8, 9, 0), + misfire="latest", + ) + with Scheduler(st, "p1", 0, sources) as sch: + sch.poll(ts(2026, 9, 8, 12, 20)) + assert st.candidates["a"] is not None + sch.edit_schedule("a", ts(2026, 9, 8, 12, 21)) # definition edit + sch.poll(ts(2026, 9, 8, 12, 21)) + assert st.candidates["a"] is None, "edits discard old-revision candidates" + assert any( + r.kind == "superseded" and r.reason == "schedule-edit" for r in st.history + ) + # No backfill before the edit: consumed advanced to edit time. + assert st.consumed["a"] >= ts(2026, 9, 8, 12, 21) + with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch: + sch.poll(ts(2026, 9, 8, 13, 0)) + (run_id,) = [r for r in st.runs if st.runs[r].state == RUNNING] + st.schedules["a"].deleted = True # delete clears pending, keeps rest + sch.poll(ts(2026, 9, 8, 14, 0)) + assert st.candidates["a"] is None + assert st.runs[run_id].state == RUNNING, "delete must not cancel active runs" + n_history = len(st.history) + assert n_history > 0, "history remains intact" + # In-flight work can still finish after delete; history is appended. + st.runs[run_id].state = COMPLETED + st.history.append( + Record( + kind="completed", + sched_id="a", + intended=st.runs[run_id].intended, + run_id=run_id, + ) + ) + assert len(st.history) == n_history + 1 + with pytest.raises(ValueError, match="schedule id reused"): + add_sched(st, "a", PeriodicSource(timedelta(hours=1), t0), sources, t0) + + +def test_restart_survives_candidate_and_rollback_never_readmits(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(minutes=10), t0), + sources, + t0 - timedelta(minutes=10), + misfire="latest", + ) + with Scheduler(st, "p1", 0, sources) as sch: + sch.poll(t0 + timedelta(minutes=25)) # candidate 12:20 + assert st.candidates["a"] is not None + with Scheduler(st, "p1", 0, sources) as sch2: # restart, still no capacity + sch2.recover(t0 + timedelta(minutes=26)) + sch2.poll(t0 + timedelta(minutes=26)) + assert st.candidates["a"] is not None + assert st.candidates["a"].intended == t0 + timedelta(minutes=20) # type: ignore[union-attr] + # Clock rollback: consumed instants are never re-admitted. + sch2.poll(t0 - timedelta(hours=1)) + admitted = [r for r in st.history if r.kind == "admitted"] + assert admitted == [] + + +def test_long_downtime_bounded_and_summarized(): + st = make_store() + sources: dict = {} + src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) + add_sched(st, "m", src, sources, ts(2023, 9, 8, 12, 0), misfire="latest") + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + sch.poll(ts(2026, 9, 8, 12, 0, 0)) # 3 years of per-minute misses + total_calls = src.next_calls + src.prev_calls + assert total_calls <= SCAN_CAP + 2, f"must not enumerate: {total_calls} calls" + # Capacity available: latest means ONE prompt admission for the latest + # missed instant (11:59), using the candidate's intended time. + admitted = [r for r in st.history if r.kind == "admitted"] + assert len(admitted) == 1 + assert admitted[0].intended == ts(2026, 9, 8, 11, 59) + assert st.candidates["m"] is None # consumed by the prompt admission + summaries = [r for r in st.history if r.kind == "interval-summary"] + assert len(summaries) == 1, "one coalesced summary, not per-minute rows" + # Same gap with no capacity: exactly ONE held candidate, zero admissions. + st1b = make_store() + sources1b: dict = {} + src1b = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) + add_sched(st1b, "m", src1b, sources1b, ts(2023, 9, 8, 12, 0), misfire="latest") + with Scheduler(st1b, "p1", 0, sources1b, {"*": "complete"}) as sch: + sch.poll(ts(2026, 9, 8, 12, 0, 0)) + assert src1b.next_calls + src1b.prev_calls <= SCAN_CAP + 2 + assert st1b.candidates["m"] is not None + assert st1b.candidates["m"].intended == ts(2026, 9, 8, 11, 59) # type: ignore[union-attr] + assert [r for r in st1b.history if r.kind == "admitted"] == [] + # skip policy: same boundedness, zero admissions, next future selected. + st2 = make_store() + sources2: dict = {} + src2 = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) + add_sched(st2, "m", src2, sources2, ts(2023, 9, 8, 12, 0), misfire="skip") + with Scheduler(st2, "p1", 4, sources2, {"*": "complete"}) as sch: + sch.poll(ts(2026, 9, 8, 12, 0, 30)) + assert src2.next_calls + src2.prev_calls <= SCAN_CAP + 2 + assert [r for r in st2.history if r.kind == "admitted"] == [] + assert st2.candidates["m"] is None + + +def test_fairness_frequent_schedule_cannot_monopolize(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "fast", + PeriodicSource(timedelta(minutes=1), t0), + sources, + t0 - timedelta(minutes=1), + ) + add_sched(st, "slow", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + with Scheduler(st, "p1", 1, sources, {"*": "hang"}) as sch: + sch.poll(t0) # rotation starts at fast (sorted first): fast admitted + assert any(r.sched_id == "fast" and r.kind == "admitted" for r in st.history) + # Complete fast's run externally, next poll must serve slow first. + for r in st.runs.values(): + r.state = COMPLETED + res = sch.poll(t0 + timedelta(seconds=30)) + slow_admitted = [ + r for r in st.history if r.sched_id == "slow" and r.kind == "admitted" + ] + assert slow_admitted, f"slow schedule starved: {res}" + + +def test_fault_before_admission_never_dispatches(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + st.faults["admission"] = "before" + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + try: + sch.poll(t0) + assert False, "fault must propagate" + except InjectedFault: + pass + assert st.runs == {}, "no dispatch before durable admission" + assert st.admissions == {} + sch.poll(t0) # retry after the fault is clean + assert len(st.runs) == 1 + + +def test_fault_between_admission_and_view_recovers_without_redispatch(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + st.faults["materialize"] = "before" + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + try: + sch.poll(t0) + assert False + except InjectedFault: + pass + assert len(st.admissions) == 1 and len(st.runs) == 0 + # Recovery NEVER executes: it completes the view and flags it + # pending. No outcome exists yet. + diags = sch.recover(t0) + assert any("pending-dispatch" in d for d in diags) + assert len(st.runs) == 1 + run = st.runs["run-a-1"] + assert run.state == RUNNING and run.needs_dispatch + assert not run.dispatched_unknown + assert [r for r in st.history if r.kind == "completed"] == [] + # The next poll dispatches the captured invocation through the + # capacity checks — exactly once, no second admission. + sch.poll(t0) + assert run.state == COMPLETED and not run.needs_dispatch + n_admitted = len([r for r in st.history if r.kind == "admitted"]) + assert n_admitted == 1, "reconciliation must be idempotent" + assert len([r for r in st.history if r.kind == "completed"]) == 1 + + +def test_recovery_pending_dispatch_waits_for_capacity(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + st.faults["materialize"] = "before" + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + try: + sch.poll(t0) + assert False + except InjectedFault: + pass + sch.recover(t0) + sch.capacity = 0 # slots full: pending dispatch must wait + sch.poll(t0 + timedelta(seconds=1)) + assert st.runs["run-a-1"].needs_dispatch + assert [r for r in st.history if r.kind == "completed"] == [] + sch.capacity = 4 + sch.poll(t0 + timedelta(seconds=2)) + assert not st.runs["run-a-1"].needs_dispatch + assert st.runs["run-a-1"].state == COMPLETED + + +def test_run_ids_survive_restart_without_reuse(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(minutes=10), t0), + sources, + t0 - timedelta(minutes=10), + ) + with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch: + sch.poll(t0) + assert st.runs["run-a-1"].intended == t0 + # Restart: the counter lives in the store, so the next admission gets + # a fresh identity instead of overwriting run-a-1. + with Scheduler(st, "p1", 4, sources, {"*": "hang"}) as sch2: + sch2.recover(t0 + timedelta(seconds=1)) # hanging run abandoned + assert st.runs["run-a-1"].state == FAILED + sch2.poll(t0 + timedelta(minutes=10)) + assert st.runs["run-a-1"].intended == t0, "old run untouched" + assert st.runs["run-a-2"].intended == t0 + timedelta(minutes=10) + assert len(st.runs) == 2 + + +def test_crash_after_dispatch_marks_abandoned_failed_without_replay(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(hours=1), t0), + sources, + t0 - timedelta(hours=1), + ) + with Scheduler(st, "p1", 4, sources, {"run-a-1": "crash", "*": "complete"}) as sch: + try: + sch.poll(t0) + assert False + except ExecutorCrashed: + pass + run = st.runs["run-a-1"] + assert run.state == RUNNING and run.dispatched_unknown + diags = sch.recover(t0 + timedelta(seconds=5)) + assert run.state == FAILED + assert "may already have occurred" in run.fail_reason + assert any("failed-closed" in d for d in diags) + # Future occurrences proceed; the failed one is never replayed. + sch.poll(t0 + timedelta(hours=1)) + intents = [r.intended for r in st.history if r.kind == "admitted"] + assert t0 not in intents[1:] or intents.count(t0) == 1 + + +def test_crash_during_resume_must_not_look_safe_to_retry(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch: + sch.poll(t0) + (run_id,) = list(st.runs) + assert st.runs[run_id].state == INTERRUPTED + # Restart: a merely-waiting interruption is safe to resume later... + with Scheduler(st, "p1", 4, sources, {"*": "crash"}) as sch2: + sch2.recover(t0) + assert st.runs[run_id].state == INTERRUPTED + assert run_id not in st.resume_attempts + try: + sch2.resume_run(run_id, t0) + assert False + except ExecutorCrashed: + pass + # ...but the crash left a durably marked ACTIVE attempt: recovery + # must fail it closed instead of presenting the old checkpoint again. + assert st.resume_attempts[run_id] == "ACTIVE" + diags = sch2.recover(t0) + assert st.runs[run_id].state == FAILED + assert "ambiguous" in st.runs[run_id].fail_reason + assert any("failed-closed" in d for d in diags) + + +def test_preflight_rejection_invents_no_run_and_freezes_input(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(hours=1), t0), + sources, + t0 - timedelta(hours=1), + misfire="latest", + ) + with Scheduler(st, "p1", 0, sources) as sch: + sch.poll(t0 + timedelta(minutes=5)) # held candidate @12:00 + assert st.candidates["a"] is not None + # Deployment edit changes the expected input before admission. + st.deployments["dep-1"] = { + "rev": 2, + "required": ["team", "report_time", "region"], + } + sch2 = Scheduler(st, "p1", 4, sources, {"*": "complete"}) + sch2.poll(t0 + timedelta(minutes=6)) + assert st.runs == {}, "preflight rejection must invent no run" + assert any(r.kind == "preflight-rejected" for r in st.history) + sch2.close() + # Admitted runs freeze their invocation: later edits change nothing. + st3 = make_store() + sources3: dict = {} + add_sched(st3, "a", OneShotSource(t0), sources3, t0 - timedelta(hours=1)) + with Scheduler(st3, "p1", 4, sources3, {"*": "hang"}) as sch: + sch.poll(t0) + (run_id,) = list(st3.runs) + before = dict(st3.runs[run_id].frozen_input) + st3.deployments["dep-1"] = {"rev": 9, "required": ["team"]} + st3.runs[run_id].state = COMPLETED + assert st3.runs[run_id].frozen_input == before + + +def test_exclusive_ownership_and_unsupported_locking(): + st = make_store() + sources: dict = {} + sch1 = Scheduler(st, "proc-A", 1, sources) + try: + Scheduler(st, "proc-B", 1, sources) + assert False, "second owner must be rejected" + except SecondOwnerError: + pass + # A held lock never expires while the owner lives (no lease timeout). + assert not hasattr(st, "lease_expiry") + sch1.close() # process death releases + sch2 = Scheduler(st, "proc-B", 1, sources) # new owner starts, recovers + sch2.close() + bad = MemStore(lockable=False) + try: + Scheduler(bad, "proc-C", 1, {}) + assert False, "unsupported locking must reject scheduler startup" + except StartupRejected: + pass + + +def test_corrupt_view_without_admission_fails_closed(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "a", + PeriodicSource(timedelta(hours=1), t0), + sources, + t0 - timedelta(hours=1), + ) + st.runs["ghost-1"] = Run("ghost-1", "a", t0, 1, {"team": "eng"}) + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + diags = sch.recover(t0) + assert any("corrupt-blocked" in d for d in diags) + assert st.schedules["a"].blocked_reason is not None + try: + sch.poll(t0 + timedelta(hours=1)) + assert False, "corrupt records fail closed with diagnostics" + except BlockedSchedule: + pass + + +def test_overlap_parallel_x_misfire_latest(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched( + st, + "p", + PeriodicSource(timedelta(minutes=5), t0), + sources, + t0 - timedelta(minutes=5), + overlap="parallel", + max_active=2, + misfire="latest", + ) + with Scheduler(st, "p1", 0, sources) as sch: # no task slots: hold + sch.poll(t0 + timedelta(minutes=12)) # 12:00,05,10 missed + cand = st.candidates["p"] + assert cand is not None and cand.intended == t0 + timedelta(minutes=10) + assert [r for r in st.history if r.kind == "admitted"] == [] + with Scheduler(st, "p1", 1, sources, {"*": "hang"}) as sch: + sch.poll(t0 + timedelta(minutes=13)) # admits held 12:10, hangs + assert st.runs["run-p-1"].intended == t0 + timedelta(minutes=10) + # Task slot taken: 12:15 is held as the one latest candidate. + sch.poll(t0 + timedelta(minutes=15)) + assert list(st.runs) == ["run-p-1"] + assert st.candidates["p"] is not None + assert st.candidates["p"].intended == t0 + timedelta(minutes=15) # type: ignore[union-attr] + # Slot frees: the held 12:15 candidate is admitted (hangs). + st.runs["run-p-1"].state = COMPLETED + sch.poll(t0 + timedelta(minutes=16)) + assert st.runs["run-p-2"].intended == t0 + timedelta(minutes=15) + # run-p-2 waits durably: schedule slot held, task slot free. Admit a + # second hanging run to reach the cap, then the next due instant is + # terminally skipped-overlap. + st.runs["run-p-2"].state = INTERRUPTED + st.history.append( + Record( + kind="interrupted", + sched_id="p", + intended=t0 + timedelta(minutes=15), + run_id="run-p-2", + ) + ) + sch.poll(t0 + timedelta(minutes=20)) # admits 12:20, hangs + assert st.runs["run-p-3"].intended == t0 + timedelta(minutes=20) + sch.poll(t0 + timedelta(minutes=25)) # 2 active >= max: terminal skip + assert any( + r.kind == "skipped-overlap" and r.intended == t0 + timedelta(minutes=25) + for r in st.history + ) + # And a later catch-up admits the earliest missed instant ASAP + # (12:30), holds only the newest (12:45), and never resurrects the + # skipped 12:25. + st.runs["run-p-2"].state = COMPLETED + st.runs["run-p-3"].state = COMPLETED + sch.poll(t0 + timedelta(minutes=45)) # 12:30..45 missed + just = [ + r + for r in st.history + if r.kind == "admitted" and r.intended == t0 + timedelta(minutes=30) + ] + assert len(just) == 1 + cand = st.candidates["p"] + assert cand is not None and cand.intended == t0 + timedelta(minutes=45) + assert all( + r.intended != t0 + timedelta(minutes=25) + for r in st.history + if r.kind == "admitted" + ) + + +def test_manual_and_other_schedule_runs_are_overlap_independent(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + add_sched(st, "b", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + # Manual runs and other schedules' runs never join this schedule's check. + st.runs["manual-1"] = Run("manual-1", "manual", t0, 1, {"team": "eng"}) + st.runs["run-b-0"] = Run("run-b-0", "b", t0, 1, {"team": "eng"}) + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + sch.poll(t0) + admitted_a = [ + r for r in st.history if r.kind == "admitted" and r.sched_id == "a" + ] + assert len(admitted_a) == 1, "overlap=skip ignores manual/other runs" + # ...while b is blocked by its OWN running run (per-schedule scope + # cuts both ways: b's check sees run-b-0, a's check does not). + skipped_b = [ + r for r in st.history if r.kind == "skipped-overlap" and r.sched_id == "b" + ] + assert len(skipped_b) == 1 and skipped_b[0].intended == t0 + + +def test_capacity_wait_then_admit_or_expire_for_skip(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + with Scheduler(st, "p1", 0, sources) as sch: # full: within allowance + assert sch.poll(t0) == {"a": "admit:held-undecided"} + assert st.consumed["a"] < t0, "undecided instants stay unconsumed" + assert st.candidates["a"] is None, "skip holds no candidate" + with Scheduler(st, "p1", 1, sources, {"*": "complete"}) as sch: + sch.poll(t0 + timedelta(seconds=30)) # still within allowance + admitted = [r for r in st.history if r.kind == "admitted"] + assert len(admitted) == 1 and admitted[0].intended == t0 + # Past the deadline instead: capacity-delayed skip expires. + st2 = make_store() + sources2: dict = {} + add_sched( + st2, + "a", + PeriodicSource(timedelta(minutes=10), t0), + sources2, + t0 - timedelta(hours=1), + ) + with Scheduler(st2, "p1", 0, sources2) as sch: + sch.poll(t0) + assert sch.poll(t0 + timedelta(seconds=61))["a"] == "skipped-misfire" + assert any( + r.kind == "skipped-misfire" and r.intended == t0 for r in st2.history + ) + + +def test_fault_after_admission_recovers_exactly_once(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + st.faults["admission"] = "after" # record persisted, crash before view + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + try: + sch.poll(t0) + assert False + except InjectedFault: + pass + assert len(st.admissions) == 1 and len(st.runs) == 0 + # Recovery materializes the view but never executes: pending. + diags = sch.recover(t0) + assert any("pending-dispatch" in d for d in diags) + run = st.runs["run-a-1"] + assert run.state == RUNNING and run.needs_dispatch + assert [r for r in st.history if r.kind == "completed"] == [] + sch.poll(t0) # sweep dispatches exactly once + assert run.state == COMPLETED and not run.needs_dispatch + assert len([r for r in st.history if r.kind == "admitted"]) == 1 + assert len([r for r in st.history if r.kind == "completed"]) == 1 + + +def test_fault_after_complete_reconciles_terminal_record(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + st.faults["complete"] = "after" # COMPLETED persisted, record lost + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + try: + sch.poll(t0) + assert False + except InjectedFault: + pass + assert st.runs["run-a-1"].state == COMPLETED + assert [r for r in st.history if r.kind == "completed"] == [] + diags = sch.recover(t0) + assert any("terminal-reconciled" in d for d in diags) + assert len([r for r in st.history if r.kind == "completed"]) == 1 + sch.poll(t0 + timedelta(hours=1)) # consumed advanced: no re-admit + assert len([r for r in st.history if r.kind == "admitted"]) == 1 + + +def test_fault_after_resume_mark_fails_closed_not_retried(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch: + sch.poll(t0) + (run_id,) = list(st.runs) + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + st.faults["resume-mark"] = "after" # ACTIVE persisted, never executed + try: + sch.resume_run(run_id, t0) + assert False + except InjectedFault: + pass + assert st.runs[run_id].state == INTERRUPTED # never re-ran + assert st.resume_attempts[run_id] == "ACTIVE" + diags = sch.recover(t0) # conservative: ambiguous, never retried + assert st.runs[run_id].state == FAILED + assert "ambiguous" in st.runs[run_id].fail_reason + assert any("failed-closed" in d for d in diags) + + +def test_fault_after_interrupt_reconciles_waiting_state(): + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + st.faults["interrupt-persist"] = "after" # INTERRUPTED kept, record lost + with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch: + try: + sch.poll(t0) + assert False + except InjectedFault: + pass + assert st.runs["run-a-1"].state == INTERRUPTED + diags = sch.recover(t0) + assert st.runs["run-a-1"].state == INTERRUPTED # still resumable + assert any("terminal-reconciled" in d for d in diags) + # A resumed run may interrupt AGAIN: re-interruption is a durable + # terminal persist of its own, and the run stays resumable after it. + assert sch.resume_run("run-a-1", t0) == "resumed-interrupted" + assert st.runs["run-a-1"].state == INTERRUPTED + assert st.resume_attempts["run-a-1"] == "DONE" + assert any( + r.kind == "interrupted" and r.reason == "resumed-reinterrupted" + for r in st.history + ) + sch.outcomes["run-a-1"] = "complete" + assert sch.resume_run("run-a-1", t0) == "resumed-complete" + + +def _interrupted_run() -> tuple[MemStore, dict, str]: + st = make_store() + sources: dict = {} + t0 = ts(2026, 9, 8, 12, 0) + add_sched(st, "a", OneShotSource(t0), sources, t0 - timedelta(hours=1)) + with Scheduler(st, "p1", 4, sources, {"*": "interrupt"}) as sch: + sch.poll(t0) + (run_id,) = list(st.runs) + assert st.runs[run_id].state == INTERRUPTED + return st, sources, run_id + + +def test_fault_after_resume_complete_reconciles_without_rerun(): + st, sources, run_id = _interrupted_run() + t0 = ts(2026, 9, 8, 12, 0) + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + st.faults["resume-complete"] = "after" # COMPLETED kept, rest lost + try: + sch.resume_run(run_id, t0) + assert False + except InjectedFault: + pass + assert st.runs[run_id].state == COMPLETED + assert st.resume_attempts[run_id] == "ACTIVE" + assert [r for r in st.history if r.kind == "completed"] == [] + diags = sch.recover(t0) + # Reconciled, never re-executed: exactly one completion, DONE marker. + assert st.resume_attempts[run_id] == "DONE" + assert any("attempt-reconciled" in d for d in diags) + assert len([r for r in st.history if r.kind == "completed"]) == 1 + sch.poll(t0 + timedelta(hours=1)) + assert len([r for r in st.history if r.kind == "completed"]) == 1 + + +def test_fault_after_resume_attempt_clear_reconciles_record(): + st, sources, run_id = _interrupted_run() + t0 = ts(2026, 9, 8, 12, 0) + with Scheduler(st, "p1", 4, sources, {"*": "complete"}) as sch: + st.faults["resume-attempt-clear"] = "after" # DONE kept, record lost + try: + sch.resume_run(run_id, t0) + assert False + except InjectedFault: + pass + assert st.runs[run_id].state == COMPLETED + assert st.resume_attempts[run_id] == "DONE" + assert [r for r in st.history if r.kind == "completed"] == [] + diags = sch.recover(t0) + assert any("terminal-reconciled" in d for d in diags) + assert len([r for r in st.history if r.kind == "completed"]) == 1 + + +def test_fault_before_resume_interrupt_fails_closed(): + st, sources, run_id = _interrupted_run() + t0 = ts(2026, 9, 8, 12, 0) + with Scheduler(st, "p1", 4, sources, {"run-a-1": "interrupt"}) as sch: + st.faults["resume-interrupt"] = "before" # re-interrupt never persisted + try: + sch.resume_run(run_id, t0) + assert False + except InjectedFault: + pass + # The ACTIVE marker superseded the old waiting checkpoint, but the + # re-interruption never landed: fail closed, never retry the old one. + assert st.runs[run_id].state == RUNNING + assert st.resume_attempts[run_id] == "ACTIVE" + assert st.runs[run_id].result_attempt != st.runs[run_id].attempt_id + diags = sch.recover(t0) + assert st.runs[run_id].state == FAILED + assert "ambiguous" in st.runs[run_id].fail_reason + assert any("failed-closed" in d for d in diags) + + +def test_fault_after_resume_interrupt_stays_resumable(): + # Exact repro shape: the re-interruption persisted WITH the attempt + # identity, then the crash hit before attempt-clearing. Recovery must + # match result to attempt and keep the run resumable — not fail it. + st, sources, run_id = _interrupted_run() + t0 = ts(2026, 9, 8, 12, 0) + with Scheduler(st, "p1", 4, sources, {"run-a-1": "interrupt"}) as sch: + st.faults["resume-interrupt"] = "after" + try: + sch.resume_run(run_id, t0) + assert False + except InjectedFault: + pass + assert st.runs[run_id].state == INTERRUPTED + assert st.resume_attempts[run_id] == "ACTIVE" + assert st.runs[run_id].result_attempt == st.runs[run_id].attempt_id + assert st.runs[run_id].attempt_id > 0 + diags = sch.recover(t0) + assert st.runs[run_id].state == INTERRUPTED, "fresh result: resumable" + assert st.resume_attempts[run_id] == "DONE" + assert any("fresh-result-resumable" in d for d in diags) + assert len([r for r in st.history if r.kind == "interrupted"]) == 1 + sch.outcomes[run_id] = "complete" + assert sch.resume_run(run_id, t0) == "resumed-complete"