Coverage for little_loops / cli / messages.py: 96%
106 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-22 16:19 -0500
1"""ll-messages: Extract user messages from Claude Code session logs."""
3from __future__ import annotations
5import argparse
6from pathlib import Path
8from little_loops.logger import Logger
11def main_messages() -> int:
12 """Entry point for ll-messages command.
14 Extract user messages from Claude Code session logs.
16 Returns:
17 Exit code (0 = success)
18 """
19 import json
20 from datetime import datetime
22 from little_loops.user_messages import (
23 CommandRecord,
24 UserMessage,
25 build_examples,
26 extract_commands,
27 extract_user_messages,
28 get_project_folder,
29 )
31 parser = argparse.ArgumentParser(
32 description="Extract user messages from Claude Code logs",
33 formatter_class=argparse.RawDescriptionHelpFormatter,
34 epilog="""
35Examples:
36 %(prog)s # Last 100 messages to file
37 %(prog)s -n 50 # Last 50 messages
38 %(prog)s --since 2026-01-01 # Messages since date
39 %(prog)s -o output.jsonl # Custom output path
40 %(prog)s --stdout # Print to terminal
41 %(prog)s --include-response-context # Include response metadata
42 %(prog)s --skip-cli # Exclude CLI commands from output
43 %(prog)s --commands-only # Extract only CLI commands
44 %(prog)s --skill capture-issue # Filter to sessions where skill was invoked
45 %(prog)s --skill capture-issue --examples-format # Output (input, output) training pairs
46 %(prog)s --skill refine-issue --examples-format --context-window 5 --stdout
48Pipeline with ll-workflows (use the conventional path so ll-workflows finds it automatically):
49 %(prog)s --output .ll/workflow-analysis/step1-patterns.jsonl
50 ll-workflows analyze --patterns .ll/workflow-analysis/step1-patterns.yaml
51""",
52 )
53 parser.add_argument(
54 "-n",
55 "--limit",
56 type=int,
57 default=100,
58 help="Maximum number of messages to extract (default: 100)",
59 )
60 parser.add_argument(
61 "--since",
62 "-S",
63 type=str,
64 help="Only include messages after this date (YYYY-MM-DD or ISO format)",
65 )
66 parser.add_argument(
67 "-o",
68 "--output",
69 type=Path,
70 help="Output file path (default: .ll/user-messages-{timestamp}.jsonl)",
71 )
72 parser.add_argument(
73 "--cwd",
74 type=Path,
75 help="Working directory to use (default: current directory)",
76 )
77 parser.add_argument(
78 "--exclude-agents",
79 action="store_true",
80 help="Exclude agent session files (agent-*.jsonl)",
81 )
82 parser.add_argument(
83 "--stdout",
84 action="store_true",
85 help="Print messages to stdout instead of writing to file",
86 )
87 parser.add_argument(
88 "-v",
89 "--verbose",
90 action="store_true",
91 help="Print verbose progress information",
92 )
93 parser.add_argument(
94 "--include-response-context",
95 action="store_true",
96 help="Include metadata from assistant responses (tools used, files modified)",
97 )
98 parser.add_argument(
99 "--skip-cli",
100 action="store_true",
101 help="Exclude CLI commands from output (included by default)",
102 )
103 parser.add_argument(
104 "--commands-only",
105 action="store_true",
106 help="Extract only CLI commands, no user messages",
107 )
108 parser.add_argument(
109 "--tools",
110 type=str,
111 default="Bash",
112 help="Comma-separated list of tools to extract commands from (default: Bash)",
113 )
114 parser.add_argument(
115 "--skill",
116 type=str,
117 help="Filter to sessions where this skill was invoked (e.g. capture-issue)",
118 )
119 parser.add_argument(
120 "--examples-format",
121 action="store_true",
122 help="Output (input, output) training pairs for prompt optimization instead of raw messages",
123 )
124 parser.add_argument(
125 "--context-window",
126 type=int,
127 default=3,
128 help="Number of preceding messages to include as context in --examples-format (default: 3)",
129 )
131 args = parser.parse_args()
133 logger = Logger(verbose=args.verbose)
135 # Parse since date if provided
136 since = None
137 if args.since:
138 try:
139 # Try ISO format first
140 since = datetime.fromisoformat(args.since.replace("Z", "+00:00"))
141 except ValueError:
142 try:
143 # Try YYYY-MM-DD format
144 since = datetime.strptime(args.since, "%Y-%m-%d")
145 except ValueError:
146 logger.error(f"Invalid date format: {args.since}")
147 logger.error("Use YYYY-MM-DD or ISO format")
148 return 1
150 # Get project folder
151 cwd = args.cwd or Path.cwd()
152 project_folder = get_project_folder(cwd)
154 if project_folder is None:
155 logger.error(f"No Claude project folder found for: {cwd}")
156 logger.error(f"Expected: ~/.claude/projects/{str(cwd).replace('/', '-')}")
157 return 1
159 logger.info(f"Project folder: {project_folder}")
160 logger.info(f"Limit: {args.limit}")
161 if since:
162 logger.info(f"Since: {since}")
164 # Parse tools list
165 tools_list = [t.strip() for t in args.tools.split(",")]
167 # Extract data based on flags
168 messages: list[UserMessage] = []
169 commands: list[CommandRecord] = []
171 if not args.commands_only:
172 messages = extract_user_messages(
173 project_folder=project_folder,
174 limit=None, # Apply limit after merging
175 since=since,
176 include_agent_sessions=not args.exclude_agents,
177 include_response_context=args.include_response_context or args.examples_format,
178 )
180 if not args.skip_cli or args.commands_only:
181 commands = extract_commands(
182 project_folder=project_folder,
183 limit=None, # Apply limit after merging
184 since=since,
185 include_agent_sessions=not args.exclude_agents,
186 tools=tools_list,
187 )
189 # Apply skill filter (session-level): keep only sessions where skill was invoked
190 if args.skill:
191 import re
193 skill_pattern = re.compile(rf"<command-name>/ll:{re.escape(args.skill)}</command-name>")
194 matching_sessions = {
195 msg.session_id for msg in messages if skill_pattern.search(msg.content)
196 }
197 messages = [msg for msg in messages if msg.session_id in matching_sessions]
198 commands = [c for c in commands if c.session_id in matching_sessions]
200 if not messages and not commands:
201 logger.warning("No user messages or commands found")
202 return 0
204 # When --examples-format is set, reshape to ExampleRecord output and return early
205 if args.examples_format:
206 if not args.skill:
207 logger.error("--examples-format requires --skill to be specified")
208 return 1
209 examples = build_examples(messages, args.skill, args.context_window)
210 examples.sort(key=lambda x: x.timestamp, reverse=True)
211 if args.limit is not None:
212 examples = examples[: args.limit]
213 logger.info(f"Found {len(examples)} examples")
214 if args.stdout:
215 for item in examples:
216 print(json.dumps(item.to_dict()))
217 else:
218 output_path = _save_combined(examples, args.output)
219 logger.success(f"Saved {len(examples)} examples to: {output_path}")
220 return 0
222 # Merge and sort by timestamp
223 combined: list[UserMessage | CommandRecord] = []
224 combined.extend(messages)
225 combined.extend(commands)
226 combined.sort(key=lambda x: x.timestamp, reverse=True)
228 # Apply limit
229 if args.limit is not None:
230 combined = combined[: args.limit]
232 msg_count = len([x for x in combined if isinstance(x, UserMessage)])
233 cmd_count = len([x for x in combined if isinstance(x, CommandRecord)])
234 logger.info(f"Found {msg_count} messages, {cmd_count} commands")
236 # Output
237 if args.stdout:
238 for record in combined:
239 print(json.dumps(record.to_dict()))
240 else:
241 output_path = _save_combined(combined, args.output)
242 logger.success(f"Saved {len(combined)} records to: {output_path}")
244 return 0
247def _save_combined(
248 items: list,
249 output_path: Path | None = None,
250) -> Path:
251 """Save combined messages and commands to JSONL file."""
252 import json
253 from datetime import datetime
255 if output_path is None:
256 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
257 output_dir = Path.cwd() / ".claude"
258 output_dir.mkdir(parents=True, exist_ok=True)
259 output_path = output_dir / f"user-messages-{timestamp}.jsonl"
261 output_path = Path(output_path)
262 output_path.parent.mkdir(parents=True, exist_ok=True)
264 with open(output_path, "w", encoding="utf-8") as f:
265 for item in items:
266 f.write(json.dumps(item.to_dict()) + "\n")
268 return output_path