Coverage for little_loops / fsm / types.py: 72%

1"""FSM result types for loop and action execution.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from dataclasses import dataclass, field 

7from typing import TYPE_CHECKING, Any 

8 

9from little_loops.subprocess_utils import TokenUsage 

10 

11if TYPE_CHECKING: 

12 from little_loops.fsm.evaluators import EvaluationResult 

13 from little_loops.fsm.interpolation import InterpolationContext 

14 from little_loops.fsm.schema import EvaluateConfig 

15 

16 

17@dataclass 

18class ExecutionResult: 

19 """Result from FSM execution. 

20 

21 Attributes: 

22 final_state: Name of the state when execution stopped 

23 iterations: Total iterations executed 

24 terminated_by: Reason for termination (terminal, max_iterations, timeout, signal, error, handoff) 

25 duration_ms: Total execution time in milliseconds 

26 captured: All captured variable values 

27 error: Error message if terminated_by is "error" 

28 handoff: True if execution stopped due to handoff signal 

29 continuation_prompt: Continuation context from handoff signal 

30 """ 

31 

32 final_state: str 

33 iterations: int 

34 terminated_by: str # "terminal", "max_iterations", "timeout", "signal", "error", "handoff", "cycle_detected" 

35 duration_ms: int 

36 captured: dict[str, dict[str, Any]] 

37 error: str | None = None 

38 handoff: bool = False 

39 continuation_prompt: str | None = None 

40 messages: list[str] = field(default_factory=list) 

41 

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

43 """Convert to dictionary for JSON serialization.""" 

44 result: dict[str, Any] = { 

45 "final_state": self.final_state, 

46 "iterations": self.iterations, 

47 "terminated_by": self.terminated_by, 

48 "duration_ms": self.duration_ms, 

49 "captured": self.captured, 

50 } 

51 if self.error is not None: 

52 result["error"] = self.error 

53 if self.handoff: 

54 result["handoff"] = self.handoff 

55 if self.continuation_prompt is not None: 

56 result["continuation_prompt"] = self.continuation_prompt 

57 if self.messages: 

58 result["messages"] = self.messages 

59 return result 

60 

61 

62@dataclass 

63class ActionResult: 

64 """Result from action execution. 

65 

66 Attributes: 

67 output: stdout from the action 

68 stderr: stderr from the action 

69 exit_code: Exit code from the action 

70 duration_ms: Execution time in milliseconds 

71 usage_events: Token usage events from host-CLI invocations (empty for shell actions) 

72 """ 

73 

74 output: str 

75 stderr: str 

76 exit_code: int 

77 duration_ms: int 

78 usage_events: list[TokenUsage] = field(default_factory=list) 

79 

80 

81# Type for event callback 

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

83 

84# Type for evaluator functions 

85# Parameter order: config, output, exit_code, context — matches evaluate() call signature 

86Evaluator = Callable[ 

87 ["EvaluateConfig", str, int, "InterpolationContext"], 

88 "EvaluationResult", 

89]