Coverage for src/lilbee/core/system.py: 100%
180 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""OS, environment, and platform helpers for lilbee."""
3import os
4import shutil
5import sys
6import threading
7from collections.abc import Iterator
8from contextlib import contextmanager
9from functools import cached_property
10from pathlib import Path
12#: Directory name for a project-local lilbee knowledge base (sibling of ``.git/``).
13LOCAL_ROOT_DIRNAME = ".lilbee"
15# Reentrant: a suppressed block can re-enter this (directly or via a native
16# helper wrapping its own stderr) and a plain Lock self-deadlocks. Nesting
17# restores correctly: the inner exit puts back the outer's devnull.
18_STDERR_LOCK = threading.RLock()
21@contextmanager
22def stderr_suppressed() -> Iterator[None]:
23 """Redirect fd 2 to /dev/null for the duration of the block.
25 Silences C-library stderr (native document extractors, GGUF readers) that
26 bypasses Python's logging. Holds a process lock so concurrent fd-2 swaps
27 can't clobber each other's saved descriptor. Wrap the whole native call, not
28 each inner iteration, so the lock doesn't serialize a hot loop.
30 On Windows, MSVC-built native extensions use GetStdHandle rather than the
31 CRT fd 2, so the fd-dup technique has no effect there. The context manager
32 is a no-op on Windows to avoid false suppression expectations.
33 """
34 if sys.platform == "win32":
35 yield
36 return
37 with _STDERR_LOCK:
38 devnull = os.open(os.devnull, os.O_WRONLY)
39 old_stderr = os.dup(2)
40 os.dup2(devnull, 2)
41 try:
42 yield
43 finally:
44 os.dup2(old_stderr, 2)
45 os.close(devnull)
46 os.close(old_stderr)
49def _dir_from_env(key: str, *home_relative: str) -> Path:
50 """Path named by environment variable *key*, or *home_relative* under the home directory.
52 The home directory is resolved only when the variable is unset, because
53 ``Path.home()`` raises where no home directory resolves.
54 """
55 value = os.environ.get(key)
56 return Path(value) if value is not None else Path.home().joinpath(*home_relative)
59def default_data_dir() -> Path:
60 """Return platform-appropriate data directory.
61 - macOS: ~/Library/Application Support/lilbee
62 - Windows: %LOCALAPPDATA%/lilbee
63 - Linux: ~/.local/share/lilbee (XDG_DATA_HOME)
64 """
65 if sys.platform == "darwin":
66 base = Path.home() / "Library" / "Application Support"
67 elif sys.platform == "win32":
68 base = _dir_from_env("LOCALAPPDATA", "AppData", "Local").expanduser()
69 else:
70 base = _dir_from_env("XDG_DATA_HOME", ".local", "share")
71 return base / "lilbee"
74def default_state_dir() -> Path:
75 """Return platform-appropriate directory for live runtime state.
76 - macOS: ~/Library/Application Support/lilbee
77 - Windows: %LOCALAPPDATA%/lilbee
78 - Linux: ~/.local/state/lilbee (XDG_STATE_HOME)
80 Deliberately not a cache directory. This holds the machine engine slot: the
81 state files recording a running llama-swap's pid and ports, the refcount
82 lock dir, and the build lock. Those records are the only handle any
83 out-of-process stop has on a running fleet, so a cleaner (or macOS evicting
84 ~/Library/Caches under disk pressure) emptying the dir mid-run would orphan
85 a fleet holding VRAM and leave the slot looking free to the next process,
86 which would then build a second fleet on top of it.
87 """
88 if sys.platform == "darwin":
89 base = Path.home() / "Library" / "Application Support"
90 elif sys.platform == "win32":
91 base = _dir_from_env("LOCALAPPDATA", "AppData", "Local").expanduser()
92 else:
93 base = _dir_from_env("XDG_STATE_HOME", ".local", "state")
94 return base / "lilbee"
97def default_cache_dir() -> Path:
98 """Return platform-appropriate directory for regenerable caches.
100 - macOS: ~/Library/Caches/lilbee
101 - Windows: %LOCALAPPDATA%/lilbee/cache
102 - Linux: ~/.cache/lilbee (XDG_CACHE_HOME)
104 The counterpart to :func:`default_state_dir`. Everything here is derived data
105 that costs time, not correctness, to lose, so a cleaner -- or macOS evicting
106 ~/Library/Caches under disk pressure -- may empty it freely. Nothing that a
107 stop path needs to find a running process belongs here.
108 """
109 if sys.platform == "darwin":
110 return Path.home() / "Library" / "Caches" / "lilbee"
111 if sys.platform == "win32":
112 base = _dir_from_env("LOCALAPPDATA", "AppData", "Local").expanduser()
113 return base / "lilbee" / "cache"
114 return _dir_from_env("XDG_CACHE_HOME", ".cache") / "lilbee"
117def find_local_root(start: Path | None = None) -> Path | None:
118 """Walk up from start (default: cwd) looking for a ``.lilbee/`` directory."""
119 start = start or Path.cwd()
120 for candidate in (start, *start.parents):
121 marker = candidate / LOCAL_ROOT_DIRNAME
122 if marker.is_dir():
123 return marker
124 return None
127def canonical_data_root(root: Path | str) -> Path:
128 """Resolve a data root to one canonical path.
130 Session file, port file, and write lock all derive from the data root, so
131 two spellings of one directory key two locks. Symlinks, relative paths, a
132 leading ``~``, and macOS ``/var`` vs ``/private/var`` each produce a pair.
133 A root that does not exist yet resolves to where it will be created.
135 Uses ``os.path`` rather than ``Path.expanduser().resolve()``: ``resolve``
136 rebuilds via ``type(self)``, which raises for a ``PosixPath`` that exists
137 on Windows (``Path()`` picks its flavour from ``os.name``, which tests patch).
138 """
139 return Path(os.path.realpath(os.path.expanduser(os.fspath(root))))
142class _Folder:
143 """One folder, with its listing read from the disk at most one time."""
145 def __init__(self, path: str) -> None:
146 self._path = path
148 @cached_property
149 def _names(self) -> list[str]:
150 return os.listdir(self._path)
152 @cached_property
153 def _listed(self) -> frozenset[str]:
154 return frozenset(self._names)
156 @cached_property
157 def _by_entry(self) -> dict[tuple[int, int], str]:
158 """The first listed name of each entry, by the device and inode ``lstat`` gives."""
159 by_entry: dict[tuple[int, int], str] = {}
160 for name in self._names:
161 entry = os.lstat(os.path.join(self._path, name))
162 by_entry.setdefault((entry.st_dev, entry.st_ino), name)
163 return by_entry
165 def listed_name(self, name: str) -> str:
166 """The name the listing gives the entry that *name* opens, else *name*."""
167 try:
168 if name in self._listed:
169 return name
170 opened = os.lstat(os.path.join(self._path, name))
171 return self._by_entry.get((opened.st_dev, opened.st_ino), name)
172 except OSError:
173 return name
176class DiskSpelling:
177 """Paths spelled as the disk spells them, with each folder read at most one time.
179 One instance serves one operation: it does not see a rename that follows a read.
180 """
182 def __init__(self) -> None:
183 self._folders: dict[str, _Folder] = {}
184 self._spelled: dict[Path, Path] = {}
186 def of(self, path: Path) -> Path:
187 """*path* with symlinks resolved and each part that exists spelled as its folder lists it.
189 Two spellings of one entry give one path, on a disk that ignores letter case
190 or Unicode form as on one that does not. A part that does not exist, or whose
191 folder cannot be listed, keeps the spelling it came with.
192 """
193 known = self._spelled.get(path)
194 if known is not None:
195 return known
196 anchor, *parts = Path(os.path.realpath(path)).parts
197 spelled = anchor
198 for part in parts:
199 folder = self._folders.setdefault(spelled, _Folder(spelled))
200 spelled = os.path.join(spelled, folder.listed_name(part))
201 self._spelled[path] = Path(spelled)
202 return self._spelled[path]
205def disk_path(path: Path) -> Path:
206 """*path* as :meth:`DiskSpelling.of` spells it, for a caller with one path to ask about."""
207 return DiskSpelling().of(path)
210def canonical_models_dir() -> Path:
211 """Return the shared models directory (always in the platform default, never per-project).
212 Multiple lilbee instances share this directory so models are downloaded once.
213 """
214 return default_data_dir() / "models"
217def is_ignored_dir(name: str, ignore_dirs: frozenset[str]) -> bool:
218 """Return True if a directory name should be skipped during traversal."""
219 return name.startswith(".") or name in ignore_dirs or name.endswith(".egg-info")
222_CTX_TIER_FLOOR = 8192
223_CTX_TIER_TABLE: tuple[tuple[int, int], ...] = (
224 # (total_bytes_threshold, target)
225 # The top tier matches AGENT_CHAT_CTX_FLOOR: a 128 GiB host is a server or
226 # pod whose GPUs can back an agent-sized window, and the dynamic picker
227 # still clamps to trained context and device memory where they cannot.
228 (128 * 1024**3, 65536),
229 (64 * 1024**3, 24576),
230 (32 * 1024**3, 16384),
231 (16 * 1024**3, 12288),
232)
235def chat_ctx_target_for_total_bytes(total_bytes: int) -> int:
236 """Pick a chat_n_ctx_target from total host RAM (floor 8192, tiers at 16/32/64/128 GiB)."""
237 if total_bytes <= 0:
238 return _CTX_TIER_FLOOR
239 for threshold, target in _CTX_TIER_TABLE:
240 if total_bytes >= threshold:
241 return target
242 return _CTX_TIER_FLOOR
245_CGROUP_ROOT = Path("/sys/fs/cgroup")
248def cgroup_memory_limit() -> int | None:
249 """Bytes this process's cgroup allows, or ``None`` when unlimited or unreadable.
251 cgroup v2 keeps the cap in ``memory.max`` (``max`` for unlimited); v1 uses
252 ``memory/memory.limit_in_bytes``, which spells unlimited as a near-int64
253 sentinel rather than a word, and so reads as a limit above installed RAM.
254 Both are read, matching the CPU quota reader in :mod:`lilbee.runtime.cpu`.
256 Every reader of host memory needs this: psutil reports the machine's
257 ``/proc/meminfo``, which a memory-capped container sees in full, so a 4 GiB
258 container on a 512 GiB machine sizes itself for the machine and is killed by
259 the OOM reaper on its first load.
260 """
261 for path in (_CGROUP_ROOT / "memory.max", _CGROUP_ROOT / "memory" / "memory.limit_in_bytes"):
262 try:
263 raw = path.read_text(encoding="utf-8").strip()
264 except OSError:
265 continue
266 if raw == "max":
267 return None
268 try:
269 return int(raw)
270 except ValueError:
271 return None
272 return None
275def cgroup_memory_used() -> int | None:
276 """Bytes this process's cgroup currently holds, or ``None`` when unreadable."""
277 for path in (
278 _CGROUP_ROOT / "memory.current",
279 _CGROUP_ROOT / "memory" / "memory.usage_in_bytes",
280 ):
281 try:
282 return int(path.read_text(encoding="utf-8").strip())
283 except (OSError, ValueError):
284 continue
285 return None
288def capped_total_memory() -> int:
289 """Total RAM this process may use in bytes; raises if the host cannot be read.
291 Bounded by the cgroup cap where one applies; a limit above installed RAM is
292 no limit at all, which is also how cgroup v1 spells unlimited.
293 """
294 import psutil
296 host_total = int(psutil.virtual_memory().total)
297 limit = cgroup_memory_limit()
298 return min(host_total, limit) if limit is not None else host_total
301def _read_total_memory_bytes() -> int:
302 """:func:`capped_total_memory`, or 0 when introspection is unavailable.
304 The config default needs an answer at import time and has a floor to fall
305 back to, so it swallows the failure. Callers sizing a real placement want the
306 exception instead: a budget silently computed from zero refuses every model
307 with no reason given.
308 """
309 try:
310 return capped_total_memory()
311 except Exception:
312 # psutil import or platform read failed; the caller falls back to the floor.
313 return 0
316def scaled_chat_ctx_target_default() -> int:
317 """Pick a chat_n_ctx_target from this host's total RAM at config-load time."""
318 return chat_ctx_target_for_total_bytes(_read_total_memory_bytes())
321# Filesystem types whose backing store is a network, where mmap page faults are
322# served over the wire and can wedge the model loader in uninterruptible I/O. The
323# exact type string a given volume reports (e.g. a RunPod network volume) is
324# confirmed on the target host and added here.
325_NETWORK_FS_TYPES = frozenset(
326 {"nfs", "nfs4", "cifs", "smb3", "smbfs", "9p", "ceph", "glusterfs", "lustre", "beegfs", "afs"}
327)
328# A /proc/mounts line is "device mountpoint fstype options ...": at least 3 fields.
329_PROC_MOUNTS_MIN_FIELDS = 3
330_PROC_MOUNTS = Path("/proc/mounts")
333def _mount_fstype(path: str, mounts_text: str) -> str:
334 """Filesystem type of the longest mount point in *mounts_text* that covers *path*."""
335 best_mount = ""
336 best_type = ""
337 for line in mounts_text.splitlines():
338 parts = line.split()
339 if len(parts) < _PROC_MOUNTS_MIN_FIELDS:
340 continue
341 mount_point, fs_type = parts[1], parts[2]
342 covers = path == mount_point or path.startswith(mount_point.rstrip("/") + "/")
343 if covers and len(mount_point) >= len(best_mount):
344 best_mount, best_type = mount_point, fs_type
345 return best_type
348def is_network_path(path: Path) -> bool:
349 """Whether *path* lives on a network filesystem.
351 mmap over a network filesystem faults pages over the wire, which can stall a
352 large-model load in uninterruptible I/O. Linux-only (reads ``/proc/mounts``);
353 returns False on other platforms and on any read failure, so local disk is the
354 safe assumption.
355 """
356 try:
357 mounts_text = _PROC_MOUNTS.read_text(encoding="utf-8")
358 except OSError:
359 return False
360 try:
361 resolved = str(path.resolve())
362 except OSError:
363 resolved = str(path)
364 fstype = _mount_fstype(resolved, mounts_text)
365 return fstype in _NETWORK_FS_TYPES or fstype.startswith("fuse.")
368_EXTRA_BIN_DIRS: tuple[str, ...] = ("~/.local/bin", "~/.bun/bin")
369_UNIX_BIN_DIRS: tuple[str, ...] = ("/opt/homebrew/bin", "/usr/local/bin")
370_WINDOWS_BIN_DIRS: tuple[str, ...] = ("~/AppData/Roaming/npm", "~/AppData/Local/Programs")
373def executable_search_path() -> str:
374 """PATH plus the directories user-level installers put executables in.
376 A server started from a desktop session inherits the login PATH, which
377 misses the package-manager and per-user install dirs a shell profile adds.
378 """
379 platform_dirs = _WINDOWS_BIN_DIRS if sys.platform == "win32" else _UNIX_BIN_DIRS
380 entries = [os.environ.get("PATH", "")]
381 entries += [str(Path(d).expanduser()) for d in (*_EXTRA_BIN_DIRS, *platform_dirs)]
382 return os.pathsep.join(entry for entry in entries if entry)
385def find_executable(name: str) -> str | None:
386 """Absolute path to the *name* executable, or None when it is not installed."""
387 return shutil.which(name, path=executable_search_path())