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

1"""Shared session list panel: filter, select, resume, rename, delete, new. 

2 

3Embedded by both the sessions drawer and the full-screen sessions view. The panel 

4owns everything self-contained (filtering, inline rename, delete confirmation) and 

5resumes or starts a chat itself. It then posts a message, so each container 

6decides how to leave. 

7""" 

8 

9from __future__ import annotations 

10 

11import asyncio 

12from dataclasses import replace 

13from pathlib import Path 

14from typing import TYPE_CHECKING, ClassVar 

15 

16from textual import events, on 

17from textual.app import ComposeResult 

18from textual.binding import Binding, BindingType 

19from textual.containers import Horizontal, Vertical 

20from textual.content import Content 

21from textual.message import Message 

22from textual.widgets import Input, ListItem, ListView, Static 

23 

24from lilbee.app.services import get_services 

25from lilbee.cli.tui import messages as msg 

26from lilbee.cli.tui.browse_bindings import BROWSE_LIST_BINDINGS 

27from lilbee.cli.tui.widgets.confirm_dialog import ConfirmDialog 

28from lilbee.core.text import strip_control_characters 

29from lilbee.sessions import ( 

30 HUMAN_ORIGINS, 

31 SessionMeta, 

32 SessionNotFoundError, 

33 SessionStore, 

34 TitleSource, 

35) 

36 

37if TYPE_CHECKING: 

38 from lilbee.cli.tui.app import LilbeeApp 

39 

40_ROW_CSS = (Path(__file__).parent / "session_list.tcss").read_text(encoding="utf-8") 

41 

42 

43class _RowText(Static): 

44 """Row text that never starts a text selection. 

45 

46 Rows are rebuilt on every store mutation and every filter keystroke, and 

47 Textual's selection path takes ``content_widget.parent`` and dereferences 

48 ``container.region`` with no None check, so a click landing on a row that has 

49 just been unparented crashed the app with AttributeError on 

50 ``_MessagePump__parent``. Selecting text is not something a pick-list row 

51 needs, and switching it off makes that path unreachable here instead of 

52 relying on the removal winning the race against the click. 

53 """ 

54 

55 ALLOW_SELECT = False 

56 

57 

58class SessionRow(ListItem): 

59 """One session: dot + title with a right-aligned age, and a meta line below.""" 

60 

61 def __init__(self, meta: SessionMeta, *, active: bool) -> None: 

62 super().__init__() 

63 self.meta = meta 

64 self._active = active 

65 

66 def compose(self) -> ComposeResult: 

67 dot = "●" if self._active else "○" 

68 title = Content.assemble( 

69 (f"{dot} ", "$success" if self._active else "$text-muted"), 

70 (strip_control_characters(self.meta.title), "bold" if self._active else ""), 

71 ) 

72 meta_line = msg.SESSIONS_ROW_META.format( 

73 count=self.meta.message_count, model=self.meta.model_ref 

74 ) 

75 with Horizontal(classes="session-row-head"): 

76 yield _RowText(title, classes="session-row-title") 

77 yield _RowText(Content(self.meta.updated_at[:10]), classes="session-row-time") 

78 yield _RowText(Content.styled(meta_line, "$text-muted"), classes="session-row-meta") 

79 

80 

81class _FilterInput(Input): 

82 """The filter box. Its editing keys run in its own queue, in order with the text typed. 

83 

84 Textual runs a focused Input's bindings (Backspace, ctrl+w, the arrows) only 

85 once the key has bubbled back to the app, which is after any row key the app 

86 is running. Running them here makes the box's queue hold every key typed into it. 

87 """ 

88 

89 async def _on_key(self, event: events.Key) -> None: 

90 """Run the box's own binding for the key; a key the box declines goes on up as before.""" 

91 for binding in self._bindings.key_to_bindings.get(event.key, ()): 

92 if await self.app.run_action(binding.action, self): 

93 event.stop() 

94 return 

95 

96 

97async def _caught_up(widget: Input) -> None: 

98 """Return once *widget* has handled every message already queued for it. 

99 

100 Also returns when the widget's queue stops, so a handler that failed on one of 

101 those messages lets the app exit instead of leaving this wait unanswered. 

102 """ 

103 done: asyncio.Future[None] = asyncio.get_running_loop().create_future() 

104 if widget.call_later(done.set_result, None): 

105 await asyncio.wait((done, widget.task), return_when=asyncio.FIRST_COMPLETED) 

106 

107 

108class SessionListPanel(Vertical): 

109 """Filterable session list with resume / rename / delete / new actions.""" 

110 

111 app: LilbeeApp # type: ignore[assignment] 

112 

113 DEFAULT_CSS: ClassVar[str] = _ROW_CSS 

114 

115 # Every key that moves the cursor or acts on a row is a priority binding, so 

116 # it runs as the app reads it, before later keys are routed. Each first waits 

117 # for the filter box to handle the keys typed into it ahead of the row key. 

118 # A focused filter still types j / k / g / G: Textual drops a binding whose 

119 # key the focused Input consumes. 

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

121 Binding("enter", "select", "Resume", show=False, priority=True), 

122 Binding("ctrl+n", "new_chat", "New", show=True, priority=True), 

123 Binding("ctrl+r", "rename", "Rename", show=False, priority=True), 

124 Binding("ctrl+d", "delete", "Delete", show=False, priority=True), 

125 Binding("escape", "close", "Close", show=False, priority=True), 

126 Binding("down", "cursor_down", "Down", show=False, priority=True), 

127 Binding("up", "cursor_up", "Up", show=False, priority=True), 

128 *(replace(binding, priority=True) for binding in BROWSE_LIST_BINDINGS), 

129 ] 

130 

131 class ChatOpened(Message): 

132 """A session was resumed or a new chat was started, and chat is showing.""" 

133 

134 class CloseRequested(Message): 

135 """The user asked to close the panel.""" 

136 

137 def __init__(self, *, focus_filter: bool = True) -> None: 

138 super().__init__() 

139 self._renaming_id: str | None = None 

140 # Sessions as of the last store read, and the live filter text. Reading 

141 # the store replays every event of every session, so it happens on mount 

142 # and after a mutation only; keystrokes filter this list in memory. 

143 self._metas: list[SessionMeta] = [] 

144 # The rows as last rendered, in list order; the selection reads these 

145 # rather than the ListView children, which mount a step later. 

146 self._shown: list[SessionMeta] = [] 

147 # The live filter text; mirrors the box except while a rename repurposes it. 

148 self._query = "" 

149 # The drawer focuses the filter for immediate type-to-switch. The 

150 # full-screen tab focuses the list instead, so the nav keys ([ ]) bubble 

151 # to the app instead of being typed into the filter. 

152 self._focus_filter = focus_filter 

153 

154 def compose(self) -> ComposeResult: 

155 yield Static(id="sessions-title") 

156 yield _FilterInput(placeholder=msg.SESSIONS_FILTER_PLACEHOLDER, id="sessions-filter") 

157 yield ListView(id="sessions-list") 

158 yield Static(id="sessions-empty") 

159 yield Static(Content.styled(msg.SESSIONS_HINT, "$text-muted"), id="sessions-hint") 

160 

161 def on_mount(self) -> None: 

162 self.refresh_list() 

163 target = "#sessions-filter" if self._focus_filter else "#sessions-list" 

164 self.screen.set_focus(self.query_one(target)) 

165 

166 def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None: 

167 """While a rename is open, ctrl+d edits the name instead of deleting the session.""" 

168 return not (action == "delete" and self._renaming_id is not None) 

169 

170 def _store(self) -> SessionStore: 

171 return get_services().session_store 

172 

173 def refresh_list(self) -> None: 

174 """Re-read the store, then render. For mount and after a mutation. 

175 

176 The filter text is not a parameter: it lives in _query and survives a 

177 reload, so deleting a row leaves the list filtered as the user left it. 

178 """ 

179 # Agent (MCP) sessions are working state, not conversations; they 

180 # never appear here. 

181 self._metas = self._store().list(origins=HUMAN_ORIGINS) 

182 self._render_rows() 

183 

184 def _render_rows(self) -> None: 

185 """Render rows from the sessions already loaded. Never touches the store. 

186 

187 Not ``_render``: Textual's Widget defines that as its own visual hook. 

188 """ 

189 lv = self.query_one("#sessions-list", ListView) 

190 lv.clear() 

191 needle = self._query.strip().lower() 

192 active_id = self.app.current_session_id() 

193 metas = [m for m in self._metas if needle in m.title.lower()] 

194 self._shown = metas 

195 for meta in metas: 

196 lv.append(SessionRow(meta, active=meta.id == active_id)) 

197 if metas: 

198 lv.index = 0 

199 title = Content.assemble( 

200 (msg.SESSIONS_VIEW, "bold"), 

201 (f" {msg.SESSIONS_COUNT.format(count=len(metas))}", "$text-muted"), 

202 ) 

203 self.query_one("#sessions-title", Static).update(title) 

204 self.query_one("#sessions-empty", Static).update( 

205 Content.styled(msg.SESSIONS_EMPTY, "$text-muted") if not metas else Content("") 

206 ) 

207 

208 def _selected(self) -> SessionMeta | None: 

209 """The session under the cursor, or the first one while the rows are still mounting.""" 

210 index = self.query_one("#sessions-list", ListView).index 

211 if not self._shown: 

212 return None 

213 return self._shown[index if index is not None and index < len(self._shown) else 0] 

214 

215 @on(Input.Changed, "#sessions-filter") 

216 def _on_filter(self, _event: Input.Changed) -> None: 

217 self._apply_filter() 

218 

219 def _apply_filter(self) -> None: 

220 """Render for the filter box's current text, unless the rows already show it.""" 

221 value = self.query_one("#sessions-filter", Input).value 

222 if self._renaming_id is None and value != self._query: 

223 self._query = value 

224 self._render_rows() 

225 

226 async def _filter_caught_up(self) -> None: 

227 """Wait for the filter box to apply the keys typed ahead, then render for them. 

228 

229 The panel's bindings are priority bindings, so they run before the filter 

230 box has handled the keys queued for it. 

231 """ 

232 await _caught_up(self.query_one("#sessions-filter", Input)) 

233 self._apply_filter() 

234 

235 @on(ListView.Selected, "#sessions-list") 

236 def _on_row_selected(self, event: ListView.Selected) -> None: 

237 """Resume a clicked row; enter never gets here, the panel's own binding takes it.""" 

238 if self._renaming_id is None: 

239 self._resume(event.item.meta if isinstance(event.item, SessionRow) else None) 

240 

241 def _resume(self, meta: SessionMeta | None) -> None: 

242 """Resume *meta* within this key's step, so focus is on the prompt before the next key.""" 

243 if meta is None: 

244 return 

245 try: 

246 self.app.resume_session(meta.id) 

247 except SessionNotFoundError: 

248 self.app.notify(msg.SESSIONS_GONE, severity="warning") 

249 self.refresh_list() 

250 return 

251 self.post_message(self.ChatOpened()) 

252 

253 async def action_select(self) -> None: 

254 """Enter: finish a rename, else resume the highlighted session.""" 

255 await self._filter_caught_up() 

256 if self._renaming_id is not None: 

257 self._commit_rename() 

258 return 

259 self._resume(self._selected()) 

260 

261 async def action_cursor_down(self) -> None: 

262 await self._filter_caught_up() 

263 self.query_one("#sessions-list", ListView).action_cursor_down() 

264 

265 async def action_cursor_up(self) -> None: 

266 await self._filter_caught_up() 

267 self.query_one("#sessions-list", ListView).action_cursor_up() 

268 

269 async def action_jump_top(self) -> None: 

270 await self._filter_caught_up() 

271 self.jump_to(0) 

272 

273 async def action_jump_bottom(self) -> None: 

274 await self._filter_caught_up() 

275 self.jump_to(-1) 

276 

277 def jump_to(self, index: int) -> None: 

278 """Move the list cursor to *index* (negative counts from the end).""" 

279 count = len(self._shown) 

280 if not count: 

281 return 

282 lv = self.query_one("#sessions-list", ListView) 

283 lv.index = count - 1 if index < 0 else min(index, count - 1) 

284 

285 def action_new_chat(self) -> None: 

286 self.app.new_chat() 

287 self.post_message(self.ChatOpened()) 

288 

289 def action_close(self) -> None: 

290 if self._renaming_id is not None: 

291 self._cancel_rename() 

292 return 

293 self.post_message(self.CloseRequested()) 

294 

295 async def action_rename(self) -> None: 

296 await self._filter_caught_up() 

297 selected = self._selected() 

298 if selected is None: 

299 return 

300 self._renaming_id = selected.id 

301 field = self.query_one("#sessions-filter", Input) 

302 field.value = selected.title 

303 field.placeholder = msg.SESSIONS_RENAME_PLACEHOLDER 

304 

305 def _commit_rename(self) -> None: 

306 field = self.query_one("#sessions-filter", Input) 

307 title = field.value.strip() 

308 if self._renaming_id is not None and title: 

309 try: 

310 self._store().set_title(self._renaming_id, title, TitleSource.CUSTOM) 

311 except SessionNotFoundError: 

312 self.app.notify(msg.SESSIONS_RENAME_GONE, severity="warning") 

313 self._finish_rename() 

314 

315 def _cancel_rename(self) -> None: 

316 self._finish_rename() 

317 

318 def _finish_rename(self) -> None: 

319 self._renaming_id = None 

320 field = self.query_one("#sessions-filter", Input) 

321 field.value = "" 

322 field.placeholder = msg.SESSIONS_FILTER_PLACEHOLDER 

323 # Resyncs _query: an already-empty box fires no Changed message. 

324 self._query = "" 

325 self.refresh_list() 

326 

327 async def action_delete(self) -> None: 

328 await self._filter_caught_up() 

329 selected = self._selected() 

330 if selected is None: 

331 return 

332 dialog = ConfirmDialog( 

333 msg.SESSIONS_DELETE_CONFIRM_TITLE, 

334 msg.SESSIONS_DELETE_CONFIRM.format(title=strip_control_characters(selected.title)), 

335 ) 

336 self.app.push_screen( 

337 dialog, lambda confirmed: self._on_delete_confirmed(selected, confirmed) 

338 ) 

339 

340 def _on_delete_confirmed(self, meta: SessionMeta, confirmed: bool | None) -> None: 

341 if not confirmed: 

342 return 

343 self._store().delete(meta.id) 

344 self.refresh_list() 

345 self.app.notify(msg.SESSIONS_DELETED.format(title=strip_control_characters(meta.title)))