SocketDev · Mikola Lysenko (mikolalysenko) · Oct 1, 2026 · Oct 1, 2026 · Oct 1, 2026 · Oct 1, 2026
diff --git a/docs/testing/pipenv-compatibility.md b/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

diff --git a/scripts/backtest-pipenv.py b/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():

diff --git a/scripts/tests/test_backtest_harnesses.py b/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