Coverage for little_loops / cli / session.py: 75%

1"""ll-session: query the unified session store (SQLite + FTS5). 

2 

3Wraps :mod:`little_loops.session_store` with a CLI surface so operators can 

4search and inspect the per-project ``.ll/session.db`` without re-parsing the 

5scattered JSON/markdown sources the analyze-* skills read. 

6 

7Subcommands: 

8 search FTS5 full-text query with BM25-ranked results 

9 recent most recent rows for an event kind (tool, file, issue, loop, correction) 

10 backfill seed the database from existing on-disk sources 

11""" 

12 

13from __future__ import annotations 

14 

15import argparse 

16from pathlib import Path 

17 

18from little_loops.cli.output import configure_output, use_color_enabled 

19from little_loops.logger import Logger 

20from little_loops.session_store import DEFAULT_DB_PATH, backfill, recent, search 

21 

22 

23def _build_parser() -> argparse.ArgumentParser: 

24 """Build the argument parser for ll-session.""" 

25 parser = argparse.ArgumentParser( 

26 prog="ll-session", 

27 description="Query the unified session store (SQLite + FTS5)", 

28 formatter_class=argparse.RawDescriptionHelpFormatter, 

29 epilog=""" 

30Examples: 

31 %(prog)s search --fts "rate limit" # Full-text search, BM25-ranked 

32 %(prog)s recent --kind loop # Recent loop events 

33 %(prog)s backfill # Seed the database from on-disk sources 

34""", 

35 ) 

36 parser.add_argument( 

37 "--db", 

38 type=Path, 

39 default=DEFAULT_DB_PATH, 

40 metavar="PATH", 

41 help="Path to the session database (default: .ll/session.db)", 

42 ) 

43 

44 subparsers = parser.add_subparsers(dest="command", help="Available commands") 

45 

46 search_parser = subparsers.add_parser("search", help="FTS5 full-text search") 

47 search_parser.add_argument("--fts", required=True, metavar="QUERY", help="FTS5 match query") 

48 search_parser.add_argument( 

49 "--limit", type=int, default=20, metavar="N", help="Maximum results (default: 20)" 

50 ) 

51 

52 recent_parser = subparsers.add_parser("recent", help="Recent events by kind") 

53 recent_parser.add_argument( 

54 "--kind", 

55 required=True, 

56 choices=["tool", "file", "issue", "loop", "correction"], 

57 help="Event kind to list", 

58 ) 

59 recent_parser.add_argument( 

60 "--limit", type=int, default=20, metavar="N", help="Maximum rows (default: 20)" 

61 ) 

62 

63 subparsers.add_parser("backfill", help="Seed the database from existing on-disk sources") 

64 

65 return parser 

66 

67 

68def _parse_args() -> argparse.Namespace: 

69 """Parse command-line arguments. Exposed for testing.""" 

70 return _build_parser().parse_args() 

71 

72 

73def main_session() -> int: 

74 """Entry point for ll-session command. 

75 

76 Returns: 

77 0 on success, 1 when no subcommand is given or on error. 

78 """ 

79 configure_output() 

80 logger = Logger(use_color=use_color_enabled()) 

81 

82 parser = _build_parser() 

83 args = parser.parse_args() 

84 

85 if not args.command: 

86 parser.print_help() 

87 return 1 

88 

89 if args.command == "search": 

90 try: 

91 results = search(args.db, query=args.fts, limit=args.limit) 

92 except ValueError as exc: 

93 logger.error(str(exc)) 

94 return 1 

95 if not results: 

96 print("No matches.") 

97 return 0 

98 for row in results: 

99 anchor = f" ({row['anchor']})" if row.get("anchor") else "" 

100 print(f"[{row['kind']}] {row['content']}{anchor}") 

101 return 0 

102 

103 if args.command == "recent": 

104 rows = recent(args.db, kind=args.kind, limit=args.limit) 

105 if not rows: 

106 print(f"No {args.kind} events.") 

107 return 0 

108 for row in rows: 

109 fields = ", ".join(f"{k}={v}" for k, v in row.items() if k != "id" and v is not None) 

110 print(fields) 

111 return 0 

112 

113 if args.command == "backfill": 

114 counts = backfill(args.db) 

115 total = sum(counts.values()) 

116 logger.success( 

117 f"Backfilled {total} rows " 

118 f"(issues={counts['issues']}, loops={counts['loops']}, tools={counts['tools']})" 

119 ) 

120 return 0 

121 

122 return 1