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 textual import on 

18from textual.app import ComposeResult 

19from textual.binding import Binding, BindingType 

20from textual.containers import Horizontal, Vertical, VerticalScroll 

21from textual.screen import Screen 

22from textual.widgets import DataTable, Input, 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, browse_back_bindings 

27from lilbee.cli.tui.task_queue import TaskType 

28from lilbee.cli.tui.thread_safe import call_from_thread 

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

30from lilbee.core.config import cfg 

31from lilbee.core.security import PathTraversalError 

32from lilbee.runtime.cancellation import TaskCancelledError 

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

34from lilbee.wiki.shared import INVALID_DRAFT_SLUG_ERROR 

35 

36if TYPE_CHECKING: 

37 from collections.abc import Callable 

38 

39 from textual.notifications import SeverityLevel 

40 

41 from lilbee.cli.tui.app import LilbeeApp 

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

43 from lilbee.wiki.drafts import DraftInfo 

44 

45log = logging.getLogger(__name__) 

46 

47 

48def _wiki_root() -> Path: 

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

50 return cfg.data_root / cfg.wiki_dir 

51 

52 

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

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

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

56 

57 

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

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

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

61 

62 

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

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

65 return msg.WIKI_DRAFTS_PUBLISHED_YES if exists else msg.WIKI_DRAFTS_PUBLISHED_NO 

66 

67 

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

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

70 if isinstance(exc, DraftAcceptError): 

71 return str(exc), "warning" 

72 if isinstance(exc, FileNotFoundError): 

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

74 if isinstance(exc, PathTraversalError): 

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

76 return INVALID_DRAFT_SLUG_ERROR, "error" 

77 return str(exc), "error" 

78 

79 

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

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

82 

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

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

85 """ 

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

87 

88 

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

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

91 

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

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

94 """ 

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

96 

97 

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

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

100 

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

102 """ 

103 app.notify(message, severity=severity) 

104 app.task_bar.reload_wiki_screens() 

105 

106 

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

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

109 

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

111 reason when no PENDING marker is present. 

112 """ 

113 return pending_kind or msg.WIKI_DRAFTS_KIND_DRIFT 

114 

115 

116class WikiDraftsScreen(Screen[None]): 

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

118 

119 app: LilbeeApp # type: ignore[assignment] 

120 

121 CSS_PATH = "wiki_drafts.tcss" 

122 AUTO_FOCUS = "#wiki-drafts-table" 

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

124 

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

126 *browse_back_bindings(escape_action="dismiss_or_back"), 

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

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

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

130 *BROWSE_LIST_BINDINGS, 

131 ] 

132 

133 def __init__(self) -> None: 

134 super().__init__() 

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

136 self._filter: str = "" 

137 

138 def compose(self) -> ComposeResult: 

139 from textual.widgets import Footer 

140 

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

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

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

144 

145 with TopBars(): 

146 yield ViewTabs() 

147 table: DataTable[str] = DataTable(id="wiki-drafts-table") 

148 table.cursor_type = "row" 

149 yield Horizontal( 

150 Vertical( 

151 Input( 

152 placeholder=msg.WIKI_DRAFTS_SEARCH_PLACEHOLDER, 

153 id="wiki-drafts-search", 

154 ), 

155 table, 

156 id="wiki-drafts-sidebar", 

157 ), 

158 Vertical( 

159 VerticalScroll( 

160 Static(msg.WIKI_DRAFTS_DIFF_EMPTY, id="wiki-drafts-diff"), 

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

162 ), 

163 id="wiki-drafts-main", 

164 ), 

165 id="wiki-drafts-layout", 

166 ) 

167 with BottomBars(): 

168 yield TaskBar() 

169 yield Footer() 

170 

171 def on_mount(self) -> None: 

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

173 table.add_columns( 

174 msg.WIKI_DRAFTS_COLUMN_SLUG, 

175 msg.WIKI_DRAFTS_COLUMN_KIND, 

176 msg.WIKI_DRAFTS_COLUMN_DRIFT, 

177 msg.WIKI_DRAFTS_COLUMN_FAITHFULNESS, 

178 msg.WIKI_DRAFTS_COLUMN_PUBLISHED, 

179 ) 

180 self.reload() 

181 

182 def reload(self) -> None: 

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

184 

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

186 screen after work that wrote or removed drafts. 

187 """ 

188 try: 

189 self._drafts = list_drafts(_wiki_root()) 

190 except Exception as exc: 

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

192 self._drafts = [] 

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

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

195 return 

196 self._populate_table() 

197 

198 def _populate_table(self) -> None: 

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

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

201 table.clear() 

202 visible = self._visible_drafts() 

203 if not visible: 

204 if self._drafts: 

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

206 else: 

207 self._show_diff(msg.WIKI_DRAFTS_EMPTY) 

208 return 

209 

210 for d in visible: 

211 table.add_row( 

212 d.slug, 

213 _kind_label(d.pending_kind), 

214 _format_drift(d.drift_ratio), 

215 _format_faithfulness(d.faithfulness_score), 

216 _format_published(d.published_exists), 

217 key=d.slug, 

218 ) 

219 self._show_diff(msg.WIKI_DRAFTS_DIFF_EMPTY) 

220 

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

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

223 if not self._filter: 

224 return self._drafts 

225 needle = self._filter.lower() 

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

227 

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

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

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

231 

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

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

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

235 if table.row_count == 0: 

236 return None 

237 try: 

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

239 except Exception: 

240 return None 

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

242 return None 

243 return str(row_key.value) 

244 

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

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

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

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

249 if key is None: 

250 return 

251 self._display_diff(str(key)) 

252 

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

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

255 try: 

256 diff = diff_draft(slug, _wiki_root()) 

257 except FileNotFoundError: 

258 self._show_diff(msg.WIKI_DRAFTS_DIFF_EMPTY) 

259 return 

260 except PathTraversalError: 

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

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

263 self._show_diff(INVALID_DRAFT_SLUG_ERROR) 

264 return 

265 except Exception as exc: 

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

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

268 return 

269 self._show_diff(diff or msg.WIKI_DRAFTS_DIFF_NONE) 

270 

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

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

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

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

275 self._populate_table() 

276 

277 def action_focus_search(self) -> None: 

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

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

280 

281 def action_dismiss_or_back(self) -> None: 

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

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

284 if search.value: 

285 search.value = "" 

286 return 

287 self.action_go_back() 

288 

289 def action_go_back(self) -> None: 

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

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

292 self.app.pop_screen() 

293 

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

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

296 if isinstance(self.focused, Input): 

297 return None 

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

299 

300 def action_cursor_down(self) -> None: 

301 table = self._table_or_none() 

302 if table is not None: 

303 table.action_cursor_down() 

304 

305 def action_cursor_up(self) -> None: 

306 table = self._table_or_none() 

307 if table is not None: 

308 table.action_cursor_up() 

309 

310 def action_jump_top(self) -> None: 

311 table = self._table_or_none() 

312 if table is not None: 

313 table.action_scroll_top() 

314 

315 def action_jump_bottom(self) -> None: 

316 table = self._table_or_none() 

317 if table is not None: 

318 table.action_scroll_bottom() 

319 

320 def action_accept(self) -> None: 

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

322 slug = self._highlighted_slug() 

323 if slug is None: 

324 return 

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

326 

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

328 if not confirmed: 

329 return 

330 self._do_accept(slug) 

331 

332 self.app.push_screen( 

333 ConfirmDialog( 

334 msg.WIKI_DRAFTS_ACCEPT_CONFIRM_TITLE, 

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

336 ), 

337 _on_confirm, 

338 ) 

339 

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

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

342 self._start_draft_task( 

343 slug, 

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

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

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

347 msg.WIKI_DRAFTS_ACCEPT_FAILED, 

348 ) 

349 

350 def action_reject(self) -> None: 

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

352 slug = self._highlighted_slug() 

353 if slug is None: 

354 return 

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

356 

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

358 if not confirmed: 

359 return 

360 self._do_reject(slug) 

361 

362 self.app.push_screen( 

363 ConfirmDialog( 

364 msg.WIKI_DRAFTS_REJECT_CONFIRM_TITLE, 

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

366 ), 

367 _on_confirm, 

368 ) 

369 

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

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

372 self._start_draft_task( 

373 slug, 

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

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

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

377 msg.WIKI_DRAFTS_REJECT_FAILED, 

378 ) 

379 

380 def _start_draft_task( 

381 self, 

382 slug: str, 

383 name: str, 

384 work: Callable[[], object], 

385 success_message: str, 

386 failure_template: str, 

387 ) -> None: 

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

389 

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

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

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

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

394 while the work was running. 

395 """ 

396 app = self.app 

397 

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

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

400 try: 

401 work() 

402 reporter.check_cancelled() 

403 except TaskCancelledError: 

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

405 # must show the disk state either way. 

406 call_from_thread(app, app.task_bar.reload_wiki_screens) 

407 raise 

408 except Exception as exc: 

409 error, severity = _draft_failure(exc, slug) 

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

411 raise RuntimeError(error) from exc 

412 _post_success(app, success_message) 

413 

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