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

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

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from dataclasses import dataclass 

7from typing import TYPE_CHECKING, Any 

8 

9if TYPE_CHECKING: 

10 from little_loops.fsm.evaluators import EvaluationResult 

11 from little_loops.fsm.interpolation import InterpolationContext 

12 from little_loops.fsm.schema import EvaluateConfig 

13 

14 

15@dataclass 

16class ExecutionResult: 

17 """Result from FSM execution. 

18 

19 Attributes: 

20 final_state: Name of the state when execution stopped 

21 iterations: Total iterations executed 

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

23 duration_ms: Total execution time in milliseconds 

24 captured: All captured variable values 

25 error: Error message if terminated_by is "error" 

26 handoff: True if execution stopped due to handoff signal 

27 continuation_prompt: Continuation context from handoff signal 

28 """ 

29 

30 final_state: str 

31 iterations: int 

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

33 duration_ms: int 

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

35 error: str | None = None 

36 handoff: bool = False 

37 continuation_prompt: str | None = None 

38 

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

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

41 result: dict[str, Any] = { 

42 "final_state": self.final_state, 

43 "iterations": self.iterations, 

44 "terminated_by": self.terminated_by, 

45 "duration_ms": self.duration_ms, 

46 "captured": self.captured, 

47 } 

48 if self.error is not None: 

49 result["error"] = self.error 

50 if self.handoff: 

51 result["handoff"] = self.handoff 

52 if self.continuation_prompt is not None: 

53 result["continuation_prompt"] = self.continuation_prompt 

54 return result 

55 

56 

57@dataclass 

58class ActionResult: 

59 """Result from action execution. 

60 

61 Attributes: 

62 output: stdout from the action 

63 stderr: stderr from the action 

64 exit_code: Exit code from the action 

65 duration_ms: Execution time in milliseconds 

66 """ 

67 

68 output: str 

69 stderr: str 

70 exit_code: int 

71 duration_ms: int 

72 

73 

74# Type for event callback 

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

76 

77# Type for evaluator functions 

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

79Evaluator = Callable[ 

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

81 "EvaluationResult", 

82]