Coverage for src/lilbee/cli/tui/screens/task_center.py: 100%
147 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Task Center screen: flight-deck-style background task monitor.
3Each task renders as a ``TaskRow`` with a three-line body (title +
4type, detail + percent, progress bar) and a heavy left rail in the
5state's color. On the active row the rail pulses at ~1 Hz, which is
6the only motion in the screen beyond the bar filling.
8State refresh is event-driven: the screen subscribes to ``TaskQueue``
9and ``_refresh_rows`` runs whenever a task is enqueued, advanced,
10updated, completed, or cancelled. A separate slow timer advances the
11spinner frame and the rail pulse so the visual heartbeat stays alive
12while the queue is idle.
13"""
15from __future__ import annotations
17import contextlib
18import logging
19from collections import Counter
20from typing import TYPE_CHECKING, ClassVar
22from textual.app import ComposeResult, ScreenStackError
23from textual.binding import Binding, BindingType
24from textual.containers import VerticalScroll
25from textual.message import Message
26from textual.screen import Screen
27from textual.timer import Timer
28from textual.widgets import Footer, Label
30from lilbee.cli.tui import messages as msg
31from lilbee.cli.tui.browse_bindings import BROWSE_LIST_BINDINGS, browse_back_bindings
32from lilbee.cli.tui.task_queue import Task, TaskStatus
33from lilbee.cli.tui.widgets.task_row import TaskRow
35if TYPE_CHECKING:
36 from lilbee.cli.tui.app import LilbeeApp
38log = logging.getLogger(__name__)
40# Spinner advance cadence. Decoupled from queue-state refresh: queue
41# events drive _refresh_rows directly, this timer only advances the
42# rotating glyph and the active-row pulse so they keep moving while
43# the queue itself is idle.
44_TICK_INTERVAL_SECONDS = 0.25
47class TaskQueueChanged(Message):
48 """Posted by TaskCenter._on_queue_change when the queue notifies.
50 Posting a Textual Message is thread-safe, so the queue can call the
51 subscriber from any thread; the message is processed on the
52 screen's main-thread message pump.
53 """
56# Quarter-circle rotation cycles every 4 ticks (~0.4 s). Visible motion
57# in the counts strip confirms background work is live when rows are
58# running (bb-18y3).
59_COUNTS_SPINNER_FRAMES = ("◐", "◓", "◑", "◒")
62class TaskCenter(Screen[None]):
63 """Live view of active + queued + recently completed tasks."""
65 CSS_PATH = "task_center.tcss"
66 AUTO_FOCUS = "#task-rows"
67 HELP = "Background task monitor.\n\nPress r to refresh, c to cancel the focused task."
69 app: LilbeeApp # type: ignore[assignment]
71 BINDINGS: ClassVar[list[BindingType]] = [
72 *browse_back_bindings(),
73 *BROWSE_LIST_BINDINGS,
74 Binding("r", "refresh_tasks", "Refresh", show=False),
75 # `c` shadows the app-wide jump to Chat here. Allowed because the
76 # footer names it: Cancel is the reason you are on this screen, and
77 # [ ] / q still leave. Never shadow `c` with a hidden binding.
78 Binding("c", "cancel_task", "Cancel", show=True),
79 Binding("C", "clear_history", "Clear done", show=False),
80 ]
82 def compose(self) -> ComposeResult:
83 from lilbee.cli.tui.widgets.bottom_bars import BottomBars
84 from lilbee.cli.tui.widgets.status_bar import ViewTabs
85 from lilbee.cli.tui.widgets.task_bar import TaskBar
86 from lilbee.cli.tui.widgets.top_bars import TopBars
88 with TopBars():
89 yield ViewTabs()
90 yield Label(msg.TASK_CENTER_TITLE, id="task-center-title")
91 yield Label("", id="task-center-counts")
92 yield VerticalScroll(id="task-rows")
93 yield Label(
94 f"{msg.TASK_CENTER_EMPTY_HEADLINE}\n{msg.TASK_CENTER_EMPTY_DETAIL}",
95 id="task-center-empty",
96 )
97 with BottomBars():
98 yield Label(msg.TASK_CENTER_HINT, id="task-center-hint")
99 yield TaskBar()
100 yield Footer()
102 def action_go_back(self) -> None:
103 self.app.go_back()
105 def on_mount(self) -> None:
106 self._tick: int = 0
107 self._rows: dict[str, TaskRow] = {}
108 self._tick_timer: Timer | None = None
109 self._refresh_rows()
110 self._focus_initial_row()
112 def on_show(self) -> None:
113 # Subscribe + tick only while visible; install_screen keeps this
114 # instance alive across switch_view, so anchoring either on
115 # on_mount would fire into a detached DOM after navigating away.
116 self.app.task_bar.queue.subscribe(self._on_queue_change)
117 if self._tick_timer is None:
118 self._tick_timer = self.set_interval(_TICK_INTERVAL_SECONDS, self._advance_tick)
119 self._refresh_rows()
121 def on_hide(self) -> None:
122 with contextlib.suppress(Exception):
123 self.app.task_bar.queue.unsubscribe(self._on_queue_change)
124 if self._tick_timer is not None:
125 self._tick_timer.stop()
126 self._tick_timer = None
128 def _on_queue_change(self) -> None:
129 """Queue notification: post a thread-safe message to the screen."""
130 self.post_message(TaskQueueChanged())
132 def on_task_queue_changed(self, _event: TaskQueueChanged) -> None:
133 """Reconcile rows when the queue posts a change."""
134 self._refresh_rows()
136 def _focus_initial_row(self) -> None:
137 """Land initial focus on the topmost active/queued row.
139 Users open the Task Center to manage live work, not to review
140 history. Without this, focus lands on the first row regardless
141 of status, so an accidental ``c`` on a terminal row is a
142 no-op rather than a status flip.
144 Falls back to the first row if there are no active/queued
145 tasks; falls back to no-op if the screen has no rows at all.
146 """
147 queue = self.app.task_bar.queue
148 for task in queue.active_tasks + queue.queued_tasks:
149 row = self._rows.get(task.task_id)
150 if row is not None:
151 row.focus()
152 return
153 # No active/queued work: leave focus on whatever AUTO_FOCUS
154 # picked (the scroll container, or the first row if one exists).
156 def action_refresh_tasks(self) -> None:
157 """Manual refresh (r). The subscription drives most updates; this
158 gives the user a way to force a reconcile if anything ever drifts."""
159 self._refresh_rows()
161 def action_clear_history(self) -> None:
162 """Drop all DONE/FAILED/CANCELLED rows (bound to capital ``C``).
164 ``clear_history`` itself emits a notification so the subscription
165 triggers the row reconcile; no manual refresh needed here."""
166 self.app.task_bar.queue.clear_history()
168 def action_cancel_task(self) -> None:
169 """Cancel the task whose row currently has focus.
171 Falls back to the first active task if no row has focus.
172 """
173 # cancel_task, not queue.cancel: only the controller aborts the transfer.
174 focused = self.focused
175 if isinstance(focused, TaskRow) and focused.id:
176 self.app.task_bar.cancel_task(focused.id.removeprefix("task-"))
177 return
178 active = self.app.task_bar.queue.active_task
179 if active is not None:
180 self.app.task_bar.cancel_task(active.task_id)
182 def action_cursor_down(self) -> None:
183 self.focus_next()
185 def action_cursor_up(self) -> None:
186 self.focus_previous()
188 def action_jump_top(self) -> None:
189 rows = list(self.query(TaskRow))
190 if rows:
191 rows[0].focus()
193 def action_jump_bottom(self) -> None:
194 rows = list(self.query(TaskRow))
195 if rows:
196 rows[-1].focus()
198 def _all_tasks(self) -> list[Task]:
199 """Tasks in display order: active first, then queued, then history."""
200 queue = self.app.task_bar.queue
201 return queue.active_tasks + queue.queued_tasks + list(reversed(queue.history))
203 def _advance_tick(self) -> None:
204 """Bump the spinner frame and re-render counts + active row pulse."""
205 # The on_hide path stops this timer, but Textual sometimes drains
206 # one final tick after the screen leaves the top of the stack and
207 # before stop() takes effect (or while the stack is being torn
208 # down on app shutdown, which makes ``self.app.screen`` itself
209 # raise). Skip in either case so the tick can't hit a torn-down
210 # DOM.
211 try:
212 if self.app.screen is not self:
213 return
214 except ScreenStackError:
215 return
216 self._tick += 1
217 tasks = self._all_tasks()
218 for task in tasks:
219 row = self._rows.get(task.task_id)
220 if row is not None:
221 row.update(task, self._tick)
222 self._update_counts(tasks)
224 def _refresh_rows(self) -> None:
225 """Reconcile rows against the queue: add new, update existing, remove stale."""
226 container = self.query_one("#task-rows", VerticalScroll)
227 tasks = self._all_tasks()
228 seen: set[str] = set()
229 for task in tasks:
230 seen.add(task.task_id)
231 row = self._rows.get(task.task_id)
232 if row is None:
233 row = TaskRow(task_id=task.task_id)
234 self._rows[task.task_id] = row
235 container.mount(row)
236 row.update(task, self._tick)
237 for tid in list(self._rows):
238 if tid not in seen:
239 row = self._rows.pop(tid)
240 try:
241 row.remove()
242 except Exception:
243 log.debug("Row %s already removed", tid, exc_info=True)
244 self._update_counts(tasks)
245 # Swap which widget occupies the 1fr row slot: scroll when
246 # there are tasks, headline when the list is empty. Hiding one
247 # of the pair (not both) keeps the empty-state headline centred
248 # in the available height instead of crowded under a ghost
249 # scroll that still claims the space.
250 empty = self.query_one("#task-center-empty", Label)
251 rows = self.query_one("#task-rows", VerticalScroll)
252 has_tasks = bool(tasks)
253 empty.display = not has_tasks
254 rows.display = has_tasks
256 def _update_counts(self, tasks: list[Task]) -> None:
257 """Top-right status strip: N running · M queued · K done.
259 Prepends a rotating spinner glyph when any task is active so
260 the header visibly moves. The rail pulse alone is too subtle
261 to communicate 'work in progress' at a glance (bb-18y3).
262 """
263 counts_label = self.query_one("#task-center-counts", Label)
264 counts: Counter[TaskStatus] = Counter(t.status for t in tasks)
265 active = counts[TaskStatus.ACTIVE]
266 queued = counts[TaskStatus.QUEUED]
267 done = counts[TaskStatus.DONE]
268 body = msg.TASK_CENTER_COUNTS.format(active=active, queued=queued, done=done)
269 if active > 0:
270 spinner = _COUNTS_SPINNER_FRAMES[self._tick % len(_COUNTS_SPINNER_FRAMES)]
271 counts_label.update(f"{spinner} {body}")
272 else:
273 counts_label.update(body)