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

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_patterns(patterns_file: Path) -> dict[str, Any]: 

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

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

35 return yaml.safe_load(f) or {}