feat: pip install checks for bad crc errors and purges cache automatically

2 files changed, 162 insertions, 1 deletions

diff --git a/app/tests/test_backends_common.py b/app/tests/test_backends_common.py
index 71f0c17..6e8ee1a 100644
--- a/app/tests/test_backends_common.py
+++ b/app/tests/test_backends_common.py
@@ -4,9 +4,13 @@ The streaming mode of ``run_console_subprocess`` (used by the in-TUI task
 view) is exercised with a real child process: output lines are captured and
 forwarded, cancellation kills the child and returns 130, an on_cancel hook
 runs first, and the no-output stall watchdog kills a wedged child and
-returns 124.
+returns 124. The on_chunk console mode (raw-byte sniffing with verbatim
+terminal forwarding — pip_install's corrupted-cache detector) is exercised
+the same way.
 """
 
+import contextlib
+import io
 import sys
 import threading
 import unittest
@@ -80,6 +84,53 @@ class RunConsoleSubprocessStreamingTests(unittest.TestCase):
             "http://host:8080/path")
 
 
+class RunConsoleSubprocessOnChunkTests(unittest.TestCase):
+    """The on_chunk console mode: raw bytes to the hook, verbatim forward."""
+
+    def test_on_chunk_receives_raw_bytes_and_output_is_forwarded(self):
+        chunks = []
+        captured = io.StringIO()
+        with contextlib.redirect_stdout(captured):
+            rc = common.run_console_subprocess(
+                [sys.executable, "-c",
+                 "import sys; sys.stdout.write('hello\\nworld\\n')"],
+                on_chunk=chunks.append)
+        self.assertEqual(rc, 0)
+        self.assertIn(b"hello\nworld\n", b"".join(chunks))
+        # The child's output still reaches the real terminal (verbatim, so
+        # carriage-return progress bars keep animating).
+        self.assertIn("hello\nworld\n", captured.getvalue())
+
+    def test_on_chunk_path_returns_the_exit_code(self):
+        rc = common.run_console_subprocess(
+            [sys.executable, "-c", "import sys; sys.exit(5)"],
+            on_chunk=lambda chunk: None)
+        self.assertEqual(rc, 5)
+
+    def test_on_chunk_cancel_kills_the_process_and_returns_130(self):
+        cancel = threading.Event()
+        cancel.set()
+        rc = common.run_console_subprocess(
+            [sys.executable, "-c", "import time; time.sleep(60)"],
+            on_chunk=lambda chunk: None, cancel=cancel)
+        self.assertEqual(rc, 130)
+
+    def test_on_chunk_carriage_return_progress_is_forwarded_raw(self):
+        # pip's download bars are \r-only: the hook must see them and the
+        # terminal must get them unsplit.
+        chunks = []
+        captured = io.StringIO()
+        with contextlib.redirect_stdout(captured):
+            rc = common.run_console_subprocess(
+                [sys.executable, "-c",
+                 "import sys; sys.stdout.write('dl 1\\rdl 2\\r'); "
+                 "sys.stdout.flush()"],
+                on_chunk=chunks.append)
+        self.assertEqual(rc, 0)
+        self.assertEqual(b"".join(chunks), b"dl 1\rdl 2\r")
+        self.assertIn("dl 1\rdl 2\r", captured.getvalue())
+
+
 class RunConsoleSubprocessStallTests(unittest.TestCase):
     """The no-output watchdog: a silent child is killed and reported 124."""
 
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), \