Address review findings on #2496: owner-aware reservation pruning, per-process disk-full diagnosis, post-truncation device names, unreadable .venv, quota errnos

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

backend/core/db_backup.py

@@ -129,7 +129,10 @@ def _reserve_snapshot(db_path: str, safe_version: str) -> tuple[str, str]:
         except FileExistsError:
             counter += 1
             continue
-        os.close(fd)
+        try:
+            os.write(fd, str(os.getpid()).encode("ascii"))  # ownership for stale_reservations
+        finally:
+            os.close(fd)
         if os.path.exists(target):
             os.unlink(reservation)
             counter += 1
@@ -137,13 +140,28 @@ def _reserve_snapshot(db_path: str, safe_version: str) -> tuple[str, str]:
         return target, reservation
 
 
-#: A reservation older than this with no finished backup belongs to a dead writer.
+#: Fallback only for a reservation whose owner PID was never recorded (a writer
+#: that died between creating the file and writing its PID).
 _RESERVATION_MAX_AGE_S = 24 * 3600
 
 
+def _reservation_owner(path: str) -> int | None:
+    try:
+        with open(path, "r", encoding="ascii") as fh:
+            pid = int(fh.read().strip())
+    except (OSError, ValueError):
+        return None
+    return pid if pid > 0 else None
+
+
 def stale_reservations(db_path: str) -> list[str]:
-    """``.reserve`` markers left by a crashed writer (or whose snapshot already
-    landed). Fresh ones may belong to a live snapshot and are never listed."""
+    """``.reserve`` markers whose writer is confirmed gone.
+
+    A reservation owned by a live process (including this one, other threads)
+    is never listed, however old: a snapshot can outlive a long suspend, and
+    removing its marker would let another startup claim the same counter.
+    ``_pid_alive`` errs toward "alive". Only a marker with no recorded owner
+    falls back to an age check."""
     directory = os.path.dirname(os.path.abspath(db_path)) or "."
     base = os.path.basename(db_path)
     try:
@@ -155,11 +173,15 @@ def stale_reservations(db_path: str) -> list[str]:
     for name in names:
         if not (name.startswith(base + ".backup-") and name.endswith(".reserve")):
             continue
-        target = name[: -len(".reserve")]
-        if not _BACKUP_SUFFIX_RE.fullmatch(target[len(base):]):
+        if not _BACKUP_SUFFIX_RE.fullmatch(name[: -len(".reserve")][len(base):]):
             continue
         path = os.path.join(directory, name)
-        if os.path.exists(os.path.join(directory, target)) or now - _mtime(path) > _RESERVATION_MAX_AGE_S:
+        owner = _reservation_owner(path)
+        if owner is not None:
+            if owner == os.getpid() or _pid_alive(owner):
+                continue
+            stale.append(path)
+        elif now - _mtime(path) > _RESERVATION_MAX_AGE_S:
             stale.append(path)
     return stale
 
@@ -261,7 +283,7 @@ def snapshot_before_migration(db_path: str, version: str) -> str | None:
             try:
                 os.remove(tmp)
             except OSError:
-                pass
+                pass  # best effort: the torn temp is pruned later; the backup error below matters more
             raise
         finally:
             src.close()
@@ -271,7 +293,7 @@ def snapshot_before_migration(db_path: str, version: str) -> str | None:
             now = time.time()
             os.utime(target, (now, now))
         except OSError:
-            pass
+            pass  # ordering hint only; the snapshot itself is already in place
         logger.info("Pre-migration DB backup written: %s (%.1f MB)", target, size / (1024 * 1024))
         prune_backups(db_path)
         return target
@@ -279,4 +301,4 @@ def snapshot_before_migration(db_path: str, version: str) -> str | None:
         try:
             os.unlink(reservation)
         except OSError:
-            pass
+            pass  # marker already gone; pruning handles any leftover

backend/core/failure.py

@@ -15,6 +15,7 @@ Guarantees:
 """
 from __future__ import annotations
 
+import errno
 import os
 import platform
 import re
@@ -275,9 +276,15 @@ _HF_CONTEXT_MARKERS = (
 
 
 _DISK_FULL_SIGNATURES = (
-    "errno 28", "no space left", "errno 122", "disk quota exceeded",
+    "errno 28", "no space left",
+    # Linux says "Disk quota exceeded"; macOS says "Disc quota exceeded".
+    "disk quota exceeded", "disc quota exceeded",
     "winerror 112", "winerror 39", "not enough space on the disk",
 )
+# EDQUOT is 122 on Linux and 69 on macOS (absent on Windows) — use the platform's.
+_DISK_FULL_ERRNOS = frozenset(
+    n for n in (getattr(errno, "ENOSPC", None), getattr(errno, "EDQUOT", None)) if n is not None
+)
 
 
 def is_disk_full_error(reason: "BaseException | str | None") -> bool:
@@ -294,7 +301,7 @@ def is_disk_full_error(reason: "BaseException | str | None") -> bool:
             exc: "BaseException | None" = reason
             while exc is not None and id(exc) not in seen:
                 seen.add(id(exc))
-                if getattr(exc, "errno", None) in (28, 122) or getattr(exc, "winerror", None) in (39, 112):
+                if getattr(exc, "errno", None) in _DISK_FULL_ERRNOS or getattr(exc, "winerror", None) in (39, 112):
                     return True
                 if any(sig in str(exc).lower() for sig in _DISK_FULL_SIGNATURES):
                     return True

backend/core/path_security.py

@@ -68,13 +68,17 @@ def portable_filename(value: object, default: str = "file", max_bytes: int = 200
         stem, ext = name, ""
     else:
         ext = "." + ext
-    stem = stem.rstrip(" .")
+    budget = max(2, max_bytes - len(ext.encode("utf-8")))
+
+    def _fit(text: str) -> str:
+        return text.encode("utf-8")[:budget].decode("utf-8", "ignore").rstrip(" .")
+
+    stem = _fit(stem) or default
     if not stem.strip("_ "):
         stem = default
-    if stem.split(".", 1)[0].upper() in _WINDOWS_RESERVED_NAMES:
-        stem = "_" + stem
-    budget = max(1, max_bytes - len(ext.encode("utf-8")))
-    stem = stem.encode("utf-8")[:budget].decode("utf-8", "ignore").rstrip(" .") or default
+    # Check device names AFTER truncation: cutting a long stem can expose "CON".
+    if stem.split(".", 1)[0].rstrip().upper() in _WINDOWS_RESERVED_NAMES:
+        stem = _fit("_" + stem)
     return stem + ext
 
 

backend/services/segmented_download.py

@@ -254,7 +254,7 @@ def _invalidate_done(part: str) -> None:
     try:
         os.remove(_manifest_path(part))
     except FileNotFoundError:
-        pass
+        pass  # already absent: the goal of invalidation is met
 
 
 def _preallocate(part: str, size: int) -> None:

backend/services/sidecar_install.py

@@ -57,7 +57,7 @@ import time
 from collections import deque
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Callable, NamedTuple, Optional
+from typing import Callable, Iterable, NamedTuple, Optional
 
 from core.config import DATA_DIR
 from core.contained_subprocess import OwnedPopen, WindowsJobPopen, spawn_owned
@@ -945,18 +945,26 @@ _DISK_FULL_REMEDIATION = (
 )
 
 
+def _redact(exc: BaseException) -> str:
+    """Exception text with home-directory paths and tokens scrubbed, since a
+    step error is persisted in the job record and log."""
+    from core.failure import sanitize
+
+    return sanitize(str(exc))
+
+
 def _is_disk_full(exc: BaseException) -> bool:
     from core.failure import is_disk_full_error
 
     return is_disk_full_error(exc)
 
 
-def _log_shows_disk_full(job: dict, tail: int = 60) -> bool:
-    """uv reports a full volume only as log text with a bare non-zero exit."""
+def _output_shows_disk_full(lines: Iterable[str]) -> bool:
+    """uv reports a full volume only as output text with a bare non-zero exit.
+    Pass the failing process's own output, never the shared job log: an earlier
+    step's recovered "No space left" line must not relabel a later failure."""
     from core.failure import is_disk_full_error
 
-    with _log_lock:
-        lines = list(job["log"])[-tail:]
     return is_disk_full_error("\n".join(lines))
 
 
@@ -971,7 +979,7 @@ def _log(job: dict, line: str) -> None:
 def _serialize_job(job: Optional[dict]) -> Optional[dict]:
     if job is None:
         return None
-    out = dict(job)
+    out = {k: v for k, v in job.items() if not k.startswith("_")}
     with _log_lock:
         out["log"] = list(job["log"])
     out["steps"] = [dict(s) for s in job["steps"]]
@@ -1185,9 +1193,9 @@ def _run_install(spec: SidecarSpec, job: dict) -> None:
             except Exception as exc:  # noqa: BLE001 — surfaced into the job
                 step["state"] = "error"
                 if _is_disk_full(exc):
-                    raise _StepError(f"{type(exc).__name__}: {exc}", _DISK_FULL_REMEDIATION) from exc
+                    raise _StepError(f"{type(exc).__name__}: {_redact(exc)}", _DISK_FULL_REMEDIATION) from exc
                 raise _StepError(
-                    f"{type(exc).__name__}: {exc}",
+                    f"{type(exc).__name__}: {_redact(exc)}",
                     "Re-run the install — it resumes from where it stopped. If it "
                     f"keeps failing, see {spec.docs_path} for the manual steps.",
                 ) from exc
@@ -1528,6 +1536,7 @@ def _step_install_deps(spec: SidecarSpec, job: dict) -> None:
         target += list(UV_PIP_CU128_ARGS)
     # Always `--python <this engine's venv>`: the install can only ever land in
     # the venv this engine owns, never the app's interpreter.
+    job.pop("_last_run_output", None)
     rc = _run_logged(
         job,
         [uv, "pip", "install", "--python", str(py), *target],
@@ -1535,7 +1544,7 @@ def _step_install_deps(spec: SidecarSpec, job: dict) -> None:
         env=uv_subprocess_env(Path(DATA_DIR) / "engines"),
     )
     if rc != 0:
-        if _log_shows_disk_full(job):
+        if _output_shows_disk_full(job.get("_last_run_output") or ()):
             raise _StepError(f"uv pip install failed (exit {rc}): no space left on device.",
                              _DISK_FULL_REMEDIATION)
         hint = (
@@ -1708,9 +1717,9 @@ def _step_fetch_weights(spec: SidecarSpec, job: dict) -> None:
             snapshot_download(**kwargs)  # nosec B615 — deliberate default-branch policy, see above
         except Exception as exc:
             if _is_disk_full(exc):
-                raise _StepError(f"Model weight download failed: {exc}", _DISK_FULL_REMEDIATION) from exc
+                raise _StepError(f"Model weight download failed: {_redact(exc)}", _DISK_FULL_REMEDIATION) from exc
             raise _StepError(
-                f"Model weight download failed: {exc}",
+                f"Model weight download failed: {_redact(exc)}",
                 "Re-run the install — the download resumes where it stopped. "
                 "Check Settings → Network (HF endpoint / proxy) if it keeps failing.",
             ) from exc
@@ -1755,6 +1764,9 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
     Returns the exit code; -1 on timeout (process tree killed) or spawn
     failure. argv-list only — never a shell string — so paths with spaces
     are safe on every platform. ``env=None`` inherits the parent environment.
+    This process's own lines are also kept in ``job["_last_run_output"]`` (reset
+    per call, hidden from the status payload) so a caller can diagnose its own
+    failure without reading earlier steps' log output.
 
     The stdout drain runs on its own daemon thread and the main flow blocks
     on ``proc.wait(timeout=…)``. That bounds the step even when a grandchild
@@ -1766,6 +1778,8 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
     # starts. POSIX links it to backend death through a control pipe; Windows
     # retains a kill-on-close Job handle in this backend process.
     popen_kwargs = _install_containment_kwargs()
+    tail: deque[str] = deque(maxlen=60)
+    job["_last_run_output"] = tail
     try:
         proc = spawn_owned(
             argv,
@@ -1786,6 +1800,7 @@ def _run_logged(job: dict, argv: list[str], *, timeout: float,
             assert proc.stdout is not None
             for line in proc.stdout:
                 _log(job, line)
+                tail.append(line)
         except (OSError, ValueError):
             pass  # pipe closed by the timeout kill — nothing left to read
 

backend/services/storage_report.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 import glob
 import os
 import shutil
+import stat
 import sys
 import tempfile
 import threading
@@ -62,6 +63,16 @@ def default_engines_dir() -> str:
     return str(Path(DATA_DIR) / "engines")
 
 
+def _has_venv(entry_path: str) -> bool:
+    """True when ``<entry>/.venv`` is a directory. Only a *missing* path means
+    "no venv"; a permission or I/O error propagates so the caller can mark the
+    scan incomplete instead of silently reclassifying an installed engine."""
+    try:
+        return stat.S_ISDIR(os.stat(os.path.join(entry_path, ".venv")).st_mode)
+    except (FileNotFoundError, NotADirectoryError):
+        return False
+
+
 def _engines_child_name(engines_dir: str, data_dir: str) -> str | None:
     """Basename of ``engines_dir`` when it is a direct child of ``data_dir`` —
     so the data category can skip it and not double-count what the engine-venv
@@ -281,7 +292,7 @@ def build_report(
         # the entry. Persistently unknown ownership remains an incomplete scan.
         for _ in range(2):
             try:
-                if e.is_dir(follow_symlinks=False) and os.path.isdir(os.path.join(e.path, ".venv")):
+                if e.is_dir(follow_symlinks=False) and _has_venv(e.path):
                     engine_dirs.append(e.path)
                 break
             except OSError:

docs/install/troubleshooting.md

@@ -687,7 +687,7 @@ A segmented download refuses a response whose status or Content-Range does not m
 
 Segmented download resume records are reused only with an existing partial file of the expected size and valid byte-range entries. If a partial file is missing, truncated, or oversized, or its sidecar is malformed, the download fetches those bytes again instead of treating preallocated zeros as completed data. Oversized partial files are resized before restarting so old trailing bytes cannot prevent verification of the new download. Stale checkpoints are removed before resizing or recreating partial files, so a failed fetch cannot make the next retry trust stale or zero-filled bytes. If that stale checkpoint cannot be removed, the restart stops before changing the partial file or destination.
 
-**Disk filled up mid-install.** A model or engine install that runs out of space stops immediately (it is not retried with backoff) and reports how much space is free and where; free space or move the model cache to a larger volume, then retry. The download resumes from the part that already finished. During first-run setup, the one-time `uv` installer download is retried up to three times on connection resets, timeouts, and HTTP 5xx/429 before it reports failure.
+**Disk filled up mid-install.** A model or engine install that runs out of space stops immediately (it is not retried with backoff) and reports how much space is free and where; free space or move the model cache to a larger volume, then retry. The download resumes from the part that already finished. During first-run setup, the one-time `uv` installer download is attempted up to three times (two retries) on connection resets, timeouts, and HTTP 5xx/429 before it reports failure.
 
 **Exports named after a video title.** Download and export names built from a video title replace characters Windows rejects (`< > : " / \ | ? *`, control characters, trailing dots/spaces, device names such as `CON`) with `_` on every OS, so `How to X: a guide?` exports as `How to X_ a guide_` instead of failing with `[Errno 22] Invalid argument`.
 

electron/src/main/portable-filename.test.ts

@@ -14,6 +14,13 @@ it('repairs trailing dots/spaces, device names and empty names', () => {
   expect(portableFilename('', 'audio.wav')).toBe('audio.wav');
 });
 
+it('repairs a device name exposed by truncation', () => {
+  const name = portableFilename(`CON${' '.repeat(197)}x.wav`);
+  expect(name.split('.')[0].toUpperCase()).not.toBe('CON');
+  expect(name.endsWith('.wav')).toBe(true);
+  expect(new TextEncoder().encode(name).length).toBeLessThanOrEqual(200);
+});
+
 it('truncates by UTF-8 bytes without splitting a character or losing the extension', () => {
   const name = portableFilename('\u65e5\u672c\u8a9e'.repeat(60) + '.mp4');
   expect(new TextEncoder().encode(name).length).toBeLessThanOrEqual(200);

electron/src/main/portable-filename.ts

@@ -33,10 +33,11 @@ export function portableFilename(value: string, fallback = 'file', maxBytes = 20
     stem = name.slice(0, dot);
     ext = name.slice(dot);
   }
-  stem = stem.replace(/[ .]+$/, '');
+  const budget = Math.max(2, maxBytes - encoder.encode(ext).length);
+  const fit = (text: string) => truncateBytes(text, budget).replace(/[ .]+$/, '');
+  stem = fit(stem) || fallback;
   if (!stem.replace(/[_ ]/g, '')) stem = fallback;
-  if (RESERVED.has(stem.split('.')[0].toUpperCase())) stem = `_${stem}`;
-  const budget = Math.max(1, maxBytes - encoder.encode(ext).length);
-  stem = truncateBytes(stem, budget).replace(/[ .]+$/, '') || fallback;
+  // Check device names AFTER truncation: cutting a long stem can expose "CON".
+  if (RESERVED.has(stem.split('.')[0].trimEnd().toUpperCase())) stem = fit(`_${stem}`);
   return stem + ext;
 }

tests/test_db_backup_concurrent_slots.py

@@ -3,10 +3,11 @@ import concurrent.futures
 import sqlite3
 import threading
 
-from core import db_backup
 
 
 def test_concurrent_snapshots_cannot_share_a_recovery_slot(tmp_path, monkeypatch):
+    from core import db_backup
+
     path = tmp_path / 'voices.db'
     with sqlite3.connect(path) as connection:
         connection.execute('CREATE TABLE voices(name TEXT)')
@@ -32,6 +33,8 @@ def test_concurrent_snapshots_cannot_share_a_recovery_slot(tmp_path, monkeypatch
 
 
 def test_abandoned_reservation_is_not_reused_or_listed(tmp_path):
+    from core import db_backup
+
     path = tmp_path / 'voices.db'
     with sqlite3.connect(path) as connection:
         connection.execute('CREATE TABLE voices(name TEXT)')
@@ -44,6 +47,8 @@ def test_abandoned_reservation_is_not_reused_or_listed(tmp_path):
 
 
 def test_failed_copy_releases_its_reservation(tmp_path, monkeypatch):
+    from core import db_backup
+
     import pytest
 
     path = tmp_path / 'voices.db'
@@ -63,22 +68,29 @@ def test_failed_copy_releases_its_reservation(tmp_path, monkeypatch):
     assert db_backup.list_backups(str(path)) == []
 
 
-def test_abandoned_reservations_are_pruned_but_live_ones_kept(tmp_path):
+def test_reservations_are_pruned_only_when_the_owner_is_gone(tmp_path, monkeypatch):
     import os
     import time
 
+    from core import db_backup
+
     path = tmp_path / 'voices.db'
     path.write_bytes(b'')
-    base = f'{path}.backup-1.0-'
-    done = tmp_path / 'voices.db.backup-1.0-1'
-    done.write_bytes(b'x')
-    (tmp_path / 'voices.db.backup-1.0-1.reserve').write_bytes(b'')       # snapshot landed
-    old = tmp_path / 'voices.db.backup-1.0-2.reserve'                    # dead writer
-    old.write_bytes(b'')
+    dead = tmp_path / 'voices.db.backup-1.0-1.reserve'
+    dead.write_text('999999')
+    live_old = tmp_path / 'voices.db.backup-1.0-2.reserve'   # owner alive, however old
+    live_old.write_text('4242')
+    mine = tmp_path / 'voices.db.backup-1.0-3.reserve'
+    mine.write_text(str(os.getpid()))
+    ownerless_old = tmp_path / 'voices.db.backup-1.0-4.reserve'
+    ownerless_old.write_bytes(b'')
+    ownerless_fresh = tmp_path / 'voices.db.backup-1.0-5.reserve'
+    ownerless_fresh.write_bytes(b'')
     past = time.time() - 48 * 3600
-    os.utime(old, (past, past))
-    live = tmp_path / 'voices.db.backup-1.0-3.reserve'                   # in flight
-    live.write_bytes(b'')
+    for p in (dead, live_old, ownerless_old):
+        os.utime(p, (past, past))
+    monkeypatch.setattr(db_backup, '_pid_alive', lambda pid: pid == 4242)
     db_backup.prune_backups(str(path))
-    assert sorted(p.name for p in tmp_path.iterdir() if p.name.endswith('.reserve')) == [live.name]
-    assert done.exists() and base
+    assert sorted(p.name for p in tmp_path.iterdir() if p.name.endswith('.reserve')) == sorted(
+        [live_old.name, mine.name, ownerless_fresh.name]
+    )

tests/test_install_disk_full.py

@@ -4,8 +4,6 @@ import errno
 
 import pytest
 
-from core.failure import is_disk_full_error
-
 
 @pytest.mark.parametrize(
     "reason",
@@ -14,13 +12,18 @@ from core.failure import is_disk_full_error
         OSError("[Errno 28] No space left on device: '/x/blobs/a.incomplete'"),
         "[WinError 112] There is not enough space on the disk",
         "OSError: [Errno 122] Disk quota exceeded",
+        "OSError: [Errno 69] Disc quota exceeded",
     ],
 )
 def test_disk_full_is_recognised(reason):
+    from core.failure import is_disk_full_error
+
     assert is_disk_full_error(reason)
 
 
 def test_disk_full_is_found_through_a_wrapped_cause():
+    from core.failure import is_disk_full_error
+
     try:
         try:
             raise OSError(errno.ENOSPC, "No space left on device")
@@ -30,7 +33,16 @@ def test_disk_full_is_found_through_a_wrapped_cause():
         assert is_disk_full_error(wrapped)
 
 
+@pytest.mark.skipif(not hasattr(errno, "EDQUOT"), reason="EDQUOT is absent on Windows")
+def test_platform_quota_errno_is_disk_full():
+    from core.failure import is_disk_full_error
+
+    assert is_disk_full_error(OSError(errno.EDQUOT, "quota"))
+
+
 def test_other_failures_are_not_disk_full():
+    from core.failure import is_disk_full_error
+
     assert not is_disk_full_error(ConnectionResetError("peer closed connection"))
     assert not is_disk_full_error("Not enough disk space to install: needs 3 GB")
     assert not is_disk_full_error(None)
@@ -50,11 +62,33 @@ def test_disk_full_message_names_free_space_and_cache(tmp_path):
     assert str(tmp_path) in text and "Free up space" in text
 
 
-def test_sidecar_install_reports_full_disk_from_uv_log():
+def test_sidecar_install_reports_full_disk_from_uv_output():
     from collections import deque
 
     from services import sidecar_install as si
 
-    job = {"log": deque(["Downloading torch", "error: failed to write: No space left on device (os error 28)"])}
-    assert si._log_shows_disk_full(job)
-    assert not si._log_shows_disk_full({"log": deque(["connection reset by peer"])})
+    assert si._output_shows_disk_full(
+        deque(["Downloading torch", "error: failed to write: No space left on device (os error 28)"])
+    )
+    assert not si._output_shows_disk_full(deque(["connection reset by peer"]))
+
+
+def test_earlier_recovered_disk_full_does_not_relabel_a_later_uv_failure(monkeypatch):
+    """A clone that logged ENOSPC then recovered stays in the shared job log;
+    only the failing process's own output may decide the diagnosis."""
+    from collections import deque
+
+    from services import sidecar_install as si
+
+    job = {"engine_id": "x", "log": deque(["clone: No space left on device", "fallback clone ok"])}
+
+    class _Proc:
+        stdout = iter(["error: no matching distribution\n"])
+
+        def wait(self, timeout=None):
+            return 1
+
+    monkeypatch.setattr(si, "spawn_owned", lambda *a, **k: _Proc())
+    assert si._run_logged(job, ["uv"], timeout=5) == 1
+    assert list(job["_last_run_output"]) == ["error: no matching distribution\n"]
+    assert not si._output_shows_disk_full(job["_last_run_output"])

tests/test_portable_filename.py

@@ -4,8 +4,6 @@ import re
 
 import pytest
 
-from core.path_security import portable_filename
-
 BACKEND = pathlib.Path(__file__).resolve().parents[1] / "backend"
 
 
@@ -23,10 +21,22 @@ BACKEND = pathlib.Path(__file__).resolve().parents[1] / "backend"
     ],
 )
 def test_portable_filename_repairs_windows_hostile_names(raw, expected):
+    from core.path_security import portable_filename
+
     assert portable_filename(raw) == expected
 
 
+def test_reserved_device_name_exposed_by_truncation_is_still_repaired():
+    from core.path_security import portable_filename
+
+    name = portable_filename("CON" + " " * 197 + "x.wav")
+    assert name.split(".")[0].upper() != "CON" and name.endswith(".wav")
+    assert len(name.encode("utf-8")) <= 200
+
+
 def test_portable_filename_truncates_by_bytes_and_keeps_extension():
+    from core.path_security import portable_filename
+
     name = portable_filename("日本語" * 60 + ".mp4")
     assert len(name.encode("utf-8")) <= 200
     assert name.endswith(".mp4") and "�" not in name

tests/test_storage_partials_regression.py

@@ -136,3 +136,28 @@ def test_transient_installed_entry_error_preserves_category_ownership(tmp_path,
     other = next(c for c in categories["data"]["children"] if c["id"] == "other")
     assert other["bytes"] == 0
     assert any(w["kind"] == "unreadable" and w["path"] == str(installed) for w in report["warnings"])
+
+
+def test_unreadable_venv_is_not_treated_as_missing(tmp_path, monkeypatch):
+    from services import storage_report
+    data = tmp_path / "data"
+    engines = data / "engines"
+    installed = engines / "installed"
+    (installed / ".venv").mkdir(parents=True)
+    (installed / "model.bin").write_bytes(b"installed bytes")
+    venv = str(installed / ".venv")
+    real_stat = storage_report.os.stat
+
+    def stat(path, *args, **kwargs):
+        if str(path) == venv:
+            raise PermissionError("traversal denied")
+        return real_stat(path, *args, **kwargs)
+
+    monkeypatch.setattr(storage_report.os, "stat", stat)
+    report = storage_report.build_report(data_dir=str(data), engines_dir=str(engines),
+        hf_cache_dir=str(tmp_path / "hf"), temp_root=str(tmp_path / "tmp"))
+    categories = {c["id"]: c for c in report["categories"]}
+    assert not categories["engine_venvs"]["complete"]
+    other = next(c for c in categories["data"]["children"] if c["id"] == "other")
+    assert other["bytes"] == 0  # not silently reclassified as application data
+    assert any(w["kind"] == "unreadable" and w["path"] == str(installed) for w in report["warnings"])