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

1"""Single-task row widget for the Task Center. 

2 

3Three lines per row. The head line uses the same ``pill()`` treatment 

4as the model cards so the screen matches the rest of the app; the 

5left rail carries the state color and pulses at 1 Hz for the active 

6row. The widget is pure-presentation: ``update(task, tick)`` writes 

7the three labels from a ``Task`` snapshot. ``TaskCenter._poll`` calls 

8it at 10 Hz. 

9""" 

10 

11from __future__ import annotations 

12 

13from time import monotonic 

14 

15from textual.app import ComposeResult 

16from textual.content import Content 

17from textual.widget import Widget 

18from textual.widgets import Label, Static 

19 

20from lilbee.cli.tui import messages as msg 

21from lilbee.cli.tui.pill import pill 

22from lilbee.cli.tui.task_queue import Task, TaskStatus, TaskType 

23from lilbee.cli.tui.widgets.progress_cell import ( 

24 frozen_indeterminate_cell, 

25 indeterminate_cell, 

26 progress_cell, 

27) 

28 

29# ~1.7 Hz rail pulse at a 10 Hz poll cadence = 3 ticks on, 3 off. 

30# Faster cadence than the original 1 Hz makes 'something is happening' 

31# visibly obvious at a glance (bb-18y3). 

32_PULSE_HALF_TICKS = 3 

33 

34_STATUS_CLASS: dict[TaskStatus, str] = { 

35 TaskStatus.QUEUED: "-queued", 

36 TaskStatus.ACTIVE: "-active", 

37 TaskStatus.DONE: "-done", 

38 TaskStatus.FAILED: "-failed", 

39 TaskStatus.CANCELLED: "-cancelled", 

40} 

41 

42_STATUS_CLASSES: tuple[str, ...] = tuple(_STATUS_CLASS.values()) 

43 

44# Pill palette: background color per task type. Sync/add/remove/import share 

45# $secondary (data-mutating ops), download uses $accent (network), wiki 

46# uses $warning (CPU-heavy generation), crawl uses $primary (external), 

47# analyze uses its own green (reads only). 

48_TASK_TYPE_BG: dict[str, str] = { 

49 TaskType.DOWNLOAD.value: "$accent", 

50 TaskType.SYNC.value: "$secondary", 

51 TaskType.ADD.value: "$secondary", 

52 TaskType.REMOVE.value: "$secondary", 

53 TaskType.IMPORT.value: "$secondary", 

54 TaskType.EXPORT.value: "$primary", 

55 TaskType.CRAWL.value: "$primary", 

56 TaskType.ANALYZE.value: "$success-darken-2", 

57 TaskType.WIKI.value: "$warning", 

58 TaskType.SETUP.value: "$warning-darken-1", 

59} 

60_TASK_TYPE_BG_FALLBACK = "$primary" 

61 

62# Pill palette: status badge. QUEUED is muted so only the running ones 

63# pop; DONE / FAILED / CANCELLED use brightened backgrounds so terminal 

64# states stand out against the matching left-rail color. 

65_STATUS_BG: dict[TaskStatus, str] = { 

66 TaskStatus.QUEUED: "$surface-lighten-2", 

67 TaskStatus.ACTIVE: "$primary", 

68 TaskStatus.DONE: "$success-lighten-2", 

69 TaskStatus.FAILED: "$error-lighten-2", 

70 TaskStatus.CANCELLED: "$warning-lighten-2", 

71} 

72 

73 

74_TERMINAL_STATUSES: frozenset[TaskStatus] = frozenset( 

75 {TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED} 

76) 

77 

78 

79def _build_head(task: Task, elapsed: str, *, cancelling: bool = False) -> Content: 

80 """Build the top line: name + type pill + status pill, elapsed trailing. 

81 

82 Kept as a module-level helper so tests can exercise the pill 

83 composition directly without spinning up the full widget tree. 

84 """ 

85 type_bg = _TASK_TYPE_BG.get(task.task_type, _TASK_TYPE_BG_FALLBACK) 

86 status_bg = _STATUS_BG[task.status] 

87 status_fg = "$text bold" if task.status in _TERMINAL_STATUSES else "$text" 

88 parts = [ 

89 Content.styled(task.name, "bold"), 

90 Content(" "), 

91 pill(task.task_type, type_bg, "$text"), 

92 Content(" "), 

93 pill(msg.TASK_STATUS_CANCELLING if cancelling else task.status.value, status_bg, status_fg), 

94 ] 

95 if elapsed: 

96 parts.append(Content(" ")) 

97 parts.append(Content.styled(elapsed, "dim")) 

98 return Content.assemble(*parts) 

99 

100 

101def _format_elapsed(task: Task) -> str: 

102 """Return elapsed time as MM:SS, a status tag, or empty. 

103 

104 Terminal states (DONE / FAILED / CANCELLED) freeze at ``completed_at`` 

105 so the timer doesn't keep climbing for rows that are just waiting out 

106 their 2-second flash before removal. 

107 """ 

108 if task.status == TaskStatus.QUEUED: 

109 return "queued" 

110 if task.started_at is None: 

111 return "" 

112 end = task.completed_at if task.completed_at is not None else monotonic() 

113 seconds = max(0, int(end - task.started_at)) 

114 mm, ss = divmod(seconds, 60) 

115 return f"{mm:02d}:{ss:02d}" 

116 

117 

118class TaskRow(Widget, can_focus=True): 

119 """One task, rendered as three stacked lines. 

120 

121 Focusable so ``Tab`` / ``j`` / ``k`` in the Task Center moves between 

122 rows and ``c`` cancels the focused task. 

123 """ 

124 

125 DEFAULT_CSS = "" # all styling lives in task_center.tcss 

126 

127 def __init__(self, task_id: str, **kwargs: object) -> None: 

128 super().__init__(id=f"task-{task_id}", **kwargs) # type: ignore[arg-type] 

129 self._task_id = task_id 

130 

131 def compose(self) -> ComposeResult: 

132 # Widget with yielded children lays them out vertically by default. 

133 # An explicit Vertical wrapper would inherit ``height: 1fr`` and 

134 # stretch each row to fill the scroll viewport, painting the 

135 # border-left rail down the whole empty stretch. 

136 yield Label("", id="row-head", classes="row-head") 

137 yield Label("", id="row-meta", classes="row-meta") 

138 yield Static("", id="row-bar", classes="row-bar") 

139 

140 def update(self, task: Task, tick: int, *, cancelling: bool = False) -> None: 

141 """Re-render from a Task snapshot. Safe to call every poll tick. 

142 

143 Quietly no-ops until the row's child labels have mounted, so the 

144 first few poll ticks (before compose settles) don't error. 

145 """ 

146 # State class: exactly one of the 5 modifier classes is active. 

147 target_class = _STATUS_CLASS.get(task.status, "") 

148 for cls in _STATUS_CLASSES: 

149 self.set_class(cls == target_class, cls) 

150 # 1 Hz rail pulse on the active row only. 

151 self.set_class( 

152 task.status == TaskStatus.ACTIVE and (tick // _PULSE_HALF_TICKS) % 2 == 0, 

153 "-pulse", 

154 ) 

155 

156 try: 

157 head = self.query_one("#row-head", Label) 

158 meta = self.query_one("#row-meta", Label) 

159 bar = self.query_one("#row-bar", Static) 

160 except Exception: 

161 return # compose hasn't finished; retry on next poll 

162 

163 elapsed = _format_elapsed(task) 

164 head.update(_build_head(task, elapsed, cancelling=cancelling)) 

165 

166 # A DONE task's detail is whatever the last progress tick wrote 

167 # ("442/610 MB", "Syncing foo.md..."): stale and confusing now that the 

168 # bar reads 100%. FAILED / CANCELLED keep their detail: it's the reason. 

169 is_done = task.status == TaskStatus.DONE 

170 parts: list[Content] = [] 

171 if not is_done and task.detail: 

172 parts.append(Content(task.detail)) 

173 if not (task.indeterminate or is_done): 

174 parts.append(Content.styled(f"{task.progress:.1f}%", "bold")) 

175 meta.update(Content(" ").join(parts)) 

176 

177 if task.indeterminate: 

178 # Terminal rows freeze the bar so a cancelled/failed/done 

179 # task doesn't keep reading as live work. 

180 if task.status in _TERMINAL_STATUSES: 

181 bar.update(frozen_indeterminate_cell()) 

182 else: 

183 bar.update(indeterminate_cell(tick)) 

184 else: 

185 bar.update(progress_cell(task.progress)) 

186 

187 def flash_completed(self) -> None: 

188 """Mark the row as 'just completed' for a 2-second visual flash.""" 

189 self.add_class("-just-completed")