Coverage for little_loops / fsm / schema.py: 35%
392 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-11 00:29 -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 "harbor_scorer",
66 ]
67 operator: str | None = None
68 target: int | float | str | None = None
69 tolerance: float | str | None = None # str for interpolation (e.g., "${context.tolerance}")
70 pattern: str | None = None
71 negate: bool = False
72 path: str | None = None
73 prompt: str | None = None
74 schema: dict[str, Any] | None = None
75 min_confidence: float = 0.5
76 uncertain_suffix: bool = False
77 source: str | None = None
78 previous: str | None = None
79 direction: Literal["minimize", "maximize"] = "minimize"
80 scope: list[str] | None = None # for diff_stall: limit git diff to these paths
81 max_stall: int = 1 # for diff_stall: consecutive no-change iterations before failure
83 def to_dict(self) -> dict[str, Any]:
84 """Convert to dictionary for JSON/YAML serialization."""
85 result: dict[str, Any] = {"type": self.type}
87 # Only include non-None optional fields
88 if self.operator is not None:
89 result["operator"] = self.operator
90 if self.target is not None:
91 result["target"] = self.target
92 if self.tolerance is not None:
93 result["tolerance"] = self.tolerance
94 if self.pattern is not None:
95 result["pattern"] = self.pattern
96 if self.negate:
97 result["negate"] = self.negate
98 if self.path is not None:
99 result["path"] = self.path
100 if self.prompt is not None:
101 result["prompt"] = self.prompt
102 if self.schema is not None:
103 result["schema"] = self.schema
104 if self.min_confidence != 0.5:
105 result["min_confidence"] = self.min_confidence
106 if self.uncertain_suffix:
107 result["uncertain_suffix"] = self.uncertain_suffix
108 if self.source is not None:
109 result["source"] = self.source
110 if self.previous is not None:
111 result["previous"] = self.previous
112 if self.direction != "minimize":
113 result["direction"] = self.direction
114 if self.scope is not None:
115 result["scope"] = self.scope
116 if self.max_stall != 1:
117 result["max_stall"] = self.max_stall
119 return result
121 @classmethod
122 def from_dict(cls, data: dict[str, Any]) -> EvaluateConfig:
123 """Create from dictionary (JSON/YAML deserialization)."""
124 return cls(
125 type=data["type"],
126 operator=data.get("operator"),
127 target=data.get("target"),
128 tolerance=data.get("tolerance"),
129 pattern=data.get("pattern"),
130 negate=data.get("negate", False),
131 path=data.get("path"),
132 prompt=data.get("prompt"),
133 schema=data.get("schema"),
134 min_confidence=data.get("min_confidence", 0.5),
135 uncertain_suffix=data.get("uncertain_suffix", False),
136 source=data.get("source"),
137 previous=data.get("previous"),
138 direction=data.get("direction", "minimize"),
139 scope=data.get("scope"),
140 max_stall=data.get("max_stall", 1),
141 )
144@dataclass
145class RouteConfig:
146 """Routing table configuration for verdict-to-state mapping.
148 Maps verdict strings from evaluators to next state names.
150 Attributes:
151 routes: Mapping from verdict string to next state name
152 default: Default state for unmatched verdicts (the "_" key)
153 error: State for evaluation/execution errors (the "_error" key)
154 """
156 routes: dict[str, str] = field(default_factory=dict)
157 default: str | None = None
158 error: str | None = None
160 def to_dict(self) -> dict[str, Any]:
161 """Convert to dictionary for JSON/YAML serialization."""
162 result = dict(self.routes)
163 if self.default is not None:
164 result["_"] = self.default
165 if self.error is not None:
166 result["_error"] = self.error
167 return result
169 @classmethod
170 def from_dict(cls, data: dict[str, Any]) -> RouteConfig:
171 """Create from dictionary (JSON/YAML deserialization)."""
172 routes = {k: v for k, v in data.items() if not k.startswith("_")}
173 return cls(
174 routes=routes,
175 default=data.get("_"),
176 error=data.get("_error"),
177 )
180@dataclass
181class ParameterSpec:
182 """Specification for a single loop input parameter.
184 Declares a typed input that callers bind via the 'with:' block on sub-loop states.
186 Attributes:
187 type: Parameter type. One of: string, integer, number, boolean, enum, path.
188 required: If True, callers must supply this parameter in 'with:' (mutually
189 exclusive with 'default').
190 default: Default value used when the caller does not bind the parameter.
191 Only valid when required is False.
192 description: Human-readable description of the parameter.
193 values: Allowed values for 'enum' type parameters.
194 """
196 type: str
197 required: bool = False
198 default: Any = None
199 description: str | None = None
200 values: list[Any] | None = None
202 def to_dict(self) -> dict[str, Any]:
203 """Convert to dictionary for JSON/YAML serialization."""
204 result: dict[str, Any] = {"type": self.type}
205 if self.required:
206 result["required"] = self.required
207 if self.default is not None:
208 result["default"] = self.default
209 if self.description is not None:
210 result["description"] = self.description
211 if self.values is not None:
212 result["values"] = self.values
213 return result
215 @classmethod
216 def from_dict(cls, data: dict[str, Any]) -> ParameterSpec:
217 """Create from dictionary (JSON/YAML deserialization)."""
218 return cls(
219 type=data["type"],
220 required=data.get("required", False),
221 default=data.get("default"),
222 description=data.get("description"),
223 values=data.get("values"),
224 )
227@dataclass
228class ThrottleConfig:
229 """Per-state tool-call progressive throttling configuration.
231 Counts successful tool calls within a single state visit and escalates
232 restrictions to self-throttle runaway states before provider limits are reached.
234 Thresholds fall back to executor module defaults when not set:
235 - normal_max: _DEFAULT_THROTTLE_NORMAL_MAX (3) — calls 1..normal_max pass through
236 - warn_max: _DEFAULT_THROTTLE_WARN_MAX (8) — calls at warn_max inject a warning event
237 - hard_max: _DEFAULT_THROTTLE_HARD_MAX (12) — calls at hard_max route to on_throttle_hard
238 - calls > hard_max: hard stop, loop marked stuck
240 States with type="learning" (FEAT-1283) are exempt from the hard_max hard-stop because
241 they legitimately make N tool calls per visit (one per unproven target). The warn_max
242 warning still applies so users can see the state is doing significant work.
243 """
245 normal_max: int | None = None
246 warn_max: int | None = None
247 hard_max: int | None = None
249 def to_dict(self) -> dict[str, Any]:
250 """Convert to dictionary for JSON/YAML serialization."""
251 result: dict[str, Any] = {}
252 if self.normal_max is not None:
253 result["normal_max"] = self.normal_max
254 if self.warn_max is not None:
255 result["warn_max"] = self.warn_max
256 if self.hard_max is not None:
257 result["hard_max"] = self.hard_max
258 return result
260 @classmethod
261 def from_dict(cls, data: dict[str, Any]) -> ThrottleConfig:
262 """Create from dictionary (JSON/YAML deserialization)."""
263 return cls(
264 normal_max=data.get("normal_max"),
265 warn_max=data.get("warn_max"),
266 hard_max=data.get("hard_max"),
267 )
270@dataclass
271class StateConfig:
272 """Configuration for a single FSM state.
274 States can have actions, evaluators, and routing. Supports both
275 shorthand (on_success/on_failure) and full routing table syntax.
277 Attributes:
278 action: Command to execute (shell, slash command, or "server/tool-name" for mcp_tool)
279 action_type: How to execute the action (prompt, slash_command, shell, mcp_tool).
280 If None, uses heuristic: / prefix = slash_command, else = shell.
281 params: MCP tool arguments (only used with action_type: mcp_tool). Supports
282 ${variable} interpolation in string values.
283 evaluate: Evaluator configuration for result interpretation
284 route: Full routing table (verdict -> state mapping)
285 on_yes: Shorthand for yes verdict routing
286 on_no: Shorthand for no verdict routing
287 on_error: Shorthand for error verdict routing
288 on_partial: Shorthand for partial verdict routing
289 next: Unconditional transition (no evaluation)
290 terminal: If True, this is an end state
291 capture: Variable name to store action output
292 timeout: Action-level timeout in seconds
293 on_maintain: State to transition to when maintain=True and loop completes
294 max_retries: Max consecutive re-entries before transitioning to on_retry_exhausted.
295 A value of N allows N retries after the initial execution (N+1 total entries).
296 Requires on_retry_exhausted to also be set.
297 on_retry_exhausted: State to transition to when max_retries consecutive re-entries
298 are exceeded. Required when max_retries is set.
299 max_rate_limit_retries: Max consecutive 429/rate-limit in-place retries for this
300 state before transitioning to on_rate_limit_exhausted. Requires
301 on_rate_limit_exhausted to also be set.
302 on_rate_limit_exhausted: State to transition to when max_rate_limit_retries
303 consecutive rate-limit retries are exceeded. Required when
304 max_rate_limit_retries is set.
305 rate_limit_backoff_base_seconds: Base seconds for exponential backoff between
306 rate-limit retries; actual sleep is base * 2^(attempt-1) + uniform(0, base).
307 Defaults to 30.
308 rate_limit_max_wait_seconds: Total wall-clock budget (seconds) for rate-limit
309 handling in this state before routing to on_rate_limit_exhausted. When unset,
310 defaults from commands.rate_limits.max_wait_seconds (21600 = 6h).
311 rate_limit_long_wait_ladder: Backoff ladder (seconds) for the long-wait tier used
312 once the short-tier retry budget is spent. Each entry is the sleep before the
313 next retry attempt. When unset, defaults from
314 commands.rate_limits.long_wait_ladder ([300, 900, 1800, 3600]).
315 loop: Name of a loop YAML to execute as a sub-FSM. Mutually exclusive with action.
316 context_passthrough: When True, pass parent context variables to child loop and
317 merge child captures back into parent context. Legacy escape hatch; prefer
318 'with_' for explicit named bindings.
319 with_: Explicit parameter bindings for sub-loop calls (YAML key: 'with'). Maps
320 declared child parameter names to parent expressions. Only valid when 'loop'
321 is set. Mutually exclusive with context_passthrough.
322 """
324 action: str | None = None
325 action_type: str | None = None
326 params: dict[str, Any] = field(default_factory=dict)
327 evaluate: EvaluateConfig | None = None
328 route: RouteConfig | None = None
329 on_yes: str | None = None
330 on_no: str | None = None
331 on_error: str | None = None
332 on_partial: str | None = None
333 on_blocked: str | None = None
334 next: str | None = None
335 terminal: bool = False
336 capture: str | None = None
337 timeout: int | None = None
338 on_maintain: str | None = None
339 max_retries: int | None = None
340 on_retry_exhausted: str | None = None
341 max_rate_limit_retries: int | None = None
342 on_rate_limit_exhausted: str | None = None
343 rate_limit_backoff_base_seconds: int | None = None
344 rate_limit_max_wait_seconds: int | None = None
345 rate_limit_long_wait_ladder: list[int] | None = None
346 loop: str | None = None
347 context_passthrough: bool = False
348 with_: dict[str, Any] = field(default_factory=dict)
349 agent: str | None = None
350 tools: list[str] | None = None
351 extra_routes: dict[str, str] = field(default_factory=dict)
352 type: str | None = None
353 throttle: ThrottleConfig | None = None
354 on_throttle_hard: str | None = None
356 def to_dict(self) -> dict[str, Any]:
357 """Convert to dictionary for JSON/YAML serialization."""
358 result: dict[str, Any] = {}
360 if self.action is not None:
361 result["action"] = self.action
362 if self.action_type is not None:
363 result["action_type"] = self.action_type
364 if self.params:
365 result["params"] = self.params
366 if self.evaluate is not None:
367 result["evaluate"] = self.evaluate.to_dict()
368 if self.route is not None:
369 result["route"] = self.route.to_dict()
370 if self.on_yes is not None:
371 result["on_yes"] = self.on_yes
372 if self.on_no is not None:
373 result["on_no"] = self.on_no
374 if self.on_error is not None:
375 result["on_error"] = self.on_error
376 if self.on_partial is not None:
377 result["on_partial"] = self.on_partial
378 if self.on_blocked is not None:
379 result["on_blocked"] = self.on_blocked
380 if self.next is not None:
381 result["next"] = self.next
382 if self.terminal:
383 result["terminal"] = self.terminal
384 if self.capture is not None:
385 result["capture"] = self.capture
386 if self.timeout is not None:
387 result["timeout"] = self.timeout
388 if self.on_maintain is not None:
389 result["on_maintain"] = self.on_maintain
390 if self.max_retries is not None:
391 result["max_retries"] = self.max_retries
392 if self.on_retry_exhausted is not None:
393 result["on_retry_exhausted"] = self.on_retry_exhausted
394 if self.max_rate_limit_retries is not None:
395 result["max_rate_limit_retries"] = self.max_rate_limit_retries
396 if self.on_rate_limit_exhausted is not None:
397 result["on_rate_limit_exhausted"] = self.on_rate_limit_exhausted
398 if self.rate_limit_backoff_base_seconds is not None:
399 result["rate_limit_backoff_base_seconds"] = self.rate_limit_backoff_base_seconds
400 if self.rate_limit_max_wait_seconds is not None:
401 result["rate_limit_max_wait_seconds"] = self.rate_limit_max_wait_seconds
402 if self.rate_limit_long_wait_ladder is not None:
403 result["rate_limit_long_wait_ladder"] = self.rate_limit_long_wait_ladder
404 if self.loop is not None:
405 result["loop"] = self.loop
406 if self.context_passthrough:
407 result["context_passthrough"] = self.context_passthrough
408 if self.with_:
409 result["with"] = self.with_
410 if self.agent is not None:
411 result["agent"] = self.agent
412 if self.tools is not None:
413 result["tools"] = self.tools
414 for verdict, target in self.extra_routes.items():
415 result[f"on_{verdict}"] = target
416 if self.type is not None:
417 result["type"] = self.type
418 if self.throttle is not None:
419 result["throttle"] = self.throttle.to_dict()
420 if self.on_throttle_hard is not None:
421 result["on_throttle_hard"] = self.on_throttle_hard
423 return result
425 @classmethod
426 def from_dict(cls, data: dict[str, Any]) -> StateConfig:
427 """Create from dictionary (JSON/YAML deserialization)."""
428 evaluate = None
429 if "evaluate" in data:
430 evaluate = EvaluateConfig.from_dict(data["evaluate"])
432 route = None
433 if "route" in data:
434 route = RouteConfig.from_dict(data["route"])
436 throttle = None
437 if "throttle" in data:
438 throttle = ThrottleConfig.from_dict(data["throttle"])
440 _known_on_keys = {
441 "on_yes",
442 "on_success",
443 "on_no",
444 "on_failure",
445 "on_error",
446 "on_partial",
447 "on_blocked",
448 "on_maintain",
449 "on_retry_exhausted",
450 "on_rate_limit_exhausted",
451 "on_throttle_hard",
452 }
453 extra_routes = {
454 key[3:]: val
455 for key, val in data.items()
456 if key.startswith("on_") and key not in _known_on_keys and isinstance(val, str)
457 }
459 return cls(
460 action=data.get("action"),
461 action_type=data.get("action_type"),
462 params=data.get("params", {}),
463 evaluate=evaluate,
464 route=route,
465 on_yes=data.get("on_yes") or data.get("on_success"),
466 on_no=data.get("on_no") or data.get("on_failure"),
467 on_error=data.get("on_error"),
468 on_partial=data.get("on_partial"),
469 on_blocked=data.get("on_blocked"),
470 next=data.get("next"),
471 terminal=data.get("terminal", False),
472 capture=data.get("capture"),
473 timeout=data.get("timeout"),
474 on_maintain=data.get("on_maintain"),
475 max_retries=data.get("max_retries"),
476 on_retry_exhausted=data.get("on_retry_exhausted"),
477 max_rate_limit_retries=data.get("max_rate_limit_retries"),
478 on_rate_limit_exhausted=data.get("on_rate_limit_exhausted"),
479 rate_limit_backoff_base_seconds=data.get("rate_limit_backoff_base_seconds"),
480 rate_limit_max_wait_seconds=data.get("rate_limit_max_wait_seconds"),
481 rate_limit_long_wait_ladder=data.get("rate_limit_long_wait_ladder"),
482 loop=data.get("loop"),
483 context_passthrough=data.get("context_passthrough", False),
484 with_=data.get("with", {}),
485 agent=data.get("agent"),
486 tools=data.get("tools"),
487 extra_routes=extra_routes,
488 type=data.get("type"),
489 throttle=throttle,
490 on_throttle_hard=data.get("on_throttle_hard"),
491 )
493 def get_referenced_states(self) -> set[str]:
494 """Get all state names referenced by this state configuration.
496 Returns:
497 Set of state names that this state can transition to.
498 """
499 refs: set[str] = set()
501 if self.on_yes is not None:
502 refs.add(self.on_yes)
503 if self.on_no is not None:
504 refs.add(self.on_no)
505 if self.on_error is not None:
506 refs.add(self.on_error)
507 if self.on_partial is not None:
508 refs.add(self.on_partial)
509 if self.on_blocked is not None:
510 refs.add(self.on_blocked)
511 if self.next is not None:
512 refs.add(self.next)
513 if self.on_maintain is not None:
514 refs.add(self.on_maintain)
515 if self.on_retry_exhausted is not None:
516 refs.add(self.on_retry_exhausted)
517 if self.on_rate_limit_exhausted is not None:
518 refs.add(self.on_rate_limit_exhausted)
519 if self.on_throttle_hard is not None:
520 refs.add(self.on_throttle_hard)
521 if self.route is not None:
522 refs.update(self.route.routes.values())
523 if self.route.default is not None:
524 refs.add(self.route.default)
525 if self.route.error is not None:
526 refs.add(self.route.error)
527 refs.update(self.extra_routes.values())
529 return refs
532@dataclass
533class LLMConfig:
534 """LLM evaluation configuration.
536 Settings for the llm_structured evaluator.
538 Attributes:
539 enabled: If False, disable LLM evaluation entirely
540 model: Model identifier for LLM calls
541 max_tokens: Maximum tokens for evaluation response
542 timeout: Timeout for LLM calls in seconds
543 """
545 enabled: bool = True
546 model: str = DEFAULT_LLM_MODEL
547 max_tokens: int = 256
548 timeout: int = 1800
550 def to_dict(self) -> dict[str, Any]:
551 """Convert to dictionary for JSON/YAML serialization."""
552 result: dict[str, Any] = {}
554 if not self.enabled:
555 result["enabled"] = self.enabled
556 if self.model != DEFAULT_LLM_MODEL:
557 result["model"] = self.model
558 if self.max_tokens != 256:
559 result["max_tokens"] = self.max_tokens
560 if self.timeout != 1800:
561 result["timeout"] = self.timeout
563 return result if result else {}
565 @classmethod
566 def from_dict(cls, data: dict[str, Any]) -> LLMConfig:
567 """Create from dictionary (JSON/YAML deserialization)."""
568 return cls(
569 enabled=data.get("enabled", True),
570 model=data.get("model", DEFAULT_LLM_MODEL),
571 max_tokens=data.get("max_tokens", 256),
572 timeout=data.get("timeout", 1800),
573 )
576@dataclass
577class LoopConfigOverrides:
578 """Per-loop ll-config overrides embedded in the loop YAML definition.
580 All fields are optional (None = use global ll-config default).
581 Precedence: CLI flags > YAML config block > global ll-config > schema defaults.
583 Attributes:
584 handoff_threshold: Override for LL_HANDOFF_THRESHOLD env var (1-100)
585 readiness_threshold: Override for commands.confidence_gate.readiness_threshold (1-100)
586 outcome_threshold: Override for commands.confidence_gate.outcome_threshold (1-100)
587 max_continuations: Override for automation.max_continuations (>=1)
588 """
590 handoff_threshold: int | None = None
591 readiness_threshold: int | None = None
592 outcome_threshold: int | None = None
593 max_continuations: int | None = None
595 def to_dict(self) -> dict[str, Any]:
596 """Convert to dictionary for JSON/YAML serialization (skip-if-None)."""
597 result: dict[str, Any] = {}
599 if self.handoff_threshold is not None:
600 result["handoff_threshold"] = self.handoff_threshold
602 confidence_gate: dict[str, Any] = {}
603 if self.readiness_threshold is not None:
604 confidence_gate["readiness_threshold"] = self.readiness_threshold
605 if self.outcome_threshold is not None:
606 confidence_gate["outcome_threshold"] = self.outcome_threshold
607 if confidence_gate:
608 result["commands"] = {"confidence_gate": confidence_gate}
610 if self.max_continuations is not None:
611 result["automation"] = {"max_continuations": self.max_continuations}
613 return result
615 @classmethod
616 def from_dict(cls, data: dict[str, Any]) -> LoopConfigOverrides:
617 """Create from dictionary (JSON/YAML deserialization)."""
618 commands = data.get("commands", {})
619 confidence_gate = commands.get("confidence_gate", {}) if isinstance(commands, dict) else {}
620 automation = data.get("automation", {})
621 continuation = data.get("continuation", {})
623 max_continuations = None
624 if isinstance(automation, dict) and "max_continuations" in automation:
625 max_continuations = automation["max_continuations"]
626 elif isinstance(continuation, dict) and "max_continuations" in continuation:
627 max_continuations = continuation["max_continuations"]
629 return cls(
630 handoff_threshold=data.get("handoff_threshold"),
631 readiness_threshold=confidence_gate.get("readiness_threshold")
632 if isinstance(confidence_gate, dict)
633 else None,
634 outcome_threshold=confidence_gate.get("outcome_threshold")
635 if isinstance(confidence_gate, dict)
636 else None,
637 max_continuations=max_continuations,
638 )
641@dataclass
642class CommandEntry:
643 """A single command entry for the loop's Commands display section.
645 Attributes:
646 cmd: Full command string to display (e.g., "ll-loop run my-loop --param x=1")
647 comment: Short description shown as a comment (e.g., "run with parameter x")
648 """
650 cmd: str
651 comment: str
654@dataclass
655class FSMLoop:
656 """Complete FSM loop definition.
658 The main dataclass representing a loop configuration.
660 Attributes:
661 name: Unique loop identifier
662 initial: Starting state name
663 states: Mapping from state name to StateConfig
664 context: User-defined shared variables
665 scope: Paths this loop operates on (for concurrency control)
666 max_iterations: Safety limit for loop iterations
667 backoff: Seconds between iterations
668 timeout: Max total runtime in seconds (loop-level)
669 maintain: If True, restart after completion
670 llm: LLM evaluation configuration
671 on_handoff: Behavior when handoff signal detected (pause/spawn/terminate)
672 commands: Optional override for the Commands section in ll-loop show
673 """
675 name: str
676 initial: str
677 states: dict[str, StateConfig]
678 description: str | None = None
679 context: dict[str, Any] = field(default_factory=dict)
680 parameters: dict[str, ParameterSpec] = field(default_factory=dict)
681 scope: list[str] = field(default_factory=list)
682 max_iterations: int = 50
683 max_edge_revisits: int = 100
684 backoff: float | None = None
685 timeout: int | None = None
686 default_timeout: int | None = None
687 maintain: bool = False
688 llm: LLMConfig = field(default_factory=LLMConfig)
689 on_handoff: Literal["pause", "spawn", "terminate"] = "pause"
690 input_key: str = "input"
691 config: LoopConfigOverrides | None = None
692 category: str = ""
693 labels: list[str] = field(default_factory=list)
694 commands: list[CommandEntry] = field(default_factory=list)
696 def to_dict(self) -> dict[str, Any]:
697 """Convert to dictionary for JSON/YAML serialization."""
698 result: dict[str, Any] = {
699 "name": self.name,
700 "initial": self.initial,
701 "states": {name: state.to_dict() for name, state in self.states.items()},
702 }
704 if self.description is not None:
705 result["description"] = self.description
706 if self.context:
707 result["context"] = self.context
708 if self.parameters:
709 result["parameters"] = {name: spec.to_dict() for name, spec in self.parameters.items()}
710 if self.scope:
711 result["scope"] = self.scope
712 if self.max_iterations != 50:
713 result["max_iterations"] = self.max_iterations
714 if self.max_edge_revisits != 100:
715 result["max_edge_revisits"] = self.max_edge_revisits
716 if self.backoff is not None:
717 result["backoff"] = self.backoff
718 if self.timeout is not None:
719 result["timeout"] = self.timeout
720 if self.default_timeout is not None:
721 result["default_timeout"] = self.default_timeout
722 if self.maintain:
723 result["maintain"] = self.maintain
724 if self.on_handoff != "pause":
725 result["on_handoff"] = self.on_handoff
726 if self.input_key != "input":
727 result["input_key"] = self.input_key
729 llm_dict = self.llm.to_dict()
730 if llm_dict:
731 result["llm"] = llm_dict
733 if self.config is not None:
734 config_dict = self.config.to_dict()
735 if config_dict:
736 result["config"] = config_dict
738 if self.category:
739 result["category"] = self.category
740 if self.labels:
741 result["labels"] = self.labels
742 if self.commands:
743 result["commands"] = [{"cmd": e.cmd, "comment": e.comment} for e in self.commands]
745 return result
747 @classmethod
748 def from_dict(cls, data: dict[str, Any]) -> FSMLoop:
749 """Create from dictionary (JSON/YAML deserialization)."""
750 states = {
751 name: StateConfig.from_dict(state_data)
752 for name, state_data in data.get("states", {}).items()
753 }
755 llm = LLMConfig()
756 if "llm" in data:
757 llm = LLMConfig.from_dict(data["llm"])
759 loop_config = None
760 if "config" in data:
761 loop_config = LoopConfigOverrides.from_dict(data["config"])
763 parameters = {
764 name: ParameterSpec.from_dict(spec) for name, spec in data.get("parameters", {}).items()
765 }
767 return cls(
768 name=data["name"],
769 initial=data["initial"],
770 states=states,
771 description=data.get("description"),
772 context=data.get("context", {}),
773 parameters=parameters,
774 scope=data.get("scope", []),
775 max_iterations=data.get("max_iterations", 50),
776 max_edge_revisits=data.get("max_edge_revisits", 100),
777 backoff=data.get("backoff"),
778 timeout=data.get("timeout"),
779 default_timeout=data.get("default_timeout"),
780 maintain=data.get("maintain", False),
781 llm=llm,
782 on_handoff=data.get("on_handoff", "pause"),
783 input_key=data.get("input_key", "input"),
784 config=loop_config,
785 category=data.get("category", ""),
786 labels=data.get("labels", []),
787 commands=[CommandEntry(**e) for e in data.get("commands", [])],
788 )
790 def get_all_state_names(self) -> set[str]:
791 """Get all defined state names.
793 Returns:
794 Set of all state names in this FSM.
795 """
796 return set(self.states.keys())
798 def get_terminal_states(self) -> set[str]:
799 """Get all terminal state names.
801 Returns:
802 Set of state names where terminal=True.
803 """
804 return {name for name, state in self.states.items() if state.terminal}
806 def get_all_referenced_states(self) -> set[str]:
807 """Get all state names referenced by transitions.
809 This includes the initial state and all states referenced
810 in routing configurations.
812 Returns:
813 Set of all referenced state names.
814 """
815 refs: set[str] = {self.initial}
816 for state in self.states.values():
817 refs.update(state.get_referenced_states())
818 return refs