Adds a script to analyze autorevert decisions and correlate with TD data and behaviour (#8333)

## What
Adds a new standalone CLI, `flake-test-fail-autorevert`, that reads
pytorch-auto-revert's already-published decisions from ClickHouse and
exports them as a long-format CSV (one row per test signal, categorized
`regression` or `flaky`), plus a companion `flake-report` command that
renders a self-contained HTML report (flakiness/regression rankings,
per-day time-series, and a pre-merge trunk-gate funnel) from that CSV.

## Why
We need visibility into what the autorevert system is actually catching
and how those regressions relate to the pre-merge gate. The exporter
answers "which test signals triggered reverts or were flagged flaky,
over a commit-landing range," and the report turns that into a shareable
view — notably the `premerge_status` classification, which shows how
many trunk/pull regressions were landraces (green pre-merge, red on
main) versus caught-but-merged, force-merged, or coverage gaps
(TD-deselected / not-in-matrix).

The tool is read-only against ClickHouse; it re-reads stored decisions
and never re-runs analysis.

---------

Signed-off-by: Jean Schmidt <contato@jschmidt.me>

Author: jeanschmidt

tools/flake-test-fail-autorevert/.env.example

@@ -0,0 +1,4 @@
+CLICKHOUSE_HOST=
+CLICKHOUSE_PORT=8443
+CLICKHOUSE_USERNAME=
+CLICKHOUSE_PASSWORD=

tools/flake-test-fail-autorevert/.gitignore

@@ -0,0 +1,6 @@
+**/report/assets/*.js
+*.report.html
+.venv/
+__pycache__/
+*.egg-info/
+.pytest_cache/

tools/flake-test-fail-autorevert/README.md

@@ -0,0 +1,163 @@
+# flake-test-fail-autorevert
+
+Standalone CLI that reads pytorch-auto-revert's **already-published** decisions
+from ClickHouse and writes a CSV. It does not re-run any analysis: it only reads
+stored decisions.
+
+The CSV is **long**: one row per signal. A commit (landing on `main` within the
+requested range) contributes one row for each test signal that either:
+
+- triggered an autorevert on a **test** regression (`category = regression`), or
+- was flagged **flaky** by autorevert (`category = flaky`).
+
+Only test-level signals are reported (signal keys containing `::`); job-level
+signals are excluded.
+
+## Output columns
+
+- `commit_sha` — full 40-char sha
+- `commit_url` — `https://github.com/{repo}/commit/{commit_sha}`
+- `commit_time` — landing time on `main` (`YYYY-MM-DD HH:MM:SS`, UTC)
+- `category` — `regression` (this signal triggered an autorevert) or `flaky`
+  (autorevert observed both a pass and a fail on this commit for this signal)
+- `workflow` — the CI workflow the signal belongs to (e.g. `trunk`, `pull`,
+  `inductor`, `periodic`, `slow`), or `unknown`. See attribution below.
+- `signal_key` — the test signal key, `file.py::test_name`
+- `advisor_verdict` — for `regression` rows, the auto-revert advisor's verdict
+  (`related`, `not_related`, `infra_issue`, `garbage`, `revert`, `unsure`) when
+  one exists, else empty. Always empty for `flaky` rows.
+- `advisor_confidence` — for `regression` rows with a verdict, the advisor's
+  confidence formatted to two decimals (e.g. `0.99`), else empty. Always empty
+  for `flaky` rows.
+- `premerge_status` — for `regression` rows on the `trunk` or `pull` workflow, the
+  pre-merge trunk-gate status of the test on the merged commit's validated head;
+  empty for all other rows. One of:
+  - `RUN_SUCCEEDED` — the test ran on the pre-merge head and passed. Only ever
+    emitted from a POSITIVE success-row observation, never from an empty read.
+  - `RUN_FAILED` — the test ran and at least one shard failed ("merged despite
+    red"). Checked before success, so a mixed pass/fail set reports as failed.
+  - `NOT_RUN:force_merge` — a REAL force merge (`skip_mandatory_checks` set on the
+    merge, i.e. `-f`) that bypassed the gate AND the test did not run at all. A
+    force merge that still ran the test reports the test's real verdict instead —
+    force_merge never masks a real outcome.
+  - `NOT_RUN:skipped` — the test ran but every run was skipped.
+  - `NOT_RUN:td_deselected` — the test's file ran but the test was deselected
+    (test dependency / target determination).
+  - `NOT_RUN:not_in_matrix` — the test's file never ran on the head (job not in
+    the matrix, or no gate jobs at all on a non-force merge).
+  - `NOT_RUN:no_merge_record` — no `default.merges` row resolved a pre-merge head
+    for this commit, so we cannot classify it. This is the honest label for a
+    ghstack **non-tip** commit (only the stack's tip PR gets a merges row keyed by
+    its squashed commit), a revert, a direct push, or data predating the merges
+    table. It is NOT an inference of force merge. See the coverage note below.
+  - `ERROR` — a query failed after retries, or the merge timestamp was missing.
+
+Rows are sorted by `(commit_time, category, workflow, signal_key)` ascending.
+
+The same `(commit, signal_key)` can appear as both a `regression` row and a
+`flaky` row: the two categories answer independent questions (did it trigger an
+autorevert vs. did autorevert observe both a pass and a fail on that commit).
+
+### How `workflow` is attributed
+
+- **Flaky rows**: the workflow is exact, taken directly from the autorevert
+  state snapshot the flaky signal was read from. A single `(commit, signal_key)`
+  legitimately observed flaky under two workflows produces two rows.
+- **Regression rows**: the reverted event stores the triggering workflows and
+  the source signals as two independent arrays (a deduped set and an ordered
+  list) that cannot be positionally zipped, so the workflow is resolved per
+  signal with a fallback: the auto-revert advisor's workflow for that
+  `(commit, signal_key)` if present, otherwise the revert event's sole workflow
+  when it triggered on exactly one workflow, otherwise `unknown`.
+
+## Environment
+
+Reads the same connection variables as the `pytorch-auto-revert` lambda:
+
+- `CLICKHOUSE_HOST` — host or URL (`https://` prefix and `:8443` suffix are
+  stripped automatically)
+- `CLICKHOUSE_PORT` — optional, defaults to `8443`
+- `CLICKHOUSE_USERNAME`
+- `CLICKHOUSE_PASSWORD`
+
+Set these in the environment or in a `.env` file in the tool dir (see
+`.env.example`), which is loaded on startup.
+
+## Run
+
+```
+cd tools/flake-test-fail-autorevert
+uv run flake-test-fail-autorevert --start 2026-07-01 --end 2026-07-14 \
+    [--repo pytorch/pytorch] [--output out.csv]
+```
+
+`uv` auto-creates and manages a local `.venv` and installs the dependencies
+declared in `pyproject.toml`, so there is no manual venv or `pip install` step.
+
+Run the tests with:
+
+```
+uv run pytest
+```
+
+### Alternative (pip)
+
+```
+pip install -e .
+flake-test-fail-autorevert --start 2026-07-01 --end 2026-07-14
+# or equivalently:
+python -m flake_test_fail_autorevert --start 2026-07-01 --end 2026-07-14
+```
+
+- `--start` / `--end` are dates (`YYYY-MM-DD`). The range is by commit landing
+  time on `main`, and `--end` is **inclusive** (the whole end day is included):
+  the effective window is `[start 00:00:00, (end + 1 day) 00:00:00)`.
+- `--repo` defaults to `pytorch/pytorch`.
+- `--output` defaults to
+  `flake_test_fail_autorevert_<start>_<end>.csv`. The path and a one-line summary
+  (`N rows across M commits: R regression, F flaky`) are printed on completion.
+
+## Notes on `premerge_status` coverage
+
+The pre-merge head is resolved from `default.merges`, which is keyed by the
+**merge command's** commit — for a ghstack stack that is only the **tip** PR's
+squashed commit. A ghstack **non-tip** commit lands its own squashed commit on
+`main` but has no `default.merges` row keyed by that commit, so its pre-merge head
+cannot be resolved and it is reported as `NOT_RUN:no_merge_record`. Autorevert
+frequently bisects a regression to a non-tip culprit, so this is a real coverage
+gap, not an edge case. There is currently no clean, reliable way to recover the
+non-tip pre-merge head from ClickHouse, so `no_merge_record` is the honest label
+rather than guessing. Reverts and direct pushes (no merges row) also land here.
+
+A real `-f` force merge, by contrast, DOES write a `default.merges` row (with
+`skip_mandatory_checks` set), so it resolves a head and its test status is queried
+normally; `NOT_RUN:force_merge` is reported only when the gate was bypassed AND the
+test genuinely did not run.
+
+## Notes on the flaky scan
+
+Flaky signals are read from the `misc.autorevert_state` JSON snapshots. Several
+independent autorevert configurations run concurrently, each publishing its own
+snapshot stream distinguished by its `workflows` set. The flaky query scans **all**
+autorevert state snapshots in the range (exhaustive — every snapshot, deduped in
+Python), so the `flaky` rows reflect every (workflow, commit, test) triple autorevert
+flagged flaky (both a passing and a failing run on that commit) while it was in the
+state window. Scanning every snapshot rather than only the day's latest is required
+because commits age out of autorevert's sliding state window mid-day, so a
+latest-only sample would miss flaky states that appeared earlier in the day.
+
+The query is run once per 6-hour chunk of the padded window (~4 chunks/day) and
+results are accumulated and deduped in Python. Cost note: each chunked query runs
+with capped parallelism and peaks at roughly 4 GiB server-side on the busiest
+observed days, taking a few seconds per chunk. Results are exhaustive (every
+snapshot in range, deduped), and very large ranges scale linearly in the number of
+chunks.
+
+DNS to the ClickHouse cloud host flaps intermittently and the shared cluster can
+return transient server errors (e.g. `MEMORY_LIMIT_EXCEEDED`), so each query is
+retried with exponential backoff on connection, name-resolution, and transient
+database errors. Genuine query bugs fail fast on the first attempt: the driver
+raises a bare `DatabaseError` for server errors with the ClickHouse numeric code in
+the message, so a deterministic code (syntax error, unknown table/column/function,
+type mismatch, bad arguments, access denied, etc.) is detected and not retried;
+unknown or transient codes default to being retried.

tools/flake-test-fail-autorevert/flake_test_fail_autorevert/__init__.py

@@ -0,0 +1 @@
+"""Read-only CLI that exports pytorch-auto-revert's published test decisions to CSV."""

tools/flake-test-fail-autorevert/flake_test_fail_autorevert/__main__.py

@@ -0,0 +1,186 @@
+import argparse
+import csv
+import logging
+import sys
+from datetime import date, datetime, timedelta
+from typing import Dict, List, Optional, Set, Tuple
+
+from dotenv import load_dotenv  # type: ignore[import-not-found]
+
+from .client import get_clickhouse_client
+from .logic import build_rows, COLUMNS, iter_time_chunks
+from .premerge import (
+    classify_with_context,
+    parse_pr_from_message,
+    PremergeContext,
+    resolve_premerge_context,
+)
+from .queries import (
+    fetch_advisor_verdicts,
+    fetch_commit_messages,
+    fetch_commit_times,
+    fetch_flaky_for_day,
+    fetch_regressions,
+)
+
+
+EVENT_PAD_DAYS = 2
+FLAKY_PAD_DAYS = 1
+FLAKY_CHUNK_HOURS = 6
+
+
+def parse_date(value: str) -> date:
+    try:
+        return datetime.strptime(value, "%Y-%m-%d").date()
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(
+            f"invalid date '{value}', expected YYYY-MM-DD"
+        ) from exc
+
+
+def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        prog="flake_test_fail_autorevert",
+        description=(
+            "Export pytorch-auto-revert's published test decisions (regression "
+            "reverts and flaky-flagged test signals) to CSV for a commit landing "
+            "range on main."
+        ),
+    )
+    parser.add_argument("--start", type=parse_date, required=True, help="YYYY-MM-DD")
+    parser.add_argument(
+        "--end", type=parse_date, required=True, help="YYYY-MM-DD (inclusive)"
+    )
+    parser.add_argument("--repo", default="pytorch/pytorch")
+    parser.add_argument("--output", default=None)
+    args = parser.parse_args(argv)
+    if args.start > args.end:
+        parser.error(f"--start ({args.start}) must be <= --end ({args.end})")
+    return args
+
+
+def default_output(start: date, end: date) -> str:
+    return f"flake_test_fail_autorevert_{start.isoformat()}_{end.isoformat()}.csv"
+
+
+def write_csv(path: str, rows: List[dict]) -> None:
+    with open(path, "w", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=COLUMNS, quoting=csv.QUOTE_MINIMAL)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def collect(args: argparse.Namespace) -> List[dict]:
+    client = get_clickhouse_client()
+
+    ev_start = datetime.combine(
+        args.start - timedelta(days=EVENT_PAD_DAYS), datetime.min.time()
+    )
+    ev_end = datetime.combine(
+        args.end + timedelta(days=EVENT_PAD_DAYS + 1), datetime.min.time()
+    )
+    regressions = fetch_regressions(client, args.repo, ev_start, ev_end)
+
+    flaky: Dict[str, Set[Tuple[str, str]]] = {}
+    flaky_start = args.start - timedelta(days=FLAKY_PAD_DAYS)
+    flaky_end = args.end + timedelta(days=FLAKY_PAD_DAYS)
+    for chunk_start, chunk_end in iter_time_chunks(
+        flaky_start, flaky_end, FLAKY_CHUNK_HOURS
+    ):
+        for workflow, signal_key, commit_sha in fetch_flaky_for_day(
+            client, args.repo, chunk_start, chunk_end
+        ):
+            flaky.setdefault(commit_sha, set()).add((workflow, signal_key))
+        n_pairs = sum(len(v) for v in flaky.values())
+        logging.info(
+            "flaky scan %s: %d distinct (workflow, signal) pairs so far",
+            chunk_start.isoformat(),
+            n_pairs,
+        )
+
+    candidate_shas = sorted(set(regressions.by_commit) | set(flaky))
+    commit_times = fetch_commit_times(client, candidate_shas)
+
+    regression_shas = sorted(regressions.by_commit)
+    verdicts = fetch_advisor_verdicts(client, args.repo, regression_shas)
+
+    rows = build_rows(
+        regressions.by_commit,
+        regressions.single_workflow,
+        flaky,
+        commit_times,
+        verdicts,
+        args.start,
+        args.end,
+        args.repo,
+    )
+
+    # Only trunk+pull regressions get a premerge lookup; everything else stays "".
+    qualifying = [
+        r
+        for r in rows
+        if r["category"] == "regression" and r["workflow"] in ("trunk", "pull")
+    ]
+    msg_shas = sorted({r["commit_sha"] for r in qualifying})
+    messages = fetch_commit_messages(client, msg_shas) if msg_shas else {}
+
+    # Initialize every row so csv.DictWriter always has the premerge_status field.
+    for r in rows:
+        r["premerge_status"] = ""
+
+    total = len(qualifying)
+    # head_sha/merge_ts/job_ids depend only on the commit, so resolve the per-commit
+    # context once and reuse it for every failing signal on that commit.
+    context_cache: Dict[str, PremergeContext] = {}
+    for i, r in enumerate(qualifying, start=1):
+        file, sep, name = r["signal_key"].partition("::")
+        if not sep:
+            continue
+        message = messages.get(r["commit_sha"], "")
+        pr = parse_pr_from_message(message)
+        context = context_cache.get(r["commit_sha"])
+        if context is None:
+            context = resolve_premerge_context(client, r["commit_sha"], repo=args.repo)
+            context_cache[r["commit_sha"]] = context
+        status = classify_with_context(client, context, file, name)
+        r["premerge_status"] = status
+        logging.info(
+            "premerge %d/%d commit=%s pr=%s signal=%s -> %s",
+            i,
+            total,
+            r["commit_sha"][:10],
+            pr,
+            r["signal_key"],
+            status,
+        )
+
+    return rows
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(message)s",
+        stream=sys.stderr,
+    )
+    load_dotenv()
+    args = parse_args(argv)
+
+    rows = collect(args)
+
+    output = args.output or default_output(args.start, args.end)
+    write_csv(output, rows)
+
+    n_reg = sum(1 for r in rows if r["category"] == "regression")
+    n_flaky = sum(1 for r in rows if r["category"] == "flaky")
+    n_commits = len({r["commit_sha"] for r in rows})
+    print(output)
+    print(
+        f"{len(rows)} rows across {n_commits} commits: "
+        f"{n_reg} regression, {n_flaky} flaky"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())