Coverage for src/lilbee/cli/sync.py: 100%
121 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Background sync, executor management, and sync status for chat mode."""
3from __future__ import annotations
5import asyncio
6import threading
7from collections.abc import Callable
8from concurrent.futures import Future, ThreadPoolExecutor
9from typing import TYPE_CHECKING
11from rich.text import Text
13from lilbee.cli import theme
14from lilbee.cli.helpers import print_prefixed
15from lilbee.data.ingest import sync
16from lilbee.runtime.asyncio_loop import is_executor_shutdown
17from lilbee.runtime.console import PlainConsole
18from lilbee.runtime.progress import (
19 EventType,
20 ExtractEvent,
21 FileStartEvent,
22 OcrStartEvent,
23 ProgressEvent,
24 SyncDoneEvent,
25 SyncWaitEvent,
26)
28if TYPE_CHECKING:
29 from lilbee.runtime.progress import DetailedProgressCallback
32def _format_sync_summary(
33 added: int, updated: int, removed: int, failed: int, skipped: int = 0, relocated: int = 0
34) -> str | None:
35 """Format sync counts into a human-readable summary, or None if nothing changed."""
36 counts = {
37 "added": added,
38 "updated": updated,
39 "removed": removed,
40 "relocated": relocated,
41 "skipped": skipped,
42 "failed": failed,
43 }
44 parts = [f"{n} {label}" for label, n in counts.items() if n]
45 return ", ".join(parts) if parts else None
48def _print_file_start(con: PlainConsole, data: ProgressEvent) -> None:
49 if not isinstance(data, FileStartEvent):
50 raise TypeError(f"Expected FileStartEvent, got {type(data).__name__}")
51 con.print(
52 Text(
53 f"Syncing [{data.current_file}/{data.total_files}]: {data.file}",
54 style=theme.MUTED,
55 ),
56 soft_wrap=True,
57 )
60def _print_done(con: PlainConsole, data: ProgressEvent) -> None:
61 if not isinstance(data, SyncDoneEvent):
62 raise TypeError(f"Expected SyncDoneEvent, got {type(data).__name__}")
63 summary = _format_sync_summary(
64 data.added, data.updated, data.removed, data.failed, data.skipped, data.relocated
65 )
66 if summary:
67 con.print(f"Synced: {summary}", style=theme.MUTED)
70def _print_wait(con: PlainConsole, data: ProgressEvent) -> None:
71 if not isinstance(data, SyncWaitEvent):
72 raise TypeError(f"Expected SyncWaitEvent, got {type(data).__name__}")
73 con.print(data.message, style=theme.MUTED, soft_wrap=True)
76def _sync_progress_printer(con: PlainConsole) -> DetailedProgressCallback:
77 """Return a callback that prints one-line status for FILE_START, SYNC_WAIT and SYNC_DONE."""
78 handlers: dict[EventType, Callable[[PlainConsole, ProgressEvent], None]] = {
79 EventType.FILE_START: _print_file_start,
80 EventType.SYNC_WAIT: _print_wait,
81 EventType.SYNC_DONE: _print_done,
82 }
84 def _callback(event_type: EventType, data: ProgressEvent) -> None:
85 handler = handlers.get(event_type)
86 if handler is not None:
87 handler(con, data)
89 return _callback
92def file_start_printer(con: PlainConsole) -> DetailedProgressCallback:
93 """Return a callback that prints one line as each file starts, for output with no live bar."""
95 def _callback(event_type: EventType, data: ProgressEvent) -> None:
96 if event_type == EventType.FILE_START:
97 _print_file_start(con, data)
99 return _callback
102_bg_executor: ThreadPoolExecutor | None = None
105def _get_executor() -> ThreadPoolExecutor:
106 """Lazy-init a single-worker executor."""
107 global _bg_executor
108 if _bg_executor is None:
109 _bg_executor = ThreadPoolExecutor(max_workers=1)
110 return _bg_executor
113def shutdown_executor() -> None:
114 """Shut down the background executor without blocking.
115 Uses wait=False + cancel_futures to avoid blocking the main thread.
116 """
117 global _bg_executor
118 if _bg_executor is None:
119 return
121 _bg_executor.shutdown(wait=False, cancel_futures=True)
122 _bg_executor = None
125def _on_sync_done(con: PlainConsole, future: Future[object], *, chat_mode: bool = False) -> None:
126 """Callback attached to background sync futures: logs errors."""
127 exc = future.exception()
128 if exc is None:
129 return
130 if isinstance(exc, asyncio.CancelledError):
131 return
132 if is_executor_shutdown(exc):
133 return
134 if chat_mode:
135 print(f"Background sync error: {exc}")
136 else:
137 print_prefixed(con, "Background sync error: ", exc, style=theme.ERROR)
140class SyncStatus:
141 """Thread-safe holder for background sync status text.
142 The background sync callback writes here; prompt_toolkit's
143 ``bottom_toolbar`` reads it on every render cycle: no cursor
144 manipulation, no flickering.
145 """
147 def __init__(self) -> None:
148 self.text: str = ""
149 self.pending: int = 0
150 self._pending_lock = threading.Lock()
152 def clear(self) -> None:
153 self.text = ""
155 def adjust_pending(self, delta: int) -> None:
156 """Atomically change the queued-sync counter (mutated from two threads)."""
157 with self._pending_lock:
158 self.pending += delta
161def _tesseract_status(data: ProgressEvent) -> str:
162 """The status line for an OCR_START event: Tesseract running on the file's pages."""
163 if not isinstance(data, OcrStartEvent):
164 raise TypeError(f"Expected OcrStartEvent, got {type(data).__name__}")
165 return f"⟳ {data.status_text}"
168def _chat_status_line(event_type: EventType, data: ProgressEvent) -> str | None:
169 """The status line an event sets in chat mode, or None for an event that sets none."""
170 if event_type == EventType.FILE_START:
171 if not isinstance(data, FileStartEvent):
172 raise TypeError(f"Expected FileStartEvent, got {type(data).__name__}")
173 return f"⟳ Syncing [{data.current_file}/{data.total_files}]: {data.file}"
174 if event_type == EventType.EXTRACT:
175 if not isinstance(data, ExtractEvent):
176 raise TypeError(f"Expected ExtractEvent, got {type(data).__name__}")
177 return f"⟳ {data.step} [{data.page}/{data.total_pages}]: {data.file}"
178 if event_type == EventType.OCR_START:
179 return _tesseract_status(data)
180 if event_type == EventType.SYNC_WAIT:
181 if not isinstance(data, SyncWaitEvent):
182 raise TypeError(f"Expected SyncWaitEvent, got {type(data).__name__}")
183 return f"⟳ {data.message}"
184 return None
187def _chat_sync_callback(status: SyncStatus) -> DetailedProgressCallback:
188 """Return a progress callback for chat-mode background sync.
189 FILE_START updates *status.text* (rendered by prompt_toolkit's bottom
190 toolbar). On DONE the status is cleared and the summary is printed via
191 ``print()`` (goes through StdoutProxy → appears above the prompt).
192 """
193 status.clear()
195 def _callback(event_type: EventType, data: ProgressEvent) -> None:
196 line = _chat_status_line(event_type, data)
197 if line is not None:
198 queue_suffix = f" (+{status.pending} queued)" if status.pending > 0 else ""
199 status.text = line + queue_suffix
200 elif event_type == EventType.SYNC_DONE:
201 status.clear()
202 if not isinstance(data, SyncDoneEvent):
203 raise TypeError(f"Expected SyncDoneEvent, got {type(data).__name__}")
204 summary = _format_sync_summary(
205 data.added, data.updated, data.removed, data.failed, data.skipped, data.relocated
206 )
207 if summary:
208 print(f"✓ Synced: {summary}")
210 return _callback
213def run_sync_background(
214 con: PlainConsole,
215 *,
216 chat_mode: bool = False,
217 sync_status: SyncStatus | None = None,
218) -> Future[object]:
219 """Submit sync to a background thread. Returns the Future."""
220 status = sync_status or SyncStatus()
222 callback = _chat_sync_callback(status) if chat_mode else _sync_progress_printer(con)
224 def _run() -> object:
225 if chat_mode:
226 status.adjust_pending(-1)
227 return asyncio.run(sync(quiet=True, on_progress=callback, started_by="lilbee chat"))
229 if chat_mode:
230 status.adjust_pending(1)
232 future = _get_executor().submit(_run)
233 future.add_done_callback(lambda f: _on_sync_done(con, f, chat_mode=chat_mode))
234 return future