parity: retry a known flaky harness once, and say so

tools/parity_hosts.json's new `known_flaky` lists the harnesses that still flake for a cause not fixed yet,
each with its reason: dedicated_server (its Java client's 1.5 s step timer), remote_fireball (wall-clock
sampling) and world_reopen. One of them that fails runs once more from scratch, and a pass then counts.

Every retry prints a RETRIED line at the end of the run and in the CI job's summary, so a harness that
passes only on its retry stays visible on PR parity and parity-main alike. The existing fresh rerun of a
failure that reused oracle results is reported the same way. A known flaky harness skips that rerun and keeps
reusing oracle results, because flaking doesn't mean its cached oracle runs were wrong. Repeated runs
(--repeat, --flake-check) are never retried.

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

docs/parity-workflow.md

@@ -133,6 +133,13 @@ when named and in the nightly audit (`--all`). LAN writes `results.json`
 with the host mode and each result beside its screenshots and logs. A skipped
 case is recorded explicitly and does not count as passing coverage.
 
+`known_flaky` in `tools/parity_hosts.json` lists harnesses that fail now and then for a
+cause not fixed yet, each with its reason. One that fails runs once more from
+scratch, and a pass then counts. `parity.py` prints a `RETRIED` line for it
+in its summary and in the CI job's summary (PR parity and parity-main), so the
+flake stays visible. No other harness gets this retry. Fix the cause, then take
+the harness off the list.
+
 The game-binary fixture audit keeps these complementary direct rule tests:
 
 - `command_harness.rs`: command feedback, styled spans and result callbacks,

tests/test_parity_runner.py

@@ -290,9 +290,6 @@ class OracleReuseTests(unittest.TestCase):
         self.assertEqual(self.output.read_text(), "b")
 
 
-if __name__ == "__main__":
-    unittest.main()
-
 
 class LargeOutputTests(unittest.TestCase):
     def test_a_pass_clears_only_its_own_directory(self):
@@ -323,3 +320,84 @@ class LargeOutputTests(unittest.TestCase):
                 listing.write_text(f"{own / 'server.txt'}\n{own / 'client.txt'}\n")
                 parity.clear_large_outputs("state_provider#2", trace)
             self.assertEqual(sorted(path.name for path in own.iterdir()), ["report.json", "server.txt"])
+
+
+class FakeExecutor:
+    """Runs nothing: each start of a harness takes the next status from its script, a list."""
+    name, dead, max_peak = "fake", False, None
+
+    def __init__(self):
+        self.started, self.finished = [], []
+
+    def free(self):
+        return 1
+
+    def accepts(self, name):
+        return True
+
+    def fits(self, expected, only_here=True):
+        return True
+
+    def start(self, label, script, reuse, expected, shard=None):
+        self.started.append((label, reuse))
+        self.finished.append((label, {"status": script.pop(0), "seconds": 1.0, "dependencies": []}))
+        return True
+
+    def poll(self):
+        finished, self.finished = self.finished, []
+        return finished
+
+
+class KnownFlakyTests(unittest.TestCase):
+    def run_harnesses(self, scripts, flaky, rounds=None):
+        executor = FakeExecutor()
+        state = {"harnesses": {name: {"status": "pass", "oracle_reuse": True} for name in scripts}}
+        with tempfile.TemporaryDirectory() as tmp, mock.patch.multiple(parity, STATE_DIR=Path(tmp), SHARED=Path(tmp)), \
+                mock.patch.object(parity, "save_state"), mock.patch.object(parity, "time") as clock:
+            clock.monotonic.return_value = 0.0
+            results, _ = parity.run(scripts, state, None, "env", [executor], True, rounds, flaky)
+        return results, executor.started
+
+    def test_a_known_flaky_failure_runs_once_more_from_scratch_and_says_so(self):
+        results, started = self.run_harnesses({"dedicated_server": ["fail", "pass"]}, {"dedicated_server": "timer"})
+        self.assertEqual(results["dedicated_server"]["status"], "pass")
+        self.assertEqual(results["dedicated_server"]["retried"], "known flaky")
+        # The first run reused oracle results; a known flaky harness doesn't lose that by flaking.
+        self.assertTrue(results["dedicated_server"]["oracle_reuse"])
+        self.assertEqual(started, [("dedicated_server", True), ("dedicated_server", False)])
+
+    def test_only_once(self):
+        results, started = self.run_harnesses({"dedicated_server": ["fail", "fail"]}, {"dedicated_server": "timer"})
+        self.assertEqual(results["dedicated_server"]["status"], "fail")
+        self.assertEqual(len(started), 2)
+
+    def test_other_harnesses_keep_only_the_reused_oracle_retry(self):
+        results, started = self.run_harnesses({"food": ["fail", "pass"]}, {"dedicated_server": "timer"})
+        self.assertEqual((results["food"]["status"], results["food"]["retried"]), ("pass", "reused oracle results"))
+        self.assertEqual(started, [("food", True), ("food", False)])
+        self.assertFalse(results["food"]["oracle_reuse"])
+
+    def test_repeated_runs_get_no_retry(self):
+        results, started = self.run_harnesses({"dedicated_server": ["pass", "fail"]}, {"dedicated_server": "timer"},
+                                              rounds={"dedicated_server": 2})
+        self.assertEqual((results["dedicated_server"]["status"], len(started)), ("fail", 2))
+
+    def test_retries_reach_the_job_summary(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            summary = Path(tmp) / "summary.md"
+            with mock.patch.dict(os.environ, {"GITHUB_STEP_SUMMARY": str(summary)}), \
+                    mock.patch("builtins.print"):
+                parity.report_retries({"dedicated_server": {"status": "pass", "retried": "known flaky"},
+                                       "food": {"status": "pass"}}, {"dedicated_server": "timer"})
+            text = summary.read_text()
+        self.assertIn("RETRIED dedicated_server: failed, then passed on its one retry (known flaky: timer", text)
+        self.assertNotIn("food", text)
+
+    def test_the_known_flaky_list_has_a_reason_per_harness(self):
+        flaky = parity.known_flaky(parity.host_config())
+        self.assertTrue(all(isinstance(reason, str) and len(reason) > 20 for reason in flaky.values()))
+        self.assertTrue(set(flaky) <= {parity.base_name(name) for name in parity.harnesses()})
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/parity.py

@@ -26,6 +26,11 @@ change costs only the Rust side. A harness that fails while reusing oracle
 results is rerun fresh, and stops reusing them if it then passes. --all and
 --fresh never reuse them.
 
+A harness listed in parity_hosts.json's "known_flaky" (each with its reason)
+that fails runs once more from scratch instead, and a pass then counts. Every
+retry is printed as a RETRIED line in the summary and in the CI job's summary.
+Fix the cause, then take the harness off the list: nothing else gets this retry.
+
 Harness probes build into .local/parity-target with release optimisations but
 without LTO and with 16 codegen units (`PROFILE` below), which builds in
 seconds instead of minutes. Rust does not let those settings change results:
@@ -1014,11 +1019,26 @@ def harness_of(label):
     return label.split("~")[0]
 
 
-def run(selected, state, hasher, environment_digest, executors, reuse_oracles, rounds=None):
+def keep_first_log(label):
+    """Move a harness's log aside before its retry writes a new one; returns where it went."""
+    log = STATE_DIR / "logs" / f"{label}.log"
+    kept = log.with_name(f"{label}.first-run.log")
+    try:
+        log.replace(kept)
+    except OSError:
+        return log  # a remote host's run: its log stays on that machine
+    return kept
+
+
+def run(selected, state, hasher, environment_digest, executors, reuse_oracles, rounds=None, known_flaky=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."""
+    must agree, and a harness that both passes and fails is reported as flaky and recorded as failing.
+    known_flaky ({name: reason}, tools/parity_hosts.json): a harness listed there that fails its single round runs
+    once more, from scratch; its record keeps `retried` (the first run's log is kept beside its own)."""
     records = state["harnesses"]
     rounds = rounds or {}
+    known_flaky = known_flaky or {}
+    second_chance = set()  # known flaky harnesses that already failed once in this run
     # 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))
@@ -1041,7 +1061,9 @@ def run(selected, state, hasher, environment_digest, executors, reuse_oracles, r
                 if pick is None:
                     break
                 name = harness_of(pick)
-                reuse = reuse_oracles and records.get(name, {}).get("oracle_reuse", True) and pick not in retried
+                # A retry runs the oracle again too: it may be the side that failed (mob_ai's heap).
+                reuse = reuse_oracles and records.get(name, {}).get("oracle_reuse", True) and pick not in retried \
+                    and name not in second_chance
                 if not executor.start(pick, selected[name], reuse, records.get(name, {}).get("peak_bytes", DEFAULT_MEMORY),
                                       SHARDS.get(name)):
                     break
@@ -1066,9 +1088,12 @@ def run(selected, state, hasher, environment_digest, executors, reuse_oracles, r
                 queue.insert(0, label)
                 continue
             passed = record["status"] == "pass"
-            if not passed and reused and label == name and records.get(name, {}).get("status") != "fail":
+            flaky_retry = rounds.get(name, 1) == 1 and base_name(name) in known_flaky
+            if not passed and reused and label == name and records.get(name, {}).get("status") != "fail" \
+                    and not flaky_retry:
                 # 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).
+                # what the first just validated, so their failures count (a flaky harness). A known flaky harness
+                # gets its own retry below instead, which keeps reusing oracle results afterwards.
                 retried.add(label)
                 queue.insert(0, label)
                 print(f"retry {label} without reused oracle results", flush=True)
@@ -1086,8 +1111,19 @@ def run(selected, state, hasher, environment_digest, executors, reuse_oracles, r
             if len(done[name]) < rounds.get(name, 1):
                 continue
             runs = done.pop(name)
+            if runs[0]["status"] != "pass" and flaky_retry and name not in second_chance:
+                second_chance.add(name)
+                first_log = keep_first_log(label)
+                print(f"RETRIED {name}: failed, and it is a known flaky harness ({known_flaky[base_name(name)]}): "
+                      f"running it once more (first run's log: {first_log})", flush=True)
+                queue.insert(0, label)
+                continue
             record = dict(runs[0], status="pass" if all(run["status"] == "pass" for run in runs) else "fail",
                           rounds=len(runs))
+            if name in second_chance:
+                record["retried"] = "known flaky"
+            elif label in retried:
+                record["retried"] = "reused oracle results"
             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)
@@ -1309,12 +1345,36 @@ def platform_skips(config):
     return dict(config.get("linux_only", {}))
 
 
+def known_flaky(config):
+    """Harnesses tools/parity_hosts.json lists as failing now and then for a cause not fixed yet: one that fails runs
+    once more, and passing then counts, reported as RETRIED. {name: reason}"""
+    return dict(config.get("known_flaky", {}))
+
+
 def named_only(config):
     """Harnesses tools/parity_hosts.json keeps out of every automatic selection (--changed, --failed, the default,
     --flake-check): they run only when named or with --all, the nightly audit. {name: reason}"""
     return dict(config.get("named_only", {}))
 
 
+def report_retries(results, flaky):
+    """Say loudly which harnesses needed their one retry (a known flaky one, or one that failed with reused oracle
+    results), here and in the CI job's summary: a pass on the retry counts, so this is where flakes stay visible."""
+    def why(name, record):
+        if record["retried"] == "known flaky":
+            return (f"known flaky: {flaky.get(base_name(name), '?')}; first run's log: "
+                    f"{STATE_DIR / 'logs' / (name + '.first-run.log')}")
+        return "its first run reused oracle results; it no longer reuses them" if record["status"] == "pass" else \
+            "its first run reused oracle results"
+    lines = [f"RETRIED {name}: failed, then {'passed' if record['status'] == 'pass' else 'failed again'} on its one "
+             f"retry ({why(name, record)})" for name, record in sorted(results.items()) if record.get("retried")]
+    for line in lines:
+        print(line, flush=True)
+    if lines and os.environ.get("GITHUB_STEP_SUMMARY"):
+        with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as summary:
+            summary.write("### Harnesses that failed once, then ran again\n" + "".join(f"- {line}\n" for line in lines))
+
+
 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)
@@ -1463,7 +1523,8 @@ 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, rounds)
+        results, elapsed = run(selected, state, hasher, environment_key(hasher), executors, reuse_oracles, rounds,
+                               known_flaky(config))
     finally:
         for executor in remotes:
             executor.stop()
@@ -1471,6 +1532,7 @@ def main():
     failed = sorted(name for name, record in results.items() if record["status"] != "pass")
     finished_run(failed)
     print(f"\n{len(results) - len(failed)} passed, {len(failed)} failed in {elapsed / 60:.1f} min.")
+    report_retries(results, known_flaky(config))
     for index, name in enumerate(failed):
         log = STATE_DIR / 'logs' / (name + '.log')
         print(f"  {name}: {log}")

tools/parity_hosts.json

@@ -42,5 +42,10 @@
  "linux_only": {},
  "named_only": {
   "lan_integration": "the whole lan_e2e suite: ~40 min a run on a loaded runner, and single checks flake under load; run it by name, or tools/lan_e2e.py"
+ },
+ "known_flaky": {
+  "dedicated_server": "its Java client moves to the next step on a 1.5 s timer, not on the server's reply: under load a respawn's add_entity/update_attributes or entity_event 19 lands in the wrong step (failed, then passed at the same head, 10-26)",
+  "remote_fireball": "samples the fireball on the wall clock, so a slow host sees it in too few samples (desktop runner, 3 times on 10-26)",
+  "world_reopen": "counted 4 minecarts instead of 3 once on a loaded Mac mini runner, then passed at the same head (10-26)"
  }
 }