app/refresh.py

"""Orchestrate the full triage:
  1. fetch_last_n(7)                       → daily reports
  2. integration-test filter + ≥5/7        → persistent failures
  3. classify failure modes
  4. attach historical first_failure_day   → cluster by regression-day
  5. join CI bisect attribution            → pinned cluster(s)
  6. render index.html

Outputs:
  output/index.html    — static report
  output/state.json    — machine-readable triage state (for the dataset push)
"""

from __future__ import annotations

import argparse
import datetime
import json
import os
import sys
from collections import Counter, defaultdict

# repo-local imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from classify import classify  # noqa: E402
from cluster import cluster_failures, is_big_model, BIG_MODEL_SKIP  # noqa: E402
from fetch import fetch_last_n  # noqa: E402
from filter import per_day_integration_failures, intersect_across_days  # noqa: E402
from history import attach_history  # noqa: E402
from persist import save  # noqa: E402
from render import render  # noqa: E402


def main(argv=None):
    p = argparse.ArgumentParser()
    p.add_argument("--cache-dir", default="/tmp/itf_cache")
    p.add_argument("--out-dir", default="/home/arthur/integration-failure-triage/output")
    p.add_argument("--window", type=int, default=7)
    p.add_argument("--min-days", type=int, default=5)
    p.add_argument("--history-days", type=int, default=90,
                   help="how far back to walk daily reports for first-failure dates")
    p.add_argument("--no-history", action="store_true", help="skip the slow history sweep")
    args = p.parse_args(argv)

    os.makedirs(args.out_dir, exist_ok=True)

    print(f"[1/5] Fetching last {args.window} daily CI reports…", flush=True)
    daily = fetch_last_n(args.window, cache_dir=args.cache_dir)
    dates_window = sorted(daily.keys())
    print(f"      dates {dates_window[0]} → {dates_window[-1]}", flush=True)

    print(f"[2/5] Filter to IntegrationTest + ≥{args.min_days}/{args.window} days…", flush=True)
    per_day = per_day_integration_failures(daily)
    kept = intersect_across_days(per_day, min_days=args.min_days)
    print(f"      {len(kept)} persistent integration-test failures", flush=True)

    if not args.no_history:
        print(f"[3/5] Historical sweep ({args.history_days} days)…", flush=True)
        kept = attach_history(kept, max_days=args.history_days, cache_dir=args.cache_dir)
        # Bucket by first_failure_day
        by_day: dict[str, list[dict]] = defaultdict(list)
        for f in kept:
            by_day[f.get("first_failure_day") or "unknown"].append(f)
        print(f"      regression-day buckets (top 10 by size):", flush=True)
        for d, items in sorted(by_day.items(), key=lambda kv: -len(kv[1]))[:10]:
            print(f"        {d}: {len(items)} failures", flush=True)
    else:
        print("[3/5] skipping history sweep", flush=True)

    print("[4/5] Cluster with CI bisect attribution…", flush=True)
    nf_latest = daily[max(daily)].get("new_failures")
    report = cluster_failures(kept, nf_latest, classify)

    # Add `big_model_skip` flag and `first_failure_day` to every failure (carried
    # through render).
    def _mark(f):
        f["big_model"] = is_big_model(f)
        return f
    for c in report["clusters"].values():
        c["failures"] = [_mark(f) for f in c["failures"]]
    report["flaky"] = [_mark(f) for f in report["flaky"]]
    report["unpinned"] = [_mark(f) for f in report["unpinned"]]

    # Report-level derived stats
    report["regression_day_buckets"] = dict(
        sorted(
            Counter(
                (f.get("first_failure_day") or "unknown")
                for f in (
                    [g for c in report["clusters"].values() for g in c["failures"]]
                    + report["flaky"]
                    + report["unpinned"]
                )
            ).items(),
            key=lambda kv: -kv[1],
        )
    )
    report["window"] = {"dates": dates_window, "min_days": args.min_days}
    report["generated_at_utc"] = (
        datetime.datetime.now(datetime.UTC).replace(tzinfo=None).isoformat(timespec="seconds")
    )

    print("[5/5] Render HTML + persist state…", flush=True)
    html_str = render(
        report,
        generated_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None),
        dates_window=dates_window,
    )
    html_path = os.path.join(args.out_dir, "index.html")
    with open(html_path, "w") as f:
        f.write(html_str)
    # Mirror state into /app/output so it's served alongside the HTML and into
    # the bucket-backed /data path.
    state_path = os.path.join(args.out_dir, "state.json")
    with open(state_path, "w") as f:
        json.dump(report, f, indent=2, default=str)
    bucket_path = save(report)
    print(f"      wrote {html_path}  ({len(html_str)} bytes)")
    print(f"      wrote {state_path}")
    print(f"      wrote {bucket_path}  (bucket-backed)")
    print()
    t = report["totals"]
    print(f"SUMMARY  total={t['total']}  clusters={t['clusters']}  "
          f"in_clusters={t['in_clusters']}  flaky={t['flaky']}  unpinned={t['unpinned']}")


if __name__ == "__main__":
    main()