Coverage for little_loops / fsm / validation.py: 80%

1"""FSM loop validation logic. 

2 

3This module provides validation for FSM loop definitions, ensuring 

4structural correctness and catching common configuration errors. 

5 

6Validation checks: 

7- Initial state exists in states dict 

8- All referenced states exist 

9- At least one terminal state 

10- Evaluator configs have required fields for their type 

11- No conflicting routing (shorthand vs full route) 

12- Numeric fields in valid ranges (max_iterations > 0, backoff >= 0, timeout > 0) 

13""" 

14 

15from __future__ import annotations 

16 

17import logging 

18from collections import deque 

19from dataclasses import dataclass 

20from enum import Enum 

21from pathlib import Path 

22from typing import Any 

23 

24import yaml 

25 

26from little_loops.fsm.fragments import resolve_fragments 

27from little_loops.fsm.schema import EvaluateConfig, FSMLoop, StateConfig 

28 

29logger = logging.getLogger(__name__) 

30 

31 

32class ValidationSeverity(Enum): 

33 """Severity level for validation issues.""" 

34 

35 ERROR = "error" 

36 WARNING = "warning" 

37 

38 

39@dataclass 

40class ValidationError: 

41 """Structured validation error. 

42 

43 Attributes: 

44 message: Human-readable error description 

45 path: Path to the problematic element (e.g., "states.check.route") 

46 severity: Error severity (error or warning) 

47 """ 

48 

49 message: str 

50 path: str | None = None 

51 severity: ValidationSeverity = ValidationSeverity.ERROR 

52 

53 def __str__(self) -> str: 

54 """Format error for display.""" 

55 prefix = f"[{self.severity.value.upper()}]" 

56 if self.path: 

57 return f"{prefix} {self.path}: {self.message}" 

58 return f"{prefix} {self.message}" 

59 

60 

61# Evaluator type to required fields mapping 

62EVALUATOR_REQUIRED_FIELDS: dict[str, list[str]] = { 

63 "exit_code": [], 

64 "output_numeric": ["operator", "target"], 

65 "output_json": ["path", "operator", "target"], 

66 "output_contains": ["pattern"], 

67 "convergence": ["target"], 

68 "diff_stall": [], 

69 "llm_structured": [], 

70 "mcp_result": [], 

71} 

72 

73# Valid comparison operators 

74VALID_OPERATORS = {"eq", "ne", "lt", "le", "gt", "ge"} 

75 

76# All top-level keys recognized by FSMLoop.from_dict() 

77KNOWN_TOP_LEVEL_KEYS: frozenset[str] = frozenset( 

78 { 

79 "name", 

80 "description", 

81 "initial", 

82 "states", 

83 "context", 

84 "scope", 

85 "max_iterations", 

86 "backoff", 

87 "timeout", 

88 "default_timeout", 

89 "maintain", 

90 "llm", 

91 "on_handoff", 

92 "input_key", 

93 "config", 

94 "category", 

95 "labels", 

96 "import", 

97 "fragments", 

98 } 

99) 

100 

101 

102def _validate_evaluator(state_name: str, evaluate: EvaluateConfig) -> list[ValidationError]: 

103 """Validate evaluator configuration for type-specific requirements. 

104 

105 Args: 

106 state_name: Name of the state containing this evaluator 

107 evaluate: The evaluator configuration to validate 

108 

109 Returns: 

110 List of validation errors found 

111 """ 

112 errors: list[ValidationError] = [] 

113 path = f"states.{state_name}.evaluate" 

114 

115 # Check that evaluator type is recognized 

116 valid_types = set(EVALUATOR_REQUIRED_FIELDS.keys()) 

117 if evaluate.type not in valid_types: 

118 errors.append( 

119 ValidationError( 

120 message=f"Unknown evaluator type '{evaluate.type}'. " 

121 f"Must be one of: {', '.join(sorted(valid_types))}", 

122 path=path, 

123 ) 

124 ) 

125 return errors # Can't check required fields for unknown type 

126 

127 # Check required fields for evaluator type 

128 required = EVALUATOR_REQUIRED_FIELDS.get(evaluate.type, []) 

129 for field_name in required: 

130 value = getattr(evaluate, field_name, None) 

131 if value is None: 

132 errors.append( 

133 ValidationError( 

134 message=f"Evaluator type '{evaluate.type}' requires '{field_name}' field", 

135 path=path, 

136 ) 

137 ) 

138 

139 # Validate operator if present 

140 if evaluate.operator is not None and evaluate.operator not in VALID_OPERATORS: 

141 errors.append( 

142 ValidationError( 

143 message=f"Invalid operator '{evaluate.operator}'. " 

144 f"Must be one of: {', '.join(sorted(VALID_OPERATORS))}", 

145 path=f"{path}.operator", 

146 ) 

147 ) 

148 

149 # Validate convergence-specific fields 

150 if evaluate.type == "convergence": 

151 if evaluate.direction not in ("minimize", "maximize"): 

152 errors.append( 

153 ValidationError( 

154 message=f"Invalid direction '{evaluate.direction}'. " 

155 "Must be 'minimize' or 'maximize'", 

156 path=f"{path}.direction", 

157 ) 

158 ) 

159 # Only validate tolerance if it's a numeric value (not an interpolation string) 

160 if ( 

161 evaluate.tolerance is not None 

162 and isinstance(evaluate.tolerance, (int, float)) 

163 and evaluate.tolerance < 0 

164 ): 

165 errors.append( 

166 ValidationError( 

167 message="Tolerance cannot be negative", 

168 path=f"{path}.tolerance", 

169 ) 

170 ) 

171 

172 # Validate llm_structured-specific fields 

173 if evaluate.type == "llm_structured": 

174 if evaluate.min_confidence < 0 or evaluate.min_confidence > 1: 

175 errors.append( 

176 ValidationError( 

177 message="min_confidence must be between 0 and 1", 

178 path=f"{path}.min_confidence", 

179 ) 

180 ) 

181 

182 # Validate diff_stall-specific fields 

183 if evaluate.type == "diff_stall": 

184 if evaluate.max_stall < 1: 

185 errors.append( 

186 ValidationError( 

187 message="max_stall must be >= 1", 

188 path=f"{path}.max_stall", 

189 ) 

190 ) 

191 

192 return errors 

193 

194 

195def _validate_state_action(state_name: str, state: StateConfig) -> list[ValidationError]: 

196 """Validate state action configuration. 

197 

198 Args: 

199 state_name: Name of the state to validate 

200 state: The state configuration to validate 

201 

202 Returns: 

203 List of validation errors found 

204 """ 

205 errors: list[ValidationError] = [] 

206 path = f"states.{state_name}" 

207 

208 # params field is only valid for mcp_tool states 

209 if state.params and state.action_type != "mcp_tool": 

210 errors.append( 

211 ValidationError( 

212 message="'params' field is only valid when action_type is 'mcp_tool'", 

213 path=f"{path}.params", 

214 ) 

215 ) 

216 

217 # loop and action are mutually exclusive 

218 if state.loop is not None and state.action is not None: 

219 errors.append( 

220 ValidationError( 

221 message="'loop' and 'action' are mutually exclusive — " 

222 "a sub-loop state cannot also have an action", 

223 path=f"{path}", 

224 ) 

225 ) 

226 

227 return errors 

228 

229 

230def _validate_state_routing(state_name: str, state: StateConfig) -> list[ValidationError]: 

231 """Validate state routing configuration. 

232 

233 Checks for conflicting routing definitions (shorthand vs full route). 

234 

235 Args: 

236 state_name: Name of the state to validate 

237 state: The state configuration to validate 

238 

239 Returns: 

240 List of validation errors/warnings found 

241 """ 

242 errors: list[ValidationError] = [] 

243 path = f"states.{state_name}" 

244 

245 has_shorthand = ( 

246 state.on_yes is not None 

247 or state.on_no is not None 

248 or state.on_error is not None 

249 or state.on_partial is not None 

250 or state.on_blocked is not None 

251 or bool(state.extra_routes) 

252 ) 

253 has_route = state.route is not None 

254 

255 # Warn about conflicting definitions 

256 if has_shorthand and has_route: 

257 errors.append( 

258 ValidationError( 

259 message="Both shorthand routing (on_yes/on_no/on_error) " 

260 "and full route table defined. Route table will take precedence.", 

261 path=path, 

262 severity=ValidationSeverity.WARNING, 

263 ) 

264 ) 

265 

266 # Check for no valid transition definition 

267 has_next = state.next is not None 

268 has_terminal = state.terminal 

269 has_loop = state.loop is not None 

270 

271 if not has_shorthand and not has_route and not has_next and not has_terminal and not has_loop: 

272 errors.append( 

273 ValidationError( 

274 message="State has no transition defined. Add routing, 'next', " 

275 "or mark as 'terminal: true'", 

276 path=path, 

277 ) 

278 ) 

279 

280 # Validate retry field pairing: max_retries requires on_retry_exhausted and vice versa 

281 if state.max_retries is not None and state.on_retry_exhausted is None: 

282 errors.append( 

283 ValidationError( 

284 message="'max_retries' requires 'on_retry_exhausted' to also be set", 

285 path=path, 

286 ) 

287 ) 

288 if state.on_retry_exhausted is not None and state.max_retries is None: 

289 errors.append( 

290 ValidationError( 

291 message="'on_retry_exhausted' requires 'max_retries' to also be set", 

292 path=path, 

293 ) 

294 ) 

295 if state.max_retries is not None and state.max_retries < 1: 

296 errors.append( 

297 ValidationError( 

298 message=f"'max_retries' must be >= 1, got {state.max_retries}", 

299 path=path, 

300 ) 

301 ) 

302 

303 return errors 

304 

305 

306def validate_fsm(fsm: FSMLoop) -> list[ValidationError]: 

307 """Validate FSM structure and return list of errors. 

308 

309 Performs comprehensive validation: 

310 - Initial state exists 

311 - All referenced states exist 

312 - At least one terminal state 

313 - Evaluator configurations are valid 

314 - Routing configurations are valid 

315 - Numeric fields are in valid ranges (max_iterations > 0, backoff >= 0, timeout > 0) 

316 

317 Args: 

318 fsm: The FSM loop to validate 

319 

320 Returns: 

321 List of validation errors (empty if valid) 

322 """ 

323 errors: list[ValidationError] = [] 

324 defined_states = fsm.get_all_state_names() 

325 

326 # Check initial state exists 

327 if fsm.initial not in defined_states: 

328 errors.append( 

329 ValidationError( 

330 message=f"Initial state '{fsm.initial}' not found in states", 

331 path="initial", 

332 ) 

333 ) 

334 

335 # Check at least one terminal state 

336 terminal_states = fsm.get_terminal_states() 

337 if not terminal_states: 

338 errors.append( 

339 ValidationError( 

340 message="No terminal state defined. At least one state must have 'terminal: true'", 

341 path="states", 

342 ) 

343 ) 

344 

345 # Validate each state 

346 for state_name, state in fsm.states.items(): 

347 # Check all referenced states exist 

348 refs = state.get_referenced_states() 

349 for ref in refs: 

350 # $current is a special token for retry 

351 if ref != "$current" and ref not in defined_states: 

352 errors.append( 

353 ValidationError( 

354 message=f"References unknown state '{ref}'", 

355 path=f"states.{state_name}", 

356 ) 

357 ) 

358 

359 # Validate action configuration 

360 errors.extend(_validate_state_action(state_name, state)) 

361 

362 # Validate evaluator if present 

363 if state.evaluate is not None: 

364 errors.extend(_validate_evaluator(state_name, state.evaluate)) 

365 

366 # Validate routing configuration 

367 errors.extend(_validate_state_routing(state_name, state)) 

368 

369 # Check numeric field ranges 

370 if fsm.max_iterations <= 0: 

371 errors.append( 

372 ValidationError( 

373 message=f"max_iterations must be > 0, got {fsm.max_iterations}", 

374 path="max_iterations", 

375 ) 

376 ) 

377 if fsm.backoff is not None and fsm.backoff < 0: 

378 errors.append( 

379 ValidationError( 

380 message=f"backoff must be >= 0, got {fsm.backoff}", 

381 path="backoff", 

382 ) 

383 ) 

384 if fsm.timeout is not None and fsm.timeout <= 0: 

385 errors.append( 

386 ValidationError( 

387 message=f"timeout must be > 0, got {fsm.timeout}", 

388 path="timeout", 

389 ) 

390 ) 

391 if fsm.llm.max_tokens <= 0: 

392 errors.append( 

393 ValidationError( 

394 message=f"llm.max_tokens must be > 0, got {fsm.llm.max_tokens}", 

395 path="llm.max_tokens", 

396 ) 

397 ) 

398 if fsm.llm.timeout <= 0: 

399 errors.append( 

400 ValidationError( 

401 message=f"llm.timeout must be > 0, got {fsm.llm.timeout}", 

402 path="llm.timeout", 

403 ) 

404 ) 

405 

406 # Check for unreachable states (warning only) 

407 reachable = _find_reachable_states(fsm) 

408 unreachable = defined_states - reachable 

409 for state_name in unreachable: 

410 errors.append( 

411 ValidationError( 

412 message="State is not reachable from initial state", 

413 path=f"states.{state_name}", 

414 severity=ValidationSeverity.WARNING, 

415 ) 

416 ) 

417 

418 return errors 

419 

420 

421def _find_reachable_states(fsm: FSMLoop) -> set[str]: 

422 """Find all states reachable from the initial state. 

423 

424 Uses breadth-first search to find all reachable states. 

425 

426 Args: 

427 fsm: The FSM loop to analyze 

428 

429 Returns: 

430 Set of reachable state names 

431 """ 

432 reachable: set[str] = set() 

433 to_visit: deque[str] = deque([fsm.initial]) 

434 

435 while to_visit: 

436 current = to_visit.popleft() 

437 if current in reachable or current not in fsm.states: 

438 continue 

439 

440 reachable.add(current) 

441 state = fsm.states[current] 

442 refs = state.get_referenced_states() 

443 

444 for ref in refs: 

445 if ref != "$current" and ref not in reachable: 

446 to_visit.append(ref) 

447 

448 return reachable 

449 

450 

451def load_and_validate(path: Path) -> tuple[FSMLoop, list[ValidationError]]: 

452 """Load YAML file and validate FSM structure. 

453 

454 Args: 

455 path: Path to the YAML file to load 

456 

457 Returns: 

458 Tuple of (validated FSMLoop instance, list of WARNING-severity ValidationErrors) 

459 

460 Raises: 

461 FileNotFoundError: If the file doesn't exist 

462 yaml.YAMLError: If the file is not valid YAML 

463 ValueError: If validation fails (contains error details) 

464 """ 

465 if not path.exists(): 

466 raise FileNotFoundError(f"FSM file not found: {path}") 

467 

468 with open(path) as f: 

469 data: dict[str, Any] = yaml.safe_load(f) 

470 

471 if not isinstance(data, dict): 

472 raise ValueError(f"FSM file must contain a YAML mapping, got {type(data)}") 

473 

474 # Check required fields before parsing 

475 missing = [] 

476 for field in ["name", "initial", "states"]: 

477 if field not in data: 

478 missing.append(field) 

479 

480 if missing: 

481 raise ValueError(f"FSM file missing required fields: {', '.join(missing)}") 

482 

483 # Check for unknown top-level keys before parsing 

484 unknown_key_warnings: list[ValidationError] = [] 

485 unknown = set(data.keys()) - KNOWN_TOP_LEVEL_KEYS 

486 if unknown: 

487 unknown_key_warnings.append( 

488 ValidationError( 

489 path="<root>", 

490 message=f"Unknown top-level keys: {', '.join(sorted(unknown))}", 

491 severity=ValidationSeverity.WARNING, 

492 ) 

493 ) 

494 

495 # Resolve fragment libraries before parsing into dataclass 

496 data = resolve_fragments(data, path.parent) 

497 

498 # Parse into dataclass 

499 fsm = FSMLoop.from_dict(data) 

500 

501 # Validate 

502 errors = validate_fsm(fsm) 

503 

504 # Filter to errors only (not warnings) for raising 

505 error_list = [e for e in errors if e.severity == ValidationSeverity.ERROR] 

506 

507 if error_list: 

508 error_messages = "\n ".join(str(e) for e in error_list) 

509 raise ValueError(f"FSM validation failed:\n {error_messages}") 

510 

511 # Collect all warnings (unknown-key warnings + structural warnings) 

512 struct_warnings = [e for e in errors if e.severity == ValidationSeverity.WARNING] 

513 all_warnings = unknown_key_warnings + struct_warnings 

514 for warning in all_warnings: 

515 logger.warning(str(warning)) 

516 

517 return fsm, all_warnings