Fix Pipenv matrix flake on 503/DNS transport blips (#419)

* Retry Pipenv cases on PyPI transport failures

The Pipenv compatibility matrix downloads real wheels from PyPI and
talks to the public patch service, but unlike the Bun, PDM and vlt
harnesses it had no transport retry. A single PyPI 503 burst ("too
many 503 error responses") turned a whole matrix leg red, and the
run had to be re-run by hand: 15 failed legs across 10 runs in the
last two weeks went green on re-run of the same commit.

Re-run a failed case from a fresh case directory, at most three
attempts, only when its error text or logs show a transport error
(pip giving up on PyPI, connection errors, a CLI request error or a
patch API 5xx). Functional failures are never retried. Earlier
attempts' logs are kept under attempts/ and listed on the row.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TrR5uqMbj9sJcU2Gpm9y9R

* Treat macOS DNS failures as Pipenv transport errors

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TrR5uqMbj9sJcU2Gpm9y9R

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: mikolalysenko

docs/testing/pipenv-compatibility.md

@@ -107,6 +107,13 @@ fixture dependency is `urllib3 1.26.18`, which has a public free-tier
 patch). Copy the binary out of `target/` first — a rebuild would swap it
 under the run. Concurrent invocations must use disjoint version/shape sets.
 
+A case that fails with a transport error in its error text or logs (pip
+giving up on PyPI, e.g. `too many 503 error responses`, a connection error,
+or a CLI request error / patch API 5xx) is re-run from a fresh case
+directory, at most three attempts in total. The failed attempts' logs stay
+under `attempts/<case>/<n>/` and the final row lists them in
+`transportRetries`. Functional failures are never retried.
+
 Per (release, shape, mode) the harness checks: the lock-only fresh checkout,
 `--dry-run` parity (hosted; the vendored preview is ledger-only by design and
 is recorded), an idempotent re-scan, the untouched Pipfile and `_meta`, the

scripts/backtest-pipenv.py

@@ -154,6 +154,52 @@ def save(path, data):
 
 DEFAULT_TIMEOUT = int(os.environ.get("BACKTEST_TIMEOUT", "900"))
 
+# A transport failure, never a functional one: pip/Pipenv giving up on PyPI
+# ("too many 503 error responses", connection errors) or the CLI's own report
+# of a request error or a patch API 5xx.
+TRANSPORT_FAILURE = re.compile(
+    r"too many 5\d\d error responses|Max retries exceeded with url|"
+    r"NewConnectionError|ConnectTimeoutError|ReadTimeoutError|"
+    r"Temporary failure in name resolution|nodename nor servname provided|Connection reset by peer|RemoteDisconnected|"
+    r"error sending request for url \(|API request failed with status 5\d\d\b"
+)
+
+
+def has_transport_failure(case, payload):
+    """Whether a failed case's error text or any of its logs shows a transport failure."""
+    if TRANSPORT_FAILURE.search(json.dumps(payload)):
+        return True
+    logs = sorted(case.glob("*.log*")) if case.is_dir() else []
+    return any(TRANSPORT_FAILURE.search(log.read_text(errors="replace")) for log in logs if log.is_file())
+
+
+def retry_transport(run_case, job, case, root, attempts=3, sleep=time.sleep):
+    """("row"|"error", payload) for one case, re-run from a clean case dir while
+    it fails for transport reasons. A failed attempt's logs are kept under
+    <root>/attempts/<case>/<n>/ and listed on the final payload."""
+    history = []
+    for attempt in range(1, attempts + 1):
+        try:
+            kind, payload = "row", run_case(job)
+        except Exception as e:
+            version, shape, mode, invocation = job
+            kind, payload = "error", {"pipenv": version, "shape": shape, "mode": mode, "invocation": invocation, "error": str(e)[-3000:], "trace": traceback.format_exc()[-1500:]}
+        failed = kind == "error" or not payload.get("passed")
+        if not failed or attempt == attempts or not has_transport_failure(case, payload):
+            if history:
+                payload["transportRetries"] = history
+                if case.is_dir():
+                    save(case / "result.json", payload)
+            return kind, payload
+        evidence = root / "attempts" / case.name / str(attempt)
+        evidence.mkdir(parents=True, exist_ok=True)
+        for log in case.glob("*.log*") if case.is_dir() else []:
+            if log.is_file():
+                shutil.copy2(log, evidence / log.name)
+        history.append({"attempt": attempt, "evidence": evidence.relative_to(root).as_posix(), "error": (payload.get("error") or "")[-300:], "failedChecks": [k for k, ok in payload.get("checks", {}).items() if not ok]})
+        print(f"{case.name}: transport failure; retrying fresh case ({attempt}/{attempts})", flush=True)
+        sleep(10 * attempt)
+
 
 class Run:
     """Run a command in its own process group, capture output, write a log.
@@ -680,11 +726,15 @@ def source_keys(text):
         return sorted({k for _, _, e in lock_entries(text) if isinstance(e, dict) for k in ("file", "path") if k in e})
 
     # -------------------------------------------------------------- one case
+    def case_dir(job):
+        version, shape, mode, invocation = job
+        suffix = "" if invocation == "in-dir" else "-" + invocation
+        return root / "captures" / f"{version}-{shape}-{mode}{suffix}"
+
     def backtest(job):
         """Run one case; persist its row (or error) as <case>/result.json."""
         version, shape, mode, invocation = job
-        suffix = "" if invocation == "in-dir" else "-" + invocation
-        case = root / "captures" / f"{version}-{shape}-{mode}{suffix}"
+        case = case_dir(job)
         try:
             row = backtest_case(job)
         except Exception as e:
@@ -699,8 +749,7 @@ def backtest_case(job):
         legacy = is_legacy(version)
         major = major_of(version)
         tool = tool_dir(version)
-        suffix = "" if invocation == "in-dir" else "-" + invocation
-        case = root / "captures" / f"{version}-{shape}-{mode}{suffix}"
+        case = case_dir(job)
         if case.exists():
             shutil.rmtree(case)
         case.mkdir(parents=True)
@@ -1277,11 +1326,8 @@ def run_group(key):
         out = []
         for m, inv in groups[key]:
             job = (v, s, m, inv)
-            try:
-                row = backtest(job)
-                out.append(("row", job, row))
-            except Exception as e:
-                out.append(("error", job, {"pipenv": v, "shape": s, "mode": m, "invocation": inv, "error": str(e)[-3000:], "trace": traceback.format_exc()[-1500:]}))
+            kind, payload = retry_transport(backtest, job, case_dir(job), root)
+            out.append((kind, job, payload))
             yield out[-1]
 
     def flush():

scripts/tests/test_backtest_harnesses.py

@@ -148,6 +148,82 @@ def test_checkout_drops_manifest_node_modules_and_old_output(self):
                                    'packages/c/package.json'])
 
 
+class PipenvTransportRetryTests(unittest.TestCase):
+    JOB = ("2023.12.1", "direct", "hosted", "in-dir")
+    PYPI_503 = ("pip._vendor.urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='files.pythonhosted.org', "
+                "port=443): Max retries exceeded with url: /packages/urllib3-1.26.18-py2.py3-none-any.whl "
+                "(Caused by ResponseError('too many 503 error responses'))")
+
+    def drive(self, outcomes, root):
+        """Drive retry_transport with one scripted outcome per attempt."""
+        case = root / "captures" / "2023.12.1-direct-hosted"
+        calls = []
+
+        def run_case(_job):
+            outcome = outcomes[len(calls)]
+            calls.append(outcome)
+            if case.exists():
+                pipenv.shutil.rmtree(case)
+            case.mkdir(parents=True)
+            (case / "install.log").write_text(outcome.get("log", ""))
+            if "raise" in outcome:
+                raise RuntimeError(outcome["raise"])
+            return {"passed": outcome["passed"], "checks": {"installedPatched": outcome["passed"]}}
+
+        sleeps = []
+        kind, payload = pipenv.retry_transport(run_case, self.JOB, case, root, sleep=sleeps.append)
+        return kind, payload, calls, sleeps
+
+    def test_pypi_503_error_is_retried_from_a_fresh_case_and_keeps_evidence(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            kind, payload, calls, sleeps = self.drive([{"raise": "pipenv install failed (exit 1):\n" + self.PYPI_503}, {"passed": True}], root)
+            self.assertEqual((kind, payload["passed"], len(calls), sleeps), ("row", True, 2, [10]))
+            retry = payload["transportRetries"][0]
+            self.assertIn("too many 503 error responses", retry["error"])
+            self.assertTrue((root / retry["evidence"] / "install.log").is_file())
+            saved = json.loads((root / "captures/2023.12.1-direct-hosted/result.json").read_text())
+            self.assertEqual(saved["transportRetries"], payload["transportRetries"])
+
+    def test_failed_check_with_transport_failure_in_a_log_is_retried(self):
+        with tempfile.TemporaryDirectory() as temp:
+            kind, payload, calls, _ = self.drive([{"passed": False, "log": self.PYPI_503}, {"passed": True}], Path(temp))
+            self.assertEqual((kind, payload["passed"], len(calls)), ("row", True, 2))
+            self.assertEqual(payload["transportRetries"][0]["failedChecks"], ["installedPatched"])
+
+    def test_patch_api_5xx_is_a_transport_failure(self):
+        with tempfile.TemporaryDirectory() as temp:
+            outcomes = [{"raise": "scan failed: API request failed with status 503: upstream connect error"}, {"passed": True}]
+            self.assertEqual(len(self.drive(outcomes, Path(temp))[2]), 2)
+
+    def test_macos_dns_failure_is_a_transport_failure(self):
+        with tempfile.TemporaryDirectory() as temp:
+            outcomes = [{"passed": False, "log": "NewConnectionError: [Errno 8] nodename nor servname provided, or not known"}, {"passed": True}]
+            self.assertEqual(len(self.drive(outcomes, Path(temp))[2]), 2)
+            self.assertTrue(pipenv.TRANSPORT_FAILURE.search("<urlopen error [Errno 8] nodename nor servname provided, or not known>"))
+
+    def test_functional_failures_are_never_retried(self):
+        with tempfile.TemporaryDirectory() as temp:
+            outcomes = [{"passed": False, "log": "ERROR: THESE PACKAGES DO NOT MATCH THE HASHES FROM THE REQUIREMENTS FILE"}]
+            kind, payload, calls, sleeps = self.drive(outcomes, Path(temp))
+            self.assertEqual((kind, payload["passed"], len(calls), sleeps), ("row", False, 1, []))
+            self.assertNotIn("transportRetries", payload)
+            kind, payload, calls, _ = self.drive([{"raise": "pipenv lock failed (exit 1): ResolutionFailure"}], Path(temp))
+            self.assertEqual((kind, len(calls)), ("error", 1))
+
+    def test_a_persistent_transport_failure_stays_red_after_three_attempts(self):
+        with tempfile.TemporaryDirectory() as temp:
+            kind, payload, calls, sleeps = self.drive([{"raise": self.PYPI_503}] * 3, Path(temp))
+            self.assertEqual((kind, len(calls), sleeps), ("error", 3, [10, 20]))
+            self.assertEqual([r["attempt"] for r in payload["transportRetries"]], [1, 2])
+
+    def test_a_passing_case_runs_once(self):
+        with tempfile.TemporaryDirectory() as temp:
+            kind, payload, calls, _ = self.drive([{"passed": True, "log": self.PYPI_503}], Path(temp))
+            self.assertEqual((kind, len(calls)), ("row", 1))
+            self.assertNotIn("transportRetries", payload)
+
+
 class PipenvShimTests(unittest.TestCase):
     def test_parallel_first_use(self):
         # Force every worker to reach symlink creation before any can create