Run Only the Monorepo Tests Your Changed Paths Touch

id: monorepo-selective-ci
namespace: company.team
description: |
  Monorepo CI runs the full package matrix on every push, so a docs edit
  costs the same as a service release. A push webhook or an input supplies the
  changed file list, Python maps each path to package test targets from an
  in-file package map, a Loop runs only the selected suites inside one
  container step, and an If escalates changed paths that no package covers
  instead of letting them pass untested. Suites report their attempts, and the
  target selection with the unmapped paths is published as outputs.

triggers:
  - id: on_push_changed_paths
    type: io.kestra.plugin.core.trigger.Webhook
    description: The push hook or CI router POSTs the changed file list here so only
      the touched package suites run for this revision.
    key: "{{ secret('MONOREPO_PUSH_WEBHOOK_KEY') }}"
  - id: nightly_path_rescan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Nightly 03:00 UTC net that reruns the selection over the last
      changed path list, shipped disabled until the webhook path is trusted in
      your environment.
    cron: "0 3 * * *"
    disabled: true

inputs:
  - id: changed_files
    type: STRING
    defaults: services/api/src/handlers.py,libs/auth/token.go,docs/setup.md,README.md
    description: Comma separated changed path list used when no webhook body
      arrived, the same list the push hook would post.
  - id: base_ref
    type: STRING
    defaults: origin/main
    description: Revision the change list was diffed against, echoed in the alerts
      so reruns against different bases stay distinguishable.
  - id: fail_on_unmapped
    type: BOOL
    defaults: true
    description: Whether changed paths that map to no test package turn the
      execution red after the Slack alert, flip to false for advisory runs.

tasks:
  # --- 1. target selection ---
  - id: select_test_targets
    type: io.kestra.plugin.scripts.python.Script
    description: Parse the changed path list from the posted push body or the input
      fallback, match every path against the package map with longest prefix
      rules, and emit the selected test targets with the unmapped leftovers
      through the outputs protocol.
    containerImage: python:3.12-slim
    env:
      CHANGED_FILES: "{{ trigger.body ?? inputs.changed_files }}"
    script: |
      import json
      import os

      DEFAULT_CHANGED_FILES = "services/api/src/handlers.py,libs/auth/token.go,docs/setup.md,README.md"

      PACKAGE_MAP = [
          {"path": "services/api/", "target": "services/api", "command": "python -m pytest tests/api -q"},
          {"path": "services/web/", "target": "services/web", "command": "npm test --prefix services/web"},
          {"path": "libs/auth/", "target": "libs/auth", "command": "go test ./libs/auth/..."},
          {"path": "libs/queue/", "target": "libs/queue", "command": "go test ./libs/queue/..."},
          {"path": "db/migrations/", "target": "db/migrations", "command": "sqlfluff lint db/migrations --rules core"},
          {"path": "deploy/helm/", "target": "deploy/helm", "command": "helm lint deploy/helm/chart"},
          {"path": "tests/e2e/", "target": "tests/e2e", "command": "npx playwright test"},
      ]
      PACKAGE_MAP.sort(key=lambda entry: -len(entry["path"]))

      raw = os.environ.get("CHANGED_FILES")
      if raw is None:
          raw = DEFAULT_CHANGED_FILES

      payload = {
          "select_ok": False,
          "files_changed": 0,
          "targets": [],
          "targets_selected": 0,
          "selected_names": "",
          "unmapped_files": [],
          "unmapped_paths": "",
          "unmapped_count": 0,
          "error": "",
      }

      text = (raw or "").strip()
      files = []
      if text.startswith("["):
          try:
              parsed = json.loads(text)
          except ValueError:
              parsed = None
          if isinstance(parsed, list):
              files = [str(item).strip() for item in parsed if str(item).strip()]
          else:
              payload["error"] = "changed file payload is not a JSON array of paths"
      elif text:
          files = [chunk.strip() for chunk in text.replace("\n", ",").split(",") if chunk.strip()]

      if payload["error"]:
          pass
      elif not files:
          payload["error"] = "changed file list is empty"
      else:
          selected = {}
          unmapped = []
          for path in files:
              match = None
              for entry in PACKAGE_MAP:
                  if path.startswith(entry["path"]):
                      match = entry
                      break
              if match is None:
                  unmapped.append(path)
                  continue
              record = selected.get(match["target"])
              if record is None:
                  record = {
                      "target": match["target"],
                      "command": match["command"],
                      "path": match["path"],
                      "matched_files": 0,
                  }
                  selected[match["target"]] = record
              record["matched_files"] += 1
          targets = list(selected.values())
          payload.update({
              "select_ok": True,
              "files_changed": len(files),
              "targets": targets,
              "targets_selected": len(targets),
              "selected_names": ", ".join(entry["target"] for entry in targets),
              "unmapped_files": unmapped,
              "unmapped_paths": ", ".join(unmapped),
              "unmapped_count": len(unmapped),
          })

      if not payload["select_ok"]:
          payload["error"] = payload["error"] or "changed paths could not be mapped to test targets"

      print("select_ok=%s files=%d targets=%d unmapped=%d" % (
          payload["select_ok"], payload["files_changed"],
          payload["targets_selected"], payload["unmapped_count"]))
      print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 2. selected suite runs ---
  - id: run_selected_suites
    type: io.kestra.plugin.core.flow.Loop
    description: Walk the selected test targets one at a time and attempt each
      package suite command in its own container step, so the execution history
      shows exactly which suites this change paid for.
    concurrencyLimit: 2
    values: "{{ outputs.select_test_targets.vars.targets }}"
    runIf: "{{ (outputs.select_test_targets.vars.targets | length) > 0 }}"
    tasks:
      - id: run_suite
        type: io.kestra.plugin.scripts.shell.Commands
        description: Attempt the mapped suite command for the current loop target,
          absorb the expected miss on a bare fixture container, and record the
          attempt for the execution history.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: alpine:3.20
        commands:
          - |
            echo "suite {{ item.value.target }} starting - {{ item.value.command }}"
            {{ item.value.command }} || echo "suite {{ item.value.target }} exited non zero, recorded for the gate"
            echo "suite {{ item.value.target }} finished after {{ item.value.matched_files }} matched file(s)"

  # --- 3. path coverage gate ---
  - id: path_coverage_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch when the selection failed or changed paths matched no test
      package, so unmapped changes escalate instead of passing untested.
    condition: "{{ not outputs.select_test_targets.vars.select_ok or
      outputs.select_test_targets.vars.unmapped_count > 0 }}"
    then:
      # --- 4. coverage escalation ---
      - id: check_selection_health
        type: io.kestra.plugin.core.flow.If
        description: Split a selection that ran from one that could not parse the
          changed path payload at all, so a broken push hook never reads as full
          coverage.
        condition: "{{ outputs.select_test_targets.vars.select_ok }}"
        then:
          - id: alert_unmapped_paths
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the unmapped changed paths, the selected targets and the file
              count so the owning team can extend the package map before the
              next push.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Monorepo selective CI skipped {{ outputs.select_test_targets.vars.unmapped_count }} changed path(s) that map to no test package against {{ inputs.base_ref }} - {{ outputs.select_test_targets.vars.unmapped_paths }}. Selected and run, {{ outputs.select_test_targets.vars.targets_selected }} target(s) ({{ outputs.select_test_targets.vars.selected_names }}) from {{ outputs.select_test_targets.vars.files_changed }} changed file(s). Execution {{ execution.id }}."
              }
          - id: enforce_unmapped_policy
            type: io.kestra.plugin.core.flow.If
            description: Apply the fail on unmapped policy, turning the run red for gating
              pipelines and staying advisory when the input is flipped off.
            condition: "{{ inputs.fail_on_unmapped }}"
            then:
              - id: fail_execution
                type: io.kestra.plugin.core.execution.Fail
                description: Fail the execution so the push pipeline stops instead of merging
                  with changed paths no suite covered.
                errorMessage: "Monorepo selective CI left {{
                  outputs.select_test_targets.vars.unmapped_count }} changed
                  path(s) unmapped against {{ inputs.base_ref }} - {{
                  outputs.select_test_targets.vars.unmapped_paths }}."
            else:
              - id: log_policy_softened
                type: io.kestra.plugin.core.log.Log
                description: Record the unmapped paths without failing, keeping the run green
                  for advisory estates.
                message: "unmapped changed paths found - {{
                  outputs.select_test_targets.vars.unmapped_count }} path(s) not
                  enforced, fail_on_unmapped is {{ inputs.fail_on_unmapped }},
                  selected suites already ran"
        else:
          - id: alert_selection_failed
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Tell the ops channel the changed path payload could not be parsed
              and that no suites ran.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {"text": "Monorepo selective CI could not map changed paths against {{ inputs.base_ref }} - {{ outputs.select_test_targets.vars.error }}. No test suites ran. Execution {{ execution.id }}."}
    else:
      - id: log_all_paths_mapped
        type: io.kestra.plugin.core.log.Log
        description: Record the fully covered verdict with the target list so the
          history doubles as evidence the change was tested.
        message: "monorepo paths mapped - {{
          outputs.select_test_targets.vars.files_changed }} changed file(s)
          selected {{ outputs.select_test_targets.vars.targets_selected }} test
          target(s) ({{ outputs.select_test_targets.vars.selected_names }}),
          every path covered against {{ inputs.base_ref }}"

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: targets_selected
    type: INT
    description: Number of package test targets selected from the changed paths, for
      example 2 when an API file and an auth library file changed.
    value: "{{ outputs.select_test_targets.vars.targets_selected }}"
  - id: files_changed
    type: INT
    description: Number of changed paths parsed from the push payload or the input
      fallback before package mapping, for example 4.
    value: "{{ outputs.select_test_targets.vars.files_changed }}"
  - id: selected_targets
    type: ARRAY
    description: Package test targets the Loop attempted, for example
      ["services/api", "libs/auth"], empty when no path mapped.
    value: "{{ outputs.select_test_targets.vars.targets }}"
  - id: unmapped_paths
    type: ARRAY
    description: Changed paths that matched no package map entry and therefore ran
      no suite, for example ["docs/setup.md", "README.md"].
    value: "{{ outputs.select_test_targets.vars.unmapped_files }}"