Coverage for little_loops / fsm / schema.py: 37%
486 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -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 LearningConfig:
272 """Per-state configuration for FEAT-1283 `type: learning` dispatch.
274 Declares the list of external-API/SDK targets a learning state must prove
275 against the learning-tests registry (ENH-1282) before advancing. On a
276 missing or stale record, the state invokes `/ll:explore-api <target>` up to
277 `max_retries` times before transitioning to a blocked target.
279 Attributes:
280 targets: Ordered list of target identifiers (e.g. "Anthropic SDK
281 streaming"). Each is slugified internally when looking up the
282 registry record. All targets must reach status="proven" for the
283 state to advance via on_yes.
284 max_retries: Maximum number of `/ll:explore-api` invocations per target
285 before the state routes to on_blocked / on_no with reason
286 ``retries_exhausted``. Counts only re-exploration attempts; the
287 initial registry lookup is free. Distinct from ENH-1115's throttle
288 counter, which measures tool-call volume; learning states are
289 already exempt from throttle hard_max via FSMExecutor._check_throttle.
290 """
292 targets: list[str]
293 max_retries: int = 2
295 def to_dict(self) -> dict[str, Any]:
296 """Convert to dictionary for JSON/YAML serialization."""
297 return {"targets": list(self.targets), "max_retries": self.max_retries}
299 @classmethod
300 def from_dict(cls, data: dict[str, Any]) -> LearningConfig:
301 """Create from dictionary (JSON/YAML deserialization)."""
302 return cls(
303 targets=list(data.get("targets") or []),
304 max_retries=int(data.get("max_retries", 2)),
305 )
308@dataclass
309class StateConfig:
310 """Configuration for a single FSM state.
312 States can have actions, evaluators, and routing. Supports both
313 shorthand (on_success/on_failure) and full routing table syntax.
315 Attributes:
316 action: Command to execute (shell, slash command, or "server/tool-name" for mcp_tool)
317 action_type: How to execute the action (prompt, slash_command, shell, mcp_tool).
318 If None, uses heuristic: / prefix = slash_command, else = shell.
319 params: MCP tool arguments (only used with action_type: mcp_tool). Supports
320 ${variable} interpolation in string values.
321 evaluate: Evaluator configuration for result interpretation
322 route: Full routing table (verdict -> state mapping)
323 on_yes: Shorthand for yes verdict routing
324 on_no: Shorthand for no verdict routing
325 on_error: Shorthand for error verdict routing
326 on_partial: Shorthand for partial verdict routing
327 next: Unconditional transition (no evaluation)
328 terminal: If True, this is an end state
329 capture: Variable name to store action output
330 timeout: Action-level timeout in seconds
331 on_maintain: State to transition to when maintain=True and loop completes
332 max_retries: Max consecutive re-entries before transitioning to on_retry_exhausted.
333 A value of N allows N retries after the initial execution (N+1 total entries).
334 Requires on_retry_exhausted to also be set.
335 on_retry_exhausted: State to transition to when max_retries consecutive re-entries
336 are exceeded. Required when max_retries is set.
337 max_rate_limit_retries: Max consecutive 429/rate-limit in-place retries for this
338 state before transitioning to on_rate_limit_exhausted. Requires
339 on_rate_limit_exhausted to also be set.
340 on_rate_limit_exhausted: State to transition to when max_rate_limit_retries
341 consecutive rate-limit retries are exceeded. Required when
342 max_rate_limit_retries is set.
343 rate_limit_backoff_base_seconds: Base seconds for exponential backoff between
344 rate-limit retries; actual sleep is base * 2^(attempt-1) + uniform(0, base).
345 Defaults to 30.
346 rate_limit_max_wait_seconds: Total wall-clock budget (seconds) for rate-limit
347 handling in this state before routing to on_rate_limit_exhausted. When unset,
348 defaults from commands.rate_limits.max_wait_seconds (21600 = 6h).
349 rate_limit_long_wait_ladder: Backoff ladder (seconds) for the long-wait tier used
350 once the short-tier retry budget is spent. Each entry is the sleep before the
351 next retry attempt. When unset, defaults from
352 commands.rate_limits.long_wait_ladder ([300, 900, 1800, 3600]).
353 loop: Name of a loop YAML to execute as a sub-FSM. Mutually exclusive with action.
354 context_passthrough: When True, pass parent context variables to child loop and
355 merge child captures back into parent context. Legacy escape hatch; prefer
356 'with_' for explicit named bindings.
357 with_: Explicit parameter bindings for sub-loop calls (YAML key: 'with'). Maps
358 declared child parameter names to parent expressions. Only valid when 'loop'
359 is set. Mutually exclusive with context_passthrough.
360 """
362 action: str | None = None
363 action_type: str | None = None
364 params: dict[str, Any] = field(default_factory=dict)
365 evaluate: EvaluateConfig | None = None
366 route: RouteConfig | None = None
367 on_yes: str | None = None
368 on_no: str | None = None
369 on_error: str | None = None
370 on_partial: str | None = None
371 on_blocked: str | None = None
372 next: str | None = None
373 terminal: bool = False
374 capture: str | None = None
375 timeout: int | None = None
376 on_maintain: str | None = None
377 max_retries: int | None = None
378 on_retry_exhausted: str | None = None
379 max_rate_limit_retries: int | None = None
380 on_rate_limit_exhausted: str | None = None
381 rate_limit_backoff_base_seconds: int | None = None
382 rate_limit_max_wait_seconds: int | None = None
383 rate_limit_long_wait_ladder: list[int] | None = None
384 loop: str | None = None
385 context_passthrough: bool = False
386 with_: dict[str, Any] = field(default_factory=dict)
387 agent: str | None = None
388 tools: list[str] | None = None
389 extra_routes: dict[str, str] = field(default_factory=dict)
390 type: str | None = None
391 throttle: ThrottleConfig | None = None
392 on_throttle_hard: str | None = None
393 learning: LearningConfig | None = None
395 def to_dict(self) -> dict[str, Any]:
396 """Convert to dictionary for JSON/YAML serialization."""
397 result: dict[str, Any] = {}
399 if self.action is not None:
400 result["action"] = self.action
401 if self.action_type is not None:
402 result["action_type"] = self.action_type
403 if self.params:
404 result["params"] = self.params
405 if self.evaluate is not None:
406 result["evaluate"] = self.evaluate.to_dict()
407 if self.route is not None:
408 result["route"] = self.route.to_dict()
409 if self.on_yes is not None:
410 result["on_yes"] = self.on_yes
411 if self.on_no is not None:
412 result["on_no"] = self.on_no
413 if self.on_error is not None:
414 result["on_error"] = self.on_error
415 if self.on_partial is not None:
416 result["on_partial"] = self.on_partial
417 if self.on_blocked is not None:
418 result["on_blocked"] = self.on_blocked
419 if self.next is not None:
420 result["next"] = self.next
421 if self.terminal:
422 result["terminal"] = self.terminal
423 if self.capture is not None:
424 result["capture"] = self.capture
425 if self.timeout is not None:
426 result["timeout"] = self.timeout
427 if self.on_maintain is not None:
428 result["on_maintain"] = self.on_maintain
429 if self.max_retries is not None:
430 result["max_retries"] = self.max_retries
431 if self.on_retry_exhausted is not None:
432 result["on_retry_exhausted"] = self.on_retry_exhausted
433 if self.max_rate_limit_retries is not None:
434 result["max_rate_limit_retries"] = self.max_rate_limit_retries
435 if self.on_rate_limit_exhausted is not None:
436 result["on_rate_limit_exhausted"] = self.on_rate_limit_exhausted
437 if self.rate_limit_backoff_base_seconds is not None:
438 result["rate_limit_backoff_base_seconds"] = self.rate_limit_backoff_base_seconds
439 if self.rate_limit_max_wait_seconds is not None:
440 result["rate_limit_max_wait_seconds"] = self.rate_limit_max_wait_seconds
441 if self.rate_limit_long_wait_ladder is not None:
442 result["rate_limit_long_wait_ladder"] = self.rate_limit_long_wait_ladder
443 if self.loop is not None:
444 result["loop"] = self.loop
445 if self.context_passthrough:
446 result["context_passthrough"] = self.context_passthrough
447 if self.with_:
448 result["with"] = self.with_
449 if self.agent is not None:
450 result["agent"] = self.agent
451 if self.tools is not None:
452 result["tools"] = self.tools
453 for verdict, target in self.extra_routes.items():
454 result[f"on_{verdict}"] = target
455 if self.type is not None:
456 result["type"] = self.type
457 if self.throttle is not None:
458 result["throttle"] = self.throttle.to_dict()
459 if self.on_throttle_hard is not None:
460 result["on_throttle_hard"] = self.on_throttle_hard
461 if self.learning is not None:
462 result["learning"] = self.learning.to_dict()
464 return result
466 @classmethod
467 def from_dict(cls, data: dict[str, Any]) -> StateConfig:
468 """Create from dictionary (JSON/YAML deserialization)."""
469 evaluate = None
470 if "evaluate" in data:
471 evaluate = EvaluateConfig.from_dict(data["evaluate"])
473 route = None
474 if "route" in data:
475 route = RouteConfig.from_dict(data["route"])
477 throttle = None
478 if "throttle" in data:
479 throttle = ThrottleConfig.from_dict(data["throttle"])
481 learning = None
482 if "learning" in data:
483 learning = LearningConfig.from_dict(data["learning"])
485 _known_on_keys = {
486 "on_yes",
487 "on_success",
488 "on_no",
489 "on_failure",
490 "on_error",
491 "on_partial",
492 "on_blocked",
493 "on_maintain",
494 "on_retry_exhausted",
495 "on_rate_limit_exhausted",
496 "on_throttle_hard",
497 }
498 extra_routes = {
499 key[3:]: val
500 for key, val in data.items()
501 if key.startswith("on_") and key not in _known_on_keys and isinstance(val, str)
502 }
504 return cls(
505 action=data.get("action"),
506 action_type=data.get("action_type"),
507 params=data.get("params", {}),
508 evaluate=evaluate,
509 route=route,
510 on_yes=data.get("on_yes") or data.get("on_success"),
511 on_no=data.get("on_no") or data.get("on_failure"),
512 on_error=data.get("on_error"),
513 on_partial=data.get("on_partial"),
514 on_blocked=data.get("on_blocked"),
515 next=data.get("next"),
516 terminal=data.get("terminal", False),
517 capture=data.get("capture"),
518 timeout=data.get("timeout"),
519 on_maintain=data.get("on_maintain"),
520 max_retries=data.get("max_retries"),
521 on_retry_exhausted=data.get("on_retry_exhausted"),
522 max_rate_limit_retries=data.get("max_rate_limit_retries"),
523 on_rate_limit_exhausted=data.get("on_rate_limit_exhausted"),
524 rate_limit_backoff_base_seconds=data.get("rate_limit_backoff_base_seconds"),
525 rate_limit_max_wait_seconds=data.get("rate_limit_max_wait_seconds"),
526 rate_limit_long_wait_ladder=data.get("rate_limit_long_wait_ladder"),
527 loop=data.get("loop"),
528 context_passthrough=data.get("context_passthrough", False),
529 with_=data.get("with", {}),
530 agent=data.get("agent"),
531 tools=data.get("tools"),
532 extra_routes=extra_routes,
533 type=data.get("type"),
534 throttle=throttle,
535 on_throttle_hard=data.get("on_throttle_hard"),
536 learning=learning,
537 )
539 def get_referenced_states(self) -> set[str]:
540 """Get all state names referenced by this state configuration.
542 Returns:
543 Set of state names that this state can transition to.
544 """
545 refs: set[str] = set()
547 if self.on_yes is not None:
548 refs.add(self.on_yes)
549 if self.on_no is not None:
550 refs.add(self.on_no)
551 if self.on_error is not None:
552 refs.add(self.on_error)
553 if self.on_partial is not None:
554 refs.add(self.on_partial)
555 if self.on_blocked is not None:
556 refs.add(self.on_blocked)
557 if self.next is not None:
558 refs.add(self.next)
559 if self.on_maintain is not None:
560 refs.add(self.on_maintain)
561 if self.on_retry_exhausted is not None:
562 refs.add(self.on_retry_exhausted)
563 if self.on_rate_limit_exhausted is not None:
564 refs.add(self.on_rate_limit_exhausted)
565 if self.on_throttle_hard is not None:
566 refs.add(self.on_throttle_hard)
567 if self.route is not None:
568 refs.update(self.route.routes.values())
569 if self.route.default is not None:
570 refs.add(self.route.default)
571 if self.route.error is not None:
572 refs.add(self.route.error)
573 refs.update(self.extra_routes.values())
575 return refs
578@dataclass
579class LLMConfig:
580 """LLM evaluation configuration.
582 Settings for the llm_structured evaluator.
584 Attributes:
585 enabled: If False, disable LLM evaluation entirely
586 model: Model identifier for LLM calls
587 max_tokens: Maximum tokens for evaluation response
588 timeout: Timeout for LLM calls in seconds
589 """
591 enabled: bool = True
592 model: str = DEFAULT_LLM_MODEL
593 max_tokens: int = 256
594 timeout: int = 1800
596 def to_dict(self) -> dict[str, Any]:
597 """Convert to dictionary for JSON/YAML serialization."""
598 result: dict[str, Any] = {}
600 if not self.enabled:
601 result["enabled"] = self.enabled
602 if self.model != DEFAULT_LLM_MODEL:
603 result["model"] = self.model
604 if self.max_tokens != 256:
605 result["max_tokens"] = self.max_tokens
606 if self.timeout != 1800:
607 result["timeout"] = self.timeout
609 return result if result else {}
611 @classmethod
612 def from_dict(cls, data: dict[str, Any]) -> LLMConfig:
613 """Create from dictionary (JSON/YAML deserialization)."""
614 return cls(
615 enabled=data.get("enabled", True),
616 model=data.get("model", DEFAULT_LLM_MODEL),
617 max_tokens=data.get("max_tokens", 256),
618 timeout=data.get("timeout", 1800),
619 )
622@dataclass
623class LoopConfigOverrides:
624 """Per-loop ll-config overrides embedded in the loop YAML definition.
626 All fields are optional (None = use global ll-config default).
627 Precedence: CLI flags > YAML config block > global ll-config > schema defaults.
629 Attributes:
630 handoff_threshold: Override for LL_HANDOFF_THRESHOLD env var (1-100)
631 readiness_threshold: Override for commands.confidence_gate.readiness_threshold (1-100)
632 outcome_threshold: Override for commands.confidence_gate.outcome_threshold (1-100)
633 max_continuations: Override for automation.max_continuations (>=1)
634 """
636 handoff_threshold: int | None = None
637 readiness_threshold: int | None = None
638 outcome_threshold: int | None = None
639 max_continuations: int | None = None
641 def to_dict(self) -> dict[str, Any]:
642 """Convert to dictionary for JSON/YAML serialization (skip-if-None)."""
643 result: dict[str, Any] = {}
645 if self.handoff_threshold is not None:
646 result["handoff_threshold"] = self.handoff_threshold
648 confidence_gate: dict[str, Any] = {}
649 if self.readiness_threshold is not None:
650 confidence_gate["readiness_threshold"] = self.readiness_threshold
651 if self.outcome_threshold is not None:
652 confidence_gate["outcome_threshold"] = self.outcome_threshold
653 if confidence_gate:
654 result["commands"] = {"confidence_gate": confidence_gate}
656 if self.max_continuations is not None:
657 result["automation"] = {"max_continuations": self.max_continuations}
659 return result
661 @classmethod
662 def from_dict(cls, data: dict[str, Any]) -> LoopConfigOverrides:
663 """Create from dictionary (JSON/YAML deserialization)."""
664 commands = data.get("commands", {})
665 confidence_gate = commands.get("confidence_gate", {}) if isinstance(commands, dict) else {}
666 automation = data.get("automation", {})
667 continuation = data.get("continuation", {})
669 max_continuations = None
670 if isinstance(automation, dict) and "max_continuations" in automation:
671 max_continuations = automation["max_continuations"]
672 elif isinstance(continuation, dict) and "max_continuations" in continuation:
673 max_continuations = continuation["max_continuations"]
675 return cls(
676 handoff_threshold=data.get("handoff_threshold"),
677 readiness_threshold=confidence_gate.get("readiness_threshold")
678 if isinstance(confidence_gate, dict)
679 else None,
680 outcome_threshold=confidence_gate.get("outcome_threshold")
681 if isinstance(confidence_gate, dict)
682 else None,
683 max_continuations=max_continuations,
684 )
687@dataclass
688class CommandEntry:
689 """A single command entry for the loop's Commands display section.
691 Attributes:
692 cmd: Full command string to display (e.g., "ll-loop run my-loop --param x=1")
693 comment: Short description shown as a comment (e.g., "run with parameter x")
694 """
696 cmd: str
697 comment: str
700@dataclass
701class TargetStateSpec:
702 """Per-state targeting specification for harness-optimize APO (ENH-1552).
704 Names a single FSM state inside a target loop file and associates it with
705 the examples file and eval fragment used during that state's optimization.
707 Attributes:
708 name: State name within the target loop
709 examples_file: Path to the examples YAML file for this state
710 eval_fragment: Eval fragment identifier used during optimization
711 """
713 name: str
714 examples_file: str
715 eval_fragment: str
717 def to_dict(self) -> dict[str, Any]:
718 """Convert to dictionary for JSON/YAML serialization."""
719 return {
720 "name": self.name,
721 "examples_file": self.examples_file,
722 "eval": self.eval_fragment,
723 }
725 @classmethod
726 def from_dict(cls, data: dict[str, Any]) -> TargetStateSpec:
727 """Create from dictionary (JSON/YAML deserialization)."""
728 return cls(
729 name=data["name"],
730 examples_file=data["examples_file"],
731 eval_fragment=data["eval"],
732 )
735@dataclass
736class TargetFileSpec:
737 """Per-file targeting specification for harness-optimize APO (ENH-1552).
739 Associates a loop YAML file (or glob pattern) with the list of states
740 to optimize within that file.
742 Attributes:
743 file: Explicit path to a loop YAML file (mutually exclusive with glob)
744 glob: Glob pattern matching one or more loop YAML files
745 states: States within the matched file(s) to target
746 """
748 file: str | None = None
749 glob: str | None = None
750 states: list[TargetStateSpec] = field(default_factory=list)
752 def to_dict(self) -> dict[str, Any]:
753 """Convert to dictionary for JSON/YAML serialization."""
754 result: dict[str, Any] = {}
755 if self.file is not None:
756 result["file"] = self.file
757 if self.glob is not None:
758 result["glob"] = self.glob
759 if self.states:
760 result["states"] = [s.to_dict() for s in self.states]
761 return result
763 @classmethod
764 def from_dict(cls, data: dict[str, Any]) -> TargetFileSpec:
765 """Create from dictionary (JSON/YAML deserialization)."""
766 return cls(
767 file=data.get("file"),
768 glob=data.get("glob"),
769 states=[TargetStateSpec.from_dict(s) for s in (data.get("states") or [])],
770 )
773@dataclass
774class RepeatedFailureConfig:
775 """Configuration for the FSM stall detector (FEAT-1637).
777 When N consecutive iterations produce an identical
778 `(state_name, exit_code, eval_verdict)` triple, the FSM either
779 aborts the run or routes to a configured recovery state.
781 Attributes:
782 window: Consecutive iterations with identical triple required to
783 fire (default 3).
784 on_repeated_failure: Either the literal ``"abort"`` (terminate
785 with ``terminated_by="stall_detected"``) or the name of a
786 declared state to route to.
787 """
789 window: int = 3
790 on_repeated_failure: str = "abort"
791 progress_paths: list[str] = field(default_factory=list)
793 def to_dict(self) -> dict[str, Any]:
794 """Convert to dictionary for JSON/YAML serialization (skip-if-default)."""
795 result: dict[str, Any] = {}
796 if self.window != 3:
797 result["window"] = self.window
798 if self.on_repeated_failure != "abort":
799 result["on_repeated_failure"] = self.on_repeated_failure
800 if self.progress_paths:
801 result["progress_paths"] = self.progress_paths
802 return result
804 @classmethod
805 def from_dict(cls, data: dict[str, Any]) -> RepeatedFailureConfig:
806 """Create from dictionary (JSON/YAML deserialization)."""
807 return cls(
808 window=data.get("window", 3),
809 on_repeated_failure=data.get("on_repeated_failure", "abort"),
810 progress_paths=data.get("progress_paths", []),
811 )
814@dataclass
815class CircuitConfig:
816 """Top-level ``circuit:`` block grouping loop-level safety knobs.
818 Currently exposes ``repeated_failure`` (the stall detector). Future
819 safety knobs (e.g. global timeouts, panic-stop guards) should be
820 added here rather than as separate top-level keys.
822 Attributes:
823 repeated_failure: Stall-detector configuration, or None to disable.
824 """
826 repeated_failure: RepeatedFailureConfig | None = None
828 def to_dict(self) -> dict[str, Any]:
829 """Convert to dictionary for JSON/YAML serialization (skip-if-None)."""
830 result: dict[str, Any] = {}
831 if self.repeated_failure is not None:
832 rf_dict = self.repeated_failure.to_dict()
833 # Always emit the key when configured (even if all fields are defaults)
834 # so the round-trip preserves "repeated_failure was set" intent.
835 result["repeated_failure"] = rf_dict
836 return result
838 @classmethod
839 def from_dict(cls, data: dict[str, Any]) -> CircuitConfig:
840 """Create from dictionary (JSON/YAML deserialization)."""
841 rf = None
842 if "repeated_failure" in data and data["repeated_failure"] is not None:
843 rf = RepeatedFailureConfig.from_dict(data["repeated_failure"])
844 return cls(repeated_failure=rf)
847@dataclass
848class FSMLoop:
849 """Complete FSM loop definition.
851 The main dataclass representing a loop configuration.
853 Attributes:
854 name: Unique loop identifier
855 initial: Starting state name
856 states: Mapping from state name to StateConfig
857 context: User-defined shared variables
858 scope: Paths this loop operates on (for concurrency control)
859 max_iterations: Safety limit for loop iterations
860 backoff: Seconds between iterations
861 timeout: Max total runtime in seconds (loop-level)
862 maintain: If True, restart after completion
863 llm: LLM evaluation configuration
864 on_handoff: Behavior when handoff signal detected (pause/spawn/terminate)
865 commands: Optional override for the Commands section in ll-loop show
866 """
868 name: str
869 initial: str
870 states: dict[str, StateConfig]
871 description: str | None = None
872 context: dict[str, Any] = field(default_factory=dict)
873 parameters: dict[str, ParameterSpec] = field(default_factory=dict)
874 scope: list[str] = field(default_factory=list)
875 max_iterations: int = 50
876 on_max_iterations: str | None = None
877 max_edge_revisits: int = 100
878 backoff: float | None = None
879 timeout: int | None = None
880 default_timeout: int | None = None
881 maintain: bool = False
882 llm: LLMConfig = field(default_factory=LLMConfig)
883 on_handoff: Literal["pause", "spawn", "terminate"] = "pause"
884 input_key: str = "input"
885 config: LoopConfigOverrides | None = None
886 category: str = ""
887 labels: list[str] = field(default_factory=list)
888 commands: list[CommandEntry] = field(default_factory=list)
889 targets: list[TargetFileSpec] = field(default_factory=list)
890 circuit: CircuitConfig | None = None
891 meta_self_eval_ok: bool = False
892 # Populated from the raw `import:` list by from_dict(); not serialized by to_dict()
893 imports: list[str] = field(default_factory=list)
895 def to_dict(self) -> dict[str, Any]:
896 """Convert to dictionary for JSON/YAML serialization."""
897 result: dict[str, Any] = {
898 "name": self.name,
899 "initial": self.initial,
900 "states": {name: state.to_dict() for name, state in self.states.items()},
901 }
903 if self.description is not None:
904 result["description"] = self.description
905 if self.context:
906 result["context"] = self.context
907 if self.parameters:
908 result["parameters"] = {name: spec.to_dict() for name, spec in self.parameters.items()}
909 if self.scope:
910 result["scope"] = self.scope
911 if self.max_iterations != 50:
912 result["max_iterations"] = self.max_iterations
913 if self.max_edge_revisits != 100:
914 result["max_edge_revisits"] = self.max_edge_revisits
915 if self.backoff is not None:
916 result["backoff"] = self.backoff
917 if self.timeout is not None:
918 result["timeout"] = self.timeout
919 if self.default_timeout is not None:
920 result["default_timeout"] = self.default_timeout
921 if self.maintain:
922 result["maintain"] = self.maintain
923 if self.on_handoff != "pause":
924 result["on_handoff"] = self.on_handoff
925 if self.on_max_iterations is not None:
926 result["on_max_iterations"] = self.on_max_iterations
927 if self.input_key != "input":
928 result["input_key"] = self.input_key
930 llm_dict = self.llm.to_dict()
931 if llm_dict:
932 result["llm"] = llm_dict
934 if self.config is not None:
935 config_dict = self.config.to_dict()
936 if config_dict:
937 result["config"] = config_dict
939 if self.category:
940 result["category"] = self.category
941 if self.labels:
942 result["labels"] = self.labels
943 if self.commands:
944 result["commands"] = [{"cmd": e.cmd, "comment": e.comment} for e in self.commands]
945 if self.targets:
946 result["targets"] = [t.to_dict() for t in self.targets]
948 if self.circuit is not None:
949 circuit_dict = self.circuit.to_dict()
950 if circuit_dict:
951 result["circuit"] = circuit_dict
953 if self.meta_self_eval_ok:
954 result["meta_self_eval_ok"] = self.meta_self_eval_ok
956 return result
958 @classmethod
959 def from_dict(cls, data: dict[str, Any]) -> FSMLoop:
960 """Create from dictionary (JSON/YAML deserialization)."""
961 states = {
962 name: StateConfig.from_dict(state_data)
963 for name, state_data in data.get("states", {}).items()
964 }
966 llm = LLMConfig()
967 if "llm" in data:
968 llm = LLMConfig.from_dict(data["llm"])
970 loop_config = None
971 if "config" in data:
972 loop_config = LoopConfigOverrides.from_dict(data["config"])
974 circuit = None
975 if "circuit" in data and data["circuit"] is not None:
976 circuit = CircuitConfig.from_dict(data["circuit"])
978 parameters = {
979 name: ParameterSpec.from_dict(spec) for name, spec in data.get("parameters", {}).items()
980 }
982 return cls(
983 name=data["name"],
984 initial=data["initial"],
985 states=states,
986 description=data.get("description"),
987 context=data.get("context", {}),
988 parameters=parameters,
989 scope=data.get("scope", []),
990 max_iterations=data.get("max_iterations", 50),
991 on_max_iterations=data.get("on_max_iterations"),
992 max_edge_revisits=data.get("max_edge_revisits", 100),
993 backoff=data.get("backoff"),
994 timeout=data.get("timeout"),
995 default_timeout=data.get("default_timeout"),
996 maintain=data.get("maintain", False),
997 llm=llm,
998 on_handoff=data.get("on_handoff", "pause"),
999 input_key=data.get("input_key", "input"),
1000 config=loop_config,
1001 category=data.get("category", ""),
1002 labels=data.get("labels", []),
1003 commands=[CommandEntry(**e) for e in data.get("commands", [])],
1004 targets=[TargetFileSpec.from_dict(t) for t in (data.get("targets") or [])],
1005 circuit=circuit,
1006 meta_self_eval_ok=data.get("meta_self_eval_ok", False),
1007 imports=data.get("import", []),
1008 )
1010 def get_all_state_names(self) -> set[str]:
1011 """Get all defined state names.
1013 Returns:
1014 Set of all state names in this FSM.
1015 """
1016 return set(self.states.keys())
1018 def get_terminal_states(self) -> set[str]:
1019 """Get all terminal state names.
1021 Returns:
1022 Set of state names where terminal=True.
1023 """
1024 return {name for name, state in self.states.items() if state.terminal}
1026 def get_all_referenced_states(self) -> set[str]:
1027 """Get all state names referenced by transitions.
1029 This includes the initial state and all states referenced
1030 in routing configurations.
1032 Returns:
1033 Set of all referenced state names.
1034 """
1035 refs: set[str] = {self.initial}
1036 for state in self.states.values():
1037 refs.update(state.get_referenced_states())
1038 if self.on_max_iterations is not None:
1039 refs.add(self.on_max_iterations)
1040 return refs