fix(cli): serialize footer installs and tolerate unknown versions

litellm/proxy/client/cli/commands/claude_settings.py

@@ -23,6 +23,7 @@ from types import MappingProxyType
 from typing import Final, TypeAlias
 
 import click
+from filelock import FileLock
 from packaging.version import InvalidVersion, Version
 from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError
 
@@ -309,6 +310,13 @@ def statusline_command(script_path: Path, platform: str = sys.platform) -> str:
     return " ".join(quote(token) for token in (sys.executable, str(script_path)))
 
 
+def _statusline_version(value: str) -> Version | None:
+    try:
+        return Version(value)
+    except InvalidVersion:
+        return None
+
+
 def _installed_statusline_version(target: Path) -> Version | None:
     try:
         with target.open("rb") as script:
@@ -318,28 +326,39 @@ def _installed_statusline_version(target: Path) -> Version | None:
     if not header.startswith(STATUSLINE_VERSION_PREFIX):
         return None
     try:
-        return Version(header.removeprefix(STATUSLINE_VERSION_PREFIX).decode("ascii").strip())
-    except (InvalidVersion, UnicodeDecodeError):
+        return _statusline_version(header.removeprefix(STATUSLINE_VERSION_PREFIX).decode("ascii").strip())
+    except UnicodeDecodeError:
         return None
 
 
-def install_statusline_script(script_path: Path | None = None, *, package_version: str = litellm_version) -> str:
+def install_statusline_script(
+    script_path: Path | None = None,
+    *,
+    package_version: str = litellm_version,
+    write: Callable[[str, bytes], None] = write_private_bytes,
+) -> str:
     target: Final = script_path or STATUSLINE_SCRIPT_PATH
     try:
         ensure_private_dir(target.parent)
-        bundled_version: Final = Version(package_version)
-        installed_version: Final = _installed_statusline_version(target)
-        if installed_version is not None and installed_version > bundled_version:
-            click.echo(
-                f"Keeping the status line from LiteLLM {installed_version}; this CLI is {bundled_version}. "
-                "Upgrade the CLI to refresh it.",
-                err=True,
+        bundled_version: Final = _statusline_version(package_version)
+        with FileLock(str(target) + ".lock", timeout=10, mode=0o600):
+            installed_version: Final = _installed_statusline_version(target)
+            if installed_version is not None and (bundled_version is None or installed_version > bundled_version):
+                cli_version: Final = str(bundled_version) if bundled_version is not None else "unknown"
+                click.echo(
+                    f"Keeping the status line from LiteLLM {installed_version}; this CLI is {cli_version}. "
+                    "Upgrade the CLI to refresh it.",
+                    err=True,
+                )
+                return statusline_command(target)
+            source: Final = Path(statusline_script.__file__).read_bytes()
+            header: Final = (
+                STATUSLINE_VERSION_PREFIX + str(bundled_version).encode("ascii") + b"\n"
+                if bundled_version is not None
+                else b""
             )
-            return statusline_command(target)
-        source: Final = Path(statusline_script.__file__).read_bytes()
-        header: Final = STATUSLINE_VERSION_PREFIX + str(bundled_version).encode("ascii") + b"\n"
-        write_private_bytes(str(target), header + source)
-    except (OSError, InvalidVersion) as e:
+            write(str(target), header + source)
+    except OSError as e:
         raise ClaudeSettingsError(f"Could not install the status line script at {target}: {e}") from e
     return statusline_command(target)
 

litellm/proxy/client/cli/commands/statusline_script.py

@@ -1,7 +1,7 @@
 """Claude Code status line and Codex Stop hook for auto-routed sessions.
 
-`lite` copies this file with a CLI version header to ~/.litellm/statusline.py and registers it as Claude
-Code's `statusLine` command and as Codex's `[[hooks.Stop]]` command, so it must stay
+`lite` copies this file to ~/.litellm/statusline.py with a CLI version header when known and registers
+it as Claude Code's `statusLine` command and as Codex's `[[hooks.Stop]]` command, so it must stay
 standard-library only and must never import litellm. Claude Code re-runs it on every
 status refresh (about every 300ms while typing), so the proxy is asked at most once per
 TTL per session and every other refresh is served from a small on-disk cache that holds

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     # When changing a floor, verify it installs + imports on every supported
     # Python with: `uv pip install --resolution=lowest-direct .`
     "fastuuid>=0.14.0,<1.0",
+    "filelock>=3.16.1,<4.0",
     "httpx[http2]>=0.28.0,<1.0",
     "openai>=2.20.0,<3.0.0",
     "python-dotenv>=1.0.0,<2.0",

tests/test_litellm/proxy/client/cli/test_claude_settings.py

@@ -5,14 +5,16 @@ import shlex
 import stat
 import sys
 import time
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
 from pathlib import Path
+from threading import Event
 from typing import Final
 from unittest.mock import patch
 
 import pytest
 from click.testing import CliRunner
 
-from litellm.litellm_core_utils.private_json import commit_staged_json
+from litellm.litellm_core_utils.private_json import commit_staged_json, write_private_bytes
 from litellm.proxy.client.cli.commands.claude_settings import (
     ANTHROPIC_DEFAULT_MODEL_ENV_KEYS,
     AUTOROUTE_BACKUP_PATH,
@@ -790,7 +792,7 @@ class TestStatusLine:
         with script.open("rb") as running:
             install_statusline_script(script)
             assert running.read() == bundled
-        assert [child.name for child in script.parent.iterdir()] == ["statusline.py"]
+        assert {child.name for child in script.parent.iterdir()} <= {"statusline.py", "statusline.py.lock"}
 
         if os.geteuid() != 0:
             script.parent.chmod(0o500)
@@ -862,6 +864,80 @@ class TestStatusLine:
         assert rig.read()["env"]["ANTHROPIC_BASE_URL"] == PROXY
         assert "Keeping the status line" in capsys.readouterr().err
 
+    @pytest.mark.parametrize("package_version", ("unknown", "", "invalid-version"))
+    @pytest.mark.parametrize("existing", (None, b"print('legacy footer')\n", b"# litellm-statusline-version: invalid\n"))
+    def test_an_unknown_cli_version_can_install_and_refresh_an_unversioned_footer(
+        self, tmp_path: Path, package_version: str, existing: bytes | None
+    ) -> None:
+        from litellm.proxy.client.cli.commands import statusline_script
+
+        script: Final = tmp_path / "statusline.py"
+        if existing is not None:
+            script.write_bytes(existing)
+
+        assert install_statusline_script(script, package_version=package_version) == statusline_command(script)
+        assert script.read_bytes() == Path(statusline_script.__file__).read_bytes()
+
+    def test_an_unknown_cli_version_preserves_a_versioned_footer(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        script: Final = tmp_path / "statusline.py"
+        command: Final = install_statusline_script(script, package_version="2.1.0")
+        installed: Final = script.read_bytes()
+
+        assert install_statusline_script(script, package_version="unknown") == command
+        assert script.read_bytes() == installed
+        assert "Keeping the status line from LiteLLM 2.1.0" in capsys.readouterr().err
+
+    @pytest.mark.parametrize(("first_version", "second_version"), (("2.0", "3.0"), ("3.0", "2.0")))
+    def test_overlapping_installs_keep_the_newest_footer(
+        self, tmp_path: Path, first_version: str, second_version: str
+    ) -> None:
+        from litellm.proxy.client.cli.commands import statusline_script
+
+        script: Final = tmp_path / "statusline.py"
+        first_writing: Final = Event()
+        release_first: Final = Event()
+        second_started: Final = Event()
+
+        def paused_write(path: str, data: bytes) -> None:
+            first_writing.set()
+            assert release_first.wait(5), "First installer was never released"
+            write_private_bytes(path, data)
+
+        def second_install() -> str:
+            second_started.set()
+            return install_statusline_script(script, package_version=second_version)
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            first: Final = pool.submit(install_statusline_script, script, package_version=first_version, write=paused_write)
+            try:
+                assert first_writing.wait(5), "First installer did not reach the write"
+                second: Final = pool.submit(second_install)
+                assert second_started.wait(5), "Second installer did not start"
+                with pytest.raises(FutureTimeoutError):
+                    second.result(timeout=0.5)
+            finally:
+                release_first.set()
+            assert first.result(timeout=5) == statusline_command(script)
+            assert second.result(timeout=5) == statusline_command(script)
+
+        assert script.read_bytes() == b"# litellm-statusline-version: 3.0\n" + Path(statusline_script.__file__).read_bytes()
+
+    def test_a_failed_install_keeps_the_footer_and_releases_the_lock(self, tmp_path: Path) -> None:
+        script: Final = tmp_path / "statusline.py"
+        install_statusline_script(script, package_version="2.0")
+        installed: Final = script.read_bytes()
+
+        def failed_write(path: str, data: bytes) -> None:
+            raise OSError("disk full")
+
+        with pytest.raises(ClaudeSettingsError, match="disk full"):
+            install_statusline_script(script, package_version="3.0", write=failed_write)
+        assert script.read_bytes() == installed
+        assert install_statusline_script(script, package_version="3.0") == statusline_command(script)
+        assert script.read_bytes().startswith(b"# litellm-statusline-version: 3.0\n")
+
     def test_configure_installs_it_and_unconfigure_removes_only_ours(self, tmp_path):
         rig = _Rig(tmp_path, {"theme": "dark"})
         script = tmp_path / "statusline.py"

uv.lock

@@ -4516,6 +4516,7 @@ dependencies = [
     { name = "boto3" },
     { name = "click" },
     { name = "fastuuid" },
+    { name = "filelock" },
     { name = "httpx", extra = ["http2"] },
     { name = "importlib-metadata" },
     { name = "jinja2" },
@@ -4770,6 +4771,7 @@ requires-dist = [
     { name = "fastapi", marker = "extra == 'proxy'", specifier = ">=0.136.3,<1.0" },
     { name = "fastapi-sso", marker = "extra == 'proxy'", specifier = ">=0.19.0,<1.0" },
     { name = "fastuuid", specifier = ">=0.14.0,<1.0" },
+    { name = "filelock", specifier = ">=3.16.1,<4.0" },
     { name = "google-cloud-aiplatform", marker = "extra == 'google'", specifier = ">=1.133.0,<2.0" },
     { name = "google-cloud-aiplatform", marker = "extra == 'proxy-runtime'", specifier = ">=1.133.0,<2.0" },
     { name = "google-cloud-iam", marker = "extra == 'extra-proxy'", specifier = ">=2.19.1,<3.0" },