Coverage for little_loops / fsm / validation.py: 13%
248 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-11 00:29 -0500
1"""FSM loop validation logic.
3This module provides validation for FSM loop definitions, ensuring
4structural correctness and catching common configuration errors.
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"""
15from __future__ import annotations
17import logging
18from collections import deque
19from dataclasses import dataclass
20from enum import Enum
21from pathlib import Path
22from typing import Any
24import yaml
26from little_loops.fsm.fragments import resolve_flow, resolve_fragments, resolve_inheritance
27from little_loops.fsm.schema import EvaluateConfig, FSMLoop, ParameterSpec, StateConfig
29logger = logging.getLogger(__name__)
32class ValidationSeverity(Enum):
33 """Severity level for validation issues."""
35 ERROR = "error"
36 WARNING = "warning"
39@dataclass
40class ValidationError:
41 """Structured validation error.
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 """
49 message: str
50 path: str | None = None
51 severity: ValidationSeverity = ValidationSeverity.ERROR
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}"
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 "harbor_scorer": [],
72}
74# Valid comparison operators
75VALID_OPERATORS = {"eq", "ne", "lt", "le", "gt", "ge"}
77# All top-level keys recognized by FSMLoop.from_dict()
78KNOWN_TOP_LEVEL_KEYS: frozenset[str] = frozenset(
79 {
80 "name",
81 "description",
82 "initial",
83 "states",
84 "context",
85 "parameters",
86 "scope",
87 "max_iterations",
88 "max_edge_revisits",
89 "backoff",
90 "timeout",
91 "default_timeout",
92 "maintain",
93 "llm",
94 "on_handoff",
95 "input_key",
96 "config",
97 "category",
98 "labels",
99 "commands",
100 "import",
101 "fragments",
102 "from",
103 "flow",
104 "state_defs",
105 }
106)
108# Valid parameter types for the 'parameters:' block
109VALID_PARAMETER_TYPES: frozenset[str] = frozenset(
110 {"string", "integer", "number", "boolean", "enum", "path"}
111)
114def _validate_evaluator(state_name: str, evaluate: EvaluateConfig) -> list[ValidationError]:
115 """Validate evaluator configuration for type-specific requirements.
117 Args:
118 state_name: Name of the state containing this evaluator
119 evaluate: The evaluator configuration to validate
121 Returns:
122 List of validation errors found
123 """
124 errors: list[ValidationError] = []
125 path = f"states.{state_name}.evaluate"
127 # Check that evaluator type is recognized
128 valid_types = set(EVALUATOR_REQUIRED_FIELDS.keys())
129 if evaluate.type not in valid_types:
130 errors.append(
131 ValidationError(
132 message=f"Unknown evaluator type '{evaluate.type}'. "
133 f"Must be one of: {', '.join(sorted(valid_types))}",
134 path=path,
135 )
136 )
137 return errors # Can't check required fields for unknown type
139 # Check required fields for evaluator type
140 required = EVALUATOR_REQUIRED_FIELDS.get(evaluate.type, [])
141 for field_name in required:
142 value = getattr(evaluate, field_name, None)
143 if value is None:
144 errors.append(
145 ValidationError(
146 message=f"Evaluator type '{evaluate.type}' requires '{field_name}' field",
147 path=path,
148 )
149 )
151 # Validate operator if present
152 if evaluate.operator is not None and evaluate.operator not in VALID_OPERATORS:
153 errors.append(
154 ValidationError(
155 message=f"Invalid operator '{evaluate.operator}'. "
156 f"Must be one of: {', '.join(sorted(VALID_OPERATORS))}",
157 path=f"{path}.operator",
158 )
159 )
161 # Validate convergence-specific fields
162 if evaluate.type == "convergence":
163 if evaluate.direction not in ("minimize", "maximize"):
164 errors.append(
165 ValidationError(
166 message=f"Invalid direction '{evaluate.direction}'. "
167 "Must be 'minimize' or 'maximize'",
168 path=f"{path}.direction",
169 )
170 )
171 # Only validate tolerance if it's a numeric value (not an interpolation string)
172 if (
173 evaluate.tolerance is not None
174 and isinstance(evaluate.tolerance, (int, float))
175 and evaluate.tolerance < 0
176 ):
177 errors.append(
178 ValidationError(
179 message="Tolerance cannot be negative",
180 path=f"{path}.tolerance",
181 )
182 )
184 # Validate llm_structured-specific fields
185 if evaluate.type == "llm_structured":
186 if evaluate.min_confidence < 0 or evaluate.min_confidence > 1:
187 errors.append(
188 ValidationError(
189 message="min_confidence must be between 0 and 1",
190 path=f"{path}.min_confidence",
191 )
192 )
194 # Validate diff_stall-specific fields
195 if evaluate.type == "diff_stall":
196 if evaluate.max_stall < 1:
197 errors.append(
198 ValidationError(
199 message="max_stall must be >= 1",
200 path=f"{path}.max_stall",
201 )
202 )
204 return errors
207def _validate_parameters(fsm: FSMLoop) -> list[ValidationError]:
208 """Validate the loop's top-level parameters: block.
210 Args:
211 fsm: The FSM loop to validate
213 Returns:
214 List of validation errors found
215 """
216 errors: list[ValidationError] = []
218 for param_name, param_spec in fsm.parameters.items():
219 path = f"parameters.{param_name}"
221 if param_spec.type not in VALID_PARAMETER_TYPES:
222 errors.append(
223 ValidationError(
224 message=(
225 f"Unknown parameter type '{param_spec.type}'. "
226 f"Must be one of: {', '.join(sorted(VALID_PARAMETER_TYPES))}"
227 ),
228 path=path,
229 )
230 )
232 if param_spec.type == "enum" and not param_spec.values:
233 errors.append(
234 ValidationError(
235 message="Parameter type 'enum' requires a 'values' list",
236 path=path,
237 )
238 )
240 if param_spec.required and param_spec.default is not None:
241 errors.append(
242 ValidationError(
243 message="Parameter cannot be both 'required: true' and have a 'default' value",
244 path=path,
245 )
246 )
248 return errors
251def _check_param_type(value: Any, spec: ParameterSpec) -> str | None:
252 """Return an error message if value does not match spec.type, else None."""
253 if spec.type == "string" and not isinstance(value, str):
254 return f"expected string, got {type(value).__name__}"
255 if spec.type == "integer" and not isinstance(value, int):
256 return f"expected integer, got {type(value).__name__}"
257 if spec.type == "number" and not isinstance(value, (int, float)):
258 return f"expected number, got {type(value).__name__}"
259 if spec.type == "boolean" and not isinstance(value, bool):
260 return f"expected boolean, got {type(value).__name__}"
261 if spec.type == "enum" and spec.values and value not in spec.values:
262 return f"expected one of {spec.values!r}, got {value!r}"
263 return None
266def _validate_with_bindings(fsm: FSMLoop, loop_dir: Path) -> list[ValidationError]:
267 """Validate with: bindings against child loop parameter contracts.
269 Called from load_and_validate (not validate_fsm) because resolving child loops
270 requires file-system access via the loop directory path.
272 Args:
273 fsm: The parent FSM loop
274 loop_dir: Directory to resolve child loop paths from
276 Returns:
277 List of validation errors found
278 """
279 errors: list[ValidationError] = []
281 for state_name, state in fsm.states.items():
282 if state.loop is None or not state.with_:
283 continue
285 # Try to resolve and load the child loop; skip if unavailable
286 try:
287 from little_loops.cli.loop._helpers import resolve_loop_path
289 loop_path = resolve_loop_path(state.loop, loop_dir)
290 child_fsm, _ = load_and_validate(loop_path)
291 except Exception:
292 continue
294 if not child_fsm.parameters:
295 continue # Child has no declared contract — nothing to cross-validate
297 path = f"states.{state_name}"
299 # Unknown with: keys (not declared by child)
300 for key in state.with_:
301 if key not in child_fsm.parameters:
302 errors.append(
303 ValidationError(
304 message=(
305 f"'with.{key}' is not a declared parameter of loop '{state.loop}'. "
306 f"Declared: {', '.join(sorted(child_fsm.parameters))}"
307 ),
308 path=f"{path}.with.{key}",
309 )
310 )
312 # Required parameters not bound
313 for param_name, param_spec in child_fsm.parameters.items():
314 if param_spec.required and param_name not in state.with_:
315 errors.append(
316 ValidationError(
317 message=(
318 f"Required parameter '{param_name}' of loop '{state.loop}' "
319 f"is not bound in 'with'"
320 ),
321 path=f"{path}.with",
322 )
323 )
325 # Statically-detectable type mismatches (skip interpolation strings)
326 for param_name, value in state.with_.items():
327 if param_name not in child_fsm.parameters:
328 continue
329 if isinstance(value, str) and "${" in value:
330 continue
331 type_error = _check_param_type(value, child_fsm.parameters[param_name])
332 if type_error:
333 errors.append(
334 ValidationError(
335 message=f"Parameter '{param_name}': {type_error}",
336 path=f"{path}.with.{param_name}",
337 )
338 )
340 return errors
343def _validate_state_action(state_name: str, state: StateConfig) -> list[ValidationError]:
344 """Validate state action configuration.
346 Args:
347 state_name: Name of the state to validate
348 state: The state configuration to validate
350 Returns:
351 List of validation errors found
352 """
353 errors: list[ValidationError] = []
354 path = f"states.{state_name}"
356 # params field is only valid for mcp_tool states
357 if state.params and state.action_type != "mcp_tool":
358 errors.append(
359 ValidationError(
360 message="'params' field is only valid when action_type is 'mcp_tool'",
361 path=f"{path}.params",
362 )
363 )
365 # loop and action are mutually exclusive
366 if state.loop is not None and state.action is not None:
367 errors.append(
368 ValidationError(
369 message="'loop' and 'action' are mutually exclusive — "
370 "a sub-loop state cannot also have an action",
371 path=f"{path}",
372 )
373 )
375 # with: requires loop: to be set
376 if state.with_ and state.loop is None:
377 errors.append(
378 ValidationError(
379 message="'with' is only valid when 'loop' is set",
380 path=f"{path}.with",
381 )
382 )
384 # with: and context_passthrough are mutually exclusive
385 if state.with_ and state.context_passthrough:
386 errors.append(
387 ValidationError(
388 message=(
389 "'with' and 'context_passthrough' are mutually exclusive — "
390 "use 'with' for explicit parameter bindings or 'context_passthrough' "
391 "for legacy bulk passthrough, not both"
392 ),
393 path=f"{path}",
394 )
395 )
397 return errors
400def _validate_state_routing(state_name: str, state: StateConfig) -> list[ValidationError]:
401 """Validate state routing configuration.
403 Checks for conflicting routing definitions (shorthand vs full route).
405 Args:
406 state_name: Name of the state to validate
407 state: The state configuration to validate
409 Returns:
410 List of validation errors/warnings found
411 """
412 errors: list[ValidationError] = []
413 path = f"states.{state_name}"
415 has_shorthand = (
416 state.on_yes is not None
417 or state.on_no is not None
418 or state.on_error is not None
419 or state.on_partial is not None
420 or state.on_blocked is not None
421 or bool(state.extra_routes)
422 )
423 has_route = state.route is not None
425 # Warn about conflicting definitions
426 if has_shorthand and has_route:
427 errors.append(
428 ValidationError(
429 message="Both shorthand routing (on_yes/on_no/on_error) "
430 "and full route table defined. Route table will take precedence.",
431 path=path,
432 severity=ValidationSeverity.WARNING,
433 )
434 )
436 # Check for no valid transition definition
437 has_next = state.next is not None
438 has_terminal = state.terminal
439 has_loop = state.loop is not None
441 if not has_shorthand and not has_route and not has_next and not has_terminal and not has_loop:
442 errors.append(
443 ValidationError(
444 message="State has no transition defined. Add routing, 'next', "
445 "or mark as 'terminal: true'",
446 path=path,
447 )
448 )
450 # Validate retry field pairing: max_retries requires on_retry_exhausted and vice versa
451 if state.max_retries is not None and state.on_retry_exhausted is None:
452 errors.append(
453 ValidationError(
454 message="'max_retries' requires 'on_retry_exhausted' to also be set",
455 path=path,
456 )
457 )
458 if state.on_retry_exhausted is not None and state.max_retries is None:
459 errors.append(
460 ValidationError(
461 message="'on_retry_exhausted' requires 'max_retries' to also be set",
462 path=path,
463 )
464 )
465 if state.max_retries is not None and state.max_retries < 1:
466 errors.append(
467 ValidationError(
468 message=f"'max_retries' must be >= 1, got {state.max_retries}",
469 path=path,
470 )
471 )
473 # Validate rate-limit retry field pairing (mirrors max_retries/on_retry_exhausted)
474 if state.max_rate_limit_retries is not None and state.on_rate_limit_exhausted is None:
475 errors.append(
476 ValidationError(
477 message="'max_rate_limit_retries' requires 'on_rate_limit_exhausted' to also be set",
478 path=path,
479 )
480 )
481 if state.on_rate_limit_exhausted is not None and state.max_rate_limit_retries is None:
482 errors.append(
483 ValidationError(
484 message="'on_rate_limit_exhausted' requires 'max_rate_limit_retries' to also be set",
485 path=path,
486 )
487 )
488 if state.max_rate_limit_retries is not None and state.max_rate_limit_retries < 1:
489 errors.append(
490 ValidationError(
491 message=f"'max_rate_limit_retries' must be >= 1, got {state.max_rate_limit_retries}",
492 path=path,
493 )
494 )
495 if (
496 state.rate_limit_backoff_base_seconds is not None
497 and state.rate_limit_backoff_base_seconds < 1
498 ):
499 errors.append(
500 ValidationError(
501 message=(
502 f"'rate_limit_backoff_base_seconds' must be >= 1, "
503 f"got {state.rate_limit_backoff_base_seconds}"
504 ),
505 path=path,
506 )
507 )
508 if state.rate_limit_max_wait_seconds is not None and state.rate_limit_max_wait_seconds < 1:
509 errors.append(
510 ValidationError(
511 message=(
512 f"'rate_limit_max_wait_seconds' must be >= 1, "
513 f"got {state.rate_limit_max_wait_seconds}"
514 ),
515 path=path,
516 )
517 )
518 if state.rate_limit_long_wait_ladder is not None:
519 if len(state.rate_limit_long_wait_ladder) == 0:
520 errors.append(
521 ValidationError(
522 message="'rate_limit_long_wait_ladder' must be non-empty if specified",
523 path=path,
524 )
525 )
526 else:
527 for idx, value in enumerate(state.rate_limit_long_wait_ladder):
528 if not isinstance(value, int) or value < 1:
529 errors.append(
530 ValidationError(
531 message=(
532 f"'rate_limit_long_wait_ladder[{idx}]' must be a "
533 f"positive integer, got {value!r}"
534 ),
535 path=path,
536 )
537 )
539 # Validate throttle config when present
540 if state.throttle is not None:
541 t = state.throttle
542 fields = {
543 "normal_max": t.normal_max,
544 "warn_max": t.warn_max,
545 "hard_max": t.hard_max,
546 }
547 for field_name, val in fields.items():
548 if val is not None and (not isinstance(val, int) or val < 1):
549 errors.append(
550 ValidationError(
551 message=f"'throttle.{field_name}' must be a positive integer, got {val!r}",
552 path=path,
553 )
554 )
555 # Enforce ordering when all three are set
556 if t.normal_max is not None and t.warn_max is not None and t.normal_max >= t.warn_max:
557 errors.append(
558 ValidationError(
559 message=(
560 f"'throttle.normal_max' ({t.normal_max}) must be less than "
561 f"'throttle.warn_max' ({t.warn_max})"
562 ),
563 path=path,
564 )
565 )
566 if t.warn_max is not None and t.hard_max is not None and t.warn_max >= t.hard_max:
567 errors.append(
568 ValidationError(
569 message=(
570 f"'throttle.warn_max' ({t.warn_max}) must be less than "
571 f"'throttle.hard_max' ({t.hard_max})"
572 ),
573 path=path,
574 )
575 )
577 return errors
580def validate_fsm(fsm: FSMLoop) -> list[ValidationError]:
581 """Validate FSM structure and return list of errors.
583 Performs comprehensive validation:
584 - Initial state exists
585 - All referenced states exist
586 - At least one terminal state
587 - Evaluator configurations are valid
588 - Routing configurations are valid
589 - Numeric fields are in valid ranges (max_iterations > 0, backoff >= 0, timeout > 0)
591 Args:
592 fsm: The FSM loop to validate
594 Returns:
595 List of validation errors (empty if valid)
596 """
597 errors: list[ValidationError] = []
598 defined_states = fsm.get_all_state_names()
600 # Warn when no top-level description: field is set. The field is optional
601 # for FSM execution but required for goal-alignment skills (debug-loop-run,
602 # audit-loop-run) and for ll-loop show --json to surface intent text.
603 if not fsm.description:
604 errors.append(
605 ValidationError(
606 path="<root>",
607 message=("No 'description' field defined. Add a top-level description: key."),
608 severity=ValidationSeverity.WARNING,
609 )
610 )
612 # Validate parameters block
613 errors.extend(_validate_parameters(fsm))
615 # Check initial state exists
616 if fsm.initial not in defined_states:
617 errors.append(
618 ValidationError(
619 message=f"Initial state '{fsm.initial}' not found in states",
620 path="initial",
621 )
622 )
624 # Check at least one terminal state
625 terminal_states = fsm.get_terminal_states()
626 if not terminal_states:
627 errors.append(
628 ValidationError(
629 message="No terminal state defined. At least one state must have 'terminal: true'",
630 path="states",
631 )
632 )
634 # Validate each state
635 for state_name, state in fsm.states.items():
636 # Check all referenced states exist
637 refs = state.get_referenced_states()
638 for ref in refs:
639 # $current is a special token for retry
640 if ref != "$current" and ref not in defined_states:
641 errors.append(
642 ValidationError(
643 message=f"References unknown state '{ref}'",
644 path=f"states.{state_name}",
645 )
646 )
648 # Validate action configuration
649 errors.extend(_validate_state_action(state_name, state))
651 # Validate evaluator if present
652 if state.evaluate is not None:
653 errors.extend(_validate_evaluator(state_name, state.evaluate))
655 # Validate routing configuration
656 errors.extend(_validate_state_routing(state_name, state))
658 # Check numeric field ranges
659 if fsm.max_iterations <= 0:
660 errors.append(
661 ValidationError(
662 message=f"max_iterations must be > 0, got {fsm.max_iterations}",
663 path="max_iterations",
664 )
665 )
666 if fsm.max_edge_revisits <= 0:
667 errors.append(
668 ValidationError(
669 message=f"max_edge_revisits must be > 0, got {fsm.max_edge_revisits}",
670 path="max_edge_revisits",
671 )
672 )
673 if fsm.backoff is not None and fsm.backoff < 0:
674 errors.append(
675 ValidationError(
676 message=f"backoff must be >= 0, got {fsm.backoff}",
677 path="backoff",
678 )
679 )
680 if fsm.timeout is not None and fsm.timeout <= 0:
681 errors.append(
682 ValidationError(
683 message=f"timeout must be > 0, got {fsm.timeout}",
684 path="timeout",
685 )
686 )
687 if fsm.llm.max_tokens <= 0:
688 errors.append(
689 ValidationError(
690 message=f"llm.max_tokens must be > 0, got {fsm.llm.max_tokens}",
691 path="llm.max_tokens",
692 )
693 )
694 if fsm.llm.timeout <= 0:
695 errors.append(
696 ValidationError(
697 message=f"llm.timeout must be > 0, got {fsm.llm.timeout}",
698 path="llm.timeout",
699 )
700 )
702 # Check for unreachable states (warning only)
703 reachable = _find_reachable_states(fsm)
704 unreachable = defined_states - reachable
705 for state_name in unreachable:
706 errors.append(
707 ValidationError(
708 message="State is not reachable from initial state",
709 path=f"states.{state_name}",
710 severity=ValidationSeverity.WARNING,
711 )
712 )
714 return errors
717def _find_reachable_states(fsm: FSMLoop) -> set[str]:
718 """Find all states reachable from the initial state.
720 Uses breadth-first search to find all reachable states.
722 Args:
723 fsm: The FSM loop to analyze
725 Returns:
726 Set of reachable state names
727 """
728 reachable: set[str] = set()
729 to_visit: deque[str] = deque([fsm.initial])
731 while to_visit:
732 current = to_visit.popleft()
733 if current in reachable or current not in fsm.states:
734 continue
736 reachable.add(current)
737 state = fsm.states[current]
738 refs = state.get_referenced_states()
740 for ref in refs:
741 if ref != "$current" and ref not in reachable:
742 to_visit.append(ref)
744 return reachable
747def load_and_validate(path: Path) -> tuple[FSMLoop, list[ValidationError]]:
748 """Load YAML file and validate FSM structure.
750 Args:
751 path: Path to the YAML file to load
753 Returns:
754 Tuple of (validated FSMLoop instance, list of WARNING-severity ValidationErrors)
756 Raises:
757 FileNotFoundError: If the file doesn't exist
758 yaml.YAMLError: If the file is not valid YAML
759 ValueError: If validation fails (contains error details)
760 """
761 if not path.exists():
762 raise FileNotFoundError(f"FSM file not found: {path}")
764 with open(path) as f:
765 data: dict[str, Any] = yaml.safe_load(f)
767 if not isinstance(data, dict):
768 raise ValueError(f"FSM file must contain a YAML mapping, got {type(data)}")
770 # Resolve `from:` inheritance before any further checks, so a child loop
771 # can omit fields its parent provides (including `initial`/`states`) and
772 # so a parent's `import:`/`fragments:` blocks survive into the merged
773 # result for the subsequent `resolve_fragments` pass.
774 data = resolve_inheritance(data, path.parent)
776 # Expand flow: linear shorthand into states: before required-fields check
777 data = resolve_flow(data)
779 # Check required fields before parsing
780 missing = []
781 for field in ["name", "initial"]:
782 if field not in data:
783 missing.append(field)
784 if "states" not in data:
785 missing.append("states (or flow)")
787 if missing:
788 raise ValueError(f"FSM file missing required fields: {', '.join(missing)}")
790 # Check for unknown top-level keys before parsing
791 unknown_key_warnings: list[ValidationError] = []
792 unknown = set(data.keys()) - KNOWN_TOP_LEVEL_KEYS
793 if unknown:
794 unknown_key_warnings.append(
795 ValidationError(
796 path="<root>",
797 message=f"Unknown top-level keys: {', '.join(sorted(unknown))}",
798 severity=ValidationSeverity.WARNING,
799 )
800 )
802 # Resolve fragment libraries before parsing into dataclass
803 data = resolve_fragments(data, path.parent)
805 # Parse into dataclass
806 fsm = FSMLoop.from_dict(data)
808 # Validate
809 errors = validate_fsm(fsm)
811 # Validate with: bindings against child loop parameters (requires file-system access)
812 errors.extend(_validate_with_bindings(fsm, path.parent))
814 # Filter to errors only (not warnings) for raising
815 error_list = [e for e in errors if e.severity == ValidationSeverity.ERROR]
817 if error_list:
818 error_messages = "\n ".join(str(e) for e in error_list)
819 raise ValueError(f"FSM validation failed:\n {error_messages}")
821 # Collect all warnings (unknown-key warnings + structural warnings)
822 struct_warnings = [e for e in errors if e.severity == ValidationSeverity.WARNING]
823 all_warnings = unknown_key_warnings + struct_warnings
824 for warning in all_warnings:
825 logger.warning(str(warning))
827 return fsm, all_warnings