Coverage for src/lilbee/cli/tui/screens/task_center.py: 100%
157 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +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 # The queue calls this under the controller lock, so it must not call the controller.
117 self.app.task_bar.queue.subscribe(self._on_queue_change)
118 # A worker's exit is not a queue change; the row leaves "cancelling" on this call.
119 self.app.task_bar.subscribe_work_end(self._refresh_rows)
120 if self._tick_timer is None:
121 self._tick_timer = self.set_interval(_TICK_INTERVAL_SECONDS, self._advance_tick)
122 self._refresh_rows()
124 def on_hide(self) -> None:
125 with contextlib.suppress(Exception):
126 self.app.task_bar.queue.unsubscribe(self._on_queue_change)
127 with contextlib.suppress(Exception):
128 self.app.task_bar.unsubscribe_work_end(self._refresh_rows)
129 if self._tick_timer is not None:
130 self._tick_timer.stop()
131 self._tick_timer = None
133 def _on_queue_change(self) -> None:
134 """Queue notification: post a thread-safe message to the screen."""
135 self.post_message(TaskQueueChanged())
137 def on_task_queue_changed(self, _event: TaskQueueChanged) -> None:
138 """Reconcile rows when the queue posts a change."""
139 self._refresh_rows()
141 def _focus_initial_row(self) -> None:
142 """Land initial focus on the topmost active/queued row.
144 Users open the Task Center to manage live work, not to review
145 history. Without this, focus lands on the first row regardless
146 of status, so an accidental ``c`` on a terminal row is a
147 no-op rather than a status flip.
149 Falls back to the first row if there are no active/queued
150 tasks; falls back to no-op if the screen has no rows at all.
151 """
152 queue = self.app.task_bar.queue
153 for task in queue.active_tasks + queue.queued_tasks:
154 row = self._rows.get(task.task_id)
155 if row is not None:
156 row.focus()
157 return
158 # No active/queued work: leave focus on whatever AUTO_FOCUS
159 # picked (the scroll container, or the first row if one exists).
161 def action_refresh_tasks(self) -> None:
162 """Manual refresh (r). The subscription drives most updates; this
163 gives the user a way to force a reconcile if anything ever drifts."""
164 self._refresh_rows()
166 def action_clear_history(self) -> None:
167 """Drop all DONE/FAILED/CANCELLED rows (bound to capital ``C``).
169 A cancelling task is not history yet: its worker still runs, so its row stays.
170 ``clear_history`` itself emits a notification so the subscription
171 triggers the row reconcile; no manual refresh needed here."""
172 self.app.task_bar.queue.clear_history(keep=self._cancelling_ids())
174 def action_cancel_task(self) -> None:
175 """Cancel the task whose row currently has focus.
177 Falls back to the first active task if no row has focus.
178 """
179 # cancel_task, not queue.cancel: only the controller aborts the transfer.
180 focused = self.focused
181 if isinstance(focused, TaskRow) and focused.id:
182 self.app.task_bar.cancel_task(focused.id.removeprefix("task-"))
183 return
184 active = self.app.task_bar.queue.active_task
185 if active is not None:
186 self.app.task_bar.cancel_task(active.task_id)
188 def action_cursor_down(self) -> None:
189 self.focus_next()
191 def action_cursor_up(self) -> None:
192 self.focus_previous()
194 def action_jump_top(self) -> None:
195 rows = list(self.query(TaskRow))
196 if rows:
197 rows[0].focus()
199 def action_jump_bottom(self) -> None:
200 rows = list(self.query(TaskRow))
201 if rows:
202 rows[-1].focus()
204 def _all_tasks(self) -> list[Task]:
205 """Tasks in display order: active first, then queued, then history."""
206 queue = self.app.task_bar.queue
207 return queue.active_tasks + queue.queued_tasks + list(reversed(queue.history))
209 def _cancelling_ids(self) -> set[str]:
210 """Ids of cancelled tasks whose worker still runs, from the record has_live_work reads."""
211 return {t.task_id for t in self.app.task_bar.cancelling_tasks()}
213 def _advance_tick(self) -> None:
214 """Bump the spinner frame and re-render counts + active row pulse."""
215 # The on_hide path stops this timer, but Textual sometimes drains
216 # one final tick after the screen leaves the top of the stack and
217 # before stop() takes effect (or while the stack is being torn
218 # down on app shutdown, which makes ``self.app.screen`` itself
219 # raise). Skip in either case so the tick can't hit a torn-down
220 # DOM.
221 try:
222 if self.app.screen is not self:
223 return
224 except ScreenStackError:
225 return
226 self._tick += 1
227 tasks = self._all_tasks()
228 cancelling = self._cancelling_ids()
229 for task in tasks:
230 row = self._rows.get(task.task_id)
231 if row is not None:
232 row.update(task, self._tick, cancelling=task.task_id in cancelling)
233 self._update_counts(tasks)
235 def _refresh_rows(self) -> None:
236 """Reconcile rows against the queue: add new, update existing, remove stale."""
237 container = self.query_one("#task-rows", VerticalScroll)
238 tasks = self._all_tasks()
239 cancelling = self._cancelling_ids()
240 seen: set[str] = set()
241 for task in tasks:
242 seen.add(task.task_id)
243 row = self._rows.get(task.task_id)
244 if row is None:
245 row = TaskRow(task_id=task.task_id)
246 self._rows[task.task_id] = row
247 container.mount(row)
248 row.update(task, self._tick, cancelling=task.task_id in cancelling)
249 for tid in list(self._rows):
250 if tid not in seen:
251 row = self._rows.pop(tid)
252 try:
253 row.remove()
254 except Exception:
255 log.debug("Row %s already removed", tid, exc_info=True)
256 self._update_counts(tasks)
257 # Swap which widget occupies the 1fr row slot: scroll when
258 # there are tasks, headline when the list is empty. Hiding one
259 # of the pair (not both) keeps the empty-state headline centred
260 # in the available height instead of crowded under a ghost
261 # scroll that still claims the space.
262 empty = self.query_one("#task-center-empty", Label)
263 rows = self.query_one("#task-rows", VerticalScroll)
264 has_tasks = bool(tasks)
265 empty.display = not has_tasks
266 rows.display = has_tasks
268 def _update_counts(self, tasks: list[Task]) -> None:
269 """Top-right status strip: N running · M queued · K done.
271 Prepends a rotating spinner glyph when any task is active so
272 the header visibly moves. The rail pulse alone is too subtle
273 to communicate 'work in progress' at a glance (bb-18y3).
274 """
275 counts_label = self.query_one("#task-center-counts", Label)
276 counts: Counter[TaskStatus] = Counter(t.status for t in tasks)
277 active = counts[TaskStatus.ACTIVE]
278 queued = counts[TaskStatus.QUEUED]
279 done = counts[TaskStatus.DONE]
280 body = msg.TASK_CENTER_COUNTS.format(active=active, queued=queued, done=done)
281 cancelling = len(self._cancelling_ids())
282 if cancelling:
283 body += msg.TASK_CENTER_COUNTS_CANCELLING.format(cancelling=cancelling)
284 if active > 0:
285 spinner = _COUNTS_SPINNER_FRAMES[self._tick % len(_COUNTS_SPINNER_FRAMES)]
286 counts_label.update(f"{spinner} {body}")
287 else:
288 counts_label.update(body)