Catch flaky harnesses: --repeat, and three runs in the gate

parity.py --repeat N runs each harness N times; its runs must agree, else it
is reported FLAKY and recorded as failing. Later rounds run on the machine
that ran the first, reusing its oracle results, and never two at once there.
With --flake-check REF (passed by gate.py, merge base with parity-base; by
supervise.py integrate, the integration branch before the merge), harnesses
whose script, oracle or probe is new or changed must pass three runs in a
row; a pass records its rounds so it is not repeated again. gate.py matches
shards of a known failure by the harness's name.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

tools/gate.py

@@ -8,7 +8,9 @@
 
 Runs, in order, stopping at the first failure: the public-tree check,
 rustfmt, clippy with warnings denied, the Rust tests, the Python tests, and
-then every vanilla harness whose inputs changed (tools/parity.py). Harnesses
+then every vanilla harness whose inputs changed (tools/parity.py); those whose
+script, oracle or probe is new or changed since the merge base with
+--flake-base must pass 3 runs in a row, which catches most flaky ones. Harnesses
 listed with --known-failures may still fail (they fail on the integration
 branch too); any other failing harness fails the gate. CI runs the
 same Rust and Python checks; the harnesses are the part CI cannot run.
@@ -41,7 +43,10 @@ def steps(args):
     yield "doc tests", ["cargo", "+stable", "test", "--locked", "--workspace", "--doc"]
     yield "python tests", [sys.executable, "-m", "unittest", "discover", "-s", "tests", "-q"]
     if not args.no_parity:
-        yield "vanilla harnesses", [sys.executable, "tools/parity.py", *(["-j", str(args.jobs)] if args.jobs is not None else [])]
+        command = [sys.executable, "tools/parity.py", *(["-j", str(args.jobs)] if args.jobs is not None else [])]
+        if args.flake_base:
+            command += ["--flake-check", args.flake_base]
+        yield "vanilla harnesses", command
 
 
 def head():
@@ -54,6 +59,8 @@ def main():
     parser.add_argument("--quick", action="store_true", help="only formatting, lints and Python tests")
     parser.add_argument("--known-failures", default="", help="comma-separated harnesses allowed to keep failing")
     parser.add_argument("-j", "--jobs", type=int, help="harnesses at once on this machine (default: tools/parity.py's)")
+    parser.add_argument("--flake-base", default="parity-base", help="harnesses new or changed since the merge base "
+                        "with this ref must pass 3 runs in a row ('' to skip; default: parity-base)")
     args = parser.parse_args()
     logs = ROOT / ".local" / "gate"
     logs.mkdir(parents=True, exist_ok=True)
@@ -76,7 +83,7 @@ def main():
             else:
                 known = set(filter(None, args.known_failures.split(",")))
                 failing = set(finished["failed"])
-                new = sorted(failing - known)
+                new = sorted(name for name in failing if name not in known and name.split("#")[0] not in known)  # shards
                 if failing:
                     print(f"Failing harnesses: {', '.join(sorted(failing))}; new: {', '.join(new) or 'none'}")
                 status = 1 if new else 0

tools/parity.py

@@ -859,28 +859,45 @@ def key_from(record, hashes, environment):
     return key(record, lambda path: hashes.get(str(Path(path).relative_to(ROOT)), "missing"), environment)
 
 
-def run(selected, state, hasher, environment_digest, executors, reuse_oracles):
+def harness_of(label):
+    """The harness a queue entry runs: food~2 is the second round of food."""
+    return label.split("~")[0]
+
+
+def run(selected, state, hasher, environment_digest, executors, reuse_oracles, rounds=None):
+    """Run the selected harnesses on the executors. rounds: how many times each must run (default once); its runs
+    must agree, and a harness that both passes and fails is reported as flaky and recorded as failing."""
     records = state["harnesses"]
-    # Longest first, so the slow harnesses don't start last.
-    queue = sorted(selected, key=lambda name: -records.get(name, {}).get("seconds", 1e9))
-    results, retried, busy = {}, set(), {}
+    rounds = rounds or {}
+    # Longest first, so the slow harnesses don't start last; a harness's later rounds after its first.
+    queue = sorted((name if index == 1 else f"{name}~{index}" for name in selected for index in range(1, rounds.get(name, 1) + 1)),
+                   key=lambda label: (-records.get(harness_of(label), {}).get("seconds", 1e9), label))
+    results, retried, busy, done = {}, set(), {}, {}
+    home = {}  # where a harness's first round ran: its later rounds run there too, reusing its oracle results
     started_all = time.monotonic()
     local = executors[-1]
     while queue or busy:
         # Remote machines first: they take the longest harnesses and leave this one lighter.
         for executor in executors:
             while queue and executor.free() > 0:
-                pick = next((name for name in queue if executor.accepts(name) and executor.fits(
-                    records.get(name, {}).get("peak_bytes", DEFAULT_MEMORY),
-                    not any(other.accepts(name) and not other.dead for other in executors[:-1]))), None)
+                # Rounds of one harness share its output directories: never two at once on one machine.
+                here = {harness_of(label) for label, (other, _) in busy.items() if other is executor}
+                pick = next((label for label in queue if harness_of(label) not in here and executor.accepts(harness_of(label))
+                             and (label == harness_of(label) or home.get(harness_of(label), executor) in (executor, None)
+                                  or home[harness_of(label)].dead)
+                             and executor.fits(records.get(harness_of(label), {}).get("peak_bytes", DEFAULT_MEMORY),
+                                               not any(other.accepts(harness_of(label)) and not other.dead
+                                                       for other in executors[:-1]))), None)
                 if pick is None:
                     break
-                reuse = reuse_oracles and records.get(pick, {}).get("oracle_reuse", True) and pick not in retried
-                if not executor.start(pick, selected[pick], reuse, records.get(pick, {}).get("peak_bytes", DEFAULT_MEMORY),
-                                      SHARDS.get(pick)):
+                name = harness_of(pick)
+                reuse = reuse_oracles and records.get(name, {}).get("oracle_reuse", True) and pick not in retried
+                if not executor.start(pick, selected[name], reuse, records.get(name, {}).get("peak_bytes", DEFAULT_MEMORY),
+                                      SHARDS.get(name)):
                     break
                 queue.remove(pick)
                 busy[pick] = (executor, reuse)
+                home.setdefault(name, executor)
         if executors[:-1] and all(executor.dead for executor in executors[:-1]) and local.max_peak is not None:
             # Every other machine is gone: this one runs everything.
             local.max_peak, local.jobs = None, max(local.jobs, default_local_jobs())
@@ -888,35 +905,46 @@ def run(selected, state, hasher, environment_digest, executors, reuse_oracles):
         finished = []
         for executor in executors:
             if isinstance(executor, LocalExecutor):
-                for name, script, passed, elapsed, trace in executor.poll():
-                    finished.append((name, outcome(name, script, passed, elapsed, trace)))
+                for label, script, passed, elapsed, trace in executor.poll():
+                    finished.append((label, outcome(label, script, passed, elapsed, trace)))
             else:
                 finished += executor.poll()
-        for name, record in finished:
-            executor, reused = busy.pop(name)
+        for label, record in finished:
+            executor, reused = busy.pop(label)
+            name = harness_of(label)
             if record is None:
-                queue.insert(0, name)
+                queue.insert(0, label)
                 continue
             passed = record["status"] == "pass"
-            if not passed and reused and records.get(name, {}).get("status") != "fail":
-                # A new failure might come from a reused oracle result: run it again from scratch.
-                retried.add(name)
-                queue.insert(0, name)
-                print(f"retry {name} without reused oracle results", flush=True)
+            if not passed and reused and label == name and records.get(name, {}).get("status") != "fail":
+                # A new failure might come from a reused oracle result: run it again from scratch. Later rounds reuse
+                # what the first just validated, so their failures count (a flaky harness).
+                retried.add(label)
+                queue.insert(0, label)
+                print(f"retry {label} without reused oracle results", flush=True)
                 continue
             # A harness that only passes fresh no longer reuses oracle results.
-            record["oracle_reuse"] = records.get(name, {}).get("oracle_reuse", True) and not (passed and name in retried)
+            record["oracle_reuse"] = records.get(name, {}).get("oracle_reuse", True) and not (passed and label in retried)
             hashes = record.pop("hashes", None)
             if record["dependencies"]:
                 record["key"] = key_from(record, hashes, environment_digest) if hashes is not None else \
                     key(record, hasher, environment_digest)
+            where = "" if executor is local else f" on {executor.name}"
+            tail = "" if passed else f"  ({STATE_DIR / 'logs' / (label + '.log')})"
+            print(f"{'PASS' if passed else 'FAIL'} {label} {record['seconds']:.0f}s{where}{tail}", flush=True)
+            done.setdefault(name, []).append(record)
+            if len(done[name]) < rounds.get(name, 1):
+                continue
+            runs = done.pop(name)
+            record = dict(runs[0], status="pass" if all(run["status"] == "pass" for run in runs) else "fail",
+                          rounds=len(runs))
+            if len({run["status"] for run in runs}) > 1:
+                record["flaky"] = [run["status"] for run in runs]
+                print(f"FLAKY {name}: {', '.join(record['flaky'])} in {len(runs)} runs", flush=True)
             records[name] = record
             results[name] = record
             save_state(state)
             publish(name, record)
-            where = "" if executor is local else f" on {executor.name}"
-            tail = "" if passed else f"  ({STATE_DIR / 'logs' / (name + '.log')})"
-            print(f"{'PASS' if passed else 'FAIL'} {name} {record['seconds']:.0f}s{where}{tail}", flush=True)
     return results, time.monotonic() - started_all
 
 
@@ -1030,6 +1058,27 @@ def trace_main(output, script, arguments):
     runpy.run_path(script, run_name="__main__")
 
 
+def changed_harnesses(found, records, ref):
+    """Harnesses new or changed since the merge base of HEAD and ref: their script, or a dependency of theirs that is
+    an oracle (Java under mods/horizon/src/test) or a probe (a crate's example or binary), changed or was added."""
+    def git(*arguments):
+        result = subprocess.run(["git", *arguments], cwd=ROOT, capture_output=True, text=True)
+        return result.stdout.split("\n") if result.returncode == 0 else None
+    base = git("merge-base", "HEAD", ref)
+    if not base or not base[0]:
+        print(f"note: no merge base with {ref}; the flake check repeats nothing", flush=True)
+        return set()
+    changed = set(filter(None, git("diff", "--name-only", base[0]) or [])) | \
+        set(filter(None, git("ls-files", "--others", "--exclude-standard") or []))
+    found_changed = set()
+    for name, script in found.items():
+        specific = [path for path in records.get(name, {}).get("dependencies", [])
+                    if path.startswith("mods/horizon/src/test/") or "/examples/" in path or "/src/bin/" in path]
+        if str(Path(script).relative_to(ROOT)) in changed or any(path in changed for path in specific):
+            found_changed.add(name)
+    return found_changed
+
+
 def finished_run(failed):
     """Record that this run completed, for tools/gate.py: a crashed runner never writes it."""
     STATE_DIR.mkdir(parents=True, exist_ok=True)
@@ -1055,6 +1104,11 @@ def main():
                         help="comma-separated machines from tools/parity_hosts.json to run harnesses on as well "
                              "(default: PARITY_HOSTS, else the file's default; '' for this machine only)")
     parser.add_argument("--timeout", type=int, default=3600, help="seconds per harness")
+    parser.add_argument("--repeat", type=int, default=1, help="run each selected harness this many times; its runs "
+                        "must agree (oracle results are reused after the first unless --fresh)")
+    parser.add_argument("--flake-check", metavar="REF", help="harnesses new or changed since the merge base with REF "
+                        "must pass --flake-rounds times in a row (tools/gate.py)")
+    parser.add_argument("--flake-rounds", type=int, default=3, help="runs required by --flake-check (default 3)")
     args = parser.parse_args()
 
     found = harnesses()
@@ -1084,6 +1138,17 @@ def main():
         chosen = [name for name in found if records.get(name, {}).get("status") == "fail"]
     else:
         chosen = [name for name in found if stale(name, state, hasher, environment)]
+    rounds = {name: args.repeat for name in chosen} if args.repeat > 1 else {}
+    if args.flake_check:
+        changed = changed_harnesses(found, records, args.flake_check)
+        for name in sorted(changed):
+            # A pass counts for a changed harness only if it held over enough runs.
+            if name not in chosen and records.get(name, {}).get("rounds", 1) < args.flake_rounds:
+                chosen.append(name)
+            if name in chosen:
+                rounds[name] = max(rounds.get(name, 1), args.flake_rounds)
+        if rounds:
+            print(f"Flake check: {', '.join(sorted(rounds))} must pass {args.flake_rounds} times.", flush=True)
     selected = {name: found[name] for name in chosen}
     if args.list or not selected:
         for name in selected:
@@ -1128,7 +1193,7 @@ def main():
         reuse_oracles = not (args.fresh or args.all)
         executors = [*remotes, LocalExecutor(jobs, int(memory_available() * 0.75), args.timeout,
                                              REMOTE_LOCAL_PEAK if remotes and args.jobs is None else None)]
-        results, elapsed = run(selected, state, hasher, environment_key(hasher), executors, reuse_oracles)
+        results, elapsed = run(selected, state, hasher, environment_key(hasher), executors, reuse_oracles, rounds)
     finally:
         for executor in remotes:
             executor.stop()

tools/supervise.py

@@ -254,7 +254,8 @@ def integrate_once(args, branches):
     if git("rev-parse", "HEAD", cwd=base_tree) == before:
         print(f"{names}: nothing new for {args.base}.")
         return []
-    gate = [sys.executable, "tools/gate.py", "--known-failures", ",".join(known)]
+    # The flake check covers what these branches bring: harnesses changed since the current integration branch.
+    gate = [sys.executable, "tools/gate.py", "--known-failures", ",".join(known), "--flake-base", before]
     gate += ["-j", str(args.jobs)] if args.jobs is not None else []
     gate += ["--no-parity"] if args.no_parity else []
     if subprocess.run(gate, cwd=base_tree).returncode: