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 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.CRAWL.value: "$primary", 

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

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

54} 

55_TASK_TYPE_BG_FALLBACK = "$primary" 

56 

57# Pill palette — status badge. QUEUED is muted so only the running ones 

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

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

60_STATUS_BG: dict[TaskStatus, str] = { 

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

62 TaskStatus.ACTIVE: "$primary", 

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

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

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

66} 

67 

68 

69_TERMINAL_STATUSES: frozenset[TaskStatus] = frozenset( 

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

71) 

72 

73 

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

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

76 

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

78 composition directly without spinning up the full widget tree. 

79 """ 

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

81 status_bg = _STATUS_BG[task.status] 

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

83 parts = [ 

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

85 Content(" "), 

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

87 Content(" "), 

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

89 ] 

90 if elapsed: 

91 parts.append(Content(" ")) 

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

93 return Content.assemble(*parts) 

94 

95 

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

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

98 

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

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

101 their 2-second flash before removal. 

102 """ 

103 if task.status == TaskStatus.QUEUED: 

104 return "queued" 

105 if task.started_at is None: 

106 return "" 

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

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

109 mm, ss = divmod(seconds, 60) 

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

111 

112 

113class TaskRow(Widget, can_focus=True): 

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

115 

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

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

118 """ 

119 

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

121 

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

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

124 self._task_id = task_id 

125 

126 def compose(self) -> ComposeResult: 

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

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

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

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

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

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

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

134 

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

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

137 

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

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

140 """ 

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

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

143 for cls in _STATUS_CLASSES: 

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

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

146 self.set_class( 

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

148 "-pulse", 

149 ) 

150 

151 try: 

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

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

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

155 except Exception: 

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

157 

158 elapsed = _format_elapsed(task) 

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

160 

161 detail = task.detail or "" 

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

163 meta_parts = [detail, pct] 

164 meta.update(" ".join(p for p in meta_parts if p)) 

165 

166 if task.indeterminate: 

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

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

169 if task.status in _TERMINAL_STATUSES: 

170 bar.update(frozen_indeterminate_cell()) 

171 else: 

172 bar.update(indeterminate_cell(tick)) 

173 else: 

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

175 

176 def flash_completed(self) -> None: 

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

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