Coverage for src/lilbee/cli/sessions.py: 100%

1"""CLI for listing and managing saved chat sessions.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import asdict 

6from pathlib import Path 

7from typing import NoReturn 

8 

9import typer 

10from rich.table import Table 

11 

12from lilbee.cli import theme 

13from lilbee.cli.app import apply_overrides, console, data_dir_option, global_option 

14from lilbee.cli.helpers import json_output 

15from lilbee.core.config import cfg 

16from lilbee.sessions import ( 

17 SESSIONS_DISABLED_HINT, 

18 SessionStore, 

19 TitleSource, 

20 sessions_enabled, 

21) 

22 

23sessions_app = typer.Typer( 

24 name="sessions", 

25 help="List and manage saved chat sessions.", 

26 no_args_is_help=True, 

27) 

28 

29_yes_option = typer.Option(False, "--yes", "-y", help="Skip the delete confirmation.") 

30_id_argument = typer.Argument(..., help="Session id, or a unique prefix of it.") 

31 

32 

33def _require_sessions() -> None: 

34 """Report that sessions are off and exit; every command reaches the store 

35 through ``_store``, so the check lives there rather than in each command. 

36 """ 

37 if sessions_enabled(): 

38 return 

39 if cfg.json_mode: 

40 json_output({"error": SESSIONS_DISABLED_HINT}) 

41 else: 

42 console.print(SESSIONS_DISABLED_HINT) 

43 raise typer.Exit(0) 

44 

45 

46def _store() -> SessionStore: 

47 _require_sessions() 

48 return SessionStore() 

49 

50 

51def _fail(message: str) -> NoReturn: 

52 if cfg.json_mode: 

53 json_output({"error": message}) 

54 else: 

55 console.print(f"[{theme.ERROR}]{message}[/{theme.ERROR}]") 

56 raise typer.Exit(1) 

57 

58 

59def _resolve_id(prefix: str) -> str: 

60 """Resolve a full id or unique prefix to a session id, or exit 1.""" 

61 matches = [meta.id for meta in _store().list() if meta.id.startswith(prefix)] 

62 if len(matches) == 1: 

63 return matches[0] 

64 if not matches: 

65 _fail(f"No session matching {prefix!r}.") 

66 _fail(f"Prefix {prefix!r} is ambiguous ({len(matches)} sessions match).") 

67 

68 

69@sessions_app.command("list") 

70def list_cmd( 

71 data_dir: Path | None = data_dir_option, 

72 use_global: bool = global_option, 

73) -> None: 

74 """List saved conversations, newest first.""" 

75 apply_overrides(data_dir=data_dir, use_global=use_global) 

76 metas = _store().list() 

77 if cfg.json_mode: 

78 json_output({"sessions": [asdict(meta) for meta in metas]}) 

79 return 

80 if not metas: 

81 console.print("No saved sessions.") 

82 return 

83 table = Table(box=None, pad_edge=False) 

84 for column in ("ID", "Title", "Msgs", "Model", "Origin", "Updated"): 

85 table.add_column(column, justify="right" if column == "Msgs" else "left") 

86 for meta in metas: 

87 table.add_row( 

88 meta.id[:8], 

89 meta.title, 

90 str(meta.message_count), 

91 meta.model_ref, 

92 meta.origin.value, 

93 meta.updated_at[:19], 

94 ) 

95 console.print(table) 

96 

97 

98@sessions_app.command("show") 

99def show_cmd( 

100 session_id: str = _id_argument, 

101 data_dir: Path | None = data_dir_option, 

102 use_global: bool = global_option, 

103) -> None: 

104 """Print a saved conversation's transcript.""" 

105 apply_overrides(data_dir=data_dir, use_global=use_global) 

106 session = _store().get(_resolve_id(session_id)) 

107 if cfg.json_mode: 

108 json_output( 

109 { 

110 "meta": asdict(session.meta), 

111 "messages": [ 

112 {"role": m.role.value, "content": m.content, "sources": list(m.sources)} 

113 for m in session.messages 

114 ], 

115 # What compaction folded older turns into (empty if never 

116 # compacted). A script that resumes from this JSON needs it, or 

117 # it rebuilds history without what was already condensed -- the 

118 # same hole the HTTP and MCP surfaces used to have. 

119 "summary": session.summary, 

120 } 

121 ) 

122 return 

123 console.print(f"[{theme.ACCENT}]{session.meta.title}[/{theme.ACCENT}]") 

124 for message in session.messages: 

125 console.print(f"[bold]{message.role.value}[/bold]: {message.content}") 

126 

127 

128@sessions_app.command("rename") 

129def rename_cmd( 

130 session_id: str = _id_argument, 

131 title: str = typer.Argument(..., help="The new title."), 

132 data_dir: Path | None = data_dir_option, 

133 use_global: bool = global_option, 

134) -> None: 

135 """Rename a saved conversation.""" 

136 apply_overrides(data_dir=data_dir, use_global=use_global) 

137 resolved = _resolve_id(session_id) 

138 _store().set_title(resolved, title, TitleSource.CUSTOM) 

139 if cfg.json_mode: 

140 json_output({"id": resolved, "title": title}) 

141 return 

142 console.print(f"Renamed to [{theme.ACCENT}]{title}[/{theme.ACCENT}].") 

143 

144 

145@sessions_app.command("delete") 

146def delete_cmd( 

147 session_id: str = _id_argument, 

148 yes: bool = _yes_option, 

149 data_dir: Path | None = data_dir_option, 

150 use_global: bool = global_option, 

151) -> None: 

152 """Delete a saved conversation.""" 

153 apply_overrides(data_dir=data_dir, use_global=use_global) 

154 resolved = _resolve_id(session_id) 

155 if ( 

156 not yes 

157 and not cfg.json_mode 

158 and not typer.confirm(f"Delete {resolved[:8]}?", default=False) 

159 ): 

160 raise typer.Abort() 

161 _store().delete(resolved) 

162 if cfg.json_mode: 

163 json_output({"id": resolved, "deleted": True}) 

164 return 

165 console.print(f"Deleted [{theme.ACCENT}]{resolved[:8]}[/{theme.ACCENT}].")