Coverage for little_loops / fsm / signal_detector.py: 65%

1"""Signal detection for FSM loop execution output. 

2 

3This module provides pattern-based signal detection for interpreting 

4special markers in action output, such as CONTEXT_HANDOFF:, FATAL_ERROR:, etc. 

5 

6The signal detection layer enables the FSM executor to respond to signals 

7emitted by commands without coupling the executor to specific signal formats. 

8""" 

9 

10from __future__ import annotations 

11 

12import re 

13from dataclasses import dataclass 

14 

15 

16@dataclass 

17class DetectedSignal: 

18 """A signal detected in command output. 

19 

20 Attributes: 

21 signal_type: Type of signal (e.g., "handoff", "error", "stop") 

22 payload: Captured content after the signal marker 

23 raw_match: The full matched string 

24 """ 

25 

26 signal_type: str 

27 payload: str | None 

28 raw_match: str 

29 

30 

31class SignalPattern: 

32 """Configurable signal pattern for detection. 

33 

34 Signal patterns use regex to match markers in command output. 

35 A capture group can be used to extract a payload. 

36 

37 Example: 

38 pattern = SignalPattern("handoff", r"CONTEXT_HANDOFF:\\s*(.+)") 

39 signal = pattern.search("CONTEXT_HANDOFF: Ready for fresh session") 

40 # signal.payload == "Ready for fresh session" 

41 """ 

42 

43 def __init__(self, name: str, pattern: str) -> None: 

44 """Initialize signal pattern. 

45 

46 Args: 

47 name: Signal type name (e.g., "handoff") 

48 pattern: Regex pattern with optional capture group for payload 

49 """ 

50 self.name = name 

51 self.regex = re.compile(pattern, re.MULTILINE) 

52 

53 def search(self, output: str) -> DetectedSignal | None: 

54 """Search for this signal pattern in output. 

55 

56 Args: 

57 output: Command output to search 

58 

59 Returns: 

60 DetectedSignal if found, None otherwise 

61 """ 

62 match = self.regex.search(output) 

63 if match: 

64 payload = match.group(1).strip() if match.groups() else None 

65 return DetectedSignal( 

66 signal_type=self.name, 

67 payload=payload, 

68 raw_match=match.group(0), 

69 ) 

70 return None 

71 

72 

73# Built-in signal patterns 

74HANDOFF_SIGNAL = SignalPattern("handoff", r"CONTEXT_HANDOFF:\s*(.+)") 

75ERROR_SIGNAL = SignalPattern("error", r"FATAL_ERROR:\s*(.+)") 

76STOP_SIGNAL = SignalPattern("stop", r"LOOP_STOP:\s*(.*)") 

77 

78 

79class SignalDetector: 

80 """Detect signals in command output. 

81 

82 Provides pattern-based signal detection with extensibility 

83 for custom signal types. The default patterns detect: 

84 

85 - CONTEXT_HANDOFF: - Signals context exhaustion, payload is continuation info 

86 - FATAL_ERROR: - Signals unrecoverable error, payload is error message 

87 - LOOP_STOP: - Signals explicit loop termination request 

88 

89 Example: 

90 detector = SignalDetector() 

91 signal = detector.detect_first("Some output\\nCONTEXT_HANDOFF: Continue") 

92 if signal and signal.signal_type == "handoff": 

93 # Handle handoff... 

94 """ 

95 

96 def __init__(self, patterns: list[SignalPattern] | None = None) -> None: 

97 """Initialize detector with patterns. 

98 

99 Args: 

100 patterns: List of signal patterns to detect. 

101 Defaults to built-in patterns (handoff, error, stop). 

102 """ 

103 self.patterns = patterns or [HANDOFF_SIGNAL, ERROR_SIGNAL, STOP_SIGNAL] 

104 

105 def detect(self, output: str) -> list[DetectedSignal]: 

106 """Detect all signals in output. 

107 

108 Args: 

109 output: Command output to scan 

110 

111 Returns: 

112 List of all detected signals 

113 """ 

114 return [ 

115 signal for pattern in self.patterns if (signal := pattern.search(output)) is not None 

116 ] 

117 

118 def detect_first(self, output: str) -> DetectedSignal | None: 

119 """Detect first matching signal in output. 

120 

121 Patterns are checked in order, so the first pattern in the list 

122 has highest priority. 

123 

124 Args: 

125 output: Command output to scan 

126 

127 Returns: 

128 First detected signal, or None if no signals found 

129 """ 

130 for pattern in self.patterns: 

131 if signal := pattern.search(output): 

132 return signal 

133 return None