Coverage for little_loops / user_messages.py: 15%
393 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -0500
1"""Extract and analyze user messages from Claude Code logs.
3Provides functionality to extract user messages from Claude Code session
4logs stored in ~/.claude/projects/.
6Usage as CLI:
7 ll-messages # Last 100 messages to file
8 ll-messages -n 50 # Last 50 messages
9 ll-messages --since 2026-01-01 # Since date
10 ll-messages -o output.jsonl # Custom output path
11 ll-messages --stdout # Print to terminal instead of file
13Usage as library:
14 from little_loops.user_messages import extract_user_messages, get_project_folder
16 project_folder = get_project_folder()
17 messages = extract_user_messages(project_folder, limit=50)
18"""
20from __future__ import annotations
22import json
23from dataclasses import dataclass
24from datetime import datetime, timedelta
25from pathlib import Path
27__all__ = [
28 "UserMessage",
29 "ResponseMetadata",
30 "CommandRecord",
31 "ExampleRecord",
32 "get_project_folder",
33 "extract_user_messages",
34 "extract_commands",
35 "build_examples",
36 "extract_conversation_turns",
37 "save_messages",
38]
41@dataclass
42class UserMessage:
43 """Extracted user message with metadata.
45 Attributes:
46 content: The text content of the user message
47 timestamp: When the message was sent
48 session_id: Claude Code session identifier
49 uuid: Unique message identifier
50 cwd: Working directory when message was sent
51 git_branch: Git branch active when message was sent
52 is_sidechain: Whether this was a sidechain message
53 """
55 content: str
56 timestamp: datetime
57 session_id: str
58 uuid: str
59 cwd: str | None = None
60 git_branch: str | None = None
61 is_sidechain: bool = False
63 response_metadata: ResponseMetadata | None = None
65 def to_dict(self) -> dict[str, object]:
66 """Convert to dictionary for JSON serialization."""
67 result: dict[str, object] = {
68 "content": self.content,
69 "timestamp": self.timestamp.isoformat(),
70 "session_id": self.session_id,
71 "uuid": self.uuid,
72 "cwd": self.cwd,
73 "git_branch": self.git_branch,
74 "is_sidechain": self.is_sidechain,
75 }
76 if self.response_metadata is not None:
77 result["response_metadata"] = self.response_metadata.to_dict()
78 return result
81@dataclass
82class ResponseMetadata:
83 """Metadata extracted from assistant response.
85 Attributes:
86 tools_used: List of tools and their usage counts
87 files_read: Files accessed via Read tool
88 files_modified: Files changed via Edit/Write tools
89 completion_status: "success", "failure", or "partial"
90 error_message: Error text if failure detected
91 """
93 tools_used: list[dict[str, str | int]]
94 files_read: list[str]
95 files_modified: list[str]
96 completion_status: str
97 error_message: str | None = None
99 def to_dict(self) -> dict[str, object]:
100 """Convert to dictionary for JSON serialization."""
101 return {
102 "tools_used": self.tools_used,
103 "files_read": self.files_read,
104 "files_modified": self.files_modified,
105 "completion_status": self.completion_status,
106 "error_message": self.error_message,
107 }
110@dataclass
111class CommandRecord:
112 """Extracted CLI command from assistant tool_use.
114 Attributes:
115 content: The command string that was executed
116 timestamp: When the command was issued
117 session_id: Claude Code session identifier
118 uuid: Unique record identifier
119 tool: Tool name (e.g., "Bash")
120 cwd: Working directory when command was issued
121 git_branch: Git branch active when command was issued
122 """
124 content: str
125 timestamp: datetime
126 session_id: str
127 uuid: str
128 tool: str
129 cwd: str | None = None
130 git_branch: str | None = None
132 def to_dict(self) -> dict[str, object]:
133 """Convert to dictionary for JSON serialization."""
134 return {
135 "type": "command",
136 "content": self.content,
137 "timestamp": self.timestamp.isoformat(),
138 "session_id": self.session_id,
139 "uuid": self.uuid,
140 "tool": self.tool,
141 "cwd": self.cwd,
142 "git_branch": self.git_branch,
143 }
146@dataclass
147class ExampleRecord:
148 """Training example pair extracted from a skill invocation session.
150 Attributes:
151 skill: The skill name (e.g., "capture-issue")
152 input: Concatenated preceding user messages as context
153 output: JSON-serialized ResponseMetadata summary (tools_used, files_modified,
154 completion_status); free-text assistant response capture is deferred
155 session_id: Claude Code session identifier
156 timestamp: When the skill was invoked
157 context_window: Number of preceding messages used as context
158 """
160 skill: str
161 input: str
162 output: str
163 session_id: str
164 timestamp: datetime
165 context_window: int
167 def to_dict(self) -> dict[str, object]:
168 """Convert to dictionary for JSON serialization."""
169 return {
170 "type": "example",
171 "skill": self.skill,
172 "input": self.input,
173 "output": self.output,
174 "session_id": self.session_id,
175 "timestamp": self.timestamp.isoformat(),
176 "context_window": self.context_window,
177 }
180def _extract_response_metadata(response_record: dict) -> ResponseMetadata | None:
181 """Extract metadata from an assistant response record.
183 Args:
184 response_record: The assistant record from JSONL
186 Returns:
187 ResponseMetadata if parseable, None otherwise
188 """
189 message_data = response_record.get("message", {})
190 content = message_data.get("content", [])
192 if not isinstance(content, list):
193 return None
195 tools_used: dict[str, int] = {}
196 files_read: list[str] = []
197 files_modified: list[str] = []
199 for block in content:
200 if not isinstance(block, dict):
201 continue
202 if block.get("type") != "tool_use":
203 continue
205 tool_name = block.get("name", "")
206 tools_used[tool_name] = tools_used.get(tool_name, 0) + 1
208 tool_input = block.get("input", {})
209 if tool_name == "Read":
210 file_path = tool_input.get("file_path")
211 if file_path:
212 files_read.append(file_path)
213 elif tool_name in ("Edit", "Write"):
214 file_path = tool_input.get("file_path")
215 if file_path:
216 files_modified.append(file_path)
218 # Detect completion status from text content
219 completion_status = _detect_completion_status(content)
220 error_message = _detect_error_message(content) if completion_status == "failure" else None
222 # Convert tools_used dict to list format
223 tools_list: list[dict[str, str | int]] = [
224 {"tool": name, "count": count} for name, count in tools_used.items()
225 ]
227 return ResponseMetadata(
228 tools_used=tools_list,
229 files_read=files_read,
230 files_modified=files_modified,
231 completion_status=completion_status,
232 error_message=error_message,
233 )
236def _aggregate_response_metadata(responses: list[dict]) -> ResponseMetadata | None:
237 """Aggregate metadata from multiple assistant response records.
239 Combines tool counts, file lists, and uses completion status from final response.
241 Args:
242 responses: List of assistant records from JSONL
244 Returns:
245 Aggregated ResponseMetadata, or None if no valid responses
246 """
247 if not responses:
248 return None
250 tools_used: dict[str, int] = {}
251 files_read: set[str] = set()
252 files_modified: set[str] = set()
253 completion_status = "success"
254 error_message: str | None = None
256 for response_record in responses:
257 message_data = response_record.get("message", {})
258 content = message_data.get("content", [])
260 if not isinstance(content, list):
261 continue
263 for block in content:
264 if not isinstance(block, dict):
265 continue
266 if block.get("type") != "tool_use":
267 continue
269 tool_name = block.get("name", "")
270 tools_used[tool_name] = tools_used.get(tool_name, 0) + 1
272 tool_input = block.get("input", {})
273 if tool_name == "Read":
274 file_path = tool_input.get("file_path")
275 if file_path:
276 files_read.add(file_path)
277 elif tool_name in ("Edit", "Write"):
278 file_path = tool_input.get("file_path")
279 if file_path:
280 files_modified.add(file_path)
282 # Use completion status from the final response
283 final_content = responses[-1].get("message", {}).get("content", [])
284 if isinstance(final_content, list):
285 completion_status = _detect_completion_status(final_content)
286 if completion_status == "failure":
287 error_message = _detect_error_message(final_content)
289 # Convert to output format
290 tools_list: list[dict[str, str | int]] = [
291 {"tool": name, "count": count} for name, count in tools_used.items()
292 ]
294 return ResponseMetadata(
295 tools_used=tools_list,
296 files_read=sorted(files_read),
297 files_modified=sorted(files_modified),
298 completion_status=completion_status,
299 error_message=error_message,
300 )
303def _detect_completion_status(content: list) -> str:
304 """Detect completion status from response content.
306 Args:
307 content: List of content blocks from assistant response
309 Returns:
310 "success", "failure", or "partial"
311 """
312 text_parts = []
313 for block in content:
314 if isinstance(block, dict) and block.get("type") == "text":
315 text_parts.append(block.get("text", ""))
317 text = " ".join(text_parts).lower()
319 # Check for error indicators
320 error_patterns = ["error", "failed", "couldn't", "unable to", "cannot"]
321 if any(pattern in text for pattern in error_patterns):
322 return "failure"
324 # Check for partial completion
325 partial_patterns = ["partially", "some of", "not all", "incomplete"]
326 if any(pattern in text for pattern in partial_patterns):
327 return "partial"
329 return "success"
332def _detect_error_message(content: list) -> str | None:
333 """Extract error message from response content.
335 Args:
336 content: List of content blocks from assistant response
338 Returns:
339 Error message if found, None otherwise
340 """
341 for block in content:
342 if isinstance(block, dict) and block.get("type") == "text":
343 text = block.get("text", "")
344 # Look for common error message patterns
345 lower_text = text.lower()
346 if "error:" in lower_text or "failed:" in lower_text:
347 # Extract the line containing the error
348 for line in text.split("\n"):
349 if "error" in line.lower() or "failed" in line.lower():
350 result = line.strip()[:200] # Limit length
351 return result if isinstance(result, str) else None
352 return None
355def get_project_folder(cwd: Path | None = None) -> Path | None:
356 """Map current directory to Claude Code project folder.
358 Converts: /home/user/foo/bar -> ~/.claude/projects/-home-user-foo-bar
360 Args:
361 cwd: Working directory to map. If None, uses current directory.
363 Returns:
364 Path to Claude project folder, or None if it doesn't exist.
365 """
366 if cwd is None:
367 cwd = Path.cwd()
369 # Convert path to dash-separated format
370 # /home/user/foo/bar -> -home-user-foo-bar
371 path_str = str(cwd.resolve())
372 encoded_path = path_str.replace("/", "-")
374 # Build project folder path
375 claude_projects = Path.home() / ".claude" / "projects"
376 project_folder = claude_projects / encoded_path
378 if project_folder.exists():
379 return project_folder
381 return None
384def extract_user_messages(
385 project_folder: Path,
386 limit: int | None = None,
387 since: datetime | None = None,
388 include_agent_sessions: bool = True,
389 include_response_context: bool = False,
390) -> list[UserMessage]:
391 """Extract user messages from all JSONL session files.
393 Filters:
394 - type == "user"
395 - message.content is string (real user input)
396 - message.content is array but [0].type != "tool_result"
398 Args:
399 project_folder: Path to Claude project folder
400 limit: Maximum number of messages to return
401 since: Only include messages after this datetime
402 include_agent_sessions: Whether to include agent-*.jsonl files
403 include_response_context: Whether to include metadata from assistant responses
405 Returns:
406 Messages sorted by timestamp, most recent first.
407 """
408 messages: list[UserMessage] = []
410 # Find all JSONL files
411 pattern = "*.jsonl"
412 jsonl_files = list(project_folder.glob(pattern))
414 for jsonl_file in jsonl_files:
415 # Skip agent sessions if requested
416 if not include_agent_sessions and jsonl_file.name.startswith("agent-"):
417 continue
419 try:
420 # If we need response context, read all records first to pair user/assistant
421 if include_response_context:
422 all_records: list[dict] = []
423 with open(jsonl_file, encoding="utf-8") as f:
424 for line in f:
425 line = line.strip()
426 if not line:
427 continue
428 try:
429 record = json.loads(line)
430 all_records.append(record)
431 except json.JSONDecodeError:
432 continue
434 # Process records, pairing user messages with their responses
435 messages.extend(_extract_messages_with_context(all_records, jsonl_file, since))
436 else:
437 # Original behavior: stream through file
438 with open(jsonl_file, encoding="utf-8") as f:
439 for line in f:
440 line = line.strip()
441 if not line:
442 continue
444 try:
445 record = json.loads(line)
446 except json.JSONDecodeError:
447 continue
449 msg = _parse_user_record(record, jsonl_file, since)
450 if msg is not None:
451 messages.append(msg)
453 except OSError:
454 # Skip files that can't be read
455 continue
457 # Sort by timestamp, most recent first
458 messages.sort(key=lambda m: m.timestamp, reverse=True)
460 # Apply limit
461 if limit is not None:
462 messages = messages[:limit]
464 return messages
467def extract_commands(
468 project_folder: Path,
469 limit: int | None = None,
470 since: datetime | None = None,
471 include_agent_sessions: bool = True,
472 tools: list[str] | None = None,
473) -> list[CommandRecord]:
474 """Extract CLI commands from assistant tool_use messages.
476 Parses assistant messages for tool_use blocks and extracts command strings.
478 Args:
479 project_folder: Path to Claude project folder
480 limit: Maximum number of commands to return
481 since: Only include commands after this datetime
482 include_agent_sessions: Whether to include agent-*.jsonl files
483 tools: Filter to specific tools (default: ["Bash"])
485 Returns:
486 Commands sorted by timestamp, most recent first.
487 """
488 if tools is None:
489 tools = ["Bash"]
491 commands: list[CommandRecord] = []
493 # Find all JSONL files
494 pattern = "*.jsonl"
495 jsonl_files = list(project_folder.glob(pattern))
497 for jsonl_file in jsonl_files:
498 # Skip agent sessions if requested
499 if not include_agent_sessions and jsonl_file.name.startswith("agent-"):
500 continue
502 try:
503 with open(jsonl_file, encoding="utf-8") as f:
504 for line in f:
505 line = line.strip()
506 if not line:
507 continue
509 try:
510 record = json.loads(line)
511 except json.JSONDecodeError:
512 continue
514 cmds = _parse_command_record(record, jsonl_file, since, tools)
515 commands.extend(cmds)
517 except OSError:
518 # Skip files that can't be read
519 continue
521 # Sort by timestamp, most recent first
522 commands.sort(key=lambda c: c.timestamp, reverse=True)
524 # Apply limit
525 if limit is not None:
526 commands = commands[:limit]
528 return commands
531def _parse_command_record(
532 record: dict,
533 jsonl_file: Path,
534 since: datetime | None,
535 tools: list[str],
536) -> list[CommandRecord]:
537 """Parse CLI commands from an assistant record.
539 Args:
540 record: The JSON record from JSONL
541 jsonl_file: Source file (for fallback timestamp)
542 since: Filter for commands after this datetime
543 tools: Tool names to extract (e.g., ["Bash"])
545 Returns:
546 List of CommandRecord for each matching tool_use block
547 """
548 # Filter for assistant messages only
549 if record.get("type") != "assistant":
550 return []
552 message_data = record.get("message", {})
553 content = message_data.get("content", [])
555 if not isinstance(content, list):
556 return []
558 # Parse timestamp
559 timestamp_str = record.get("timestamp", "")
560 try:
561 timestamp_str = timestamp_str.replace("Z", "+00:00")
562 timestamp = datetime.fromisoformat(timestamp_str)
563 if timestamp.tzinfo is not None:
564 timestamp = timestamp.replace(tzinfo=None)
565 except (ValueError, AttributeError):
566 timestamp = datetime.fromtimestamp(jsonl_file.stat().st_mtime)
568 # Apply since filter
569 if since and timestamp < since:
570 return []
572 commands: list[CommandRecord] = []
574 for block in content:
575 if not isinstance(block, dict):
576 continue
577 if block.get("type") != "tool_use":
578 continue
580 tool_name = block.get("name", "")
581 if tool_name not in tools:
582 continue
584 tool_input = block.get("input", {})
585 command_str = tool_input.get("command", "")
586 if not command_str:
587 continue
589 commands.append(
590 CommandRecord(
591 content=command_str,
592 timestamp=timestamp,
593 session_id=record.get("sessionId", ""),
594 uuid=record.get("uuid", ""),
595 tool=tool_name,
596 cwd=record.get("cwd"),
597 git_branch=record.get("gitBranch"),
598 )
599 )
601 return commands
604def _parse_user_record(
605 record: dict,
606 jsonl_file: Path,
607 since: datetime | None,
608) -> UserMessage | None:
609 """Parse a single user record into a UserMessage.
611 Args:
612 record: The JSON record from JSONL
613 jsonl_file: Source file (for fallback timestamp)
614 since: Filter for messages after this datetime
616 Returns:
617 UserMessage if valid user message, None otherwise
618 """
619 # Filter for user messages only
620 if record.get("type") != "user":
621 return None
623 message_data = record.get("message", {})
624 content = message_data.get("content")
626 # Skip if no content
627 if content is None:
628 return None
630 # Check if this is a real user message or tool_result
631 if isinstance(content, str):
632 # String content = real user message
633 message_content = content
634 elif isinstance(content, list):
635 # Array content - check first element
636 if len(content) > 0 and content[0].get("type") == "tool_result":
637 # This is a tool result, skip it
638 return None
639 # Extract text from array (could be text blocks)
640 text_parts = []
641 for block in content:
642 if isinstance(block, dict):
643 if block.get("type") == "text":
644 text_parts.append(block.get("text", ""))
645 elif "content" in block:
646 text_parts.append(str(block.get("content", "")))
647 message_content = "\n".join(text_parts) if text_parts else str(content)
648 else:
649 return None
651 # Parse timestamp
652 timestamp_str = record.get("timestamp", "")
653 try:
654 # Handle ISO 8601 format with Z suffix
655 timestamp_str = timestamp_str.replace("Z", "+00:00")
656 timestamp = datetime.fromisoformat(timestamp_str)
657 # Convert to naive datetime for consistent comparison
658 if timestamp.tzinfo is not None:
659 timestamp = timestamp.replace(tzinfo=None)
660 except (ValueError, AttributeError):
661 # Use file modification time as fallback
662 timestamp = datetime.fromtimestamp(jsonl_file.stat().st_mtime)
664 # Apply since filter
665 if since and timestamp < since:
666 return None
668 # Create message object
669 return UserMessage(
670 content=message_content,
671 timestamp=timestamp,
672 session_id=record.get("sessionId", ""),
673 uuid=record.get("uuid", ""),
674 cwd=record.get("cwd"),
675 git_branch=record.get("gitBranch"),
676 is_sidechain=record.get("isSidechain", False),
677 )
680def _extract_messages_with_context(
681 records: list[dict],
682 jsonl_file: Path,
683 since: datetime | None,
684) -> list[UserMessage]:
685 """Extract user messages with response context from a list of records.
687 Pairs each user message with ALL following assistant responses until the
688 next user message, aggregating tool usage and file changes.
690 Args:
691 records: List of all records from a JSONL file
692 jsonl_file: Source file (for fallback timestamp)
693 since: Filter for messages after this datetime
695 Returns:
696 List of UserMessages with response_metadata populated
697 """
698 messages: list[UserMessage] = []
700 current_msg: UserMessage | None = None
701 current_responses: list[dict] = []
703 for record in records:
704 if record.get("type") == "user":
705 if current_msg is not None:
706 current_msg.response_metadata = _aggregate_response_metadata(current_responses)
707 messages.append(current_msg)
708 current_msg = _parse_user_record(record, jsonl_file, since)
709 current_responses = []
710 elif record.get("type") == "assistant" and current_msg is not None:
711 current_responses.append(record)
713 # Emit the final group
714 if current_msg is not None:
715 current_msg.response_metadata = _aggregate_response_metadata(current_responses)
716 messages.append(current_msg)
718 return messages
721def _extract_turn_pairs(
722 records: list[dict],
723 jsonl_file: Path,
724 since: datetime | None,
725) -> list[tuple[str, str]]:
726 """Extract (user_text, assistant_text) pairs from a record stream.
728 Collects all assistant text blocks between consecutive user messages and
729 pairs them with the preceding user message.
731 Args:
732 records: All records from a JSONL session file
733 jsonl_file: Source file (for _parse_user_record fallback timestamp)
734 since: Filter for turns after this datetime
736 Returns:
737 List of (user_text, assistant_text) pairs in chronological order
738 """
739 pairs: list[tuple[str, str]] = []
740 current_user: str | None = None
741 assistant_texts: list[str] = []
743 for record in records:
744 if record.get("type") == "user":
745 if current_user is not None and assistant_texts:
746 pairs.append((current_user, "\n\n".join(assistant_texts)))
747 msg = _parse_user_record(record, jsonl_file, since)
748 current_user = msg.content if msg is not None else None
749 assistant_texts = []
750 elif record.get("type") == "assistant" and current_user is not None:
751 content = record.get("message", {}).get("content", [])
752 if isinstance(content, list):
753 for block in content:
754 if isinstance(block, dict) and block.get("type") == "text":
755 text = block.get("text", "").strip()
756 if text:
757 assistant_texts.append(text)
759 if current_user is not None and assistant_texts:
760 pairs.append((current_user, "\n\n".join(assistant_texts)))
762 return pairs
765def _mtime(path: Path) -> float:
766 """Return file modification time as a Unix float, or 0.0 if inaccessible."""
767 try:
768 return path.stat().st_mtime
769 except OSError:
770 return 0.0
773def extract_conversation_turns(
774 project_folder: Path,
775 since: datetime | None = None,
776 context_window: int = 3,
777 include_agent_sessions: bool = True,
778) -> list[list[tuple[str, str]]]:
779 """Extract conversation turns (user + assistant) from session logs.
781 Reads JSONL session files and extracts alternating user/assistant turn pairs,
782 grouping them into sliding windows of context_window turn-pairs each.
784 Args:
785 project_folder: Path to Claude project folder
786 since: Only include turns from sessions containing messages after this datetime
787 context_window: Number of (user, assistant) turn pairs per output window
788 include_agent_sessions: Whether to include agent-*.jsonl files
790 Returns:
791 List of conversation windows; each window is a list of (role, content) tuples
792 alternating between "user" and "assistant".
793 """
794 windows: list[list[tuple[str, str]]] = []
796 jsonl_files = list(project_folder.glob("*.jsonl"))
797 if since is not None:
798 cutoff_ts = (since - timedelta(seconds=60)).timestamp()
799 jsonl_files = [f for f in jsonl_files if _mtime(f) >= cutoff_ts]
801 for jsonl_file in jsonl_files:
802 if not include_agent_sessions and jsonl_file.name.startswith("agent-"):
803 continue
805 try:
806 all_records: list[dict] = []
807 with open(jsonl_file, encoding="utf-8") as f:
808 for line in f:
809 line = line.strip()
810 if not line:
811 continue
812 try:
813 all_records.append(json.loads(line))
814 except json.JSONDecodeError:
815 continue
817 turn_pairs = _extract_turn_pairs(all_records, jsonl_file, since)
819 # Emit sliding windows of context_window turn-pairs
820 n = len(turn_pairs)
821 if n == 0:
822 continue
823 for i in range(max(1, n - context_window + 1)):
824 window_pairs = turn_pairs[i : i + context_window]
825 window: list[tuple[str, str]] = []
826 for user_text, assistant_text in window_pairs:
827 window.append(("user", user_text))
828 window.append(("assistant", assistant_text))
829 windows.append(window)
830 except OSError:
831 continue
833 return windows
836def save_messages(
837 messages: list[UserMessage],
838 output_path: Path | None = None,
839) -> Path:
840 """Save messages to timestamped JSONL file.
842 Args:
843 messages: List of UserMessage objects to save
844 output_path: Output file path. If None, uses default location.
846 Returns:
847 Path to the saved file.
848 """
849 if output_path is None:
850 # Default: ./.ll/user-messages-{timestamp}.jsonl
851 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
852 output_dir = Path.cwd() / ".ll"
853 output_dir.mkdir(parents=True, exist_ok=True)
854 output_path = output_dir / f"user-messages-{timestamp}.jsonl"
856 output_path = Path(output_path)
857 output_path.parent.mkdir(parents=True, exist_ok=True)
859 with open(output_path, "w", encoding="utf-8") as f:
860 for msg in messages:
861 f.write(json.dumps(msg.to_dict()) + "\n")
863 return output_path
866def build_examples(
867 messages: list[UserMessage],
868 skill: str,
869 context_window: int = 3,
870) -> list[ExampleRecord]:
871 """Build training example pairs from skill invocation sessions.
873 Groups messages by session, identifies skill trigger records (the user-side record
874 whose content contains ``<command-name>/ll:SKILL_NAME</command-name>``), and pairs
875 each trigger with the N preceding messages as input context.
877 Args:
878 messages: UserMessage list (already filtered to skill-matching sessions)
879 skill: The skill name to build examples for (e.g. "capture-issue")
880 context_window: Number of preceding messages to include as context (default 3)
882 Returns:
883 List of ExampleRecord objects, one per skill trigger record found.
884 """
885 import re
887 skill_pattern = re.compile(rf"<command-name>/ll:{re.escape(skill)}</command-name>")
889 # Group by session_id, sorted ascending by timestamp
890 sessions: dict[str, list[UserMessage]] = {}
891 for msg in messages:
892 sessions.setdefault(msg.session_id, []).append(msg)
893 for session_msgs in sessions.values():
894 session_msgs.sort(key=lambda m: m.timestamp)
896 examples: list[ExampleRecord] = []
897 for session_id, session_msgs in sessions.items():
898 for idx, msg in enumerate(session_msgs):
899 if not skill_pattern.search(msg.content):
900 continue
902 # Collect N preceding messages as context
903 preceding = session_msgs[max(0, idx - context_window) : idx]
904 input_text = "\n\n".join(m.content for m in preceding)
906 # Serialize response_metadata as output
907 if msg.response_metadata is not None:
908 output_str = json.dumps(msg.response_metadata.to_dict())
909 else:
910 output_str = "{}"
912 examples.append(
913 ExampleRecord(
914 skill=skill,
915 input=input_text,
916 output=output_str,
917 session_id=session_id,
918 timestamp=msg.timestamp,
919 context_window=context_window,
920 )
921 )
923 return examples
926def print_messages_to_stdout(messages: list[UserMessage]) -> None:
927 """Print messages to stdout in JSONL format.
929 Args:
930 messages: List of UserMessage objects to print
931 """
932 import sys
934 for msg in messages:
935 print(json.dumps(msg.to_dict()), file=sys.stdout)