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 step executions (state enters) 

24 terminated_by: Reason for termination. Values: "terminal", "max_steps" (step cap reached; 

25 legacy "max_iterations" renamed), "max_iterations_reached" (full-pass cap reached), 

26 "timeout", "signal", "error", "handoff", "cycle_detected", "stall_detected". 

27 duration_ms: Total execution time in milliseconds 

28 captured: All captured variable values 

29 error: Error message if terminated_by is "error" 

30 handoff: True if execution stopped due to handoff signal 

31 continuation_prompt: Continuation context from handoff signal 

32 """ 

33 

34 final_state: str 

35 iterations: int 

36 terminated_by: str # "terminal", "max_steps", "max_iterations_reached", "timeout", "signal", "error", "handoff", "cycle_detected" 

37 duration_ms: int 

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

39 error: str | None = None 

40 handoff: bool = False 

41 continuation_prompt: str | None = None 

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

43 

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

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

46 result: dict[str, Any] = { 

47 "final_state": self.final_state, 

48 "iterations": self.iterations, 

49 "terminated_by": self.terminated_by, 

50 "duration_ms": self.duration_ms, 

51 "captured": self.captured, 

52 } 

53 if self.error is not None: 

54 result["error"] = self.error 

55 if self.handoff: 

56 result["handoff"] = self.handoff 

57 if self.continuation_prompt is not None: 

58 result["continuation_prompt"] = self.continuation_prompt 

59 if self.messages: 

60 result["messages"] = self.messages 

61 return result 

62 

63 

64@dataclass 

65class ActionResult: 

66 """Result from action execution. 

67 

68 Attributes: 

69 output: stdout from the action 

70 stderr: stderr from the action 

71 exit_code: Exit code from the action 

72 duration_ms: Execution time in milliseconds 

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

74 """ 

75 

76 output: str 

77 stderr: str 

78 exit_code: int 

79 duration_ms: int 

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

81 

82 

83# Type for event callback 

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

85 

86# Type for evaluator functions 

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

88Evaluator = Callable[ 

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

90 "EvaluationResult", 

91]