Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

id: change-risk-scoring-cab-agenda
namespace: company.team

description: |
  Change risk scoring for the change advisory board (ITIL 4 change enablement): score every
  change request on the criticality and blast radius of what it touches, its size, the backout
  plan and test evidence, the team's recent change failure rate and whether it lands in a
  freeze window or collides with another change. Low-risk changes are pre-approved, medium
  ones go to a peer review, high ones to the CAB agenda. Outcomes recorded after
  implementation feed the team's failure rate, kept in KV.

inputs:
  - id: cab_date
    type: STRING
    displayName: CAB date
    defaults: "2026-10-06"
    validator: ^\d{4}-\d{2}-\d{2}$

  - id: outcomes
    type: JSON
    displayName: Outcomes of implemented changes
    description: >-
      Result of changes implemented since the last CAB, keyed by change ID,
      SUCCESS or FAILED, for example {"CHG-1001": "FAILED"}. A failed change is
      one that caused an incident or a rollback.
    defaults: "{}"

  - id: scenario
    type: SELECT
    displayName: Demo scenario
    description: ISSUES adds a change for an unknown service and a change whose
      window ends before it starts, which must stop the scoring.
    values:
      - CLEAN
      - ISSUES
    defaults: CLEAN

variables:
  state_key: change_outcomes

concurrency:
  limit: 1

tasks:
  - id: state
    type: io.kestra.plugin.core.output.OutputValues
    values:
      s: "{{ (kv(vars.state_key, errorOnMissing=false) ?? {'outcomes': {}}) | toJson
        }}"

  - id: score
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: >
      One DuckDB session: service catalog with criticality and dependencies,
      freeze calendar, past change outcomes and the change requests (demo
      generator or your ITSM and CMDB), then the risk factors, the score, the
      route and the collisions.
    outputFiles:
      - changes
      - agenda
      - blockers
    fetchType: FETCH_ONE
    sql: |
      SET VARIABLE cab = DATE '{{ inputs.cab_date }}';
      -- The demo change history is fixed, so every CAB sees the same requests.
      SET VARIABLE anchor = DATE '2026-10-06';
      CREATE TABLE st AS SELECT '{{ outputs.state.values.s | replace({"'": "''"}) }}'::JSON AS j;

      -- =======================================================================
      -- Service catalog, dependencies and freeze calendar. Replace with your CMDB.
      -- =======================================================================
      CREATE TABLE services (service VARCHAR, owner_team VARCHAR, criticality INT);  -- 1 = business critical ... 4 = internal tool
      INSERT INTO services VALUES ('payments-api', 'Payments', 1), ('checkout-web', 'Web', 1), ('auth', 'Identity', 1), ('search', 'Discovery', 2),
        ('notifications', 'Platform', 2), ('reporting', 'Data', 3), ('wiki', 'IT', 4), ('core-db', 'Platform', 1);
      CREATE TABLE depends_on (service VARCHAR, dependency VARCHAR);
      INSERT INTO depends_on VALUES ('checkout-web', 'payments-api'), ('checkout-web', 'auth'), ('payments-api', 'core-db'), ('auth', 'core-db'),
        ('search', 'core-db'), ('reporting', 'core-db'), ('notifications', 'auth'), ('checkout-web', 'search');
      CREATE TABLE freezes (name VARCHAR, starts_at TIMESTAMP, ends_at TIMESTAMP, min_criticality INT);
      INSERT INTO freezes VALUES ('Quarter close', getvariable('anchor') + INTERVAL 20 DAY, getvariable('anchor') + INTERVAL 24 DAY, 2),
                                 ('Marketing launch', getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 18 HOUR, getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 21 HOUR, 1);

      -- Changes implemented over the last 90 days, with their outcome, per team.
      CREATE TABLE history AS
      SELECT 'H-' || i AS change_id, ['Payments', 'Web', 'Identity', 'Discovery', 'Platform', 'Data', 'IT'][1 + i % 7] AS team,
             getvariable('anchor') - (i % 90)::INT AS implemented_on,
             CASE WHEN i % 7 = 4 AND i % 3 = 0 THEN 'FAILED' WHEN i % 7 = 1 AND i % 9 = 0 THEN 'FAILED' WHEN i % 17 = 0 THEN 'FAILED' ELSE 'SUCCESS' END AS outcome
      FROM range(1, 211) t(i);

      -- =======================================================================
      -- Change requests for this CAB. Replace with your ITSM export.
      -- =======================================================================
      CREATE TABLE changes (change_id VARCHAR, title VARCHAR, service VARCHAR, team VARCHAR, kind VARCHAR, window_start TIMESTAMP, window_end TIMESTAMP,
                            files_changed INT, has_backout BOOLEAN, tested_in_staging BOOLEAN, needs_downtime BOOLEAN);
      INSERT INTO changes VALUES
        ('CHG-2001', 'Rotate TLS certificates', 'wiki', 'IT', 'STANDARD', getvariable('anchor') + INTERVAL 1 DAY + INTERVAL 10 HOUR, getvariable('anchor') + INTERVAL 1 DAY + INTERVAL 11 HOUR, 2, true, true, false),
        ('CHG-2002', 'Upgrade core-db to 16.4', 'core-db', 'Platform', 'NORMAL', getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 22 HOUR, getvariable('anchor') + INTERVAL 4 DAY + INTERVAL 2 HOUR, 14, true, true, true),
        ('CHG-2003', 'New payment provider routing', 'payments-api', 'Payments', 'NORMAL', getvariable('anchor') + INTERVAL 2 DAY + INTERVAL 14 HOUR, getvariable('anchor') + INTERVAL 2 DAY + INTERVAL 16 HOUR, 63, false, true, false),
        ('CHG-2004', 'Search ranking tweak', 'search', 'Discovery', 'NORMAL', getvariable('anchor') + INTERVAL 2 DAY + INTERVAL 10 HOUR, getvariable('anchor') + INTERVAL 2 DAY + INTERVAL 11 HOUR, 4, true, true, false),
        ('CHG-2005', 'Checkout copy change', 'checkout-web', 'Web', 'NORMAL', getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 19 HOUR, getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 20 HOUR, 3, true, true, false),
        ('CHG-2006', 'Report export to Parquet', 'reporting', 'Data', 'NORMAL', getvariable('anchor') + INTERVAL 1 DAY + INTERVAL 15 HOUR, getvariable('anchor') + INTERVAL 1 DAY + INTERVAL 16 HOUR, 22, true, false, false),
        ('CHG-2007', 'MFA policy change', 'auth', 'Identity', 'NORMAL', getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 23 HOUR, getvariable('anchor') + INTERVAL 3 DAY + INTERVAL 23 HOUR + INTERVAL 45 MINUTE, 9, true, true, false),
        ('CHG-2008', 'Hotfix: notification retries', 'notifications', 'Platform', 'EMERGENCY', getvariable('anchor') + INTERVAL 0 DAY + INTERVAL 18 HOUR, getvariable('anchor') + INTERVAL 0 DAY + INTERVAL 19 HOUR, 5, true, false, false);
      {% if inputs.scenario == 'ISSUES' %}
      INSERT INTO changes VALUES ('CHG-2090', 'Unknown service tweak', 'legacy-erp', 'IT', 'NORMAL', getvariable('anchor') + INTERVAL 1 DAY, getvariable('anchor') + INTERVAL 1 DAY + INTERVAL 1 HOUR, 1, true, true, false),
                                 ('CHG-2091', 'Window typo', 'wiki', 'IT', 'STANDARD', getvariable('anchor') + INTERVAL 2 DAY, getvariable('anchor') + INTERVAL 1 DAY, 1, true, true, false);
      {% endif %}

      -- Outcomes recorded at this CAB, plus those kept from earlier CABs.
      CREATE TABLE recorded AS
      SELECT key AS change_id, value ->> '$' AS outcome FROM json_each((SELECT j->'outcomes' FROM st))
      UNION ALL SELECT key, value ->> '$' FROM json_each('{{ inputs.outcomes | toJson }}');
      CREATE TABLE past_changes (change_id VARCHAR, team VARCHAR, implemented_on DATE);
      INSERT INTO past_changes VALUES ('CHG-1001', 'Payments', getvariable('anchor') - 3), ('CHG-1002', 'Payments', getvariable('anchor') - 2), ('CHG-1003', 'Web', getvariable('anchor') - 4), ('CHG-1004', 'Payments', getvariable('anchor') - 1);
      CREATE TABLE all_history AS
      SELECT change_id, team, implemented_on, outcome FROM history
      UNION ALL SELECT p.change_id, p.team, p.implemented_on, r.outcome FROM past_changes p JOIN (SELECT change_id, last(outcome) AS outcome FROM recorded GROUP BY change_id) r USING (change_id);

      CREATE TABLE blockers AS
      SELECT 'UNKNOWN_SERVICE' AS code, change_id AS ref, 'Service ' || service || ' is not in the CMDB: impact cannot be assessed' AS detail FROM changes c WHERE NOT EXISTS (SELECT 1 FROM services s WHERE s.service = c.service)
      UNION ALL SELECT 'WINDOW_INVERTED', change_id, 'Window ends ' || window_end || ' before it starts ' || window_start FROM changes WHERE window_end <= window_start
      UNION ALL SELECT 'BAD_OUTCOME', change_id, 'Outcome ' || outcome || ' is not SUCCESS or FAILED' FROM recorded WHERE outcome NOT IN ('SUCCESS', 'FAILED')
      UNION ALL SELECT 'UNKNOWN_CHANGE', change_id, 'Outcome recorded for a change that was not implemented' FROM recorded r
        WHERE NOT EXISTS (SELECT 1 FROM past_changes p WHERE p.change_id = r.change_id);

      -- =======================================================================
      -- Risk factors. Blast radius = services that depend on the changed one,
      -- directly or through others, weighted by their criticality.
      -- =======================================================================
      CREATE TABLE reach AS
      WITH RECURSIVE up(root, service) AS (
        SELECT service, service FROM services
        UNION
        SELECT up.root, d.service FROM up JOIN depends_on d ON d.dependency = up.service)
      SELECT root AS service, count(*) - 1 AS dependents, min(s.criticality) AS worst_criticality
      FROM up JOIN services s ON s.service = up.service GROUP BY root;
      CREATE TABLE team_cfr AS
      SELECT team, count(*) AS changes_90d, count(*) FILTER (WHERE outcome = 'FAILED') AS failed_90d,
             round(100.0 * count(*) FILTER (WHERE outcome = 'FAILED') / count(*), 1) AS cfr_pct
      FROM all_history WHERE implemented_on > getvariable('cab') - 90 GROUP BY team;

      CREATE TABLE f AS
      SELECT c.*, s.criticality, r.dependents, r.worst_criticality, coalesce(t.cfr_pct, 0) AS team_cfr_pct,
             (SELECT string_agg(z.name, ', ') FROM freezes z WHERE c.window_start < z.ends_at AND c.window_end > z.starts_at AND r.worst_criticality <= z.min_criticality) AS freeze_name,
             (SELECT string_agg(o.change_id, ', ') FROM changes o JOIN reach ro ON ro.service = o.service
                WHERE o.change_id <> c.change_id AND c.window_start < o.window_end AND c.window_end > o.window_start
                  AND (o.service = c.service OR o.service IN (SELECT dependency FROM depends_on WHERE service = c.service) OR c.service IN (SELECT dependency FROM depends_on WHERE service = o.service))) AS collides_with,
             isodow(c.window_start) BETWEEN 1 AND 5 AND hour(c.window_start) BETWEEN 9 AND 17 AS business_hours
      FROM changes c JOIN services s USING (service) JOIN reach r USING (service) LEFT JOIN team_cfr t USING (team)
      WHERE c.window_end > c.window_start;

      -- Score: sum of points per factor, capped at 100.
      CREATE TABLE scored AS
      SELECT *,
             [30, 20, 10, 0][criticality] AS p_criticality,
             least(20, dependents * 4) AS p_blast,
             CASE WHEN files_changed > 50 THEN 15 WHEN files_changed > 15 THEN 8 ELSE 0 END AS p_size,
             CASE WHEN NOT has_backout THEN 15 ELSE 0 END AS p_backout,
             CASE WHEN NOT tested_in_staging THEN 10 ELSE 0 END AS p_untested,
             CASE WHEN team_cfr_pct > 15 THEN 15 WHEN team_cfr_pct > 10 THEN 8 ELSE 0 END AS p_team,
             CASE WHEN needs_downtime THEN 10 WHEN business_hours AND criticality <= 2 THEN 5 ELSE 0 END AS p_timing
      FROM f;
      CREATE TABLE result AS
      SELECT *, least(100, p_criticality + p_blast + p_size + p_backout + p_untested + p_team + p_timing)::INT AS risk_score,
             concat_ws(', ',
               CASE WHEN p_criticality > 0 THEN 'criticality ' || criticality END, CASE WHEN p_blast > 0 THEN dependents || ' dependent services' END,
               CASE WHEN p_size > 0 THEN files_changed || ' files' END, CASE WHEN p_backout > 0 THEN 'no backout plan' END,
               CASE WHEN p_untested > 0 THEN 'not tested in staging' END, CASE WHEN p_team > 0 THEN team || ' failure rate ' || team_cfr_pct || '%' END,
               CASE WHEN needs_downtime THEN 'downtime' WHEN p_timing > 0 THEN 'business hours' END) AS factors
      FROM scored;
      ALTER TABLE result ADD COLUMN route VARCHAR; ALTER TABLE result ADD COLUMN reason VARCHAR;
      UPDATE result SET
        route = CASE WHEN freeze_name IS NOT NULL AND kind <> 'EMERGENCY' THEN 'REJECT_FREEZE'
                     WHEN kind = 'EMERGENCY' THEN 'ECAB_RETRO_REVIEW'
                     WHEN collides_with IS NOT NULL THEN 'CAB'
                     WHEN risk_score < 25 THEN 'PRE_APPROVED'
                     WHEN risk_score < 50 THEN 'PEER_REVIEW'
                     ELSE 'CAB' END,
        reason = CASE WHEN freeze_name IS NOT NULL AND kind <> 'EMERGENCY' THEN 'Window inside the ' || freeze_name || ' freeze: reschedule'
                      WHEN kind = 'EMERGENCY' THEN 'Emergency change: implement with ECAB approval, review at the next CAB'
                      WHEN collides_with IS NOT NULL THEN 'Overlaps ' || collides_with || ' on a related service'
                      ELSE 'Score ' || risk_score END;

      CREATE TABLE agenda AS
      SELECT row_number() OVER (ORDER BY route = 'CAB' DESC, risk_score DESC) AS item, change_id, title, service, team, window_start, risk_score, route, reason, factors
      FROM result WHERE route IN ('CAB', 'ECAB_RETRO_REVIEW', 'REJECT_FREEZE');

      COPY (SELECT change_id, title, service, team, kind, window_start, window_end, criticality, dependents, team_cfr_pct, freeze_name, collides_with,
                   p_criticality, p_blast, p_size, p_backout, p_untested, p_team, p_timing, risk_score, route, reason, factors FROM result ORDER BY risk_score DESC)
        TO '{{ outputFiles.changes }}' (HEADER, DELIMITER ',');
      COPY agenda TO '{{ outputFiles.agenda }}' (HEADER, DELIMITER ',');
      COPY blockers TO '{{ outputFiles.blockers }}' (HEADER, DELIMITER ',');

      SELECT
        (SELECT count(*) FROM result)::INT AS changes,
        (SELECT string_agg(route || ' ' || n, ', ' ORDER BY route) FROM (SELECT route, count(*) AS n FROM result GROUP BY route)) AS routes,
        (SELECT string_agg(change_id || ' ' || risk_score || ' ' || route, ', ' ORDER BY risk_score DESC) FROM result) AS scores,
        (SELECT coalesce(string_agg(change_id || ': ' || reason, '; ' ORDER BY change_id), '') FROM result WHERE route IN ('REJECT_FREEZE', 'CAB')) AS agenda_detail,
        (SELECT string_agg(team || ' ' || cfr_pct || '%', ', ' ORDER BY cfr_pct DESC) FROM team_cfr WHERE team IN (SELECT team FROM changes)) AS team_cfr,
        coalesce((SELECT json_group_object(change_id, outcome) FROM (SELECT change_id, last(outcome) AS outcome FROM recorded GROUP BY change_id) HAVING count(*) > 0)::VARCHAR, '{}') AS outcomes_all,
        (SELECT count(*) FROM blockers)::INT AS blockers,
        (SELECT coalesce(string_agg(code || ' ' || ref, ', ' ORDER BY code, ref), '') FROM blockers) AS blocker_detail;

  - id: result
    type: io.kestra.plugin.core.output.OutputValues
    values:
      r: "{{ (outputs.score.outputs | last).row | toJson }}"

  - id: log_score
    type: io.kestra.plugin.core.log.Log
    message: |
      CAB {{ inputs.cab_date }} ({{ inputs.scenario }}): {{ outputs.result.values.r | jq('"\(.changes) changes. \(.routes)."') | first }}
      {{ outputs.result.values.r | jq('"Scores: \(.scores)."') | first }}
      {{ outputs.result.values.r | jq('"Agenda: \(.agenda_detail)"') | first }}
      {{ outputs.result.values.r | jq('"Team change failure rate, 90 days: \(.team_cfr)."') | first }}
      Blockers: {{ outputs.result.values.r | jq('.blockers') | first }} {{ outputs.result.values.r | jq('.blocker_detail') | first }}

  - id: evidence_blockers
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "cab/{{ inputs.cab_date }}/blockers.csv": "{{ outputs.score.outputFiles.blockers }}"

  - id: gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.result.values.r | jq('.blockers > 0') | first }}"
    then:
      - id: blocked
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Change scoring stopped: {{ outputs.result.values.r |
          jq('.blocker_detail') | first }}"

  - id: evidence
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "cab/{{ inputs.cab_date }}/changes.csv": "{{ outputs.score.outputFiles.changes }}"
      "cab/{{ inputs.cab_date }}/agenda.csv": "{{ outputs.score.outputFiles.agenda }}"

  - id: save
    type: io.kestra.plugin.core.kv.Set
    description: Outcomes are kept, so a failed change raises its team's failure
      rate in the next scoring.
    key: "{{ vars.state_key }}"
    kvType: JSON
    value: "{{ {'outcomes':
      fromJson(fromJson(outputs.result.values.r).outcomes_all)} | toJson }}"

triggers:
  - id: weekly_cab
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 7 * * 1"
    disabled: true
    inputs:
      cab_date: "{{ trigger.date | date('yyyy-MM-dd') }}"