Coverage for little_loops / fsm / handoff_handler.py: 53%

1"""Handoff handling for FSM loop execution. 

2 

3This module provides behavior handlers for context handoff signals, 

4supporting pause, spawn, and terminate behaviors. 

5 

6The handler is Claude-specific and knows how to spawn continuation sessions 

7via the Claude CLI. 

8""" 

9 

10from __future__ import annotations 

11 

12import subprocess 

13from dataclasses import dataclass 

14from enum import Enum 

15 

16 

17class HandoffBehavior(Enum): 

18 """Behavior when a handoff signal is detected. 

19 

20 - TERMINATE: Stop loop execution immediately, no state preservation 

21 - PAUSE: Save state with continuation prompt and exit (default) 

22 - SPAWN: Save state and spawn a new Claude session to continue 

23 """ 

24 

25 TERMINATE = "terminate" 

26 PAUSE = "pause" 

27 SPAWN = "spawn" 

28 

29 

30@dataclass 

31class HandoffResult: 

32 """Result from handling a handoff signal. 

33 

34 Attributes: 

35 behavior: The behavior that was applied 

36 continuation_prompt: The continuation prompt from the signal 

37 spawned_process: Popen object if spawn behavior was used 

38 """ 

39 

40 behavior: HandoffBehavior 

41 continuation_prompt: str | None 

42 spawned_process: subprocess.Popen[str] | None = None 

43 

44 

45class HandoffHandler: 

46 """Handle context handoff signals. 

47 

48 Provides configurable behavior for when handoff signals are detected 

49 in loop action output. 

50 

51 Example: 

52 handler = HandoffHandler(HandoffBehavior.PAUSE) 

53 result = handler.handle("fix-types", "Continue from iteration 5") 

54 # result.behavior == HandoffBehavior.PAUSE 

55 # State should be saved by caller 

56 """ 

57 

58 def __init__(self, behavior: HandoffBehavior = HandoffBehavior.PAUSE) -> None: 

59 """Initialize handler with behavior. 

60 

61 Args: 

62 behavior: How to handle handoff signals (default: pause) 

63 """ 

64 self.behavior = behavior 

65 

66 def handle(self, loop_name: str, continuation: str | None) -> HandoffResult: 

67 """Handle a detected handoff signal. 

68 

69 For PAUSE and SPAWN behaviors, the caller (executor) is responsible 

70 for saving state with the continuation prompt. 

71 

72 Args: 

73 loop_name: Name of the loop for spawn commands 

74 continuation: Continuation prompt from the signal 

75 

76 Returns: 

77 HandoffResult with behavior taken and any spawned process 

78 """ 

79 if self.behavior == HandoffBehavior.TERMINATE: 

80 return HandoffResult(self.behavior, continuation) 

81 

82 if self.behavior == HandoffBehavior.PAUSE: 

83 # State saving handled by executor 

84 return HandoffResult(self.behavior, continuation) 

85 

86 if self.behavior == HandoffBehavior.SPAWN: 

87 process = self._spawn_continuation(loop_name, continuation) 

88 return HandoffResult(self.behavior, continuation, process) 

89 

90 # Should never reach here due to enum exhaustiveness, 

91 # but satisfy type checker 

92 return HandoffResult(self.behavior, continuation) 

93 

94 def _spawn_continuation( 

95 self, loop_name: str, continuation: str | None 

96 ) -> subprocess.Popen[str]: 

97 """Spawn new Claude session to continue loop. 

98 

99 Creates a new Claude CLI process with a prompt instructing it 

100 to resume the loop execution. 

101 

102 Args: 

103 loop_name: Name of the loop to resume 

104 continuation: Continuation context from handoff 

105 

106 Returns: 

107 Popen object for the spawned process 

108 """ 

109 prompt_parts = [f"Continue loop execution. Run: ll-loop resume {loop_name}"] 

110 if continuation: 

111 prompt_parts.append(f"\n\n{continuation}") 

112 prompt = "".join(prompt_parts) 

113 

114 cmd = ["claude", "-p", prompt] 

115 return subprocess.Popen( 

116 cmd, 

117 text=True, 

118 start_new_session=True, 

119 stdout=subprocess.DEVNULL, 

120 stderr=subprocess.DEVNULL, 

121 stdin=subprocess.DEVNULL, 

122 )