1 files changed, 110 insertions, 0 deletions
diff --git a/app/tests/test_backends_envs.py b/app/tests/test_backends_envs.py
index e38f009..78d033c 100644
--- a/app/tests/test_backends_envs.py
+++ b/app/tests/test_backends_envs.py
@@ -1,5 +1,7 @@
 """Tests for the managed Python environment (backends/envs.py)."""
 
+import contextlib
+import io
 import json
 import os
 import sys
@@ -204,6 +206,114 @@ class PipInstallTests(unittest.TestCase):
         self.assertNotIn("-U", calls[0])
 
 
+class PipInstallCorruptionRetryTests(unittest.TestCase):
+    """The corrupted-pip-cache self-heal: sniff markers, purge, retry once.
+
+    A wheel truncated inside pip's HTTP cache (~/.cache/pip) makes pip die
+    mid-unpack with zipfile.BadZipFile / "Bad CRC-32" — a generic exit 2
+    that otherwise just tells the user to install manually. The retry path
+    only triggers when those markers appear in pip's output.
+    """
+
+    CORRUPT = "zipfile.BadZipFile: Bad CRC-32 for file 'pynini.libs/x.so'"
+
+    def test_corrupted_cache_failure_purges_and_retries_once(self):
+        calls = []
+
+        def fake_run(argv, **kwargs):
+            calls.append(("install", list(argv)))
+            hook = kwargs.get("on_chunk")
+            if hook is not None:
+                hook(self.CORRUPT.encode("utf-8"))
+            return 2 if len(calls) == 1 else 0
+
+        def fake_quiet(argv, **kwargs):
+            calls.append(("purge", list(argv)))
+            return None
+
+        captured = io.StringIO()
+        with patch.object(envs, "env_exists", return_value=True), \
+                patch.object(envs.common, "run_console_subprocess",
+                             side_effect=fake_run), \
+                patch.object(envs.common, "run_console_subprocess_quiet",
+                             side_effect=fake_quiet), \
+                contextlib.redirect_stdout(captured):
+            rc = envs.pip_install(["sglang-omni"])
+        self.assertEqual(rc, 0)
+        # install (failed) -> purge -> install again, same argv.
+        self.assertEqual([kind for kind, _argv in calls],
+                         ["install", "purge", "install"])
+        self.assertEqual(calls[0][1], calls[2][1])
+        self.assertIn("cache", calls[1][1])
+        self.assertIn("purge", calls[1][1])
+        output = captured.getvalue()
+        self.assertIn("[WARNING]", output)
+        self.assertIn("retrying once", output)
+
+    def test_ordinary_failure_is_not_retried(self):
+        calls = []
+
+        def fake_run(argv, **kwargs):
+            calls.append(list(argv))
+            return 2
+
+        with patch.object(envs, "env_exists", return_value=True), \
+                patch.object(envs.common, "run_console_subprocess",
+                             side_effect=fake_run) as run, \
+                patch.object(envs.common, "run_console_subprocess_quiet") \
+                as quiet:
+            rc = envs.pip_install(["sglang-omni"])
+        self.assertEqual(rc, 2)
+        self.assertEqual(run.call_count, 1)
+        quiet.assert_not_called()
+
+    def test_retry_failure_returns_the_second_exit_code(self):
+        def fake_run(argv, **kwargs):
+            hook = kwargs.get("on_chunk")
+            if hook is not None:
+                hook(self.CORRUPT.encode("utf-8"))
+            return 2
+
+        def fake_quiet(argv, **kwargs):
+            return None
+
+        with patch.object(envs, "env_exists", return_value=True), \
+                patch.object(envs.common, "run_console_subprocess",
+                             side_effect=fake_run) as run, \
+                patch.object(envs.common, "run_console_subprocess_quiet",
+                             side_effect=fake_quiet):
+            rc = envs.pip_install(["sglang-omni"])
+        self.assertEqual(rc, 2)
+        self.assertEqual(run.call_count, 2)
+
+    def test_emit_path_sniffs_lines_and_forwards_them_unchanged(self):
+        lines = []
+        calls = []
+
+        def fake_run(argv, **kwargs):
+            calls.append(("install", list(argv)))
+            wrapped = kwargs.get("emit")
+            if wrapped is not None and len(calls) == 1:
+                wrapped(self.CORRUPT)
+            return 2 if len(calls) == 1 else 0
+
+        def fake_quiet(argv, **kwargs):
+            calls.append(("purge", list(argv)))
+            return None
+
+        with patch.object(envs, "env_exists", return_value=True), \
+                patch.object(envs.common, "run_console_subprocess",
+                             side_effect=fake_run), \
+                patch.object(envs.common, "run_console_subprocess_quiet",
+                             side_effect=fake_quiet):
+            rc = envs.pip_install(["sglang-omni"], emit=lines.append)
+        self.assertEqual(rc, 0)
+        self.assertEqual([kind for kind, _argv in calls],
+                         ["install", "purge", "install"])
+        # The task view still received the corruption line verbatim.
+        self.assertEqual(lines, [self.CORRUPT])
+
+
 class PipUninstallTests(unittest.TestCase):
     def test_missing_env_is_success_without_running_pip(self):
         with patch.object(envs, "env_exists", return_value=False), \