CI recurrence study: Tier 0 fingerprinting, collector, and harness

studies/ci-recurrence/.gitignore

@@ -0,0 +1,4 @@
+cache/
+out/
+__pycache__/
+*.pyc

studies/ci-recurrence/README.md

@@ -0,0 +1,129 @@
+# CI failure recurrence study
+
+**Question.** Over 90 days on one repo, walk failure events chronologically. A
+failure is a RECURRENCE if its fingerprint appeared earlier in the window.
+`rate = recurrences / total_failures`.
+
+**Decision rule.** Below roughly 30% at the most defensible fingerprinting, the
+premise is dead.
+
+## Status
+
+Tier 0 and the collection/analysis path are built and tested. **No recurrence
+rate has been computed yet** — the session this was written in cannot reach the
+GitHub Actions API for `pytorch/pytorch` or `ankidroid/Anki-Android` (see
+`findings.md`). Run the two commands below on the study machine and the numbers
+drop out.
+
+The Tier 1 clustering loop is deliberately **not built yet**. It should not be
+written until the Tier 0 collapse rate says how many singletons it would have
+to chew through — that number decides whether Tier 1 is a night of local
+inference or a week of it.
+
+## Run it
+
+```bash
+export GITHUB_TOKEN=<classic PAT with NO scopes ticked>
+
+uv run fetch.py   --repo ankidroid/Anki-Android --since-days 90
+uv run analyze.py --repo ankidroid/Anki-Android --top 50
+```
+
+### What token
+
+**A classic PAT with zero scopes checked.** Both targets are public, and every
+endpoint this study touches is public-readable:
+
+| Endpoint | Needs |
+|---|---|
+| `/repos/{o}/{r}/actions/runs` | nothing, for a public repo |
+| `/repos/{o}/{r}/actions/runs/{id}/jobs` | nothing, for a public repo |
+| `/repos/{o}/{r}/check-runs/{id}/annotations` | nothing, for a public repo |
+
+The token is not there for access, it is there for **rate limit**:
+unauthenticated is 60 requests/hour, which makes a 90-day crawl impossible; any
+valid token raises that to 5,000/hour regardless of its scopes.
+
+Do **not** tick `public_repo`. That scope grants *write* to every public repo
+you can see — push, issues, the lot — and buys this study nothing. A scopeless
+token can only read public data, which is exactly the blast radius we want for
+something crawling two repos we don't own.
+
+A fine-grained PAT also works: set the resource owner to your own account and
+choose **"Public repositories (read-only)"**. Note you cannot scope a
+fine-grained PAT *to* `pytorch/pytorch` — fine-grained tokens only target repos
+you own, so the public-read option is the route. The scopeless classic token is
+simpler and no less safe.
+
+`fetch.py` caches every API response under `cache/` keyed by URL hash, so a
+re-run costs no API calls and the window can be rebuilt offline. `analyze.py`
+touches no network and no model — rerun it freely.
+
+For pytorch, scope the crawl or it will run for hours:
+
+```bash
+uv run fetch.py --repo pytorch/pytorch --since-days 90 \
+  --workflow trunk --workflow pull --max-runs 5000
+```
+
+Tests (no network, no model, no tokens):
+
+```bash
+uv run test_pipeline.py
+```
+
+## The cascade
+
+| Tier | Who | Handles | Cost |
+|---|---|---|---|
+| 0 | regex normalizer | everything with a matching fingerprint | free, reproducible, hashable |
+| 1 | local model via Ollama | Tier 0 singletons only | electricity |
+| 2 | Gemini Flash | low-confidence Tier 1, or when local is the bottleneck | metered |
+| 3 | coordinator | audit of 20 sampled labels per batch | attention |
+
+Tier 0 is the one that matters. An LLM label is not reproducible and not
+hashable; a regex fingerprint is both. Every event Tier 0 places in a group of
+2+ is an event no worker ever has to look at.
+
+## Files
+
+| File | What |
+|---|---|
+| `schema.sql` | SQLite: queue, cache, and checkpoint in one file |
+| `normalize.py` | **Tier 0.** Normalization rules + fingerprint |
+| `fetch.py` | GitHub Actions collector, disk-cached, read-only |
+| `analyze.py` | Flake split, Tier 0 collapse stats, chronological walk |
+| `prompts.py` | **Tier 1 worker prompt.** Version it on every edit |
+| `dispatch.py` | The whole harness: one `ask()`, content-hash cache, retry, tokens |
+| `test_pipeline.py` | Known-answer tests for the model-free parts |
+
+## Three decisions worth arguing with
+
+**Exit codes survive number-stripping.** `exit code 137` is an OOM kill and
+`exit code 143` is a timeout; `exit code 1` is a test failing. Stripping them as
+"bare ints" merges infrastructure failures into test failures and inflates the
+rate. `KEEP_EXIT_CODES` in `normalize.py` toggles it so the collapse rate can be
+reported both ways.
+
+**Path basenames survive, directory prefixes don't.** `test_nn.py` and
+`test_optim.py` are different failures; `/home/runner/work/...` vs
+`/opt/actions-runner/_work/...` is the same failure on a different machine. So
+absolute paths collapse to `<PATH>/test_nn.py`. Repo-relative paths from Check
+Run annotations are left fully intact — they are identical on every runner, so
+they are signal, not noise.
+
+**The worker is told to answer "different" when torn.** Merging is the
+destructive direction: every bad merge turns a first-seen failure into a
+recurrence and pushes the rate up, toward the 30% line we are testing against.
+The reported rate is therefore a lower bound, which is the only version worth
+betting on.
+
+## One correctness trap, documented so nobody re-introduces it
+
+You cannot find flakes by listing `status=failure` runs. When a failed run is
+re-run and passes, GitHub **rewrites the run's conclusion to `success`** — the
+flaky run vanishes from the failure list, taking the fail-then-pass evidence
+with it. So `fetch.py` lists all *completed* runs and pulls jobs for any run
+that is non-success **or** has `run_attempt > 1`, and records every job outcome
+including successes. Filtering on failure would silently drop the flakes and
+leave a rate that cannot be corrected after the fact.

studies/ci-recurrence/analyze.py

@@ -0,0 +1,193 @@
+"""Flake split, Tier 0 collapse stats, and the chronological recurrence walk.
+
+THE DEFINITION, verbatim from the brief:
+  Walk failure events chronologically over the window. A failure is a
+  RECURRENCE if its fingerprint appeared earlier in the window.
+  rate = recurrences / total_failures.
+
+Nothing here calls a model. Tier 0 numbers come out of this file alone, which
+is the point: they are reproducible and cost nothing.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import Counter
+
+import db
+
+
+# ----------------------------------------------------------------- flake split
+def mark_flakes(con, repo: str) -> dict[str, int]:
+    """Same head SHA, re-run, fail then pass => flake, not recurrence.
+
+    Two shapes, both requiring an empty diff (identical head SHA, so nothing
+    changed between the fail and the pass):
+      A. same run_id + job_name, a LATER attempt succeeded  (the classic re-run)
+      B. same head_sha + workflow + job_name, a later run succeeded
+    """
+    con.execute("UPDATE failure_event SET is_flake=0, flake_reason=NULL WHERE repo=?", (repo,))
+
+    # A: later successful attempt of the same job in the same run
+    con.execute(
+        """UPDATE failure_event SET is_flake=1, flake_reason='rerun_attempt_passed'
+           WHERE repo=?1 AND EXISTS (
+             SELECT 1 FROM job_outcome jo
+              WHERE jo.repo=?1 AND jo.run_id=failure_event.run_id
+                AND jo.job_name=failure_event.job_name
+                AND jo.conclusion='success'
+                AND jo.run_attempt > failure_event.run_attempt)""",
+        (repo,),
+    )
+    # B: later successful run on the identical head SHA
+    con.execute(
+        """UPDATE failure_event SET is_flake=1, flake_reason='same_sha_later_pass'
+           WHERE repo=?1 AND is_flake=0 AND EXISTS (
+             SELECT 1 FROM job_outcome jo
+              WHERE jo.repo=?1 AND jo.head_sha=failure_event.head_sha
+                AND jo.job_name=failure_event.job_name
+                AND IFNULL(jo.workflow,'')=IFNULL(failure_event.workflow,'')
+                AND jo.conclusion='success'
+                AND jo.started_at > failure_event.started_at)""",
+        (repo,),
+    )
+    con.commit()
+    rows = con.execute(
+        "SELECT IFNULL(flake_reason,'not_flake') r, COUNT(*) c FROM failure_event WHERE repo=? GROUP BY r",
+        (repo,),
+    ).fetchall()
+    return {r["r"]: r["c"] for r in rows}
+
+
+# ------------------------------------------------------------- recurrence walk
+def walk(con, repo: str, *, key: str = "fingerprint", exclude_flakes: bool = False) -> dict:
+    """Chronological walk. Returns totals and the recurrence rate."""
+    sql = (
+        f"SELECT {key} AS k, started_at FROM failure_event "
+        "WHERE repo=? AND fingerprint != '' "
+        + ("AND is_flake=0 " if exclude_flakes else "")
+        + "ORDER BY started_at ASC, job_id ASC"
+    )
+    seen: set[str] = set()
+    total = recurrences = 0
+    for row in con.execute(sql, (repo,)):
+        k = row["k"]
+        if not k:
+            continue
+        total += 1
+        if k in seen:
+            recurrences += 1
+        else:
+            seen.add(k)
+    return {
+        "total_failures": total,
+        "recurrences": recurrences,
+        "distinct": len(seen),
+        "rate": (recurrences / total) if total else 0.0,
+    }
+
+
+# -------------------------------------------------------------- tier 0 metrics
+def tier0_stats(con, repo: str) -> dict:
+    total = con.execute("SELECT COUNT(*) c FROM failure_event WHERE repo=?", (repo,)).fetchone()["c"]
+    with_ev = con.execute(
+        "SELECT COUNT(*) c FROM failure_event WHERE repo=? AND fingerprint != ''", (repo,)
+    ).fetchone()["c"]
+    by_kind = {
+        r["evidence_kind"]: r["c"]
+        for r in con.execute(
+            "SELECT evidence_kind, COUNT(*) c FROM failure_event WHERE repo=? GROUP BY evidence_kind",
+            (repo,),
+        )
+    }
+    fps = [
+        r["c"]
+        for r in con.execute(
+            "SELECT fingerprint, COUNT(*) c FROM failure_event "
+            "WHERE repo=? AND fingerprint != '' GROUP BY fingerprint",
+            (repo,),
+        )
+    ]
+    distinct = len(fps)
+    singletons = sum(1 for c in fps if c == 1)
+    clustered_events = sum(c for c in fps if c > 1)
+    return {
+        "failure_events": total,
+        "events_with_evidence": with_ev,
+        "log_coverage": (with_ev / total) if total else 0.0,
+        "evidence_kinds": by_kind,
+        "distinct_fingerprints": distinct,
+        # Fraction of events that Tier 0 placed in a group of 2+. This is the
+        # "Tier 0 handled it" number: those events never need a worker token.
+        "tier0_collapse": (clustered_events / with_ev) if with_ev else 0.0,
+        "singleton_fingerprints": singletons,
+        # What Tier 1 would actually have to chew through.
+        "tier1_candidates": singletons,
+    }
+
+
+def top_fingerprints(con, repo: str, n: int = 50) -> list[dict]:
+    rows = con.execute(
+        """SELECT fingerprint, COUNT(*) c,
+                  MIN(started_at) first_seen, MAX(started_at) last_seen,
+                  MIN(norm_message) exemplar, MIN(evidence_kind) kind
+             FROM failure_event WHERE repo=? AND fingerprint != ''
+            GROUP BY fingerprint ORDER BY c DESC, first_seen ASC LIMIT ?""",
+        (repo, n),
+    ).fetchall()
+    return [dict(r) for r in rows]
+
+
+def report(repo: str, out_json: str | None = None) -> dict:
+    con = db.connect()
+    flakes = mark_flakes(con, repo)
+    stats = tier0_stats(con, repo)
+    result = {
+        "repo": repo,
+        "window_start": db.meta_get(con, f"window_start:{repo}"),
+        "since_days": db.meta_get(con, f"since_days:{repo}", "90"),
+        "runs_scanned": db.meta_get(con, f"runs_scanned:{repo}", "0"),
+        "tier0": stats,
+        "flake_breakdown": flakes,
+        "recurrence": {
+            "tier0_with_flakes": walk(con, repo, key="fingerprint", exclude_flakes=False),
+            "tier0_without_flakes": walk(con, repo, key="fingerprint", exclude_flakes=True),
+        },
+        "top_fingerprints": top_fingerprints(con, repo, 50),
+    }
+    if out_json:
+        with open(out_json, "w") as f:
+            json.dump(result, f, indent=2)
+    return result
+
+
+def main() -> int:
+    p = argparse.ArgumentParser()
+    p.add_argument("--repo", required=True)
+    p.add_argument("--json", default=None)
+    p.add_argument("--top", type=int, default=50)
+    a = p.parse_args()
+    r = report(a.repo, a.json)
+    t, rec = r["tier0"], r["recurrence"]
+    print(f"repo                  {r['repo']}   window {r['since_days']}d from {r['window_start']}")
+    print(f"runs scanned          {r['runs_scanned']}")
+    print(f"failure events        {t['failure_events']}")
+    print(f"log coverage          {t['log_coverage']:.1%}  ({t['evidence_kinds']})")
+    print(f"distinct fingerprints {t['distinct_fingerprints']}")
+    print(f"TIER 0 COLLAPSE       {t['tier0_collapse']:.1%}")
+    print(f"tier 1 candidates     {t['tier1_candidates']} singletons")
+    print(f"flakes                {r['flake_breakdown']}")
+    print(f"RECURRENCE with flakes    {rec['tier0_with_flakes']['rate']:.1%} "
+          f"({rec['tier0_with_flakes']['recurrences']}/{rec['tier0_with_flakes']['total_failures']})")
+    print(f"RECURRENCE without flakes {rec['tier0_without_flakes']['rate']:.1%} "
+          f"({rec['tier0_without_flakes']['recurrences']}/{rec['tier0_without_flakes']['total_failures']})")
+    print()
+    print(f"--- top {a.top} Tier 0 fingerprints ---")
+    for i, fp in enumerate(r["top_fingerprints"][: a.top], 1):
+        print(f"{i:3}. {fp['fingerprint']}  n={fp['c']:<5} {fp['kind']:<10} {fp['exemplar'][:110]}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

studies/ci-recurrence/db.py

@@ -0,0 +1,32 @@
+"""SQLite handle + tiny helpers. SQLite is the queue, cache, and checkpoint."""
+
+from __future__ import annotations
+
+import pathlib
+import sqlite3
+from datetime import datetime, timezone
+
+ROOT = pathlib.Path(__file__).parent
+DB_PATH = ROOT / "out" / "study.db"
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def connect(path: pathlib.Path | str = DB_PATH) -> sqlite3.Connection:
+    path = pathlib.Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    con = sqlite3.connect(path)
+    con.row_factory = sqlite3.Row
+    con.executescript((ROOT / "schema.sql").read_text())
+    return con
+
+
+def meta_set(con: sqlite3.Connection, key: str, value: str) -> None:
+    con.execute("INSERT INTO meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value))
+
+
+def meta_get(con: sqlite3.Connection, key: str, default: str = "") -> str:
+    row = con.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone()
+    return row["value"] if row else default