Coverage for src/lilbee/cli/tui/widgets/task_bar.py: 100%
194 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-17 10:02 +0000
1"""TaskBar widget: slim 1-line status indicator polling the shared TaskQueue."""
3from __future__ import annotations
5import contextlib
6import logging
7from collections.abc import Callable
8from pathlib import Path
9from typing import TYPE_CHECKING, ClassVar
11from textual.app import ComposeResult
12from textual.timer import Timer
13from textual.widgets import Label, Static
15from lilbee.catalog.formatting import display_label_for_ref
16from lilbee.cli.tui import messages as msg
17from lilbee.cli.tui.task_queue import TaskQueue, TaskStatus
18from lilbee.cli.tui.widgets.task_bar_controller import TaskBarController
19from lilbee.providers.warm_progress import WarmPhase, WarmProgress
21if TYPE_CHECKING:
22 from lilbee.cli.tui.app import LilbeeApp
24log = logging.getLogger(__name__)
26_CSS_FILE = Path(__file__).parent / "task_bar.tcss"
28_DONE_FLASH_SECONDS = 2.0
29_POLL_INTERVAL_ACTIVE_S = 0.1
30_POLL_INTERVAL_IDLE_S = 1.0
32# Pulsing-dot cadence: on/off flip at half of this tick count.
33# 10 Hz poll x 5 = 500 ms per half cycle, which is a 1 Hz dot pulse,
34# matching the active-row rail pulse in the Task Center.
35_DOT_PULSE_HALF_TICKS = 5
36_DOT_GLYPH = "●"
38# Warm bar width; fill/track glyphs come from msg.progress_bar_glyphs() at
39# render time, matching the fleet panel bars.
40_WARM_BAR_WIDTH = 12
43# Indeterminate sweep (the engine-load phases have no byte signal): a lit window
44# that walks the track so the bar reads as "working", not stalled. Advanced by the
45# poll tick. Preferred over a spinner glyph: the moving bar reads as progress at a
46# glance, which is what a multi-second model load needs.
47_WARM_SWEEP_WIDTH = 3
50def _progress_bar(fraction: float) -> str:
51 """A determinate fill bar for the byte-progress (reading-weights) phase."""
52 fill, track = msg.progress_bar_glyphs()
53 filled = round(max(0.0, min(1.0, fraction)) * _WARM_BAR_WIDTH)
54 return fill * filled + track * (_WARM_BAR_WIDTH - filled)
57def _sweep_bar(tick: int) -> str:
58 """An indeterminate bar with a lit window of fixed width walking (and wrapping)
59 across the track, keyed to *tick*, so it always shows motion, never a blank."""
60 fill, track = msg.progress_bar_glyphs()
61 start = tick % _WARM_BAR_WIDTH
62 lit = {(start + offset) % _WARM_BAR_WIDTH for offset in range(_WARM_SWEEP_WIDTH)}
63 return "".join(fill if i in lit else track for i in range(_WARM_BAR_WIDTH))
66def _warm_detail(progress: WarmProgress | None, tick: int = 0) -> str | None:
67 """A moving bar plus the phase word for the warm line.
69 A determinate byte bar while paging weights, an indeterminate sweep while the
70 engine starts and loads (no byte signal). None once past an active phase, so
71 the caller drops the line entirely instead of showing a stalled indicator.
72 """
73 if progress is None:
74 return None
75 if progress.phase is WarmPhase.STARTING:
76 return f"{_sweep_bar(tick)} {msg.TASKBAR_WARM_STARTING}"
77 if progress.phase is WarmPhase.LOADING_ENGINE:
78 return f"{_sweep_bar(tick)} {msg.TASKBAR_WARM_LOADING}"
79 if progress.phase is WarmPhase.READING_WEIGHTS:
80 fraction = progress.bytes_done / progress.bytes_total if progress.bytes_total else 0.0
81 pct = int(fraction * 100)
82 return f"{_progress_bar(fraction)} {msg.TASKBAR_WARM_READING.format(pct=pct)}"
83 return None
86class TaskBar(Static):
87 """Slim 1-line status indicator for background tasks.
89 Shows a compact summary when tasks are active and hides when idle.
90 Detailed progress (spinners, progress bars, task panels) lives in
91 the Task Center screen, accessible via ``t``.
92 """
94 app: LilbeeApp # type: ignore[assignment]
96 # NOTE: no ``dock: bottom`` here. TaskBar is always mounted inside a
97 # ``BottomBars`` container that owns the dock; multiple dock-bottom
98 # siblings overlap at the same row in Textual (see BottomBars docstring).
99 DEFAULT_CSS: ClassVar[str] = _CSS_FILE.read_text(encoding="utf-8")
101 def __init__(self, **kwargs: object) -> None:
102 super().__init__(**kwargs) # type: ignore[arg-type]
103 self._tick_count = 0
104 # Timestamp (tick count) at which the current flash started.
105 # None when no flash is active. The 2 s completion/failure
106 # flash holds the coloured dot + summary past queue drain.
107 self._flash_until_tick: int | None = None
108 self._flash_outcome: TaskStatus | None = None
109 # Failures among the just-finished batch (not all of persistent history).
110 self._flash_failed_count: int = 0
111 # Task ids we've already flashed on. Task Center rows linger in
112 # history after DONE/FAILED/CANCELLED so the user can review
113 # recent work; without this gate the bar would re-flash the same
114 # task every poll because ``history[-1]`` keeps matching.
115 self._flashed_ids: set[str] = set()
116 # Fingerprint of the most recently painted label state. Each
117 # tick fires at 10 Hz; if nothing visible has changed (no new
118 # tasks, no progress shift, no pulse-phase flip) the heavy
119 # ``Label.update`` -- which re-segments + re-styles the line --
120 # is skipped. Visible idle cost drops from "every tick" to "on
121 # actual change", recovering ~5-8 ms/sec on idle screens.
122 self._last_render_fingerprint: tuple[object, ...] | None = None
123 # Poll handle. Set in on_mount and cleared in on_unmount; declared
124 # here so on_unmount can read it directly without a getattr fallback.
125 self._interval: Timer | None = None
126 # True when no work is visible: the timer drops to 1 Hz here since
127 # the fingerprint cache short-circuits the render path. Flips
128 # back on the first non-idle event.
129 self._idle_mode: bool = True
131 def compose(self) -> ComposeResult:
132 yield Label("", id="task-status-label")
134 def on_mount(self) -> None:
135 self._refresh_display()
136 # Capture the handle so we can cancel the poll on unmount. Without
137 # this, a screen push/pop cycle leaves the previous TaskBar's
138 # interval firing against a detached widget, racing with the new
139 # TaskBar and occasionally setting ``display=False`` on the live
140 # instance. Start at the idle cadence; the first tick re-arms at
141 # 10 Hz if work is already in flight.
142 self._interval = self.set_interval(_POLL_INTERVAL_IDLE_S, self._tick)
144 def on_unmount(self) -> None:
145 if self._interval is not None:
146 self._interval.stop()
147 self._interval = None
149 @property
150 def _controller(self) -> TaskBarController:
151 return self.app.task_bar
153 @property
154 def queue(self) -> TaskQueue:
155 """Expose the shared queue for callers that iterate or advance it."""
156 return self._controller.queue
158 def add_task(
159 self,
160 name: str,
161 task_type: str,
162 fn: Callable[[], None] | None = None,
163 *,
164 indeterminate: bool = False,
165 ) -> str:
166 """Enqueue a task via the app's controller. Returns the task_id."""
167 return self._controller.add_task(name, task_type, fn, indeterminate=indeterminate)
169 def update_task(
170 self,
171 task_id: str,
172 progress: float,
173 detail: str = "",
174 *,
175 indeterminate: bool | None = None,
176 ) -> None:
177 self._controller.update_task(task_id, progress, detail, indeterminate=indeterminate)
179 def complete_task(self, task_id: str) -> None:
180 self._controller.complete_task(task_id)
182 def fail_task(self, task_id: str, detail: str = "") -> None:
183 self._controller.fail_task(task_id, detail)
185 def cancel_task(self, task_id: str) -> None:
186 self._controller.cancel_task(task_id)
188 def _tick(self) -> None:
189 """Poll the shared queue and re-render."""
190 self._tick_count += 1
191 self._refresh_display()
193 def _sync_poll_cadence(self, fully_idle: bool) -> None:
194 """Re-arm the poll timer at idle/active cadence on state transitions."""
195 if fully_idle == self._idle_mode:
196 return
197 self._idle_mode = fully_idle
198 if self._interval is not None:
199 self._interval.stop()
200 interval = _POLL_INTERVAL_IDLE_S if fully_idle else _POLL_INTERVAL_ACTIVE_S
201 self._interval = self.set_interval(interval, self._tick)
203 def _refresh_display(self) -> None:
204 """Rebuild the 1-line status label from the shared queue.
206 Visual language:
207 - Leading ``●`` pulses ``$primary`` <-> ``$primary-lighten-2`` at 1 Hz
208 when anything is active. Dim ``$text-muted`` when only queued tasks
209 remain, ``$success`` during a completion flash, ``$error`` during
210 a failure flash.
211 - The text either reads ``{name} {pct}`` (one active, zero queued),
212 ``{N} tasks running`` (plural), ``{N} queued`` (throttle mode),
213 or the flash copy.
214 - Right-aligned muted-italic ``Press t for Tasks`` hint.
215 """
216 queue = self.queue
217 active = queue.active_tasks
218 queued = queue.queued_tasks
219 history = queue.history
221 # Drop flashed-id entries for tasks the user has cleared from
222 # history. Without this prune, the set grows unbounded over a
223 # long session even though any id not in history can't re-flash.
224 if self._flashed_ids:
225 live_ids = {t.task_id for t in history}
226 self._flashed_ids &= live_ids
228 in_flash = self._flash_until_tick is not None and self._tick_count <= self._flash_until_tick
229 if not in_flash:
230 self._flash_until_tick = None
231 self._flash_outcome = None
232 # Flash on the freshest completion that hasn't been flashed
233 # yet. History now persists (rows show as DONE in Task
234 # Center until cleared), so we must gate by task_id instead
235 # of "history is non-empty".
236 if not active and not queued and history:
237 new_done = [
238 t
239 for t in history
240 if t.task_id not in self._flashed_ids
241 and t.status in (TaskStatus.DONE, TaskStatus.FAILED)
242 ]
243 if new_done:
244 for t in new_done:
245 self._flashed_ids.add(t.task_id)
246 self._flash_until_tick = self._tick_count + int(
247 _DONE_FLASH_SECONDS / _POLL_INTERVAL_ACTIVE_S
248 )
249 self._flash_failed_count = sum(
250 1 for t in new_done if t.status == TaskStatus.FAILED
251 )
252 self._flash_outcome = (
253 TaskStatus.FAILED if self._flash_failed_count else TaskStatus.DONE
254 )
256 idle = not active and not queued and not in_flash and self._flash_outcome is None
257 pending = self._controller.pending_sync_count if idle else 0
258 spawning_roles = sorted(self._controller.spawning_roles) if idle else []
259 # Computed regardless of task activity: a chat warm holds the user's input
260 # disabled, so it outranks a passive task summary (which stays in the Task
261 # Center). Hiding it behind a background sync left a dead input unexplained.
262 warm_line = self._warm_line()
263 fully_idle = idle and pending == 0 and not spawning_roles and warm_line is None
264 self._sync_poll_cadence(fully_idle)
265 if fully_idle:
266 self.display = False
267 self._last_render_fingerprint = None
268 return
270 self.display = True
271 dot_color, summary = self._status_line(
272 active, queued, spawning_roles, pending, warm_line, idle=idle
273 )
274 hint_text = self._hint_copy()
275 # Fingerprint captures every variable the label content depends
276 # on. Recomputing it is essentially free; the win comes from
277 # skipping ``Label.update`` when nothing visible has changed,
278 # since update re-segments and re-styles the whole line.
279 fingerprint: tuple[object, ...] = (
280 dot_color,
281 summary,
282 hint_text,
283 in_flash,
284 self._flash_outcome,
285 pending,
286 tuple(spawning_roles),
287 warm_line,
288 )
289 if fingerprint == self._last_render_fingerprint:
290 return
291 self._last_render_fingerprint = fingerprint
293 label_text = f" [{dot_color}]{_DOT_GLYPH}[/] {summary} [i dim]{hint_text}[/]"
294 with contextlib.suppress(Exception):
295 label = self.query_one("#task-status-label", Label)
296 label.update(label_text)
298 def _status_line(
299 self,
300 active: list, # type: ignore[type-arg]
301 queued: list, # type: ignore[type-arg]
302 spawning_roles: list[str],
303 pending: int,
304 warm_line: str | None,
305 *,
306 idle: bool,
307 ) -> tuple[str, str]:
308 """Pick the dot color and summary text for the current bar state."""
309 if warm_line is not None:
310 return "$primary", warm_line
311 if idle and spawning_roles:
312 return "$primary", self._spawning_workers_template(spawning_roles)
313 if idle and pending > 0:
314 return "$text-muted", self._pending_sync_template(pending).format(count=pending)
315 return self._compose_segments(active, queued)
317 def _warm_line(self) -> str | None:
318 """The cold-start chat warm line, or None when chat isn't warming."""
319 from lilbee.app.placement import active_chat_warm_progress
321 progress = active_chat_warm_progress()
322 detail = _warm_detail(progress, self._tick_count)
323 if detail is None:
324 return None
325 name = (
326 display_label_for_ref(progress.model_ref)
327 if progress is not None and progress.model_ref
328 else msg.TASKBAR_WARM_FALLBACK_NAME
329 )
330 return msg.TASKBAR_WARM_LINE.format(name=name, detail=detail)
332 def _spawning_workers_template(self, roles: list[str]) -> str:
333 """Render the active worker-warmup hint for the bottom bar."""
334 labels = ", ".join(role.replace("_", " ") for role in roles)
335 template = msg.TASKBAR_STARTING_WORKER if len(roles) == 1 else msg.TASKBAR_STARTING_WORKERS
336 return template.format(labels=labels)
338 def _pending_sync_template(self, pending: int) -> str:
339 """Pick the singular/plural hint, swapping in the Esc-prefixed copy
340 when a chat ``Input`` swallows printable characters before bindings fire.
341 """
342 from textual.widgets import Input
344 try:
345 input_focused = isinstance(self.app.focused, Input)
346 except Exception:
347 input_focused = False
348 if pending == 1:
349 return (
350 msg.TASKBAR_SYNC_PENDING_ONE_INPUT
351 if input_focused
352 else msg.TASKBAR_SYNC_PENDING_ONE
353 )
354 return (
355 msg.TASKBAR_SYNC_PENDING_PLURAL_INPUT
356 if input_focused
357 else msg.TASKBAR_SYNC_PENDING_PLURAL
358 )
360 def _hint_copy(self) -> str:
361 """Return the right-aligned hint, context-aware.
363 When a chat ``Input`` (or similar) is focused the ``t`` keypress is
364 eaten before the app-level binding fires, so the user needs
365 ``Esc then t``. Every other screen (wizard grid, catalog,
366 settings, task center) lets ``t`` bubble, so a shorter ``Press t
367 for Tasks`` is accurate and easier to scan.
368 """
369 from textual.widgets import Input
371 try:
372 focused = self.app.focused
373 except Exception:
374 return msg.TASKBAR_HINT
375 if isinstance(focused, Input):
376 return msg.TASKBAR_HINT_INPUT
377 return msg.TASKBAR_HINT
379 def _compose_segments(self, active: list, queued: list) -> tuple[str, str]:
380 """Return (dot color, text summary) for the current state."""
381 # Pulsing even/odd cadence, shared with TaskRow's rail pulse.
382 on_beat = (self._tick_count // _DOT_PULSE_HALF_TICKS) % 2 == 0
384 if self._flash_outcome == TaskStatus.DONE:
385 return "$success", msg.TASKBAR_ALL_DONE
386 if self._flash_outcome == TaskStatus.FAILED:
387 count = self._flash_failed_count
388 key = msg.TASKBAR_FAILED if count == 1 else msg.TASKBAR_FAILED_PLURAL
389 return "$error", key.format(count=count)
391 parts: list[str] = []
392 if active:
393 count = len(active)
394 task = active[0]
395 if count == 1 and not queued:
396 pct = "" if task.indeterminate else f" [b]{task.progress:.1f}%[/b]"
397 parts.append(f"[b]{task.name}[/b]{pct}")
398 else:
399 key = msg.TASKBAR_ONE if count == 1 else msg.TASKBAR_MULTIPLE
400 parts.append(key.format(count=count))
401 parts.append(f"[b]{task.name}[/b]")
402 if queued:
403 parts.append(f"[dim]{msg.TASKBAR_QUEUED_COUNT.format(count=len(queued))}[/dim]")
405 dot_color = ("$primary" if on_beat else "$primary-lighten-2") if active else "$text-muted"
406 return dot_color, " · ".join(parts)