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

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 os 

13import subprocess 

14from dataclasses import dataclass 

15from enum import Enum 

16 

17from little_loops.host_runner import resolve_host 

18 

19 

20class HandoffBehavior(Enum): 

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

22 

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

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

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

26 """ 

27 

28 TERMINATE = "terminate" 

29 PAUSE = "pause" 

30 SPAWN = "spawn" 

31 

32 

33@dataclass 

34class HandoffResult: 

35 """Result from handling a handoff signal. 

36 

37 Attributes: 

38 behavior: The behavior that was applied 

39 continuation_prompt: The continuation prompt from the signal 

40 spawned_process: Popen object if spawn behavior was used 

41 """ 

42 

43 behavior: HandoffBehavior 

44 continuation_prompt: str | None 

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

46 

47 

48class HandoffHandler: 

49 """Handle context handoff signals. 

50 

51 Provides configurable behavior for when handoff signals are detected 

52 in loop action output. 

53 

54 Example: 

55 handler = HandoffHandler(HandoffBehavior.PAUSE) 

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

57 # result.behavior == HandoffBehavior.PAUSE 

58 # State should be saved by caller 

59 """ 

60 

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

62 """Initialize handler with behavior. 

63 

64 Args: 

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

66 """ 

67 self.behavior = behavior 

68 

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

70 """Handle a detected handoff signal. 

71 

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

73 for saving state with the continuation prompt. 

74 

75 Args: 

76 loop_name: Name of the loop for spawn commands 

77 continuation: Continuation prompt from the signal 

78 

79 Returns: 

80 HandoffResult with behavior taken and any spawned process 

81 """ 

82 if self.behavior == HandoffBehavior.TERMINATE: 

83 return HandoffResult(self.behavior, continuation) 

84 

85 if self.behavior == HandoffBehavior.PAUSE: 

86 # State saving handled by executor 

87 return HandoffResult(self.behavior, continuation) 

88 

89 if self.behavior == HandoffBehavior.SPAWN: 

90 process = self._spawn_continuation(loop_name, continuation) 

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

92 

93 # Should never reach here due to enum exhaustiveness, 

94 # but satisfy type checker 

95 return HandoffResult(self.behavior, continuation) 

96 

97 def _spawn_continuation( 

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

99 ) -> subprocess.Popen[str]: 

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

101 

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

103 to resume the loop execution. 

104 

105 Args: 

106 loop_name: Name of the loop to resume 

107 continuation: Continuation context from handoff 

108 

109 Returns: 

110 Popen object for the spawned process 

111 """ 

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

113 if continuation: 

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

115 prompt = "".join(prompt_parts) 

116 

117 invocation = resolve_host().build_detached(prompt=prompt) 

118 # Legacy argv had no perm-skip; strip it for no-behavior-change refactor. 

119 args = [a for a in invocation.args if a != "--dangerously-skip-permissions"] 

120 return subprocess.Popen( 

121 [invocation.binary, *args], 

122 text=True, 

123 start_new_session=True, 

124 stdout=subprocess.DEVNULL, 

125 stderr=subprocess.DEVNULL, 

126 stdin=subprocess.DEVNULL, 

127 env={**os.environ, **invocation.env}, 

128 )