Schedulers as data generators, and a generator on every automation

documentation/changelog.rst

@@ -14,10 +14,19 @@ v1.1.0 | September XX, 2026
              If you maintain indexes of your own on this table, note that the reordered primary key leads with ``(sensor_id, source_id, event_start, belief_horizon)``, so any index you keep on a prefix of that is now redundant and can be dropped once the migration has run.
              The migration names the ones it finds and leaves them in place, as it cannot know which ones you meant to keep.
 
+.. warning:: A scheduler's data source now also records the flex config it computed under, where previously one data source per scheduler version recorded every schedule that scheduler made.
+             Schedules computed under different flex configs are therefore recorded by different data sources, and a sensor can carry schedules from several of them, as it already could for forecasts.
+             Values describing a single moment stay out of that config, so a ``soc-at-start``, or a ``soc-targets`` entry at a given datetime, does not make every run a new data source.
+             What does is a change to what the site and its devices can do, such as a device's ``power-capacity``.
+             After such a change, a sensor holds the schedule computed under each configuration, where the newer schedule used to supersede the older one, so a chart of that sensor draws both, and the asset's KPIs total both, as they report what the chart draws.
+             Select a data source to see the schedule computed under one configuration.
+             One scheduling request still records under a single data source, including the per-device jobs of a sequential schedule.
+
 New features
 -------------
 
 * Automations: recurring tasks defined per asset, computing forecasts or schedules, managed with new CLI commands (``flexmeasures add|edit|delete automation``), run by ``flexmeasures jobs run-automations``, and viewable in a new UI page and API endpoints (``[GET] /assets/(id)/automations``); each automation interprets its recurrence in its own timezone, and runs missed while the runner was down are caught up once, coalesced into one current forecast; a forecast automation points at a data source holding its forecaster configuration, while a schedule automation stores what the schedule trigger endpoint accepts, and schedules from each run's own time, unless the trigger message fixes a ``start``; an automation's details link to the sensors it reads from and writes to, a sensor's page lists the automations feeding it, and deleting a sensor warns about the automations that use it; jobs now also record whether they were created via the CLI, the API or an automation [see `PR #2290 <https://www.github.com/FlexMeasures/flexmeasures/pull/2290>`_, `PR #2396 <https://www.github.com/FlexMeasures/flexmeasures/pull/2396>`_ and `PR #2293 <https://www.github.com/FlexMeasures/flexmeasures/pull/2293>`_]
+* A scheduler's data source now also records the flex config the scheduler computed under, so a schedule can be traced back to the configuration that produced it, and a schedule automation points at such a data source, the way a forecast automation points at its forecaster's [see `PR #2444 <https://www.github.com/FlexMeasures/flexmeasures/pull/2444>`_]
 * In the UI, the full record of the data source selected on a sensor page can be inspected, backed by a new API endpoint (``[GET] /sources/(id)``) [see `PR #2290 <https://www.github.com/FlexMeasures/flexmeasures/pull/2290>`_]
 * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 <https://www.github.com/FlexMeasures/flexmeasures/pull/2433>`_]
 

documentation/cli/change_log.rst

@@ -13,6 +13,7 @@ since v1.0.0 | August 11, 2026
 * Add ``flexmeasures edit secret`` to store an encrypted secret on an account or asset.
 * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset.
 * Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, with ``--type forecasting`` or ``--type scheduling`` saying which task to automate). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted.
+* ``flexmeasures add automation --type scheduling`` refuses a flex config field which fixes a moment in time, such as ``soc-at-start`` or a ``soc-targets`` entry with a ``datetime``, naming the field: a recurring schedule automation computes a fresh schedule on every run, so such a value would be stale on the next one.
 * Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once.
 * ``flexmeasures delete sensor`` now warns which automations read from or write to a sensor before it is deleted, as an automation refers to its sensors by ID and would fail on its next run.
 * ``flexmeasures show data-sources`` now shows the account a data source belongs to, and lists the sensors holding data recorded by a single source with ``--show-sensors``.

documentation/features/automations.rst

@@ -44,7 +44,16 @@ Automating schedules
 A schedule automation's parameters form a schedule trigger message, as accepted by the `[POST] /assets/(id)/schedules/trigger <../api/v3_0.html#post--api-v3_0-assets-id-schedules-trigger>`_ API endpoint (without the asset id).
 Use the canonical API field names, including ``flex-model``, ``flex-context`` and ``force-new-job-creation``.
 The message is passed in a file, through ``--parameters``, and validated when the automation is created.
-No forecaster or data source is involved, so the forecaster options above do not apply to a schedule automation, and are refused when combined with ``--type scheduling``.
+The forecaster options above configure a forecaster, so they do not apply here, and are refused when combined with ``--type scheduling``.
+
+A schedule automation has a data generator too, but you do not name it separately.
+It is put together from choices you have already made: the flex config in the trigger message, the flex config saved on the asset tree, and the scheduler that the asset resolves to.
+Because those live in two places, and the asset can be edited without touching the automation, the runner puts the generator together again on every run, and moves the automation to another data source when the combination has changed.
+Editing an asset's flex-model is therefore a configuration change, and shows up as one: the schedules computed before and after it carry different data sources.
+
+Because the schedule is recomputed on every run, the flex config may only describe the site and its devices, not one moment.
+A field with a fixed moment in it, such as ``soc-at-start`` or a ``soc-targets`` entry with a ``datetime``, is refused when the automation is created, and the error names the field.
+Refer to a sensor instead, which says where to look rather than what was true once.
 
 Omit the ``start`` field to calculate it afresh from the server time on each run.
 It is floored to the fixed, positive ``resolution`` when given, or otherwise to the minute.

flexmeasures/api/v3_0/tests/test_automations_api_fresh_db.py

@@ -6,6 +6,7 @@
 from flask import url_for
 
 from flexmeasures.data.models.automations import Automation
+from flexmeasures.data.services.automations import resolve_schedule_generator
 from flexmeasures import Forecaster
 from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.services.data_sources import get_data_generator
@@ -105,12 +106,14 @@ def test_schedule_details_include_stored_flex_sensors(
         "site-power-capacity": "2 MVA",
         "consumption-price": {"sensor": price_sensor.id},
     }
+    parameters = {"duration": "PT1H"}
     automation = Automation(
         asset=asset,
         type="scheduling",
         name="Minimal schedule details",
         cronstr="0 6 * * *",
-        parameters={"duration": "PT1H"},
+        parameters=parameters,
+        generator_id=resolve_schedule_generator(asset.id, parameters).id,
     )
     fresh_db.session.add(automation)
     fresh_db.session.commit()

flexmeasures/cli/data_add.py

@@ -52,7 +52,11 @@
     populate_initial_structure,
     add_default_asset_types,
 )
-from flexmeasures.data.services.automations import prepare_schedule_trigger_message
+from flexmeasures.data.schemas.scheduling import find_momentary_flex_config_fields
+from flexmeasures.data.services.automations import (
+    prepare_schedule_trigger_message,
+    resolve_schedule_generator,
+)
 from flexmeasures.data.services.data_sources import (
     get_or_create_source,
     get_data_generator,
@@ -1674,6 +1678,29 @@ def add_forecast(  # noqa: C901
         raise
 
 
+def _check_schedule_automation_parameters(parameters: dict, asset) -> DataSource:
+    """Validate a schedule automation's trigger message, and return the data generator it will run with.
+
+    The message has to be a valid schedule trigger, and its flex config has to describe the site and its devices,
+    rather than one moment: the automation computes a fresh schedule on every run,
+    so a value tied to a fixed moment would be stale on the next one.
+    """
+    try:
+        message = prepare_schedule_trigger_message(parameters, asset.id)
+        AssetTriggerSchema().load(message)
+    except ValidationError as e:
+        click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR)
+        raise click.Abort()
+    momentary_fields = find_momentary_flex_config_fields(message)
+    if momentary_fields:
+        raise click.UsageError(
+            f"{flexmeasures_inflection.join_words_into_a_list(momentary_fields)} fixes a moment in time,"
+            " so it cannot configure a recurring schedule automation, which computes a fresh schedule on every run."
+            " Refer to a sensor instead of a fixed value, or leave the field out."
+        )
+    return resolve_schedule_generator(asset.id, parameters)
+
+
 @fm_add_data.command("automation")
 @with_appcontext
 @click.option(
@@ -1861,13 +1888,9 @@ def add_automation(
         db.session.flush()
         generator_id = generator.id
     else:  # scheduling
-        try:
-            AssetTriggerSchema().load(
-                prepare_schedule_trigger_message(parameters, asset.id)
-            )
-        except ValidationError as e:
-            click.secho(f"Invalid schedule parameters: {e.messages}", **MsgStyle.ERROR)
-            raise click.Abort()
+        # The scheduler and its configuration make up the automation's data generator,
+        # the same way a forecaster and its configuration do for a forecast automation.
+        generator_id = _check_schedule_automation_parameters(parameters, asset).id
         if "start" in parameters:
             click.secho(
                 "Warning: the schedule 'start' is fixed, so each run will compute the same period."

flexmeasures/cli/tests/test_automations.py

@@ -766,6 +766,70 @@ def test_add_automation_rejects_malformed_yaml_file(
     assert "Traceback" not in result.output
 
 
+def test_add_schedule_automation_rejects_momentary_flex_fields(
+    app, fresh_db, setup_dummy_data, tmp_path
+):
+    """A flex config field describing one moment cannot configure a recurring schedule automation.
+
+    Such a value is stale on the next run, and it would misdescribe the automation on its data source,
+    which records the configuration the scheduler computes under.
+    """
+    from flexmeasures.cli.data_add import add_automation
+
+    runner = app.test_cli_runner()
+    parameters_file = tmp_path / "parameters.yml"
+
+    # a state of charge that held at one moment
+    parameters_file.write_text(
+        'duration: "PT12H"\n'
+        "flex-model:\n"
+        "  - sensor: 1\n"
+        '    soc-at-start: "5 kWh"\n'
+    )
+    result = runner.invoke(
+        add_automation,
+        [
+            "--asset", "1",
+            "--name", "Momentary state of charge",
+            "--cron", "0 * * * *",
+            "--type", "scheduling",
+            "--parameters", str(parameters_file),
+        ],
+    )  # fmt: skip
+    assert result.exit_code == 2, result.output
+    assert "flex-model[0].soc-at-start fixes a moment in time" in result.output
+    assert "Traceback" not in result.output
+
+    # a target tied to a datetime
+    parameters_file.write_text(
+        'duration: "PT12H"\n'
+        "flex-model:\n"
+        "  - sensor: 1\n"
+        "    soc-targets:\n"
+        '      - datetime: "2026-01-15T10:00+01:00"\n'
+        '        value: "5 kWh"\n'
+    )
+    result = runner.invoke(
+        add_automation,
+        [
+            "--asset", "1",
+            "--name", "Momentary target",
+            "--cron", "0 * * * *",
+            "--type", "scheduling",
+            "--parameters", str(parameters_file),
+        ],
+    )  # fmt: skip
+    assert result.exit_code == 2, result.output
+    assert "flex-model[0].soc-targets[0] fixes a moment in time" in result.output
+
+    assert (
+        fresh_db.session.execute(
+            select(Automation).filter_by(name="Momentary state of charge")
+        ).scalar_one_or_none()
+        is None
+    )
+
+
 def test_add_schedule_automation(app, fresh_db, setup_dummy_data, tmp_path):
     """Create a schedules automation; parameters are validated as a schedule trigger message."""
     from flexmeasures.cli.data_add import add_automation
@@ -805,7 +869,13 @@ def test_add_schedule_automation(app, fresh_db, setup_dummy_data, tmp_path):
         select(Automation).filter_by(name="Half-day schedules")
     ).scalar_one()
     assert automation.type == "scheduling"
-    assert automation.generator_id is None
+    # The scheduler and the flex config it computes under are the automation's data generator.
+    assert automation.generator is not None
+    assert automation.generator.type == "scheduler"
+    assert (
+        automation.generator.attributes["data_generator"]["config"]["asset"]
+        == automation.asset_id
+    )
     assert automation.parameters == {"duration": "PT12H"}
 
     # a fixed start draws a warning
@@ -959,16 +1029,21 @@ def test_run_schedule_automation_dispatch(app, fresh_db, setup_dummy_data, monke
     """
     from flexmeasures.data.models.generic_assets import GenericAsset
     from flexmeasures.data.services import scheduling
-    from flexmeasures.data.services.automations import run_automation
+    from flexmeasures.data.services.automations import (
+        resolve_schedule_generator,
+        run_automation,
+    )
     from flexmeasures.utils.time_utils import server_now
 
     asset = fresh_db.session.get(GenericAsset, 1)
+    parameters = {"duration": "PT12H", "resolution": "PT15M"}
     automation = Automation(
         asset_id=asset.id,
         type="scheduling",
         name="Test schedules",
         cronstr="0 * * * *",
-        parameters={"duration": "PT12H", "resolution": "PT15M"},
+        parameters=parameters,
+        generator_id=resolve_schedule_generator(asset.id, parameters).id,
     )
     fresh_db.session.add(automation)
     fresh_db.session.flush()

flexmeasures/data/migrations/versions/3e91c47b0a58_merge_schedule_automations_with_main.py

@@ -1,18 +1,18 @@
 """merge the schedule automation migrations with main
 
-Two migrations branched off the same revision: those adding schedule automations,
+Two migrations branched off the same revision: the one adding an automation's timezone and cursor,
 and those reordering the timed belief primary key and adding the sensor data source association.
 They touch different tables, so this merge only rejoins them and has nothing of its own to do.
 
 Revision ID: 3e91c47b0a58
-Revises: c63896a97a8e, 84f268f5153c
+Revises: 9f2b6e1d4a73, 84f268f5153c
 Create Date: 2026-09-02 10:30:00.000000
 
 """
 
 # revision identifiers, used by Alembic.
 revision = "3e91c47b0a58"
-down_revision = ("c63896a97a8e", "84f268f5153c")
+down_revision = ("9f2b6e1d4a73", "84f268f5153c")
 branch_labels = None
 depends_on = None
 

flexmeasures/data/migrations/versions/a71d6f2c9b04_name_automation_types_after_the_task.py

@@ -2,7 +2,6 @@
 
 The rest of the codebase calls these tasks "forecasting" and "scheduling" (queue names, job types),
 so the automation types follow suit: 'forecasts' becomes 'forecasting' and 'schedules' becomes 'scheduling'.
-The check constraint requiring a data generator for forecast automations is recreated with the new value.
 
 Revision ID: a71d6f2c9b04
 Revises: 3e91c47b0a58
@@ -20,22 +19,10 @@
 
 
 def upgrade():
-    op.drop_constraint("forecast_generator", "automation", type_="check")
     op.execute("UPDATE automation SET type = 'forecasting' WHERE type = 'forecasts'")
     op.execute("UPDATE automation SET type = 'scheduling' WHERE type = 'schedules'")
-    op.create_check_constraint(
-        "forecast_generator",
-        "automation",
-        "type != 'forecasting' OR generator_id IS NOT NULL",
-    )
 
 
 def downgrade():
-    op.drop_constraint("forecast_generator", "automation", type_="check")
     op.execute("UPDATE automation SET type = 'forecasts' WHERE type = 'forecasting'")
     op.execute("UPDATE automation SET type = 'schedules' WHERE type = 'scheduling'")
-    op.create_check_constraint(
-        "forecast_generator",
-        "automation",
-        "type != 'forecasts' OR generator_id IS NOT NULL",
-    )