Coverage for src/lilbee/runtime/splash.py: 100%
79 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-17 10:02 +0000
1"""Splash animation lifecycle: starts and stops the animation subprocess.
3The animation itself lives in ``_splash_runner.py`` (stdlib-only, zero lilbee
4imports). This module manages the subprocess, pipe-based IPC, and cleanup.
6IPC uses an OS pipe: parent holds the write end, child polls the read end.
7When the parent closes the write end (or dies), the child sees EOF and exits.
8This guarantees no orphan processes: the OS closes the pipe on parent death.
9"""
11from __future__ import annotations
13import atexit
14import contextlib
15import os
16import subprocess
17import sys
18from dataclasses import dataclass
19from typing import Any
21from lilbee.runtime._splash_runner import TAKEOVER_BYTE
23_SPLASH_FD_ENV = "_LILBEE_SPLASH_FD"
25_SHOW_CURSOR = "\033[?25h"
27_STOP_TIMEOUT = 3.0
30@dataclass
31class SplashHandle:
32 """Opaque handle returned by ``start()`` for use with ``stop()``."""
34 process: subprocess.Popen[bytes]
35 write_fd: int
38_active_handle: SplashHandle | None = None
41def _should_skip() -> bool:
42 """Return True when the splash animation should be suppressed."""
43 if not os.isatty(2):
44 return True
45 return bool(os.environ.get("LILBEE_NO_SPLASH", ""))
48def _spawn_args(read_fd: int) -> tuple[str, dict[str, Any]]:
49 """Per-platform argv reference and Popen kwargs for the pipe's read end.
51 POSIX passes the fd itself; pass_fds keeps only read_fd open in the child
52 (close_fds=False would leak all open descriptors, including any held by
53 libraries). subprocess cannot pass fds on Windows, so the child gets the
54 pipe's OS handle number on argv instead and reopens it as a fd
55 (msvcrt.open_osfhandle in the runner; inherited handles keep their value
56 in the child). handle_list restricts inheritance to exactly this handle,
57 the same no-leak guarantee pass_fds gives on POSIX.
58 """
59 if sys.platform == "win32": # pragma: no cover - Windows-only, covered on the Windows CI leg
60 import msvcrt
62 handle = msvcrt.get_osfhandle(read_fd)
63 os.set_handle_inheritable(handle, True)
64 startupinfo = subprocess.STARTUPINFO()
65 startupinfo.lpAttributeList = {"handle_list": [handle]}
66 return str(handle), {"startupinfo": startupinfo}
67 os.set_inheritable(read_fd, True)
68 return str(read_fd), {"pass_fds": (read_fd,)}
71def start() -> SplashHandle | None:
72 """Launch the splash animation subprocess.
73 Returns a handle for ``stop()``, or None if the splash was skipped.
74 The caller must eventually call ``stop(handle)`` to clean up.
75 """
76 global _active_handle
78 if _should_skip():
79 return None
81 read_fd, write_fd = os.pipe()
82 pipe_ref, spawn_kwargs = _spawn_args(read_fd)
84 # Trusted: sys.executable is this interpreter, module path is static,
85 # the one runtime value (pipe_ref) derives from an int from os.pipe().
86 proc = subprocess.Popen( # noqa: S603
87 [sys.executable, "-m", "lilbee.runtime._splash_runner", pipe_ref],
88 stderr=None,
89 stdout=subprocess.DEVNULL,
90 stdin=subprocess.DEVNULL,
91 **spawn_kwargs,
92 )
94 os.close(read_fd)
96 os.environ[_SPLASH_FD_ENV] = str(write_fd)
98 handle = SplashHandle(process=proc, write_fd=write_fd)
99 _active_handle = handle
101 atexit.register(_atexit_cleanup)
103 return handle
106def stop(handle: SplashHandle | None) -> None:
107 """Stop the splash animation and wait for the subprocess to exit."""
108 global _active_handle
110 if handle is None:
111 return
113 _close_write_fd(handle.write_fd)
115 try:
116 handle.process.wait(timeout=_STOP_TIMEOUT)
117 except subprocess.TimeoutExpired:
118 handle.process.kill()
119 handle.process.wait(timeout=1.0)
121 os.environ.pop(_SPLASH_FD_ENV, None)
123 _active_handle = None
125 _restore_cursor()
128def dismiss() -> None:
129 """Signal the splash to stop from the TUI side.
130 Called once the TUI is ready to paint. Writes the takeover byte so the
131 subprocess exits without touching the terminal (its frame clear and
132 cursor-show would land on the Textual alt-screen and leave a visible
133 cursor for the whole session), then closes the pipe, waits for the
134 subprocess, and clears the active handle so ``atexit`` does not re-run
135 ``stop()``.
136 """
137 global _active_handle
139 fd_str = os.environ.pop(_SPLASH_FD_ENV, None)
140 if fd_str is not None:
141 _signal_takeover(int(fd_str))
142 _close_write_fd(int(fd_str))
144 handle = _active_handle
145 if handle is None:
146 return
147 _active_handle = None
149 # Signal and close the write end. This may double-signal/close the same
150 # fd that the env-var path already handled; both helpers suppress
151 # OSError so that is harmless.
152 # No _restore_cursor() here: we are inside Textual's alt-screen,
153 # where writing cursor-show would produce a visible artifact.
154 _signal_takeover(handle.write_fd)
155 _close_write_fd(handle.write_fd)
156 try:
157 handle.process.wait(timeout=_STOP_TIMEOUT)
158 except subprocess.TimeoutExpired:
159 handle.process.kill()
160 handle.process.wait(timeout=1.0)
163def _close_write_fd(fd: int) -> None:
164 """Close a pipe write fd, ignoring errors if already closed."""
165 with contextlib.suppress(OSError):
166 os.close(fd)
169def _signal_takeover(fd: int) -> None:
170 """Send the TUI-takeover byte, ignoring errors if the pipe is gone."""
171 with contextlib.suppress(OSError):
172 os.write(fd, TAKEOVER_BYTE)
175def _restore_cursor() -> None:
176 """Belt-and-suspenders cursor restore on stderr."""
177 try:
178 sys.stderr.write(_SHOW_CURSOR)
179 sys.stderr.flush()
180 except OSError:
181 pass # stderr may be closed during interpreter shutdown
184def _atexit_cleanup() -> None:
185 """Last-resort cleanup if stop() was never called."""
186 if _active_handle is not None:
187 stop(_active_handle)