Coverage for src / lilbee / 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 os 

16import subprocess 

17import sys 

18from dataclasses import dataclass 

19 

20_SPLASH_FD_ENV = "_LILBEE_SPLASH_FD" 

21 

22_SHOW_CURSOR = "\033[?25h" 

23 

24_STOP_TIMEOUT = 3.0 

25 

26 

27@dataclass 

28class SplashHandle: 

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

30 

31 process: subprocess.Popen[bytes] 

32 write_fd: int 

33 

34 

35_active_handle: SplashHandle | None = None 

36 

37 

38def _should_skip() -> bool: 

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

40 if not os.isatty(2): 

41 return True 

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

43 

44 

45def start() -> SplashHandle | None: 

46 """Launch the splash animation subprocess. 

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

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

49 """ 

50 global _active_handle 

51 

52 if _should_skip(): 

53 return None 

54 

55 read_fd, write_fd = os.pipe() 

56 os.set_inheritable(read_fd, True) 

57 

58 # Trusted: sys.executable is this interpreter, module path is static, 

59 # the one runtime value (read_fd) is an int from os.pipe(). 

60 proc = subprocess.Popen( # noqa: S603 

61 [sys.executable, "-m", "lilbee._splash_runner", str(read_fd)], 

62 close_fds=False, 

63 stderr=None, 

64 stdout=subprocess.DEVNULL, 

65 stdin=subprocess.DEVNULL, 

66 ) 

67 

68 os.close(read_fd) 

69 

70 os.environ[_SPLASH_FD_ENV] = str(write_fd) 

71 

72 handle = SplashHandle(process=proc, write_fd=write_fd) 

73 _active_handle = handle 

74 

75 atexit.register(_atexit_cleanup) 

76 

77 return handle 

78 

79 

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

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

82 global _active_handle 

83 

84 if handle is None: 

85 return 

86 

87 _close_write_fd(handle.write_fd) 

88 

89 try: 

90 handle.process.wait(timeout=_STOP_TIMEOUT) 

91 except subprocess.TimeoutExpired: 

92 handle.process.kill() 

93 handle.process.wait(timeout=1.0) 

94 

95 os.environ.pop(_SPLASH_FD_ENV, None) 

96 

97 _active_handle = None 

98 

99 _restore_cursor() 

100 

101 

102def dismiss() -> None: 

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

104 Called by the chat screen's ``on_show()`` to dismiss the splash once 

105 the TUI is ready to paint. Closes the pipe so the subprocess sees EOF, 

106 waits for it to exit, and clears the active handle so ``atexit`` does 

107 not re-run ``stop()`` (which would write ``\\033[?25h`` into the 

108 Textual alt-screen and leave a cursor artifact at (0,0)). 

109 """ 

110 global _active_handle 

111 

112 fd_str = os.environ.pop(_SPLASH_FD_ENV, None) 

113 if fd_str is not None: 

114 _close_write_fd(int(fd_str)) 

115 

116 handle = _active_handle 

117 if handle is None: 

118 return 

119 _active_handle = None 

120 

121 # Close the write end so the subprocess sees EOF. This may double-close 

122 # the same fd that the env-var path already closed; _close_write_fd 

123 # suppresses OSError so that is harmless. 

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

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

126 _close_write_fd(handle.write_fd) 

127 try: 

128 handle.process.wait(timeout=_STOP_TIMEOUT) 

129 except subprocess.TimeoutExpired: 

130 handle.process.kill() 

131 handle.process.wait(timeout=1.0) 

132 

133 

134def _close_write_fd(fd: int) -> None: 

135 """Close a pipe write fd, ignoring errors if already closed.""" 

136 with contextlib.suppress(OSError): 

137 os.close(fd) 

138 

139 

140def _restore_cursor() -> None: 

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

142 try: 

143 sys.stderr.write(_SHOW_CURSOR) 

144 sys.stderr.flush() 

145 except OSError: 

146 pass 

147 

148 

149def _atexit_cleanup() -> None: 

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

151 if _active_handle is not None: 

152 stop(_active_handle)