Coverage for little_loops / fsm / schema.py: 62%
306 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""FSM loop schema dataclasses.
3This module defines the type-safe dataclasses that represent FSM loop
4definitions. These match the universal FSM schema described in the
5design documentation.
7The schema supports:
8- Multiple evaluator types (exit_code, output_numeric, etc.)
9- Two-layer transition system (evaluate + route)
10- Both shorthand (on_success/on_failure) and full routing syntax
11- Context variables and captured values
12- LLM evaluation configuration
13"""
15from __future__ import annotations
17from dataclasses import dataclass, field
18from typing import Any, Literal
20# Default LLM model for structured evaluation
21DEFAULT_LLM_MODEL: str = "sonnet"
24@dataclass
25class EvaluateConfig:
26 """Evaluator configuration for action result interpretation.
28 The evaluator determines how to interpret an action's output and
29 produce a verdict string for routing.
31 Attributes:
32 type: Evaluator type. One of:
33 - exit_code: Map exit codes to verdicts (default for shell)
34 - output_numeric: Compare numeric output to target
35 - output_json: Extract and compare JSON path value
36 - output_contains: Pattern matching on stdout
37 - convergence: Compare current vs previous value toward target
38 - llm_structured: Use LLM with structured output (default for slash)
39 operator: Comparison operator (eq, ne, lt, le, gt, ge)
40 target: Target value for comparison
41 tolerance: Acceptable distance from target (for convergence)
42 pattern: Pattern string for output_contains
43 negate: If True, invert the match result (output_contains)
44 path: JSON path for output_json (jq-style)
45 prompt: Custom prompt for llm_structured
46 schema: Custom JSON schema for llm_structured response
47 min_confidence: Minimum confidence threshold for llm_structured
48 uncertain_suffix: If True, append _uncertain to low-confidence verdicts
49 source: Override default source (current action output)
50 previous: Previous value reference for convergence
51 direction: Optimization direction for convergence (minimize/maximize)
52 scope: Paths to limit git diff to for diff_stall evaluator
53 max_stall: Consecutive no-change iterations before failure (diff_stall)
54 """
56 type: Literal[
57 "exit_code",
58 "output_numeric",
59 "output_json",
60 "output_contains",
61 "convergence",
62 "diff_stall",
63 "llm_structured",
64 "mcp_result",
65 ]
66 operator: str | None = None
67 target: int | float | str | None = None
68 tolerance: float | str | None = None # str for interpolation (e.g., "${context.tolerance}")
69 pattern: str | None = None
70 negate: bool = False
71 path: str | None = None
72 prompt: str | None = None
73 schema: dict[str, Any] | None = None
74 min_confidence: float = 0.5
75 uncertain_suffix: bool = False
76 source: str | None = None
77 previous: str | None = None
78 direction: Literal["minimize", "maximize"] = "minimize"
79 scope: list[str] | None = None # for diff_stall: limit git diff to these paths
80 max_stall: int = 1 # for diff_stall: consecutive no-change iterations before failure
82 def to_dict(self) -> dict[str, Any]:
83 """Convert to dictionary for JSON/YAML serialization."""
84 result: dict[str, Any] = {"type": self.type}
86 # Only include non-None optional fields
87 if self.operator is not None:
88 result["operator"] = self.operator
89 if self.target is not None:
90 result["target"] = self.target
91 if self.tolerance is not None:
92 result["tolerance"] = self.tolerance
93 if self.pattern is not None:
94 result["pattern"] = self.pattern
95 if self.negate:
96 result["negate"] = self.negate
97 if self.path is not None:
98 result["path"] = self.path
99 if self.prompt is not None:
100 result["prompt"] = self.prompt
101 if self.schema is not None:
102 result["schema"] = self.schema
103 if self.min_confidence != 0.5:
104 result["min_confidence"] = self.min_confidence
105 if self.uncertain_suffix:
106 result["uncertain_suffix"] = self.uncertain_suffix
107 if self.source is not None:
108 result["source"] = self.source
109 if self.previous is not None:
110 result["previous"] = self.previous
111 if self.direction != "minimize":
112 result["direction"] = self.direction
113 if self.scope is not None:
114 result["scope"] = self.scope
115 if self.max_stall != 1:
116 result["max_stall"] = self.max_stall
118 return result
120 @classmethod
121 def from_dict(cls, data: dict[str, Any]) -> EvaluateConfig:
122 """Create from dictionary (JSON/YAML deserialization)."""
123 return cls(
124 type=data["type"],
125 operator=data.get("operator"),
126 target=data.get("target"),
127 tolerance=data.get("tolerance"),
128 pattern=data.get("pattern"),
129 negate=data.get("negate", False),
130 path=data.get("path"),
131 prompt=data.get("prompt"),
132 schema=data.get("schema"),
133 min_confidence=data.get("min_confidence", 0.5),
134 uncertain_suffix=data.get("uncertain_suffix", False),
135 source=data.get("source"),
136 previous=data.get("previous"),
137 direction=data.get("direction", "minimize"),
138 scope=data.get("scope"),
139 max_stall=data.get("max_stall", 1),
140 )
143@dataclass
144class RouteConfig:
145 """Routing table configuration for verdict-to-state mapping.
147 Maps verdict strings from evaluators to next state names.
149 Attributes:
150 routes: Mapping from verdict string to next state name
151 default: Default state for unmatched verdicts (the "_" key)
152 error: State for evaluation/execution errors (the "_error" key)
153 """
155 routes: dict[str, str] = field(default_factory=dict)
156 default: str | None = None
157 error: str | None = None
159 def to_dict(self) -> dict[str, Any]:
160 """Convert to dictionary for JSON/YAML serialization."""
161 result = dict(self.routes)
162 if self.default is not None:
163 result["_"] = self.default
164 if self.error is not None:
165 result["_error"] = self.error
166 return result
168 @classmethod
169 def from_dict(cls, data: dict[str, Any]) -> RouteConfig:
170 """Create from dictionary (JSON/YAML deserialization)."""
171 routes = {k: v for k, v in data.items() if not k.startswith("_")}
172 return cls(
173 routes=routes,
174 default=data.get("_"),
175 error=data.get("_error"),
176 )
179@dataclass
180class StateConfig:
181 """Configuration for a single FSM state.
183 States can have actions, evaluators, and routing. Supports both
184 shorthand (on_success/on_failure) and full routing table syntax.
186 Attributes:
187 action: Command to execute (shell, slash command, or "server/tool-name" for mcp_tool)
188 action_type: How to execute the action (prompt, slash_command, shell, mcp_tool).
189 If None, uses heuristic: / prefix = slash_command, else = shell.
190 params: MCP tool arguments (only used with action_type: mcp_tool). Supports
191 ${variable} interpolation in string values.
192 evaluate: Evaluator configuration for result interpretation
193 route: Full routing table (verdict -> state mapping)
194 on_yes: Shorthand for yes verdict routing
195 on_no: Shorthand for no verdict routing
196 on_error: Shorthand for error verdict routing
197 on_partial: Shorthand for partial verdict routing
198 next: Unconditional transition (no evaluation)
199 terminal: If True, this is an end state
200 capture: Variable name to store action output
201 timeout: Action-level timeout in seconds
202 on_maintain: State to transition to when maintain=True and loop completes
203 max_retries: Max consecutive re-entries before transitioning to on_retry_exhausted.
204 A value of N allows N retries after the initial execution (N+1 total entries).
205 Requires on_retry_exhausted to also be set.
206 on_retry_exhausted: State to transition to when max_retries consecutive re-entries
207 are exceeded. Required when max_retries is set.
208 loop: Name of a loop YAML to execute as a sub-FSM. Mutually exclusive with action.
209 context_passthrough: When True, pass parent context variables to child loop and
210 merge child captures back into parent context.
211 """
213 action: str | None = None
214 action_type: str | None = None
215 params: dict[str, Any] = field(default_factory=dict)
216 evaluate: EvaluateConfig | None = None
217 route: RouteConfig | None = None
218 on_yes: str | None = None
219 on_no: str | None = None
220 on_error: str | None = None
221 on_partial: str | None = None
222 on_blocked: str | None = None
223 next: str | None = None
224 terminal: bool = False
225 capture: str | None = None
226 timeout: int | None = None
227 on_maintain: str | None = None
228 max_retries: int | None = None
229 on_retry_exhausted: str | None = None
230 loop: str | None = None
231 context_passthrough: bool = False
232 agent: str | None = None
233 tools: list[str] | None = None
234 extra_routes: dict[str, str] = field(default_factory=dict)
236 def to_dict(self) -> dict[str, Any]:
237 """Convert to dictionary for JSON/YAML serialization."""
238 result: dict[str, Any] = {}
240 if self.action is not None:
241 result["action"] = self.action
242 if self.action_type is not None:
243 result["action_type"] = self.action_type
244 if self.params:
245 result["params"] = self.params
246 if self.evaluate is not None:
247 result["evaluate"] = self.evaluate.to_dict()
248 if self.route is not None:
249 result["route"] = self.route.to_dict()
250 if self.on_yes is not None:
251 result["on_yes"] = self.on_yes
252 if self.on_no is not None:
253 result["on_no"] = self.on_no
254 if self.on_error is not None:
255 result["on_error"] = self.on_error
256 if self.on_partial is not None:
257 result["on_partial"] = self.on_partial
258 if self.on_blocked is not None:
259 result["on_blocked"] = self.on_blocked
260 if self.next is not None:
261 result["next"] = self.next
262 if self.terminal:
263 result["terminal"] = self.terminal
264 if self.capture is not None:
265 result["capture"] = self.capture
266 if self.timeout is not None:
267 result["timeout"] = self.timeout
268 if self.on_maintain is not None:
269 result["on_maintain"] = self.on_maintain
270 if self.max_retries is not None:
271 result["max_retries"] = self.max_retries
272 if self.on_retry_exhausted is not None:
273 result["on_retry_exhausted"] = self.on_retry_exhausted
274 if self.loop is not None:
275 result["loop"] = self.loop
276 if self.context_passthrough:
277 result["context_passthrough"] = self.context_passthrough
278 if self.agent is not None:
279 result["agent"] = self.agent
280 if self.tools is not None:
281 result["tools"] = self.tools
282 for verdict, target in self.extra_routes.items():
283 result[f"on_{verdict}"] = target
285 return result
287 @classmethod
288 def from_dict(cls, data: dict[str, Any]) -> StateConfig:
289 """Create from dictionary (JSON/YAML deserialization)."""
290 evaluate = None
291 if "evaluate" in data:
292 evaluate = EvaluateConfig.from_dict(data["evaluate"])
294 route = None
295 if "route" in data:
296 route = RouteConfig.from_dict(data["route"])
298 _known_on_keys = {
299 "on_yes",
300 "on_success",
301 "on_no",
302 "on_failure",
303 "on_error",
304 "on_partial",
305 "on_blocked",
306 "on_maintain",
307 "on_retry_exhausted",
308 }
309 extra_routes = {
310 key[3:]: val
311 for key, val in data.items()
312 if key.startswith("on_") and key not in _known_on_keys and isinstance(val, str)
313 }
315 return cls(
316 action=data.get("action"),
317 action_type=data.get("action_type"),
318 params=data.get("params", {}),
319 evaluate=evaluate,
320 route=route,
321 on_yes=data.get("on_yes") or data.get("on_success"),
322 on_no=data.get("on_no") or data.get("on_failure"),
323 on_error=data.get("on_error"),
324 on_partial=data.get("on_partial"),
325 on_blocked=data.get("on_blocked"),
326 next=data.get("next"),
327 terminal=data.get("terminal", False),
328 capture=data.get("capture"),
329 timeout=data.get("timeout"),
330 on_maintain=data.get("on_maintain"),
331 max_retries=data.get("max_retries"),
332 on_retry_exhausted=data.get("on_retry_exhausted"),
333 loop=data.get("loop"),
334 context_passthrough=data.get("context_passthrough", False),
335 agent=data.get("agent"),
336 tools=data.get("tools"),
337 extra_routes=extra_routes,
338 )
340 def get_referenced_states(self) -> set[str]:
341 """Get all state names referenced by this state configuration.
343 Returns:
344 Set of state names that this state can transition to.
345 """
346 refs: set[str] = set()
348 if self.on_yes is not None:
349 refs.add(self.on_yes)
350 if self.on_no is not None:
351 refs.add(self.on_no)
352 if self.on_error is not None:
353 refs.add(self.on_error)
354 if self.on_partial is not None:
355 refs.add(self.on_partial)
356 if self.on_blocked is not None:
357 refs.add(self.on_blocked)
358 if self.next is not None:
359 refs.add(self.next)
360 if self.on_maintain is not None:
361 refs.add(self.on_maintain)
362 if self.on_retry_exhausted is not None:
363 refs.add(self.on_retry_exhausted)
364 if self.route is not None:
365 refs.update(self.route.routes.values())
366 if self.route.default is not None:
367 refs.add(self.route.default)
368 if self.route.error is not None:
369 refs.add(self.route.error)
370 refs.update(self.extra_routes.values())
372 return refs
375@dataclass
376class LLMConfig:
377 """LLM evaluation configuration.
379 Settings for the llm_structured evaluator.
381 Attributes:
382 enabled: If False, disable LLM evaluation entirely
383 model: Model identifier for LLM calls
384 max_tokens: Maximum tokens for evaluation response
385 timeout: Timeout for LLM calls in seconds
386 """
388 enabled: bool = True
389 model: str = DEFAULT_LLM_MODEL
390 max_tokens: int = 256
391 timeout: int = 1800
393 def to_dict(self) -> dict[str, Any]:
394 """Convert to dictionary for JSON/YAML serialization."""
395 result: dict[str, Any] = {}
397 if not self.enabled:
398 result["enabled"] = self.enabled
399 if self.model != DEFAULT_LLM_MODEL:
400 result["model"] = self.model
401 if self.max_tokens != 256:
402 result["max_tokens"] = self.max_tokens
403 if self.timeout != 1800:
404 result["timeout"] = self.timeout
406 return result if result else {}
408 @classmethod
409 def from_dict(cls, data: dict[str, Any]) -> LLMConfig:
410 """Create from dictionary (JSON/YAML deserialization)."""
411 return cls(
412 enabled=data.get("enabled", True),
413 model=data.get("model", DEFAULT_LLM_MODEL),
414 max_tokens=data.get("max_tokens", 256),
415 timeout=data.get("timeout", 1800),
416 )
419@dataclass
420class LoopConfigOverrides:
421 """Per-loop ll-config overrides embedded in the loop YAML definition.
423 All fields are optional (None = use global ll-config default).
424 Precedence: CLI flags > YAML config block > global ll-config > schema defaults.
426 Attributes:
427 handoff_threshold: Override for LL_HANDOFF_THRESHOLD env var (1-100)
428 readiness_threshold: Override for commands.confidence_gate.readiness_threshold (1-100)
429 outcome_threshold: Override for commands.confidence_gate.outcome_threshold (1-100)
430 max_continuations: Override for automation.max_continuations (>=1)
431 """
433 handoff_threshold: int | None = None
434 readiness_threshold: int | None = None
435 outcome_threshold: int | None = None
436 max_continuations: int | None = None
438 def to_dict(self) -> dict[str, Any]:
439 """Convert to dictionary for JSON/YAML serialization (skip-if-None)."""
440 result: dict[str, Any] = {}
442 if self.handoff_threshold is not None:
443 result["handoff_threshold"] = self.handoff_threshold
445 confidence_gate: dict[str, Any] = {}
446 if self.readiness_threshold is not None:
447 confidence_gate["readiness_threshold"] = self.readiness_threshold
448 if self.outcome_threshold is not None:
449 confidence_gate["outcome_threshold"] = self.outcome_threshold
450 if confidence_gate:
451 result["commands"] = {"confidence_gate": confidence_gate}
453 if self.max_continuations is not None:
454 result["automation"] = {"max_continuations": self.max_continuations}
456 return result
458 @classmethod
459 def from_dict(cls, data: dict[str, Any]) -> LoopConfigOverrides:
460 """Create from dictionary (JSON/YAML deserialization)."""
461 commands = data.get("commands", {})
462 confidence_gate = commands.get("confidence_gate", {}) if isinstance(commands, dict) else {}
463 automation = data.get("automation", {})
464 continuation = data.get("continuation", {})
466 max_continuations = None
467 if isinstance(automation, dict) and "max_continuations" in automation:
468 max_continuations = automation["max_continuations"]
469 elif isinstance(continuation, dict) and "max_continuations" in continuation:
470 max_continuations = continuation["max_continuations"]
472 return cls(
473 handoff_threshold=data.get("handoff_threshold"),
474 readiness_threshold=confidence_gate.get("readiness_threshold")
475 if isinstance(confidence_gate, dict)
476 else None,
477 outcome_threshold=confidence_gate.get("outcome_threshold")
478 if isinstance(confidence_gate, dict)
479 else None,
480 max_continuations=max_continuations,
481 )
484@dataclass
485class FSMLoop:
486 """Complete FSM loop definition.
488 The main dataclass representing a loop configuration.
490 Attributes:
491 name: Unique loop identifier
492 initial: Starting state name
493 states: Mapping from state name to StateConfig
494 context: User-defined shared variables
495 scope: Paths this loop operates on (for concurrency control)
496 max_iterations: Safety limit for loop iterations
497 backoff: Seconds between iterations
498 timeout: Max total runtime in seconds (loop-level)
499 maintain: If True, restart after completion
500 llm: LLM evaluation configuration
501 on_handoff: Behavior when handoff signal detected (pause/spawn/terminate)
502 """
504 name: str
505 initial: str
506 states: dict[str, StateConfig]
507 description: str | None = None
508 context: dict[str, Any] = field(default_factory=dict)
509 scope: list[str] = field(default_factory=list)
510 max_iterations: int = 50
511 backoff: float | None = None
512 timeout: int | None = None
513 default_timeout: int | None = None
514 maintain: bool = False
515 llm: LLMConfig = field(default_factory=LLMConfig)
516 on_handoff: Literal["pause", "spawn", "terminate"] = "pause"
517 input_key: str = "input"
518 config: LoopConfigOverrides | None = None
519 category: str = ""
520 labels: list[str] = field(default_factory=list)
522 def to_dict(self) -> dict[str, Any]:
523 """Convert to dictionary for JSON/YAML serialization."""
524 result: dict[str, Any] = {
525 "name": self.name,
526 "initial": self.initial,
527 "states": {name: state.to_dict() for name, state in self.states.items()},
528 }
530 if self.description is not None:
531 result["description"] = self.description
532 if self.context:
533 result["context"] = self.context
534 if self.scope:
535 result["scope"] = self.scope
536 if self.max_iterations != 50:
537 result["max_iterations"] = self.max_iterations
538 if self.backoff is not None:
539 result["backoff"] = self.backoff
540 if self.timeout is not None:
541 result["timeout"] = self.timeout
542 if self.default_timeout is not None:
543 result["default_timeout"] = self.default_timeout
544 if self.maintain:
545 result["maintain"] = self.maintain
546 if self.on_handoff != "pause":
547 result["on_handoff"] = self.on_handoff
548 if self.input_key != "input":
549 result["input_key"] = self.input_key
551 llm_dict = self.llm.to_dict()
552 if llm_dict:
553 result["llm"] = llm_dict
555 if self.config is not None:
556 config_dict = self.config.to_dict()
557 if config_dict:
558 result["config"] = config_dict
560 if self.category:
561 result["category"] = self.category
562 if self.labels:
563 result["labels"] = self.labels
565 return result
567 @classmethod
568 def from_dict(cls, data: dict[str, Any]) -> FSMLoop:
569 """Create from dictionary (JSON/YAML deserialization)."""
570 states = {
571 name: StateConfig.from_dict(state_data)
572 for name, state_data in data.get("states", {}).items()
573 }
575 llm = LLMConfig()
576 if "llm" in data:
577 llm = LLMConfig.from_dict(data["llm"])
579 loop_config = None
580 if "config" in data:
581 loop_config = LoopConfigOverrides.from_dict(data["config"])
583 return cls(
584 name=data["name"],
585 initial=data["initial"],
586 states=states,
587 description=data.get("description"),
588 context=data.get("context", {}),
589 scope=data.get("scope", []),
590 max_iterations=data.get("max_iterations", 50),
591 backoff=data.get("backoff"),
592 timeout=data.get("timeout"),
593 default_timeout=data.get("default_timeout"),
594 maintain=data.get("maintain", False),
595 llm=llm,
596 on_handoff=data.get("on_handoff", "pause"),
597 input_key=data.get("input_key", "input"),
598 config=loop_config,
599 category=data.get("category", ""),
600 labels=data.get("labels", []),
601 )
603 def get_all_state_names(self) -> set[str]:
604 """Get all defined state names.
606 Returns:
607 Set of all state names in this FSM.
608 """
609 return set(self.states.keys())
611 def get_terminal_states(self) -> set[str]:
612 """Get all terminal state names.
614 Returns:
615 Set of state names where terminal=True.
616 """
617 return {name for name, state in self.states.items() if state.terminal}
619 def get_all_referenced_states(self) -> set[str]:
620 """Get all state names referenced by transitions.
622 This includes the initial state and all states referenced
623 in routing configurations.
625 Returns:
626 Set of all referenced state names.
627 """
628 refs: set[str] = {self.initial}
629 for state in self.states.values():
630 refs.update(state.get_referenced_states())
631 return refs