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.sessions import ( 

29 HUMAN_ORIGINS, 

30 SessionMeta, 

31 SessionNotFoundError, 

32 SessionStore, 

33 TitleSource, 

34) 

35 

36if TYPE_CHECKING: 

37 from lilbee.cli.tui.app import LilbeeApp 

38 

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

40 

41 

42class _RowText(Static): 

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

44 

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

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

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

48 just been unparented crashed the app with AttributeError on 

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

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

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

52 """ 

53 

54 ALLOW_SELECT = False 

55 

56 

57class SessionRow(ListItem): 

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

59 

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

61 super().__init__() 

62 self.meta = meta 

63 self._active = active 

64 

65 def compose(self) -> ComposeResult: 

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

67 title = Content.assemble( 

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

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

70 ) 

71 meta_line = msg.SESSIONS_ROW_META.format( 

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

73 ) 

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

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

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

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

78 

79 

80class _FilterInput(Input): 

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

82 

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

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

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

86 """ 

87 

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

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

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

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

92 event.stop() 

93 return 

94 

95 

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

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

98 

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

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

101 """ 

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

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

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

105 

106 

107class SessionListPanel(Vertical): 

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

109 

110 app: LilbeeApp # type: ignore[assignment] 

111 

112 DEFAULT_CSS: ClassVar[str] = _ROW_CSS 

113 

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

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

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

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

118 # key the focused Input consumes. 

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

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

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

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

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

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

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

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

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

128 ] 

129 

130 class ChatOpened(Message): 

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

132 

133 class CloseRequested(Message): 

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

135 

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

137 super().__init__() 

138 self._renaming_id: str | None = None 

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

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

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

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

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

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

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

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

147 self._query = "" 

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

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

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

151 self._focus_filter = focus_filter 

152 

153 def compose(self) -> ComposeResult: 

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

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

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

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

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

159 

160 def on_mount(self) -> None: 

161 self.refresh_list() 

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

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

164 

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

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

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

168 

169 def _store(self) -> SessionStore: 

170 return get_services().session_store 

171 

172 def refresh_list(self) -> None: 

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

174 

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

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

177 """ 

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

179 # never appear here. 

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

181 self._render_rows() 

182 

183 def _render_rows(self) -> None: 

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

185 

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

187 """ 

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

189 lv.clear() 

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

191 active_id = self.app.current_session_id() 

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

193 self._shown = metas 

194 for meta in metas: 

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

196 if metas: 

197 lv.index = 0 

198 title = Content.assemble( 

199 (msg.SESSIONS_VIEW, "bold"), 

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

201 ) 

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

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

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

205 ) 

206 

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

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

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

210 if not self._shown: 

211 return None 

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

213 

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

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

216 self._apply_filter() 

217 

218 def _apply_filter(self) -> None: 

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

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

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

222 self._query = value 

223 self._render_rows() 

224 

225 async def _filter_caught_up(self) -> None: 

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

227 

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

229 box has handled the keys queued for it. 

230 """ 

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

232 self._apply_filter() 

233 

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

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

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

237 if self._renaming_id is None: 

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

239 

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

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

242 if meta is None: 

243 return 

244 try: 

245 self.app.resume_session(meta.id) 

246 except SessionNotFoundError: 

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

248 self.refresh_list() 

249 return 

250 self.post_message(self.ChatOpened()) 

251 

252 async def action_select(self) -> None: 

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

254 await self._filter_caught_up() 

255 if self._renaming_id is not None: 

256 self._commit_rename() 

257 return 

258 self._resume(self._selected()) 

259 

260 async def action_cursor_down(self) -> None: 

261 await self._filter_caught_up() 

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

263 

264 async def action_cursor_up(self) -> None: 

265 await self._filter_caught_up() 

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

267 

268 async def action_jump_top(self) -> None: 

269 await self._filter_caught_up() 

270 self.jump_to(0) 

271 

272 async def action_jump_bottom(self) -> None: 

273 await self._filter_caught_up() 

274 self.jump_to(-1) 

275 

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

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

278 count = len(self._shown) 

279 if not count: 

280 return 

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

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

283 

284 def action_new_chat(self) -> None: 

285 self.app.new_chat() 

286 self.post_message(self.ChatOpened()) 

287 

288 def action_close(self) -> None: 

289 if self._renaming_id is not None: 

290 self._cancel_rename() 

291 return 

292 self.post_message(self.CloseRequested()) 

293 

294 async def action_rename(self) -> None: 

295 await self._filter_caught_up() 

296 selected = self._selected() 

297 if selected is None: 

298 return 

299 self._renaming_id = selected.id 

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

301 field.value = selected.title 

302 field.placeholder = msg.SESSIONS_RENAME_PLACEHOLDER 

303 

304 def _commit_rename(self) -> None: 

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

306 title = field.value.strip() 

307 if self._renaming_id is not None and title: 

308 try: 

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

310 except SessionNotFoundError: 

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

312 self._finish_rename() 

313 

314 def _cancel_rename(self) -> None: 

315 self._finish_rename() 

316 

317 def _finish_rename(self) -> None: 

318 self._renaming_id = None 

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

320 field.value = "" 

321 field.placeholder = msg.SESSIONS_FILTER_PLACEHOLDER 

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

323 self._query = "" 

324 self.refresh_list() 

325 

326 async def action_delete(self) -> None: 

327 await self._filter_caught_up() 

328 selected = self._selected() 

329 if selected is None: 

330 return 

331 dialog = ConfirmDialog( 

332 msg.SESSIONS_DELETE_CONFIRM_TITLE, 

333 msg.SESSIONS_DELETE_CONFIRM.format(title=selected.title), 

334 ) 

335 self.app.push_screen( 

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

337 ) 

338 

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

340 if not confirmed: 

341 return 

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

343 self.refresh_list() 

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