Coverage for little_loops / workflow_sequence / io.py: 0%

1"""File I/O helpers for workflow sequence analysis.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import sys 

7from pathlib import Path 

8from typing import Any 

9 

10import yaml 

11 

12 

13def _load_messages(messages_file: Path) -> list[dict[str, Any]]: 

14 """Load messages from JSONL file.""" 

15 messages = [] 

16 skipped = 0 

17 with open(messages_file, encoding="utf-8") as f: 

18 for line_num, raw_line in enumerate(f, 1): 

19 line = raw_line.strip() 

20 if not line: 

21 continue 

22 try: 

23 messages.append(json.loads(line)) 

24 except json.JSONDecodeError as e: 

25 skipped += 1 

26 print(f"Warning: skipping malformed line {line_num}: {e}", file=sys.stderr) 

27 if skipped: 

28 print(f"Warning: skipped {skipped} malformed line(s) in {messages_file}", file=sys.stderr) 

29 return messages 

30 

31 

32def _load_messages_from_db(db_path: Path) -> list[dict[str, Any]]: 

33 """Load user messages from the unified session DB (ENH-1621). 

34 

35 Reads ``message_events`` rows seeded by 

36 :func:`little_loops.session_store._backfill_messages` and shapes them like 

37 the JSONL records :func:`_load_messages` returns so the rest of 

38 :func:`analyze_workflows` consumes them unchanged. 

39 

40 Returns ``[]`` when the DB is missing, the table is empty, or the schema 

41 pre-dates the v2 migration — the caller treats an empty result as the 

42 fallback trigger. 

43 """ 

44 if not db_path.exists(): 

45 return [] 

46 try: 

47 from little_loops.session_store import connect 

48 

49 conn = connect(db_path) 

50 except Exception: # pragma: no cover - defensive 

51 return [] 

52 try: 

53 try: 

54 rows = conn.execute( 

55 "SELECT id, ts, session_id, content FROM message_events ORDER BY id" 

56 ).fetchall() 

57 except Exception: 

58 return [] 

59 finally: 

60 conn.close() 

61 messages: list[dict[str, Any]] = [] 

62 for row in rows: 

63 messages.append( 

64 { 

65 "uuid": f"msg-{row['id']}", 

66 "timestamp": row["ts"], 

67 "session_id": row["session_id"], 

68 "content": row["content"] or "", 

69 "git_branch": None, 

70 } 

71 ) 

72 return messages 

73 

74 

75def _load_patterns(patterns_file: Path) -> dict[str, Any]: 

76 """Load patterns from Step 1 YAML output.""" 

77 with open(patterns_file, encoding="utf-8") as f: 

78 return yaml.safe_load(f) or {}