Feat/2393 durable automation runs

[BelhsanHmida]

Description

• Gave every scheduled automation run a durable record, so a run whose queueing failed before any job existed is retryable instead of lost.
• Added three tables: automation_run (one row per scheduled run, unique on (automation_id, scheduled_at, schedule_revision)), automation_run_job (the jobs a run intends to create, with deterministic RQ job IDs) and automation_run_attempt (one row per dispatch attempt).
• Added Automation.schedule_revision, counted up whenever flexmeasures edit automation rebases the cursor, so runs of the old and the new schedule stay distinct.
• Moved run ownership from the Redis SET NX guard to the database, keeping the same cursor compare-and-swap. Removes claim_due_automation() and the Redis guard.
• Gave each claim a ten-minute lease, so a runner that died mid-queueing hands its work over, while one still queueing is left alone.
• Persisted the run's plan before the first enqueue, so a retry re-queues only the jobs still missing, with the parameters the run was planned with.
• Tracked dispatch state (pending, claimed, partially_queued, queued, failed) separately from execution state (pending, running, succeeded, failed, canceled), the latter derived from all job intents so a late success cannot hide an earlier failure.
• Recorded job start, success and failure against the run, rolling the session back first so an outcome survives a job that failed on the database.
• Added schedule_revision and a run_stats object to the automations API, and the same run status to the asset's Automations page. Backward compatible: no existing field changed.
• Made flexmeasures jobs run-automations report the run and attempt it works on, and retry runs whose queueing did not finish.
• Added changelog item in documentation/changelog.rst

Look & Feel

flexmeasures jobs run-automations, on a run whose previous attempt queued only part of its jobs:

Automation 42 ('Partial run') run 7 queued 3 forecasting job(s), scheduled for 2026-08-05 01:00:00+00:00.

and when an attempt fails, naming the run and the attempt rather than just the automation:

Automation 42 ('Partial run') run 7 failed while dispatching attempt 2: lost Redis connection

[GET] /assets/(id)/automations/(automation_id) gains run_stats:

"run_stats": {
  "total": 4,
  "dispatch": {"queued": 3, "partially_queued": 1},
  "execution": {"succeeded": 3, "pending": 1},
  "latest_run": {
    "id": 12,
    "scheduled_at": "2026-07-11T04:00:00+00:00",
    "schedule_revision": 1,
    "dispatch_state": "partially_queued",
    "execution_state": "pending",
    "attempt_count": 2,
    "intended_job_count": 2,
    "queued_job_count": 1,
    "claim_owner": "runner-1@host",
    "last_error": {"type": "ConnectionError", "message": "lost Redis connection"},
    "latest_attempt": {"attempt_no": 2, "owner": "runner-1@host", "outcome": "failed", "queued_job_count": 1},
    "jobs": [
      {"logical_job_key": "cycle-001", "rq_job_id": "automation-run-12-cycle-001", "status": "queued"},
      {"logical_job_key": "wrap-up", "rq_job_id": "automation-run-12-wrap-up", "status": "pending"}
    ]
  },
  "recent_runs": []
}

The automation details modal on the asset's Automations page shows the schedule revision and a Durable runs block with the same totals, dispatch and execution breakdown, and latest run — next to the existing Recently created jobs counts, which come from Redis and disappear as jobs expire, whereas the run records do not.

How to test

# 1. create an automation and let it become due
flexmeasures add automation --asset 3 --name "Daily PV forecasts" --cron "* * * * *" --sensor 12
flexmeasures jobs run-automations

# 2. inspect the durable run, on the asset's Automations page or through the API
curl -H "Authorization: $AUTH_TOKEN" \
  "$HOST/api/v3_0/assets/3/automations/<automation_id>" | jq .run_stats

# 3. break the dispatch halfway (stop Redis, or kill the runner mid-queueing), then run it again.
#    The retry queues only the jobs still missing, on the same run record, with attempt_count
#    incremented — no second run row, and no duplicate jobs.

Automated coverage:

pytest \
  flexmeasures/data/tests/test_automation_runs_fresh_db.py \
  flexmeasures/data/tests/test_automation_scheduling_fresh_db.py \
  flexmeasures/data/tests/test_automations_fresh_db.py \
  flexmeasures/cli/tests/test_automations.py \
  flexmeasures/api/v3_0/tests/test_automations_api.py \
  flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py

Related Items

Closes #2393.

---

Sign-off

• I agree to contribute to the project under Apache 2 License.
• To the best of my knowledge, the proposed patch is not based on code under GPL or other license that is incompatible with FlexMeasures

[read-the-docs-community]

Documentation build overview

📚 flexmeasures | 🛠️ Build #34508798 | 📁 Comparing 45f8bb7 against latest (d4e0805)

  🔍 Preview build  

52 files changed · + 3 added · ± 49 modified

+ Added

• _autosummary/flexmeasures.data.schemas.scheduling.config.html
• _autosummary/flexmeasures.data.services.data_generators.html
• _autosummary/flexmeasures.data.services.reporting.html

± Modified

• changelog.html
• configuration.html
• genindex.html
• http-routingtable.html
• py-modindex.html
• _autosummary/flexmeasures.api.common.utils.deprecation_utils.html
• _autosummary/flexmeasures.api.v3_0.assets.html
• _autosummary/flexmeasures.api.v3_0.sensors.html
• _autosummary/flexmeasures.cli.jobs.html
• _autosummary/flexmeasures.data.models.automations.html
• and 39 more...

[Copilot]

🟡 Changes recommended

record_automation_job_started can overwrite a previously recorded failed run back to “running”, and several new docstrings violate the repo’s line-break-after-punctuation rule.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR introduces durable, database-backed automation run records to make scheduled automation dispatch idempotent and safely retryable, while exposing run/attempt/job outcome details through the API, CLI output, and UI.

Changes:

• Add durable run/attempt/job tables plus claiming/lease logic to support safe retries after pre-enqueue and partial-enqueue failures.
• Wire durable run identity through queued forecasting jobs and record durable execution outcomes (started/succeeded/failed).
• Expose schedule_revision and run_stats via API/OpenAPI, surface the same in the asset Automations UI, and document the new behavior.

File summaries

File	Description
flexmeasures/ui/templates/assets/asset_automations.html	Display schedule_revision and durable run_stats in the automation details modal
flexmeasures/ui/static/openapi-specs.json	Update OpenAPI examples/description for schedule_revision and run_stats
flexmeasures/data/tests/test_automation_scheduling_fresh_db.py	Update scheduling tests to claim durable runs instead of legacy claims
flexmeasures/data/tests/test_automation_runs_fresh_db.py	Add regression coverage for durable run claiming, retry semantics, and durable outcome recording
flexmeasures/data/services/forecasting.py	Record durable job failure info from forecasting exception handler when run metadata is present
flexmeasures/data/services/automations.py	Core implementation of durable run claiming, job intents, dispatch tracking, reconciliation, and serialization/stats
flexmeasures/data/schemas/automations.py	Add schedule_revision to automation schema output
flexmeasures/data/models/forecasting/pipelines/train_predict.py	Plan deterministic job IDs for automation runs and record durable job outcomes
flexmeasures/data/models/data_sources.py	Extend job trigger metadata to include automation_run_id
flexmeasures/data/models/automations.py	Add models for AutomationRun, AutomationRunAttempt, AutomationRunJob, and Automation.schedule_revision
flexmeasures/data/migrations/versions/f3d8e2c9a741_add_durable_automation_runs.py	Alembic migration creating new durable run tables and adding schedule_revision
flexmeasures/cli/tests/test_automations.py	Update runner CLI test to expect durable run/attempt reporting
flexmeasures/cli/jobs.py	Replace Redis guard with durable run dispatch + retry behavior and improved CLI messaging
flexmeasures/cli/data_edit.py	Increment schedule_revision on schedule rebase (cursor reset)
flexmeasures/api/v3_0/tests/test_sensor_data.py	Ensure DB error listener is always removed to avoid leaking into later tests
flexmeasures/api/v3_0/tests/test_automations_api.py	Add API assertions/examples for schedule_revision and run_stats outcome distinctions
flexmeasures/api/v3_0/assets.py	Include durable run_stats in automation detail API responses
documentation/features/automations.rst	Document durable runs, retries, leases, and dispatch vs. execution state concepts
documentation/cli/change_log.rst	Document durable run recording and retry semantics for the automation runner CLI
documentation/changelog.rst	Add a main changelog entry describing durable automation run tracking
documentation/api/change_log.rst	Document new API fields (schedule_revision, run_stats) and their semantics

Review details

Suppressed comments (6)

flexmeasures/data/services/automations.py:226

• Docstring line wrap breaks mid-phrase (line ends with "to"), violating the repository rule to only break lines after punctuation.

    A run is only up for grabs once no other runner holds a live claim on it, because the dispatch state turns to
    'partially_queued' while the owning runner is still queueing the rest of its jobs.

flexmeasures/data/services/automations.py:461

• Docstring wraps mid-phrase (line ends with "where" / "own"), violating the repo-wide rule to only break lines after punctuation.

    The job may well have failed on the database itself, which leaves the session in an aborted transaction where
    every further statement is refused. Roll back first, so that the failure is still recorded. The job's own
    uncommitted work is lost either way, since it is failing.

flexmeasures/data/services/automations.py:418

• Docstring wraps mid-phrase (line ends with "the"), violating the repo-wide rule to only break lines after punctuation.

    A failed job keeps the whole run failed: a later job succeeding, as the wrap-up job does whatever became of the
    cycle jobs it reports on, must not put the run back to 'running' and bury the failure.

flexmeasures/data/tests/test_automation_runs_fresh_db.py:330

• Docstring wraps mid-phrase (line ends with "durable"), violating the repo-wide rule to only break lines after punctuation.

    All jobs are already in Redis, so a later runner must adopt the abandoned claim, reconcile the durable
    intents against Redis, and complete the dispatch without queueing anything again.

flexmeasures/data/tests/test_automation_runs_fresh_db.py:382

• Docstring wraps mid-phrase (line ends with "run"), violating the repo-wide rule to only break lines after punctuation.

    The occurrence stays claimed until the lease runs out, after which a later runner adopts the same durable run
    instead of creating a second one.

flexmeasures/data/tests/test_automation_runs_fresh_db.py:676

• Docstring wraps mid-phrase (line ends with "as"), violating the repo-wide rule to only break lines after punctuation.

    The wrap-up job succeeds whatever became of the cycle jobs it reports on, so it must not report the run as
    merely still running and bury the failure an operator needs to see.

• Files reviewed: 21/21 changed files
• Comments generated: 4
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Flix6x]

Merged main (which now carries schedule automations, #2293) into this branch. Four conflicts, plus two things the merge needed a decision on.

Conflicts

• documentation/cli/change_log.rst — kept main's --type forecasting|scheduling wording, with this PR's durable-claim sentence replacing "at most one queueing attempt per automation per minute", which this PR supersedes.
• flexmeasures/cli/jobs.py — the run-automations line now reports both: the run and its scheduled time (this PR) and the automation's type and asset (main). No longer hardcodes "forecasting".
• flexmeasures/data/services/automations.py — union of both sides, except run_automation, which main had split into _run_forecast_automation / _run_schedule_automation. Both now take the AutomationRun, so a schedule run also uses the parameters it was planned with, and names itself on the job's trigger.
• flexmeasures/ui/static/openapi-specs.json — regenerated rather than hand-merged.

Decisions

1. A schedule run is not dispatched a second time. This PR's retry safety rests on job IDs derived from the run, which only the forecasting pipeline produces; create_*_scheduling_job gets a fresh UUID per dispatch, and passing a single job_id into the sequential variant would give every device job the same ID. Retrying a schedule run would therefore have duplicated its schedules — a regression the merge would have introduced silently. get_dispatchable_automation_runs now resumes only forecast runs; a schedule run is still recorded, claimed, reported and shown, but left where it failed. Docs and both changelogs say so. Extending deterministic job IDs to the scheduling path is worth a follow-up.
2. The migration chains after main's merge revision. f3d8e2c9a741 pointed at 8f4a1d0c2e77, which after Schedules as automations #2293 left two heads; it now follows c7a2f13b9e04. Verified: a single head, 121 revisions reachable from base.

documentation/changelog.rst was rebuilt from main's copy plus this PR's entry — the automatic merge had duplicated the two automations entries that #2293 rewrote, and dropped #2507's bugfix line.

Tests: 1149 passed, 1 xfailed across data, cli, api/v3_0 and ui. Two new tests, each shown to fail when the behaviour it covers is broken:

• test_a_durable_run_schedules_what_it_was_planned_with — a run dispatched from its record uses its stored parameters, not the automation's edited ones, and carries automation_run_id on the job.
• test_only_a_forecast_run_is_dispatched_a_second_time — the retry gate above.

🤖 Generated with Claude Code

https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk

[Flix6x]

This half-overclaims, and the half that is wrong has a visible consequence.

The parameters part is true. claim_due_automation_run stores parameters=dict(automation.parameters or {}), so an edit after the run was claimed does not reach it.

The timings part is not. Only timings that were explicitly given are in those parameters. One that defaults to the run time was never stored, so it is re-resolved on every attempt: TrainPredictPipelineParametersSchema sets predict_start = floored_now when start is absent (flexmeasures/data/schemas/forecasting/pipeline.py:658), and belief_time defaults to now alongside it. ensure_automation_run_job_intents reuses the stored intents, but the job specs handed to it are recomputed each attempt, so identity is preserved across a retry while the resolved window is not.

The consequence is that one run's jobs can disagree about what they are forecasting.

Repro

Drop this in as flexmeasures/data/tests/test_repro_timings.py and run it. A forecast automation with no explicit --start; attempt 1 at 01:00; one job dropped from Redis to simulate a dispatch that queued only part of its work; attempt 2 at 03:00.

from datetime import datetime, timezone

from rq.job import Job
from sqlalchemy import select

from flexmeasures.cli.tests.utils import to_flags
from flexmeasures.data.models.automations import Automation


def test_a_resumed_run_reresolves_a_start_it_never_stored(
    app, fresh_db, setup_fresh_test_forecast_data, freeze_server_now, clean_redis
):
    from flexmeasures.cli.data_add import add_automation
    from flexmeasures.data.services.automations import (
        dispatch_automation_run,
        get_dispatchable_automation_runs,
    )

    sensor = setup_fresh_test_forecast_data["solar-sensor"]

    # A forecast automation with no explicit --start: its start defaults to the run time.
    freeze_server_now(datetime(2026, 8, 5, 0, 58, tzinfo=timezone.utc))
    runner = app.test_cli_runner()
    result = runner.invoke(
        add_automation,
        to_flags(
            {
                "asset": sensor.generic_asset_id,
                "name": "No explicit start",
                "cron": "0 1 * * *",
                "timezone": "UTC",
                "sensor": sensor.id,
                "duration": "PT2H",
                "forecast-frequency": "PT1H",
                "max-forecast-horizon": "PT2H",
                "retrain-frequency": "PT1H",
            }
        ),
    )
    assert result.exit_code == 0, result.output
    automation = fresh_db.session.scalars(select(Automation)).one()

    # Attempt 1, at 01:00.
    freeze_server_now(datetime(2026, 8, 5, 1, 0, tzinfo=timezone.utc))
    claimed = get_dispatchable_automation_runs(owner="runner:1")
    assert len(claimed) == 1
    run = claimed[0].run
    dispatch_automation_run(claimed[0])
    fresh_db.session.refresh(run)
    print("\nrun.parameters (snapshot):", dict(run.parameters))

    connection = app.queues["forecasting"].connection
    starts = lambda: {
        i.logical_job_key: Job.fetch(i.rq_job_id, connection=connection).meta.get("start")
        for i in run.job_intents
    }
    first = starts()
    print("attempt 1 job starts:", first)

    # Simulate an attempt that queued only part of its jobs: drop one job from Redis,
    # put its intent back to pending, and hand the run back as resumable.
    victim = next(i for i in run.job_intents if i.logical_job_key == "cycle-002")
    Job.fetch(victim.rq_job_id, connection=connection).delete()
    victim.status = "pending"
    victim.enqueued_at = None
    run.dispatch_state = "partially_queued"
    run.dispatch_completed_at = None
    run.claim_owner = None
    run.claim_expires_at = None
    fresh_db.session.commit()

    # Attempt 2, two hours later.
    freeze_server_now(datetime(2026, 8, 5, 3, 0, tzinfo=timezone.utc))
    resumed = [c for c in get_dispatchable_automation_runs(owner="runner:2") if c.run.id == run.id]
    assert resumed, "the run was not picked up again"
    dispatch_automation_run(resumed[0])
    fresh_db.session.refresh(run)
    print("attempt 2 job starts:", starts())

Output:

run.parameters (snapshot): {'sensor': '2', 'duration': 'PT2H', 'forecast-frequency': 'PT1H', 'max-forecast-horizon': 'PT2H'}
attempt 1 job starts: {'cycle-001': '2026-08-05T01:00:00+00:00', 'cycle-002': '2026-08-05T01:00:00+00:00', 'wrap-up': '2026-08-05T01:00:00+00:00'}
attempt 2 job starts: {'cycle-001': '2026-08-05T01:00:00+00:00', 'cycle-002': '2026-08-05T03:00:00+00:00', 'wrap-up': '2026-08-05T01:00:00+00:00'}

No start in the stored parameters, because none was ever given. After the resume, cycle-001 and cycle-002 of the same run forecast from different starts, and the wrap-up job reports on the 01:00 window.

Suggestion

The smallest fix is to say what is actually guaranteed, e.g.:

Because the plan is stored, a retry hours later still uses the parameters the run was planned with, even if the automation has been edited since. Timings the automation left to the run time are not part of those parameters, so they are resolved afresh on each attempt: a resumed run's jobs can therefore cover a later window than the ones its first attempt queued.

The larger fix is to resolve the run's timings once, at claim time, and store them on AutomationRun.parameters — which would make the sentence true as written, and give a resumed run one coherent window. That is a behaviour change, so it may belong in its own PR; the docs edit is enough to make this one accurate.

Same question applies to AutomationRun.parameters for schedule automations, where prepare_schedule_trigger_message defaults start to server_now(). It does not bite today, since schedule runs are not retried, but it would as soon as they are (#2510).

[BelhsanHmida]

Checked it you're right. Ran your test locally and got the same split: cycle-002 at 03:00 while cycle-001 and the wrap-up stayed at 01:00. pushed docs update commits

I have the behaviour fix working locally too: pin the resolved start onto the run on first dispatch, so later attempts read it back. Start only pinning end as well trips the "duration with either start or end, but not both" check. It settles belief_time too, since the schema derives that from the start whenever one is given.

Keeping it out of this PR as you suggested. i'll make a follow-up pr