Coverage for src/lilbee/cli/tui/widgets/task

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.content import Content

13from textual.timer import Timer

14from textual.widgets import Label, Static

16from lilbee.catalog.formatting import display_label_for_ref

17from lilbee.cli.tui import messages as msg

18from lilbee.cli.tui.task_queue import Task, TaskQueue, TaskStatus

19from lilbee.cli.tui.widgets.task_bar_controller import TaskBarController

20from lilbee.providers.warm_progress import WarmPhase, WarmProgress

22if TYPE_CHECKING:

23 from lilbee.cli.tui.app import LilbeeApp

25log = logging.getLogger(__name__)

27_CSS_FILE = Path(__file__).parent / "task_bar.tcss"

29_DONE_FLASH_SECONDS = 2.0

30_POLL_INTERVAL_ACTIVE_S = 0.1

31_POLL_INTERVAL_IDLE_S = 1.0

33# Pulsing-dot cadence: on/off flip at half of this tick count.

34# 10 Hz poll x 5 = 500 ms per half cycle, which is a 1 Hz dot pulse,

35# matching the active-row rail pulse in the Task Center.

36_DOT_PULSE_HALF_TICKS = 5

37_DOT_GLYPH = "●"

39# Warm bar width; fill/track glyphs come from msg.progress_bar_glyphs() at

40# render time, matching the fleet panel bars.

41_WARM_BAR_WIDTH = 12

44# Indeterminate sweep (the engine-load phases have no byte signal): a lit window

45# that walks the track so the bar reads as "working", not stalled. Advanced by the

46# poll tick. Preferred over a spinner glyph: the moving bar reads as progress at a

47# glance, which is what a multi-second model load needs.

48_WARM_SWEEP_WIDTH = 3

51def _progress_bar(fraction: float) -> str:

52 """A determinate fill bar for the byte-progress (reading-weights) phase."""

53 fill, track = msg.progress_bar_glyphs()

54 filled = round(max(0.0, min(1.0, fraction)) * _WARM_BAR_WIDTH)

55 return fill * filled + track * (_WARM_BAR_WIDTH - filled)

58def _sweep_bar(tick: int) -> str:

59 """An indeterminate bar with a lit window of fixed width walking (and wrapping)

60 across the track, keyed to *tick*, so it always shows motion, never a blank."""

61 fill, track = msg.progress_bar_glyphs()

62 start = tick % _WARM_BAR_WIDTH

63 lit = {(start + offset) % _WARM_BAR_WIDTH for offset in range(_WARM_SWEEP_WIDTH)}

64 return "".join(fill if i in lit else track for i in range(_WARM_BAR_WIDTH))

67def _warm_detail(progress: WarmProgress | None, tick: int = 0) -> str | None:

68 """A moving bar plus the phase word for the warm line.

70 A determinate byte bar while paging weights, an indeterminate sweep while the

71 engine starts and loads (no byte signal). None once past an active phase, so

72 the caller drops the line entirely instead of showing a stalled indicator.

73 """

74 if progress is None:

75 return None

76 if progress.phase is WarmPhase.STARTING:

77 return f"{_sweep_bar(tick)} {msg.TASKBAR_WARM_STARTING}"

78 if progress.phase is WarmPhase.LOADING_ENGINE:

79 return f"{_sweep_bar(tick)} {msg.TASKBAR_WARM_LOADING}"

80 if progress.phase is WarmPhase.READING_WEIGHTS:

81 fraction = progress.bytes_done / progress.bytes_total if progress.bytes_total else 0.0

82 pct = int(fraction * 100)

83 return f"{_progress_bar(fraction)} {msg.TASKBAR_WARM_READING.format(pct=pct)}"

84 return None

87class TaskBar(Static):

88 """Slim 1-line status indicator for background tasks.

90 Shows a compact summary when tasks are active and hides when idle.

91 Detailed progress (spinners, progress bars, task panels) lives in

92 the Task Center screen, accessible via ``t``.

93 """

95 app: LilbeeApp # type: ignore[assignment]

97 # NOTE: no ``dock: bottom`` here. TaskBar is always mounted inside a

98 # ``BottomBars`` container that owns the dock; multiple dock-bottom

99 # siblings overlap at the same row in Textual (see BottomBars docstring).

100 DEFAULT_CSS: ClassVar[str] = _CSS_FILE.read_text(encoding="utf-8")

101

102 def __init__(self, **kwargs: object) -> None:

103 super().__init__(**kwargs) # type: ignore[arg-type]

104 self._tick_count = 0

105 # Timestamp (tick count) at which the current flash started.

106 # None when no flash is active. The 2 s completion/failure

107 # flash holds the coloured dot + summary past queue drain.

108 self._flash_until_tick: int | None = None

109 self._flash_outcome: TaskStatus | None = None

110 # Failures among the just-finished batch (not all of persistent history).

111 self._flash_failed_count: int = 0

112 # Task ids we've already flashed on. Task Center rows linger in

113 # history after DONE/FAILED/CANCELLED so the user can review

114 # recent work; without this gate the bar would re-flash the same

115 # task every poll because ``history[-1]`` keeps matching.

116 self._flashed_ids: set[str] = set()

117 # Fingerprint of the most recently painted label state. Each

118 # tick fires at 10 Hz; if nothing visible has changed (no new

119 # tasks, no progress shift, no pulse-phase flip) the heavy

120 # ``Label.update`` -- which re-segments + re-styles the line --

121 # is skipped. Visible idle cost drops from "every tick" to "on

122 # actual change", recovering ~5-8 ms/sec on idle screens.

123 self._last_render_fingerprint: tuple[object, ...] | None = None

124 # Poll handle. Set in on_mount and cleared in on_unmount; declared

125 # here so on_unmount can read it directly without a getattr fallback.

126 self._interval: Timer | None = None

127 # True when no work is visible: the timer drops to 1 Hz here since

128 # the fingerprint cache short-circuits the render path. Flips

129 # back on the first non-idle event.

130 self._idle_mode: bool = True

131

132 def compose(self) -> ComposeResult:

133 yield Label("", id="task-status-label")

134

135 def on_mount(self) -> None:

136 self._refresh_display()

137 # Capture the handle so we can cancel the poll on unmount. Without

138 # this, a screen push/pop cycle leaves the previous TaskBar's

139 # interval firing against a detached widget, racing with the new

140 # TaskBar and occasionally setting ``display=False`` on the live

141 # instance. Start at the idle cadence; the first tick re-arms at

142 # 10 Hz if work is already in flight.

143 self._interval = self.set_interval(_POLL_INTERVAL_IDLE_S, self._tick)

144

145 def on_unmount(self) -> None:

146 if self._interval is not None:

147 self._interval.stop()

148 self._interval = None

149

150 @property

151 def _controller(self) -> TaskBarController:

152 return self.app.task_bar

153

154 @property

155 def queue(self) -> TaskQueue:

156 """Expose the shared queue for callers that iterate or advance it."""

157 return self._controller.queue

158

159 def add_task(

160 self,

161 name: str,

162 task_type: str,

163 fn: Callable[[], None] | None = None,

164 *,

165 indeterminate: bool = False,

166 ) -> str:

167 """Enqueue a task via the app's controller. Returns the task_id."""

168 return self._controller.add_task(name, task_type, fn, indeterminate=indeterminate)

169

170 def update_task(

171 self,

172 task_id: str,

173 progress: float,

174 detail: str = "",

175 *,

176 indeterminate: bool | None = None,

177 ) -> None:

178 self._controller.update_task(task_id, progress, detail, indeterminate=indeterminate)

179

180 def complete_task(self, task_id: str) -> None:

181 self._controller.complete_task(task_id)

182

183 def fail_task(self, task_id: str, detail: str = "") -> None:

184 self._controller.fail_task(task_id, detail)

185

186 def cancel_task(self, task_id: str) -> None:

187 self._controller.cancel_task(task_id)

188

189 def _tick(self) -> None:

190 """Poll the shared queue and re-render."""

191 self._tick_count += 1

192 self._refresh_display()

193

194 def _sync_poll_cadence(self, fully_idle: bool) -> None:

195 """Re-arm the poll timer at idle/active cadence on state transitions."""

196 if fully_idle == self._idle_mode:

197 return

198 self._idle_mode = fully_idle

199 if self._interval is not None:

200 self._interval.stop()

201 interval = _POLL_INTERVAL_IDLE_S if fully_idle else _POLL_INTERVAL_ACTIVE_S

202 self._interval = self.set_interval(interval, self._tick)

203

204 def _refresh_display(self) -> None:

205 """Rebuild the 1-line status label from the shared queue.

206

207 Visual language:

208 - Leading ``●`` pulses ``$primary`` <-> ``$primary-lighten-2`` at 1 Hz

209 when anything is active. Dim ``$text-muted`` when only queued tasks

210 remain, ``$success`` during a completion flash, ``$error`` during

211 a failure flash.

212 - The text either reads ``{name} {pct}`` (one active, zero queued),

213 ``{N} tasks running`` (plural), ``{N} queued`` (throttle mode),

214 or the flash copy.

215 - Right-aligned muted-italic ``Press t for Tasks`` hint.

216 """

217 queue = self.queue

218 active = queue.active_tasks

219 queued = queue.queued_tasks

220 history = queue.history

221

222 # Drop flashed-id entries for tasks the user has cleared from

223 # history. Without this prune, the set grows unbounded over a

224 # long session even though any id not in history can't re-flash.

225 if self._flashed_ids:

226 live_ids = {t.task_id for t in history}

227 self._flashed_ids &= live_ids

228

229 in_flash = self._flash_until_tick is not None and self._tick_count <= self._flash_until_tick

230 if not in_flash:

231 self._flash_until_tick = None

232 self._flash_outcome = None

233 # Flash on the freshest completion that hasn't been flashed

234 # yet. History now persists (rows show as DONE in Task

235 # Center until cleared), so we must gate by task_id instead

236 # of "history is non-empty".

237 if not active and not queued and history:

238 new_done = [

239 t

240 for t in history

241 if t.task_id not in self._flashed_ids

242 and t.status in (TaskStatus.DONE, TaskStatus.FAILED)

243 ]

244 if new_done:

245 for t in new_done:

246 self._flashed_ids.add(t.task_id)

247 self._flash_until_tick = self._tick_count + int(

248 _DONE_FLASH_SECONDS / _POLL_INTERVAL_ACTIVE_S

249 )

250 self._flash_failed_count = sum(

251 1 for t in new_done if t.status == TaskStatus.FAILED

252 )

253 self._flash_outcome = (

254 TaskStatus.FAILED if self._flash_failed_count else TaskStatus.DONE

255 )

256

257 cancelling = self._controller.cancelling_tasks()

258 idle = (

259 not active

260 and not queued

261 and not cancelling

262 and not in_flash

263 and self._flash_outcome is None

264 )

265 pending = self._controller.pending_sync_count if idle else 0

266 spawning_roles = sorted(self._controller.spawning_roles) if idle else []

267 # Computed regardless of task activity: a chat warm holds the user's input

268 # disabled, so it outranks a passive task summary (which stays in the Task

269 # Center). Hiding it behind a background sync left a dead input unexplained.

270 warm_line = self._warm_line()

271 fully_idle = idle and pending == 0 and not spawning_roles and warm_line is None

272 self._sync_poll_cadence(fully_idle)

273 if fully_idle:

274 self.display = False

275 self._last_render_fingerprint = None

276 return

277

278 self.display = True

279 dot_color, summary = self._status_line(

280 active, queued, spawning_roles, pending, warm_line, idle=idle, cancelling=cancelling

281 )

282 hint_text = self._hint_copy()

283 # Fingerprint captures every variable the label content depends

284 # on. Recomputing it is essentially free; the win comes from

285 # skipping ``Label.update`` when nothing visible has changed,

286 # since update re-segments and re-styles the whole line.

287 fingerprint: tuple[object, ...] = (

288 dot_color,

289 summary,

290 hint_text,

291 in_flash,

292 self._flash_outcome,

293 pending,

294 tuple(spawning_roles),

295 warm_line,

296 )

297 if fingerprint == self._last_render_fingerprint:

298 return

299 self._last_render_fingerprint = fingerprint

300

301 label_text = Content.assemble(

302 " ",

303 Content.from_markup(f"[{dot_color}]{_DOT_GLYPH}[/]"),

304 " ",

305 summary,

306 " ",

307 Content.from_markup(f"[i dim]{hint_text}[/]"),

308 )

309 with contextlib.suppress(Exception):

310 label = self.query_one("#task-status-label", Label)

311 label.update(label_text)

312

313 def _status_line(

314 self,

315 active: list, # type: ignore[type-arg]

316 queued: list, # type: ignore[type-arg]

317 spawning_roles: list[str],

318 pending: int,

319 warm_line: str | None,

320 *,

321 idle: bool,

322 cancelling: list[Task] | None = None,

323 ) -> tuple[str, Content]:

324 """Pick the dot color and summary text for the current bar state."""

325 if warm_line is not None:

326 return "$primary", Content(warm_line)

327 if idle and spawning_roles:

328 return "$primary", Content(self._spawning_workers_template(spawning_roles))

329 if idle and pending > 0:

330 template = self._pending_sync_template(pending)

331 return "$text-muted", Content.from_markup(template.format(count=pending))

332 return self._compose_segments(active, queued, cancelling or [])

333

334 def _warm_line(self) -> str | None:

335 """The cold-start chat warm line, or None when chat isn't warming."""

336 from lilbee.app.placement import active_chat_warm_progress

337

338 progress = active_chat_warm_progress()

339 detail = _warm_detail(progress, self._tick_count)

340 if detail is None:

341 return None

342 name = (

343 display_label_for_ref(progress.model_ref)

344 if progress is not None and progress.model_ref

345 else msg.TASKBAR_WARM_FALLBACK_NAME

346 )

347 return msg.TASKBAR_WARM_LINE.format(name=name, detail=detail)

348

349 def _spawning_workers_template(self, roles: list[str]) -> str:

350 """Render the active worker-warmup hint for the bottom bar."""

351 labels = ", ".join(role.replace("_", " ") for role in roles)

352 template = msg.TASKBAR_STARTING_WORKER if len(roles) == 1 else msg.TASKBAR_STARTING_WORKERS

353 return template.format(labels=labels)

354

355 def _pending_sync_template(self, pending: int) -> str:

356 """Pick the singular/plural hint, swapping in the Esc-prefixed copy

357 when a chat ``Input`` swallows printable characters before bindings fire.

358 """

359 from textual.widgets import Input

360

361 try:

362 input_focused = isinstance(self.app.focused, Input)

363 except Exception:

364 input_focused = False

365 if pending == 1:

366 return (

367 msg.TASKBAR_SYNC_PENDING_ONE_INPUT

368 if input_focused

369 else msg.TASKBAR_SYNC_PENDING_ONE

370 )

371 return (

372 msg.TASKBAR_SYNC_PENDING_PLURAL_INPUT

373 if input_focused

374 else msg.TASKBAR_SYNC_PENDING_PLURAL

375 )

376

377 def _hint_copy(self) -> str:

378 """Return the right-aligned hint, context-aware.

379

380 When a chat ``Input`` (or similar) is focused the ``t`` keypress is

381 eaten before the app-level binding fires, so the user needs

382 ``Esc then t``. Every other screen (wizard grid, catalog,

383 settings, task center) lets ``t`` bubble, so a shorter ``Press t

384 for Tasks`` is accurate and easier to scan.

385 """

386 from textual.widgets import Input

387

388 try:

389 focused = self.app.focused

390 except Exception:

391 return msg.TASKBAR_HINT

392 if isinstance(focused, Input):

393 return msg.TASKBAR_HINT_INPUT

394 return msg.TASKBAR_HINT

395

396 def _compose_segments(

397 self, active: list, queued: list, cancelling: list[Task] | None = None

398 ) -> tuple[str, Content]:

399 """Return (dot color, text summary) for the current state."""

400 # Pulsing even/odd cadence, shared with TaskRow's rail pulse.

401 on_beat = (self._tick_count // _DOT_PULSE_HALF_TICKS) % 2 == 0

402

403 if self._flash_outcome == TaskStatus.DONE:

404 return "$success", Content.from_markup(msg.TASKBAR_ALL_DONE)

405 if self._flash_outcome == TaskStatus.FAILED:

406 count = self._flash_failed_count

407 key = msg.TASKBAR_FAILED if count == 1 else msg.TASKBAR_FAILED_PLURAL

408 return "$error", Content.from_markup(key.format(count=count))

409

410 parts: list[Content] = []

411 if active:

412 count = len(active)

413 task = active[0]

414 if count == 1 and not queued:

415 pct = "" if task.indeterminate else f" [b]{task.progress:.1f}%[/b]"

416 parts.append(Content.from_markup(f"[b]$name[/b]{pct}", name=task.name))

417 else:

418 key = msg.TASKBAR_ONE if count == 1 else msg.TASKBAR_MULTIPLE

419 parts.append(Content.from_markup(key.format(count=count)))

420 parts.append(Content.from_markup("[b]$name[/b]", name=task.name))

421 if queued:

422 queued_text = msg.TASKBAR_QUEUED_COUNT.format(count=len(queued))

423 parts.append(Content.from_markup(f"[dim]{queued_text}[/dim]"))

424 for task in cancelling or []:

425 parts.append(

426 Content.from_markup(

427 "$label [b]$name[/b]", label=msg.TASKBAR_CANCELLING, name=task.name

428 )

429 )

430

431 dot_color = ("$primary" if on_beat else "$primary-lighten-2") if active else "$text-muted"

432 return dot_color, Content(" · ").join(parts)

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

200 statements