Coverage for src/lilbee/cli/sessions.py: 100%
122 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""CLI for listing and managing saved chat sessions."""
3from __future__ import annotations
5from dataclasses import asdict
6from pathlib import Path
7from typing import NoReturn
9import typer
10from rich.table import Table
11from rich.text import Text
13from lilbee.app.session_export import session_markdown, write_session_markdown
14from lilbee.app.session_sources import message_payload
15from lilbee.cli import theme
16from lilbee.cli.app import apply_overrides, console, data_dir_option, global_option
17from lilbee.cli.helpers import json_output
18from lilbee.core.config import cfg
19from lilbee.core.text import strip_control_characters
20from lilbee.runtime.console import styled
21from lilbee.sessions import (
22 SESSIONS_DISABLED_HINT,
23 SessionForkRangeError,
24 SessionOrigin,
25 SessionOwnershipError,
26 SessionStore,
27 TitleSource,
28 sessions_enabled,
29)
31sessions_app = typer.Typer(
32 name="sessions",
33 help="List and manage saved chat sessions.",
34 no_args_is_help=True,
35)
37_yes_option = typer.Option(False, "--yes", "-y", help="Skip the delete confirmation.")
38_id_argument = typer.Argument(..., help="Session id, or a unique prefix of it.")
39_messages_option = typer.Option(
40 None, "--messages", help="Copy only the first N messages (default: all of them)."
41)
42_output_option = typer.Option(
43 None,
44 "--output",
45 "-o",
46 help="Write to this file, or into this directory, instead of printing to stdout.",
47)
50def _require_sessions() -> None:
51 """Report that sessions are off and exit; every command reaches the store
52 through ``_store``, so the check lives there rather than in each command.
53 """
54 if sessions_enabled():
55 return
56 if cfg.json_mode:
57 json_output({"error": SESSIONS_DISABLED_HINT})
58 else:
59 console.print(SESSIONS_DISABLED_HINT)
60 raise typer.Exit(0)
63def _store() -> SessionStore:
64 _require_sessions()
65 return SessionStore()
68def _fail(message: str) -> NoReturn:
69 if cfg.json_mode:
70 json_output({"error": message})
71 else:
72 # Text, not markup: messages carry user paths and prefixes verbatim.
73 console.print(Text(message, style=theme.ERROR), soft_wrap=True)
74 raise typer.Exit(1)
77def _resolve_id(prefix: str) -> str:
78 """Resolve a full id or unique prefix to a session id, or exit 1."""
79 matches = [meta.id for meta in _store().list() if meta.id.startswith(prefix)]
80 if len(matches) == 1:
81 return matches[0]
82 if not matches:
83 _fail(f"No session matching {prefix!r}.")
84 _fail(f"Prefix {prefix!r} is ambiguous ({len(matches)} sessions match).")
87@sessions_app.command("list")
88def list_cmd(
89 data_dir: Path | None = data_dir_option,
90 use_global: bool = global_option,
91) -> None:
92 """List saved conversations, newest first."""
93 apply_overrides(data_dir=data_dir, use_global=use_global)
94 metas = _store().list()
95 if cfg.json_mode:
96 json_output({"sessions": [asdict(meta) for meta in metas]})
97 return
98 if not metas:
99 console.print("No saved sessions.")
100 return
101 table = Table(box=None, pad_edge=False)
102 for column in ("ID", "Title", "Msgs", "Model", "Origin", "Updated"):
103 table.add_column(column, justify="right" if column == "Msgs" else "left")
104 for meta in metas:
105 table.add_row(
106 meta.id[:8],
107 Text(strip_control_characters(meta.title)),
108 str(meta.message_count),
109 Text(meta.model_ref),
110 meta.origin.value,
111 meta.updated_at[:19],
112 )
113 console.print(table)
116@sessions_app.command("show")
117def show_cmd(
118 session_id: str = _id_argument,
119 data_dir: Path | None = data_dir_option,
120 use_global: bool = global_option,
121) -> None:
122 """Print a saved conversation's transcript."""
123 apply_overrides(data_dir=data_dir, use_global=use_global)
124 session = _store().get(_resolve_id(session_id))
125 if cfg.json_mode:
126 json_output(
127 {
128 "meta": asdict(session.meta),
129 "messages": [message_payload(m) for m in session.messages],
130 # What compaction folded older turns into (empty if never
131 # compacted). A script that resumes from this JSON needs it, or
132 # it rebuilds history without what was already condensed -- the
133 # same hole the HTTP and MCP surfaces used to have.
134 "summary": session.summary,
135 }
136 )
137 return
138 console.print(
139 Text(strip_control_characters(session.meta.title), style=theme.ACCENT), soft_wrap=True
140 )
141 for message in session.messages:
142 console.print(
143 Text.assemble(
144 (message.role.value, "bold"),
145 ": ",
146 strip_control_characters(message.content, keep_lines=True),
147 ),
148 soft_wrap=True,
149 )
152@sessions_app.command("fork")
153def fork_cmd(
154 session_id: str = _id_argument,
155 messages: int | None = _messages_option,
156 data_dir: Path | None = data_dir_option,
157 use_global: bool = global_option,
158) -> None:
159 """Start a new conversation from a copy of a saved one."""
160 apply_overrides(data_dir=data_dir, use_global=use_global)
161 store = _store()
162 try:
163 fork_id = store.fork(
164 _resolve_id(session_id), message_count=messages, origin=SessionOrigin.CLI
165 )
166 except (SessionOwnershipError, SessionForkRangeError) as exc:
167 _fail(str(exc))
168 meta = store.get(fork_id).meta
169 if cfg.json_mode:
170 json_output({"meta": asdict(meta)})
171 return
172 console.print(
173 Text.assemble(
174 "Forked to ",
175 (strip_control_characters(meta.title), theme.ACCENT),
176 f" ({fork_id[:8]}).",
177 ),
178 soft_wrap=True,
179 )
182@sessions_app.command("export")
183def export_cmd(
184 session_id: str = _id_argument,
185 output: str | None = _output_option,
186 data_dir: Path | None = data_dir_option,
187 use_global: bool = global_option,
188) -> None:
189 """Export a saved conversation as markdown."""
190 apply_overrides(data_dir=data_dir, use_global=use_global)
191 resolved = _resolve_id(session_id)
192 session = _store().get(resolved)
193 if output is None:
194 markdown = session_markdown(session)
195 if cfg.json_mode:
196 json_output({"id": resolved, "markdown": markdown})
197 return
198 # Bytes go to the binary stream, so a redirect is UTF-8 on every platform.
199 typer.echo(markdown.encode("utf-8"), nl=False)
200 return
201 try:
202 path = write_session_markdown(session, output)
203 except OSError as exc:
204 _fail(f"Could not write the export: {exc}")
205 if cfg.json_mode:
206 json_output({"id": resolved, "path": str(path)})
207 return
208 # A Text, not markup: a Windows separator before "[" would read as an escape.
209 # soft_wrap keeps a long path on one line, so it copies whole.
210 console.print(Text.assemble("Exported to ", (str(path), theme.ACCENT), "."), soft_wrap=True)
213@sessions_app.command("rename")
214def rename_cmd(
215 session_id: str = _id_argument,
216 title: str = typer.Argument(..., help="The new title."),
217 data_dir: Path | None = data_dir_option,
218 use_global: bool = global_option,
219) -> None:
220 """Rename a saved conversation."""
221 apply_overrides(data_dir=data_dir, use_global=use_global)
222 resolved = _resolve_id(session_id)
223 _store().set_title(resolved, title, TitleSource.CUSTOM)
224 if cfg.json_mode:
225 json_output({"id": resolved, "title": title})
226 return
227 console.print(Text.assemble("Renamed to ", (title, theme.ACCENT), "."), soft_wrap=True)
230@sessions_app.command("delete")
231def delete_cmd(
232 session_id: str = _id_argument,
233 yes: bool = _yes_option,
234 data_dir: Path | None = data_dir_option,
235 use_global: bool = global_option,
236) -> None:
237 """Delete a saved conversation."""
238 apply_overrides(data_dir=data_dir, use_global=use_global)
239 resolved = _resolve_id(session_id)
240 if (
241 not yes
242 and not cfg.json_mode
243 and not typer.confirm(f"Delete {resolved[:8]}?", default=False)
244 ):
245 raise typer.Abort()
246 _store().delete(resolved)
247 if cfg.json_mode:
248 json_output({"id": resolved, "deleted": True})
249 return
250 console.print(styled("Deleted ", (resolved[:8], theme.ACCENT), "."))