Coverage for src / lilbee / cli / tui / screens / task_center.py: 100%
138 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-05-15 20:55 +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, block-char bar) and a thick 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.task_queue import Task, TaskStatus
32from lilbee.cli.tui.widgets.task_row import TaskRow
34if TYPE_CHECKING:
35 from lilbee.cli.tui.app import LilbeeApp
37log = logging.getLogger(__name__)
39# Spinner advance cadence. Decoupled from queue-state refresh: queue
40# events drive _refresh_rows directly, this timer only advances the
41# rotating glyph and the active-row pulse so they keep moving while
42# the queue itself is idle.
43_TICK_INTERVAL_SECONDS = 0.25
46class TaskQueueChanged(Message):
47 """Posted by TaskCenter._on_queue_change when the queue notifies.
49 Posting a Textual Message is thread-safe, so the queue can call the
50 subscriber from any thread; the message is processed on the
51 screen's main-thread message pump.
52 """
55# Quarter-circle rotation cycles every 4 ticks (~0.4 s). Visible motion
56# in the counts strip confirms background work is live when rows are
57# running (bb-18y3).
58_COUNTS_SPINNER_FRAMES = ("◐", "◓", "◑", "◒")
61class TaskCenter(Screen[None]):
62 """Live view of active + queued + recently completed tasks."""
64 CSS_PATH = "task_center.tcss"
65 AUTO_FOCUS = "#task-rows"
66 HELP = "Background task monitor.\n\nPress r to refresh, c to cancel the focused task."
68 app: LilbeeApp # type: ignore[assignment]
70 BINDINGS: ClassVar[list[BindingType]] = [
71 Binding("q", "go_back", "Back", show=True),
72 Binding("escape", "go_back", "Back", show=False),
73 Binding("r", "refresh_tasks", "Refresh", show=True),
74 Binding("c", "cancel_task", "Cancel", show=True),
75 Binding("C", "clear_history", "Clear done", show=True),
76 Binding("j", "cursor_down", "Down", show=False),
77 Binding("k", "cursor_up", "Up", show=False),
78 ]
80 def compose(self) -> ComposeResult:
81 from lilbee.cli.tui.widgets.bottom_bars import BottomBars
82 from lilbee.cli.tui.widgets.status_bar import ViewTabs
83 from lilbee.cli.tui.widgets.task_bar import TaskBar
84 from lilbee.cli.tui.widgets.top_bars import TopBars
86 with TopBars():
87 yield ViewTabs()
88 yield Label(msg.TASK_CENTER_TITLE, id="task-center-title")
89 yield Label("", id="task-center-counts")
90 yield VerticalScroll(id="task-rows")
91 yield Label(
92 f"{msg.TASK_CENTER_EMPTY_HEADLINE}\n{msg.TASK_CENTER_EMPTY_DETAIL}",
93 id="task-center-empty",
94 )
95 with BottomBars():
96 yield Label(msg.TASK_CENTER_HINT, id="task-center-hint")
97 yield TaskBar()
98 yield Footer()
100 def action_go_back(self) -> None:
101 self.app.switch_view("Chat")
103 def on_mount(self) -> None:
104 self._tick: int = 0
105 self._rows: dict[str, TaskRow] = {}
106 self._tick_timer: Timer | None = None
107 self._refresh_rows()
108 self._focus_initial_row()
110 def on_show(self) -> None:
111 # Subscribe + tick only while visible; install_screen keeps this
112 # instance alive across switch_view, so anchoring either on
113 # on_mount would fire into a detached DOM after navigating away.
114 self.app.task_bar.queue.subscribe(self._on_queue_change)
115 if self._tick_timer is None:
116 self._tick_timer = self.set_interval(_TICK_INTERVAL_SECONDS, self._advance_tick)
117 self._refresh_rows()
119 def on_hide(self) -> None:
120 with contextlib.suppress(Exception):
121 self.app.task_bar.queue.unsubscribe(self._on_queue_change)
122 if self._tick_timer is not None:
123 self._tick_timer.stop()
124 self._tick_timer = None
126 def _on_queue_change(self) -> None:
127 """Queue notification: post a thread-safe message to the screen."""
128 self.post_message(TaskQueueChanged())
130 def on_task_queue_changed(self, _event: TaskQueueChanged) -> None:
131 """Reconcile rows when the queue posts a change."""
132 self._refresh_rows()
134 def _focus_initial_row(self) -> None:
135 """Land initial focus on the topmost active/queued row.
137 Users open the Task Center to manage live work, not to review
138 history. Without this, focus lands on the first row regardless
139 of status, so an accidental ``c`` on a terminal row is a
140 no-op rather than a status flip.
142 Falls back to the first row if there are no active/queued
143 tasks; falls back to no-op if the screen has no rows at all.
144 """
145 queue = self.app.task_bar.queue
146 for task in queue.active_tasks + queue.queued_tasks:
147 row = self._rows.get(task.task_id)
148 if row is not None:
149 row.focus()
150 return
151 # No active/queued work: leave focus on whatever AUTO_FOCUS
152 # picked (the scroll container, or the first row if one exists).
154 def action_refresh_tasks(self) -> None:
155 """Manual refresh (r). The subscription drives most updates; this
156 gives the user a way to force a reconcile if anything ever drifts."""
157 self._refresh_rows()
159 def action_clear_history(self) -> None:
160 """Drop all DONE/FAILED/CANCELLED rows (bound to capital ``C``).
162 ``clear_history`` itself emits a notification so the subscription
163 triggers the row reconcile; no manual refresh needed here."""
164 self.app.task_bar.queue.clear_history()
166 def action_cancel_task(self) -> None:
167 """Cancel the task whose row currently has focus.
169 Falls back to the first active task if no row has focus.
170 """
171 focused = self.focused
172 if isinstance(focused, TaskRow) and focused.id:
173 self.app.task_bar.queue.cancel(focused.id.removeprefix("task-"))
174 return
175 active = self.app.task_bar.queue.active_task
176 if active is not None:
177 self.app.task_bar.queue.cancel(active.task_id)
179 def action_cursor_down(self) -> None:
180 self.focus_next()
182 def action_cursor_up(self) -> None:
183 self.focus_previous()
185 def _all_tasks(self) -> list[Task]:
186 """Tasks in display order: active first, then queued, then history."""
187 queue = self.app.task_bar.queue
188 return queue.active_tasks + queue.queued_tasks + list(reversed(queue.history))
190 def _advance_tick(self) -> None:
191 """Bump the spinner frame and re-render counts + active row pulse."""
192 # The on_hide path stops this timer, but Textual sometimes drains
193 # one final tick after the screen leaves the top of the stack and
194 # before stop() takes effect (or while the stack is being torn
195 # down on app shutdown, which makes ``self.app.screen`` itself
196 # raise). Skip in either case so the tick can't hit a torn-down
197 # DOM.
198 try:
199 if self.app.screen is not self:
200 return
201 except ScreenStackError:
202 return
203 self._tick += 1
204 tasks = self._all_tasks()
205 for task in tasks:
206 row = self._rows.get(task.task_id)
207 if row is not None:
208 row.update(task, self._tick)
209 self._update_counts(tasks)
211 def _refresh_rows(self) -> None:
212 """Reconcile rows against the queue: add new, update existing, remove stale."""
213 container = self.query_one("#task-rows", VerticalScroll)
214 tasks = self._all_tasks()
215 seen: set[str] = set()
216 for task in tasks:
217 seen.add(task.task_id)
218 row = self._rows.get(task.task_id)
219 if row is None:
220 row = TaskRow(task_id=task.task_id)
221 self._rows[task.task_id] = row
222 container.mount(row)
223 row.update(task, self._tick)
224 for tid in list(self._rows):
225 if tid not in seen:
226 row = self._rows.pop(tid)
227 try:
228 row.remove()
229 except Exception:
230 log.debug("Row %s already removed", tid, exc_info=True)
231 self._update_counts(tasks)
232 # Swap which widget occupies the 1fr row slot: scroll when
233 # there are tasks, headline when the list is empty. Hiding one
234 # of the pair (not both) keeps the empty-state headline centred
235 # in the available height instead of crowded under a ghost
236 # scroll that still claims the space.
237 empty = self.query_one("#task-center-empty", Label)
238 rows = self.query_one("#task-rows", VerticalScroll)
239 has_tasks = bool(tasks)
240 empty.display = not has_tasks
241 rows.display = has_tasks
243 def _update_counts(self, tasks: list[Task]) -> None:
244 """Top-right status strip: N running · M queued · K done.
246 Prepends a rotating spinner glyph when any task is active so
247 the header visibly moves. The rail pulse alone is too subtle
248 to communicate 'work in progress' at a glance (bb-18y3).
249 """
250 counts_label = self.query_one("#task-center-counts", Label)
251 counts: Counter[TaskStatus] = Counter(t.status for t in tasks)
252 active = counts[TaskStatus.ACTIVE]
253 queued = counts[TaskStatus.QUEUED]
254 done = counts[TaskStatus.DONE]
255 body = msg.TASK_CENTER_COUNTS.format(active=active, queued=queued, done=done)
256 if active > 0:
257 spinner = _COUNTS_SPINNER_FRAMES[self._tick % len(_COUNTS_SPINNER_FRAMES)]
258 counts_label.update(f"{spinner} {body}")
259 else:
260 counts_label.update(body)