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.pill import pill 

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

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

23 frozen_indeterminate_cell, 

24 indeterminate_cell, 

25 progress_cell, 

26) 

27 

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

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

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

31_PULSE_HALF_TICKS = 3 

32 

33_STATUS_CLASS: dict[TaskStatus, str] = { 

34 TaskStatus.QUEUED: "-queued", 

35 TaskStatus.ACTIVE: "-active", 

36 TaskStatus.DONE: "-done", 

37 TaskStatus.FAILED: "-failed", 

38 TaskStatus.CANCELLED: "-cancelled", 

39} 

40 

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

42 

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

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

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

46_TASK_TYPE_BG: dict[str, str] = { 

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

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

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

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

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

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

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

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

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

56} 

57_TASK_TYPE_BG_FALLBACK = "$primary" 

58 

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

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

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

62_STATUS_BG: dict[TaskStatus, str] = { 

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

64 TaskStatus.ACTIVE: "$primary", 

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

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

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

68} 

69 

70 

71_TERMINAL_STATUSES: frozenset[TaskStatus] = frozenset( 

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

73) 

74 

75 

76def _build_head(task: Task, elapsed: str) -> Content: 

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

78 

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

80 composition directly without spinning up the full widget tree. 

81 """ 

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

83 status_bg = _STATUS_BG[task.status] 

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

85 parts = [ 

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

87 Content(" "), 

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

89 Content(" "), 

90 pill(task.status.value, status_bg, status_fg), 

91 ] 

92 if elapsed: 

93 parts.append(Content(" ")) 

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

95 return Content.assemble(*parts) 

96 

97 

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

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

100 

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

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

103 their 2-second flash before removal. 

104 """ 

105 if task.status == TaskStatus.QUEUED: 

106 return "queued" 

107 if task.started_at is None: 

108 return "" 

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

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

111 mm, ss = divmod(seconds, 60) 

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

113 

114 

115class TaskRow(Widget, can_focus=True): 

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

117 

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

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

120 """ 

121 

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

123 

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

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

126 self._task_id = task_id 

127 

128 def compose(self) -> ComposeResult: 

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

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

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

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

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

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

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

136 

137 def update(self, task: Task, tick: int) -> None: 

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

139 

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

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

142 """ 

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

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

145 for cls in _STATUS_CLASSES: 

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

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

148 self.set_class( 

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

150 "-pulse", 

151 ) 

152 

153 try: 

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

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

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

157 except Exception: 

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

159 

160 elapsed = _format_elapsed(task) 

161 head.update(_build_head(task, elapsed)) 

162 

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

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

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

166 is_done = task.status == TaskStatus.DONE 

167 detail = "" if is_done else (task.detail or "") 

168 pct = "" if task.indeterminate or is_done else f"[b]{task.progress:.1f}%[/b]" 

169 meta.update(" ".join(p for p in (detail, pct) if p)) 

170 

171 if task.indeterminate: 

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

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

174 if task.status in _TERMINAL_STATUSES: 

175 bar.update(frozen_indeterminate_cell()) 

176 else: 

177 bar.update(indeterminate_cell(tick)) 

178 else: 

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

180 

181 def flash_completed(self) -> None: 

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

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