Coverage for src/lilbee/cli/tui/__init__.py: 100%

1"""Textual TUI for lilbee -- full-screen interactive knowledge base.""" 

2 

3from __future__ import annotations 

4 

5import io 

6import logging 

7import os 

8import sys 

9from dataclasses import dataclass 

10from pathlib import Path 

11 

12from lilbee.app.services import exit_when_stragglers_would_hang, reset_services_on_exit 

13from lilbee.cli.tui.log_routing import setup_tui_log_file 

14from lilbee.core.security import OWNER_ONLY_MODE 

15 

16 

17def _silence_stderr_log_handlers() -> None: 

18 """Drop stderr/stdout-streaming log handlers before mounting the TUI. (bb-82ce) 

19 

20 lilbee logs route to ``cfg.data_root/logs/tui.log`` for the duration of 

21 the TUI session via :func:`setup_tui_log_file`. FileHandler instances 

22 are skipped explicitly here so that file route survives. 

23 """ 

24 root = logging.getLogger() 

25 for handler in list(root.handlers): 

26 if not isinstance(handler, logging.StreamHandler): 

27 continue 

28 if isinstance(handler, logging.FileHandler): 

29 continue 

30 if handler.stream in (sys.stderr, sys.stdout): 

31 root.removeHandler(handler) 

32 

33 

34@dataclass 

35class _StderrRedirect: 

36 """State captured by ``_redirect_native_stderr_to`` for restoration on exit.""" 

37 

38 saved_fd: int 

39 saved_sys_stderr: object 

40 saved_sys_dunder_stderr: object 

41 

42 

43def _redirect_native_stderr_to(log_path: Path) -> _StderrRedirect | None: 

44 """Send native fd-2 writes to *log_path* without breaking Textual's render. 

45 

46 Textual's Linux/macOS driver writes its alternate-screen ANSI to 

47 ``sys.__stderr__`` (see 

48 ``textual.drivers.linux_driver.LinuxDriver.write``). Native deps 

49 like xberg's vendored tesseract write directly to fd 2 and leak 

50 onto the same buffer, e.g. "Detected N diacritics", which corrupts 

51 the TUI. 

52 

53 Strategy: dup the original fd 2 to a saved fd, repoint 

54 ``sys.__stderr__`` and ``sys.stderr`` at that saved fd so Textual 

55 keeps drawing to the real terminal, then dup2 fd 2 itself to the 

56 log file so any fd-2 writer (xberg, tesseract, poppler) lands 

57 in ``tui.log`` instead of on top of the screen. 

58 """ 

59 try: 

60 log_fd = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, OWNER_ONLY_MODE) 

61 except OSError: 

62 return None 

63 saved_fd = os.dup(2) 

64 terminal_stderr = io.TextIOWrapper( 

65 io.FileIO(saved_fd, "w", closefd=False), 

66 encoding=sys.stderr.encoding or "utf-8", 

67 errors="replace", 

68 write_through=True, 

69 ) 

70 saved_sys_stderr = sys.stderr 

71 saved_sys_dunder_stderr = sys.__stderr__ 

72 sys.stderr = terminal_stderr 

73 sys.__stderr__ = terminal_stderr # type: ignore[misc] 

74 os.dup2(log_fd, 2) 

75 os.close(log_fd) 

76 return _StderrRedirect( 

77 saved_fd=saved_fd, 

78 saved_sys_stderr=saved_sys_stderr, 

79 saved_sys_dunder_stderr=saved_sys_dunder_stderr, 

80 ) 

81 

82 

83def _restore_native_stderr(redirect: _StderrRedirect | None) -> None: 

84 """Undo ``_redirect_native_stderr_to``; safe to call when *redirect* is None.""" 

85 if redirect is None: 

86 return 

87 os.dup2(redirect.saved_fd, 2) 

88 os.close(redirect.saved_fd) 

89 sys.stderr = redirect.saved_sys_stderr # type: ignore[assignment] 

90 sys.__stderr__ = redirect.saved_sys_dunder_stderr # type: ignore[misc,assignment] 

91 

92 

93def _prestart_mp_resource_tracker() -> None: 

94 """Start the multiprocessing resource tracker while stderr has a real descriptor. 

95 

96 Textual replaces ``sys.stderr`` with a stream whose ``fileno()`` is -1, and a 

97 tracker launched after that crashes with ``bad value(s) in fds_to_keep``. 

98 Frozen builds run it too: ``__main__._dispatch_frozen_child`` intercepts the 

99 tracker's ``-c`` reinvocation. Windows has no tracker. 

100 """ 

101 if sys.platform == "win32": 

102 return 

103 try: 

104 from multiprocessing import resource_tracker 

105 

106 resource_tracker.ensure_running() 

107 except (OSError, RuntimeError, ValueError, ImportError): 

108 # Best effort: the first worker spawn surfaces a real error at call time. 

109 pass 

110 

111 

112def run_tui(*, initial_view: str | None = None) -> None: 

113 """Launch the full-screen Textual TUI. 

114 

115 *initial_view* deep-links to a named view (e.g. ``"Catalog"``) after 

116 the default chat screen is mounted. Used by ``lilbee model browse``. 

117 """ 

118 # heavy: cli.sync transitively imports ingest -> store -> pyarrow (~1.8s 

119 # cold-start on bare runners); only needed at TUI shutdown. (bb-oae5) 

120 from lilbee.cli.sync import shutdown_executor 

121 from lilbee.cli.tui.app import LilbeeApp 

122 

123 _prestart_mp_resource_tracker() 

124 log_path = setup_tui_log_file() 

125 _silence_stderr_log_handlers() 

126 stderr_redirect = _redirect_native_stderr_to(log_path) 

127 

128 app = LilbeeApp(initial_view=initial_view) 

129 try: 

130 app.run() 

131 except KeyboardInterrupt: 

132 pass # Ctrl-C exits the TUI; cleanup runs in the finally block 

133 finally: 

134 # Tasks stop through their own cancel while the executors and services live. 

135 app.task_bar.stop_all() 

136 _restore_native_stderr(stderr_redirect) 

137 shutdown_executor() 

138 reset_services_on_exit() 

139 exit_when_stragglers_would_hang()