Tell a flaky test apart from a genuinely broken one by flip rate

id: flaky-test-flip-rate-detector
namespace: company.team

inputs:
  - id: test_image
    type: STRING
    displayName: Test container image
    description: Image the suite runs in. It needs whatever your tests need, plus a
      JUnit XML reporter.
    defaults: ghcr.io/kestra-io/pydata:latest

  - id: setup_commands
    type: STRING
    displayName: Setup commands
    description: >
      Shell run once before the reruns. Point it at your own checkout, for
      example `git clone --depth 1 --branch main https://github.com/acme/api
      repo && cd repo && pip install -r requirements-dev.txt`. The default
      installs pytest and uses the bundled demo suite.
    defaults: pip install --quiet pytest

  - id: test_command
    type: STRING
    displayName: Test command
    description: >
      One run of your suite. It must write JUnit XML to `$JUNIT_REPORT` and it
      may exit non-zero. `$RUN_INDEX` is 1..rerun_count, so a test can key off
      it. Jest `--reporters=jest-junit` and Maven surefire reports parse
      unchanged.
    defaults: python -m pytest demo_suite -q --junit-xml="$JUNIT_REPORT"

  - id: rerun_count
    type: INT
    displayName: Reruns per execution
    description: How many times to run the suite on this one unchanged commit. Each
      rerun costs a full suite, so keep it small and let the history do the
      work.
    defaults: 6

  - id: history_window
    type: INT
    displayName: History window
    description: Most outcomes kept per test. A test not seen for this many
      executions is forgotten, so the KV entry cannot grow without bound.
    defaults: 40

  - id: flake_threshold
    type: FLOAT
    displayName: Flake threshold
    description: Flip rate at or above this scores FLAKE. Flip rate is outcome
      changes divided by observations minus one, so 0.1 catches one flip in ten.
    defaults: 0.1

  - id: confidence_window
    type: FLOAT
    displayName: Confidence window
    description: Verdicts drawn from fewer observations than this are marked low
      confidence. A short window under-reports flakes, it never over-reports
      them.
    defaults: 10

  - id: quarantine_limit
    type: INT
    displayName: Quarantine limit
    description: Most tests the proposal will name. A cap keeps one bad day from
      proposing the whole suite.
    defaults: 10

  - id: history_key
    type: STRING
    displayName: KV history key
    description: Namespace KV entry holding the outcome history. Use one key per suite.
    defaults: flaky_test_history

  - id: file_github_issue
    type: BOOL
    displayName: Update a GitHub tracker comment
    description: Keep one self-updating comment on a GitHub issue. Needs the
      GITHUB_TOKEN secret. Off by default so the flow runs with no secret
      configured.
    defaults: false

  - id: github_repository
    type: STRING
    displayName: GitHub repository
    description: Repository holding the tracker issue, as owner/repo. Only used when
      file_github_issue is true.
    defaults: acme/api

  - id: github_issue_number
    type: INT
    displayName: GitHub issue number
    description: The long-lived flaky test tracker issue. One comment on it is
      updated in place, never appended to.
    defaults: 1

concurrency:
  limit: 1

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run every night. The history only becomes conclusive by
      accumulating across executions, so this trigger is the point of the flow
      rather than a convenience.
    cron: "0 3 * * *"

tasks:
  - id: load_history
    type: io.kestra.plugin.core.kv.Get
    description: Read the outcome history written by earlier executions. Missing on
      the first run, which is fine.
    key: "{{ inputs.history_key }}"
    errorOnMissing: false

  - id: detect
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: >
      The reruns and the scorer share one working directory, so the JUnit
      reports pass between them as files. Do not add outputFiles to rerun_suite,
      it moves the reports out of the shared directory before the scorer can
      read them.
    tasks:
      - id: rerun_suite
        type: io.kestra.plugin.scripts.shell.Commands
        description: Run the suite rerun_count times on the same commit, same image,
          same inputs. One container, so the reruns cost one image pull and one
          setup.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: "{{ inputs.test_image }}"
        beforeCommands:
          - "{{ inputs.setup_commands }}"
        inputFiles:
          demo_suite/test_flaky_demo.py: |
            """A demo suite with one of each outcome shape, so the scoring is visible.

            RUN_INDEX makes every flip deterministic. A real flake is a race, a clock
            or a shared fixture. What the detector reads is the outcome pattern, never
            the cause, so a modulo stands in for a race without changing the maths.
            """
            import os

            import pytest

            RUN = int(os.environ.get("RUN_INDEX", "1"))


            def test_stable_pass():
                assert 2 + 2 == 4


            def test_flaky_clock():
                # Fails on every even run. Flips on every rerun, so flip rate is 1.0.
                assert RUN % 2 == 1, "run %d: lock was still held" % RUN


            def test_genuinely_broken():
                # Fails on every run. Reads red in CI exactly like the flake above,
                # and never flips. This is the contrast the whole flow exists for.
                raise RuntimeError("off-by-one in the paginator")


            @pytest.fixture
            def unreachable_db():
                raise RuntimeError("fixture cannot reach the test database")


            def test_broken_fixture(unreachable_db):
                # A setup failure lands under <error>, not <failure>. Both are a fail.
                assert unreachable_db


            def test_flaky_rare():
                # Fails once every seven runs, so the default rerun_count of 6 never
                # sees it fail and scores it stable. That is the honest limit of a
                # short window. Raise rerun_count to 14, or let the nightly history
                # accumulate, and it shows up.
                assert RUN % 7 != 0, "run %d: retry budget exhausted" % RUN


            def test_conditionally_skipped():
                # Skipped on every third run. A skip is not an outcome, so it is
                # dropped and this test's window stays narrower than the others.
                if RUN % 3 == 0:
                    pytest.skip("not run on this shard")
                assert True
        env:
          RERUN_COUNT: "{{ inputs.rerun_count }}"
          TEST_COMMAND: "{{ inputs.test_command }}"
        commands:
          - mkdir -p reports
          - |
            i=1
            while [ "$i" -le "$RERUN_COUNT" ]; do
              RUN_INDEX="$i"
              JUNIT_REPORT="reports/run-$(printf '%04d' "$i").xml"
              export RUN_INDEX JUNIT_REPORT
              echo "--- rerun $i of $RERUN_COUNT"
              sh -c "$TEST_COMMAND" || echo "rerun $i exited non-zero, which is expected while hunting flakes"
              i=$((i + 1))
            done
          - |
            written=$(ls -1 reports/*.xml 2>/dev/null | wc -l)
            echo "reports written: $written of $RERUN_COUNT reruns"
            if [ "$written" -eq 0 ]; then
              echo "No JUnit XML was written. test_command must write to \$JUNIT_REPORT." >&2
              exit 1
            fi

      - id: score_flips
        type: io.kestra.plugin.scripts.python.Script
        description: Parse every report, append each test's outcome to its history, then
          score by flip rate. A flip rate above zero is nondeterminism. A run of
          failures with no flip is a real bug.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: "{{ inputs.test_image }}"
        inputFiles:
          history.json: "{{ outputs.load_history.value ?? '{}' }}"
        outputFiles:
          - reports/*.xml
        env:
          HISTORY_WINDOW: "{{ inputs.history_window }}"
          FLAKE_THRESHOLD: "{{ inputs.flake_threshold }}"
          CONFIDENCE_WINDOW: "{{ inputs.confidence_window }}"
          QUARANTINE_LIMIT: "{{ inputs.quarantine_limit }}"
        script: |
          import glob
          import json
          import os
          import xml.etree.ElementTree as ET

          WINDOW = max(2, int(os.environ["HISTORY_WINDOW"]))
          THRESHOLD = float(os.environ["FLAKE_THRESHOLD"])
          CONFIDENCE = int(float(os.environ["CONFIDENCE_WINDOW"]))
          LIMIT = max(1, int(os.environ["QUARANTINE_LIMIT"]))
          MAX_REPORT_BYTES = 32 * 1024 * 1024
          MAX_TESTS = 5000

          # Reports are zero padded by the rerun loop, so a plain sort is run order.
          paths = sorted(glob.glob("reports/*.xml"))

          observed = {}
          parsed = 0
          skipped_rows = 0
          for path in paths:
              if os.path.getsize(path) > MAX_REPORT_BYTES:
                  print("skipping oversized report:", path)
                  continue
              try:
                  root = ET.parse(path).getroot()
              except ET.ParseError as exc:
                  print("unparseable report, ignored:", path, exc)
                  continue
              parsed += 1
              # iter() includes the element itself, so this handles the pytest
              # <testsuites> wrapper and the bare <testsuite> root that surefire,
              # jest-junit and Ant emit.
              for suite in root.iter("testsuite"):
                  for case in suite.iter("testcase"):
                      name = case.get("name") or "(unnamed)"
                      cls = case.get("classname") or ""
                      tid = cls + "::" + name if cls else name
                      kinds = set(child.tag for child in case)
                      # A skip is an absence, not an outcome. Counting it would make
                      # a conditionally skipped test look like it flipped.
                      if "skipped" in kinds:
                          skipped_rows += 1
                          continue
                      # Pass is the absence of failure and error. pytest omits the
                      # type attribute on <failure>, and it files an exception in a
                      # test body under <failure> while surefire files it under
                      # <error>, so neither attribute nor tag choice is reliable.
                      ok = not (kinds & set(["failure", "error"]))
                      if len(observed) >= MAX_TESTS and tid not in observed:
                          continue
                      observed.setdefault(tid, []).append("1" if ok else "0")

          state = {}
          try:
              state = json.loads(open("history.json").read() or "{}")
          except ValueError:
              print("history was not valid JSON, starting a new one")
          if not isinstance(state, dict):
              state = {}
          runs = int(state.get("runs", 0)) + 1
          tests = state.get("tests")
          if not isinstance(tests, dict):
              tests = {}

          for tid, bits in observed.items():
              row = tests.get(tid)
              if not isinstance(row, dict):
                  row = {"bits": "", "last_run": runs}
              row["bits"] = (str(row.get("bits", "")) + "".join(bits))[-WINDOW:]
              row["last_run"] = runs
              tests[tid] = row

          # Forget anything not seen for a whole window of executions.
          tests = dict(
              (tid, row)
              for tid, row in tests.items()
              if runs - int(row.get("last_run", runs)) < WINDOW
          )

          rows = []
          for tid, row in tests.items():
              bits = str(row.get("bits", ""))
              width = len(bits)
              flips = sum(1 for a, b in zip(bits, bits[1:]) if a != b)
              rate = flips / float(width - 1) if width > 1 else 0.0
              if width < 2:
                  verdict = "NEW"
              elif rate >= THRESHOLD:
                  verdict = "FLAKE"
              elif "1" not in bits:
                  verdict = "BROKEN"
              elif "0" not in bits:
                  verdict = "STABLE"
              elif bits[-1] == "0":
                  verdict = "REGRESSED"
              else:
                  verdict = "STABLE"
              rows.append({
                  "id": tid,
                  "name": tid.split("::")[-1],
                  "bits": bits,
                  "window": width,
                  "flips": flips,
                  "rate": round(rate, 3),
                  "verdict": verdict,
                  "low_confidence": width < CONFIDENCE,
                  "seen_now": tid in observed,
              })

          order = {"FLAKE": 0, "REGRESSED": 1, "BROKEN": 2, "NEW": 3, "STABLE": 4}
          rows.sort(key=lambda r: (order[r["verdict"]], -r["rate"], -r["window"], r["id"]))

          head = "| test | history (oldest first) | window | flips | flip rate | verdict |"
          table = [head, "| --- | --- | --- | --- | --- | --- |"]
          for r in rows:
              mark = r["verdict"] + (" (low confidence)" if r["low_confidence"] else "")
              table.append("| `%s` | `%s` | %d | %d | %.3f | %s |" % (
                  r["id"], r["bits"], r["window"], r["flips"], r["rate"], mark))
          table = "\n".join(table)

          flakes = [r for r in rows if r["verdict"] == "FLAKE"][:LIMIT]
          broken = [r for r in rows if r["verdict"] == "BROKEN"]
          regressed = [r for r in rows if r["verdict"] == "REGRESSED"]

          if flakes:
              quarantine = "\n".join(
                  "- `%s` flips %.0f%% of the time over %d observations" % (
                      r["id"], r["rate"] * 100, r["window"])
                  for r in flakes)
              deselect = 'pytest -k "' + " and ".join(
                  "not " + r["name"] for r in flakes) + '"'
          else:
              quarantine = "- nothing to quarantine in this window"
              deselect = "(no quarantine needed)"

          widths = [r["window"] for r in rows] or [0]
          for r in rows:
              print("%-9s %-46s bits=%-12s window=%d flips=%d rate=%.3f%s" % (
                  r["verdict"], r["id"], r["bits"], r["window"], r["flips"],
                  r["rate"], "  LOW CONFIDENCE" if r["low_confidence"] else ""))

          out = {}
          out["runs"] = runs
          out["reports_parsed"] = parsed
          out["skipped_rows_dropped"] = skipped_rows
          out["tests_scored"] = len(rows)
          out["flake_count"] = len([r for r in rows if r["verdict"] == "FLAKE"])
          out["broken_count"] = len(broken)
          out["regressed_count"] = len(regressed)
          out["stable_count"] = len([r for r in rows if r["verdict"] == "STABLE"])
          out["new_count"] = len([r for r in rows if r["verdict"] == "NEW"])
          out["min_window"] = min(widths)
          out["max_window"] = max(widths)
          out["low_confidence_count"] = len([r for r in rows if r["low_confidence"]])
          out["table"] = table
          out["quarantine"] = quarantine
          out["deselect"] = deselect
          out["history"] = json.dumps({"version": 1, "runs": runs, "tests": tests})
          print("::" + json.dumps({"outputs": out}) + "::")

  - id: save_history
    type: io.kestra.plugin.core.kv.Set
    description: Persist the widened history. No ttl, because kv.Get throws on an
      expired key and losing the history silently would make every verdict low
      confidence again.
    key: "{{ inputs.history_key }}"
    value: "{{ outputs.score_flips.vars.history }}"
    kvType: STRING

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: The whole verdict table, with every window size next to its verdict.
    message: |
      Flip rate over {{ outputs.score_flips.vars.runs }} execution(s), {{ inputs.rerun_count }} reruns each.
      Parsed {{ outputs.score_flips.vars.reports_parsed }} JUnit report(s). Dropped {{ outputs.score_flips.vars.skipped_rows_dropped }} skipped row(s).
      {{ outputs.score_flips.vars.tests_scored }} test(s) scored: {{ outputs.score_flips.vars.flake_count }} flake, {{ outputs.score_flips.vars.broken_count }} broken, {{ outputs.score_flips.vars.regressed_count }} regressed, {{ outputs.score_flips.vars.stable_count }} stable, {{ outputs.score_flips.vars.new_count }} new.
      Windows run {{ outputs.score_flips.vars.min_window }} to {{ outputs.score_flips.vars.max_window }} observations. {{ outputs.score_flips.vars.low_confidence_count }} verdict(s) below the {{ inputs.confidence_window }} observation confidence window.

      {{ outputs.score_flips.vars.table }}

  - id: triage
    type: io.kestra.plugin.core.flow.If
    description: A flake gets quarantined. A deterministic failure never does, it
      gets fixed.
    condition: "{{ outputs.score_flips.vars.flake_count > 0 }}"
    then:
      - id: propose_quarantine
        type: io.kestra.plugin.core.log.Log
        message: |
          Quarantine proposal, at most {{ inputs.quarantine_limit }} test(s):
          {{ outputs.score_flips.vars.quarantine }}

          Deselect them with:
          {{ outputs.score_flips.vars.deselect }}

          Not quarantined: {{ outputs.score_flips.vars.broken_count }} test(s) scored BROKEN and {{ outputs.score_flips.vars.regressed_count }} scored REGRESSED. They fail every time, so they are bugs to fix rather than noise to hide.

      - id: publish_tracker
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.file_github_issue }}"
        then:
          - id: tracker_comment
            type: io.kestra.plugin.github.issues.Comment
            description: One comment on the tracker issue, updated in place by updateTag.
              The history is only readable as a single comment that changes,
              never as a new comment per night.
            oauthToken: "{{ secret('GITHUB_TOKEN') }}"
            repository: "{{ inputs.github_repository }}"
            issueNumber: "{{ inputs.github_issue_number }}"
            updateTag: kestra-flaky-test-flip-rate
            body: |
              ## Flaky test flip rate

              {{ outputs.score_flips.vars.runs }} execution(s) of {{ inputs.rerun_count }} reruns each, all on one unchanged commit. A test that changes outcome without the code changing is nondeterministic. A test that fails every single time is a bug.

              {{ outputs.score_flips.vars.table }}

              ### Quarantine proposal

              {{ outputs.score_flips.vars.quarantine }}

              ```
              {{ outputs.score_flips.vars.deselect }}
              ```

              {{ outputs.score_flips.vars.broken_count }} test(s) scored BROKEN and {{ outputs.score_flips.vars.regressed_count }} scored REGRESSED. Those are excluded from the proposal on purpose.

              Windows run {{ outputs.score_flips.vars.min_window }} to {{ outputs.score_flips.vars.max_window }} observations. A verdict under {{ inputs.confidence_window }} observations is marked low confidence, because a short window hides flakes rather than inventing them.

              <sub>Kestra execution `{{ execution.id }}`</sub>
        else:
          - id: tracker_skipped
            type: io.kestra.plugin.core.log.Log
            message: "file_github_issue is false, so nothing was posted. With it true this
              would update the comment tagged kestra-flaky-test-flip-rate on {{
              inputs.github_repository }}#{{ inputs.github_issue_number }},
              using the GITHUB_TOKEN secret."
    else:
      - id: no_flakes
        type: io.kestra.plugin.core.log.Log
        message: |
          No test flipped at or above a flip rate of {{ inputs.flake_threshold }} in this window, so nothing is proposed for quarantine.
          {{ outputs.score_flips.vars.broken_count }} test(s) scored BROKEN and {{ outputs.score_flips.vars.regressed_count }} scored REGRESSED. Those fail deterministically, so they are real failures.
          {{ outputs.score_flips.vars.low_confidence_count }} verdict(s) came from fewer than {{ inputs.confidence_window }} observations. A narrow window under-reports flakes, so let the nightly trigger widen it before trusting a STABLE verdict.