Coverage for little_loops / events.py: 39%

1"""Event system for little-loops extension architecture. 

2 

3Provides structured event types and a multi-observer event bus that extensions 

4can subscribe to for receiving lifecycle events from FSM loops, issue management, 

5and automation workflows. 

6 

7Public exports: 

8 LLEvent: Structured event dataclass with type, timestamp, and payload 

9 EventBus: Multi-observer event dispatcher with optional JSONL file sink 

10""" 

11 

12from __future__ import annotations 

13 

14import fnmatch 

15import json 

16import logging 

17from collections.abc import Callable 

18from dataclasses import dataclass, field 

19from pathlib import Path 

20from typing import Any 

21 

22logger = logging.getLogger(__name__) 

23 

24# Type for event callback (matches existing EventCallback in fsm/executor.py) 

25EventCallback = Callable[[dict[str, Any]], None] 

26 

27 

28@dataclass 

29class LLEvent: 

30 """Structured event emitted by little-loops subsystems. 

31 

32 Attributes: 

33 type: Event type identifier (e.g. "fsm.state_enter", "fsm.loop_complete") 

34 timestamp: ISO 8601 timestamp string 

35 payload: Type-specific event data 

36 """ 

37 

38 type: str 

39 timestamp: str 

40 payload: dict[str, Any] = field(default_factory=dict) 

41 

42 def to_dict(self) -> dict[str, Any]: 

43 """Convert to dictionary for JSON serialization. 

44 

45 Produces a flat dict compatible with the existing FSM event format: 

46 {"event": type, "ts": timestamp, ...payload} 

47 """ 

48 return {"event": self.type, "ts": self.timestamp, **self.payload} 

49 

50 @classmethod 

51 def from_dict(cls, data: dict[str, Any]) -> LLEvent: 

52 """Create from dictionary (JSON deserialization). 

53 

54 Accepts the flat dict format: {"event": ..., "ts": ..., ...payload} 

55 """ 

56 copy = dict(data) 

57 event_type = copy.pop("event", copy.pop("type", "unknown")) 

58 ts = copy.pop("ts", copy.pop("timestamp", "")) 

59 return cls(type=event_type, timestamp=ts, payload=copy) 

60 

61 @classmethod 

62 def from_raw_event(cls, raw: dict[str, Any]) -> LLEvent: 

63 """Convert existing executor event dict to LLEvent without mutating the original.""" 

64 return cls.from_dict(dict(raw)) 

65 

66 

67class EventBus: 

68 """Multi-observer event dispatcher. 

69 

70 Dispatches event dicts to all registered observers. Exceptions in individual 

71 observers are caught and logged, not propagated. 

72 """ 

73 

74 def __init__(self) -> None: 

75 self._observers: list[tuple[EventCallback, list[str] | None]] = [] 

76 self._file_sinks: list[Path] = [] 

77 

78 def register(self, callback: EventCallback, filter: str | list[str] | None = None) -> None: 

79 """Register an observer to receive events. 

80 

81 Args: 

82 callback: Callable that receives raw event dicts. 

83 filter: Optional glob pattern(s) to match against the event's ``"event"`` key. 

84 A single string (e.g. ``"issue.*"``) or a list of strings 

85 (e.g. ``["issue.*", "parallel.*"]``). ``None`` (default) means 

86 the observer receives every event — preserving existing behaviour. 

87 FSM executor events use bare names (``"state_enter"``, ``"loop_*"``); 

88 other subsystems use dotted namespaces (``"issue.*"``, ``"parallel.*"``). 

89 """ 

90 patterns: list[str] | None = None 

91 if filter is not None: 

92 patterns = [filter] if isinstance(filter, str) else list(filter) 

93 self._observers.append((callback, patterns)) 

94 

95 def unregister(self, callback: EventCallback) -> None: 

96 """Remove an observer. No-op if not registered.""" 

97 for i, (cb, _) in enumerate(self._observers): 

98 if cb is callback: 

99 del self._observers[i] 

100 return 

101 

102 def add_file_sink(self, path: Path) -> None: 

103 """Add a JSONL file sink. Events will be appended to this file.""" 

104 path.parent.mkdir(parents=True, exist_ok=True) 

105 self._file_sinks.append(path) 

106 

107 def emit(self, event: dict[str, Any]) -> None: 

108 """Dispatch event to all observers and file sinks. 

109 

110 Observer exceptions are caught and logged to prevent one observer 

111 from blocking others. 

112 """ 

113 event_type = event.get("event", "") 

114 for observer, filter_patterns in self._observers: 

115 if filter_patterns is not None and not any( 

116 fnmatch.fnmatch(event_type, p) for p in filter_patterns 

117 ): 

118 continue 

119 try: 

120 observer(event) 

121 except Exception: 

122 logger.warning("EventBus observer raised an exception", exc_info=True) 

123 

124 for sink_path in self._file_sinks: 

125 try: 

126 with open(sink_path, "a", encoding="utf-8") as f: 

127 f.write(json.dumps(event) + "\n") 

128 except Exception: 

129 logger.warning("EventBus file sink write failed: %s", sink_path, exc_info=True) 

130 

131 @staticmethod 

132 def read_events(path: Path) -> list[LLEvent]: 

133 """Read events from a JSONL file. 

134 

135 Returns: 

136 List of LLEvent instances, empty if file doesn't exist. 

137 Malformed lines are silently skipped. 

138 """ 

139 if not path.exists(): 

140 return [] 

141 events: list[LLEvent] = [] 

142 with open(path, encoding="utf-8") as f: 

143 for line in f: 

144 line = line.strip() 

145 if line: 

146 try: 

147 events.append(LLEvent.from_dict(json.loads(line))) 

148 except json.JSONDecodeError: 

149 continue 

150 return events