Coverage for src/lilbee/runtime/splash.py: 100%

1"""Splash animation lifecycle: starts and stops the animation subprocess. 

2 

3The animation itself lives in ``_splash_runner.py`` (stdlib-only, zero lilbee 

4imports). This module manages the subprocess, pipe-based IPC, and cleanup. 

5 

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""" 

10 

11from __future__ import annotations 

12 

13import atexit 

14import contextlib 

15import io 

16import os 

17import subprocess 

18import sys 

19from dataclasses import dataclass 

20from typing import Any 

21 

22from lilbee.runtime._splash_runner import TAKEOVER_BYTE 

23 

24_SHOW_CURSOR = "\033[?25h" 

25 

26_STOP_TIMEOUT = 3.0 

27 

28 

29@dataclass 

30class SplashHandle: 

31 """Opaque handle returned by ``start()`` for use with ``stop()``.""" 

32 

33 process: subprocess.Popen[bytes] 

34 pipe: io.FileIO 

35 

36 

37_active_handle: SplashHandle | None = None 

38 

39 

40def _should_skip() -> bool: 

41 """Return True when the splash animation should be suppressed.""" 

42 if not os.isatty(2): 

43 return True 

44 return bool(os.environ.get("LILBEE_NO_SPLASH", "")) 

45 

46 

47def _spawn_args(read_fd: int) -> tuple[str, dict[str, Any]]: 

48 """Per-platform argv reference and Popen kwargs for the pipe's read end. 

49 

50 POSIX passes the fd itself; pass_fds keeps only read_fd open in the child 

51 (close_fds=False would leak all open descriptors, including any held by 

52 libraries). subprocess cannot pass fds on Windows, so the child gets the 

53 pipe's OS handle number on argv instead and reopens it as a fd 

54 (msvcrt.open_osfhandle in the runner; inherited handles keep their value 

55 in the child). handle_list restricts inheritance to exactly this handle, 

56 the same no-leak guarantee pass_fds gives on POSIX. 

57 """ 

58 if sys.platform == "win32": # pragma: no cover - Windows-only, covered on the Windows CI leg 

59 import msvcrt 

60 

61 handle = msvcrt.get_osfhandle(read_fd) 

62 os.set_handle_inheritable(handle, True) 

63 startupinfo = subprocess.STARTUPINFO() 

64 startupinfo.lpAttributeList = {"handle_list": [handle]} 

65 return str(handle), {"startupinfo": startupinfo} 

66 os.set_inheritable(read_fd, True) 

67 return str(read_fd), {"pass_fds": (read_fd,)} 

68 

69 

70def start() -> SplashHandle | None: 

71 """Launch the splash animation subprocess. 

72 Returns a handle for ``stop()``, or None if the splash was skipped. 

73 The caller must eventually call ``stop(handle)`` to clean up. 

74 """ 

75 global _active_handle 

76 

77 if _should_skip(): 

78 return None 

79 

80 read_fd, write_fd = os.pipe() 

81 try: 

82 pipe_ref, spawn_kwargs = _spawn_args(read_fd) 

83 

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 ) 

93 except BaseException: 

94 os.close(write_fd) 

95 raise 

96 finally: 

97 os.close(read_fd) 

98 

99 handle = SplashHandle(process=proc, pipe=io.FileIO(write_fd, "wb")) 

100 _active_handle = handle 

101 

102 atexit.register(_atexit_cleanup) 

103 

104 return handle 

105 

106 

107def stop(handle: SplashHandle | None) -> None: 

108 """Stop the splash animation and wait for the subprocess to exit.""" 

109 global _active_handle 

110 

111 if handle is None: 

112 return 

113 

114 handle.pipe.close() 

115 

116 try: 

117 handle.process.wait(timeout=_STOP_TIMEOUT) 

118 except subprocess.TimeoutExpired: 

119 handle.process.kill() 

120 handle.process.wait(timeout=1.0) 

121 

122 _active_handle = None 

123 

124 _restore_cursor() 

125 

126 

127def dismiss() -> None: 

128 """Signal the splash to stop from the TUI side. 

129 Called once the TUI is ready to paint. Writes the takeover byte so the 

130 subprocess exits without touching the terminal (its frame clear and 

131 cursor-show would land on the Textual alt-screen and leave a visible 

132 cursor for the whole session), then closes the pipe, waits for the 

133 subprocess, and clears the active handle so ``atexit`` does not re-run 

134 ``stop()``. 

135 """ 

136 global _active_handle 

137 

138 handle = _active_handle 

139 if handle is None: 

140 return 

141 _active_handle = None 

142 

143 # Closing the owned file is idempotent, so a repeat dismiss closes nothing. 

144 # No _restore_cursor() here: we are inside Textual's alt-screen, 

145 # where writing cursor-show would produce a visible artifact. 

146 _signal_takeover(handle.pipe) 

147 handle.pipe.close() 

148 try: 

149 handle.process.wait(timeout=_STOP_TIMEOUT) 

150 except subprocess.TimeoutExpired: 

151 handle.process.kill() 

152 handle.process.wait(timeout=1.0) 

153 

154 

155def _signal_takeover(pipe: io.FileIO) -> None: 

156 """Send the TUI-takeover byte, ignoring a pipe that is gone or closed.""" 

157 with contextlib.suppress(OSError, ValueError): 

158 pipe.write(TAKEOVER_BYTE) 

159 

160 

161def _restore_cursor() -> None: 

162 """Belt-and-suspenders cursor restore on stderr.""" 

163 try: 

164 sys.stderr.write(_SHOW_CURSOR) 

165 sys.stderr.flush() 

166 except OSError: 

167 pass # stderr may be closed during interpreter shutdown 

168 

169 

170def _atexit_cleanup() -> None: 

171 """Last-resort cleanup if stop() was never called.""" 

172 if _active_handle is not None: 

173 stop(_active_handle)