Coverage for src/lilbee/cli/tui/screens/wiki_drafts.py: 100%

1"""Wiki drafts review screen: browse, diff, accept, or reject pending drafts. 

2 

3The screen pairs a left-hand :class:`DataTable` of drafts with a 

4right-hand scrollable :class:`Static` that renders the unified diff of 

5the highlighted draft against its published counterpart. Accept and 

6reject are confirmed through the shared :class:`ConfirmDialog` modal. 

7Keybindings follow the rest of the TUI: vim j/k to navigate, ``/`` to 

8search, ``a`` / ``r`` for accept / reject, ``q`` / Esc to back out. 

9""" 

10 

11from __future__ import annotations 

12 

13import logging 

14from pathlib import Path 

15from typing import TYPE_CHECKING, ClassVar 

16 

17from rich.text import Text 

18from textual import on 

19from textual.app import ComposeResult 

20from textual.binding import Binding, BindingType 

21from textual.containers import Horizontal, Vertical, VerticalScroll 

22from textual.screen import Screen 

23from textual.widgets import DataTable, Input, Static 

24 

25from lilbee.app.services import get_services 

26from lilbee.cli.tui import messages as msg 

27from lilbee.cli.tui.browse_bindings import BROWSE_LIST_BINDINGS, browse_back_bindings 

28from lilbee.cli.tui.task_queue import TaskType 

29from lilbee.cli.tui.thread_safe import call_from_thread 

30from lilbee.cli.tui.widgets.task_bar import TaskBar 

31from lilbee.core.config import cfg 

32from lilbee.core.security import PathTraversalError 

33from lilbee.runtime.cancellation import TaskCancelledError 

34from lilbee.wiki.drafts import DraftAcceptError, accept_draft, diff_draft, list_drafts, reject_draft 

35from lilbee.wiki.shared import INVALID_DRAFT_SLUG_ERROR 

36 

37if TYPE_CHECKING: 

38 from collections.abc import Callable 

39 

40 from textual.notifications import SeverityLevel 

41 

42 from lilbee.cli.tui.app import LilbeeApp 

43 from lilbee.cli.tui.widgets.task_bar_controller import ProgressReporter 

44 from lilbee.wiki.drafts import DraftInfo 

45 

46log = logging.getLogger(__name__) 

47 

48 

49def _wiki_root() -> Path: 

50 """Resolve the wiki root directory from config.""" 

51 return cfg.data_root / cfg.wiki_dir 

52 

53 

54def _format_drift(drift: float | None) -> str: 

55 """Render a drift ratio as a percentage, or ``-`` when absent.""" 

56 return f"{drift:.0%}" if drift is not None else "-" 

57 

58 

59def _format_faithfulness(score: float | None) -> str: 

60 """Render a faithfulness score with two decimals, or ``-`` when absent.""" 

61 return f"{score:.2f}" if score is not None else "-" 

62 

63 

64def _format_published(exists: bool) -> str: 

65 """Render the published-counterpart flag as a human yes/no.""" 

66 return msg.WIKI_DRAFTS_PUBLISHED_YES if exists else msg.WIKI_DRAFTS_PUBLISHED_NO 

67 

68 

69def _draft_failure(exc: Exception, slug: str) -> tuple[str, SeverityLevel]: 

70 """Map a failed draft mutation to its user-facing text and toast severity.""" 

71 if isinstance(exc, DraftAcceptError): 

72 return str(exc), "warning" 

73 if isinstance(exc, FileNotFoundError): 

74 return msg.WIKI_DRAFTS_MISSING.format(slug=slug), "error" 

75 if isinstance(exc, PathTraversalError): 

76 # Generic text: the exception carries the absolute candidate path. 

77 return INVALID_DRAFT_SLUG_ERROR, "error" 

78 return str(exc), "error" 

79 

80 

81def _post_success(app: LilbeeApp, message: str) -> None: 

82 """Toast a completed draft mutation from the worker thread. 

83 

84 No reload here: the WIKI task's done hook rescans the wiki screens, and 

85 each rescan re-walks every page's frontmatter from disk. 

86 """ 

87 call_from_thread(app, app.notify, message, severity="information") 

88 

89 

90def _post_failure(app: LilbeeApp, message: str, severity: SeverityLevel) -> None: 

91 """Marshal a failed draft mutation back to the event loop from the worker thread. 

92 

93 Targets the app, not the screen: a WIKI task queues behind a running 

94 build, so the screen that started it may be gone by the time it lands. 

95 """ 

96 call_from_thread(app, _apply_failure, app, message, severity) 

97 

98 

99def _apply_failure(app: LilbeeApp, message: str, severity: SeverityLevel) -> None: 

100 """Toast the failure and re-read the drafts a partial mutation may have changed. 

101 

102 No done hook fires for a failed task, so this path reloads itself. 

103 """ 

104 app.notify(message, severity=severity) 

105 app.task_bar.reload_wiki_screens() 

106 

107 

108def _kind_label(pending_kind: str | None) -> str: 

109 """Map a pending_kind value to its display label. 

110 

111 ``None`` surfaces as "drift" because drift is the default review 

112 reason when no PENDING marker is present. 

113 """ 

114 return pending_kind or msg.WIKI_DRAFTS_KIND_DRIFT 

115 

116 

117class WikiDraftsScreen(Screen[None]): 

118 """Review-surface screen for pending wiki drafts.""" 

119 

120 app: LilbeeApp # type: ignore[assignment] 

121 

122 CSS_PATH = "wiki_drafts.tcss" 

123 AUTO_FOCUS = "#wiki-drafts-table" 

124 HELP = "Review pending wiki drafts. j/k navigate, a accept, r reject, / search, q back." 

125 

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

127 *browse_back_bindings(escape_action="dismiss_or_back"), 

128 Binding("a", "accept", "Accept", show=True), 

129 Binding("r", "reject", "Reject", show=True), 

130 Binding("slash", "focus_search", "Search", show=True), 

131 *BROWSE_LIST_BINDINGS, 

132 ] 

133 

134 def __init__(self) -> None: 

135 super().__init__() 

136 self._drafts: list[DraftInfo] = [] 

137 self._filter: str = "" 

138 

139 def compose(self) -> ComposeResult: 

140 from textual.widgets import Footer 

141 

142 from lilbee.cli.tui.widgets.bottom_bars import BottomBars 

143 from lilbee.cli.tui.widgets.status_bar import ViewTabs 

144 from lilbee.cli.tui.widgets.top_bars import TopBars 

145 

146 with TopBars(): 

147 yield ViewTabs() 

148 table: DataTable[str | Text] = DataTable(id="wiki-drafts-table") 

149 table.cursor_type = "row" 

150 yield Horizontal( 

151 Vertical( 

152 Input( 

153 placeholder=msg.WIKI_DRAFTS_SEARCH_PLACEHOLDER, 

154 id="wiki-drafts-search", 

155 ), 

156 table, 

157 id="wiki-drafts-sidebar", 

158 ), 

159 Vertical( 

160 VerticalScroll( 

161 Static(msg.WIKI_DRAFTS_DIFF_EMPTY, id="wiki-drafts-diff", markup=False), 

162 id="wiki-drafts-diff-scroll", 

163 ), 

164 id="wiki-drafts-main", 

165 ), 

166 id="wiki-drafts-layout", 

167 ) 

168 with BottomBars(): 

169 yield TaskBar() 

170 yield Footer() 

171 

172 def on_mount(self) -> None: 

173 table = self.query_one("#wiki-drafts-table", DataTable) 

174 table.add_columns( 

175 msg.WIKI_DRAFTS_COLUMN_SLUG, 

176 msg.WIKI_DRAFTS_COLUMN_KIND, 

177 msg.WIKI_DRAFTS_COLUMN_DRIFT, 

178 msg.WIKI_DRAFTS_COLUMN_FAITHFULNESS, 

179 msg.WIKI_DRAFTS_COLUMN_PUBLISHED, 

180 ) 

181 self.reload() 

182 

183 def reload(self) -> None: 

184 """Re-read drafts from disk and repopulate the table. 

185 

186 Public entry point for the task bar, which refreshes an open drafts 

187 screen after work that wrote or removed drafts. 

188 """ 

189 try: 

190 self._drafts = list_drafts(_wiki_root()) 

191 except Exception as exc: 

192 log.warning("Failed to list wiki drafts", exc_info=True) 

193 self._drafts = [] 

194 self.query_one("#wiki-drafts-table", DataTable).clear() 

195 self._show_diff(msg.WIKI_DRAFTS_LOAD_FAILED.format(error=exc)) 

196 return 

197 self._populate_table() 

198 

199 def _populate_table(self) -> None: 

200 """Render the filtered view of the already-loaded drafts.""" 

201 table = self.query_one("#wiki-drafts-table", DataTable) 

202 table.clear() 

203 visible = self._visible_drafts() 

204 if not visible: 

205 if self._drafts: 

206 self._show_diff(msg.WIKI_DRAFTS_NO_MATCHES.format(filter=self._filter)) 

207 else: 

208 self._show_diff(msg.WIKI_DRAFTS_EMPTY) 

209 return 

210 

211 for d in visible: 

212 table.add_row( 

213 Text(d.slug), 

214 _kind_label(d.pending_kind), 

215 _format_drift(d.drift_ratio), 

216 _format_faithfulness(d.faithfulness_score), 

217 _format_published(d.published_exists), 

218 key=d.slug, 

219 ) 

220 self._show_diff(msg.WIKI_DRAFTS_DIFF_EMPTY) 

221 

222 def _visible_drafts(self) -> list[DraftInfo]: 

223 """Apply the current filter to the loaded draft list.""" 

224 if not self._filter: 

225 return self._drafts 

226 needle = self._filter.lower() 

227 return [d for d in self._drafts if needle in d.slug.lower()] 

228 

229 def _show_diff(self, text: str) -> None: 

230 """Update the diff pane with *text*.""" 

231 self.query_one("#wiki-drafts-diff", Static).update(text) 

232 

233 def _highlighted_slug(self) -> str | None: 

234 """Return the slug of the highlighted row, or ``None`` when empty.""" 

235 table = self.query_one("#wiki-drafts-table", DataTable) 

236 if table.row_count == 0: 

237 return None 

238 try: 

239 row_key, _ = table.coordinate_to_cell_key(table.cursor_coordinate) 

240 except Exception: 

241 return None 

242 if row_key is None or row_key.value is None: 

243 return None 

244 return str(row_key.value) 

245 

246 @on(DataTable.RowHighlighted, "#wiki-drafts-table") 

247 def _on_row_highlighted(self, event: DataTable.RowHighlighted) -> None: 

248 """Load the diff for the newly highlighted row.""" 

249 key = event.row_key.value if event.row_key is not None else None 

250 if key is None: 

251 return 

252 self._display_diff(str(key)) 

253 

254 def _display_diff(self, slug: str) -> None: 

255 """Compute and render the unified diff for *slug*.""" 

256 try: 

257 diff = diff_draft(slug, _wiki_root()) 

258 except FileNotFoundError: 

259 self._show_diff(msg.WIKI_DRAFTS_DIFF_EMPTY) 

260 return 

261 except PathTraversalError: 

262 # Traversal slug: show the generic, path-free message its sibling 

263 # transports use rather than leaking the absolute candidate path. 

264 self._show_diff(INVALID_DRAFT_SLUG_ERROR) 

265 return 

266 except Exception as exc: 

267 log.debug("Failed to compute diff for %s", slug, exc_info=True) 

268 self._show_diff(msg.WIKI_DRAFTS_DIFF_FAILED.format(error=exc)) 

269 return 

270 self._show_diff(diff or msg.WIKI_DRAFTS_DIFF_NONE) 

271 

272 @on(Input.Changed, "#wiki-drafts-search") 

273 def _on_search_changed(self, event: Input.Changed) -> None: 

274 """Filter the drafts already in memory; typing never re-reads disk.""" 

275 self._filter = event.value.strip() 

276 self._populate_table() 

277 

278 def action_focus_search(self) -> None: 

279 """Focus the search input (``/`` keybinding).""" 

280 self.query_one("#wiki-drafts-search", Input).focus() 

281 

282 def action_dismiss_or_back(self) -> None: 

283 """Clear the search if active, otherwise back out to the wiki screen.""" 

284 search = self.query_one("#wiki-drafts-search", Input) 

285 if search.value: 

286 search.value = "" 

287 return 

288 self.action_go_back() 

289 

290 def action_go_back(self) -> None: 

291 """Pop back to the wiki screen, unless this is the only screen on the stack.""" 

292 if len(self.app.screen_stack) > 1: 

293 self.app.pop_screen() 

294 

295 def _table_or_none(self) -> DataTable[str | Text] | None: 

296 """Return the drafts table unless an Input is focused.""" 

297 if isinstance(self.focused, Input): 

298 return None 

299 return self.query_one("#wiki-drafts-table", DataTable) 

300 

301 def action_cursor_down(self) -> None: 

302 table = self._table_or_none() 

303 if table is not None: 

304 table.action_cursor_down() 

305 

306 def action_cursor_up(self) -> None: 

307 table = self._table_or_none() 

308 if table is not None: 

309 table.action_cursor_up() 

310 

311 def action_jump_top(self) -> None: 

312 table = self._table_or_none() 

313 if table is not None: 

314 table.action_scroll_top() 

315 

316 def action_jump_bottom(self) -> None: 

317 table = self._table_or_none() 

318 if table is not None: 

319 table.action_scroll_bottom() 

320 

321 def action_accept(self) -> None: 

322 """Prompt for confirmation, then accept the highlighted draft.""" 

323 slug = self._highlighted_slug() 

324 if slug is None: 

325 return 

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

327 

328 def _on_confirm(confirmed: bool | None) -> None: 

329 if not confirmed: 

330 return 

331 self._do_accept(slug) 

332 

333 self.app.push_screen( 

334 ConfirmDialog( 

335 msg.WIKI_DRAFTS_ACCEPT_CONFIRM_TITLE, 

336 msg.WIKI_DRAFTS_ACCEPT_CONFIRM_MESSAGE.format(slug=slug), 

337 ), 

338 _on_confirm, 

339 ) 

340 

341 def _do_accept(self, slug: str) -> None: 

342 """Accept the draft on the task bar and refresh the list when it lands.""" 

343 self._start_draft_task( 

344 slug, 

345 msg.WIKI_DRAFTS_ACCEPT_TASK.format(slug=slug), 

346 lambda: accept_draft(slug, _wiki_root(), get_services().store), 

347 msg.WIKI_DRAFTS_ACCEPTED.format(slug=slug), 

348 msg.WIKI_DRAFTS_ACCEPT_FAILED, 

349 ) 

350 

351 def action_reject(self) -> None: 

352 """Prompt for confirmation, then reject the highlighted draft.""" 

353 slug = self._highlighted_slug() 

354 if slug is None: 

355 return 

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

357 

358 def _on_confirm(confirmed: bool | None) -> None: 

359 if not confirmed: 

360 return 

361 self._do_reject(slug) 

362 

363 self.app.push_screen( 

364 ConfirmDialog( 

365 msg.WIKI_DRAFTS_REJECT_CONFIRM_TITLE, 

366 msg.WIKI_DRAFTS_REJECT_CONFIRM_MESSAGE.format(slug=slug), 

367 ), 

368 _on_confirm, 

369 ) 

370 

371 def _do_reject(self, slug: str) -> None: 

372 """Reject the draft on the task bar and refresh the list when it lands.""" 

373 self._start_draft_task( 

374 slug, 

375 msg.WIKI_DRAFTS_REJECT_TASK.format(slug=slug), 

376 lambda: reject_draft(slug, _wiki_root()), 

377 msg.WIKI_DRAFTS_REJECTED.format(slug=slug), 

378 msg.WIKI_DRAFTS_REJECT_FAILED, 

379 ) 

380 

381 def _start_draft_task( 

382 self, 

383 slug: str, 

384 name: str, 

385 work: Callable[[], object], 

386 success_message: str, 

387 failure_template: str, 

388 ) -> None: 

389 """Run a draft mutation as a WIKI task, then toast and refresh. 

390 

391 accept_draft takes the wiki build mutex, so running it inline would 

392 freeze the UI for the length of whatever build holds the mutex. 

393 Failures re-raise the mapped text so the task row records what the 

394 toast said, and the outcome is checked against a cancel that landed 

395 while the work was running. 

396 """ 

397 app = self.app 

398 

399 def _target(reporter: ProgressReporter) -> None: 

400 reporter.update(0, slug, indeterminate=True) 

401 try: 

402 work() 

403 reporter.check_cancelled() 

404 except TaskCancelledError: 

405 # The mutation may have landed before the cancel; the table 

406 # must show the disk state either way. 

407 call_from_thread(app, app.task_bar.reload_wiki_screens) 

408 raise 

409 except Exception as exc: 

410 error, severity = _draft_failure(exc, slug) 

411 _post_failure(app, failure_template.format(error=error), severity) 

412 raise RuntimeError(error) from exc 

413 _post_success(app, success_message) 

414 

415 app.task_bar.start_task(name, TaskType.WIKI, _target, indeterminate=True)