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