Give scheduling jobs deterministic IDs, so a schedule automation run can be retried

[Flix6x]

Summary

#2457 gave automation runs durable records and safe retries, but only forecast runs are actually dispatched a second time. Schedule runs are recorded, claimed, reported and shown like any other, and then left where they failed. This issue is about lifting that restriction.

The gate is one clause in get_dispatchable_automation_runs (flexmeasures/data/services/automations.py):

# Only a forecast run can be dispatched a second time safely.
# Its jobs carry IDs derived from the run, so a retry recognizes the ones it already queued.
# A schedule run's jobs get a fresh ID on every dispatch, so retrying one would duplicate its schedules,
# which is why such a run is recorded and reported, but left where it failed.
AutomationRun.automation_type == "forecasting",

Why the restriction exists

The whole of #2457's retry safety rests on one property: a job's RQ ID is a function of the run and a logical key, rather than a random UUID. TrainPredictPipeline._plan_cycle_jobs builds IDs like automation-run-12-cycle-001 and automation-run-12-wrap-up, records an AutomationRunJob intent per job before enqueueing any of them, and on a retry asks Redis whether each ID already exists (reconcile_automation_job_intent). That is what separates "a retry resumes" from "a retry duplicates".

Scheduling has no equivalent. create_scheduling_job takes job_id: str | None = None and hands it to Job.create; nothing in the automation path sets it, so RQ invents a UUID. Two dispatches of the same run produce two unrelated IDs, and there is no way to tell an already-queued schedule from a missing one.

@job_cache("scheduling") is not a substitute. It dedupes on an argument hash with a Redis TTL (FLEXMEASURES_JOB_CACHE_TTL, one hour by default), so it is gone exactly when a run outlives its claim lease or Redis restarts — which is the failure mode durable runs exist to survive. It also hashes the call arguments, so anything that varies between attempts breaks it: AutomationRun.parameters is a snapshot of the automation's raw parameters (see claim_due_automation_run), so a schedule automation that omits start resolves a fresh one per attempt via prepare_schedule_trigger_message and hashes differently. And per #2404 the cache can hand back a wedged job rather than a fresh one.

What needs to happen

1. Give each device job its own ID. create_sequential_scheduling_job currently forwards a single job_id to every create_scheduling_job call in its loop (flexmeasures/data/services/scheduling.py:514). That is harmless today because every caller passes None; the moment a real ID is passed, all device jobs collide on it. This needs a per-device ID before anything else can be built on top.
2. Give the wrap-up job an ID. The final cb_done_sequential_scheduling_job job is built with a bare Job.create and has no job_id parameter at all.
3. Register job intents for schedule jobs, with logical keys along the lines of device-<sensor id> and wrap-up, and queue="scheduling", so the existing ensure_automation_run_job_intents / mark_automation_job_queued / reconcile_automation_job_intent machinery applies unchanged.
4. Rebuild the depends_on chain on a resumed dispatch against the jobs that already exist, instead of re-creating the chain from scratch.
5. Record schedule job outcomes against the run. Today only the forecasting pipeline calls record_automation_job_started / _succeeded / _failed, so a schedule run's execution_state never leaves pending and the Automations page shows it that way indefinitely. This is arguably the more visible half of the gap, and is worth doing even if 1–4 are deferred.
6. Then remove the gate and its test (test_only_a_forecast_run_is_dispatched_a_second_time), and update documentation/features/automations.rst and both changelogs, which currently spell the restriction out.

Notes

• The fallback scheduler is not a constraint here. Since Feat/retire fallback scheduler #2252 no core scheduler sets fallback_scheduler_class, so trigger_optional_fallback never fires in core; the hook survives for plugins that set it. The only thing to watch is that a fallback job, which inherits the original's trigger meta and is created with force_new_job_creation=True, must not be handed the deterministic ID of the job it replaces.
• Overlap with Retiring the storage fallback scheduler leaves failed sequential job chains wedged, with no terminal state for clients #2404. That issue covers what happens to a sequential chain whose first subjob fails: the dependents stay deferred forever with no terminal state. Items 2–4 here touch the same code and the same chain semantics, so the two are worth sequencing together — a deterministic ID scheme is easier to reason about once a failed chain has a defined end state.
• Steps 1 and 2 change shared scheduling code that the API trigger endpoint, flexmeasures add schedule and any plugin scheduler all go through, so they deserve their own tests independent of automations.

Acceptance criteria

• A schedule automation run whose dispatch fails after queueing some of its device jobs is picked up again, and queues only the jobs it still owes.
• A schedule run's execution state reflects what its jobs did, rather than staying pending.
• Re-dispatching a schedule run never produces a second schedule for a device that already had one queued for that run.