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

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/history.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, print_json, 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/history.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", "message"], 

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 recent_parser.add_argument( 

63 "--json", action="store_true", dest="json", help="Output as JSON array" 

64 ) 

65 

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

67 

68 return parser 

69 

70 

71def _parse_args() -> argparse.Namespace: 

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

73 return _build_parser().parse_args() 

74 

75 

76def main_session() -> int: 

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

78 

79 Returns: 

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

81 """ 

82 configure_output() 

83 logger = Logger(use_color=use_color_enabled()) 

84 

85 parser = _build_parser() 

86 args = parser.parse_args() 

87 

88 if not args.command: 

89 parser.print_help() 

90 return 1 

91 

92 if args.command == "search": 

93 try: 

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

95 except ValueError as exc: 

96 logger.error(str(exc)) 

97 return 1 

98 if not results: 

99 print("No matches.") 

100 return 0 

101 for row in results: 

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

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

104 return 0 

105 

106 if args.command == "recent": 

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

108 if args.json: 

109 print_json(list(rows)) 

110 return 0 

111 if not rows: 

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

113 return 0 

114 for row in rows: 

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

116 print(fields) 

117 return 0 

118 

119 if args.command == "backfill": 

120 counts = backfill(args.db) 

121 total = sum(counts.values()) 

122 logger.success( 

123 f"Backfilled {total} rows " 

124 f"(issues={counts['issues']}, loops={counts['loops']}, " 

125 f"tools={counts['tools']}, messages={counts.get('messages', 0)})" 

126 ) 

127 return 0 

128 

129 return 1