Coverage for little_loops / cli / logs.py: 11%
1134 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""ll-logs: Discover, extract, analyze log entries from ~/.claude/projects/."""
3from __future__ import annotations
5import argparse
6import json
7import re
8import shutil
9import sqlite3
10import sys
11import time
12from collections import Counter, defaultdict
13from dataclasses import dataclass, field
14from datetime import UTC, datetime, timedelta
15from pathlib import Path
17from little_loops.analytics.association import compute_lift, compute_pmi
18from little_loops.cli.loop.info import ( # private symbol: cross-module coupling; verify signature on upgrade
19 _format_history_event,
20)
21from little_loops.cli.output import configure_output, print_json, table, use_color_enabled
22from little_loops.cli_args import add_json_arg
23from little_loops.config import BRConfig
24from little_loops.logger import Logger
25from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context, resolve_history_db
26from little_loops.user_messages import get_project_folder
28_COMMAND_NAME_RE = re.compile(r"<command-name>/ll:")
29BRIDGE_MARKER = "Bridged from `commands/"
32def _is_ll_relevant(record: dict) -> bool:
33 """Return True if a JSONL record indicates ll activity.
35 Detects three signal types:
36 (a) queue-operation enqueue with /ll: content
37 (b) user records with <command-name>/ll: pattern in message content
38 """
39 record_type = record.get("type")
41 # (a) queue-operation: only enqueue records with /ll: content signal ll activity
42 if record_type == "queue-operation":
43 return (
44 record.get("operation") == "enqueue"
45 and isinstance(record.get("content"), str)
46 and record["content"].startswith("/ll:")
47 )
49 # (b) user records: check message content for <command-name>/ll: pattern
50 if record_type == "user":
51 message = record.get("message", {})
52 if not isinstance(message, dict):
53 return False
54 content = message.get("content")
55 if isinstance(content, str):
56 return bool(_COMMAND_NAME_RE.search(content))
57 if isinstance(content, list):
58 for block in content:
59 if isinstance(block, dict):
60 text = block.get("text", "")
61 if isinstance(text, str) and _COMMAND_NAME_RE.search(text):
62 return True
64 # (c) assistant records: check for Bash tool-use invoking an ll- command
65 if record_type == "assistant":
66 message = record.get("message", {})
67 content = message.get("content", [])
68 if isinstance(content, list):
69 for block in content:
70 if (
71 isinstance(block, dict)
72 and block.get("type") == "tool_use"
73 and block.get("name") == "Bash"
74 ):
75 cmd = block.get("input", {}).get("command", "")
76 if re.search(r"\bll-\w+", cmd):
77 return True
79 return False
82def _has_ll_activity(project_folder: Path) -> bool:
83 """Return True if any non-agent JSONL file in project_folder has ll activity."""
84 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
86 for jsonl_file in jsonl_files:
87 try:
88 with open(jsonl_file, encoding="utf-8") as f:
89 for line in f:
90 line = line.strip()
91 if not line:
92 continue
93 try:
94 record = json.loads(line)
95 except json.JSONDecodeError:
96 continue
97 if _is_ll_relevant(record):
98 return True
99 except OSError:
100 continue
102 return False
105def _extract_cwd_from_project(project_dir: Path) -> Path | None:
106 """Extract the project working directory from cwd fields in JSONL records.
108 Claude Code encodes project paths by replacing '/' with '-', which is
109 lossy for paths containing hyphens. Reading the cwd field from JSONL
110 records gives the canonical path without ambiguity.
111 """
112 jsonl_files = [f for f in project_dir.glob("*.jsonl") if not f.name.startswith("agent-")]
113 for jsonl_file in jsonl_files:
114 try:
115 with open(jsonl_file, encoding="utf-8") as f:
116 for line in f:
117 line = line.strip()
118 if not line:
119 continue
120 try:
121 record = json.loads(line)
122 cwd = record.get("cwd")
123 if isinstance(cwd, str) and cwd:
124 return Path(cwd)
125 except json.JSONDecodeError:
126 continue
127 except OSError:
128 continue
129 return None
132def discover_all_projects(
133 logger: Logger, *, host: str | None = None, existing_only: bool = False
134) -> list[Path]:
135 """Discover all projects with ll activity for the given host.
137 Iterates the host's session directory (e.g. ``~/.claude/projects/`` for
138 Claude Code, ``~/.codex/projects/`` for Codex), resolves each directory
139 name back to an absolute path, checks for ll-relevant JSONL records, and
140 returns a sorted list of paths that exist on disk.
142 Args:
143 logger: Logger instance for diagnostics.
144 host: Host identifier. If None, auto-detects from ``LL_HOOK_HOST``
145 env var (default ``"claude-code"``).
146 existing_only: When True, silently skip paths that don't exist on disk
147 (no debug message). Useful for scripted consumers that want clean
148 stderr as well as clean stdout.
150 Returns:
151 Sorted list of decoded absolute paths for projects with ll activity.
152 """
153 import os as _os
155 if host is None:
156 host = _os.environ.get("LL_HOOK_HOST", "claude-code")
158 if host == "claude-code":
159 projects_root = Path.home() / ".claude" / "projects"
160 elif host == "codex":
161 projects_root = Path.home() / ".codex" / "projects"
162 elif host == "opencode":
163 projects_root = Path.home() / ".opencode" / "projects"
164 elif host == "pi":
165 projects_root = Path.home() / ".pi" / "projects"
166 else:
167 return []
169 if not projects_root.exists():
170 return []
172 results: list[Path] = []
174 for project_dir in projects_root.iterdir():
175 if not project_dir.is_dir():
176 continue
178 # Prefer cwd field from JSONL records; fall back to lossy decode.
179 # The lossy decode ("-Users-foo-bar" -> "/Users/foo/bar") breaks for
180 # paths that contain hyphens (e.g. "little-loops", macOS per-user
181 # temp dirs like /tmp/claude-501/).
182 decoded_path = _extract_cwd_from_project(project_dir) or Path(
183 project_dir.name.replace("-", "/")
184 )
186 if not decoded_path.exists():
187 if not existing_only:
188 logger.debug(f"Decoded path does not exist: {decoded_path}")
189 continue
191 if _has_ll_activity(project_dir):
192 results.append(decoded_path)
194 return sorted(results)
197def _cmd_matches(record: dict, cmd: str) -> bool:
198 """Return True if record contains a Bash tool-use whose command includes cmd."""
199 message = record.get("message", {})
200 content = message.get("content", [])
201 if isinstance(content, list):
202 for block in content:
203 if (
204 isinstance(block, dict)
205 and block.get("type") == "tool_use"
206 and block.get("name") == "Bash"
207 ):
208 command = block.get("input", {}).get("command", "")
209 if cmd in command:
210 return True
211 return False
214_LL_BASH_RE = re.compile(r"\b(ll-[\w-]+)")
215_QUEUE_SKILL_RE = re.compile(r"^/ll:(\S+)")
216_COMMAND_NAME_SKILL_RE = re.compile(r"<command-name>/ll:(\S+)")
217_CONTENT_FREE_RE = re.compile(r"^exit\s+code\s+\d+$", re.IGNORECASE)
220@dataclass
221class InvocationEvent:
222 """A single ll invocation event extracted from a JSONL record."""
224 tool_name: str
225 timestamp: str
226 session_id: str
229def _extract_ll_event_streams(
230 project_folder: Path, *, cutoff: datetime | None = None
231) -> dict[str, list[InvocationEvent]]:
232 """Extract per-session ordered ll-invocation event streams from JSONL files.
234 Walks JSONL files (skipping ``agent-*``), filters to records with ll activity,
235 extracts the tool/skill name, and returns a dict mapping ``sessionId`` to a
236 timestamp-sorted list of ``InvocationEvent``.
238 Args:
239 project_folder: Path to the claude project session directory.
240 cutoff: If set, exclude records with timestamps before this datetime.
242 Returns:
243 Dict of ``{session_id: [InvocationEvent, ...]}`` with events sorted by timestamp.
244 """
245 events_by_session: dict[str, list[InvocationEvent]] = {}
247 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
248 if not jsonl_files:
249 return events_by_session
251 all_events: list[InvocationEvent] = []
253 for jsonl_file in jsonl_files:
254 try:
255 with open(jsonl_file, encoding="utf-8") as f:
256 for line in f:
257 line = line.strip()
258 if not line:
259 continue
260 try:
261 record = json.loads(line)
262 except json.JSONDecodeError:
263 continue
265 tool_name = _extract_tool_name(record)
266 if tool_name is None:
267 continue
269 ts = record.get("timestamp", "")
270 sid = record.get("sessionId", "")
272 evt = InvocationEvent(tool_name=tool_name, timestamp=ts, session_id=sid)
273 all_events.append(evt)
274 except OSError:
275 continue
277 # Apply wall-clock cutoff filter
278 if cutoff is not None:
279 all_events = [e for e in all_events if _parse_iso_timestamp(e.timestamp) >= cutoff]
281 # Bucket by session and sort
282 for evt in all_events:
283 events_by_session.setdefault(evt.session_id, []).append(evt)
285 for session_id in events_by_session:
286 events_by_session[session_id].sort(key=lambda e: e.timestamp)
288 return events_by_session
291@dataclass
292class _InvocationSignal:
293 """Raw ll invocation signal extracted from a JSONL record.
295 Shared by ``_extract_tool_name`` and ``_extract_eval_invocation`` — single
296 source of truth for the three-signal detection logic.
297 """
299 tool_name: str # matched skill/tool name, e.g. "scan-codebase" or "ll-issues"
300 runner: str # signal source: "queue-operation" | "user" | "bash"
301 input_context: str # raw matched text (full cmd for bash; user/queue text otherwise)
304def _detect_ll_signal(record: dict) -> _InvocationSignal | None:
305 """Extract the ll invocation signal from a JSONL record.
307 Detects three signal types:
308 (a) queue-operation enqueue with ``/ll:<name>`` → tool_name=name, runner=queue-operation
309 (b) user records with ``<command-name>/ll:<name>`` → tool_name=name, runner=user
310 (c) assistant Bash tool-use invoking ``ll-<tool>`` → tool_name=match, runner=bash
312 ``input_context`` holds the raw matched text used by eval-export consumers.
313 Returns ``None`` for records carrying no ll invocation signal.
314 """
315 record_type = record.get("type")
317 # (a) queue-operation enqueue
318 if record_type == "queue-operation" and record.get("operation") == "enqueue":
319 content = record.get("content", "")
320 if isinstance(content, str) and content.startswith("/ll:"):
321 m = _QUEUE_SKILL_RE.match(content)
322 if m:
323 return _InvocationSignal(m.group(1), "queue-operation", content)
325 # (b) user records with <command-name>/ll: pattern
326 if record_type == "user":
327 message = record.get("message", {})
328 if not isinstance(message, dict):
329 return None
330 content = message.get("content")
331 text = ""
332 if isinstance(content, str):
333 text = content
334 elif isinstance(content, list):
335 for block in content:
336 if isinstance(block, dict):
337 text = block.get("text", "")
338 if text:
339 break
340 if text:
341 m = _COMMAND_NAME_SKILL_RE.search(text)
342 if m:
343 name = m.group(1)
344 if name.endswith("</command-name>"):
345 name = name[: -len("</command-name>")]
346 return _InvocationSignal(name, "user", text)
348 # (c) assistant Bash tool-use invoking ll-<tool>
349 if record_type == "assistant":
350 message = record.get("message", {})
351 content = message.get("content", [])
352 if isinstance(content, list):
353 for block in content:
354 if (
355 isinstance(block, dict)
356 and block.get("type") == "tool_use"
357 and block.get("name") == "Bash"
358 ):
359 cmd = block.get("input", {}).get("command", "")
360 m = _LL_BASH_RE.search(cmd)
361 if m:
362 return _InvocationSignal(m.group(1), "bash", cmd)
364 return None
367def _extract_tool_name(record: dict) -> str | None:
368 """Extract the ll tool/skill name from a JSONL record."""
369 sig = _detect_ll_signal(record)
370 return sig.tool_name if sig else None
373def _parse_iso_timestamp(ts: str) -> datetime:
374 """Parse an ISO 8601 timestamp string to a timezone-aware datetime.
376 Handles both ``Z``-suffixed and ``+00:00`` offset formats. Returns
377 ``datetime.min`` with UTC tzinfo for unparseable input.
378 """
379 if not ts:
380 return datetime.min.replace(tzinfo=UTC)
381 try:
382 # Handle Z suffix
383 if ts.endswith("Z"):
384 ts = ts[:-1] + "+00:00"
385 dt = datetime.fromisoformat(ts)
386 if dt.tzinfo is None:
387 dt = dt.replace(tzinfo=UTC)
388 return dt
389 except (ValueError, TypeError):
390 return datetime.min.replace(tzinfo=UTC)
393@dataclass
394class Edge:
395 """A transition edge within an n-gram chain."""
397 from_: str
398 to: str
399 freq: float
400 pmi: float | None = None
401 lift: float | None = None
404@dataclass
405class ChainResult:
406 """An n-gram chain with occurrence count and per-edge transition frequencies."""
408 chain: list[str]
409 count: int
410 edges: list[Edge]
411 pmi: float | None = None
412 lift: float | None = None
414 def to_dict(self) -> dict:
415 edges_out = []
416 for e in self.edges:
417 ed: dict = {"from": e.from_, "to": e.to, "freq": e.freq}
418 if e.pmi is not None:
419 ed["pmi"] = e.pmi
420 if e.lift is not None:
421 ed["lift"] = e.lift
422 edges_out.append(ed)
423 result: dict = {"chain": self.chain, "count": self.count, "edges": edges_out}
424 if self.pmi is not None:
425 result["pmi"] = self.pmi
426 if self.lift is not None:
427 result["lift"] = self.lift
428 return result
431def _count_ngrams(
432 events_by_session: dict[str, list[InvocationEvent]],
433 min_len: int = 2,
434) -> tuple[Counter, Counter]:
435 """Count n-grams and unigrams across per-session event streams.
437 Args:
438 events_by_session: Per-session ordered event streams.
439 min_len: Minimum n-gram length (window size).
441 Returns:
442 Tuple of (ngram_counter, unigram_counter).
443 ngram_counter maps ``(tool_1, tool_2, ...)`` tuples to occurrence counts.
444 unigram_counter maps individual tool names to occurrence counts.
445 """
446 counter: Counter = Counter()
447 unigram_counter: Counter = Counter()
448 for events in events_by_session.values():
449 names = [e.tool_name for e in events]
450 for name in names:
451 unigram_counter[name] += 1
452 for n in range(min_len, len(names) + 1):
453 for i in range(len(names) - n + 1):
454 ngram = tuple(names[i : i + n])
455 counter[ngram] += 1
456 return counter, unigram_counter
459def _build_chain_results(
460 counter: Counter,
461 unigram_counter: Counter | None = None,
462 min_count: int = 1,
463 top: int | None = None,
464) -> list[ChainResult]:
465 """Build ranked ``ChainResult`` list from n-gram counter.
467 Args:
468 counter: n-gram counter from ``_count_ngrams``.
469 unigram_counter: Unigram counter from ``_count_ngrams``. When provided,
470 PMI and lift scores are attached to each edge and chain result.
471 min_count: Minimum occurrence count to include.
472 top: If set, limit to top N chains by frequency.
474 Returns:
475 List of ``ChainResult`` sorted by count descending.
476 """
477 all_transitions: Counter = Counter()
478 out_degree: Counter = Counter()
479 for ngram_key, count in counter.items():
480 for i in range(len(ngram_key) - 1):
481 pair = (ngram_key[i], ngram_key[i + 1])
482 all_transitions[pair] += count
483 out_degree[ngram_key[i]] += count
485 total_unigrams = sum(unigram_counter.values()) if unigram_counter else 0
487 results: list[ChainResult] = []
488 for ngram, count in counter.most_common():
489 if count < min_count:
490 continue
491 edges = _compute_edges(
492 ngram, all_transitions, out_degree, unigram_counter, total_unigrams, counter
493 )
495 chain_pmi: float | None = None
496 chain_lift: float | None = None
497 if edges and all(e.lift is not None for e in edges):
498 chain_lift = min(e.lift for e in edges if e.lift is not None) # type: ignore[type-var]
499 chain_pmi = min(e.pmi for e in edges if e.pmi is not None) # type: ignore[type-var]
501 results.append(
502 ChainResult(chain=list(ngram), count=count, edges=edges, pmi=chain_pmi, lift=chain_lift)
503 )
505 if top is not None:
506 results = results[:top]
508 return results
511def _compute_edges(
512 ngram: tuple[str, ...],
513 all_transitions: Counter,
514 out_degree: Counter,
515 unigram_counter: Counter | None = None,
516 total_unigrams: int = 0,
517 ngram_counter: Counter | None = None,
518) -> list[Edge]:
519 """Compute per-edge transition frequencies and PMI/lift for an n-gram chain.
521 For each adjacent pair ``(from, to)`` in the chain, computes the frequency
522 as the proportion of times ``from → to`` appears out of all transitions
523 originating from ``from`` across the entire corpus. When ``unigram_counter``
524 and ``ngram_counter`` are provided, also computes PMI and lift for each edge
525 using the raw bigram count from ``ngram_counter`` (not the overcounted
526 ``all_transitions`` which accumulates across all n-gram lengths).
527 """
528 edges: list[Edge] = []
529 for i in range(len(ngram) - 1):
530 from_ = ngram[i]
531 to = ngram[i + 1]
532 pair = (from_, to)
533 total_out = out_degree.get(from_, 0)
534 freq = all_transitions.get(pair, 0) / total_out if total_out > 0 else 0.0
536 edge_pmi: float | None = None
537 edge_lift: float | None = None
538 if unigram_counter and total_unigrams > 0 and ngram_counter is not None:
539 # Use the raw bigram count (not all_transitions which overcounts from longer n-grams)
540 count_ab = ngram_counter.get(pair, 0)
541 count_a = unigram_counter.get(from_, 0)
542 count_b = unigram_counter.get(to, 0)
543 if count_ab > 0 and count_a > 0 and count_b > 0:
544 edge_lift = round(compute_lift(count_ab, count_a, count_b, total_unigrams), 4)
545 edge_pmi = round(compute_pmi(count_ab, count_a, count_b, total_unigrams), 4)
547 edges.append(Edge(from_=from_, to=to, freq=round(freq, 4), pmi=edge_pmi, lift=edge_lift))
549 return edges
552def _cmd_sequences(args: argparse.Namespace, logger: Logger) -> int:
553 """Extract n-grams of ll invocations from JSONL log files."""
554 if args.project:
555 cwd_path: Path = args.project
556 project_folder = get_project_folder(cwd_path)
557 if project_folder is None:
558 logger.error(f"No session project folder found for: {cwd_path}")
559 return 1
560 project_items = [(cwd_path, project_folder)]
561 else:
562 decoded_paths = discover_all_projects(logger)
563 project_items = []
564 for decoded_path in decoded_paths:
565 folder = get_project_folder(decoded_path)
566 if folder is not None:
567 project_items.append((decoded_path, folder))
569 cutoff = (
570 datetime.now(UTC) - timedelta(days=args.window_days)
571 if args.window_days is not None
572 else None
573 )
575 # Aggregate events across all projects
576 all_events: dict[str, list[InvocationEvent]] = {}
577 for _cwd_path, project_folder in project_items:
578 events = _extract_ll_event_streams(project_folder, cutoff=cutoff)
579 for sid, evt_list in events.items():
580 all_events.setdefault(sid, []).extend(evt_list)
582 # Sort each session's events by timestamp
583 for sid in all_events:
584 all_events[sid].sort(key=lambda e: e.timestamp)
586 # Count n-grams
587 counter, unigram_counter = _count_ngrams(all_events, min_len=args.min_len)
588 results = _build_chain_results(counter, unigram_counter, min_count=args.min_count, top=args.top)
590 if args.json:
591 print_json([r.to_dict() for r in results])
592 else:
593 if not results:
594 print("No sequences found.")
595 return 0
597 # Print ranked table
598 for rank, r in enumerate(results, 1):
599 chain_str = " → ".join(r.chain)
600 print(f"{rank}. [{r.count}] {chain_str}")
601 for edge in r.edges:
602 print(f" {edge.from_} → {edge.to}: {edge.freq:.4f}")
604 return 0
607def generate_index(logs_dir: Path) -> None:
608 """Generate logs/index.md summarising extracted projects."""
609 rows = []
611 if logs_dir.exists():
612 for subdir in sorted(logs_dir.iterdir()):
613 if not subdir.is_dir():
614 continue
616 jsonl_files = [f for f in subdir.glob("*.jsonl") if not f.name.startswith("agent-")]
617 if not jsonl_files:
618 continue
620 timestamps: list[str] = []
621 for jsonl_file in jsonl_files:
622 try:
623 with open(jsonl_file, encoding="utf-8") as f:
624 for line in f:
625 line = line.strip()
626 if not line:
627 continue
628 try:
629 record = json.loads(line)
630 except json.JSONDecodeError:
631 continue
632 ts = record.get("timestamp")
633 if ts:
634 timestamps.append(ts)
635 except OSError:
636 continue
638 if timestamps:
639 earliest = min(timestamps)[:10]
640 latest = max(timestamps)[:10]
641 date_range = f"{earliest} – {latest}" if earliest != latest else earliest
642 else:
643 date_range = ""
645 rows.append((subdir.name, len(jsonl_files), date_range))
647 lines = ["# Logs Index", ""]
648 if rows:
649 lines.append("| Project | Sessions | Date Range |")
650 lines.append("|---------|----------|------------|")
651 for name, count, date_range in rows:
652 lines.append(f"| {name} | {count} | {date_range} |")
653 else:
654 lines.append("*No projects extracted yet.*")
655 lines.append("")
657 logs_dir.mkdir(parents=True, exist_ok=True)
658 (logs_dir / "index.md").write_text("\n".join(lines), encoding="utf-8")
661def _cmd_extract(args: argparse.Namespace, logger: Logger) -> int:
662 """Extract ll-relevant JSONL records to logs/<slug>/<session-id>.jsonl."""
663 if args.project:
664 cwd_path: Path = args.project
665 project_folder = get_project_folder(cwd_path)
666 if project_folder is None:
667 logger.error(f"No session project folder found for: {cwd_path}")
668 return 1
669 project_items = [(cwd_path, project_folder)]
670 else:
671 decoded_paths = discover_all_projects(logger)
672 project_items = []
673 for decoded_path in decoded_paths:
674 folder = get_project_folder(decoded_path)
675 if folder is not None:
676 project_items.append((decoded_path, folder))
678 for cwd_path, project_folder in project_items:
679 slug = cwd_path.resolve().name
680 buckets: dict[str, list[dict]] = {}
682 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
683 for jsonl_file in jsonl_files:
684 try:
685 with open(jsonl_file, encoding="utf-8") as f:
686 for line in f:
687 line = line.strip()
688 if not line:
689 continue
690 try:
691 record = json.loads(line)
692 except json.JSONDecodeError:
693 continue
694 if _is_ll_relevant(record):
695 session_id = record.get("sessionId", "")
696 buckets.setdefault(session_id, []).append(record)
697 except OSError:
698 continue
700 if args.cmd:
701 filtered: dict[str, list[dict]] = {}
702 for session_id, records in buckets.items():
703 matching = [r for r in records if _cmd_matches(r, args.cmd)]
704 if matching:
705 filtered[session_id] = matching
706 buckets = filtered
708 out_base = Path.cwd() / "logs" / slug
709 for session_id, records in buckets.items():
710 out_file = out_base / f"{session_id}.jsonl"
711 out_file.parent.mkdir(parents=True, exist_ok=True)
712 with open(out_file, "w", encoding="utf-8") as f:
713 for record in records:
714 f.write(json.dumps(record) + "\n")
716 generate_index(Path.cwd() / "logs")
717 return 0
720def _cmd_tail(args: argparse.Namespace, loops_dir: Path) -> int:
721 """Stream live events from an active loop session."""
722 events_file = loops_dir / ".running" / f"{args.loop}.events.jsonl"
724 if not events_file.exists():
725 print(f"No active session for loop '{args.loop}'", file=sys.stderr)
726 return 1
728 width = shutil.get_terminal_size().columns
729 try:
730 with open(events_file, encoding="utf-8") as f:
731 f.seek(0, 2)
732 while True:
733 line = f.readline()
734 if line:
735 line = line.strip()
736 if line:
737 try:
738 event = json.loads(line)
739 except json.JSONDecodeError:
740 continue
741 formatted = _format_history_event(event, verbose=False, width=width)
742 if formatted is not None:
743 print(formatted)
744 else:
745 time.sleep(0.1)
746 except KeyboardInterrupt:
747 return 0
749 return 0
752_CORRECTION_WINDOW_SEC = 30
755def _aggregate_skill_stats(
756 db_path: Path,
757 *,
758 cutoff: datetime | None = None,
759) -> dict[str, dict[str, int]] | None:
760 """Aggregate per-skill invocation and correction counts from history.db.
762 Returns None when the database is absent, or an empty dict when the database
763 has no skill_events rows. Corrections are attributed to the most recent skill
764 event in the same session within _CORRECTION_WINDOW_SEC seconds.
765 """
766 if not db_path.exists():
767 return None
769 conn = sqlite3.connect(str(db_path))
770 conn.row_factory = sqlite3.Row
771 try:
772 try:
773 skill_rows = conn.execute(
774 "SELECT ts, session_id, skill_name FROM skill_events ORDER BY ts"
775 ).fetchall()
776 except sqlite3.OperationalError:
777 return None
779 if not skill_rows:
780 return {}
782 if cutoff is not None:
783 skill_rows = [r for r in skill_rows if _parse_iso_timestamp(r["ts"] or "") >= cutoff]
785 stats: dict[str, dict[str, int]] = defaultdict(lambda: {"invocations": 0, "corrections": 0})
786 for row in skill_rows:
787 stats[row["skill_name"] or "unknown"]["invocations"] += 1
789 session_skills: dict[str, list[tuple[str, str]]] = defaultdict(list)
790 for row in skill_rows:
791 sid = row["session_id"] or ""
792 session_skills[sid].append((row["ts"] or "", row["skill_name"] or "unknown"))
794 try:
795 corr_rows = conn.execute(
796 "SELECT ts, session_id FROM user_corrections ORDER BY ts"
797 ).fetchall()
798 except sqlite3.OperationalError:
799 corr_rows = []
801 for corr in corr_rows:
802 c_ts = corr["ts"] or ""
803 sid = corr["session_id"] or ""
804 candidates = session_skills.get(sid, [])
805 best_skill: str | None = None
806 best_ts: str = ""
807 for s_ts, s_name in candidates:
808 if s_ts <= c_ts and s_ts >= best_ts:
809 best_ts = s_ts
810 best_skill = s_name
811 if best_skill is not None:
812 elapsed = (
813 _parse_iso_timestamp(c_ts) - _parse_iso_timestamp(best_ts)
814 ).total_seconds()
815 if 0 <= elapsed <= _CORRECTION_WINDOW_SEC:
816 stats[best_skill]["corrections"] += 1
818 return dict(stats)
819 finally:
820 conn.close()
823def _load_catalog_names(root_dir: Path) -> set[str]:
824 """Load normalized skill/command names from skills/ and commands/ under root_dir.
826 Excludes bridge skills (containing BRIDGE_MARKER) and skills/commands with
827 disable-model-invocation: true. Normalizes names by stripping the "ll-" prefix
828 so catalog names match the skill_events.skill_name recording convention.
829 """
830 import yaml
832 names: set[str] = set()
834 skills_dir = root_dir / "skills"
835 if skills_dir.is_dir():
836 for skill_md in sorted(skills_dir.glob("*/SKILL.md")):
837 try:
838 text = skill_md.read_text()
839 except OSError:
840 continue
841 if BRIDGE_MARKER in text:
842 continue
843 name: str = skill_md.parent.name
844 if text.startswith("---"):
845 end = text.find("---", 3)
846 if end != -1:
847 try:
848 fm = yaml.safe_load(text[3:end]) or {}
849 except yaml.YAMLError:
850 fm = {}
851 if isinstance(fm, dict):
852 dmi = fm.get("disable-model-invocation")
853 if (isinstance(dmi, bool) and dmi) or (
854 isinstance(dmi, str) and dmi.strip().lower() in ("true", "yes", "1")
855 ):
856 continue
857 name = str(fm.get("name") or name)
858 if name.startswith("ll-"):
859 name = name[3:]
860 if name:
861 names.add(name)
863 commands_dir = root_dir / "commands"
864 if commands_dir.is_dir():
865 for cmd_md in sorted(commands_dir.glob("*.md")):
866 stem = cmd_md.stem
867 try:
868 text = cmd_md.read_text()
869 except OSError:
870 names.add(stem)
871 continue
872 if text.startswith("---"):
873 end = text.find("---", 3)
874 if end != -1:
875 try:
876 fm = yaml.safe_load(text[3:end]) or {}
877 except yaml.YAMLError:
878 fm = {}
879 if isinstance(fm, dict):
880 dmi = fm.get("disable-model-invocation")
881 if (isinstance(dmi, bool) and dmi) or (
882 isinstance(dmi, str) and dmi.strip().lower() in ("true", "yes", "1")
883 ):
884 continue
885 names.add(stem)
887 return names
890def _cmd_dead_skills(args: argparse.Namespace, logger: Logger) -> int:
891 """List catalog skills/commands never or rarely invoked within the window."""
892 if args.project:
893 db_paths = [Path(args.project) / ".ll" / "history.db"]
894 catalog_root = Path(args.project)
895 else:
896 decoded_paths = discover_all_projects(logger)
897 db_paths = [p / ".ll" / "history.db" for p in decoded_paths]
898 catalog_root = Path.cwd()
900 cutoff = (
901 datetime.now(UTC) - timedelta(days=args.window_days)
902 if args.window_days is not None
903 else None
904 )
906 merged: dict[str, int] = defaultdict(int)
907 for db_path in db_paths:
908 result = _aggregate_skill_stats(db_path, cutoff=cutoff)
909 if result is None:
910 continue
911 for skill, counts in result.items():
912 merged[skill] += counts["invocations"]
914 catalog_names = _load_catalog_names(catalog_root)
915 if not catalog_names:
916 logger.warning(
917 "No catalog skills found — run from an ll project root with skills/ directory."
918 )
919 return 0
921 threshold = args.threshold
922 rows = []
923 for name in sorted(catalog_names):
924 count = merged.get(name, 0)
925 if count == 0:
926 rows.append({"skill": name, "invocations": 0, "tier": "never"})
927 elif count <= threshold:
928 rows.append({"skill": name, "invocations": count, "tier": "rarely"})
930 if args.json:
931 print_json(rows)
932 return 0
934 if not rows:
935 print("No dead or rarely-invoked skills found.")
936 return 0
938 headers = ["Skill", "Invocations", "Tier"]
939 table_rows = [[str(r["skill"]), str(r["invocations"]), str(r["tier"])] for r in rows]
940 print(table(headers, table_rows))
941 return 0
944def _load_cli_allowlist(root: Path) -> frozenset[str]:
945 """Return ll-* CLI names from [project.scripts] in scripts/pyproject.toml.
947 Returns an empty frozenset if the file cannot be read; the allowlist check
948 is skipped when the set is empty so fallback behavior is open (no filtering).
949 """
950 import tomllib
952 pyproject = root / "scripts" / "pyproject.toml"
953 try:
954 with open(pyproject, "rb") as f:
955 data = tomllib.load(f)
956 except (OSError, ValueError):
957 return frozenset()
958 scripts = data.get("project", {}).get("scripts", {})
959 return frozenset(k for k in scripts if k.startswith("ll-"))
962def _is_content_free_error(error_text: str) -> bool:
963 """Return True if error_text carries no signal beyond a bare exit code."""
964 return bool(_CONTENT_FREE_RE.match(error_text.strip()))
967_STACK_FRAME_RE = re.compile(r'\s*File "[^"]+", line \d+[^\n]*')
968_ABS_PATH_RE = re.compile(r"/(?:[^\s,;\"']+/)+[^\s,;\"']+")
969_LINE_NUM_RE = re.compile(r"\bline \d+\b")
970_LL_VERIFY_RE = re.compile(r"^ll-verify-\w+")
973def _normalize_error_sig(text: str) -> str:
974 """Strip volatile parts (paths, line numbers, stack frames) from error text.
976 Returns a stable string suitable as a cluster key.
977 """
978 text = _STACK_FRAME_RE.sub("", text)
979 text = _ABS_PATH_RE.sub("<path>", text)
980 text = _LINE_NUM_RE.sub("line N", text)
981 return re.sub(r"\s+", " ", text).strip()[:300]
984def _extract_error_text(content: object) -> str:
985 """Extract plain text from a tool_result content field (string or list of text blocks)."""
986 if isinstance(content, str):
987 return content
988 if isinstance(content, list):
989 parts: list[str] = []
990 for item in content:
991 if isinstance(item, dict):
992 text = item.get("text", "")
993 if isinstance(text, str):
994 parts.append(text)
995 return "\n".join(parts)
996 return ""
999@dataclass
1000class _FailureCluster:
1001 """Aggregated failure cluster keyed on (cwd_path, tool_name, normalized_sig)."""
1003 tool_name: str
1004 normalized_sig: str
1005 count: int
1006 sample_error: str
1007 session_ids: list[str]
1008 cwd_path: Path = field(default_factory=lambda: Path("."))
1011def _cmd_scan_failures(args: argparse.Namespace, logger: Logger) -> int:
1012 """Mine failed ll-* Bash calls from interactive session JSONL logs."""
1013 from little_loops.issue_lifecycle import FailureType, classify_failure
1015 _cli_allowlist = _load_cli_allowlist(Path.cwd())
1017 if args.project:
1018 cwd_path: Path = args.project
1019 project_folder = get_project_folder(cwd_path)
1020 if project_folder is None:
1021 logger.error(f"No session project folder found for: {cwd_path}")
1022 return 1
1023 project_items = [(cwd_path, project_folder)]
1024 else:
1025 decoded_paths = discover_all_projects(logger)
1026 project_items = []
1027 for decoded_path in decoded_paths:
1028 folder = get_project_folder(decoded_path)
1029 if folder is not None:
1030 project_items.append((decoded_path, folder))
1032 # raw_clusters maps (cwd_path, tool_name, normalized_sig) -> (count, sample_error, session_ids, latest_ts)
1033 raw_clusters: dict[tuple[Path, str, str], tuple[int, str, list[str], str]] = {}
1035 for _cwd_path, project_folder in project_items:
1036 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
1038 for jsonl_file in jsonl_files:
1039 # pending maps tool_use_id -> (ll_tool_name, timestamp)
1040 pending: dict[str, tuple[str, str]] = {}
1042 try:
1043 with open(jsonl_file, encoding="utf-8") as f:
1044 for line in f:
1045 line = line.strip()
1046 if not line:
1047 continue
1048 try:
1049 record = json.loads(line)
1050 except json.JSONDecodeError:
1051 continue
1053 record_type = record.get("type")
1054 ts = record.get("timestamp", "")
1055 session_id = record.get("sessionId", "")
1057 if record_type == "assistant":
1058 message = record.get("message", {})
1059 content = message.get("content", [])
1060 if not isinstance(content, list):
1061 continue
1062 for block in content:
1063 if not isinstance(block, dict):
1064 continue
1065 if block.get("type") != "tool_use" or block.get("name") != "Bash":
1066 continue
1067 cmd = block.get("input", {}).get("command", "")
1068 m = _LL_BASH_RE.search(cmd)
1069 if not m:
1070 continue
1071 tool_name = m.group(1)
1072 # Skip tokens that are not real ll CLIs (e.g. ll-labs, ll-marketing)
1073 if _cli_allowlist and tool_name not in _cli_allowlist:
1074 continue
1075 block_id = block.get("id", "")
1076 if block_id:
1077 pending[block_id] = (tool_name, ts)
1079 elif record_type == "user":
1080 message = record.get("message", {})
1081 content = message.get("content", [])
1082 if not isinstance(content, list) or not content:
1083 continue
1084 for block in content:
1085 if not isinstance(block, dict):
1086 continue
1087 if block.get("type") != "tool_result":
1088 continue
1089 tool_use_id = block.get("tool_use_id", "")
1090 if tool_use_id not in pending:
1091 continue
1092 tool_name, _invoke_ts = pending.pop(tool_use_id)
1094 # Skip ll-verify-* tools — exit 1 is expected gate behavior
1095 if _LL_VERIFY_RE.match(tool_name):
1096 continue
1098 is_error_flag = block.get("is_error") is True
1099 raw_content = block.get("content", "")
1100 error_text = _extract_error_text(raw_content)
1101 has_traceback = "Traceback (most recent call last)" in error_text
1103 if not (is_error_flag or has_traceback):
1104 continue
1106 returncode = 1 if is_error_flag else 0
1107 failure_type, _reason = classify_failure(error_text, returncode)
1108 if failure_type in (
1109 FailureType.TRANSIENT,
1110 FailureType.NON_RECOVERABLE,
1111 ):
1112 continue
1114 normalized_sig = _normalize_error_sig(error_text)
1115 key = (_cwd_path, tool_name, normalized_sig)
1117 if key in raw_clusters:
1118 cnt, sample, sids, _lts = raw_clusters[key]
1119 if session_id not in sids:
1120 sids.append(session_id)
1121 raw_clusters[key] = (cnt + 1, sample, sids, ts)
1122 else:
1123 raw_clusters[key] = (1, error_text[:500], [session_id], ts)
1124 except OSError:
1125 continue
1127 # Apply wall-clock cutoff filter
1128 if args.window_days is not None:
1129 cutoff = datetime.now(UTC) - timedelta(days=args.window_days)
1130 raw_clusters = {
1131 k: v for k, v in raw_clusters.items() if _parse_iso_timestamp(v[3]) >= cutoff
1132 }
1134 # Drop content-free clusters (bare "Exit code N" with no error body)
1135 raw_clusters = {k: v for k, v in raw_clusters.items() if not _is_content_free_error(v[1])}
1137 clusters: list[_FailureCluster] = sorted(
1138 [
1139 _FailureCluster(
1140 tool_name=k[1],
1141 normalized_sig=k[2],
1142 count=v[0],
1143 sample_error=v[1],
1144 session_ids=v[2],
1145 cwd_path=k[0],
1146 )
1147 for k, v in raw_clusters.items()
1148 ],
1149 key=lambda c: c.count,
1150 reverse=True,
1151 )
1153 if not clusters:
1154 if not args.json:
1155 print("No ll-* failures found.")
1156 else:
1157 print_json([])
1158 return 0
1160 if args.capture:
1161 capture_foreign = getattr(args, "capture_foreign", False)
1162 return _capture_failure_clusters(clusters, logger, capture_foreign=capture_foreign)
1164 if args.json:
1165 print_json(
1166 [
1167 {
1168 "tool": c.tool_name,
1169 "count": c.count,
1170 "normalized_sig": c.normalized_sig,
1171 "sample_error": c.sample_error,
1172 "session_ids": c.session_ids,
1173 }
1174 for c in clusters
1175 ]
1176 )
1177 return 0
1179 for c in clusters:
1180 print(f"[{c.count}x] {c.tool_name}")
1181 print(f" Sessions: {', '.join(c.session_ids[:5])}")
1182 for sl in c.sample_error.splitlines()[:5]:
1183 print(f" {sl}")
1184 print()
1186 return 0
1189def _capture_failure_clusters(
1190 clusters: list[_FailureCluster], logger: Logger, capture_foreign: bool = False
1191) -> int:
1192 """Create bug issue files for each distinct failure cluster (--capture mode)."""
1193 from little_loops.issue_lifecycle import create_issue_from_failure
1194 from little_loops.issue_parser import IssueInfo
1196 config = BRConfig(Path.cwd())
1197 current_project = Path.cwd().resolve()
1198 created = 0
1199 skipped_foreign = 0
1201 for c in clusters:
1202 if not capture_foreign and c.cwd_path.resolve() != current_project:
1203 skipped_foreign += 1
1204 continue
1205 stub_info = IssueInfo(
1206 path=Path(f"cli/{c.tool_name}"),
1207 issue_type="bugs",
1208 priority="P1",
1209 issue_id=c.tool_name,
1210 title=f"Tool failure in {c.tool_name}",
1211 )
1212 result = create_issue_from_failure(c.sample_error, stub_info, config, logger)
1213 if result is not None:
1214 logger.info(f"Created: {result.name}")
1215 created += 1
1217 if skipped_foreign:
1218 logger.info(
1219 f"Skipped {skipped_foreign} cluster(s) from other projects "
1220 "(use --capture-foreign to include them)."
1221 )
1222 logger.info(f"Captured {created} failure cluster(s) as bug issues.")
1223 return 0
1226def _cmd_stats(args: argparse.Namespace, logger: Logger) -> int:
1227 """Aggregate skill invocation frequency and correction rate from history.db."""
1228 if args.project:
1229 db_paths = [args.project / ".ll" / "history.db"]
1230 else:
1231 decoded_paths = discover_all_projects(logger)
1232 db_paths = [p / ".ll" / "history.db" for p in decoded_paths]
1234 cutoff = (
1235 datetime.now(UTC) - timedelta(days=args.window_days)
1236 if args.window_days is not None
1237 else None
1238 )
1240 merged: dict[str, dict[str, int]] = defaultdict(lambda: {"invocations": 0, "corrections": 0})
1241 found_any_db = False
1242 for db_path in db_paths:
1243 result = _aggregate_skill_stats(db_path, cutoff=cutoff)
1244 if result is None:
1245 continue
1246 found_any_db = True
1247 for skill, counts in result.items():
1248 merged[skill]["invocations"] += counts["invocations"]
1249 merged[skill]["corrections"] += counts["corrections"]
1251 if not merged:
1252 if not found_any_db:
1253 logger.warning("No history.db found — run with an active ll project.")
1254 else:
1255 print("No skill events recorded yet.")
1256 return 0
1258 sort_key = getattr(args, "sort", "freq")
1259 if sort_key == "corrections":
1260 ranked = sorted(merged.items(), key=lambda kv: kv[1]["corrections"], reverse=True)
1261 else:
1262 ranked = sorted(merged.items(), key=lambda kv: kv[1]["invocations"], reverse=True)
1264 if args.json:
1265 rows_json = [
1266 {
1267 "skill": skill,
1268 "invocations": counts["invocations"],
1269 "corrections": counts["corrections"],
1270 "correction_rate": (
1271 round(counts["corrections"] / counts["invocations"], 4)
1272 if counts["invocations"] > 0
1273 else 0.0
1274 ),
1275 }
1276 for skill, counts in ranked
1277 ]
1278 print_json(rows_json)
1279 return 0
1281 headers = ["Skill", "Invocations", "Corrections", "Corr%"]
1282 rows = []
1283 for skill, counts in ranked:
1284 inv = counts["invocations"]
1285 corr = counts["corrections"]
1286 corr_pct = f"{corr / inv * 100:.1f}%" if inv > 0 else "0.0%"
1287 rows.append([skill, str(inv), str(corr), corr_pct])
1289 print(table(headers, rows))
1290 return 0
1293def _resolve_session_log(session_ref: str, db_path: Path) -> Path | None:
1294 """Resolve a session reference (session ID or JSONL path) to a JSONL file path.
1296 Tries in order:
1297 1. Direct file path if the ref resolves to an existing ``.jsonl`` file
1298 2. DB lookup of ``session_id → jsonl_path`` in the sessions table
1299 Returns None if unresolvable.
1300 """
1301 candidate = Path(session_ref)
1302 if candidate.suffix == ".jsonl" and candidate.exists():
1303 return candidate
1305 if db_path.exists():
1306 conn = sqlite3.connect(str(db_path))
1307 conn.row_factory = sqlite3.Row
1308 try:
1309 row = conn.execute(
1310 "SELECT jsonl_path FROM sessions WHERE session_id = ?",
1311 (session_ref,),
1312 ).fetchone()
1313 if row and row["jsonl_path"]:
1314 return Path(row["jsonl_path"])
1315 except sqlite3.OperationalError:
1316 pass
1317 finally:
1318 conn.close()
1320 return None
1323def _events_from_jsonl(jsonl_path: Path) -> list[InvocationEvent]:
1324 """Extract ll invocation events from a single JSONL file, sorted by timestamp."""
1325 events: list[InvocationEvent] = []
1326 try:
1327 with open(jsonl_path, encoding="utf-8") as f:
1328 for line in f:
1329 line = line.strip()
1330 if not line:
1331 continue
1332 try:
1333 record = json.loads(line)
1334 except json.JSONDecodeError:
1335 continue
1336 tool_name = _extract_tool_name(record)
1337 if tool_name is None:
1338 continue
1339 ts = record.get("timestamp", "")
1340 sid = record.get("sessionId", "")
1341 events.append(InvocationEvent(tool_name=tool_name, timestamp=ts, session_id=sid))
1342 except OSError:
1343 pass
1344 events.sort(key=lambda e: e.timestamp)
1345 return events
1348@dataclass
1349class SessionDiff:
1350 """Behavioral diff between two ll sessions."""
1352 session_a: str
1353 session_b: str
1354 skills_added: list[str]
1355 skills_removed: list[str]
1356 count_deltas: dict[str, dict[str, int]]
1357 sequence_diff: list[str]
1359 def to_dict(self) -> dict:
1360 return {
1361 "session_a": self.session_a,
1362 "session_b": self.session_b,
1363 "skills_added": self.skills_added,
1364 "skills_removed": self.skills_removed,
1365 "count_deltas": self.count_deltas,
1366 "sequence_diff": self.sequence_diff,
1367 }
1370def _compute_session_diff(
1371 session_a: str,
1372 events_a: list[InvocationEvent],
1373 session_b: str,
1374 events_b: list[InvocationEvent],
1375) -> SessionDiff:
1376 """Compute the behavioral diff between two session event streams."""
1377 import difflib
1378 from collections import Counter as _Counter
1380 names_a = [e.tool_name for e in events_a]
1381 names_b = [e.tool_name for e in events_b]
1383 set_a = set(names_a)
1384 set_b = set(names_b)
1385 skills_added = sorted(set_b - set_a)
1386 skills_removed = sorted(set_a - set_b)
1388 counter_a: Counter = _Counter(names_a)
1389 counter_b: Counter = _Counter(names_b)
1390 count_deltas: dict[str, dict[str, int]] = {}
1391 for skill in sorted(set_a | set_b):
1392 ca = counter_a.get(skill, 0)
1393 cb = counter_b.get(skill, 0)
1394 if ca != cb:
1395 count_deltas[skill] = {"a": ca, "b": cb, "delta": cb - ca}
1397 label_a = f"session_a ({session_a[:8]})" if len(session_a) > 8 else f"session_a ({session_a})"
1398 label_b = f"session_b ({session_b[:8]})" if len(session_b) > 8 else f"session_b ({session_b})"
1399 sequence_diff = list(
1400 difflib.unified_diff(names_a, names_b, fromfile=label_a, tofile=label_b, lineterm="")
1401 )
1403 return SessionDiff(
1404 session_a=session_a,
1405 session_b=session_b,
1406 skills_added=skills_added,
1407 skills_removed=skills_removed,
1408 count_deltas=count_deltas,
1409 sequence_diff=sequence_diff,
1410 )
1413def _cmd_diff(args: argparse.Namespace, logger: Logger) -> int:
1414 """Compare two sessions' ll-invocation behavior."""
1415 db_path = resolve_history_db()
1417 path_a = _resolve_session_log(args.session_a, db_path)
1418 if path_a is None:
1419 logger.error(f"Cannot resolve session: {args.session_a}")
1420 return 1
1422 path_b = _resolve_session_log(args.session_b, db_path)
1423 if path_b is None:
1424 logger.error(f"Cannot resolve session: {args.session_b}")
1425 return 1
1427 events_a = _events_from_jsonl(path_a)
1428 events_b = _events_from_jsonl(path_b)
1430 diff = _compute_session_diff(args.session_a, events_a, args.session_b, events_b)
1432 if args.json:
1433 print_json(diff.to_dict())
1434 return 0
1436 if (
1437 not diff.skills_added
1438 and not diff.skills_removed
1439 and not diff.count_deltas
1440 and not diff.sequence_diff
1441 ):
1442 print("No behavioral differences found.")
1443 return 0
1445 if diff.skills_added:
1446 print(f"Skills added ({len(diff.skills_added)}):")
1447 for s in diff.skills_added:
1448 print(f" + {s}")
1450 if diff.skills_removed:
1451 if diff.skills_added:
1452 print()
1453 print(f"Skills removed ({len(diff.skills_removed)}):")
1454 for s in diff.skills_removed:
1455 print(f" - {s}")
1457 if diff.count_deltas:
1458 print()
1459 print("Invocation count changes:")
1460 for skill, counts in sorted(diff.count_deltas.items()):
1461 delta_str = f"+{counts['delta']}" if counts["delta"] > 0 else str(counts["delta"])
1462 print(f" {skill}: {counts['a']} → {counts['b']} ({delta_str})")
1464 if diff.sequence_diff:
1465 print()
1466 print("Sequence diff:")
1467 for line in diff.sequence_diff:
1468 print(f" {line}")
1470 return 0
1473_ISSUE_ID_RE = re.compile(r"\b[A-Z]+-\d+\b")
1476@dataclass
1477class _EvalInvocation:
1478 """A reconstructed ll-harness invocation extracted from a JSONL record.
1480 Carries the runner kind and raw (un-redacted) input-context text that the
1481 EvalFixture export needs but that ``InvocationEvent`` discards.
1482 """
1484 runner: str # "skill" | "cmd"
1485 target: str # skill name (runner==skill) or full shell command (runner==cmd)
1486 session_id: str
1487 timestamp: str
1488 input_context: str # raw user-message text; "" when none (e.g. Bash invocations)
1491def _extract_eval_invocation(record: dict) -> _EvalInvocation | None:
1492 """Reconstruct a single ll-harness invocation from a JSONL record.
1494 Delegates signal detection to ``_detect_ll_signal`` and wraps the result
1495 in an ``_EvalInvocation`` with ``session_id`` and ``timestamp`` from the
1496 record. Runner mapping: queue-operation/user → "skill"; bash → "cmd"
1497 (target becomes the full command; input_context is "").
1499 Returns None for records that carry no ll invocation signal.
1500 """
1501 sig = _detect_ll_signal(record)
1502 if sig is None:
1503 return None
1504 sid = record.get("sessionId", "")
1505 ts = record.get("timestamp", "")
1506 if sig.runner == "bash":
1507 return _EvalInvocation("cmd", sig.input_context, sid, ts, "")
1508 return _EvalInvocation("skill", sig.tool_name, sid, ts, sig.input_context)
1511def _record_has_error(record: dict) -> bool:
1512 """True if a JSONL record is a ``tool_result`` flagged ``is_error``.
1514 Used as the session-level ``failed`` outcome signal: session logs expose no
1515 output-quality judgment, only execution evidence (ARCHITECTURE-017).
1516 """
1517 if record.get("type") != "user":
1518 return False
1519 message = record.get("message", {})
1520 if not isinstance(message, dict):
1521 return False
1522 content = message.get("content")
1523 if isinstance(content, list):
1524 for block in content:
1525 if (
1526 isinstance(block, dict)
1527 and block.get("type") == "tool_result"
1528 and block.get("is_error")
1529 ):
1530 return True
1531 return False
1534def _classify_outcome(metadata: dict, *, has_error: bool) -> str:
1535 """Map session metadata + error signal to an EvalFixture execution outcome.
1537 Precedence ``failed`` > ``corrected`` > ``accepted``; ``unknown`` when the DB
1538 returned no metadata. Per ARCHITECTURE-017 the taxonomy is EXECUTION, not
1539 output quality. ``metadata`` is the dict from
1540 ``history_reader.lookup_session_metadata`` (``{}`` when the session is absent).
1541 """
1542 if has_error:
1543 return "failed"
1544 if not metadata:
1545 return "unknown"
1546 if metadata.get("has_corrections"):
1547 return "corrected"
1548 return "accepted"
1551def _redact_input_context(text: str) -> tuple[str | None, bool]:
1552 """Best-effort, non-blocking redaction of user-message text.
1554 Applies ``pii.redact_pii`` (email/phone/SSN) then ``_ABS_PATH_RE`` (absolute
1555 paths). Returns ``(redacted_text_or_None, pii_detected)`` where ``pii_detected``
1556 is True when either pass altered the text. Never raises and never drops a
1557 record for unredactable content (ARCHITECTURE-017).
1558 """
1559 if not text:
1560 return None, False
1561 from little_loops.pii import redact_pii
1563 redacted = redact_pii(text)
1564 redacted = _ABS_PATH_RE.sub("<path>", redacted)
1565 return redacted, redacted != text
1568def _build_eval_fixture(inv: _EvalInvocation, outcome: str) -> dict:
1569 """Map a reconstructed invocation + outcome to an EvalFixture v1 record.
1571 Pure function (no I/O) — the mapping core covered by unit tests. Schema and
1572 field semantics per decision ARCHITECTURE-017 in ``.ll/decisions.yaml``: the
1573 fixture replays into ``ll-harness <runner> <target> [runner_args...]
1574 [--exit-code N] [--semantic TEXT] [--timeout S]`` (ll-harness has no loader).
1575 """
1576 input_context, pii_detected = _redact_input_context(inv.input_context)
1577 issue_match = _ISSUE_ID_RE.search(inv.input_context) if inv.input_context else None
1578 issue_id = issue_match.group(0) if issue_match else None
1579 skill_name = inv.target if inv.runner == "skill" else None
1580 return {
1581 "runner": inv.runner,
1582 "target": inv.target,
1583 "session_id": inv.session_id,
1584 "timestamp": inv.timestamp,
1585 "outcome": outcome,
1586 "runner_args": [],
1587 "exit_code": None,
1588 "semantic": None,
1589 "timeout": 120,
1590 "input_context": input_context,
1591 "issue_id": issue_id,
1592 "skill_name": skill_name,
1593 "pii_detected": pii_detected,
1594 }
1597def _fixture_to_harness_argv(fixture: dict) -> list[str]:
1598 """Serialize an EvalFixture record back into an ``ll-harness`` argv.
1600 ll-harness has no fixture loader (ARCHITECTURE-017); a fixture replays by
1601 serializing its fields into the harness CLI arg surface. Used by the
1602 round-trip test to prove every exported fixture is a valid harness invocation.
1603 """
1604 argv: list[str] = [fixture["runner"], fixture["target"]]
1605 argv.extend(fixture.get("runner_args") or [])
1606 if fixture.get("exit_code") is not None:
1607 argv.extend(["--exit-code", str(fixture["exit_code"])])
1608 if fixture.get("semantic") is not None:
1609 argv.extend(["--semantic", str(fixture["semantic"])])
1610 timeout = fixture.get("timeout")
1611 if timeout is not None and timeout != 120:
1612 argv.extend(["--timeout", str(timeout)])
1613 return argv
1616def _cmd_eval_export(args: argparse.Namespace) -> int:
1617 """Export ll-harness eval fixtures reconstructed from session logs (FEAT-1971).
1619 Walks the current project's JSONL logs, reconstructs each ll invocation, sources
1620 an execution outcome from ``history_reader.lookup_session_metadata``, redacts the
1621 input context, and writes EvalFixture v1 records (YAML default, JSON with
1622 ``--json``). Schema + outcome taxonomy: decision ARCHITECTURE-017 in
1623 ``.ll/decisions.yaml``.
1624 """
1625 from little_loops.history_reader import lookup_session_metadata
1627 cwd_path = Path(args.project) if args.project else Path.cwd()
1628 project_folder = get_project_folder(cwd_path)
1629 if project_folder is None:
1630 print(f"No session project folder found for: {cwd_path}", file=sys.stderr)
1631 return 1
1632 db_path = resolve_history_db(cwd_path / ".ll" / "history.db")
1634 # Single JSONL pass: collect raw invocations + per-session error flags together
1635 # (avoids the double-parse the decision warns against).
1636 invocations: list[_EvalInvocation] = []
1637 session_has_error: dict[str, bool] = {}
1638 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
1639 for jsonl_file in jsonl_files:
1640 try:
1641 with open(jsonl_file, encoding="utf-8") as f:
1642 for line in f:
1643 line = line.strip()
1644 if not line:
1645 continue
1646 try:
1647 record = json.loads(line)
1648 except json.JSONDecodeError:
1649 continue
1650 if _record_has_error(record):
1651 sid = record.get("sessionId", "")
1652 if sid:
1653 session_has_error[sid] = True
1654 inv = _extract_eval_invocation(record)
1655 if inv is not None:
1656 invocations.append(inv)
1657 except OSError:
1658 continue
1660 # Stable, deterministic order: by timestamp then session.
1661 invocations.sort(key=lambda e: (e.timestamp, e.session_id))
1663 metadata_cache: dict[str, dict] = {}
1664 fixtures: list[dict] = []
1665 skipped = 0
1666 for inv in invocations:
1667 # --skill: keep only skill-runner invocations of the named target.
1668 if args.skill and not (inv.runner == "skill" and inv.target == args.skill):
1669 continue
1671 if inv.session_id not in metadata_cache:
1672 metadata_cache[inv.session_id] = lookup_session_metadata(inv.session_id, db=db_path)
1673 outcome = _classify_outcome(
1674 metadata_cache[inv.session_id],
1675 has_error=session_has_error.get(inv.session_id, False),
1676 )
1677 # No extractable execution outcome -> skip with a logged count.
1678 if outcome == "unknown":
1679 skipped += 1
1680 continue
1682 fixture = _build_eval_fixture(inv, outcome)
1684 # --issue: match the extracted issue_id or a literal occurrence in target.
1685 if args.issue and args.issue != fixture["issue_id"] and args.issue not in inv.target:
1686 continue
1688 fixtures.append(fixture)
1689 if args.limit and len(fixtures) >= args.limit:
1690 break
1692 if args.json:
1693 output = json.dumps(fixtures, indent=2)
1694 else:
1695 import yaml
1697 output = yaml.safe_dump(fixtures, sort_keys=False, default_flow_style=False)
1699 if args.out:
1700 out_path = Path(args.out)
1701 try:
1702 out_path.parent.mkdir(parents=True, exist_ok=True)
1703 out_path.write_text(output, encoding="utf-8")
1704 except OSError as exc:
1705 print(f"Failed to write {out_path}: {exc}", file=sys.stderr)
1706 return 1
1707 print(f"Wrote {len(fixtures)} fixture(s) to {out_path}", file=sys.stderr)
1708 else:
1709 print(output, end="" if output.endswith("\n") else "\n")
1711 if skipped:
1712 print(
1713 f"Skipped {skipped} invocation(s) with no extractable outcome",
1714 file=sys.stderr,
1715 )
1717 return 0
1720def _build_parser() -> argparse.ArgumentParser:
1721 """Build the argument parser for ll-logs."""
1722 parser = argparse.ArgumentParser(
1723 prog="ll-logs",
1724 description="Discover and extract ll-relevant JSONL entries from Claude Code logs",
1725 formatter_class=argparse.RawDescriptionHelpFormatter,
1726 epilog="""
1727Examples:
1728 %(prog)s discover # List all projects with ll activity
1729 %(prog)s tail --loop <name> # Stream live events from an active loop session
1730 %(prog)s extract --all # Extract all projects to logs/
1731 %(prog)s extract --project /path # Extract one project to logs/<slug>/
1732 %(prog)s extract --all --cmd ll-history # Filter to ll-history invocations
1733""",
1734 )
1736 subparsers = parser.add_subparsers(dest="command", help="Available commands")
1737 discover_parser = subparsers.add_parser(
1738 "discover",
1739 help="List all Claude projects with ll activity (one path per line, sorted)",
1740 )
1741 add_json_arg(discover_parser)
1742 discover_parser.add_argument(
1743 "--existing-only",
1744 action="store_true",
1745 default=False,
1746 help="Only emit paths that currently exist on disk; suppress all diagnostic output.",
1747 )
1749 tail_parser = subparsers.add_parser(
1750 "tail",
1751 help="Stream live events from an active loop session",
1752 )
1753 tail_parser.add_argument("--loop", required=True, metavar="NAME", help="Loop name to tail")
1754 tail_parser.add_argument(
1755 "--project", type=Path, metavar="DIR", help="Project root to tail loops from (default: CWD)"
1756 )
1758 extract_parser = subparsers.add_parser(
1759 "extract",
1760 help="Extract ll-relevant JSONL records to logs/<slug>/<session-id>.jsonl",
1761 )
1762 target_group = extract_parser.add_mutually_exclusive_group(required=True)
1763 target_group.add_argument(
1764 "--project",
1765 type=Path,
1766 metavar="DIR",
1767 help="Working directory of the target project",
1768 )
1769 target_group.add_argument(
1770 "--all",
1771 action="store_true",
1772 help="Extract all projects with ll activity",
1773 )
1774 extract_parser.add_argument(
1775 "--cmd",
1776 metavar="TOOL",
1777 help="Filter to records containing this ll- tool name (e.g. ll-history)",
1778 )
1780 sequences_parser = subparsers.add_parser(
1781 "sequences",
1782 help="Extract tool-chain n-grams of ll invocations from JSONL logs",
1783 )
1784 sequences_target = sequences_parser.add_mutually_exclusive_group(required=True)
1785 sequences_target.add_argument(
1786 "--project",
1787 type=Path,
1788 metavar="DIR",
1789 help="Working directory of the target project",
1790 )
1791 sequences_target.add_argument(
1792 "--all",
1793 action="store_true",
1794 help="Analyze all projects with ll activity",
1795 )
1796 sequences_parser.add_argument(
1797 "--min-len",
1798 type=int,
1799 default=2,
1800 metavar="N",
1801 help="Minimum n-gram length (default: 2)",
1802 )
1803 sequences_parser.add_argument(
1804 "--min-count",
1805 type=int,
1806 default=1,
1807 metavar="M",
1808 help="Minimum occurrence count to include (default: 1)",
1809 )
1810 sequences_parser.add_argument(
1811 "--top",
1812 type=int,
1813 default=None,
1814 metavar="N",
1815 help="Limit output to top N chains by frequency",
1816 )
1817 sequences_parser.add_argument(
1818 "--window-days",
1819 type=int,
1820 default=None,
1821 metavar="D",
1822 help="Only consider records within the last D calendar days",
1823 )
1824 add_json_arg(sequences_parser)
1826 stats_parser = subparsers.add_parser(
1827 "stats",
1828 help="Aggregate skill invocation frequency and correction rate from history.db",
1829 )
1830 stats_target = stats_parser.add_mutually_exclusive_group(required=True)
1831 stats_target.add_argument(
1832 "--project",
1833 type=Path,
1834 metavar="DIR",
1835 help="Working directory of the target project",
1836 )
1837 stats_target.add_argument(
1838 "--all",
1839 action="store_true",
1840 help="Aggregate across all projects with ll activity",
1841 )
1842 stats_parser.add_argument(
1843 "--window-days",
1844 type=int,
1845 default=None,
1846 metavar="D",
1847 help="Only consider records within the last D calendar days",
1848 )
1849 stats_parser.add_argument(
1850 "--sort",
1851 choices=["freq", "corrections"],
1852 default="freq",
1853 help="Sort output by invocation frequency or correction count (default: freq)",
1854 )
1855 add_json_arg(stats_parser)
1857 scan_failures_parser = subparsers.add_parser(
1858 "scan-failures",
1859 help="Mine failed ll-* calls from interactive session logs and propose bug issues",
1860 )
1861 scan_failures_target = scan_failures_parser.add_mutually_exclusive_group(required=True)
1862 scan_failures_target.add_argument(
1863 "--project",
1864 type=Path,
1865 metavar="DIR",
1866 help="Working directory of the target project",
1867 )
1868 scan_failures_target.add_argument(
1869 "--all",
1870 action="store_true",
1871 help="Scan all projects with ll activity",
1872 )
1873 scan_failures_parser.add_argument(
1874 "--window-days",
1875 type=int,
1876 default=None,
1877 metavar="D",
1878 help="Only consider records within the last D calendar days",
1879 )
1880 scan_failures_parser.add_argument(
1881 "--capture",
1882 action="store_true",
1883 help="Create bug issue files for each failure cluster (one per tool+error signature)",
1884 )
1885 scan_failures_parser.add_argument(
1886 "--capture-foreign",
1887 action="store_true",
1888 help="Allow --capture to include failures from projects other than the current directory (only meaningful with --all)",
1889 )
1890 add_json_arg(scan_failures_parser)
1892 dead_skills_parser = subparsers.add_parser(
1893 "dead-skills",
1894 help="List catalog skills/commands with zero or low invocations across the corpus",
1895 )
1896 dead_skills_target = dead_skills_parser.add_mutually_exclusive_group(required=True)
1897 dead_skills_target.add_argument(
1898 "--project",
1899 type=Path,
1900 metavar="DIR",
1901 help="Working directory of the target project (also used as catalog root)",
1902 )
1903 dead_skills_target.add_argument(
1904 "--all",
1905 action="store_true",
1906 help="Aggregate across all projects; catalog loaded from current directory",
1907 )
1908 dead_skills_parser.add_argument(
1909 "--window-days",
1910 type=int,
1911 default=None,
1912 metavar="D",
1913 help="Only consider records within the last D calendar days",
1914 )
1915 dead_skills_parser.add_argument(
1916 "--threshold",
1917 type=int,
1918 default=3,
1919 metavar="N",
1920 help="Skills with invocations <= N are 'rarely' invoked (default: 3)",
1921 )
1922 add_json_arg(dead_skills_parser)
1924 diff_parser = subparsers.add_parser(
1925 "diff",
1926 help="Compare two sessions' ll-invocation behavior (skills, sequences, counts)",
1927 )
1928 diff_parser.add_argument(
1929 "session_a", metavar="SESSION_A", help="First session ID or JSONL file path"
1930 )
1931 diff_parser.add_argument(
1932 "session_b", metavar="SESSION_B", help="Second session ID or JSONL file path"
1933 )
1934 add_json_arg(diff_parser)
1936 eval_export_parser = subparsers.add_parser(
1937 "eval-export",
1938 help="Export eval fixtures from ll-harness session logs",
1939 )
1940 eval_export_parser.add_argument(
1941 "--project",
1942 type=Path,
1943 metavar="DIR",
1944 help="Project working directory (default: current directory)",
1945 )
1946 eval_export_parser.add_argument(
1947 "--skill",
1948 metavar="NAME",
1949 help="Filter by skill name",
1950 )
1951 eval_export_parser.add_argument(
1952 "--issue",
1953 metavar="ID",
1954 help="Filter by issue ID in session context",
1955 )
1956 eval_export_parser.add_argument(
1957 "--limit",
1958 type=int,
1959 default=0,
1960 metavar="N",
1961 help="Cap output records (0 = unlimited)",
1962 )
1963 eval_export_parser.add_argument(
1964 "--out",
1965 metavar="PATH",
1966 help="Write output to file (default: stdout)",
1967 )
1968 add_json_arg(eval_export_parser, help_text="JSON output instead of YAML (default: YAML)")
1970 return parser
1973def _parse_args() -> argparse.Namespace:
1974 """Parse command-line arguments. Exposed for testing."""
1975 return _build_parser().parse_args()
1978def main_logs() -> int:
1979 """Entry point for ll-logs command.
1981 Returns:
1982 0 on success, 1 when no subcommand given or on error.
1983 """
1984 with cli_event_context(DEFAULT_DB_PATH, "ll-logs", sys.argv[1:]):
1985 configure_output()
1986 logger = Logger(use_color=use_color_enabled())
1988 parser = _build_parser()
1989 args = parser.parse_args()
1991 if not args.command:
1992 parser.print_help()
1993 return 1
1995 if args.command == "discover":
1996 projects = discover_all_projects(logger, existing_only=args.existing_only)
1997 if args.json:
1998 print_json({"paths": [str(p) for p in projects]})
1999 else:
2000 for path in projects:
2001 print(path)
2002 return 0
2004 if args.command == "tail":
2005 project_root = args.project if args.project else Path.cwd()
2006 config = BRConfig(project_root)
2007 loops_dir = Path(config.loops.loops_dir)
2008 return _cmd_tail(args, loops_dir)
2010 if args.command == "extract":
2011 return _cmd_extract(args, logger)
2013 if args.command == "sequences":
2014 return _cmd_sequences(args, logger)
2016 if args.command == "stats":
2017 return _cmd_stats(args, logger)
2019 if args.command == "scan-failures":
2020 return _cmd_scan_failures(args, logger)
2022 if args.command == "dead-skills":
2023 return _cmd_dead_skills(args, logger)
2025 if args.command == "diff":
2026 return _cmd_diff(args, logger)
2028 if args.command == "eval-export":
2029 return _cmd_eval_export(args)
2031 return 1