Carbon-Aware Batch Job Scheduler with Deadline-Safe Deferral

id: carbon-aware-batch-job-scheduler
namespace: company.sustainability
description: |
  Receive a deferrable batch job, compare Great Britain's regional carbon-intensity
  forecast with its deadline, and either run it now or schedule the executor subflow
  in a lower-carbon half-hour window.

inputs:
  - id: job_id
    type: STRING
    defaults: nightly-feature-build
    description: Unique business identifier for the batch job. Use a new value for
      each production request.

  - id: deadline
    type: DATETIME
    required: false
    description: Hard UTC completion deadline. When omitted, the planner uses 12
      hours after the execution starts.

  - id: estimated_runtime_minutes
    type: INT
    defaults: 30
    description: Expected workload duration. A selected start must leave this much
      time before the deadline.

  - id: estimated_energy_kwh
    type: FLOAT
    defaults: 2.5
    description: Estimated electricity used by one run, used only to estimate, not
      claim, avoided operational emissions.

  - id: records_to_process
    type: INT
    defaults: 100000
    description: Demo workload size passed to the companion executor. Replace the
      executor with your production batch flow.

  - id: region_id
    type: INT
    defaults: 13
    description: Carbon Intensity API region id. Region 13 is London; use the API's
      regional endpoint to find other ids.

  - id: run_now_below_gco2_kwh
    type: INT
    defaults: 100
    description: Run immediately when the current forecast is already at or below
      this gCO2/kWh threshold.

  - id: minimum_savings_percent
    type: FLOAT
    defaults: 10.0
    description: Minimum forecast reduction required before deferring a job.

  - id: minimum_deferral_minutes
    type: INT
    defaults: 30
    description: Ignore forecast slots too close to now to be operationally useful.

  - id: forecast_mode
    type: SELECT
    defaults: LIVE
    values:
      - LIVE
      - SAMPLE
      - UNAVAILABLE_DEMO
    description: LIVE uses the public forecast; SAMPLE deterministically
      demonstrates deferral; UNAVAILABLE_DEMO demonstrates fallback behavior.

  - id: fail_closed_on_forecast_error
    type: BOOL
    defaults: false
    description: Fail instead of running now when no usable forecast is available.

tasks:
  - id: fetch_carbon_forecast
    type: io.kestra.plugin.core.http.Request
    description: Fetch 48 hours of half-hour regional intensity forecasts. HTTP and
      transport failures are handed to the planner as missing data.
    uri: >-
      https://api.carbonintensity.org.uk/regional/intensity/{{ now() |
      date("yyyy-MM-dd'T'HH:mm'Z'", timeZone="UTC") }}/fw48h/regionid/{{
      inputs.region_id }}
    method: GET
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT20S
    allowFailure: true

  - id: build_execution_plan
    type: io.kestra.plugin.scripts.python.Script
    description: Validate the request, select the cleanest deadline-safe slot, and
      calculate forecast-based impact estimates.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    inputFiles:
      forecast.json: "{{ outputs.fetch_carbon_forecast.body ?? '{\"data\": []}' }}"
      request.json: |
        {
          "job_id": {{ inputs.job_id | toJson }},
          "deadline": {{ (inputs.deadline ?? '') | toJson }},
          "estimated_runtime_minutes": {{ inputs.estimated_runtime_minutes }},
          "estimated_energy_kwh": {{ inputs.estimated_energy_kwh }},
          "run_now_below_gco2_kwh": {{ inputs.run_now_below_gco2_kwh }},
          "minimum_savings_percent": {{ inputs.minimum_savings_percent }},
          "minimum_deferral_minutes": {{ inputs.minimum_deferral_minutes }},
          "forecast_mode": {{ inputs.forecast_mode | toJson }},
          "fail_closed_on_forecast_error": {{ inputs.fail_closed_on_forecast_error }}
        }
    script: |
      import json
      from datetime import datetime, timedelta, timezone

      def parse_datetime(value):
          if not value:
              return None
          parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
          return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)

      def iso(value):
          return value.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")

      with open("request.json", encoding="utf-8") as handle:
          request = json.load(handle)

      now = datetime.now(timezone.utc)
      deadline = parse_datetime(request.get("deadline")) or now + timedelta(hours=12)
      runtime = int(request["estimated_runtime_minutes"])
      min_deferral = int(request["minimum_deferral_minutes"])
      energy_kwh = float(request["estimated_energy_kwh"])

      if deadline <= now:
          raise ValueError("deadline must be in the future")
      if runtime <= 0 or min_deferral < 0 or energy_kwh < 0:
          raise ValueError("runtime must be positive; deferral and energy estimates cannot be negative")

      mode = request["forecast_mode"]
      forecast_error = None
      slots = []

      if mode == "SAMPLE":
          # Relative timestamps keep the deterministic demo valid whenever it is run.
          for minutes, intensity in [(0, 250), (60, 230), (120, 80), (180, 120)]:
              start = now + timedelta(minutes=minutes)
              slots.append({"start": start, "end": start + timedelta(minutes=30), "intensity": intensity})
      elif mode == "UNAVAILABLE_DEMO":
          forecast_error = "forecast deliberately suppressed by UNAVAILABLE_DEMO"
      else:
          try:
              with open("forecast.json", encoding="utf-8-sig") as handle:
                  payload = json.load(handle)
              periods = payload.get("data", [])
              if isinstance(periods, dict):
                  periods = periods.get("data", [])
              for period in periods:
                  forecast = (period.get("intensity") or {}).get("forecast")
                  start = parse_datetime(period.get("from"))
                  end = parse_datetime(period.get("to"))
                  if forecast is not None and start and end and end > now:
                      slots.append({"start": start, "end": end, "intensity": int(forecast)})
              if not slots:
                  forecast_error = "API response contained no usable future forecast periods"
          except Exception as exc:
              forecast_error = f"forecast response could not be parsed: {exc}"

      if not slots:
          if request["fail_closed_on_forecast_error"]:
              raise RuntimeError(forecast_error or "forecast unavailable")
          plan = {
              "job_id": request["job_id"],
              "decision": "RUN_NOW",
              "reason": "FORECAST_UNAVAILABLE_FALLBACK",
              "detail": forecast_error or "forecast unavailable",
              "schedule_at": iso(now),
              "deadline": iso(deadline),
              "current_intensity_gco2_kwh": -1,
              "selected_intensity_gco2_kwh": -1,
              "forecast_savings_percent": 0.0,
              "estimated_baseline_emissions_kg": -1.0,
              "estimated_selected_emissions_kg": -1.0,
              "estimated_avoided_emissions_kg": -1.0,
              "forecast_mode": mode,
              "forecast_periods_evaluated": 0,
          }
      else:
          slots.sort(key=lambda slot: slot["start"])
          current = next((slot for slot in slots if slot["start"] <= now < slot["end"]), slots[0])
          latest_start = deadline - timedelta(minutes=runtime)
          earliest_deferred_start = now + timedelta(minutes=min_deferral)
          candidates = [
              slot for slot in slots
              if slot["start"] >= earliest_deferred_start and slot["start"] <= latest_start
          ]
          best = min(candidates, key=lambda slot: (slot["intensity"], slot["start"])) if candidates else current
          current_intensity = current["intensity"]
          selected_intensity = best["intensity"]
          savings_percent = 0.0 if current_intensity <= 0 else max(0.0, (current_intensity - selected_intensity) / current_intensity * 100)

          if latest_start <= earliest_deferred_start:
              decision, reason, selected = "RUN_NOW", "DEADLINE_APPROACHING", current
          elif current_intensity <= int(request["run_now_below_gco2_kwh"]):
              decision, reason, selected = "RUN_NOW", "CURRENT_WINDOW_ALREADY_CLEAN", current
          elif candidates and selected_intensity < current_intensity and savings_percent >= float(request["minimum_savings_percent"]):
              decision, reason, selected = "DEFER", "LOWER_CARBON_WINDOW_FOUND", best
          else:
              decision, reason, selected = "RUN_NOW", "NO_MATERIAL_SAVINGS_BEFORE_DEADLINE", current

          selected_intensity = selected["intensity"]
          baseline_kg = energy_kwh * current_intensity / 1000
          selected_kg = energy_kwh * selected_intensity / 1000
          plan = {
              "job_id": request["job_id"],
              "decision": decision,
              "reason": reason,
              "detail": "All values are forecast estimates; actual savings require measured workload energy.",
              "schedule_at": iso(selected["start"] if decision == "DEFER" else now),
              "deadline": iso(deadline),
              "current_intensity_gco2_kwh": current_intensity,
              "selected_intensity_gco2_kwh": selected_intensity,
              "forecast_savings_percent": round(0.0 if current_intensity <= 0 else max(0.0, (current_intensity - selected_intensity) / current_intensity * 100), 2),
              "estimated_baseline_emissions_kg": round(baseline_kg, 4),
              "estimated_selected_emissions_kg": round(selected_kg, 4),
              "estimated_avoided_emissions_kg": round(max(0.0, baseline_kg - selected_kg), 4),
              "forecast_mode": mode,
              "forecast_periods_evaluated": len(slots),
          }

      print(json.dumps(plan, indent=2))
      print("::" + json.dumps({"outputs": plan}) + "::")

  - id: record_plan
    type: io.kestra.plugin.core.kv.Set
    description: Persist the complete scheduling decision for dashboards, audits,
      and later comparison with measured energy data.
    key: "carbon_plan_{{ execution.id }}"
    kvType: JSON
    value: "{{ outputs.build_execution_plan.vars | toJson }}"

  - id: route_execution
    type: io.kestra.plugin.core.flow.If
    description: Schedule the executor in the selected future window, or start it
      immediately when waiting is not worthwhile or safe.
    condition: "{{ outputs.build_execution_plan.vars.decision == 'DEFER' }}"
    then:
      - id: schedule_for_cleaner_window
        type: io.kestra.plugin.core.flow.Subflow
        description: Create a future-dated execution without keeping this planner
          execution running while it waits.
        namespace: company.sustainability
        flowId: carbon-aware-batch-job-executor
        wait: false
        transmitFailed: false
        scheduleDate: "{{ outputs.build_execution_plan.vars.schedule_at }}"
        labels:
          carbon-aware-decision: deferred
          planner-execution: "{{ execution.id }}"
        inputs:
          job_id: "{{ inputs.job_id }}"
          records_to_process: "{{ inputs.records_to_process }}"
          estimated_energy_kwh: "{{ inputs.estimated_energy_kwh }}"
          decision: "{{ outputs.build_execution_plan.vars.decision }}"
          decision_reason: "{{ outputs.build_execution_plan.vars.reason }}"
          planned_start: "{{ outputs.build_execution_plan.vars.schedule_at }}"
          deadline: "{{ outputs.build_execution_plan.vars.deadline }}"
          baseline_intensity_gco2_kwh: "{{ outputs.build_execution_plan.vars.current_intensity_gco2_kwh }}"
          planned_intensity_gco2_kwh: "{{ outputs.build_execution_plan.vars.selected_intensity_gco2_kwh }}"
          estimated_avoided_emissions_kg: "{{
            outputs.build_execution_plan.vars.estimated_avoided_emissions_kg }}"
          planner_execution_id: "{{ execution.id }}"
    else:
      - id: execute_now
        type: io.kestra.plugin.core.flow.Subflow
        description: Start the companion workload immediately and wait so failures
          propagate to the caller.
        namespace: company.sustainability
        flowId: carbon-aware-batch-job-executor
        wait: true
        transmitFailed: true
        labels:
          carbon-aware-decision: run-now
          planner-execution: "{{ execution.id }}"
        inputs:
          job_id: "{{ inputs.job_id }}"
          records_to_process: "{{ inputs.records_to_process }}"
          estimated_energy_kwh: "{{ inputs.estimated_energy_kwh }}"
          decision: "{{ outputs.build_execution_plan.vars.decision }}"
          decision_reason: "{{ outputs.build_execution_plan.vars.reason }}"
          planned_start: "{{ outputs.build_execution_plan.vars.schedule_at }}"
          deadline: "{{ outputs.build_execution_plan.vars.deadline }}"
          baseline_intensity_gco2_kwh: "{{ outputs.build_execution_plan.vars.current_intensity_gco2_kwh }}"
          planned_intensity_gco2_kwh: "{{ outputs.build_execution_plan.vars.selected_intensity_gco2_kwh }}"
          estimated_avoided_emissions_kg: "{{
            outputs.build_execution_plan.vars.estimated_avoided_emissions_kg }}"
          planner_execution_id: "{{ execution.id }}"

  - id: report_decision
    type: io.kestra.plugin.core.log.Log
    message: >-
      Job {{ inputs.job_id }}: {{ outputs.build_execution_plan.vars.decision }}
      at {{ outputs.build_execution_plan.vars.schedule_at }} because {{
      outputs.build_execution_plan.vars.reason }}. Forecast {{
      outputs.build_execution_plan.vars.current_intensity_gco2_kwh }} -> {{
      outputs.build_execution_plan.vars.selected_intensity_gco2_kwh }} gCO2/kWh;
      estimated avoided emissions {{
      outputs.build_execution_plan.vars.estimated_avoided_emissions_kg }}
      kgCO2e.

outputs:
  - id: decision
    type: STRING
    description: RUN_NOW or DEFER.
    value: "{{ outputs.build_execution_plan.vars.decision }}"

  - id: schedule_at
    type: STRING
    description: UTC start selected by the planner.
    value: "{{ outputs.build_execution_plan.vars.schedule_at }}"

  - id: plan
    type: JSON
    description: Full forecast, deadline, reasoning, and estimated-impact record.
    value: "{{ outputs.build_execution_plan.vars | toJson }}"

triggers:
  - id: submit_batch_job
    type: io.kestra.plugin.core.trigger.Webhook
    description: Accept a batch job from CI, an ML platform, or a data portal. Keep
      the key secret because callers can create executions.
    key: "{{ secret('CARBON_SCHEDULER_WEBHOOK_KEY') }}"
    inputs:
      job_id: "{{ (trigger.body ?? {}).job_id ?? 'webhook-batch-job' }}"
      deadline: "{{ (trigger.body ?? {}).deadline ?? (now() | dateAdd(12, 'HOURS')) }}"
      estimated_runtime_minutes: "{{ (trigger.body ?? {}).estimated_runtime_minutes ?? 30 }}"
      estimated_energy_kwh: "{{ (trigger.body ?? {}).estimated_energy_kwh ?? 2.5 }}"
      records_to_process: "{{ (trigger.body ?? {}).records_to_process ?? 100000 }}"
      region_id: "{{ (trigger.body ?? {}).region_id ?? 13 }}"
      run_now_below_gco2_kwh: "{{ (trigger.body ?? {}).run_now_below_gco2_kwh ?? 100 }}"
      minimum_savings_percent: "{{ (trigger.body ?? {}).minimum_savings_percent ?? 10.0 }}"
      minimum_deferral_minutes: "{{ (trigger.body ?? {}).minimum_deferral_minutes ?? 30 }}"
      forecast_mode: "{{ (trigger.body ?? {}).forecast_mode ?? 'LIVE' }}"
      fail_closed_on_forecast_error: "{{ (trigger.body ?? {}).fail_closed_on_forecast_error ?? false }}"