Coverage for src/lilbee/cli/sync.py: 100%

1"""Background sync, executor management, and sync status for chat mode.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import threading 

7from collections.abc import Callable 

8from concurrent.futures import Future, ThreadPoolExecutor 

9from typing import TYPE_CHECKING 

10 

11from rich.console import Console 

12 

13from lilbee.cli import theme 

14from lilbee.data.ingest import sync 

15from lilbee.runtime.asyncio_loop import is_executor_shutdown 

16from lilbee.runtime.progress import ( 

17 EventType, 

18 ExtractEvent, 

19 FileStartEvent, 

20 ProgressEvent, 

21 SyncDoneEvent, 

22) 

23 

24if TYPE_CHECKING: 

25 from lilbee.runtime.progress import DetailedProgressCallback 

26 

27 

28def _format_sync_summary( 

29 added: int, updated: int, removed: int, failed: int, skipped: int = 0, relocated: int = 0 

30) -> str | None: 

31 """Format sync counts into a human-readable summary, or None if nothing changed.""" 

32 counts = { 

33 "added": added, 

34 "updated": updated, 

35 "removed": removed, 

36 "relocated": relocated, 

37 "skipped": skipped, 

38 "failed": failed, 

39 } 

40 parts = [f"{n} {label}" for label, n in counts.items() if n] 

41 return ", ".join(parts) if parts else None 

42 

43 

44def _print_file_start(con: Console, data: ProgressEvent) -> None: 

45 if not isinstance(data, FileStartEvent): 

46 raise TypeError(f"Expected FileStartEvent, got {type(data).__name__}") 

47 m = theme.MUTED 

48 con.print(f"[{m}]Syncing [{data.current_file}/{data.total_files}]: {data.file}[/{m}]") 

49 

50 

51def _print_done(con: Console, data: ProgressEvent) -> None: 

52 if not isinstance(data, SyncDoneEvent): 

53 raise TypeError(f"Expected SyncDoneEvent, got {type(data).__name__}") 

54 summary = _format_sync_summary( 

55 data.added, data.updated, data.removed, data.failed, data.skipped, data.relocated 

56 ) 

57 if summary: 

58 con.print(f"[{theme.MUTED}]Synced: {summary}[/{theme.MUTED}]") 

59 

60 

61def _sync_progress_printer(con: Console) -> DetailedProgressCallback: 

62 """Return a callback that prints one-line status for FILE_START and DONE events.""" 

63 handlers: dict[EventType, Callable[[Console, ProgressEvent], None]] = { 

64 EventType.FILE_START: _print_file_start, 

65 EventType.DONE: _print_done, 

66 } 

67 

68 def _callback(event_type: EventType, data: ProgressEvent) -> None: 

69 handler = handlers.get(event_type) 

70 if handler is not None: 

71 handler(con, data) 

72 

73 return _callback 

74 

75 

76_bg_executor: ThreadPoolExecutor | None = None 

77 

78 

79def _get_executor() -> ThreadPoolExecutor: 

80 """Lazy-init a single-worker executor.""" 

81 global _bg_executor 

82 if _bg_executor is None: 

83 _bg_executor = ThreadPoolExecutor(max_workers=1) 

84 return _bg_executor 

85 

86 

87def shutdown_executor() -> None: 

88 """Shut down the background executor without blocking. 

89 Uses wait=False + cancel_futures to avoid blocking the main thread. 

90 """ 

91 global _bg_executor 

92 if _bg_executor is None: 

93 return 

94 

95 _bg_executor.shutdown(wait=False, cancel_futures=True) 

96 _bg_executor = None 

97 

98 

99def _on_sync_done(con: Console, future: Future[object], *, chat_mode: bool = False) -> None: 

100 """Callback attached to background sync futures: logs errors.""" 

101 exc = future.exception() 

102 if exc is None: 

103 return 

104 if isinstance(exc, asyncio.CancelledError): 

105 return 

106 if is_executor_shutdown(exc): 

107 return 

108 if chat_mode: 

109 print(f"Background sync error: {exc}") 

110 else: 

111 con.print(f"[{theme.ERROR}]Background sync error:[/{theme.ERROR}] {exc}") 

112 

113 

114class SyncStatus: 

115 """Thread-safe holder for background sync status text. 

116 The background sync callback writes here; prompt_toolkit's 

117 ``bottom_toolbar`` reads it on every render cycle: no cursor 

118 manipulation, no flickering. 

119 """ 

120 

121 def __init__(self) -> None: 

122 self.text: str = "" 

123 self.pending: int = 0 

124 self._pending_lock = threading.Lock() 

125 

126 def clear(self) -> None: 

127 self.text = "" 

128 

129 def adjust_pending(self, delta: int) -> None: 

130 """Atomically change the queued-sync counter (mutated from two threads).""" 

131 with self._pending_lock: 

132 self.pending += delta 

133 

134 

135def _chat_sync_callback(status: SyncStatus) -> DetailedProgressCallback: 

136 """Return a progress callback for chat-mode background sync. 

137 FILE_START updates *status.text* (rendered by prompt_toolkit's bottom 

138 toolbar). On DONE the status is cleared and the summary is printed via 

139 ``print()`` (goes through StdoutProxy → appears above the prompt). 

140 """ 

141 status.clear() 

142 

143 def _callback(event_type: EventType, data: ProgressEvent) -> None: 

144 queue_suffix = f" (+{status.pending} queued)" if status.pending > 0 else "" 

145 if event_type == EventType.FILE_START: 

146 if not isinstance(data, FileStartEvent): 

147 raise TypeError(f"Expected FileStartEvent, got {type(data).__name__}") 

148 status.text = ( 

149 f"⟳ Syncing [{data.current_file}/{data.total_files}]: {data.file}{queue_suffix}" 

150 ) 

151 elif event_type == EventType.EXTRACT: 

152 if not isinstance(data, ExtractEvent): 

153 raise TypeError(f"Expected ExtractEvent, got {type(data).__name__}") 

154 status.text = ( 

155 f"⟳ Vision OCR [{data.page}/{data.total_pages}]: {data.file}{queue_suffix}" 

156 ) 

157 elif event_type == EventType.DONE: 

158 status.clear() 

159 if not isinstance(data, SyncDoneEvent): 

160 raise TypeError(f"Expected SyncDoneEvent, got {type(data).__name__}") 

161 summary = _format_sync_summary( 

162 data.added, data.updated, data.removed, data.failed, data.skipped, data.relocated 

163 ) 

164 if summary: 

165 print(f"✓ Synced: {summary}") 

166 

167 return _callback 

168 

169 

170def run_sync_background( 

171 con: Console, 

172 *, 

173 chat_mode: bool = False, 

174 sync_status: SyncStatus | None = None, 

175) -> Future[object]: 

176 """Submit sync to a background thread. Returns the Future.""" 

177 status = sync_status or SyncStatus() 

178 

179 callback = _chat_sync_callback(status) if chat_mode else _sync_progress_printer(con) 

180 

181 def _run() -> object: 

182 if chat_mode: 

183 status.adjust_pending(-1) 

184 return asyncio.run(sync(quiet=True, on_progress=callback)) 

185 

186 if chat_mode: 

187 status.adjust_pending(1) 

188 

189 future = _get_executor().submit(_run) 

190 future.add_done_callback(lambda f: _on_sync_done(con, f, chat_mode=chat_mode)) 

191 return future