Coverage for src / lilbee / cli / tui / screens / task_center.py: 100%

1"""Task Center screen -- flight-deck-style background task monitor. 

2 

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. 

7 

8The render path is poll-based: ``_poll`` runs on the main thread at 

94 Hz, reads the shared ``TaskQueue``, and reconciles rows in place by 

10task_id. There's no subscriber chain; tasks owned by the controller 

11write into the lock-protected queue from worker threads and the poll 

12picks them up next tick. 

13""" 

14 

15from __future__ import annotations 

16 

17import logging 

18from typing import TYPE_CHECKING, ClassVar 

19 

20from textual.app import ComposeResult 

21from textual.binding import Binding, BindingType 

22from textual.containers import VerticalScroll 

23from textual.screen import Screen 

24from textual.widgets import Footer, Label 

25 

26from lilbee.cli.tui import messages as msg 

27from lilbee.cli.tui.task_queue import Task, TaskStatus 

28from lilbee.cli.tui.widgets.task_row import TaskRow 

29 

30if TYPE_CHECKING: 

31 from lilbee.cli.tui.app import LilbeeApp 

32 

33log = logging.getLogger(__name__) 

34 

35_POLL_INTERVAL_SECONDS = 0.25 

36 

37# Quarter-circle rotation cycles every 4 ticks (~0.4 s). Visible motion 

38# in the counts strip confirms background work is live when rows are 

39# running (bb-18y3). 

40_COUNTS_SPINNER_FRAMES = ("◐", "◓", "◑", "◒") 

41 

42 

43class TaskCenter(Screen[None]): 

44 """Live view of active + queued + recently completed tasks.""" 

45 

46 CSS_PATH = "task_center.tcss" 

47 AUTO_FOCUS = "#task-rows" 

48 HELP = "Background task monitor.\n\nPress r to refresh, c to cancel the focused task." 

49 

50 app: LilbeeApp 

51 

52 BINDINGS: ClassVar[list[BindingType]] = [ 

53 Binding("q", "go_back", "Back", show=True), 

54 Binding("escape", "go_back", "Back", show=False), 

55 Binding("r", "refresh_tasks", "Refresh", show=True), 

56 Binding("c", "cancel_task", "Cancel", show=True), 

57 Binding("C", "clear_history", "Clear done", show=True), 

58 Binding("j", "cursor_down", "Down", show=False), 

59 Binding("k", "cursor_up", "Up", show=False), 

60 ] 

61 

62 def compose(self) -> ComposeResult: 

63 from lilbee.cli.tui.widgets.bottom_bars import BottomBars 

64 from lilbee.cli.tui.widgets.status_bar import ViewTabs 

65 from lilbee.cli.tui.widgets.task_bar import TaskBar 

66 from lilbee.cli.tui.widgets.top_bars import TopBars 

67 

68 with TopBars(): 

69 yield ViewTabs() 

70 yield Label(msg.TASK_CENTER_TITLE, id="task-center-title") 

71 yield Label("", id="task-center-counts") 

72 yield VerticalScroll(id="task-rows") 

73 yield Label( 

74 f"{msg.TASK_CENTER_EMPTY_HEADLINE}\n{msg.TASK_CENTER_EMPTY_DETAIL}", 

75 id="task-center-empty", 

76 ) 

77 with BottomBars(): 

78 yield Label(msg.TASK_CENTER_HINT, id="task-center-hint") 

79 yield TaskBar() 

80 yield Footer() 

81 

82 def action_go_back(self) -> None: 

83 """Return to Chat (or pop if we're on a detached test app).""" 

84 from lilbee.cli.tui.app import LilbeeApp 

85 

86 if isinstance(self.app, LilbeeApp): # test apps aren't LilbeeApp 

87 self.app.switch_view("Chat") 

88 else: 

89 self.app.pop_screen() 

90 

91 def on_mount(self) -> None: 

92 self._tick: int = 0 

93 self._rows: dict[str, TaskRow] = {} 

94 self._poll() 

95 self._focus_initial_row() 

96 self.set_interval(_POLL_INTERVAL_SECONDS, self._poll) 

97 

98 def _focus_initial_row(self) -> None: 

99 """Land initial focus on the topmost active/queued row. 

100 

101 Users open the Task Center to manage live work, not to review 

102 history. Without this, focus lands on the first row regardless 

103 of status, so an accidental ``c`` on a terminal row is a 

104 no-op rather than a status flip. 

105 

106 Falls back to the first row if there are no active/queued 

107 tasks; falls back to no-op if the screen has no rows at all. 

108 """ 

109 queue = self.app.task_bar.queue 

110 for task in queue.active_tasks + queue.queued_tasks: 

111 row = self._rows.get(task.task_id) 

112 if row is not None: 

113 row.focus() 

114 return 

115 # No active/queued work: leave focus on whatever AUTO_FOCUS 

116 # picked (the scroll container, or the first row if one exists). 

117 

118 def action_refresh_tasks(self) -> None: 

119 """Manual refresh (r). No-op beyond forcing an immediate poll.""" 

120 self._poll() 

121 

122 def action_clear_history(self) -> None: 

123 """Drop all DONE/FAILED/CANCELLED rows (bound to capital ``C``).""" 

124 self.app.task_bar.queue.clear_history() 

125 self._poll() 

126 

127 def action_cancel_task(self) -> None: 

128 """Cancel the task whose row currently has focus. 

129 

130 Falls back to the first active task if no row has focus. 

131 """ 

132 focused = self.focused 

133 if isinstance(focused, TaskRow) and focused.id: 

134 self.app.task_bar.queue.cancel(focused.id.removeprefix("task-")) 

135 return 

136 active = self.app.task_bar.queue.active_task 

137 if active is not None: 

138 self.app.task_bar.queue.cancel(active.task_id) 

139 

140 def action_cursor_down(self) -> None: 

141 self.focus_next() 

142 

143 def action_cursor_up(self) -> None: 

144 self.focus_previous() 

145 

146 def _all_tasks(self) -> list[Task]: 

147 """Tasks in display order: active first, then queued, then history.""" 

148 queue = self.app.task_bar.queue 

149 return queue.active_tasks + queue.queued_tasks + list(reversed(queue.history)) 

150 

151 def _poll(self) -> None: 

152 """4 Hz reconciliation: add new rows, update existing, remove stale.""" 

153 self._tick += 1 

154 container = self.query_one("#task-rows", VerticalScroll) 

155 tasks = self._all_tasks() 

156 seen: set[str] = set() 

157 for task in tasks: 

158 seen.add(task.task_id) 

159 row = self._rows.get(task.task_id) 

160 if row is None: 

161 row = TaskRow(task_id=task.task_id) 

162 self._rows[task.task_id] = row 

163 container.mount(row) 

164 row.update(task, self._tick) 

165 for tid in list(self._rows): 

166 if tid not in seen: 

167 row = self._rows.pop(tid) 

168 try: 

169 row.remove() 

170 except Exception: 

171 log.debug("Row %s already removed", tid, exc_info=True) 

172 self._update_counts(tasks) 

173 # Swap which widget occupies the 1fr row slot: scroll when 

174 # there are tasks, headline when the list is empty. Hiding one 

175 # of the pair (not both) keeps the empty-state headline centred 

176 # in the available height instead of crowded under a ghost 

177 # scroll that still claims the space. 

178 empty = self.query_one("#task-center-empty", Label) 

179 rows = self.query_one("#task-rows", VerticalScroll) 

180 has_tasks = bool(tasks) 

181 empty.display = not has_tasks 

182 rows.display = has_tasks 

183 

184 def _update_counts(self, tasks: list[Task]) -> None: 

185 """Top-right status strip: N running · M queued · K done. 

186 

187 Prepends a rotating spinner glyph when any task is active so 

188 the header visibly moves. The rail pulse alone is too subtle 

189 to communicate 'work in progress' at a glance (bb-18y3). 

190 """ 

191 counts_label = self.query_one("#task-center-counts", Label) 

192 active = queued = done = 0 

193 for t in tasks: 

194 if t.status == TaskStatus.ACTIVE: 

195 active += 1 

196 elif t.status == TaskStatus.QUEUED: 

197 queued += 1 

198 elif t.status == TaskStatus.DONE: 

199 done += 1 

200 body = msg.TASK_CENTER_COUNTS.format(active=active, queued=queued, done=done) 

201 if active > 0: 

202 spinner = _COUNTS_SPINNER_FRAMES[self._tick % len(_COUNTS_SPINNER_FRAMES)] 

203 counts_label.update(f"{spinner} {body}") 

204 else: 

205 counts_label.update(body)