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

4 files changed, 264 insertions, 7 deletions

diff --git a/app/backends/common.py b/app/backends/common.py
index b5d6576..a12a366 100644
--- a/app/backends/common.py
+++ b/app/backends/common.py
@@ -372,7 +372,8 @@ def write_prompt_text(wav_dir: Path,
 def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
                            *, emit=None, cancel=None, on_cancel=None,
                            stall_timeout: Optional[float] = None,
-                           env: Optional[Dict[str, str]] = None) -> int:
+                           env: Optional[Dict[str, str]] = None,
+                           on_chunk=None) -> int:
     """Run a subprocess, streaming output to the console or to EMIT.
 
     With EMIT None the child inherits the real terminal and its output
@@ -382,6 +383,15 @@ def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
     updates like git's or tqdm's surface as lines), and each line is passed
     to EMIT — the in-TUI task view path.
 
+    ON_CHUNK (a ``callable(bytes)``) asks for the child's output *and* a
+    look at its raw bytes: the run then takes the piped path even without
+    EMIT, forwarding every chunk verbatim to the real terminal (so
+    carriage-return progress still animates) and handing each chunk to
+    ON_CHUNK — the hook pip_install uses to sniff pip's output for a
+    corrupted-wheel failure while the user watches it live. ON_CHUNK must
+    not raise and must not write to the terminal (the forwarding here
+    already does).
+
     CANCEL is an optional ``threading.Event``: once set, ON_CANCEL (if given)
     is called (e.g. to touch a ``--cancel-file``), then the child's process
     group is terminated (SIGTERM, escalating to SIGKILL after a grace
@@ -401,7 +411,7 @@ def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
     Returns the process exit code.
     """
     import subprocess
-    if emit is None:
+    if emit is None and on_chunk is None:
         try:
             result = subprocess.run(
                 argv, cwd=str(cwd) if cwd is not None else None, env=env)
@@ -423,7 +433,11 @@ def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
     try:
         proc = subprocess.Popen(argv, **popen_kwargs)
     except OSError as exc:
-        emit(f"[ERROR] Could not run {' '.join(argv)}: {exc}")
+        message = f"[ERROR] Could not run {' '.join(argv)}: {exc}"
+        if emit is not None:
+            emit(message)
+        else:
+            print(message)
         return 1
 
     cancelled = False
@@ -448,6 +462,16 @@ def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
                 if not chunk:
                     break
                 last_output = time.monotonic()
+                if on_chunk is not None:
+                    on_chunk(chunk)
+                if emit is None:
+                    # The on_chunk console path: forward each chunk
+                    # verbatim so carriage-return progress bars still
+                    # animate exactly as they did when the child owned
+                    # the terminal.
+                    sys.stdout.write(chunk.decode("utf-8", errors="replace"))
+                    sys.stdout.flush()
+                    continue
                 pending += chunk.decode("utf-8", errors="replace")
                 parts = re.split(r"[\r\n]+", pending)
                 pending = parts.pop()
@@ -483,8 +507,12 @@ def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
         if (stall_timeout is not None
                 and time.monotonic() - last_output > stall_timeout):
             stalled = True
-            emit(f"[ERROR] No output for {int(stall_timeout)}s — assuming "
-                 "the process hung; stopping it.")
+            message = (f"[ERROR] No output for {int(stall_timeout)}s — "
+                       "assuming the process hung; stopping it.")
+            if emit is not None:
+                emit(message)
+            else:
+                print(message)
             _terminate_process_group(proc)
             break
         time.sleep(0.1)
diff --git a/app/backends/envs.py b/app/backends/envs.py
index 98f1cc1..7e2b12d 100644
--- a/app/backends/envs.py
+++ b/app/backends/envs.py
@@ -211,6 +211,57 @@ def install_requirements(skip_optional: bool = False) -> int:
         [str(env_python()), "-m", "pip", "install", *installable])
 
 
+# Substrings of pip's output that identify a wheel corrupted inside pip's
+# HTTP cache (a truncated download from an interrupted install or a disk
+# that filled mid-write — pip faithfully re-serves the bad bytes from
+# ~/.cache/pip on every later run). Unpacking dies deep in the install
+# phase with ``zipfile.BadZipFile`` / "Bad CRC-32", long after resolution
+# looked healthy. Matching the traceback markers rather than pip's exit
+# code (a generic 2) is what lets pip_install retry *only* this failure
+# mode instead of re-downloading gigabytes on every ordinary pip failure.
+_WHEEL_CORRUPTION_MARKERS = (b"BadZipFile", b"Bad CRC-32")
+
+
+def _corruption_scanner(state: Dict[str, bool]):
+    """A text/bytes callback that flags wheel-corruption markers in STATE.
+
+    One closure serves both streaming paths: raw chunk bytes from
+    run_console_subprocess's on_chunk console hook and decoded lines from
+    a wrapped EMIT. Sets ``state["corrupt"]`` once and stays silent
+    otherwise (it must never print — the forwarding caller already shows
+    the output — and never raise).
+    """
+    def _scan(text) -> None:
+        if state["corrupt"]:
+            return
+        blob = text if isinstance(text, bytes) else \
+            text.encode("utf-8", errors="replace")
+        for marker in _WHEEL_CORRUPTION_MARKERS:
+            if marker in blob:
+                state["corrupt"] = True
+                return
+    return _scan
+
+
+def _pip_install_once(argv: List[str], *, emit, cancel) -> Tuple[int, bool]:
+    """One pip attempt; returns (exit code, wheel-corruption-seen).
+
+    The console path (no EMIT) passes an on_chunk scanner so pip's output
+    still streams verbatim to the terminal while being sniffed; the EMIT
+    path wraps EMIT so the task view receives every line unchanged.
+    """
+    state: Dict[str, bool] = {"corrupt": False}
+    scan = _corruption_scanner(state)
+    if emit is None:
+        rc = common.run_console_subprocess(argv, cancel=cancel, on_chunk=scan)
+    else:
+        def wrapped(line: str) -> None:
+            scan(line)
+            emit(line)
+        rc = common.run_console_subprocess(argv, emit=wrapped, cancel=cancel)
+    return rc, state["corrupt"]
+
+
 def pip_install(packages: List[str], *, emit=None, cancel=None,
                 env_dir: Optional[Path] = None,
                 upgrade: bool = False,
@@ -231,6 +282,13 @@ def pip_install(packages: List[str], *, emit=None, cancel=None,
     launching one. Returns pip's exit code. With EMIT given (the in-TUI task
     view) pip runs with ``--progress-bar off`` so its output is clean status
     lines rather than carriage-return progress spam.
+
+    A failed attempt whose output shows a wheel corrupted inside pip's HTTP
+    cache (zipfile.BadZipFile / Bad CRC-32 — a truncated download pip keeps
+    re-serving; the cache lives in the user's home, so it survives repo
+    re-clones and venv deletes) gets exactly one self-heal retry: the pip
+    cache is purged and the same install reruns, re-downloading every wheel
+    afresh. Ordinary failures (no corruption markers) return immediately.
     """
     if not env_exists(env_dir) and create_env(env_dir, interpreter) != 0:
         return 1
@@ -244,7 +302,17 @@ def pip_install(packages: List[str], *, emit=None, cancel=None,
         argv.append("off")
     argv.extend(extra_args or [])
     argv.extend(packages)
-    return common.run_console_subprocess(argv, emit=emit, cancel=cancel)
+    rc, corrupted = _pip_install_once(argv, emit=emit, cancel=cancel)
+    if rc != 0 and corrupted:
+        print(f"[WARNING] pip failed unpacking a wheel corrupted in its "
+              f"cache (~/.cache/pip — typically a truncated download; it "
+              f"survives re-cloning this project). Purging the pip cache "
+              f"and retrying once...")
+        common.run_console_subprocess_quiet(
+            [str(env_python(env_dir)), "-m", "pip", "cache", "purge"])
+        print("[INFO] pip cache purged; re-running the install...")
+        rc, _corrupted = _pip_install_once(argv, emit=emit, cancel=cancel)
+    return rc
 
 
 def pip_uninstall(packages: List[str], *, emit=None,
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), \