Coverage for little_loops / cli / logs.py: 10%
1126 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:30 -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, *, window_days: int | 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 window_days: If set, filter records to within N days of the latest record.
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 # Collect all raw events first for window-days filtering
245 all_events: list[InvocationEvent] = []
246 latest_ts: str | None = None
248 for jsonl_file in jsonl_files:
249 try:
250 with open(jsonl_file, encoding="utf-8") as f:
251 for line in f:
252 line = line.strip()
253 if not line:
254 continue
255 try:
256 record = json.loads(line)
257 except json.JSONDecodeError:
258 continue
260 tool_name = _extract_tool_name(record)
261 if tool_name is None:
262 continue
264 ts = record.get("timestamp", "")
265 sid = record.get("sessionId", "")
267 evt = InvocationEvent(tool_name=tool_name, timestamp=ts, session_id=sid)
268 all_events.append(evt)
270 if ts and (latest_ts is None or ts > latest_ts):
271 latest_ts = ts
272 except OSError:
273 continue
275 # Apply window-days filter
276 if window_days is not None and latest_ts:
277 cutoff = _parse_iso_timestamp(latest_ts) - timedelta(days=window_days)
278 all_events = [e for e in all_events if _parse_iso_timestamp(e.timestamp) >= cutoff]
280 # Bucket by session and sort
281 for evt in all_events:
282 events_by_session.setdefault(evt.session_id, []).append(evt)
284 for session_id in events_by_session:
285 events_by_session[session_id].sort(key=lambda e: e.timestamp)
287 return events_by_session
290def _extract_tool_name(record: dict) -> str | None:
291 """Extract the ll tool/skill name from a JSONL record.
293 Detects three signal types (matching ``_is_ll_relevant``):
294 (a) queue-operation enqueue with ``/ll:<name>`` → skill name
295 (b) user records with ``<command-name>/ll:<name>`` → skill name
296 (c) assistant Bash tool-use invoking ``ll-<tool>`` → CLI tool name
297 """
298 record_type = record.get("type")
300 # (a) queue-operation enqueue
301 if record_type == "queue-operation" and record.get("operation") == "enqueue":
302 content = record.get("content", "")
303 if isinstance(content, str) and content.startswith("/ll:"):
304 m = _QUEUE_SKILL_RE.match(content)
305 if m:
306 return m.group(1)
308 # (b) user records with <command-name>/ll: pattern
309 if record_type == "user":
310 message = record.get("message", {})
311 if not isinstance(message, dict):
312 return None
313 content = message.get("content")
314 text = ""
315 if isinstance(content, str):
316 text = content
317 elif isinstance(content, list):
318 for block in content:
319 if isinstance(block, dict):
320 text = block.get("text", "")
321 if text:
322 break
323 if text:
324 m = _COMMAND_NAME_SKILL_RE.search(text)
325 if m:
326 # Extract skill name, stripping trailing </command-name> if present
327 name = m.group(1)
328 if name.endswith("</command-name>"):
329 name = name[: -len("</command-name>")]
330 return name
332 # (c) assistant Bash tool-use invoking ll-<tool>
333 if record_type == "assistant":
334 message = record.get("message", {})
335 content = message.get("content", [])
336 if isinstance(content, list):
337 for block in content:
338 if (
339 isinstance(block, dict)
340 and block.get("type") == "tool_use"
341 and block.get("name") == "Bash"
342 ):
343 cmd = block.get("input", {}).get("command", "")
344 m = _LL_BASH_RE.search(cmd)
345 if m:
346 return m.group(1)
348 return None
351def _parse_iso_timestamp(ts: str) -> datetime:
352 """Parse an ISO 8601 timestamp string to a timezone-aware datetime.
354 Handles both ``Z``-suffixed and ``+00:00`` offset formats. Returns
355 ``datetime.min`` with UTC tzinfo for unparseable input.
356 """
357 if not ts:
358 return datetime.min.replace(tzinfo=UTC)
359 try:
360 # Handle Z suffix
361 if ts.endswith("Z"):
362 ts = ts[:-1] + "+00:00"
363 dt = datetime.fromisoformat(ts)
364 if dt.tzinfo is None:
365 dt = dt.replace(tzinfo=UTC)
366 return dt
367 except (ValueError, TypeError):
368 return datetime.min.replace(tzinfo=UTC)
371@dataclass
372class Edge:
373 """A transition edge within an n-gram chain."""
375 from_: str
376 to: str
377 freq: float
380@dataclass
381class ChainResult:
382 """An n-gram chain with occurrence count and per-edge transition frequencies."""
384 chain: list[str]
385 count: int
386 edges: list[Edge]
388 def to_dict(self) -> dict:
389 return {
390 "chain": self.chain,
391 "count": self.count,
392 "edges": [{"from": e.from_, "to": e.to, "freq": e.freq} for e in self.edges],
393 }
396def _count_ngrams(
397 events_by_session: dict[str, list[InvocationEvent]],
398 min_len: int = 2,
399) -> Counter:
400 """Count n-grams across per-session event streams.
402 Args:
403 events_by_session: Per-session ordered event streams.
404 min_len: Minimum n-gram length (window size).
406 Returns:
407 Counter mapping ``(tool_1, tool_2, ...)`` tuples to occurrence counts.
408 """
409 counter: Counter = Counter()
410 for events in events_by_session.values():
411 names = [e.tool_name for e in events]
412 # For chains shorter than min_len, still consider them as single n-grams
413 # but note: min_len is the minimum, so we use range(min_len, len(names)+1)
414 for n in range(min_len, len(names) + 1):
415 for i in range(len(names) - n + 1):
416 ngram = tuple(names[i : i + n])
417 counter[ngram] += 1
418 return counter
421def _build_chain_results(
422 counter: Counter,
423 min_count: int = 1,
424 top: int | None = None,
425) -> list[ChainResult]:
426 """Build ranked ``ChainResult`` list from n-gram counter.
428 Args:
429 counter: n-gram counter from ``_count_ngrams``.
430 min_count: Minimum occurrence count to include.
431 top: If set, limit to top N chains by frequency.
433 Returns:
434 List of ``ChainResult`` sorted by count descending.
435 """
436 results: list[ChainResult] = []
437 for ngram, count in counter.most_common():
438 if count < min_count:
439 continue
440 edges = _compute_edges(ngram, counter)
441 results.append(ChainResult(chain=list(ngram), count=count, edges=edges))
443 if top is not None:
444 results = results[:top]
446 return results
449def _compute_edges(ngram: tuple[str, ...], counter: Counter) -> list[Edge]:
450 """Compute per-edge transition frequencies for an n-gram chain.
452 For each adjacent pair ``(from, to)`` in the chain, computes the frequency
453 as the proportion of times ``from → to`` appears out of all transitions
454 originating from ``from`` across the entire corpus.
455 """
456 edges: list[Edge] = []
457 # Build a transition counter for all pairs in the corpus
458 all_transitions: Counter = Counter()
459 out_degree: Counter = Counter()
460 for ngram_key, count in counter.items():
461 for i in range(len(ngram_key) - 1):
462 pair = (ngram_key[i], ngram_key[i + 1])
463 all_transitions[pair] += count
464 out_degree[ngram_key[i]] += count
466 for i in range(len(ngram) - 1):
467 from_ = ngram[i]
468 to = ngram[i + 1]
469 pair = (from_, to)
470 total_out = out_degree.get(from_, 0)
471 freq = all_transitions.get(pair, 0) / total_out if total_out > 0 else 0.0
472 edges.append(Edge(from_=from_, to=to, freq=round(freq, 4)))
474 return edges
477def _cmd_sequences(args: argparse.Namespace, logger: Logger) -> int:
478 """Extract n-grams of ll invocations from JSONL log files."""
479 if args.project:
480 cwd_path: Path = args.project
481 project_folder = get_project_folder(cwd_path)
482 if project_folder is None:
483 logger.error(f"No session project folder found for: {cwd_path}")
484 return 1
485 project_items = [(cwd_path, project_folder)]
486 else:
487 decoded_paths = discover_all_projects(logger)
488 project_items = []
489 for decoded_path in decoded_paths:
490 folder = get_project_folder(decoded_path)
491 if folder is not None:
492 project_items.append((decoded_path, folder))
494 # Aggregate events across all projects
495 all_events: dict[str, list[InvocationEvent]] = {}
496 for _cwd_path, project_folder in project_items:
497 events = _extract_ll_event_streams(project_folder, window_days=args.window_days)
498 for sid, evt_list in events.items():
499 all_events.setdefault(sid, []).extend(evt_list)
501 # Sort each session's events by timestamp
502 for sid in all_events:
503 all_events[sid].sort(key=lambda e: e.timestamp)
505 # Count n-grams
506 counter = _count_ngrams(all_events, min_len=args.min_len)
507 results = _build_chain_results(counter, min_count=args.min_count, top=args.top)
509 if args.json:
510 print_json([r.to_dict() for r in results])
511 else:
512 if not results:
513 print("No sequences found.")
514 return 0
516 # Print ranked table
517 for rank, r in enumerate(results, 1):
518 chain_str = " → ".join(r.chain)
519 print(f"{rank}. [{r.count}] {chain_str}")
520 for edge in r.edges:
521 print(f" {edge.from_} → {edge.to}: {edge.freq:.4f}")
523 return 0
526def generate_index(logs_dir: Path) -> None:
527 """Generate logs/index.md summarising extracted projects."""
528 rows = []
530 if logs_dir.exists():
531 for subdir in sorted(logs_dir.iterdir()):
532 if not subdir.is_dir():
533 continue
535 jsonl_files = [f for f in subdir.glob("*.jsonl") if not f.name.startswith("agent-")]
536 if not jsonl_files:
537 continue
539 timestamps: list[str] = []
540 for jsonl_file in jsonl_files:
541 try:
542 with open(jsonl_file, encoding="utf-8") as f:
543 for line in f:
544 line = line.strip()
545 if not line:
546 continue
547 try:
548 record = json.loads(line)
549 except json.JSONDecodeError:
550 continue
551 ts = record.get("timestamp")
552 if ts:
553 timestamps.append(ts)
554 except OSError:
555 continue
557 if timestamps:
558 earliest = min(timestamps)[:10]
559 latest = max(timestamps)[:10]
560 date_range = f"{earliest} – {latest}" if earliest != latest else earliest
561 else:
562 date_range = ""
564 rows.append((subdir.name, len(jsonl_files), date_range))
566 lines = ["# Logs Index", ""]
567 if rows:
568 lines.append("| Project | Sessions | Date Range |")
569 lines.append("|---------|----------|------------|")
570 for name, count, date_range in rows:
571 lines.append(f"| {name} | {count} | {date_range} |")
572 else:
573 lines.append("*No projects extracted yet.*")
574 lines.append("")
576 logs_dir.mkdir(parents=True, exist_ok=True)
577 (logs_dir / "index.md").write_text("\n".join(lines), encoding="utf-8")
580def _cmd_extract(args: argparse.Namespace, logger: Logger) -> int:
581 """Extract ll-relevant JSONL records to logs/<slug>/<session-id>.jsonl."""
582 if args.project:
583 cwd_path: Path = args.project
584 project_folder = get_project_folder(cwd_path)
585 if project_folder is None:
586 logger.error(f"No session project folder found for: {cwd_path}")
587 return 1
588 project_items = [(cwd_path, project_folder)]
589 else:
590 decoded_paths = discover_all_projects(logger)
591 project_items = []
592 for decoded_path in decoded_paths:
593 folder = get_project_folder(decoded_path)
594 if folder is not None:
595 project_items.append((decoded_path, folder))
597 for cwd_path, project_folder in project_items:
598 slug = cwd_path.resolve().name
599 buckets: dict[str, list[dict]] = {}
601 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
602 for jsonl_file in jsonl_files:
603 try:
604 with open(jsonl_file, encoding="utf-8") as f:
605 for line in f:
606 line = line.strip()
607 if not line:
608 continue
609 try:
610 record = json.loads(line)
611 except json.JSONDecodeError:
612 continue
613 if _is_ll_relevant(record):
614 session_id = record.get("sessionId", "")
615 buckets.setdefault(session_id, []).append(record)
616 except OSError:
617 continue
619 if args.cmd:
620 filtered: dict[str, list[dict]] = {}
621 for session_id, records in buckets.items():
622 matching = [r for r in records if _cmd_matches(r, args.cmd)]
623 if matching:
624 filtered[session_id] = matching
625 buckets = filtered
627 out_base = Path.cwd() / "logs" / slug
628 for session_id, records in buckets.items():
629 out_file = out_base / f"{session_id}.jsonl"
630 out_file.parent.mkdir(parents=True, exist_ok=True)
631 with open(out_file, "w", encoding="utf-8") as f:
632 for record in records:
633 f.write(json.dumps(record) + "\n")
635 generate_index(Path.cwd() / "logs")
636 return 0
639def _cmd_tail(args: argparse.Namespace, loops_dir: Path) -> int:
640 """Stream live events from an active loop session."""
641 events_file = loops_dir / ".running" / f"{args.loop}.events.jsonl"
643 if not events_file.exists():
644 print(f"No active session for loop '{args.loop}'", file=sys.stderr)
645 return 1
647 width = shutil.get_terminal_size().columns
648 try:
649 with open(events_file, encoding="utf-8") as f:
650 f.seek(0, 2)
651 while True:
652 line = f.readline()
653 if line:
654 line = line.strip()
655 if line:
656 try:
657 event = json.loads(line)
658 except json.JSONDecodeError:
659 continue
660 formatted = _format_history_event(event, verbose=False, width=width)
661 if formatted is not None:
662 print(formatted)
663 else:
664 time.sleep(0.1)
665 except KeyboardInterrupt:
666 return 0
668 return 0
671_CORRECTION_WINDOW_SEC = 30
674def _aggregate_skill_stats(
675 db_path: Path,
676 *,
677 window_days: int | None = None,
678) -> dict[str, dict[str, int]] | None:
679 """Aggregate per-skill invocation and correction counts from history.db.
681 Returns None when the database is absent, or an empty dict when the database
682 has no skill_events rows. Corrections are attributed to the most recent skill
683 event in the same session within _CORRECTION_WINDOW_SEC seconds.
684 """
685 if not db_path.exists():
686 return None
688 conn = sqlite3.connect(str(db_path))
689 conn.row_factory = sqlite3.Row
690 try:
691 try:
692 skill_rows = conn.execute(
693 "SELECT ts, session_id, skill_name FROM skill_events ORDER BY ts"
694 ).fetchall()
695 except sqlite3.OperationalError:
696 return None
698 if not skill_rows:
699 return {}
701 if window_days is not None:
702 latest_ts = skill_rows[-1]["ts"] or ""
703 cutoff = _parse_iso_timestamp(latest_ts) - timedelta(days=window_days)
704 skill_rows = [r for r in skill_rows if _parse_iso_timestamp(r["ts"] or "") >= cutoff]
706 stats: dict[str, dict[str, int]] = defaultdict(lambda: {"invocations": 0, "corrections": 0})
707 for row in skill_rows:
708 stats[row["skill_name"] or "unknown"]["invocations"] += 1
710 session_skills: dict[str, list[tuple[str, str]]] = defaultdict(list)
711 for row in skill_rows:
712 sid = row["session_id"] or ""
713 session_skills[sid].append((row["ts"] or "", row["skill_name"] or "unknown"))
715 try:
716 corr_rows = conn.execute(
717 "SELECT ts, session_id FROM user_corrections ORDER BY ts"
718 ).fetchall()
719 except sqlite3.OperationalError:
720 corr_rows = []
722 for corr in corr_rows:
723 c_ts = corr["ts"] or ""
724 sid = corr["session_id"] or ""
725 candidates = session_skills.get(sid, [])
726 best_skill: str | None = None
727 best_ts: str = ""
728 for s_ts, s_name in candidates:
729 if s_ts <= c_ts and s_ts >= best_ts:
730 best_ts = s_ts
731 best_skill = s_name
732 if best_skill is not None:
733 elapsed = (
734 _parse_iso_timestamp(c_ts) - _parse_iso_timestamp(best_ts)
735 ).total_seconds()
736 if 0 <= elapsed <= _CORRECTION_WINDOW_SEC:
737 stats[best_skill]["corrections"] += 1
739 return dict(stats)
740 finally:
741 conn.close()
744def _load_catalog_names(root_dir: Path) -> set[str]:
745 """Load normalized skill/command names from skills/ and commands/ under root_dir.
747 Excludes bridge skills (containing BRIDGE_MARKER) and skills/commands with
748 disable-model-invocation: true. Normalizes names by stripping the "ll-" prefix
749 so catalog names match the skill_events.skill_name recording convention.
750 """
751 import yaml
753 names: set[str] = set()
755 skills_dir = root_dir / "skills"
756 if skills_dir.is_dir():
757 for skill_md in sorted(skills_dir.glob("*/SKILL.md")):
758 try:
759 text = skill_md.read_text()
760 except OSError:
761 continue
762 if BRIDGE_MARKER in text:
763 continue
764 name: str = skill_md.parent.name
765 if text.startswith("---"):
766 end = text.find("---", 3)
767 if end != -1:
768 try:
769 fm = yaml.safe_load(text[3:end]) or {}
770 except yaml.YAMLError:
771 fm = {}
772 if isinstance(fm, dict):
773 dmi = fm.get("disable-model-invocation")
774 if (isinstance(dmi, bool) and dmi) or (
775 isinstance(dmi, str) and dmi.strip().lower() in ("true", "yes", "1")
776 ):
777 continue
778 name = str(fm.get("name") or name)
779 if name.startswith("ll-"):
780 name = name[3:]
781 if name:
782 names.add(name)
784 commands_dir = root_dir / "commands"
785 if commands_dir.is_dir():
786 for cmd_md in sorted(commands_dir.glob("*.md")):
787 stem = cmd_md.stem
788 try:
789 text = cmd_md.read_text()
790 except OSError:
791 names.add(stem)
792 continue
793 if text.startswith("---"):
794 end = text.find("---", 3)
795 if end != -1:
796 try:
797 fm = yaml.safe_load(text[3:end]) or {}
798 except yaml.YAMLError:
799 fm = {}
800 if isinstance(fm, dict):
801 dmi = fm.get("disable-model-invocation")
802 if (isinstance(dmi, bool) and dmi) or (
803 isinstance(dmi, str) and dmi.strip().lower() in ("true", "yes", "1")
804 ):
805 continue
806 names.add(stem)
808 return names
811def _cmd_dead_skills(args: argparse.Namespace, logger: Logger) -> int:
812 """List catalog skills/commands never or rarely invoked within the window."""
813 if args.project:
814 db_paths = [Path(args.project) / ".ll" / "history.db"]
815 catalog_root = Path(args.project)
816 else:
817 decoded_paths = discover_all_projects(logger)
818 db_paths = [p / ".ll" / "history.db" for p in decoded_paths]
819 catalog_root = Path.cwd()
821 merged: dict[str, int] = defaultdict(int)
822 for db_path in db_paths:
823 result = _aggregate_skill_stats(db_path, window_days=args.window_days)
824 if result is None:
825 continue
826 for skill, counts in result.items():
827 merged[skill] += counts["invocations"]
829 catalog_names = _load_catalog_names(catalog_root)
830 if not catalog_names:
831 logger.warning(
832 "No catalog skills found — run from an ll project root with skills/ directory."
833 )
834 return 0
836 threshold = args.threshold
837 rows = []
838 for name in sorted(catalog_names):
839 count = merged.get(name, 0)
840 if count == 0:
841 rows.append({"skill": name, "invocations": 0, "tier": "never"})
842 elif count <= threshold:
843 rows.append({"skill": name, "invocations": count, "tier": "rarely"})
845 if args.json:
846 print_json(rows)
847 return 0
849 if not rows:
850 print("No dead or rarely-invoked skills found.")
851 return 0
853 headers = ["Skill", "Invocations", "Tier"]
854 table_rows = [[str(r["skill"]), str(r["invocations"]), str(r["tier"])] for r in rows]
855 print(table(headers, table_rows))
856 return 0
859def _load_cli_allowlist(root: Path) -> frozenset[str]:
860 """Return ll-* CLI names from [project.scripts] in scripts/pyproject.toml.
862 Returns an empty frozenset if the file cannot be read; the allowlist check
863 is skipped when the set is empty so fallback behavior is open (no filtering).
864 """
865 import tomllib
867 pyproject = root / "scripts" / "pyproject.toml"
868 try:
869 with open(pyproject, "rb") as f:
870 data = tomllib.load(f)
871 except (OSError, ValueError):
872 return frozenset()
873 scripts = data.get("project", {}).get("scripts", {})
874 return frozenset(k for k in scripts if k.startswith("ll-"))
877def _is_content_free_error(error_text: str) -> bool:
878 """Return True if error_text carries no signal beyond a bare exit code."""
879 return bool(_CONTENT_FREE_RE.match(error_text.strip()))
882_STACK_FRAME_RE = re.compile(r'\s*File "[^"]+", line \d+[^\n]*')
883_ABS_PATH_RE = re.compile(r"/(?:[^\s,;\"']+/)+[^\s,;\"']+")
884_LINE_NUM_RE = re.compile(r"\bline \d+\b")
885_LL_VERIFY_RE = re.compile(r"^ll-verify-\w+")
888def _normalize_error_sig(text: str) -> str:
889 """Strip volatile parts (paths, line numbers, stack frames) from error text.
891 Returns a stable string suitable as a cluster key.
892 """
893 text = _STACK_FRAME_RE.sub("", text)
894 text = _ABS_PATH_RE.sub("<path>", text)
895 text = _LINE_NUM_RE.sub("line N", text)
896 return re.sub(r"\s+", " ", text).strip()[:300]
899def _extract_error_text(content: object) -> str:
900 """Extract plain text from a tool_result content field (string or list of text blocks)."""
901 if isinstance(content, str):
902 return content
903 if isinstance(content, list):
904 parts: list[str] = []
905 for item in content:
906 if isinstance(item, dict):
907 text = item.get("text", "")
908 if isinstance(text, str):
909 parts.append(text)
910 return "\n".join(parts)
911 return ""
914@dataclass
915class _FailureCluster:
916 """Aggregated failure cluster keyed on (cwd_path, tool_name, normalized_sig)."""
918 tool_name: str
919 normalized_sig: str
920 count: int
921 sample_error: str
922 session_ids: list[str]
923 cwd_path: Path = field(default_factory=lambda: Path("."))
926def _cmd_scan_failures(args: argparse.Namespace, logger: Logger) -> int:
927 """Mine failed ll-* Bash calls from interactive session JSONL logs."""
928 from little_loops.issue_lifecycle import FailureType, classify_failure
930 _cli_allowlist = _load_cli_allowlist(Path.cwd())
932 if args.project:
933 cwd_path: Path = args.project
934 project_folder = get_project_folder(cwd_path)
935 if project_folder is None:
936 logger.error(f"No session project folder found for: {cwd_path}")
937 return 1
938 project_items = [(cwd_path, project_folder)]
939 else:
940 decoded_paths = discover_all_projects(logger)
941 project_items = []
942 for decoded_path in decoded_paths:
943 folder = get_project_folder(decoded_path)
944 if folder is not None:
945 project_items.append((decoded_path, folder))
947 # raw_clusters maps (cwd_path, tool_name, normalized_sig) -> (count, sample_error, session_ids, latest_ts)
948 raw_clusters: dict[tuple[Path, str, str], tuple[int, str, list[str], str]] = {}
949 latest_ts_overall: str = ""
951 for _cwd_path, project_folder in project_items:
952 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
954 for jsonl_file in jsonl_files:
955 try:
956 with open(jsonl_file, encoding="utf-8") as f:
957 lines = f.readlines()
958 except OSError:
959 continue
961 # pending maps tool_use_id -> (ll_tool_name, timestamp)
962 pending: dict[str, tuple[str, str]] = {}
964 for line in lines:
965 line = line.strip()
966 if not line:
967 continue
968 try:
969 record = json.loads(line)
970 except json.JSONDecodeError:
971 continue
973 record_type = record.get("type")
974 ts = record.get("timestamp", "")
975 session_id = record.get("sessionId", "")
977 if ts and ts > latest_ts_overall:
978 latest_ts_overall = ts
980 if record_type == "assistant":
981 message = record.get("message", {})
982 content = message.get("content", [])
983 if not isinstance(content, list):
984 continue
985 for block in content:
986 if not isinstance(block, dict):
987 continue
988 if block.get("type") != "tool_use" or block.get("name") != "Bash":
989 continue
990 cmd = block.get("input", {}).get("command", "")
991 m = _LL_BASH_RE.search(cmd)
992 if not m:
993 continue
994 tool_name = m.group(1)
995 # Skip tokens that are not real ll CLIs (e.g. ll-labs, ll-marketing)
996 if _cli_allowlist and tool_name not in _cli_allowlist:
997 continue
998 block_id = block.get("id", "")
999 if block_id:
1000 pending[block_id] = (tool_name, ts)
1002 elif record_type == "user":
1003 message = record.get("message", {})
1004 content = message.get("content", [])
1005 if not isinstance(content, list) or not content:
1006 continue
1007 for block in content:
1008 if not isinstance(block, dict):
1009 continue
1010 if block.get("type") != "tool_result":
1011 continue
1012 tool_use_id = block.get("tool_use_id", "")
1013 if tool_use_id not in pending:
1014 continue
1015 tool_name, _invoke_ts = pending.pop(tool_use_id)
1017 # Skip ll-verify-* tools — exit 1 is expected gate behavior
1018 if _LL_VERIFY_RE.match(tool_name):
1019 continue
1021 is_error_flag = block.get("is_error") is True
1022 raw_content = block.get("content", "")
1023 error_text = _extract_error_text(raw_content)
1024 has_traceback = "Traceback (most recent call last)" in error_text
1026 if not (is_error_flag or has_traceback):
1027 continue
1029 returncode = 1 if is_error_flag else 0
1030 failure_type, _reason = classify_failure(error_text, returncode)
1031 if failure_type == FailureType.TRANSIENT:
1032 continue
1034 normalized_sig = _normalize_error_sig(error_text)
1035 key = (_cwd_path, tool_name, normalized_sig)
1037 if key in raw_clusters:
1038 cnt, sample, sids, _lts = raw_clusters[key]
1039 if session_id not in sids:
1040 sids.append(session_id)
1041 raw_clusters[key] = (cnt + 1, sample, sids, ts)
1042 else:
1043 raw_clusters[key] = (1, error_text[:500], [session_id], ts)
1045 # Apply window-days filter using the latest seen timestamp as anchor
1046 if args.window_days is not None and latest_ts_overall:
1047 cutoff = _parse_iso_timestamp(latest_ts_overall) - timedelta(days=args.window_days)
1048 raw_clusters = {
1049 k: v for k, v in raw_clusters.items() if _parse_iso_timestamp(v[3]) >= cutoff
1050 }
1052 # Drop content-free clusters (bare "Exit code N" with no error body)
1053 raw_clusters = {k: v for k, v in raw_clusters.items() if not _is_content_free_error(v[1])}
1055 clusters: list[_FailureCluster] = sorted(
1056 [
1057 _FailureCluster(
1058 tool_name=k[1],
1059 normalized_sig=k[2],
1060 count=v[0],
1061 sample_error=v[1],
1062 session_ids=v[2],
1063 cwd_path=k[0],
1064 )
1065 for k, v in raw_clusters.items()
1066 ],
1067 key=lambda c: c.count,
1068 reverse=True,
1069 )
1071 if not clusters:
1072 if not args.json:
1073 print("No ll-* failures found.")
1074 else:
1075 print_json([])
1076 return 0
1078 if args.capture:
1079 capture_foreign = getattr(args, "capture_foreign", False)
1080 return _capture_failure_clusters(clusters, logger, capture_foreign=capture_foreign)
1082 if args.json:
1083 print_json(
1084 [
1085 {
1086 "tool": c.tool_name,
1087 "count": c.count,
1088 "normalized_sig": c.normalized_sig,
1089 "sample_error": c.sample_error,
1090 "session_ids": c.session_ids,
1091 }
1092 for c in clusters
1093 ]
1094 )
1095 return 0
1097 for c in clusters:
1098 print(f"[{c.count}x] {c.tool_name}")
1099 print(f" Sessions: {', '.join(c.session_ids[:5])}")
1100 for sl in c.sample_error.splitlines()[:5]:
1101 print(f" {sl}")
1102 print()
1104 return 0
1107def _capture_failure_clusters(
1108 clusters: list[_FailureCluster], logger: Logger, capture_foreign: bool = False
1109) -> int:
1110 """Create bug issue files for each distinct failure cluster (--capture mode)."""
1111 from little_loops.issue_lifecycle import create_issue_from_failure
1112 from little_loops.issue_parser import IssueInfo
1114 config = BRConfig(Path.cwd())
1115 current_project = Path.cwd().resolve()
1116 created = 0
1117 skipped_foreign = 0
1119 for c in clusters:
1120 if not capture_foreign and c.cwd_path.resolve() != current_project:
1121 skipped_foreign += 1
1122 continue
1123 stub_info = IssueInfo(
1124 path=Path(f"cli/{c.tool_name}"),
1125 issue_type="bugs",
1126 priority="P1",
1127 issue_id=c.tool_name,
1128 title=f"Tool failure in {c.tool_name}",
1129 )
1130 result = create_issue_from_failure(c.sample_error, stub_info, config, logger)
1131 if result is not None:
1132 logger.info(f"Created: {result.name}")
1133 created += 1
1135 if skipped_foreign:
1136 logger.info(
1137 f"Skipped {skipped_foreign} cluster(s) from other projects "
1138 "(use --capture-foreign to include them)."
1139 )
1140 logger.info(f"Captured {created} failure cluster(s) as bug issues.")
1141 return 0
1144def _cmd_stats(args: argparse.Namespace, logger: Logger) -> int:
1145 """Aggregate skill invocation frequency and correction rate from history.db."""
1146 if args.project:
1147 db_paths = [Path(args.project) / ".ll" / "history.db"]
1148 else:
1149 decoded_paths = discover_all_projects(logger)
1150 db_paths = [p / ".ll" / "history.db" for p in decoded_paths]
1152 merged: dict[str, dict[str, int]] = defaultdict(lambda: {"invocations": 0, "corrections": 0})
1153 found_any_db = False
1154 for db_path in db_paths:
1155 result = _aggregate_skill_stats(db_path, window_days=args.window_days)
1156 if result is None:
1157 continue
1158 found_any_db = True
1159 for skill, counts in result.items():
1160 merged[skill]["invocations"] += counts["invocations"]
1161 merged[skill]["corrections"] += counts["corrections"]
1163 if not merged:
1164 if not found_any_db:
1165 logger.warning("No history.db found — run with an active ll project.")
1166 else:
1167 print("No skill events recorded yet.")
1168 return 0
1170 sort_key = getattr(args, "sort", "freq")
1171 if sort_key == "corrections":
1172 ranked = sorted(merged.items(), key=lambda kv: kv[1]["corrections"], reverse=True)
1173 else:
1174 ranked = sorted(merged.items(), key=lambda kv: kv[1]["invocations"], reverse=True)
1176 if args.json:
1177 rows_json = [
1178 {
1179 "skill": skill,
1180 "invocations": counts["invocations"],
1181 "corrections": counts["corrections"],
1182 "correction_rate": (
1183 round(counts["corrections"] / counts["invocations"], 4)
1184 if counts["invocations"] > 0
1185 else 0.0
1186 ),
1187 "errors": None,
1188 "error_rate": None,
1189 }
1190 for skill, counts in ranked
1191 ]
1192 print_json(rows_json)
1193 return 0
1195 headers = ["Skill", "Invocations", "Corrections", "Corr%", "Errors"]
1196 rows = []
1197 for skill, counts in ranked:
1198 inv = counts["invocations"]
1199 corr = counts["corrections"]
1200 corr_pct = f"{corr / inv * 100:.1f}%" if inv > 0 else "0.0%"
1201 rows.append([skill, str(inv), str(corr), corr_pct, "N/A"])
1203 print(table(headers, rows))
1204 return 0
1207def _resolve_session_log(session_ref: str, db_path: Path) -> Path | None:
1208 """Resolve a session reference (session ID or JSONL path) to a JSONL file path.
1210 Tries in order:
1211 1. Direct file path if the ref resolves to an existing ``.jsonl`` file
1212 2. DB lookup of ``session_id → jsonl_path`` in the sessions table
1213 Returns None if unresolvable.
1214 """
1215 candidate = Path(session_ref)
1216 if candidate.suffix == ".jsonl" and candidate.exists():
1217 return candidate
1219 if db_path.exists():
1220 conn = sqlite3.connect(str(db_path))
1221 conn.row_factory = sqlite3.Row
1222 try:
1223 row = conn.execute(
1224 "SELECT jsonl_path FROM sessions WHERE session_id = ?",
1225 (session_ref,),
1226 ).fetchone()
1227 if row and row["jsonl_path"]:
1228 return Path(row["jsonl_path"])
1229 except sqlite3.OperationalError:
1230 pass
1231 finally:
1232 conn.close()
1234 return None
1237def _events_from_jsonl(jsonl_path: Path) -> list[InvocationEvent]:
1238 """Extract ll invocation events from a single JSONL file, sorted by timestamp."""
1239 events: list[InvocationEvent] = []
1240 try:
1241 with open(jsonl_path, encoding="utf-8") as f:
1242 for line in f:
1243 line = line.strip()
1244 if not line:
1245 continue
1246 try:
1247 record = json.loads(line)
1248 except json.JSONDecodeError:
1249 continue
1250 tool_name = _extract_tool_name(record)
1251 if tool_name is None:
1252 continue
1253 ts = record.get("timestamp", "")
1254 sid = record.get("sessionId", "")
1255 events.append(InvocationEvent(tool_name=tool_name, timestamp=ts, session_id=sid))
1256 except OSError:
1257 pass
1258 events.sort(key=lambda e: e.timestamp)
1259 return events
1262@dataclass
1263class SessionDiff:
1264 """Behavioral diff between two ll sessions."""
1266 session_a: str
1267 session_b: str
1268 skills_added: list[str]
1269 skills_removed: list[str]
1270 count_deltas: dict[str, dict[str, int]]
1271 sequence_diff: list[str]
1273 def to_dict(self) -> dict:
1274 return {
1275 "session_a": self.session_a,
1276 "session_b": self.session_b,
1277 "skills_added": self.skills_added,
1278 "skills_removed": self.skills_removed,
1279 "count_deltas": self.count_deltas,
1280 "sequence_diff": self.sequence_diff,
1281 }
1284def _compute_session_diff(
1285 session_a: str,
1286 events_a: list[InvocationEvent],
1287 session_b: str,
1288 events_b: list[InvocationEvent],
1289) -> SessionDiff:
1290 """Compute the behavioral diff between two session event streams."""
1291 import difflib
1292 from collections import Counter as _Counter
1294 names_a = [e.tool_name for e in events_a]
1295 names_b = [e.tool_name for e in events_b]
1297 set_a = set(names_a)
1298 set_b = set(names_b)
1299 skills_added = sorted(set_b - set_a)
1300 skills_removed = sorted(set_a - set_b)
1302 counter_a: Counter = _Counter(names_a)
1303 counter_b: Counter = _Counter(names_b)
1304 count_deltas: dict[str, dict[str, int]] = {}
1305 for skill in sorted(set_a | set_b):
1306 ca = counter_a.get(skill, 0)
1307 cb = counter_b.get(skill, 0)
1308 if ca != cb:
1309 count_deltas[skill] = {"a": ca, "b": cb, "delta": cb - ca}
1311 label_a = f"session_a ({session_a[:8]})" if len(session_a) > 8 else f"session_a ({session_a})"
1312 label_b = f"session_b ({session_b[:8]})" if len(session_b) > 8 else f"session_b ({session_b})"
1313 sequence_diff = list(
1314 difflib.unified_diff(names_a, names_b, fromfile=label_a, tofile=label_b, lineterm="")
1315 )
1317 return SessionDiff(
1318 session_a=session_a,
1319 session_b=session_b,
1320 skills_added=skills_added,
1321 skills_removed=skills_removed,
1322 count_deltas=count_deltas,
1323 sequence_diff=sequence_diff,
1324 )
1327def _cmd_diff(args: argparse.Namespace, logger: Logger) -> int:
1328 """Compare two sessions' ll-invocation behavior."""
1329 db_path = resolve_history_db()
1331 path_a = _resolve_session_log(args.session_a, db_path)
1332 if path_a is None:
1333 logger.error(f"Cannot resolve session: {args.session_a}")
1334 return 1
1336 path_b = _resolve_session_log(args.session_b, db_path)
1337 if path_b is None:
1338 logger.error(f"Cannot resolve session: {args.session_b}")
1339 return 1
1341 events_a = _events_from_jsonl(path_a)
1342 events_b = _events_from_jsonl(path_b)
1344 diff = _compute_session_diff(args.session_a, events_a, args.session_b, events_b)
1346 if args.json:
1347 print_json(diff.to_dict())
1348 return 0
1350 if (
1351 not diff.skills_added
1352 and not diff.skills_removed
1353 and not diff.count_deltas
1354 and not diff.sequence_diff
1355 ):
1356 print("No behavioral differences found.")
1357 return 0
1359 if diff.skills_added:
1360 print(f"Skills added ({len(diff.skills_added)}):")
1361 for s in diff.skills_added:
1362 print(f" + {s}")
1364 if diff.skills_removed:
1365 if diff.skills_added:
1366 print()
1367 print(f"Skills removed ({len(diff.skills_removed)}):")
1368 for s in diff.skills_removed:
1369 print(f" - {s}")
1371 if diff.count_deltas:
1372 print()
1373 print("Invocation count changes:")
1374 for skill, counts in sorted(diff.count_deltas.items()):
1375 delta_str = f"+{counts['delta']}" if counts["delta"] > 0 else str(counts["delta"])
1376 print(f" {skill}: {counts['a']} → {counts['b']} ({delta_str})")
1378 if diff.sequence_diff:
1379 print()
1380 print("Sequence diff:")
1381 for line in diff.sequence_diff:
1382 print(f" {line}")
1384 return 0
1387_ISSUE_ID_RE = re.compile(r"\b[A-Z]+-\d+\b")
1390@dataclass
1391class _EvalInvocation:
1392 """A reconstructed ll-harness invocation extracted from a JSONL record.
1394 Carries the runner kind and raw (un-redacted) input-context text that the
1395 EvalFixture export needs but that ``InvocationEvent`` discards.
1396 """
1398 runner: str # "skill" | "cmd"
1399 target: str # skill name (runner==skill) or full shell command (runner==cmd)
1400 session_id: str
1401 timestamp: str
1402 input_context: str # raw user-message text; "" when none (e.g. Bash invocations)
1405def _extract_eval_invocation(record: dict) -> _EvalInvocation | None:
1406 """Reconstruct a single ll-harness invocation from a JSONL record.
1408 Mirrors ``_extract_tool_name``'s three signals but also captures the runner
1409 kind and raw input-context text needed for EvalFixture export:
1410 (a) queue enqueue ``/ll:<name>`` -> runner=skill, target=name
1411 (b) user ``<command-name>/ll:<name>`` -> runner=skill, target=name
1412 (c) assistant Bash ``ll-<tool>`` -> runner=cmd, target=<full command>
1414 Returns None for records that carry no ll invocation signal.
1415 """
1416 record_type = record.get("type")
1417 sid = record.get("sessionId", "")
1418 ts = record.get("timestamp", "")
1420 # (a) queue-operation enqueue
1421 if record_type == "queue-operation" and record.get("operation") == "enqueue":
1422 content = record.get("content", "")
1423 if isinstance(content, str) and content.startswith("/ll:"):
1424 m = _QUEUE_SKILL_RE.match(content)
1425 if m:
1426 return _EvalInvocation("skill", m.group(1), sid, ts, content)
1428 # (b) user records with <command-name>/ll: pattern
1429 if record_type == "user":
1430 message = record.get("message", {})
1431 if not isinstance(message, dict):
1432 return None
1433 content = message.get("content")
1434 text = ""
1435 if isinstance(content, str):
1436 text = content
1437 elif isinstance(content, list):
1438 for block in content:
1439 if isinstance(block, dict):
1440 text = block.get("text", "")
1441 if text:
1442 break
1443 if text:
1444 m = _COMMAND_NAME_SKILL_RE.search(text)
1445 if m:
1446 name = m.group(1)
1447 if name.endswith("</command-name>"):
1448 name = name[: -len("</command-name>")]
1449 return _EvalInvocation("skill", name, sid, ts, text)
1451 # (c) assistant Bash tool-use invoking ll-<tool>
1452 if record_type == "assistant":
1453 message = record.get("message", {})
1454 content = message.get("content", [])
1455 if isinstance(content, list):
1456 for block in content:
1457 if (
1458 isinstance(block, dict)
1459 and block.get("type") == "tool_use"
1460 and block.get("name") == "Bash"
1461 ):
1462 cmd = block.get("input", {}).get("command", "")
1463 if _LL_BASH_RE.search(cmd):
1464 return _EvalInvocation("cmd", cmd, sid, ts, "")
1466 return None
1469def _record_has_error(record: dict) -> bool:
1470 """True if a JSONL record is a ``tool_result`` flagged ``is_error``.
1472 Used as the session-level ``failed`` outcome signal: session logs expose no
1473 output-quality judgment, only execution evidence (ARCHITECTURE-017).
1474 """
1475 if record.get("type") != "user":
1476 return False
1477 message = record.get("message", {})
1478 if not isinstance(message, dict):
1479 return False
1480 content = message.get("content")
1481 if isinstance(content, list):
1482 for block in content:
1483 if (
1484 isinstance(block, dict)
1485 and block.get("type") == "tool_result"
1486 and block.get("is_error")
1487 ):
1488 return True
1489 return False
1492def _classify_outcome(metadata: dict, *, has_error: bool) -> str:
1493 """Map session metadata + error signal to an EvalFixture execution outcome.
1495 Precedence ``failed`` > ``corrected`` > ``accepted``; ``unknown`` when the DB
1496 returned no metadata. Per ARCHITECTURE-017 the taxonomy is EXECUTION, not
1497 output quality. ``metadata`` is the dict from
1498 ``history_reader.lookup_session_metadata`` (``{}`` when the session is absent).
1499 """
1500 if has_error:
1501 return "failed"
1502 if not metadata:
1503 return "unknown"
1504 if metadata.get("has_corrections"):
1505 return "corrected"
1506 return "accepted"
1509def _redact_input_context(text: str) -> tuple[str | None, bool]:
1510 """Best-effort, non-blocking redaction of user-message text.
1512 Applies ``pii.redact_pii`` (email/phone/SSN) then ``_ABS_PATH_RE`` (absolute
1513 paths). Returns ``(redacted_text_or_None, pii_detected)`` where ``pii_detected``
1514 is True when either pass altered the text. Never raises and never drops a
1515 record for unredactable content (ARCHITECTURE-017).
1516 """
1517 if not text:
1518 return None, False
1519 from little_loops.pii import redact_pii
1521 redacted = redact_pii(text)
1522 redacted = _ABS_PATH_RE.sub("<path>", redacted)
1523 return redacted, redacted != text
1526def _build_eval_fixture(inv: _EvalInvocation, outcome: str) -> dict:
1527 """Map a reconstructed invocation + outcome to an EvalFixture v1 record.
1529 Pure function (no I/O) — the mapping core covered by unit tests. Schema and
1530 field semantics per decision ARCHITECTURE-017 in ``.ll/decisions.yaml``: the
1531 fixture replays into ``ll-harness <runner> <target> [runner_args...]
1532 [--exit-code N] [--semantic TEXT] [--timeout S]`` (ll-harness has no loader).
1533 """
1534 input_context, pii_detected = _redact_input_context(inv.input_context)
1535 issue_match = _ISSUE_ID_RE.search(inv.input_context) if inv.input_context else None
1536 issue_id = issue_match.group(0) if issue_match else None
1537 skill_name = inv.target if inv.runner == "skill" else None
1538 return {
1539 "runner": inv.runner,
1540 "target": inv.target,
1541 "session_id": inv.session_id,
1542 "timestamp": inv.timestamp,
1543 "outcome": outcome,
1544 "runner_args": [],
1545 "exit_code": None,
1546 "semantic": None,
1547 "timeout": 120,
1548 "input_context": input_context,
1549 "issue_id": issue_id,
1550 "skill_name": skill_name,
1551 "pii_detected": pii_detected,
1552 }
1555def _fixture_to_harness_argv(fixture: dict) -> list[str]:
1556 """Serialize an EvalFixture record back into an ``ll-harness`` argv.
1558 ll-harness has no fixture loader (ARCHITECTURE-017); a fixture replays by
1559 serializing its fields into the harness CLI arg surface. Used by the
1560 round-trip test to prove every exported fixture is a valid harness invocation.
1561 """
1562 argv: list[str] = [fixture["runner"], fixture["target"]]
1563 argv.extend(fixture.get("runner_args") or [])
1564 if fixture.get("exit_code") is not None:
1565 argv.extend(["--exit-code", str(fixture["exit_code"])])
1566 if fixture.get("semantic") is not None:
1567 argv.extend(["--semantic", str(fixture["semantic"])])
1568 timeout = fixture.get("timeout")
1569 if timeout is not None and timeout != 120:
1570 argv.extend(["--timeout", str(timeout)])
1571 return argv
1574def _cmd_eval_export(args: argparse.Namespace) -> int:
1575 """Export ll-harness eval fixtures reconstructed from session logs (FEAT-1971).
1577 Walks the current project's JSONL logs, reconstructs each ll invocation, sources
1578 an execution outcome from ``history_reader.lookup_session_metadata``, redacts the
1579 input context, and writes EvalFixture v1 records (YAML default, JSON with
1580 ``--json``). Schema + outcome taxonomy: decision ARCHITECTURE-017 in
1581 ``.ll/decisions.yaml``.
1582 """
1583 from little_loops.history_reader import lookup_session_metadata
1585 cwd_path = Path(args.project) if args.project else Path.cwd()
1586 project_folder = get_project_folder(cwd_path)
1587 if project_folder is None:
1588 print(f"No session project folder found for: {cwd_path}", file=sys.stderr)
1589 return 1
1590 from little_loops.session_store import resolve_history_db
1592 db_path = resolve_history_db(cwd_path / ".ll" / "history.db")
1594 # Single JSONL pass: collect raw invocations + per-session error flags together
1595 # (avoids the double-parse the decision warns against).
1596 invocations: list[_EvalInvocation] = []
1597 session_has_error: dict[str, bool] = {}
1598 jsonl_files = [f for f in project_folder.glob("*.jsonl") if not f.name.startswith("agent-")]
1599 for jsonl_file in jsonl_files:
1600 try:
1601 with open(jsonl_file, encoding="utf-8") as f:
1602 for line in f:
1603 line = line.strip()
1604 if not line:
1605 continue
1606 try:
1607 record = json.loads(line)
1608 except json.JSONDecodeError:
1609 continue
1610 if _record_has_error(record):
1611 sid = record.get("sessionId", "")
1612 if sid:
1613 session_has_error[sid] = True
1614 inv = _extract_eval_invocation(record)
1615 if inv is not None:
1616 invocations.append(inv)
1617 except OSError:
1618 continue
1620 # Stable, deterministic order: by timestamp then session.
1621 invocations.sort(key=lambda e: (e.timestamp, e.session_id))
1623 metadata_cache: dict[str, dict] = {}
1624 fixtures: list[dict] = []
1625 skipped = 0
1626 for inv in invocations:
1627 # --skill: keep only skill-runner invocations of the named target.
1628 if args.skill and not (inv.runner == "skill" and inv.target == args.skill):
1629 continue
1631 if inv.session_id not in metadata_cache:
1632 metadata_cache[inv.session_id] = lookup_session_metadata(inv.session_id, db=db_path)
1633 outcome = _classify_outcome(
1634 metadata_cache[inv.session_id],
1635 has_error=session_has_error.get(inv.session_id, False),
1636 )
1637 # No extractable execution outcome -> skip with a logged count.
1638 if outcome == "unknown":
1639 skipped += 1
1640 continue
1642 fixture = _build_eval_fixture(inv, outcome)
1644 # --issue: match the extracted issue_id or a literal occurrence in target.
1645 if args.issue and args.issue != fixture["issue_id"] and args.issue not in inv.target:
1646 continue
1648 fixtures.append(fixture)
1649 if args.limit and len(fixtures) >= args.limit:
1650 break
1652 if args.json:
1653 output = json.dumps(fixtures, indent=2)
1654 else:
1655 import yaml
1657 output = yaml.safe_dump(fixtures, sort_keys=False, default_flow_style=False)
1659 if args.out:
1660 out_path = Path(args.out)
1661 try:
1662 out_path.parent.mkdir(parents=True, exist_ok=True)
1663 out_path.write_text(output, encoding="utf-8")
1664 except OSError as exc:
1665 print(f"Failed to write {out_path}: {exc}", file=sys.stderr)
1666 return 1
1667 print(f"Wrote {len(fixtures)} fixture(s) to {out_path}", file=sys.stderr)
1668 else:
1669 print(output, end="" if output.endswith("\n") else "\n")
1671 if skipped:
1672 print(
1673 f"Skipped {skipped} invocation(s) with no extractable outcome",
1674 file=sys.stderr,
1675 )
1677 return 0
1680def _build_parser() -> argparse.ArgumentParser:
1681 """Build the argument parser for ll-logs."""
1682 parser = argparse.ArgumentParser(
1683 prog="ll-logs",
1684 description="Discover and extract ll-relevant JSONL entries from Claude Code logs",
1685 formatter_class=argparse.RawDescriptionHelpFormatter,
1686 epilog="""
1687Examples:
1688 %(prog)s discover # List all projects with ll activity
1689 %(prog)s tail --loop <name> # Stream live events from an active loop session
1690 %(prog)s extract --all # Extract all projects to logs/
1691 %(prog)s extract --project /path # Extract one project to logs/<slug>/
1692 %(prog)s extract --all --cmd ll-history # Filter to ll-history invocations
1693""",
1694 )
1696 subparsers = parser.add_subparsers(dest="command", help="Available commands")
1697 discover_parser = subparsers.add_parser(
1698 "discover",
1699 help="List all Claude projects with ll activity (one path per line, sorted)",
1700 )
1701 add_json_arg(discover_parser)
1703 tail_parser = subparsers.add_parser(
1704 "tail",
1705 help="Stream live events from an active loop session",
1706 )
1707 tail_parser.add_argument("--loop", required=True, metavar="NAME", help="Loop name to tail")
1709 extract_parser = subparsers.add_parser(
1710 "extract",
1711 help="Extract ll-relevant JSONL records to logs/<slug>/<session-id>.jsonl",
1712 )
1713 target_group = extract_parser.add_mutually_exclusive_group(required=True)
1714 target_group.add_argument(
1715 "--project",
1716 type=Path,
1717 metavar="DIR",
1718 help="Working directory of the target project",
1719 )
1720 target_group.add_argument(
1721 "--all",
1722 action="store_true",
1723 help="Extract all projects with ll activity",
1724 )
1725 extract_parser.add_argument(
1726 "--cmd",
1727 metavar="TOOL",
1728 help="Filter to records containing this ll- tool name (e.g. ll-history)",
1729 )
1731 sequences_parser = subparsers.add_parser(
1732 "sequences",
1733 help="Extract tool-chain n-grams of ll invocations from JSONL logs",
1734 )
1735 sequences_target = sequences_parser.add_mutually_exclusive_group(required=True)
1736 sequences_target.add_argument(
1737 "--project",
1738 type=Path,
1739 metavar="DIR",
1740 help="Working directory of the target project",
1741 )
1742 sequences_target.add_argument(
1743 "--all",
1744 action="store_true",
1745 help="Analyze all projects with ll activity",
1746 )
1747 sequences_parser.add_argument(
1748 "--min-len",
1749 type=int,
1750 default=2,
1751 metavar="N",
1752 help="Minimum n-gram length (default: 2)",
1753 )
1754 sequences_parser.add_argument(
1755 "--min-count",
1756 type=int,
1757 default=1,
1758 metavar="M",
1759 help="Minimum occurrence count to include (default: 1)",
1760 )
1761 sequences_parser.add_argument(
1762 "--top",
1763 type=int,
1764 default=None,
1765 metavar="N",
1766 help="Limit output to top N chains by frequency",
1767 )
1768 sequences_parser.add_argument(
1769 "--window-days",
1770 type=int,
1771 default=None,
1772 metavar="D",
1773 help="Only consider records within D days of latest record",
1774 )
1775 add_json_arg(sequences_parser)
1777 stats_parser = subparsers.add_parser(
1778 "stats",
1779 help="Aggregate skill invocation frequency and correction rate from history.db",
1780 )
1781 stats_target = stats_parser.add_mutually_exclusive_group(required=True)
1782 stats_target.add_argument(
1783 "--project",
1784 type=Path,
1785 metavar="DIR",
1786 help="Working directory of the target project",
1787 )
1788 stats_target.add_argument(
1789 "--all",
1790 action="store_true",
1791 help="Aggregate across all projects with ll activity",
1792 )
1793 stats_parser.add_argument(
1794 "--window-days",
1795 type=int,
1796 default=None,
1797 metavar="D",
1798 help="Only consider records within D days of latest record",
1799 )
1800 stats_parser.add_argument(
1801 "--sort",
1802 choices=["freq", "corrections"],
1803 default="freq",
1804 help="Sort output by invocation frequency or correction count (default: freq)",
1805 )
1806 add_json_arg(stats_parser)
1808 scan_failures_parser = subparsers.add_parser(
1809 "scan-failures",
1810 help="Mine failed ll-* calls from interactive session logs and propose bug issues",
1811 )
1812 scan_failures_target = scan_failures_parser.add_mutually_exclusive_group(required=True)
1813 scan_failures_target.add_argument(
1814 "--project",
1815 type=Path,
1816 metavar="DIR",
1817 help="Working directory of the target project",
1818 )
1819 scan_failures_target.add_argument(
1820 "--all",
1821 action="store_true",
1822 help="Scan all projects with ll activity",
1823 )
1824 scan_failures_parser.add_argument(
1825 "--window-days",
1826 type=int,
1827 default=None,
1828 metavar="D",
1829 help="Only consider records within D days of latest record",
1830 )
1831 scan_failures_parser.add_argument(
1832 "--capture",
1833 action="store_true",
1834 help="Create bug issue files for each failure cluster (one per tool+error signature)",
1835 )
1836 scan_failures_parser.add_argument(
1837 "--capture-foreign",
1838 action="store_true",
1839 help="Allow --capture to include failures from projects other than the current directory (only meaningful with --all)",
1840 )
1841 add_json_arg(scan_failures_parser)
1843 dead_skills_parser = subparsers.add_parser(
1844 "dead-skills",
1845 help="List catalog skills/commands with zero or low invocations across the corpus",
1846 )
1847 dead_skills_target = dead_skills_parser.add_mutually_exclusive_group(required=True)
1848 dead_skills_target.add_argument(
1849 "--project",
1850 type=Path,
1851 metavar="DIR",
1852 help="Working directory of the target project (also used as catalog root)",
1853 )
1854 dead_skills_target.add_argument(
1855 "--all",
1856 action="store_true",
1857 help="Aggregate across all projects; catalog loaded from current directory",
1858 )
1859 dead_skills_parser.add_argument(
1860 "--window-days",
1861 type=int,
1862 default=None,
1863 metavar="D",
1864 help="Only consider records within D days of latest record",
1865 )
1866 dead_skills_parser.add_argument(
1867 "--threshold",
1868 type=int,
1869 default=3,
1870 metavar="N",
1871 help="Skills with invocations <= N are 'rarely' invoked (default: 3)",
1872 )
1873 add_json_arg(dead_skills_parser)
1875 diff_parser = subparsers.add_parser(
1876 "diff",
1877 help="Compare two sessions' ll-invocation behavior (skills, sequences, counts)",
1878 )
1879 diff_parser.add_argument(
1880 "session_a", metavar="SESSION_A", help="First session ID or JSONL file path"
1881 )
1882 diff_parser.add_argument(
1883 "session_b", metavar="SESSION_B", help="Second session ID or JSONL file path"
1884 )
1885 add_json_arg(diff_parser)
1887 eval_export_parser = subparsers.add_parser(
1888 "eval-export",
1889 help="Export eval fixtures from ll-harness session logs",
1890 )
1891 eval_export_parser.add_argument(
1892 "--project",
1893 type=Path,
1894 metavar="DIR",
1895 help="Project working directory (default: current directory)",
1896 )
1897 eval_export_parser.add_argument(
1898 "--skill",
1899 metavar="NAME",
1900 help="Filter by skill name",
1901 )
1902 eval_export_parser.add_argument(
1903 "--issue",
1904 metavar="ID",
1905 help="Filter by issue ID in session context",
1906 )
1907 eval_export_parser.add_argument(
1908 "--limit",
1909 type=int,
1910 default=0,
1911 metavar="N",
1912 help="Cap output records (0 = unlimited)",
1913 )
1914 eval_export_parser.add_argument(
1915 "--out",
1916 metavar="PATH",
1917 help="Write output to file (default: stdout)",
1918 )
1919 eval_export_parser.add_argument(
1920 "--json",
1921 action="store_true",
1922 help="JSON output instead of YAML (default: YAML)",
1923 )
1925 return parser
1928def _parse_args() -> argparse.Namespace:
1929 """Parse command-line arguments. Exposed for testing."""
1930 return _build_parser().parse_args()
1933def main_logs() -> int:
1934 """Entry point for ll-logs command.
1936 Returns:
1937 0 on success, 1 when no subcommand given or on error.
1938 """
1939 with cli_event_context(DEFAULT_DB_PATH, "ll-logs", sys.argv[1:]):
1940 configure_output()
1941 logger = Logger(use_color=use_color_enabled())
1943 parser = _build_parser()
1944 args = parser.parse_args()
1946 if not args.command:
1947 parser.print_help()
1948 return 1
1950 if args.command == "discover":
1951 projects = discover_all_projects(logger)
1952 if args.json:
1953 print_json({"paths": [str(p) for p in projects]})
1954 else:
1955 for path in projects:
1956 print(path)
1957 return 0
1959 if args.command == "tail":
1960 config = BRConfig(Path.cwd())
1961 loops_dir = Path(config.loops.loops_dir)
1962 return _cmd_tail(args, loops_dir)
1964 if args.command == "extract":
1965 return _cmd_extract(args, logger)
1967 if args.command == "sequences":
1968 return _cmd_sequences(args, logger)
1970 if args.command == "stats":
1971 return _cmd_stats(args, logger)
1973 if args.command == "scan-failures":
1974 return _cmd_scan_failures(args, logger)
1976 if args.command == "dead-skills":
1977 return _cmd_dead_skills(args, logger)
1979 if args.command == "diff":
1980 return _cmd_diff(args, logger)
1982 if args.command == "eval-export":
1983 return _cmd_eval_export(args)
1985 return 1