Trigger a single automation run on demand, via CLI, API and UI

documentation/api/change_log.rst

@@ -5,6 +5,10 @@ API change log
 
 .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL (e.g. `/api/v3_0`), allowing developers to upgrade at their own pace.
 
+v3.0-34 | September 2, 2026
+"""""""""""""""""""""""""""
+- Added ``POST /api/v3_0/assets/<id>/automations/<automation_id>/trigger``, to run one automation now, once, on top of its recurring runs. The response is the standard job response, extended with ``n_jobs``: how many jobs the run queued. An on-demand run does not affect the automation's recurrence, and inactive automations can be triggered, too. Triggering requires the same permission as writing data under the asset, and falls under the stricter rate limit that the other triggering endpoints share.
+
 v3.0-33 | September 1, 2026
 """""""""""""""""""""""""""
 - Added ``GET /api/v3_0/assets/<id>/automations`` and ``GET /api/v3_0/assets/<id>/automations/<automation_id>`` for listing and inspecting forecast automations, including the sensors an automation reads from and writes to. Each automation shows the IANA ``timezone`` in which its cron expression is interpreted, and a ``cursor``: the offset-aware UTC time of the most recent run it committed to. The cursor advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. Asset job entries now include ``created_via`` provenance; automation identity is included only when the caller may read that automation.

documentation/changelog.rst

@@ -17,6 +17,7 @@ v1.1.0 | September XX, 2026
 New features
 -------------
 
+* A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets/<id>/automations/<automation_id>/trigger``) and the asset's *Automations* page (a *Run now* button), which is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived [see `PR #2460 <https://www.github.com/FlexMeasures/flexmeasures/pull/2460>`_]
 * 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>`_]
 
 Infrastructure / Support

documentation/cli/change_log.rst

@@ -14,6 +14,7 @@ since v1.0.0 | August 11, 2026
 * 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; for now, computing forecasts). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted.
 * 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.
+* Add ``flexmeasures jobs run-automation --automation <id>`` to queue the jobs for a single run of one automation, now, on top of its recurring runs. This leaves the automation's cursor alone, so its next recurring run still happens as scheduled, and inactive automations can be run this way, too.
 * ``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

@@ -56,6 +56,24 @@ If the process crashes, or queueing fails after creating some jobs, that run is
 
 The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed.
 
+Running one automation on demand
+--------------------------------
+
+Besides its recurring runs, a single automation can be run now, once.
+This is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived.
+
+.. code-block:: bash
+
+    flexmeasures jobs run-automation --automation 4
+
+The same is available in the API, as `[POST] /assets/(id)/automations/(automation_id)/trigger <../api/v3_0.html#post--api-v3_0-assets-id-automations-automation_id-trigger>`_, and in the UI, as the *Run now* button on the asset's *Automations* page.
+
+The automation runs with the parameters it was created with, and the jobs it queues are recorded as its jobs, just like the jobs of a recurring run.
+An on-demand run does not affect the automation's recurrence: its cursor (see :ref:`automation_cursor`) stays where it was, so the next recurring run still happens as scheduled, and a run missed while the runner was down is still caught up.
+Inactive automations can be run this way, too, which is how you can try one out before activating it.
+
+Unlike a recurring run, an on-demand run is not protected against being started twice: asking for two runs in a row queues two runs.
+
 Viewing automations
 -------------------
 

flexmeasures/api/v3_0/assets.py

@@ -54,6 +54,7 @@
     describe_cronstr,
     get_automation_job_stats,
     resolve_automation_sensors,
+    run_automation,
 )
 from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
 from flexmeasures.data.queries.generic_assets import (
@@ -1575,6 +1576,114 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset):
         automation_data["redis_connection_err"] = redis_connection_err
         return automation_data, 200
 
+    @route("/<id>/automations/<int:automation_id>/trigger", methods=["POST"])
+    @limit_triggers()
+    @use_kwargs(
+        {
+            "asset": AssetIdField(data_key="id"),
+            "automation_id": fields.Int(),
+        },
+        location="path",
+    )
+    # Running an automation writes data under the asset, which is what create-children means here.
+    # The sensors it writes to were checked against the same permission when the automation was created,
+    # and its output scope is checked again on each run (see validate_forecast_output_scope).
+    @permission_required_for_context("create-children", ctx_arg_name="asset")
+    @as_json
+    def trigger_automation(self, id: int, automation_id: int, asset: GenericAsset):
+        """
+        .. :quickref: Assets; Trigger a single run of an automation.
+
+        ---
+        post:
+          summary: Trigger a single run of an automation.
+          description: |
+            Run one automation now, once, in addition to its recurring runs.
+            This is useful to try out a new automation, to re-run one after fixing what made it fail,
+            or to refresh its results after late input data arrived.
+
+            The automation runs with the parameters it was created with,
+            and the jobs it queues are recorded as its jobs, just like the jobs of a recurring run.
+
+            An on-demand run does not affect the automation's recurrence: its cursor stays where it was,
+            so the next recurring run still happens as scheduled, and a missed run is still caught up.
+            Inactive automations can be triggered, too, which is how you can try one out before activating it.
+          security:
+            - ApiKeyAuth: []
+          parameters:
+            - in: path
+              name: id
+              required: true
+              description: ID of the asset the automation is defined on.
+              schema:
+                type: integer
+            - in: path
+              name: automation_id
+              required: true
+              description: ID of the automation to run.
+              schema:
+                type: integer
+          responses:
+            202:
+              description: PROCESSING
+              content:
+                application/json:
+                  examples:
+                    triggered:
+                      summary: Automation run accepted
+                      description: |
+                        The automation queued its jobs, which will be picked up by a worker.
+                        The `job` field holds the Universally Unique Identifier (UUID) of the job to follow,
+                        and `n_jobs` says how many jobs the run queued in total.
+                      value:
+                        status: ACCEPTED
+                        job: "364bfd06-c1fa-430b-8d25-8f5a547651fb"
+                        job-url: "/api/v3_0/jobs/364bfd06-c1fa-430b-8d25-8f5a547651fb"
+                        n_jobs: 2
+                        message: "Request has been accepted for processing."
+            401:
+              description: UNAUTHORIZED
+            403:
+              description: INVALID_SENDER
+            404:
+              description: NOT_FOUND
+            422:
+              description: UNPROCESSABLE_ENTITY
+          tags:
+            - Assets
+        """
+        automation = db.session.get(Automation, automation_id)
+        if automation is None or automation.asset_id != asset.id:
+            return {
+                "message": f"Asset {asset.id} has no automation with id {automation_id}."
+            }, 404
+        try:
+            returns = run_automation(automation)
+        except (NotImplementedError, ValueError, ValidationError) as e:
+            db.session.rollback()
+            return unprocessable_entity(
+                e.messages if isinstance(e, ValidationError) else str(e)
+            )
+        job_id = (returns or {}).get("job_id")
+        if job_id is None:
+            db.session.rollback()
+            current_app.logger.error(
+                "Automation %s ran on demand, but reported no job: %r",
+                automation.id,
+                returns,
+            )
+            return unprocessable_entity(
+                f"Automation {automation.id} did not queue any job."
+            )
+        AssetAuditLog.add_record(
+            asset,
+            f"Triggered a run of automation '{automation.name}' ({automation.id}).",
+        )
+        db.session.commit()
+        response, status_code = request_accepted_for_processing(job_id)
+        response["n_jobs"] = returns.get("n_jobs")
+        return response, status_code
+
     @route("/<id>/jobs", methods=["GET"])
     @use_kwargs(
         {"asset": AssetIdField(data_key="id")},

flexmeasures/api/v3_0/tests/test_automations_api.py

@@ -172,3 +172,141 @@ def test_get_nonexistent_automation(
             url_for("AssetAPI:get_automation", id=battery.id, automation_id=9999),
         )
     assert response.status_code == 404
+
+
+@pytest.mark.parametrize(
+    "requesting_user, expected_status_code",
+    [
+        (None, 401),  # not logged in
+        ("test_prosumer_user@seita.nl", 202),  # same account
+        ("test_dummy_user_3@seita.nl", 403),  # different account
+    ],
+    indirect=["requesting_user"],
+)
+def test_trigger_automation_auth(
+    app,
+    add_battery_assets,
+    add_automations,
+    requesting_user,
+    expected_status_code,
+    mocker,
+):
+    battery = add_battery_assets["Test battery"]
+    automation = add_automations[0]
+    run_automation = mocker.patch(
+        "flexmeasures.api.v3_0.assets.run_automation",
+        return_value={"job_id": "364bfd06-c1fa-430b-8d25-8f5a547651fb", "n_jobs": 2},
+    )
+    with app.test_client() as client:
+        response = client.post(
+            url_for(
+                "AssetAPI:trigger_automation",
+                id=battery.id,
+                automation_id=automation.id,
+            ),
+        )
+    assert response.status_code == expected_status_code
+    if expected_status_code == 202:
+        assert run_automation.call_args.args[0].id == automation.id
+    else:
+        run_automation.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True
+)
+def test_trigger_automation(
+    app,
+    db,
+    add_battery_assets,
+    add_automations,
+    requesting_user,
+    mocker,
+):
+    """Triggering a run reports the queued job, and leaves the automation's recurrence alone."""
+    battery = add_battery_assets["Test battery"]
+    automation = add_automations[1]  # inactive automations can be triggered, too.
+    cursor_before = automation.cursor
+    mocker.patch(
+        "flexmeasures.api.v3_0.assets.run_automation",
+        return_value={"job_id": "364bfd06-c1fa-430b-8d25-8f5a547651fb", "n_jobs": 2},
+    )
+    with app.test_client() as client:
+        response = client.post(
+            url_for(
+                "AssetAPI:trigger_automation",
+                id=battery.id,
+                automation_id=automation.id,
+            ),
+        )
+    assert response.status_code == 202
+    assert response.json["status"] == "ACCEPTED"
+    assert response.json["job"] == "364bfd06-c1fa-430b-8d25-8f5a547651fb"
+    assert response.json["n_jobs"] == 2
+    db.session.expire_all()
+    assert automation.cursor == cursor_before
+    assert automation.active is False
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True
+)
+def test_trigger_automation_that_cannot_run(
+    app,
+    add_battery_assets,
+    add_automations,
+    requesting_user,
+    mocker,
+):
+    """A run which cannot be set up is reported as such, rather than as a queued job."""
+    battery = add_battery_assets["Test battery"]
+    automation = add_automations[0]
+    mocker.patch(
+        "flexmeasures.api.v3_0.assets.run_automation",
+        side_effect=ValueError(
+            "Forecast automation output sensor 3 must belong to asset 1 or one of its descendants."
+        ),
+    )
+    with app.test_client() as client:
+        response = client.post(
+            url_for(
+                "AssetAPI:trigger_automation",
+                id=battery.id,
+                automation_id=automation.id,
+            ),
+        )
+    assert response.status_code == 422
+    assert "must belong to asset" in str(response.json["message"])
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True
+)
+@pytest.mark.parametrize("via_other_asset", [True, False])
+def test_trigger_unknown_automation(
+    app,
+    add_battery_assets,
+    add_automations,
+    requesting_user,
+    via_other_asset,
+    mocker,
+):
+    """Triggering an automation the asset does not have returns 404, without running anything."""
+    run_automation = mocker.patch("flexmeasures.api.v3_0.assets.run_automation")
+    if via_other_asset:
+        # an existing automation, requested through an asset it does not belong to.
+        asset = add_battery_assets["Test small battery"]
+        automation_id = add_automations[0].id
+    else:
+        asset = add_battery_assets["Test battery"]
+        automation_id = 9999
+    with app.test_client() as client:
+        response = client.post(
+            url_for(
+                "AssetAPI:trigger_automation",
+                id=asset.id,
+                automation_id=automation_id,
+            ),
+        )
+    assert response.status_code == 404
+    run_automation.assert_not_called()

flexmeasures/cli/jobs.py

@@ -26,12 +26,16 @@
     ScheduledJobRegistry,
     StartedJobRegistry,
 )
+from marshmallow import ValidationError
 from sqlalchemy.orm import configure_mappers
 from tabulate import tabulate
 import pandas as pd
 
 from flexmeasures.data import db
+from flexmeasures.data.models.audit_log import AssetAuditLog
+from flexmeasures.data.models.automations import Automation
 from flexmeasures.data.schemas import AssetIdField, SensorIdField
+from flexmeasures.data.schemas.automations import AutomationIdField
 from flexmeasures.data.services.automations import (
     claim_due_automation,
     floor_to_minute,
@@ -127,6 +131,59 @@ def run_automations():
         raise click.exceptions.Exit(1)
 
 
+@fm_jobs.command("run-automation")
+@with_appcontext
+@click.option(
+    "--automation",
+    "automation",
+    type=AutomationIdField(),
+    required=True,
+    help="ID of the automation to run.",
+)
+def run_one_automation(automation: Automation):
+    """
+    Queue the jobs for a single run of one automation, now.
+
+    \b
+        flexmeasures jobs run-automation --automation 4
+
+    Use this to try out a new automation, to re-run one after fixing what made it fail,
+    or to refresh its results after late input data arrived.
+    The automation runs with the parameters it was created with,
+    and the jobs it queues are recorded as its jobs, just like the jobs of a recurring run.
+
+    This does not affect the automation's recurrence: its cursor stays where it was,
+    so the next recurring run still happens as scheduled (see `flexmeasures jobs run-automations`).
+    Inactive automations can be run this way, too, which is how you can try one out before activating it.
+    """
+    try:
+        returns = run_automation(automation)
+    except (NotImplementedError, ValueError, ValidationError) as e:
+        db.session.rollback()
+        click.secho(
+            f"Automation {automation.id} ('{automation.name}') failed to queue jobs: {e}",
+            **MsgStyle.ERROR,
+        )
+        raise click.Abort()
+    if not returns or returns.get("job_id") is None:
+        db.session.rollback()
+        click.secho(
+            f"Automation {automation.id} ('{automation.name}') did not queue any job.",
+            **MsgStyle.ERROR,
+        )
+        raise click.Abort()
+    n_jobs = returns.get("n_jobs")
+    AssetAuditLog.add_record(
+        automation.asset,
+        f"Triggered a run of automation '{automation.name}' ({automation.id}) via CLI.",
+    )
+    db.session.commit()
+    click.secho(
+        f"Automation {automation.id} ('{automation.name}') queued {n_jobs} forecasting job(s) for asset {automation.asset_id}.",
+        **MsgStyle.SUCCESS,
+    )
+
+
 @fm_jobs.command("stats")
 @with_appcontext
 @click.option(