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

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 — in simulation, -n/--max-steps caps steps for brevity 

203 if getattr(args, "max_steps", None): 

204 fsm.max_steps = args.max_steps 

205 if getattr(args, "max_iterations", None): 

206 fsm.max_iterations = args.max_iterations 

207 else: 

208 # Limit steps for simulation safety (cap at 20 unless overridden) 

209 if fsm.max_steps > 20: 

210 logger.info(f"Limiting simulation to 20 steps (max_steps: {fsm.max_steps})") 

211 fsm.max_steps = 20 

212 

213 # Inject runner-managed context variables so ${context.run_dir} resolves during 

214 # simulation — the real runner does this in run.py before FSMExecutor is created. 

215 if "run_dir" not in fsm.context: 

216 fsm.context["run_dir"] = str(loops_dir / "runs" / f"{loop_name}-simulate") + "/" 

217 if "input_hash" not in fsm.context and isinstance(fsm.context.get("input"), str): 

218 import hashlib 

219 

220 fsm.context["input_hash"] = hashlib.sha256(fsm.context["input"].encode()).hexdigest()[:12] 

221 

222 # Create simulation runner 

223 sim_runner = SimulationActionRunner(scenario=args.scenario) 

224 

225 # Track simulation state 

226 states_visited: list[str] = [] 

227 

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

229 """Display simulation progress.""" 

230 event_type = event.get("event") 

231 

232 if event_type == "state_enter": 

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

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

235 states_visited.append(state) 

236 print() 

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

238 

239 elif event_type == "action_start": 

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

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

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

243 

244 elif event_type == "evaluate": 

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

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

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

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

249 

250 elif event_type == "route": 

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

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

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

254 

255 # Print header 

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

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

258 

259 # Run simulation 

260 executor = FSMExecutor( 

261 fsm, 

262 event_callback=simulation_callback, 

263 action_runner=sim_runner, 

264 circuit=circuit, 

265 ) 

266 result = executor.run() 

267 

268 # Print summary 

269 print() 

270 print("=== Summary ===") 

271 arrow = " \u2192 " 

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

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

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

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

276 

277 return 0