Coverage for little_loops / user_messages.py: 15%

1"""Extract and analyze user messages from Claude Code logs. 

2 

3Provides functionality to extract user messages from Claude Code session 

4logs stored in ~/.claude/projects/. 

5 

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 

12 

13Usage as library: 

14 from little_loops.user_messages import extract_user_messages, get_project_folder 

15 

16 project_folder = get_project_folder() 

17 messages = extract_user_messages(project_folder, limit=50) 

18""" 

19 

20from __future__ import annotations 

21 

22import json 

23import os 

24from dataclasses import dataclass 

25from datetime import datetime, timedelta 

26from pathlib import Path 

27 

28__all__ = [ 

29 "UserMessage", 

30 "ResponseMetadata", 

31 "CommandRecord", 

32 "ExampleRecord", 

33 "get_project_folder", 

34 "extract_user_messages", 

35 "extract_commands", 

36 "build_examples", 

37 "extract_conversation_turns", 

38 "save_messages", 

39] 

40 

41 

42@dataclass 

43class UserMessage: 

44 """Extracted user message with metadata. 

45 

46 Attributes: 

47 content: The text content of the user message 

48 timestamp: When the message was sent 

49 session_id: Claude Code session identifier 

50 uuid: Unique message identifier 

51 cwd: Working directory when message was sent 

52 git_branch: Git branch active when message was sent 

53 is_sidechain: Whether this was a sidechain message 

54 """ 

55 

56 content: str 

57 timestamp: datetime 

58 session_id: str 

59 uuid: str 

60 cwd: str | None = None 

61 git_branch: str | None = None 

62 is_sidechain: bool = False 

63 

64 response_metadata: ResponseMetadata | None = None 

65 

66 def to_dict(self) -> dict[str, object]: 

67 """Convert to dictionary for JSON serialization.""" 

68 result: dict[str, object] = { 

69 "content": self.content, 

70 "timestamp": self.timestamp.isoformat(), 

71 "session_id": self.session_id, 

72 "uuid": self.uuid, 

73 "cwd": self.cwd, 

74 "git_branch": self.git_branch, 

75 "is_sidechain": self.is_sidechain, 

76 } 

77 if self.response_metadata is not None: 

78 result["response_metadata"] = self.response_metadata.to_dict() 

79 return result 

80 

81 

82@dataclass 

83class ResponseMetadata: 

84 """Metadata extracted from assistant response. 

85 

86 Attributes: 

87 tools_used: List of tools and their usage counts 

88 files_read: Files accessed via Read tool 

89 files_modified: Files changed via Edit/Write tools 

90 completion_status: "success", "failure", or "partial" 

91 error_message: Error text if failure detected 

92 """ 

93 

94 tools_used: list[dict[str, str | int]] 

95 files_read: list[str] 

96 files_modified: list[str] 

97 completion_status: str 

98 error_message: str | None = None 

99 

100 def to_dict(self) -> dict[str, object]: 

101 """Convert to dictionary for JSON serialization.""" 

102 return { 

103 "tools_used": self.tools_used, 

104 "files_read": self.files_read, 

105 "files_modified": self.files_modified, 

106 "completion_status": self.completion_status, 

107 "error_message": self.error_message, 

108 } 

109 

110 

111@dataclass 

112class CommandRecord: 

113 """Extracted CLI command from assistant tool_use. 

114 

115 Attributes: 

116 content: The command string that was executed 

117 timestamp: When the command was issued 

118 session_id: Claude Code session identifier 

119 uuid: Unique record identifier 

120 tool: Tool name (e.g., "Bash") 

121 cwd: Working directory when command was issued 

122 git_branch: Git branch active when command was issued 

123 """ 

124 

125 content: str 

126 timestamp: datetime 

127 session_id: str 

128 uuid: str 

129 tool: str 

130 cwd: str | None = None 

131 git_branch: str | None = None 

132 

133 def to_dict(self) -> dict[str, object]: 

134 """Convert to dictionary for JSON serialization.""" 

135 return { 

136 "type": "command", 

137 "content": self.content, 

138 "timestamp": self.timestamp.isoformat(), 

139 "session_id": self.session_id, 

140 "uuid": self.uuid, 

141 "tool": self.tool, 

142 "cwd": self.cwd, 

143 "git_branch": self.git_branch, 

144 } 

145 

146 

147@dataclass 

148class ExampleRecord: 

149 """Training example pair extracted from a skill invocation session. 

150 

151 Attributes: 

152 skill: The skill name (e.g., "capture-issue") 

153 input: Concatenated preceding user messages as context 

154 output: JSON-serialized ResponseMetadata summary (tools_used, files_modified, 

155 completion_status); free-text assistant response capture is deferred 

156 session_id: Claude Code session identifier 

157 timestamp: When the skill was invoked 

158 context_window: Number of preceding messages used as context 

159 """ 

160 

161 skill: str 

162 input: str 

163 output: str 

164 session_id: str 

165 timestamp: datetime 

166 context_window: int 

167 

168 def to_dict(self) -> dict[str, object]: 

169 """Convert to dictionary for JSON serialization.""" 

170 return { 

171 "type": "example", 

172 "skill": self.skill, 

173 "input": self.input, 

174 "output": self.output, 

175 "session_id": self.session_id, 

176 "timestamp": self.timestamp.isoformat(), 

177 "context_window": self.context_window, 

178 } 

179 

180 

181def _extract_response_metadata(response_record: dict) -> ResponseMetadata | None: 

182 """Extract metadata from an assistant response record. 

183 

184 Args: 

185 response_record: The assistant record from JSONL 

186 

187 Returns: 

188 ResponseMetadata if parseable, None otherwise 

189 """ 

190 message_data = response_record.get("message", {}) 

191 content = message_data.get("content", []) 

192 

193 if not isinstance(content, list): 

194 return None 

195 

196 tools_used: dict[str, int] = {} 

197 files_read: list[str] = [] 

198 files_modified: list[str] = [] 

199 

200 for block in content: 

201 if not isinstance(block, dict): 

202 continue 

203 if block.get("type") != "tool_use": 

204 continue 

205 

206 tool_name = block.get("name", "") 

207 tools_used[tool_name] = tools_used.get(tool_name, 0) + 1 

208 

209 tool_input = block.get("input", {}) 

210 if tool_name == "Read": 

211 file_path = tool_input.get("file_path") 

212 if file_path: 

213 files_read.append(file_path) 

214 elif tool_name in ("Edit", "Write"): 

215 file_path = tool_input.get("file_path") 

216 if file_path: 

217 files_modified.append(file_path) 

218 

219 # Detect completion status from text content 

220 completion_status = _detect_completion_status(content) 

221 error_message = _detect_error_message(content) if completion_status == "failure" else None 

222 

223 # Convert tools_used dict to list format 

224 tools_list: list[dict[str, str | int]] = [ 

225 {"tool": name, "count": count} for name, count in tools_used.items() 

226 ] 

227 

228 return ResponseMetadata( 

229 tools_used=tools_list, 

230 files_read=files_read, 

231 files_modified=files_modified, 

232 completion_status=completion_status, 

233 error_message=error_message, 

234 ) 

235 

236 

237def _aggregate_response_metadata(responses: list[dict]) -> ResponseMetadata | None: 

238 """Aggregate metadata from multiple assistant response records. 

239 

240 Combines tool counts, file lists, and uses completion status from final response. 

241 

242 Args: 

243 responses: List of assistant records from JSONL 

244 

245 Returns: 

246 Aggregated ResponseMetadata, or None if no valid responses 

247 """ 

248 if not responses: 

249 return None 

250 

251 tools_used: dict[str, int] = {} 

252 files_read: set[str] = set() 

253 files_modified: set[str] = set() 

254 completion_status = "success" 

255 error_message: str | None = None 

256 

257 for response_record in responses: 

258 message_data = response_record.get("message", {}) 

259 content = message_data.get("content", []) 

260 

261 if not isinstance(content, list): 

262 continue 

263 

264 for block in content: 

265 if not isinstance(block, dict): 

266 continue 

267 if block.get("type") != "tool_use": 

268 continue 

269 

270 tool_name = block.get("name", "") 

271 tools_used[tool_name] = tools_used.get(tool_name, 0) + 1 

272 

273 tool_input = block.get("input", {}) 

274 if tool_name == "Read": 

275 file_path = tool_input.get("file_path") 

276 if file_path: 

277 files_read.add(file_path) 

278 elif tool_name in ("Edit", "Write"): 

279 file_path = tool_input.get("file_path") 

280 if file_path: 

281 files_modified.add(file_path) 

282 

283 # Use completion status from the final response 

284 final_content = responses[-1].get("message", {}).get("content", []) 

285 if isinstance(final_content, list): 

286 completion_status = _detect_completion_status(final_content) 

287 if completion_status == "failure": 

288 error_message = _detect_error_message(final_content) 

289 

290 # Convert to output format 

291 tools_list: list[dict[str, str | int]] = [ 

292 {"tool": name, "count": count} for name, count in tools_used.items() 

293 ] 

294 

295 return ResponseMetadata( 

296 tools_used=tools_list, 

297 files_read=sorted(files_read), 

298 files_modified=sorted(files_modified), 

299 completion_status=completion_status, 

300 error_message=error_message, 

301 ) 

302 

303 

304def _detect_completion_status(content: list) -> str: 

305 """Detect completion status from response content. 

306 

307 Args: 

308 content: List of content blocks from assistant response 

309 

310 Returns: 

311 "success", "failure", or "partial" 

312 """ 

313 text_parts = [] 

314 for block in content: 

315 if isinstance(block, dict) and block.get("type") == "text": 

316 text_parts.append(block.get("text", "")) 

317 

318 text = " ".join(text_parts).lower() 

319 

320 # Check for error indicators 

321 error_patterns = ["error", "failed", "couldn't", "unable to", "cannot"] 

322 if any(pattern in text for pattern in error_patterns): 

323 return "failure" 

324 

325 # Check for partial completion 

326 partial_patterns = ["partially", "some of", "not all", "incomplete"] 

327 if any(pattern in text for pattern in partial_patterns): 

328 return "partial" 

329 

330 return "success" 

331 

332 

333def _detect_error_message(content: list) -> str | None: 

334 """Extract error message from response content. 

335 

336 Args: 

337 content: List of content blocks from assistant response 

338 

339 Returns: 

340 Error message if found, None otherwise 

341 """ 

342 for block in content: 

343 if isinstance(block, dict) and block.get("type") == "text": 

344 text = block.get("text", "") 

345 # Look for common error message patterns 

346 lower_text = text.lower() 

347 if "error:" in lower_text or "failed:" in lower_text: 

348 # Extract the line containing the error 

349 for line in text.split("\n"): 

350 if "error" in line.lower() or "failed" in line.lower(): 

351 result = line.strip()[:200] # Limit length 

352 return result if isinstance(result, str) else None 

353 return None 

354 

355 

356def get_project_folder(cwd: Path | None = None, *, host: str | None = None) -> Path | None: 

357 """Map current directory to the host's session-log project folder. 

358 

359 Converts the working directory into the host-specific encoded path and 

360 probes the host's session directory for matching JSONL files. 

361 

362 Args: 

363 cwd: Working directory to map. If None, uses current directory. 

364 host: Host identifier (``"claude-code"``, ``"codex"``, ``"opencode"``, 

365 ``"pi"``). If None, auto-detects from ``LL_HOOK_HOST`` env var 

366 (default ``"claude-code"``). 

367 

368 Returns: 

369 Path to the host's project session folder, or None if not found. 

370 

371 Future hosts (e.g. FEAT-992 Pi) add a new branch here rather than a 

372 new code path elsewhere. 

373 """ 

374 if cwd is None: 

375 cwd = Path.cwd() 

376 if host is None: 

377 host = os.environ.get("LL_HOOK_HOST", "claude-code") 

378 

379 # Convert path to dash-separated format 

380 # /home/user/foo/bar -> -home-user-foo-bar 

381 path_str = str(cwd.resolve()) 

382 encoded_path = path_str.replace("/", "-") 

383 

384 if host == "claude-code": 

385 return _get_claude_project_folder(encoded_path) 

386 elif host == "codex": 

387 return _get_codex_project_folder(encoded_path) 

388 elif host == "opencode": 

389 return _get_opencode_project_folder(encoded_path) 

390 elif host == "pi": 

391 return _get_pi_project_folder(encoded_path) 

392 return None 

393 

394 

395def _get_claude_project_folder(encoded_path: str) -> Path | None: 

396 """Probe the Claude Code session directory.""" 

397 project_folder = Path.home() / ".claude" / "projects" / encoded_path 

398 return project_folder if project_folder.exists() else None 

399 

400 

401def _get_codex_project_folder(encoded_path: str) -> Path | None: 

402 """Probe the Codex session directory.""" 

403 project_folder = Path.home() / ".codex" / "projects" / encoded_path 

404 return project_folder if project_folder.exists() else None 

405 

406 

407def _get_opencode_project_folder(encoded_path: str) -> Path | None: 

408 """Probe the OpenCode session directory.""" 

409 project_folder = Path.home() / ".opencode" / "projects" / encoded_path 

410 return project_folder if project_folder.exists() else None 

411 

412 

413def _get_pi_project_folder(encoded_path: str) -> Path | None: 

414 """Probe the Pi session directory (stub; Pi adapter deferred per FEAT-992).""" 

415 project_folder = Path.home() / ".pi" / "projects" / encoded_path 

416 return project_folder if project_folder.exists() else None 

417 

418 

419def extract_user_messages( 

420 project_folder: Path, 

421 limit: int | None = None, 

422 since: datetime | None = None, 

423 include_agent_sessions: bool = True, 

424 include_response_context: bool = False, 

425) -> list[UserMessage]: 

426 """Extract user messages from all JSONL session files. 

427 

428 Filters: 

429 - type == "user" 

430 - message.content is string (real user input) 

431 - message.content is array but [0].type != "tool_result" 

432 

433 Args: 

434 project_folder: Path to Claude project folder 

435 limit: Maximum number of messages to return 

436 since: Only include messages after this datetime 

437 include_agent_sessions: Whether to include agent-*.jsonl files 

438 include_response_context: Whether to include metadata from assistant responses 

439 

440 Returns: 

441 Messages sorted by timestamp, most recent first. 

442 """ 

443 messages: list[UserMessage] = [] 

444 

445 # Find all JSONL files 

446 pattern = "*.jsonl" 

447 jsonl_files = list(project_folder.glob(pattern)) 

448 

449 for jsonl_file in jsonl_files: 

450 # Skip agent sessions if requested 

451 if not include_agent_sessions and jsonl_file.name.startswith("agent-"): 

452 continue 

453 

454 try: 

455 # If we need response context, read all records first to pair user/assistant 

456 if include_response_context: 

457 all_records: list[dict] = [] 

458 with open(jsonl_file, encoding="utf-8") as f: 

459 for line in f: 

460 line = line.strip() 

461 if not line: 

462 continue 

463 try: 

464 record = json.loads(line) 

465 all_records.append(record) 

466 except json.JSONDecodeError: 

467 continue 

468 

469 # Process records, pairing user messages with their responses 

470 messages.extend(_extract_messages_with_context(all_records, jsonl_file, since)) 

471 else: 

472 # Original behavior: stream through file 

473 with open(jsonl_file, encoding="utf-8") as f: 

474 for line in f: 

475 line = line.strip() 

476 if not line: 

477 continue 

478 

479 try: 

480 record = json.loads(line) 

481 except json.JSONDecodeError: 

482 continue 

483 

484 msg = _parse_user_record(record, jsonl_file, since) 

485 if msg is not None: 

486 messages.append(msg) 

487 

488 except OSError: 

489 # Skip files that can't be read 

490 continue 

491 

492 # Sort by timestamp, most recent first 

493 messages.sort(key=lambda m: m.timestamp, reverse=True) 

494 

495 # Apply limit 

496 if limit is not None: 

497 messages = messages[:limit] 

498 

499 return messages 

500 

501 

502def extract_commands( 

503 project_folder: Path, 

504 limit: int | None = None, 

505 since: datetime | None = None, 

506 include_agent_sessions: bool = True, 

507 tools: list[str] | None = None, 

508) -> list[CommandRecord]: 

509 """Extract CLI commands from assistant tool_use messages. 

510 

511 Parses assistant messages for tool_use blocks and extracts command strings. 

512 

513 Args: 

514 project_folder: Path to Claude project folder 

515 limit: Maximum number of commands to return 

516 since: Only include commands after this datetime 

517 include_agent_sessions: Whether to include agent-*.jsonl files 

518 tools: Filter to specific tools (default: ["Bash"]) 

519 

520 Returns: 

521 Commands sorted by timestamp, most recent first. 

522 """ 

523 if tools is None: 

524 tools = ["Bash"] 

525 

526 commands: list[CommandRecord] = [] 

527 

528 # Find all JSONL files 

529 pattern = "*.jsonl" 

530 jsonl_files = list(project_folder.glob(pattern)) 

531 

532 for jsonl_file in jsonl_files: 

533 # Skip agent sessions if requested 

534 if not include_agent_sessions and jsonl_file.name.startswith("agent-"): 

535 continue 

536 

537 try: 

538 with open(jsonl_file, encoding="utf-8") as f: 

539 for line in f: 

540 line = line.strip() 

541 if not line: 

542 continue 

543 

544 try: 

545 record = json.loads(line) 

546 except json.JSONDecodeError: 

547 continue 

548 

549 cmds = _parse_command_record(record, jsonl_file, since, tools) 

550 commands.extend(cmds) 

551 

552 except OSError: 

553 # Skip files that can't be read 

554 continue 

555 

556 # Sort by timestamp, most recent first 

557 commands.sort(key=lambda c: c.timestamp, reverse=True) 

558 

559 # Apply limit 

560 if limit is not None: 

561 commands = commands[:limit] 

562 

563 return commands 

564 

565 

566def _parse_command_record( 

567 record: dict, 

568 jsonl_file: Path, 

569 since: datetime | None, 

570 tools: list[str], 

571) -> list[CommandRecord]: 

572 """Parse CLI commands from an assistant record. 

573 

574 Args: 

575 record: The JSON record from JSONL 

576 jsonl_file: Source file (for fallback timestamp) 

577 since: Filter for commands after this datetime 

578 tools: Tool names to extract (e.g., ["Bash"]) 

579 

580 Returns: 

581 List of CommandRecord for each matching tool_use block 

582 """ 

583 # Filter for assistant messages only 

584 if record.get("type") != "assistant": 

585 return [] 

586 

587 message_data = record.get("message", {}) 

588 content = message_data.get("content", []) 

589 

590 if not isinstance(content, list): 

591 return [] 

592 

593 # Parse timestamp 

594 timestamp_str = record.get("timestamp", "") 

595 try: 

596 timestamp_str = timestamp_str.replace("Z", "+00:00") 

597 timestamp = datetime.fromisoformat(timestamp_str) 

598 if timestamp.tzinfo is not None: 

599 timestamp = timestamp.replace(tzinfo=None) 

600 except (ValueError, AttributeError): 

601 timestamp = datetime.fromtimestamp(jsonl_file.stat().st_mtime) 

602 

603 # Apply since filter 

604 if since and timestamp < since: 

605 return [] 

606 

607 commands: list[CommandRecord] = [] 

608 

609 for block in content: 

610 if not isinstance(block, dict): 

611 continue 

612 if block.get("type") != "tool_use": 

613 continue 

614 

615 tool_name = block.get("name", "") 

616 if tool_name not in tools: 

617 continue 

618 

619 tool_input = block.get("input", {}) 

620 command_str = tool_input.get("command", "") 

621 if not command_str: 

622 continue 

623 

624 commands.append( 

625 CommandRecord( 

626 content=command_str, 

627 timestamp=timestamp, 

628 session_id=record.get("sessionId", ""), 

629 uuid=record.get("uuid", ""), 

630 tool=tool_name, 

631 cwd=record.get("cwd"), 

632 git_branch=record.get("gitBranch"), 

633 ) 

634 ) 

635 

636 return commands 

637 

638 

639def _parse_user_record( 

640 record: dict, 

641 jsonl_file: Path, 

642 since: datetime | None, 

643) -> UserMessage | None: 

644 """Parse a single user record into a UserMessage. 

645 

646 Args: 

647 record: The JSON record from JSONL 

648 jsonl_file: Source file (for fallback timestamp) 

649 since: Filter for messages after this datetime 

650 

651 Returns: 

652 UserMessage if valid user message, None otherwise 

653 """ 

654 # Filter for user messages only 

655 if record.get("type") != "user": 

656 return None 

657 

658 message_data = record.get("message", {}) 

659 content = message_data.get("content") 

660 

661 # Skip if no content 

662 if content is None: 

663 return None 

664 

665 # Check if this is a real user message or tool_result 

666 if isinstance(content, str): 

667 # String content = real user message 

668 message_content = content 

669 elif isinstance(content, list): 

670 # Array content - check first element 

671 if len(content) > 0 and content[0].get("type") == "tool_result": 

672 # This is a tool result, skip it 

673 return None 

674 # Extract text from array (could be text blocks) 

675 text_parts = [] 

676 for block in content: 

677 if isinstance(block, dict): 

678 if block.get("type") == "text": 

679 text_parts.append(block.get("text", "")) 

680 elif "content" in block: 

681 text_parts.append(str(block.get("content", ""))) 

682 message_content = "\n".join(text_parts) if text_parts else str(content) 

683 else: 

684 return None 

685 

686 # Parse timestamp 

687 timestamp_str = record.get("timestamp", "") 

688 try: 

689 # Handle ISO 8601 format with Z suffix 

690 timestamp_str = timestamp_str.replace("Z", "+00:00") 

691 timestamp = datetime.fromisoformat(timestamp_str) 

692 # Convert to naive datetime for consistent comparison 

693 if timestamp.tzinfo is not None: 

694 timestamp = timestamp.replace(tzinfo=None) 

695 except (ValueError, AttributeError): 

696 # Use file modification time as fallback 

697 timestamp = datetime.fromtimestamp(jsonl_file.stat().st_mtime) 

698 

699 # Apply since filter 

700 if since and timestamp < since: 

701 return None 

702 

703 # Create message object 

704 return UserMessage( 

705 content=message_content, 

706 timestamp=timestamp, 

707 session_id=record.get("sessionId", ""), 

708 uuid=record.get("uuid", ""), 

709 cwd=record.get("cwd"), 

710 git_branch=record.get("gitBranch"), 

711 is_sidechain=record.get("isSidechain", False), 

712 ) 

713 

714 

715def _extract_messages_with_context( 

716 records: list[dict], 

717 jsonl_file: Path, 

718 since: datetime | None, 

719) -> list[UserMessage]: 

720 """Extract user messages with response context from a list of records. 

721 

722 Pairs each user message with ALL following assistant responses until the 

723 next user message, aggregating tool usage and file changes. 

724 

725 Args: 

726 records: List of all records from a JSONL file 

727 jsonl_file: Source file (for fallback timestamp) 

728 since: Filter for messages after this datetime 

729 

730 Returns: 

731 List of UserMessages with response_metadata populated 

732 """ 

733 messages: list[UserMessage] = [] 

734 

735 current_msg: UserMessage | None = None 

736 current_responses: list[dict] = [] 

737 

738 for record in records: 

739 if record.get("type") == "user": 

740 if current_msg is not None: 

741 current_msg.response_metadata = _aggregate_response_metadata(current_responses) 

742 messages.append(current_msg) 

743 current_msg = _parse_user_record(record, jsonl_file, since) 

744 current_responses = [] 

745 elif record.get("type") == "assistant" and current_msg is not None: 

746 current_responses.append(record) 

747 

748 # Emit the final group 

749 if current_msg is not None: 

750 current_msg.response_metadata = _aggregate_response_metadata(current_responses) 

751 messages.append(current_msg) 

752 

753 return messages 

754 

755 

756def _extract_turn_pairs( 

757 records: list[dict], 

758 jsonl_file: Path, 

759 since: datetime | None, 

760) -> list[tuple[str, str]]: 

761 """Extract (user_text, assistant_text) pairs from a record stream. 

762 

763 Collects all assistant text blocks between consecutive user messages and 

764 pairs them with the preceding user message. 

765 

766 Args: 

767 records: All records from a JSONL session file 

768 jsonl_file: Source file (for _parse_user_record fallback timestamp) 

769 since: Filter for turns after this datetime 

770 

771 Returns: 

772 List of (user_text, assistant_text) pairs in chronological order 

773 """ 

774 pairs: list[tuple[str, str]] = [] 

775 current_user: str | None = None 

776 assistant_texts: list[str] = [] 

777 

778 for record in records: 

779 if record.get("type") == "user": 

780 if current_user is not None and assistant_texts: 

781 pairs.append((current_user, "\n\n".join(assistant_texts))) 

782 msg = _parse_user_record(record, jsonl_file, since) 

783 current_user = msg.content if msg is not None else None 

784 assistant_texts = [] 

785 elif record.get("type") == "assistant" and current_user is not None: 

786 content = record.get("message", {}).get("content", []) 

787 if isinstance(content, list): 

788 for block in content: 

789 if isinstance(block, dict) and block.get("type") == "text": 

790 text = block.get("text", "").strip() 

791 if text: 

792 assistant_texts.append(text) 

793 

794 if current_user is not None and assistant_texts: 

795 pairs.append((current_user, "\n\n".join(assistant_texts))) 

796 

797 return pairs 

798 

799 

800def _mtime(path: Path) -> float: 

801 """Return file modification time as a Unix float, or 0.0 if inaccessible.""" 

802 try: 

803 return path.stat().st_mtime 

804 except OSError: 

805 return 0.0 

806 

807 

808def extract_conversation_turns( 

809 project_folder: Path, 

810 since: datetime | None = None, 

811 context_window: int = 3, 

812 include_agent_sessions: bool = True, 

813 reader: str = "auto", 

814) -> list[list[tuple[str, str]]]: 

815 """Extract conversation turns (user + assistant) from session logs. 

816 

817 By default (``reader="auto"``), tries ``history.db`` first and falls back to 

818 JSONL parsing when the database is missing, empty, or predates schema v11. 

819 Use ``reader="db"`` to require DB-only (errors if unavailable) or 

820 ``reader="jsonl"`` to skip the DB and use the existing JSONL path directly. 

821 

822 Args: 

823 project_folder: Path to Claude project folder 

824 since: Only include turns from sessions containing messages after this datetime 

825 context_window: Number of (user, assistant) turn pairs per output window 

826 include_agent_sessions: Whether to include agent-*.jsonl files 

827 reader: ``"auto"`` (DB-first, JSONL fallback), ``"db"`` (DB only), 

828 or ``"jsonl"`` (JSONL only, current behavior) 

829 

830 Returns: 

831 List of conversation windows; each window is a list of (role, content) tuples 

832 alternating between "user" and "assistant". 

833 """ 

834 from little_loops.history_reader import conversation_turns as db_conversation_turns 

835 from little_loops.session_store import resolve_history_db 

836 

837 reader = reader.lower() 

838 

839 # -- DB path: try history_reader first, optionally fall back to JSONL -- 

840 if reader in ("auto", "db"): 

841 db_path = resolve_history_db(project_folder / ".ll" / "history.db") 

842 db_windows = db_conversation_turns( 

843 db_path=db_path, 

844 since=since, 

845 context_window=context_window, 

846 ) 

847 if db_windows: 

848 if not include_agent_sessions: 

849 # DB results can't currently filter agent sessions; 

850 # fall through to JSONL when filtering is needed. 

851 pass 

852 else: 

853 return db_windows 

854 

855 if reader == "db": 

856 # DB-only mode: error when DB returns nothing 

857 raise RuntimeError( 

858 "history.db missing or predates v11; " 

859 "run `ll-session backfill` to migrate. " 

860 "Use --reader jsonl for JSONL-only mode." 

861 ) 

862 

863 # auto mode: fall through to JSONL parsing below 

864 if db_windows: 

865 import logging 

866 

867 logging.getLogger(__name__).warning( 

868 "history.db returned no results; falling back to JSONL parsing. " 

869 "Run `ll-session backfill` to populate the database." 

870 ) 

871 

872 # -- JSONL path (fallback for "auto", direct for "jsonl") -- 

873 windows: list[list[tuple[str, str]]] = [] 

874 

875 jsonl_files = list(project_folder.glob("*.jsonl")) 

876 if since is not None: 

877 cutoff_ts = (since - timedelta(seconds=60)).timestamp() 

878 jsonl_files = [f for f in jsonl_files if _mtime(f) >= cutoff_ts] 

879 

880 for jsonl_file in jsonl_files: 

881 if not include_agent_sessions and jsonl_file.name.startswith("agent-"): 

882 continue 

883 

884 try: 

885 all_records: list[dict] = [] 

886 with open(jsonl_file, encoding="utf-8") as f: 

887 for line in f: 

888 line = line.strip() 

889 if not line: 

890 continue 

891 try: 

892 all_records.append(json.loads(line)) 

893 except json.JSONDecodeError: 

894 continue 

895