Coverage for little_loops / cli / loop / testing.py: 85%

1"""ll-loop testing subcommands: test, simulate.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6from pathlib import Path 

7 

8from little_loops.cli.loop._helpers import load_loop 

9from little_loops.fsm.rate_limit_circuit import RateLimitCircuit 

10from little_loops.logger import Logger 

11 

12 

13def cmd_test( 

14 loop_name: str, 

15 args: argparse.Namespace, 

16 loops_dir: Path, 

17 logger: Logger, 

18) -> int: 

19 """Run a single test iteration to verify loop configuration. 

20 

21 Executes the target state's action and evaluation, then reports 

22 what the loop would do without actually transitioning further. 

23 """ 

24 from little_loops.fsm.evaluators import EvaluationResult, evaluate, evaluate_exit_code 

25 from little_loops.fsm.executor import DefaultActionRunner 

26 from little_loops.fsm.interpolation import InterpolationContext 

27 

28 try: 

29 fsm = load_loop(loop_name, loops_dir, logger) 

30 except FileNotFoundError as e: 

31 logger.error(str(e)) 

32 return 1 

33 except ValueError as e: 

34 logger.error(f"Validation error: {e}") 

35 return 1 

36 

37 # Determine target state 

38 target = args.state if args.state else fsm.initial 

39 if target not in fsm.states: 

40 logger.error(f"State '{target}' not found. Available: {', '.join(fsm.states)}") 

41 return 1 

42 state_config = fsm.states[target] 

43 

44 print(f"## Test Iteration: {loop_name}") 

45 print() 

46 print(f"State: {target}") 

47 

48 # If no action, report and exit 

49 if not state_config.action: 

50 print(f"State '{target}' has no action to test") 

51 print() 

52 print("\u2713 Loop structure is valid (no check action to execute)") 

53 return 0 

54 

55 action = state_config.action 

56 is_slash = action.startswith("/") or state_config.action_type in ( 

57 "prompt", 

58 "slash_command", 

59 ) 

60 

61 print(f"Action: {action}") 

62 print() 

63 

64 if is_slash: 

65 from little_loops.fsm.executor import ActionResult, SimulationActionRunner 

66 

67 exit_code_arg = getattr(args, "exit_code", None) 

68 if exit_code_arg is not None: 

69 sim_exit_code = exit_code_arg 

70 print(f"[SIMULATED] Using --exit-code {sim_exit_code}") 

71 else: 

72 sim_runner = SimulationActionRunner() 

73 sim_result = sim_runner.run(action, timeout=120, is_slash_command=True) 

74 sim_exit_code = sim_result.exit_code 

75 print() 

76 result = ActionResult( 

77 output=f"[simulated output for: {action}]", 

78 stderr="", 

79 exit_code=sim_exit_code, 

80 duration_ms=0, 

81 ) 

82 else: 

83 # Run the action 

84 runner = DefaultActionRunner() 

85 timeout = state_config.timeout or 120 

86 result = runner.run(action, timeout=timeout, is_slash_command=False) 

87 

88 print(f"Exit code: {result.exit_code}") 

89 

90 # Truncate output for display 

91 output_lines = result.output.strip().split("\n") 

92 if len(output_lines) > 10: 

93 extra = len(output_lines) - 10 

94 output_preview = "\n".join(output_lines[:10]) + f"\n... ({extra} more lines)" 

95 elif len(result.output) > 500: 

96 output_preview = result.output[:500] + "..." 

97 else: 

98 output_preview = result.output.strip() if result.output.strip() else "(empty)" 

99 

100 print(f"Output:\n{output_preview}") 

101 

102 if result.stderr: 

103 stderr_lines = result.stderr.strip().split("\n") 

104 if len(stderr_lines) > 5: 

105 extra = len(stderr_lines) - 5 

106 stderr_preview = "\n".join(stderr_lines[:5]) + f"\n... ({extra} more lines)" 

107 else: 

108 stderr_preview = result.stderr.strip() 

109 print(f"Stderr:\n{stderr_preview}") 

110 

111 print() 

112 

113 # Evaluate 

114 ctx = InterpolationContext() 

115 eval_result: EvaluationResult 

116 

117 if state_config.evaluate: 

118 eval_result = evaluate( 

119 config=state_config.evaluate, 

120 output=result.output, 

121 exit_code=result.exit_code, 

122 context=ctx, 

123 ) 

124 evaluator_type: str = state_config.evaluate.type 

125 else: 

126 # Default to exit_code evaluation 

127 eval_result = evaluate_exit_code(result.exit_code) 

128 evaluator_type = "exit_code (default)" 

129 

130 print(f"Evaluator: {evaluator_type}") 

131 print(f"Verdict: {eval_result.verdict.upper()}") 

132 

133 if eval_result.details: 

134 for key, value in eval_result.details.items(): 

135 if key != "exit_code" or evaluator_type != "exit_code (default)": 

136 print(f" {key}: {value}") 

137 

138 # Determine next state based on verdict 

139 verdict = eval_result.verdict 

140 next_state = None 

141 

142 if state_config.route: 

143 routes = state_config.route.routes 

144 if verdict in routes: 

145 next_state = routes[verdict] 

146 elif state_config.route.default: 

147 next_state = state_config.route.default 

148 else: 

149 if verdict == "yes" and state_config.on_yes: 

150 next_state = state_config.on_yes 

151 elif verdict == "no" and state_config.on_no: 

152 next_state = state_config.on_no 

153 elif verdict == "error" and state_config.on_error: 

154 next_state = state_config.on_error 

155 elif verdict in state_config.extra_routes: 

156 next_state = state_config.extra_routes[verdict] 

157 

158 print() 

159 if next_state: 

160 print(f"Would transition: {target} \u2192 {next_state}") 

161 else: 

162 print(f"Would transition: {target} \u2192 (no route for '{verdict}')") 

163 

164 # Summary 

165 print() 

166 has_error = eval_result.verdict == "error" or "error" in eval_result.details 

167 if has_error: 

168 print("\u26a0 Loop has issues - review the error details above") 

169 return 1 

170 else: 

171 print("\u2713 Loop appears to be configured correctly") 

172 return 0 

173 

174 

175def cmd_simulate( 

176 loop_name: str, 

177 args: argparse.Namespace, 

178 loops_dir: Path, 

179 logger: Logger, 

180 circuit: RateLimitCircuit | None = None, 

181) -> int: 

182 """Run interactive simulation of loop execution. 

183 

184 Traces through loop logic without executing commands, allowing users 

185 to verify state transitions and understand loop behavior. 

186 

187 The ``circuit`` kwarg lets tests redirect the shared 429 circuit-breaker 

188 state file (normally under ``.loops/tmp/``) to a ``tmp_path``; the CLI 

189 dispatcher does not pass one. 

190 """ 

191 from little_loops.fsm.executor import FSMExecutor, SimulationActionRunner 

192 

193 try: 

194 fsm = load_loop(loop_name, loops_dir, logger) 

195 except FileNotFoundError as e: 

196 logger.error(str(e)) 

197 return 1 

198 except ValueError as e: 

199 logger.error(f"Validation error: {e}") 

200 return 1 

201 

202 # Apply CLI overrides 

203 if args.max_iterations: 

204 fsm.max_iterations = args.max_iterations 

205 else: 

206 # Limit iterations for simulation safety (cap at 20 unless overridden) 

207 if fsm.max_iterations > 20: 

208 logger.info( 

209 f"Limiting simulation to 20 iterations (max_iterations: {fsm.max_iterations})" 

210 ) 

211 fsm.max_iterations = 20 

212 

213 # Create simulation runner 

214 sim_runner = SimulationActionRunner(scenario=args.scenario) 

215 

216 # Track simulation state 

217 states_visited: list[str] = [] 

218 

219 def simulation_callback(event: dict) -> None: 

220 """Display simulation progress.""" 

221 event_type = event.get("event") 

222 

223 if event_type == "state_enter": 

224 iteration = event.get("iteration", 0) 

225 state = event.get("state", "") 

226 states_visited.append(state) 

227 print() 

228 print(f"[{iteration}] State: {state}") 

229 

230 elif event_type == "action_start": 

231 action = event.get("action", "") 

232 action_display = action[:70] + "..." if len(action) > 70 else action 

233 print(f" Action: {action_display}") 

234 

235 elif event_type == "evaluate": 

236 evaluator = event.get("type", "exit_code") 

237 verdict = event.get("verdict", "") 

238 print(f" Evaluator: {evaluator}") 

239 print(f" Result: {verdict.upper()}") 

240 

241 elif event_type == "route": 

242 from_state = event.get("from", "") 

243 to_state = event.get("to", "") 

244 print(f" Transition: {from_state} \u2192 {to_state}") 

245 

246 # Print header 

247 mode_str = f"scenario={args.scenario}" if args.scenario else "interactive" 

248 print(f"=== SIMULATION: {fsm.name} ({mode_str}) ===") 

249 

250 # Run simulation 

251 executor = FSMExecutor( 

252 fsm, 

253 event_callback=simulation_callback, 

254 action_runner=sim_runner, 

255 circuit=circuit, 

256 ) 

257 result = executor.run() 

258 

259 # Print summary 

260 print() 

261 print("=== Summary ===") 

262 arrow = " \u2192 " 

263 print(f"States visited: {arrow.join(states_visited)}") 

264 print(f"Iterations: {result.iterations}") 

265 print(f"Would have executed {len(sim_runner.calls)} commands") 

266 print(f"Terminated by: {result.terminated_by}") 

267 

268 return 0