Coverage for little_loops / cli / logs.py: 10%
261 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -0500
1"""ll-logs: Discover and extract ll-relevant JSONL entries from ~/.claude/projects/."""
3from __future__ import annotations
5import argparse
6import json
7import re
8import shutil
9import sys
10import time
11from pathlib import Path
13from little_loops.cli.loop.info import ( # private symbol: cross-module coupling; verify signature on upgrade
14 _format_history_event,
15)
16from little_loops.cli.output import configure_output, print_json, use_color_enabled
17from little_loops.cli_args import add_json_arg
18from little_loops.config import BRConfig
19from little_loops.logger import Logger
20from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
21from little_loops.user_messages import get_project_folder
23_COMMAND_NAME_RE = re.compile(r"<command-name>/ll:")
26def _is_ll_relevant(record: dict) -> bool:
27 """Return True if a JSONL record indicates ll activity.
29 Detects three signal types:
30 (a) queue-operation enqueue with /ll: content
31 (b) user records with <command-name>/ll: pattern in message content
32 """
33 record_type = record.get("type")
35 # (a) queue-operation: only enqueue records with /ll: content signal ll activity
36 if record_type == "queue-operation":
37 return (
38 record.get("operation") == "enqueue"
39 and isinstance(record.get("content"), str)
40 and record["content"].startswith("/ll:")
41 )
43 # (b) user records: check message content for <command-name>/ll: pattern
44 if record_type == "user":
45 message = record.get("message", {})
46 if not isinstance(message, dict):
47 return False
48 content = message.get("content")
49 if isinstance(content, str):
50 return bool(_COMMAND_NAME_RE.search(content))
51 if isinstance(content, list):
52 for block in content:
53 if isinstance(block, dict):
54 text = block.get("text", "")
55 if isinstance(text, str) and _COMMAND_NAME_RE.search(text):
56 return True
58 # (c) assistant records: check for Bash tool-use invoking an ll- command
59 if record_type == "assistant":
60 message = record.get("message", {})
61 content = message.get("content", [])
62 if isinstance(content, list):
63 for block in content:
64 if (
65 isinstance(block, dict)
66 and block.get("type") == "tool_use"
67 and block.get("name") == "Bash"
68 ):
69 cmd = block.get("input", {}).get("command", "")
70 if re.search(r"\bll-\w+", cmd):
71 return True
73 return False
76def _has_ll_activity(project_folder: Path) -> bool:
77 """Return True if any non-agent JSONL file in project_folder has ll activity."""
78 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
80 for jsonl_file in jsonl_files:
81 try:
82 with open(jsonl_file, encoding="utf-8") as f:
83 for line in f:
84 line = line.strip()
85 if not line:
86 continue
87 try:
88 record = json.loads(line)
89 except json.JSONDecodeError:
90 continue
91 if _is_ll_relevant(record):
92 return True
93 except OSError:
94 continue
96 return False
99def _extract_cwd_from_project(project_dir: Path) -> Path | None:
100 """Extract the project working directory from cwd fields in JSONL records.
102 Claude Code encodes project paths by replacing '/' with '-', which is
103 lossy for paths containing hyphens. Reading the cwd field from JSONL
104 records gives the canonical path without ambiguity.
105 """
106 jsonl_files = [f for f in project_dir.glob("*.jsonl") if not f.name.startswith("agent-")]
107 for jsonl_file in jsonl_files:
108 try:
109 with open(jsonl_file, encoding="utf-8") as f:
110 for line in f:
111 line = line.strip()
112 if not line:
113 continue
114 try:
115 record = json.loads(line)
116 cwd = record.get("cwd")
117 if isinstance(cwd, str) and cwd:
118 return Path(cwd)
119 except json.JSONDecodeError:
120 continue
121 except OSError:
122 continue
123 return None
126def discover_all_projects(logger: Logger) -> list[Path]:
127 """Discover all Claude projects with ll activity.
129 Iterates ~/.claude/projects/, resolves each directory name back to an
130 absolute path (preferring the cwd field from JSONL records over the
131 lossy '-'-based decode), checks for ll-relevant JSONL records, and
132 returns a sorted list of paths that exist on disk.
134 Args:
135 logger: Logger instance for warnings.
137 Returns:
138 Sorted list of decoded absolute paths for projects with ll activity.
139 """
140 claude_projects = Path.home() / ".claude" / "projects"
142 if not claude_projects.exists():
143 return []
145 results: list[Path] = []
147 for project_dir in claude_projects.iterdir():
148 if not project_dir.is_dir():
149 continue
151 # Prefer cwd field from JSONL records; fall back to lossy decode.
152 # The lossy decode ("-Users-foo-bar" -> "/Users/foo/bar") breaks for
153 # paths that contain hyphens (e.g. "little-loops", macOS per-user
154 # temp dirs like /tmp/claude-501/).
155 decoded_path = _extract_cwd_from_project(project_dir) or Path(
156 project_dir.name.replace("-", "/")
157 )
159 if not decoded_path.exists():
160 logger.warning(f"Decoded path does not exist: {decoded_path}")
161 continue
163 if _has_ll_activity(project_dir):
164 results.append(decoded_path)
166 return sorted(results)
169def _cmd_matches(record: dict, cmd: str) -> bool:
170 """Return True if record contains a Bash tool-use whose command includes cmd."""
171 message = record.get("message", {})
172 content = message.get("content", [])
173 if isinstance(content, list):
174 for block in content:
175 if (
176 isinstance(block, dict)
177 and block.get("type") == "tool_use"
178 and block.get("name") == "Bash"
179 ):
180 command = block.get("input", {}).get("command", "")
181 if cmd in command:
182 return True
183 return False
186def generate_index(logs_dir: Path) -> None:
187 """Generate logs/index.md summarising extracted projects."""
188 rows = []
190 if logs_dir.exists():
191 for subdir in sorted(logs_dir.iterdir()):
192 if not subdir.is_dir():
193 continue
195 jsonl_files = [f for f in subdir.glob("*.jsonl") if not f.name.startswith("agent-")]
196 if not jsonl_files:
197 continue
199 timestamps: list[str] = []
200 for jsonl_file in jsonl_files:
201 try:
202 with open(jsonl_file, encoding="utf-8") as f:
203 for line in f:
204 line = line.strip()
205 if not line:
206 continue
207 try:
208 record = json.loads(line)
209 except json.JSONDecodeError:
210 continue
211 ts = record.get("timestamp")
212 if ts:
213 timestamps.append(ts)
214 except OSError:
215 continue
217 if timestamps:
218 earliest = min(timestamps)[:10]
219 latest = max(timestamps)[:10]
220 date_range = f"{earliest} – {latest}" if earliest != latest else earliest
221 else:
222 date_range = ""
224 rows.append((subdir.name, len(jsonl_files), date_range))
226 lines = ["# Logs Index", ""]
227 if rows:
228 lines.append("| Project | Sessions | Date Range |")
229 lines.append("|---------|----------|------------|")
230 for name, count, date_range in rows:
231 lines.append(f"| {name} | {count} | {date_range} |")
232 else:
233 lines.append("*No projects extracted yet.*")
234 lines.append("")
236 logs_dir.mkdir(parents=True, exist_ok=True)
237 (logs_dir / "index.md").write_text("\n".join(lines), encoding="utf-8")
240def _cmd_extract(args: argparse.Namespace, logger: Logger) -> int:
241 """Extract ll-relevant JSONL records to logs/<slug>/<session-id>.jsonl."""
242 if args.project:
243 cwd_path: Path = args.project
244 project_folder = get_project_folder(cwd_path)
245 if project_folder is None:
246 logger.error(f"No Claude project folder found for: {cwd_path}")
247 logger.error(f"Expected: ~/.claude/projects/{str(cwd_path).replace('/', '-')}")
248 return 1
249 project_items = [(cwd_path, project_folder)]
250 else:
251 decoded_paths = discover_all_projects(logger)
252 project_items = []
253 for decoded_path in decoded_paths:
254 folder = get_project_folder(decoded_path)
255 if folder is not None:
256 project_items.append((decoded_path, folder))
258 for cwd_path, project_folder in project_items:
259 slug = cwd_path.resolve().name
260 buckets: dict[str, list[dict]] = {}
262 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
263 for jsonl_file in jsonl_files:
264 try:
265 with open(jsonl_file, encoding="utf-8") as f:
266 for line in f:
267 line = line.strip()
268 if not line:
269 continue
270 try:
271 record = json.loads(line)
272 except json.JSONDecodeError:
273 continue
274 if _is_ll_relevant(record):
275 session_id = record.get("sessionId", "")
276 buckets.setdefault(session_id, []).append(record)
277 except OSError:
278 continue
280 if args.cmd:
281 filtered: dict[str, list[dict]] = {}
282 for session_id, records in buckets.items():
283 matching = [r for r in records if _cmd_matches(r, args.cmd)]
284 if matching:
285 filtered[session_id] = matching
286 buckets = filtered
288 out_base = Path.cwd() / "logs" / slug
289 for session_id, records in buckets.items():
290 out_file = out_base / f"{session_id}.jsonl"
291 out_file.parent.mkdir(parents=True, exist_ok=True)
292 with open(out_file, "w", encoding="utf-8") as f:
293 for record in records:
294 f.write(json.dumps(record) + "\n")
296 generate_index(Path.cwd() / "logs")
297 return 0
300def _cmd_tail(args: argparse.Namespace, loops_dir: Path) -> int:
301 """Stream live events from an active loop session."""
302 events_file = loops_dir / ".running" / f"{args.loop}.events.jsonl"
304 if not events_file.exists():
305 print(f"No active session for loop '{args.loop}'", file=sys.stderr)
306 return 1
308 width = shutil.get_terminal_size().columns
309 try:
310 with open(events_file, encoding="utf-8") as f:
311 f.seek(0, 2)
312 while True:
313 line = f.readline()
314 if line:
315 line = line.strip()
316 if line:
317 try:
318 event = json.loads(line)
319 except json.JSONDecodeError:
320 continue
321 formatted = _format_history_event(event, verbose=False, width=width)
322 if formatted is not None:
323 print(formatted)
324 else:
325 time.sleep(0.1)
326 except KeyboardInterrupt:
327 return 0
329 return 0
332def _build_parser() -> argparse.ArgumentParser:
333 """Build the argument parser for ll-logs."""
334 parser = argparse.ArgumentParser(
335 prog="ll-logs",
336 description="Discover and extract ll-relevant JSONL entries from Claude Code logs",
337 formatter_class=argparse.RawDescriptionHelpFormatter,
338 epilog="""
339Examples:
340 %(prog)s discover # List all projects with ll activity
341 %(prog)s tail --loop <name> # Stream live events from an active loop session
342 %(prog)s extract --all # Extract all projects to logs/
343 %(prog)s extract --project /path # Extract one project to logs/<slug>/
344 %(prog)s extract --all --cmd ll-history # Filter to ll-history invocations
345""",
346 )
348 subparsers = parser.add_subparsers(dest="command", help="Available commands")
349 discover_parser = subparsers.add_parser(
350 "discover",
351 help="List all Claude projects with ll activity (one path per line, sorted)",
352 )
353 add_json_arg(discover_parser)
355 tail_parser = subparsers.add_parser(
356 "tail",
357 help="Stream live events from an active loop session",
358 )
359 tail_parser.add_argument("--loop", required=True, metavar="NAME", help="Loop name to tail")
361 extract_parser = subparsers.add_parser(
362 "extract",
363 help="Extract ll-relevant JSONL records to logs/<slug>/<session-id>.jsonl",
364 )
365 target_group = extract_parser.add_mutually_exclusive_group(required=True)
366 target_group.add_argument(
367 "--project",
368 type=Path,
369 metavar="DIR",
370 help="Working directory of the target project",
371 )
372 target_group.add_argument(
373 "--all",
374 action="store_true",
375 help="Extract all projects with ll activity",
376 )
377 extract_parser.add_argument(
378 "--cmd",
379 metavar="TOOL",
380 help="Filter to records containing this ll- tool name (e.g. ll-history)",
381 )
383 return parser
386def _parse_args() -> argparse.Namespace:
387 """Parse command-line arguments. Exposed for testing."""
388 return _build_parser().parse_args()
391def main_logs() -> int:
392 """Entry point for ll-logs command.
394 Returns:
395 0 on success, 1 when no subcommand given or on error.
396 """
397 with cli_event_context(DEFAULT_DB_PATH, "ll-logs", sys.argv[1:]):
398 configure_output()
399 logger = Logger(use_color=use_color_enabled())
401 parser = _build_parser()
402 args = parser.parse_args()
404 if not args.command:
405 parser.print_help()
406 return 1
408 if args.command == "discover":
409 projects = discover_all_projects(logger)
410 if args.json:
411 print_json({"paths": [str(p) for p in projects]})
412 else:
413 for path in projects:
414 print(path)
415 return 0
417 if args.command == "tail":
418 config = BRConfig(Path.cwd())
419 loops_dir = Path(config.loops.loops_dir)
420 return _cmd_tail(args, loops_dir)
422 if args.command == "extract":
423 return _cmd_extract(args, logger)
425 return 1