Carbon-Aware Batch Job Executor Companion Flow

id: carbon-aware-batch-job-executor
namespace: company.sustainability
description: |
  Companion subflow for the carbon-aware scheduler. It runs a deterministic
  CPU batch workload, captures an artifact, and records the planned carbon context.

inputs:
  - id: job_id
    type: STRING
    defaults: manual-demo-job

  - id: records_to_process
    type: INT
    defaults: 100000

  - id: estimated_energy_kwh
    type: FLOAT
    defaults: 2.5

  - id: decision
    type: STRING
    defaults: RUN_NOW

  - id: decision_reason
    type: STRING
    defaults: MANUAL_EXECUTION

  - id: planned_start
    type: STRING
    defaults: not-scheduled

  - id: deadline
    type: STRING
    defaults: not-specified

  - id: baseline_intensity_gco2_kwh
    type: INT
    defaults: -1

  - id: planned_intensity_gco2_kwh
    type: INT
    defaults: -1

  - id: estimated_avoided_emissions_kg
    type: FLOAT
    defaults: -1.0

  - id: planner_execution_id
    type: STRING
    defaults: manual

tasks:
  - id: run_batch_workload
    type: io.kestra.plugin.scripts.python.Script
    description: Demonstrate replaceable batch work by aggregating deterministic
      records and writing a verifiable result artifact.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    outputFiles:
      - job-result.json
    env:
      JOB_ID: "{{ inputs.job_id }}"
      RECORDS_TO_PROCESS: "{{ inputs.records_to_process }}"
      PLANNER_EXECUTION_ID: "{{ inputs.planner_execution_id }}"
    script: |
      import hashlib
      import json
      import os
      import time
      from datetime import datetime, timezone

      job_id = os.environ["JOB_ID"]
      records = int(os.environ["RECORDS_TO_PROCESS"])
      if records <= 0 or records > 5_000_000:
          raise ValueError("records_to_process must be between 1 and 5,000,000")

      started = time.perf_counter()
      aggregate = 0
      digest = hashlib.sha256()
      for record_id in range(records):
          value = (record_id * 2654435761) % 1_000_003
          aggregate = (aggregate + value) % 1_000_000_007
          digest.update(f"{record_id}:{value}\n".encode())

      result = {
          "job_id": job_id,
          "records_processed": records,
          "aggregate": aggregate,
          "sha256": digest.hexdigest(),
          "duration_seconds": round(time.perf_counter() - started, 4),
          "completed_at": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
          "planner_execution_id": os.environ["PLANNER_EXECUTION_ID"],
      }
      with open("job-result.json", "w", encoding="utf-8") as handle:
          json.dump(result, handle, indent=2)
      print(json.dumps(result, indent=2))
      print("::" + json.dumps({"outputs": result}) + "::")

  - id: record_job_result
    type: io.kestra.plugin.core.kv.Set
    description: Store the workload result beside its planning assumptions for audit
      and later reconciliation with measured energy.
    key: "carbon_job_result_{{ execution.id }}"
    kvType: JSON
    value: |
      {
        "job_id": {{ inputs.job_id | toJson }},
        "state": "COMPLETED",
        "decision": {{ inputs.decision | toJson }},
        "decision_reason": {{ inputs.decision_reason | toJson }},
        "planner_execution_id": {{ inputs.planner_execution_id | toJson }},
        "executor_execution_id": "{{ execution.id }}",
        "planned_start": {{ inputs.planned_start | toJson }},
        "actual_completion": {{ outputs.run_batch_workload.vars.completed_at | toJson }},
        "deadline": {{ inputs.deadline | toJson }},
        "records_processed": {{ outputs.run_batch_workload.vars.records_processed }},
        "result_sha256": {{ outputs.run_batch_workload.vars.sha256 | toJson }},
        "result_uri": {{ outputs.run_batch_workload.outputFiles['job-result.json'] | toJson }},
        "estimated_energy_kwh": {{ inputs.estimated_energy_kwh }},
        "baseline_intensity_gco2_kwh": {{ inputs.baseline_intensity_gco2_kwh }},
        "planned_intensity_gco2_kwh": {{ inputs.planned_intensity_gco2_kwh }},
        "estimated_avoided_emissions_kg": {{ inputs.estimated_avoided_emissions_kg }},
        "impact_warning": "Forecast estimate only; no workload energy was measured."
      }

  - id: summarize_completion
    type: io.kestra.plugin.core.log.Log
    message: >-
      Completed carbon-aware job {{ inputs.job_id }}: {{
      outputs.run_batch_workload.vars.records_processed }} records in {{
      outputs.run_batch_workload.vars.duration_seconds }}s, result {{
      outputs.run_batch_workload.outputFiles['job-result.json'] }}. Planning
      reason: {{ inputs.decision_reason }}.

outputs:
  - id: result_file
    type: FILE
    description: JSON artifact containing record count, checksum, aggregate, and
      completion timestamp.
    value: "{{ outputs.run_batch_workload.outputFiles['job-result.json'] }}"

  - id: records_processed
    type: INT
    value: "{{ outputs.run_batch_workload.vars.records_processed }}"

  - id: result_sha256
    type: STRING
    value: "{{ outputs.run_batch_workload.vars.sha256 }}"

errors:
  - id: record_executor_failure
    type: io.kestra.plugin.core.kv.Set
    description: Preserve a durable failure marker even when the workload itself
      cannot produce an artifact.
    key: "carbon_job_result_{{ execution.id }}"
    kvType: JSON
    value: |
      {
        "job_id": {{ inputs.job_id | toJson }},
        "state": "FAILED",
        "planner_execution_id": {{ inputs.planner_execution_id | toJson }},
        "executor_execution_id": "{{ execution.id }}",
        "decision": {{ inputs.decision | toJson }},
        "decision_reason": {{ inputs.decision_reason | toJson }},
        "error": {{ error.message | toJson }}
      }