Coverage for src / lilbee / cli / tui / widgets / task_bar.py: 100%
156 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-05-15 20:55 +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.cli.tui import messages as msg
16from lilbee.cli.tui.task_queue import TaskQueue, TaskStatus
17from lilbee.cli.tui.widgets.task_bar_controller import TaskBarController
19if TYPE_CHECKING:
20 from lilbee.cli.tui.app import LilbeeApp
22log = logging.getLogger(__name__)
24_CSS_FILE = Path(__file__).parent / "task_bar.tcss"
26_DONE_FLASH_SECONDS = 2.0
27_POLL_INTERVAL_ACTIVE_S = 0.1
28_POLL_INTERVAL_IDLE_S = 1.0
30# Pulsing-dot cadence: on/off flip at half of this tick count.
31# 10 Hz poll x 5 = 500 ms per half cycle, which is a 1 Hz dot pulse,
32# matching the active-row rail pulse in the Task Center.
33_DOT_PULSE_HALF_TICKS = 5
34_DOT_GLYPH = "●"
37class TaskBar(Static):
38 """Slim 1-line status indicator for background tasks.
40 Shows a compact summary when tasks are active and hides when idle.
41 Detailed progress (spinners, progress bars, task panels) lives in
42 the Task Center screen, accessible via ``t``.
43 """
45 app: LilbeeApp # type: ignore[assignment]
47 # NOTE: no ``dock: bottom`` here. TaskBar is always mounted inside a
48 # ``BottomBars`` container that owns the dock; multiple dock-bottom
49 # siblings overlap at the same row in Textual (see BottomBars docstring).
50 DEFAULT_CSS: ClassVar[str] = _CSS_FILE.read_text(encoding="utf-8")
52 def __init__(self, **kwargs: object) -> None:
53 super().__init__(**kwargs) # type: ignore[arg-type]
54 self._tick_count = 0
55 # Timestamp (tick count) at which the current flash started.
56 # None when no flash is active. The 2 s completion/failure
57 # flash holds the coloured dot + summary past queue drain.
58 self._flash_until_tick: int | None = None
59 self._flash_outcome: TaskStatus | None = None
60 # Task ids we've already flashed on. Task Center rows linger in
61 # history after DONE/FAILED/CANCELLED so the user can review
62 # recent work; without this gate the bar would re-flash the same
63 # task every poll because ``history[-1]`` keeps matching.
64 self._flashed_ids: set[str] = set()
65 # Fingerprint of the most recently painted label state. Each
66 # tick fires at 10 Hz; if nothing visible has changed (no new
67 # tasks, no progress shift, no pulse-phase flip) the heavy
68 # ``Label.update`` -- which re-segments + re-styles the line --
69 # is skipped. Visible idle cost drops from "every tick" to "on
70 # actual change", recovering ~5-8 ms/sec on idle screens.
71 self._last_render_fingerprint: tuple[object, ...] | None = None
72 # Poll handle. Set in on_mount and cleared in on_unmount; declared
73 # here so on_unmount can read it directly without a getattr fallback.
74 self._interval: Timer | None = None
75 # True when no work is visible: the timer drops to 1 Hz here since
76 # the fingerprint cache short-circuits the render path. Flips
77 # back on the first non-idle event.
78 self._idle_mode: bool = True
80 def compose(self) -> ComposeResult:
81 yield Label("", id="task-status-label")
83 def on_mount(self) -> None:
84 self._refresh_display()
85 # Capture the handle so we can cancel the poll on unmount. Without
86 # this, a screen push/pop cycle leaves the previous TaskBar's
87 # interval firing against a detached widget, racing with the new
88 # TaskBar and occasionally setting ``display=False`` on the live
89 # instance. Start at the idle cadence; the first tick re-arms at
90 # 10 Hz if work is already in flight.
91 self._interval = self.set_interval(_POLL_INTERVAL_IDLE_S, self._tick)
93 def on_unmount(self) -> None:
94 if self._interval is not None:
95 self._interval.stop()
96 self._interval = None
98 @property
99 def _controller(self) -> TaskBarController:
100 return self.app.task_bar
102 @property
103 def queue(self) -> TaskQueue:
104 """Expose the shared queue for callers that iterate or advance it."""
105 return self._controller.queue
107 def add_task(
108 self,
109 name: str,
110 task_type: str,
111 fn: Callable[[], None] | None = None,
112 *,
113 indeterminate: bool = False,
114 ) -> str:
115 """Enqueue a task via the app's controller. Returns the task_id."""
116 return self._controller.add_task(name, task_type, fn, indeterminate=indeterminate)
118 def update_task(
119 self,
120 task_id: str,
121 progress: float,
122 detail: str = "",
123 *,
124 indeterminate: bool | None = None,
125 ) -> None:
126 self._controller.update_task(task_id, progress, detail, indeterminate=indeterminate)
128 def complete_task(self, task_id: str) -> None:
129 self._controller.complete_task(task_id)
131 def fail_task(self, task_id: str, detail: str = "") -> None:
132 self._controller.fail_task(task_id, detail)
134 def cancel_task(self, task_id: str) -> None:
135 self._controller.cancel_task(task_id)
137 def _tick(self) -> None:
138 """Poll the shared queue and re-render."""
139 self._tick_count += 1
140 self._refresh_display()
142 def _sync_poll_cadence(self, fully_idle: bool) -> None:
143 """Re-arm the poll timer at idle/active cadence on state transitions."""
144 if fully_idle == self._idle_mode:
145 return
146 self._idle_mode = fully_idle
147 if self._interval is not None:
148 self._interval.stop()
149 interval = _POLL_INTERVAL_IDLE_S if fully_idle else _POLL_INTERVAL_ACTIVE_S
150 self._interval = self.set_interval(interval, self._tick)
152 def _refresh_display(self) -> None:
153 """Rebuild the 1-line status label from the shared queue.
155 Visual language:
156 - Leading ``●`` pulses ``$primary`` <-> ``$primary-lighten-2`` at 1 Hz
157 when anything is active. Dim ``$text-muted`` when only queued tasks
158 remain, ``$success`` during a completion flash, ``$error`` during
159 a failure flash.
160 - The text either reads ``{name} {pct}`` (one active, zero queued),
161 ``{N} tasks running`` (plural), ``{N} queued`` (throttle mode),
162 or the flash copy.
163 - Right-aligned muted-italic ``Press t for Tasks`` hint.
164 """
165 queue = self.queue
166 active = queue.active_tasks
167 queued = queue.queued_tasks
168 history = queue.history
170 # Drop flashed-id entries for tasks the user has cleared from
171 # history. Without this prune, the set grows unbounded over a
172 # long session even though any id not in history can't re-flash.
173 if self._flashed_ids:
174 live_ids = {t.task_id for t in history}
175 self._flashed_ids &= live_ids
177 in_flash = self._flash_until_tick is not None and self._tick_count <= self._flash_until_tick
178 if not in_flash:
179 self._flash_until_tick = None
180 self._flash_outcome = None
181 # Flash on the freshest completion that hasn't been flashed
182 # yet. History now persists (rows show as DONE in Task
183 # Center until cleared), so we must gate by task_id instead
184 # of "history is non-empty".
185 if not active and not queued and history:
186 last = history[-1]
187 if last.task_id not in self._flashed_ids and last.status in (
188 TaskStatus.DONE,
189 TaskStatus.FAILED,
190 ):
191 self._flashed_ids.add(last.task_id)
192 self._flash_until_tick = self._tick_count + int(
193 _DONE_FLASH_SECONDS / _POLL_INTERVAL_ACTIVE_S
194 )
195 self._flash_outcome = last.status
197 idle = not active and not queued and not in_flash and self._flash_outcome is None
198 pending = self._controller.pending_sync_count if idle else 0
199 spawning_roles = sorted(self._controller.spawning_roles) if idle else []
200 fully_idle = idle and pending == 0 and not spawning_roles
201 self._sync_poll_cadence(fully_idle)
202 if fully_idle:
203 self.display = False
204 self._last_render_fingerprint = None
205 return
207 self.display = True
208 if idle and spawning_roles:
209 dot_color = "$primary"
210 summary = self._spawning_workers_template(spawning_roles)
211 elif idle and pending > 0:
212 dot_color = "$text-muted"
213 key = self._pending_sync_template(pending)
214 summary = key.format(count=pending)
215 else:
216 dot_color, summary = self._compose_segments(active, queued)
217 hint_text = self._hint_copy()
218 # Fingerprint captures every variable the label content depends
219 # on. Recomputing it is essentially free; the win comes from
220 # skipping ``Label.update`` when nothing visible has changed,
221 # since update re-segments and re-styles the whole line.
222 fingerprint: tuple[object, ...] = (
223 dot_color,
224 summary,
225 hint_text,
226 in_flash,
227 self._flash_outcome,
228 pending,
229 tuple(spawning_roles),
230 )
231 if fingerprint == self._last_render_fingerprint:
232 return
233 self._last_render_fingerprint = fingerprint
235 label_text = f" [{dot_color}]{_DOT_GLYPH}[/] {summary} [i dim]{hint_text}[/]"
236 with contextlib.suppress(Exception):
237 label = self.query_one("#task-status-label", Label)
238 label.update(label_text)
240 def _spawning_workers_template(self, roles: list[str]) -> str:
241 """Render the active worker-warmup hint for the bottom bar."""
242 labels = ", ".join(role.replace("_", " ") for role in roles)
243 template = msg.TASKBAR_STARTING_WORKER if len(roles) == 1 else msg.TASKBAR_STARTING_WORKERS
244 return template.format(labels=labels)
246 def _pending_sync_template(self, pending: int) -> str:
247 """Pick the singular/plural hint, swapping in the Esc-prefixed copy
248 when a chat ``Input`` swallows printable characters before bindings fire.
249 """
250 from textual.widgets import Input
252 try:
253 input_focused = isinstance(self.app.focused, Input)
254 except Exception:
255 input_focused = False
256 if pending == 1:
257 return (
258 msg.TASKBAR_SYNC_PENDING_ONE_INPUT
259 if input_focused
260 else msg.TASKBAR_SYNC_PENDING_ONE
261 )
262 return (
263 msg.TASKBAR_SYNC_PENDING_PLURAL_INPUT
264 if input_focused
265 else msg.TASKBAR_SYNC_PENDING_PLURAL
266 )
268 def _hint_copy(self) -> str:
269 """Return the right-aligned hint, context-aware.
271 When a chat ``Input`` (or similar) is focused the ``t`` keypress is
272 eaten before the app-level binding fires, so the user needs
273 ``Esc then t``. Every other screen (wizard grid, catalog,
274 settings, task center) lets ``t`` bubble, so a shorter ``Press t
275 for Tasks`` is accurate and easier to scan.
276 """
277 from textual.widgets import Input
279 try:
280 focused = self.app.focused
281 except Exception:
282 return msg.TASKBAR_HINT
283 if isinstance(focused, Input):
284 return msg.TASKBAR_HINT_INPUT
285 return msg.TASKBAR_HINT
287 def _compose_segments(self, active: list, queued: list) -> tuple[str, str]:
288 """Return (dot color, text summary) for the current state."""
289 # Pulsing even/odd cadence, shared with TaskRow's rail pulse.
290 on_beat = (self._tick_count // _DOT_PULSE_HALF_TICKS) % 2 == 0
292 if self._flash_outcome == TaskStatus.DONE:
293 return "$success", msg.TASKBAR_ALL_DONE
294 if self._flash_outcome == TaskStatus.FAILED:
295 count = sum(1 for t in self.queue.history if t.status == TaskStatus.FAILED)
296 key = msg.TASKBAR_FAILED if count == 1 else msg.TASKBAR_FAILED_PLURAL
297 return "$error", key.format(count=count)
299 parts: list[str] = []
300 if active:
301 count = len(active)
302 task = active[0]
303 if count == 1 and not queued:
304 pct = "" if task.indeterminate else f" [b]{task.progress:.1f}%[/b]"
305 parts.append(f"[b]{task.name}[/b]{pct}")
306 else:
307 key = msg.TASKBAR_ONE if count == 1 else msg.TASKBAR_MULTIPLE
308 parts.append(key.format(count=count))
309 parts.append(f"[b]{task.name}[/b]")
310 if queued:
311 parts.append(f"[dim]{msg.TASKBAR_QUEUED_COUNT.format(count=len(queued))}[/dim]")
313 dot_color = ("$primary" if on_beat else "$primary-lighten-2") if active else "$text-muted"
314 return dot_color, " · ".join(parts)