Coverage for little_loops / cli / messages.py: 8%
130 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:11 -0500
1"""ll-messages: Extract user messages from Claude Code session logs."""
3from __future__ import annotations
5import argparse
6import sys
7from pathlib import Path
9from little_loops.logger import Logger
10from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
13def main_messages() -> int:
14 """Entry point for ll-messages command.
16 Extract user messages from Claude Code session logs.
18 Returns:
19 Exit code (0 = success)
20 """
21 with cli_event_context(DEFAULT_DB_PATH, "ll-messages", sys.argv[1:]):
22 import json
23 from datetime import datetime
25 from little_loops.user_messages import (
26 CommandRecord,
27 UserMessage,
28 build_examples,
29 extract_commands,
30 extract_user_messages,
31 get_project_folder,
32 )
34 parser = argparse.ArgumentParser(
35 description="Extract user messages from Claude Code logs",
36 formatter_class=argparse.RawDescriptionHelpFormatter,
37 epilog="""
38Examples:
39 %(prog)s # Last 100 messages to file
40 %(prog)s -n 50 # Last 50 messages
41 %(prog)s --since 2026-01-01 # Messages since date
42 %(prog)s -o output.jsonl # Custom output path
43 %(prog)s --stdout # Print to terminal
44 %(prog)s --include-response-context # Include response metadata
45 %(prog)s --skip-cli # Exclude CLI commands from output
46 %(prog)s --commands-only # Extract only CLI commands
47 %(prog)s --skill capture-issue # Filter to sessions where skill was invoked
48 %(prog)s --skill capture-issue --examples-format # Output (input, output) training pairs
49 %(prog)s --skill refine-issue --examples-format --context-window 5 --stdout
50 %(prog)s --sft-format chatml --stdout
51 %(prog)s --sft-format sharegpt --context-window 3 --since 2026-05-01 --stdout
52 %(prog)s --sft-format alpaca --output data/sft/raw.jsonl
53 %(prog)s --sft-format chatml --reader db --stdout
54 %(prog)s --sft-format chatml --reader jsonl --stdout
56Pipeline with ll-workflows (use the conventional path so ll-workflows finds it automatically):
57 %(prog)s --output .ll/workflow-analysis/step1-patterns.jsonl
58 ll-workflows analyze --patterns .ll/workflow-analysis/step1-patterns.yaml
59""",
60 )
61 parser.add_argument(
62 "-n",
63 "--limit",
64 type=int,
65 default=100,
66 help="Maximum number of messages to extract (default: 100)",
67 )
68 parser.add_argument(
69 "--since",
70 "-S",
71 type=str,
72 help="Only include messages after this date (YYYY-MM-DD or ISO format)",
73 )
74 parser.add_argument(
75 "-o",
76 "--output",
77 type=Path,
78 help="Output file path (default: .ll/user-messages-{timestamp}.jsonl)",
79 )
80 parser.add_argument(
81 "--cwd",
82 type=Path,
83 help="Working directory to use (default: current directory)",
84 )
85 parser.add_argument(
86 "--exclude-agents",
87 action="store_true",
88 help="Exclude agent session files (agent-*.jsonl)",
89 )
90 parser.add_argument(
91 "--stdout",
92 action="store_true",
93 help="Print messages to stdout instead of writing to file",
94 )
95 parser.add_argument(
96 "-v",
97 "--verbose",
98 action="store_true",
99 help="Print verbose progress information",
100 )
101 parser.add_argument(
102 "--include-response-context",
103 action="store_true",
104 help="Include metadata from assistant responses (tools used, files modified)",
105 )
106 parser.add_argument(
107 "--skip-cli",
108 action="store_true",
109 help="Exclude CLI commands from output (included by default)",
110 )
111 parser.add_argument(
112 "--commands-only",
113 action="store_true",
114 help="Extract only CLI commands, no user messages",
115 )
116 parser.add_argument(
117 "--tools",
118 type=str,
119 default="Bash",
120 help="Comma-separated list of tools to extract commands from (default: Bash)",
121 )
122 parser.add_argument(
123 "--skill",
124 type=str,
125 help="Filter to sessions where this skill was invoked (e.g. capture-issue)",
126 )
127 format_group = parser.add_mutually_exclusive_group()
128 format_group.add_argument(
129 "--examples-format",
130 action="store_true",
131 help="Output (input, output) training pairs for prompt optimization instead of raw messages",
132 )
133 format_group.add_argument(
134 "--sft-format",
135 choices=["chatml", "alpaca", "sharegpt"],
136 help="Output conversation turns in SFT training format as JSON-lines (chatml, alpaca, sharegpt)",
137 )
138 parser.add_argument(
139 "--context-window",
140 type=int,
141 default=3,
142 help="Number of preceding messages to include as context in --examples-format or --sft-format (default: 3)",
143 )
144 parser.add_argument(
145 "--reader",
146 choices=["auto", "db", "jsonl"],
147 default="auto",
148 help="Data source for --sft-format: auto (DB first, JSONL fallback), "
149 "db (DB only, error if unavailable), jsonl (JSONL only) (default: auto)",
150 )
152 args = parser.parse_args()
154 logger = Logger(verbose=args.verbose)
156 # Parse since date if provided
157 since = None
158 if args.since:
159 try:
160 # Try ISO format first
161 since = datetime.fromisoformat(args.since.replace("Z", "+00:00"))
162 except ValueError:
163 try:
164 # Try YYYY-MM-DD format
165 since = datetime.strptime(args.since, "%Y-%m-%d")
166 except ValueError:
167 logger.error(f"Invalid date format: {args.since}")
168 logger.error("Use YYYY-MM-DD or ISO format")
169 return 1
171 # Get project folder
172 cwd = args.cwd or Path.cwd()
173 project_folder = get_project_folder(cwd)
175 if project_folder is None:
176 logger.error(f"No session project folder found for: {cwd}")
177 return 1
179 logger.info(f"Project folder: {project_folder}")
180 logger.info(f"Limit: {args.limit}")
181 if since:
182 logger.info(f"Since: {since}")
184 # Parse tools list
185 tools_list = [t.strip() for t in args.tools.split(",")]
187 # Extract data based on flags
188 messages: list[UserMessage] = []
189 commands: list[CommandRecord] = []
191 if not args.commands_only:
192 messages = extract_user_messages(
193 project_folder=project_folder,
194 limit=None, # Apply limit after merging
195 since=since,
196 include_agent_sessions=not args.exclude_agents,
197 include_response_context=args.include_response_context or args.examples_format,
198 )
200 if not args.skip_cli or args.commands_only:
201 commands = extract_commands(
202 project_folder=project_folder,
203 limit=None, # Apply limit after merging
204 since=since,
205 include_agent_sessions=not args.exclude_agents,
206 tools=tools_list,
207 )
209 # Apply skill filter (session-level): keep only sessions where skill was invoked
210 if args.skill:
211 import re
213 skill_pattern = re.compile(rf"<command-name>/ll:{re.escape(args.skill)}</command-name>")
214 matching_sessions = {
215 msg.session_id for msg in messages if skill_pattern.search(msg.content)
216 }
217 messages = [msg for msg in messages if msg.session_id in matching_sessions]
218 commands = [c for c in commands if c.session_id in matching_sessions]
220 if not messages and not commands:
221 logger.warning("No user messages or commands found")
222 return 0
224 # When --examples-format is set, reshape to ExampleRecord output and return early
225 if args.examples_format:
226 if not args.skill:
227 logger.error("--examples-format requires --skill to be specified")
228 return 1
229 examples = build_examples(messages, args.skill, args.context_window)
230 examples.sort(key=lambda x: x.timestamp, reverse=True)
231 if args.limit is not None:
232 examples = examples[: args.limit]
233 logger.info(f"Found {len(examples)} examples")
234 if args.stdout:
235 for item in examples:
236 print(json.dumps(item.to_dict()))
237 else:
238 output_path = _save_combined(examples, args.output)
239 logger.success(f"Saved {len(examples)} examples to: {output_path}")
240 return 0
242 # When --sft-format is set, emit conversation turns in SFT training format
243 if args.sft_format:
244 from little_loops.sft_formatter import to_alpaca, to_chatml, to_sharegpt
245 from little_loops.user_messages import extract_conversation_turns
247 formatter = {"chatml": to_chatml, "alpaca": to_alpaca, "sharegpt": to_sharegpt}[
248 args.sft_format
249 ]
250 windows = extract_conversation_turns(
251 project_folder=project_folder,
252 since=since,
253 context_window=args.context_window,
254 include_agent_sessions=not args.exclude_agents,
255 reader=args.reader,
256 )
257 if args.limit is not None:
258 windows = windows[: args.limit]
259 logger.info(f"Found {len(windows)} conversation windows")
260 if args.stdout:
261 for window in windows:
262 print(json.dumps(formatter(window)))
263 else:
264 output_path = _save_combined([_SFTItem(formatter(w)) for w in windows], args.output)
265 logger.success(f"Saved {len(windows)} windows to: {output_path}")
266 return 0
268 # Merge and sort by timestamp
269 combined: list[UserMessage | CommandRecord] = []
270 combined.extend(messages)
271 combined.extend(commands)
272 combined.sort(key=lambda x: x.timestamp, reverse=True)
274 # Apply limit
275 if args.limit is not None:
276 combined = combined[: args.limit]
278 msg_count = len([x for x in combined if isinstance(x, UserMessage)])
279 cmd_count = len([x for x in combined if isinstance(x, CommandRecord)])
280 logger.info(f"Found {msg_count} messages, {cmd_count} commands")
282 # Output
283 if args.stdout:
284 for record in combined:
285 print(json.dumps(record.to_dict()))
286 else:
287 output_path = _save_combined(combined, args.output)
288 logger.success(f"Saved {len(combined)} records to: {output_path}")
290 return 0
293def _save_combined(
294 items: list,
295 output_path: Path | None = None,
296) -> Path:
297 """Save combined messages and commands to JSONL file."""
298 import json
299 from datetime import datetime
301 if output_path is None:
302 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
303 output_dir = Path.cwd() / ".claude"
304 output_dir.mkdir(parents=True, exist_ok=True)
305 output_path = output_dir / f"user-messages-{timestamp}.jsonl"
307 output_path = Path(output_path)
308 output_path.parent.mkdir(parents=True, exist_ok=True)
310 with open(output_path, "w", encoding="utf-8") as f:
311 for item in items:
312 f.write(json.dumps(item.to_dict()) + "\n")
314 return output_path
317class _SFTItem:
318 """Thin wrapper to make an SFT dict compatible with _save_combined()."""
320 def __init__(self, data: dict) -> None:
321 self._data = data
323 def to_dict(self) -> dict:
324 return self._data