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