Coverage for little_loops / fsm / schema.py: 44%
540 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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 pairs: List of producer/consumer pair dicts for contract evaluator
55 """
57 type: Literal[
58 "exit_code",
59 "output_numeric",
60 "output_json",
61 "output_contains",
62 "convergence",
63 "diff_stall",
64 "action_stall",
65 "llm_structured",
66 "mcp_result",
67 "harbor_scorer",
68 "comparator",
69 "contract",
70 ]
71 operator: str | None = None
72 target: int | float | str | None = None
73 tolerance: float | str | None = None # str for interpolation (e.g., "${context.tolerance}")
74 pattern: str | None = None
75 negate: bool = False
76 path: str | None = None
77 prompt: str | None = None
78 schema: dict[str, Any] | None = None
79 min_confidence: float = 0.5
80 uncertain_suffix: bool = False
81 source: str | None = None
82 previous: str | None = None
83 direction: Literal["minimize", "maximize"] = "minimize"
84 scope: list[str] | None = None # for diff_stall: limit git diff to these paths
85 max_stall: int = 1 # for diff_stall: consecutive no-change iterations before failure
86 track: list[str] | None = None # for action_stall: context keys to track (default: ["action"])
87 max_repeat: int = 2 # for action_stall: consecutive identical iterations before failure
88 baseline_path: str | None = None # for comparator: path to .loops/baselines/<loop>/ dir
89 auto_promote: bool = False # for comparator: write output to baseline on yes verdict
90 min_pairs: int = 1 # for comparator: number of blind A/B comparisons to run
91 pairs: list[dict] | None = None # for contract: list of producer/consumer pair dicts
93 def to_dict(self) -> dict[str, Any]:
94 """Convert to dictionary for JSON/YAML serialization."""
95 result: dict[str, Any] = {"type": self.type}
97 # Only include non-None optional fields
98 if self.operator is not None:
99 result["operator"] = self.operator
100 if self.target is not None:
101 result["target"] = self.target
102 if self.tolerance is not None:
103 result["tolerance"] = self.tolerance
104 if self.pattern is not None:
105 result["pattern"] = self.pattern
106 if self.negate:
107 result["negate"] = self.negate
108 if self.path is not None:
109 result["path"] = self.path
110 if self.prompt is not None:
111 result["prompt"] = self.prompt
112 if self.schema is not None:
113 result["schema"] = self.schema
114 if self.min_confidence != 0.5:
115 result["min_confidence"] = self.min_confidence
116 if self.uncertain_suffix:
117 result["uncertain_suffix"] = self.uncertain_suffix
118 if self.source is not None:
119 result["source"] = self.source
120 if self.previous is not None:
121 result["previous"] = self.previous
122 if self.direction != "minimize":
123 result["direction"] = self.direction
124 if self.scope is not None:
125 result["scope"] = self.scope
126 if self.max_stall != 1:
127 result["max_stall"] = self.max_stall
128 if self.track is not None:
129 result["track"] = self.track
130 if self.max_repeat != 2:
131 result["max_repeat"] = self.max_repeat
132 if self.baseline_path is not None:
133 result["baseline_path"] = self.baseline_path
134 if self.auto_promote:
135 result["auto_promote"] = self.auto_promote
136 if self.min_pairs != 1:
137 result["min_pairs"] = self.min_pairs
138 if self.pairs is not None:
139 result["pairs"] = self.pairs
141 return result
143 @classmethod
144 def from_dict(cls, data: dict[str, Any]) -> EvaluateConfig:
145 """Create from dictionary (JSON/YAML deserialization)."""
146 return cls(
147 type=data["type"],
148 operator=data.get("operator"),
149 target=data.get("target"),
150 tolerance=data.get("tolerance"),
151 pattern=data.get("pattern"),
152 negate=data.get("negate", False),
153 path=data.get("path"),
154 prompt=data.get("prompt"),
155 schema=data.get("schema"),
156 min_confidence=data.get("min_confidence", 0.5),
157 uncertain_suffix=data.get("uncertain_suffix", False),
158 source=data.get("source"),
159 previous=data.get("previous"),
160 direction=data.get("direction", "minimize"),
161 scope=data.get("scope"),
162 max_stall=data.get("max_stall", 1),
163 track=data.get("track"),
164 max_repeat=data.get("max_repeat", 2),
165 baseline_path=data.get("baseline_path"),
166 auto_promote=data.get("auto_promote", False),
167 min_pairs=data.get("min_pairs", 1),
168 pairs=data.get("pairs"),
169 )
172@dataclass
173class RouteConfig:
174 """Routing table configuration for verdict-to-state mapping.
176 Maps verdict strings from evaluators to next state names.
178 Attributes:
179 routes: Mapping from verdict string to next state name
180 default: Default state for unmatched verdicts (the "_" key)
181 error: State for evaluation/execution errors (the "_error" key)
182 """
184 routes: dict[str, str] = field(default_factory=dict)
185 default: str | None = None
186 error: str | None = None
188 def to_dict(self) -> dict[str, Any]:
189 """Convert to dictionary for JSON/YAML serialization."""
190 result = dict(self.routes)
191 if self.default is not None:
192 result["_"] = self.default
193 if self.error is not None:
194 result["_error"] = self.error
195 return result
197 @classmethod
198 def from_dict(cls, data: dict[str, Any]) -> RouteConfig:
199 """Create from dictionary (JSON/YAML deserialization)."""
200 routes = {k: v for k, v in data.items() if not k.startswith("_")}
201 return cls(
202 routes=routes,
203 default=data.get("_"),
204 error=data.get("_error"),
205 )
208@dataclass
209class ParameterSpec:
210 """Specification for a single loop input parameter.
212 Declares a typed input that callers bind via the 'with:' block on sub-loop states.
214 Attributes:
215 type: Parameter type. One of: string, integer, number, boolean, enum, path.
216 required: If True, callers must supply this parameter in 'with:' (mutually
217 exclusive with 'default').
218 default: Default value used when the caller does not bind the parameter.
219 Only valid when required is False.
220 description: Human-readable description of the parameter.
221 values: Allowed values for 'enum' type parameters.
222 """
224 type: str
225 required: bool = False
226 default: Any = None
227 description: str | None = None
228 values: list[Any] | None = None
230 def to_dict(self) -> dict[str, Any]:
231 """Convert to dictionary for JSON/YAML serialization."""
232 result: dict[str, Any] = {"type": self.type}
233 if self.required:
234 result["required"] = self.required
235 if self.default is not None:
236 result["default"] = self.default
237 if self.description is not None:
238 result["description"] = self.description
239 if self.values is not None:
240 result["values"] = self.values
241 return result
243 @classmethod
244 def from_dict(cls, data: dict[str, Any]) -> ParameterSpec:
245 """Create from dictionary (JSON/YAML deserialization)."""
246 return cls(
247 type=data["type"],
248 required=data.get("required", False),
249 default=data.get("default"),
250 description=data.get("description"),
251 values=data.get("values"),
252 )
255@dataclass
256class ThrottleConfig:
257 """Per-state tool-call progressive throttling configuration.
259 Counts successful tool calls within a single state visit and escalates
260 restrictions to self-throttle runaway states before provider limits are reached.
262 Thresholds fall back to executor module defaults when not set:
263 - normal_max: _DEFAULT_THROTTLE_NORMAL_MAX (3) — calls 1..normal_max pass through
264 - warn_max: _DEFAULT_THROTTLE_WARN_MAX (8) — calls at warn_max inject a warning event
265 - hard_max: _DEFAULT_THROTTLE_HARD_MAX (12) — calls at hard_max route to on_throttle_hard
266 - calls > hard_max: hard stop, loop marked stuck
268 States with type="learning" (FEAT-1283) are exempt from the hard_max hard-stop because
269 they legitimately make N tool calls per visit (one per unproven target). The warn_max
270 warning still applies so users can see the state is doing significant work.
271 """
273 normal_max: int | None = None
274 warn_max: int | None = None
275 hard_max: int | None = None
277 def to_dict(self) -> dict[str, Any]:
278 """Convert to dictionary for JSON/YAML serialization."""
279 result: dict[str, Any] = {}
280 if self.normal_max is not None:
281 result["normal_max"] = self.normal_max
282 if self.warn_max is not None:
283 result["warn_max"] = self.warn_max
284 if self.hard_max is not None:
285 result["hard_max"] = self.hard_max
286 return result
288 @classmethod
289 def from_dict(cls, data: dict[str, Any]) -> ThrottleConfig:
290 """Create from dictionary (JSON/YAML deserialization)."""
291 return cls(
292 normal_max=data.get("normal_max"),
293 warn_max=data.get("warn_max"),
294 hard_max=data.get("hard_max"),
295 )
298@dataclass
299class LearningConfig:
300 """Per-state configuration for FEAT-1283 `type: learning` dispatch.
302 Declares the list of external-API/SDK targets a learning state must prove
303 against the learning-tests registry (ENH-1282) before advancing. On a
304 missing or stale record, the state invokes `/ll:explore-api <target>` up to
305 `max_retries` times before transitioning to a blocked target.
307 Attributes:
308 targets: Ordered list of target identifiers (e.g. "Anthropic SDK
309 streaming"). Each is slugified internally when looking up the
310 registry record. All targets must reach status="proven" for the
311 state to advance via on_yes.
312 targets_csv: Runtime-interpolated comma-separated target string (e.g.
313 "${context.targets}"). Resolved and CSV-split by the executor at
314 execution time. Alternative to the static ``targets`` list for loops
315 that receive targets as a context CSV string. Exactly one of
316 ``targets`` (non-empty) or ``targets_csv`` must be set.
317 max_retries: Maximum number of `/ll:explore-api` invocations per target
318 before the state routes to on_blocked / on_no with reason
319 ``retries_exhausted``. Counts only re-exploration attempts; the
320 initial registry lookup is free. Distinct from ENH-1115's throttle
321 counter, which measures tool-call volume; learning states are
322 already exempt from throttle hard_max via FSMExecutor._check_throttle.
323 max_retries_expr: Runtime-interpolated retry limit (e.g.
324 "${context.max_retries}"). Resolved via interpolate() and int()-cast
325 at execution time. Takes precedence over ``max_retries`` when set.
326 """
328 targets: list[str] = field(default_factory=list)
329 targets_csv: str | None = None
330 max_retries: int = 2
331 max_retries_expr: str | None = None
333 def to_dict(self) -> dict[str, Any]:
334 """Convert to dictionary for JSON/YAML serialization."""
335 result: dict[str, Any] = {"targets": list(self.targets), "max_retries": self.max_retries}
336 if self.targets_csv is not None:
337 result["targets_csv"] = self.targets_csv
338 if self.max_retries_expr is not None:
339 result["max_retries_expr"] = self.max_retries_expr
340 return result
342 @classmethod
343 def from_dict(cls, data: dict[str, Any]) -> LearningConfig:
344 """Create from dictionary (JSON/YAML deserialization)."""
345 return cls(
346 targets=list(data.get("targets") or []),
347 targets_csv=data.get("targets_csv") or None,
348 max_retries=int(data.get("max_retries", 2)),
349 max_retries_expr=data.get("max_retries_expr") or None,
350 )
353@dataclass
354class StateConfig:
355 """Configuration for a single FSM state.
357 States can have actions, evaluators, and routing. Supports both
358 shorthand (on_success/on_failure) and full routing table syntax.
360 Attributes:
361 action: Command to execute (shell, slash command, or "server/tool-name" for mcp_tool)
362 action_type: How to execute the action (prompt, slash_command, shell, mcp_tool).
363 If None, uses heuristic: / prefix = slash_command, else = shell.
364 params: MCP tool arguments (only used with action_type: mcp_tool). Supports
365 ${variable} interpolation in string values.
366 evaluate: Evaluator configuration for result interpretation
367 route: Full routing table (verdict -> state mapping)
368 on_yes: Shorthand for yes verdict routing
369 on_no: Shorthand for no verdict routing
370 on_error: Shorthand for error verdict routing
371 on_partial: Shorthand for partial verdict routing
372 next: Unconditional transition (no evaluation)
373 terminal: If True, this is an end state
374 capture: Variable name to store action output
375 timeout: Action-level timeout in seconds
376 on_maintain: State to transition to when maintain=True and loop completes
377 max_retries: Max consecutive re-entries before transitioning to on_retry_exhausted.
378 A value of N allows N retries after the initial execution (N+1 total entries).
379 Requires on_retry_exhausted to also be set.
380 on_retry_exhausted: State to transition to when max_retries consecutive re-entries
381 are exceeded. Required when max_retries is set.
382 max_rate_limit_retries: Max consecutive 429/rate-limit in-place retries for this
383 state before transitioning to on_rate_limit_exhausted. Requires
384 on_rate_limit_exhausted to also be set.
385 on_rate_limit_exhausted: State to transition to when max_rate_limit_retries
386 consecutive rate-limit retries are exceeded. Required when
387 max_rate_limit_retries is set.
388 rate_limit_backoff_base_seconds: Base seconds for exponential backoff between
389 rate-limit retries; actual sleep is base * 2^(attempt-1) + uniform(0, base).
390 Defaults to 30.
391 rate_limit_max_wait_seconds: Total wall-clock budget (seconds) for rate-limit
392 handling in this state before routing to on_rate_limit_exhausted. When unset,
393 defaults from commands.rate_limits.max_wait_seconds (21600 = 6h).
394 rate_limit_long_wait_ladder: Backoff ladder (seconds) for the long-wait tier used
395 once the short-tier retry budget is spent. Each entry is the sleep before the
396 next retry attempt. When unset, defaults from
397 commands.rate_limits.long_wait_ladder ([300, 900, 1800, 3600]).
398 loop: Name of a loop YAML to execute as a sub-FSM. Mutually exclusive with action.
399 context_passthrough: When True, pass parent context variables to child loop and
400 merge child captures back into parent context. Legacy escape hatch; prefer
401 'with_' for explicit named bindings.
402 with_: Explicit parameter bindings for sub-loop calls (YAML key: 'with'). Maps
403 declared child parameter names to parent expressions. Only valid when 'loop'
404 is set. Mutually exclusive with context_passthrough.
405 fragment_name: Original fragment name from the 'fragment:' YAML key. Populated by
406 resolve_fragments; None for non-fragment states. Used for validation and debugging.
407 fragment_bindings: Parameter bindings for fragment references (YAML key 'with' on a
408 fragment state, renamed to 'fragment_bindings' by resolve_fragments). Distinct from
409 with_ (sub-loop bindings) and params (MCP tool args). Only valid when fragment_name
410 is set.
411 fragment_parameters: Parsed ParameterSpec declarations from the fragment definition's
412 'parameters:' block. Populated by resolve_fragments for validation and runtime checks.
413 """
415 action: str | None = None
416 action_type: str | None = None
417 params: dict[str, Any] = field(default_factory=dict)
418 evaluate: EvaluateConfig | None = None
419 route: RouteConfig | None = None
420 on_yes: str | None = None
421 on_no: str | None = None
422 on_error: str | None = None
423 on_partial: str | None = None
424 on_blocked: str | None = None
425 next: str | None = None
426 terminal: bool = False
427 capture: str | None = None
428 append_to_messages: str | None = None
429 timeout: int | None = None
430 on_maintain: str | None = None
431 max_retries: int | None = None
432 on_retry_exhausted: str | None = None
433 retryable_exit_codes: list[int] | None = None
434 max_rate_limit_retries: int | None = None
435 on_rate_limit_exhausted: str | None = None
436 rate_limit_backoff_base_seconds: int | None = None
437 rate_limit_max_wait_seconds: int | None = None
438 rate_limit_long_wait_ladder: list[int] | None = None
439 loop: str | None = None
440 context_passthrough: bool = False
441 with_: dict[str, Any] = field(default_factory=dict)
442 fragment_name: str | None = None
443 fragment_bindings: dict[str, Any] = field(default_factory=dict)
444 fragment_parameters: dict[str, ParameterSpec] = field(default_factory=dict)
445 agent: str | None = None
446 tools: list[str] | None = None
447 extra_routes: dict[str, str] = field(default_factory=dict)
448 type: str | None = None
449 throttle: ThrottleConfig | None = None
450 on_throttle_hard: str | None = None
451 learning: LearningConfig | None = None
453 def to_dict(self) -> dict[str, Any]:
454 """Convert to dictionary for JSON/YAML serialization."""
455 result: dict[str, Any] = {}
457 if self.action is not None:
458 result["action"] = self.action
459 if self.action_type is not None:
460 result["action_type"] = self.action_type
461 if self.params:
462 result["params"] = self.params
463 if self.evaluate is not None:
464 result["evaluate"] = self.evaluate.to_dict()
465 if self.route is not None:
466 result["route"] = self.route.to_dict()
467 if self.on_yes is not None:
468 result["on_yes"] = self.on_yes
469 if self.on_no is not None:
470 result["on_no"] = self.on_no
471 if self.on_error is not None:
472 result["on_error"] = self.on_error
473 if self.on_partial is not None:
474 result["on_partial"] = self.on_partial
475 if self.on_blocked is not None:
476 result["on_blocked"] = self.on_blocked
477 if self.next is not None:
478 result["next"] = self.next
479 if self.terminal:
480 result["terminal"] = self.terminal
481 if self.capture is not None:
482 result["capture"] = self.capture
483 if self.append_to_messages is not None:
484 result["append_to_messages"] = self.append_to_messages
485 if self.timeout is not None:
486 result["timeout"] = self.timeout
487 if self.on_maintain is not None:
488 result["on_maintain"] = self.on_maintain
489 if self.max_retries is not None:
490 result["max_retries"] = self.max_retries
491 if self.on_retry_exhausted is not None:
492 result["on_retry_exhausted"] = self.on_retry_exhausted
493 if self.retryable_exit_codes is not None:
494 result["retryable_exit_codes"] = self.retryable_exit_codes
495 if self.max_rate_limit_retries is not None:
496 result["max_rate_limit_retries"] = self.max_rate_limit_retries
497 if self.on_rate_limit_exhausted is not None:
498 result["on_rate_limit_exhausted"] = self.on_rate_limit_exhausted
499 if self.rate_limit_backoff_base_seconds is not None:
500 result["rate_limit_backoff_base_seconds"] = self.rate_limit_backoff_base_seconds
501 if self.rate_limit_max_wait_seconds is not None:
502 result["rate_limit_max_wait_seconds"] = self.rate_limit_max_wait_seconds
503 if self.rate_limit_long_wait_ladder is not None:
504 result["rate_limit_long_wait_ladder"] = self.rate_limit_long_wait_ladder
505 if self.loop is not None:
506 result["loop"] = self.loop
507 if self.context_passthrough:
508 result["context_passthrough"] = self.context_passthrough
509 if self.with_:
510 result["with"] = self.with_
511 if self.fragment_bindings:
512 result["fragment_bindings"] = self.fragment_bindings
513 if self.agent is not None:
514 result["agent"] = self.agent
515 if self.tools is not None:
516 result["tools"] = self.tools
517 for verdict, target in self.extra_routes.items():
518 result[f"on_{verdict}"] = target
519 if self.type is not None:
520 result["type"] = self.type
521 if self.throttle is not None:
522 result["throttle"] = self.throttle.to_dict()
523 if self.on_throttle_hard is not None:
524 result["on_throttle_hard"] = self.on_throttle_hard
525 if self.learning is not None:
526 result["learning"] = self.learning.to_dict()
528 return result
530 @classmethod
531 def from_dict(cls, data: dict[str, Any]) -> StateConfig:
532 """Create from dictionary (JSON/YAML deserialization)."""
533 evaluate = None
534 if "evaluate" in data:
535 evaluate = EvaluateConfig.from_dict(data["evaluate"])
537 route = None
538 if "route" in data:
539 route = RouteConfig.from_dict(data["route"])
541 throttle = None
542 if "throttle" in data:
543 throttle = ThrottleConfig.from_dict(data["throttle"])
545 learning = None
546 if "learning" in data:
547 learning = LearningConfig.from_dict(data["learning"])
549 _known_on_keys = {
550 "on_yes",
551 "on_success",
552 "on_no",
553 "on_failure",
554 "on_error",
555 "on_partial",
556 "on_blocked",
557 "on_maintain",
558 "on_retry_exhausted",
559 "on_rate_limit_exhausted",
560 "on_throttle_hard",
561 }
562 extra_routes = {
563 key[3:]: val
564 for key, val in data.items()
565 if key.startswith("on_") and key not in _known_on_keys and isinstance(val, str)
566 }
568 return cls(
569 action=data.get("action"),
570 action_type=data.get("action_type"),
571 params=data.get("params", {}),
572 evaluate=evaluate,
573 route=route,
574 on_yes=data.get("on_yes") or data.get("on_success"),
575 on_no=data.get("on_no") or data.get("on_failure"),
576 on_error=data.get("on_error"),
577 on_partial=data.get("on_partial"),
578 on_blocked=data.get("on_blocked"),
579 next=data.get("next"),
580 terminal=data.get("terminal", False),
581 capture=data.get("capture"),
582 append_to_messages=data.get("append_to_messages"),
583 timeout=data.get("timeout"),
584 on_maintain=data.get("on_maintain"),
585 max_retries=data.get("max_retries"),
586 on_retry_exhausted=data.get("on_retry_exhausted"),
587 retryable_exit_codes=data.get("retryable_exit_codes"),
588 max_rate_limit_retries=data.get("max_rate_limit_retries"),
589 on_rate_limit_exhausted=data.get("on_rate_limit_exhausted"),
590 rate_limit_backoff_base_seconds=data.get("rate_limit_backoff_base_seconds"),
591 rate_limit_max_wait_seconds=data.get("rate_limit_max_wait_seconds"),
592 rate_limit_long_wait_ladder=data.get("rate_limit_long_wait_ladder"),
593 loop=data.get("loop"),
594 context_passthrough=data.get("context_passthrough", False),
595 with_=data.get("with", {}),
596 agent=data.get("agent"),
597 tools=data.get("tools"),
598 extra_routes=extra_routes,
599 type=data.get("type"),
600 throttle=throttle,
601 on_throttle_hard=data.get("on_throttle_hard"),
602 learning=learning,
603 fragment_name=data.get("fragment_name"),
604 fragment_bindings=data.get("fragment_bindings", {}),
605 fragment_parameters={
606 name: ParameterSpec.from_dict(spec)
607 for name, spec in data.get("fragment_parameters", {}).items()
608 },
609 )
611 def get_referenced_states(self) -> set[str]:
612 """Get all state names referenced by this state configuration.
614 Returns:
615 Set of state names that this state can transition to.
616 """
617 refs: set[str] = set()
619 if self.on_yes is not None:
620 refs.add(self.on_yes)
621 if self.on_no is not None:
622 refs.add(self.on_no)
623 if self.on_error is not None:
624 refs.add(self.on_error)
625 if self.on_partial is not None:
626 refs.add(self.on_partial)
627 if self.on_blocked is not None:
628 refs.add(self.on_blocked)
629 if self.next is not None:
630 refs.add(self.next)
631 if self.on_maintain is not None:
632 refs.add(self.on_maintain)
633 if self.on_retry_exhausted is not None:
634 refs.add(self.on_retry_exhausted)
635 if self.on_rate_limit_exhausted is not None:
636 refs.add(self.on_rate_limit_exhausted)
637 if self.on_throttle_hard is not None:
638 refs.add(self.on_throttle_hard)
639 if self.route is not None:
640 refs.update(self.route.routes.values())
641 if self.route.default is not None:
642 refs.add(self.route.default)
643 if self.route.error is not None:
644 refs.add(self.route.error)
645 refs.update(self.extra_routes.values())
647 return refs
650@dataclass
651class LLMConfig:
652 """LLM evaluation configuration.
654 Settings for the llm_structured evaluator.
656 Attributes:
657 enabled: If False, disable LLM evaluation entirely
658 model: Model identifier for LLM calls
659 max_tokens: Maximum tokens for evaluation response
660 timeout: Timeout for LLM calls in seconds
661 """
663 enabled: bool = True
664 model: str = DEFAULT_LLM_MODEL
665 max_tokens: int = 256
666 timeout: int = 1800
668 def to_dict(self) -> dict[str, Any]:
669 """Convert to dictionary for JSON/YAML serialization."""
670 result: dict[str, Any] = {}
672 if not self.enabled:
673 result["enabled"] = self.enabled
674 if self.model != DEFAULT_LLM_MODEL:
675 result["model"] = self.model
676 if self.max_tokens != 256:
677 result["max_tokens"] = self.max_tokens
678 if self.timeout != 1800:
679 result["timeout"] = self.timeout
681 return result if result else {}
683 @classmethod
684 def from_dict(cls, data: dict[str, Any]) -> LLMConfig:
685 """Create from dictionary (JSON/YAML deserialization)."""
686 return cls(
687 enabled=data.get("enabled", True),
688 model=data.get("model", DEFAULT_LLM_MODEL),
689 max_tokens=data.get("max_tokens", 256),
690 timeout=data.get("timeout", 1800),
691 )
694@dataclass
695class LoopConfigOverrides:
696 """Per-loop ll-config overrides embedded in the loop YAML definition.
698 All fields are optional (None = use global ll-config default).
699 Precedence: CLI flags > YAML config block > global ll-config > schema defaults.
701 Attributes:
702 handoff_threshold: Override for LL_HANDOFF_THRESHOLD env var (1-100)
703 readiness_threshold: Override for commands.confidence_gate.readiness_threshold (1-100)
704 outcome_threshold: Override for commands.confidence_gate.outcome_threshold (1-100)
705 max_continuations: Override for automation.max_continuations (>=1)
706 """
708 handoff_threshold: int | None = None
709 readiness_threshold: int | None = None
710 outcome_threshold: int | None = None
711 max_continuations: int | None = None
713 def to_dict(self) -> dict[str, Any]:
714 """Convert to dictionary for JSON/YAML serialization (skip-if-None)."""
715 result: dict[str, Any] = {}
717 if self.handoff_threshold is not None:
718 result["handoff_threshold"] = self.handoff_threshold
720 confidence_gate: dict[str, Any] = {}
721 if self.readiness_threshold is not None:
722 confidence_gate["readiness_threshold"] = self.readiness_threshold
723 if self.outcome_threshold is not None:
724 confidence_gate["outcome_threshold"] = self.outcome_threshold
725 if confidence_gate:
726 result["commands"] = {"confidence_gate": confidence_gate}
728 if self.max_continuations is not None:
729 result["automation"] = {"max_continuations": self.max_continuations}
731 return result
733 @classmethod
734 def from_dict(cls, data: dict[str, Any]) -> LoopConfigOverrides:
735 """Create from dictionary (JSON/YAML deserialization)."""
736 commands = data.get("commands", {})
737 confidence_gate = commands.get("confidence_gate", {}) if isinstance(commands, dict) else {}
738 automation = data.get("automation", {})
739 continuation = data.get("continuation", {})
741 max_continuations = None
742 if isinstance(automation, dict) and "max_continuations" in automation:
743 max_continuations = automation["max_continuations"]
744 elif isinstance(continuation, dict) and "max_continuations" in continuation:
745 max_continuations = continuation["max_continuations"]
747 return cls(
748 handoff_threshold=data.get("handoff_threshold"),
749 readiness_threshold=confidence_gate.get("readiness_threshold")
750 if isinstance(confidence_gate, dict)
751 else None,
752 outcome_threshold=confidence_gate.get("outcome_threshold")
753 if isinstance(confidence_gate, dict)
754 else None,
755 max_continuations=max_continuations,
756 )
759@dataclass
760class CommandEntry:
761 """A single command entry for the loop's Commands display section.
763 Attributes:
764 cmd: Full command string to display (e.g., "ll-loop run my-loop --param x=1")
765 comment: Short description shown as a comment (e.g., "run with parameter x")
766 """
768 cmd: str
769 comment: str
772@dataclass
773class TargetStateSpec:
774 """Per-state targeting specification for harness-optimize APO (ENH-1552).
776 Names a single FSM state inside a target loop file and associates it with
777 the examples file and eval fragment used during that state's optimization.
779 Attributes:
780 name: State name within the target loop
781 examples_file: Path to the examples YAML file for this state
782 eval_fragment: Eval fragment identifier used during optimization
783 """
785 name: str
786 examples_file: str
787 eval_fragment: str
789 def to_dict(self) -> dict[str, Any]:
790 """Convert to dictionary for JSON/YAML serialization."""
791 return {
792 "name": self.name,
793 "examples_file": self.examples_file,
794 "eval": self.eval_fragment,
795 }
797 @classmethod
798 def from_dict(cls, data: dict[str, Any]) -> TargetStateSpec:
799 """Create from dictionary (JSON/YAML deserialization)."""
800 return cls(
801 name=data["name"],
802 examples_file=data["examples_file"],
803 eval_fragment=data["eval"],
804 )
807@dataclass
808class TargetFileSpec:
809 """Per-file targeting specification for harness-optimize APO (ENH-1552).
811 Associates a loop YAML file (or glob pattern) with the list of states
812 to optimize within that file.
814 Attributes:
815 file: Explicit path to a loop YAML file (mutually exclusive with glob)
816 glob: Glob pattern matching one or more loop YAML files
817 states: States within the matched file(s) to target
818 """
820 file: str | None = None
821 glob: str | None = None
822 states: list[TargetStateSpec] = field(default_factory=list)
824 def to_dict(self) -> dict[str, Any]:
825 """Convert to dictionary for JSON/YAML serialization."""
826 result: dict[str, Any] = {}
827 if self.file is not None:
828 result["file"] = self.file
829 if self.glob is not None:
830 result["glob"] = self.glob
831 if self.states:
832 result["states"] = [s.to_dict() for s in self.states]
833 return result
835 @classmethod
836 def from_dict(cls, data: dict[str, Any]) -> TargetFileSpec:
837 """Create from dictionary (JSON/YAML deserialization)."""
838 return cls(
839 file=data.get("file"),
840 glob=data.get("glob"),
841 states=[TargetStateSpec.from_dict(s) for s in (data.get("states") or [])],
842 )
845@dataclass
846class RepeatedFailureConfig:
847 """Configuration for the FSM stall detector (FEAT-1637).
849 When N consecutive iterations produce an identical
850 `(state_name, exit_code, eval_verdict)` triple, the FSM either
851 aborts the run or routes to a configured recovery state.
853 Attributes:
854 window: Consecutive iterations with identical triple required to
855 fire (default 3).
856 on_repeated_failure: Either the literal ``"abort"`` (terminate
857 with ``terminated_by="stall_detected"``) or the name of a
858 declared state to route to.
859 """
861 window: int = 3
862 on_repeated_failure: str = "abort"
863 progress_paths: list[str] = field(default_factory=list)
864 exclude_paths: list[str] = field(default_factory=list)
866 def to_dict(self) -> dict[str, Any]:
867 """Convert to dictionary for JSON/YAML serialization (skip-if-default)."""
868 result: dict[str, Any] = {}
869 if self.window != 3:
870 result["window"] = self.window
871 if self.on_repeated_failure != "abort":
872 result["on_repeated_failure"] = self.on_repeated_failure
873 if self.progress_paths:
874 result["progress_paths"] = self.progress_paths
875 if self.exclude_paths:
876 result["exclude_paths"] = self.exclude_paths
877 return result
879 @classmethod
880 def from_dict(cls, data: dict[str, Any]) -> RepeatedFailureConfig:
881 """Create from dictionary (JSON/YAML deserialization)."""
882 return cls(
883 window=data.get("window", 3),
884 on_repeated_failure=data.get("on_repeated_failure", "abort"),
885 progress_paths=data.get("progress_paths", []),
886 exclude_paths=data.get("exclude_paths", []),
887 )
890@dataclass
891class CircuitConfig:
892 """Top-level ``circuit:`` block grouping loop-level safety knobs.
894 Currently exposes ``repeated_failure`` (the stall detector). Future
895 safety knobs (e.g. global timeouts, panic-stop guards) should be
896 added here rather than as separate top-level keys.
898 Attributes:
899 repeated_failure: Stall-detector configuration, or None to disable.
900 """
902 repeated_failure: RepeatedFailureConfig | None = None
904 def to_dict(self) -> dict[str, Any]:
905 """Convert to dictionary for JSON/YAML serialization (skip-if-None)."""
906 result: dict[str, Any] = {}
907 if self.repeated_failure is not None:
908 rf_dict = self.repeated_failure.to_dict()
909 # Always emit the key when configured (even if all fields are defaults)
910 # so the round-trip preserves "repeated_failure was set" intent.
911 result["repeated_failure"] = rf_dict
912 return result
914 @classmethod
915 def from_dict(cls, data: dict[str, Any]) -> CircuitConfig:
916 """Create from dictionary (JSON/YAML deserialization)."""
917 rf = None
918 if "repeated_failure" in data and data["repeated_failure"] is not None:
919 rf = RepeatedFailureConfig.from_dict(data["repeated_failure"])
920 return cls(repeated_failure=rf)
923@dataclass
924class FSMLoop:
925 """Complete FSM loop definition.
927 The main dataclass representing a loop configuration.
929 Attributes:
930 name: Unique loop identifier
931 initial: Starting state name
932 states: Mapping from state name to StateConfig
933 context: User-defined shared variables
934 scope: Paths this loop operates on (for concurrency control). Supports ${context.<var>} template variables that are resolved at runtime.
935 max_iterations: Safety limit for loop iterations
936 backoff: Seconds between iterations
937 timeout: Max total runtime in seconds (loop-level)
938 maintain: If True, restart after completion
939 llm: LLM evaluation configuration
940 on_handoff: Behavior when handoff signal detected (pause/spawn/terminate)
941 commands: Optional override for the Commands section in ll-loop show
942 """
944 name: str
945 initial: str
946 states: dict[str, StateConfig]
947 description: str | None = None
948 context: dict[str, Any] = field(default_factory=dict)
949 parameters: dict[str, ParameterSpec] = field(default_factory=dict)
950 scope: list[str] = field(default_factory=list)
951 max_iterations: int = 50
952 on_max_iterations: str | None = None
953 max_edge_revisits: int = 100
954 backoff: float | None = None
955 timeout: int | None = None
956 default_timeout: int | None = None
957 maintain: bool = False
958 llm: LLMConfig = field(default_factory=LLMConfig)
959 on_handoff: Literal["pause", "spawn", "terminate"] = "pause"
960 input_key: str = "input"
961 config: LoopConfigOverrides | None = None
962 category: str = ""
963 labels: list[str] = field(default_factory=list)
964 required_inputs: list[str] = field(default_factory=list)
965 commands: list[CommandEntry] = field(default_factory=list)
966 targets: list[TargetFileSpec] = field(default_factory=list)
967 circuit: CircuitConfig | None = None
968 meta_self_eval_ok: bool = False
969 shared_state_ok: bool = False
970 partial_route_ok: bool = False
971 artifact_versioning: bool = False
972 artifact_versioning_ok: bool = False
973 # Populated from the raw `import:` list by from_dict(); not serialized by to_dict()
974 imports: list[str] = field(default_factory=list)
976 def to_dict(self) -> dict[str, Any]:
977 """Convert to dictionary for JSON/YAML serialization."""
978 result: dict[str, Any] = {
979 "name": self.name,
980 "initial": self.initial,
981 "states": {name: state.to_dict() for name, state in self.states.items()},
982 }
984 if self.description is not None:
985 result["description"] = self.description
986 if self.context:
987 result["context"] = self.context
988 if self.parameters:
989 result["parameters"] = {name: spec.to_dict() for name, spec in self.parameters.items()}
990 if self.scope:
991 result["scope"] = self.scope
992 if self.max_iterations != 50:
993 result["max_iterations"] = self.max_iterations
994 if self.max_edge_revisits != 100:
995 result["max_edge_revisits"] = self.max_edge_revisits
996 if self.backoff is not None:
997 result["backoff"] = self.backoff
998 if self.timeout is not None:
999 result["timeout"] = self.timeout
1000 if self.default_timeout is not None:
1001 result["default_timeout"] = self.default_timeout
1002 if self.maintain:
1003 result["maintain"] = self.maintain
1004 if self.on_handoff != "pause":
1005 result["on_handoff"] = self.on_handoff
1006 if self.on_max_iterations is not None:
1007 result["on_max_iterations"] = self.on_max_iterations
1008 if self.input_key != "input":
1009 result["input_key"] = self.input_key
1011 llm_dict = self.llm.to_dict()
1012 if llm_dict:
1013 result["llm"] = llm_dict
1015 if self.config is not None:
1016 config_dict = self.config.to_dict()
1017 if config_dict:
1018 result["config"] = config_dict
1020 if self.category:
1021 result["category"] = self.category
1022 if self.labels:
1023 result["labels"] = self.labels
1024 if self.required_inputs:
1025 result["required_inputs"] = self.required_inputs
1026 if self.commands:
1027 result["commands"] = [{"cmd": e.cmd, "comment": e.comment} for e in self.commands]
1028 if self.targets:
1029 result["targets"] = [t.to_dict() for t in self.targets]
1031 if self.circuit is not None:
1032 circuit_dict = self.circuit.to_dict()
1033 if circuit_dict:
1034 result["circuit"] = circuit_dict
1036 if self.meta_self_eval_ok:
1037 result["meta_self_eval_ok"] = self.meta_self_eval_ok
1038 if self.shared_state_ok:
1039 result["shared_state_ok"] = self.shared_state_ok
1040 if self.partial_route_ok:
1041 result["partial_route_ok"] = self.partial_route_ok
1042 if self.artifact_versioning:
1043 result["artifact_versioning"] = self.artifact_versioning
1044 if self.artifact_versioning_ok:
1045 result["artifact_versioning_ok"] = self.artifact_versioning_ok
1047 return result
1049 @classmethod
1050 def from_dict(cls, data: dict[str, Any]) -> FSMLoop:
1051 """Create from dictionary (JSON/YAML deserialization)."""
1052 states = {
1053 name: StateConfig.from_dict(state_data)
1054 for name, state_data in data.get("states", {}).items()
1055 }
1057 llm = LLMConfig()
1058 if "llm" in data:
1059 llm = LLMConfig.from_dict(data["llm"])
1061 loop_config = None
1062 if "config" in data:
1063 loop_config = LoopConfigOverrides.from_dict(data["config"])
1065 circuit = None
1066 if "circuit" in data and data["circuit"] is not None:
1067 circuit = CircuitConfig.from_dict(data["circuit"])
1069 parameters = {
1070 name: ParameterSpec.from_dict(spec) for name, spec in data.get("parameters", {}).items()
1071 }
1073 return cls(
1074 name=data["name"],
1075 initial=data["initial"],
1076 states=states,
1077 description=data.get("description"),
1078 context=data.get("context", {}),
1079 parameters=parameters,
1080 scope=data.get("scope", []),
1081 max_iterations=data.get("max_iterations", 50),
1082 on_max_iterations=data.get("on_max_iterations"),
1083 max_edge_revisits=data.get("max_edge_revisits", 100),
1084 backoff=data.get("backoff"),
1085 timeout=data.get("timeout"),
1086 default_timeout=data.get("default_timeout"),
1087 maintain=data.get("maintain", False),
1088 llm=llm,
1089 on_handoff=data.get("on_handoff", "pause"),
1090 input_key=data.get("input_key", "input"),
1091 config=loop_config,
1092 category=data.get("category", ""),
1093 labels=data.get("labels", []),
1094 required_inputs=data.get("required_inputs", []),
1095 commands=[CommandEntry(**e) for e in data.get("commands", [])],
1096 targets=[TargetFileSpec.from_dict(t) for t in (data.get("targets") or [])],
1097 circuit=circuit,
1098 meta_self_eval_ok=data.get("meta_self_eval_ok", False),
1099 shared_state_ok=data.get("shared_state_ok", False),
1100 partial_route_ok=data.get("partial_route_ok", False),
1101 artifact_versioning=data.get("artifact_versioning", False),
1102 artifact_versioning_ok=data.get("artifact_versioning_ok", False),
1103 imports=data.get("import", []),
1104 )
1106 def get_all_state_names(self) -> set[str]:
1107 """Get all defined state names.
1109 Returns:
1110 Set of all state names in this FSM.
1111 """
1112 return set(self.states.keys())
1114 def get_terminal_states(self) -> set[str]:
1115 """Get all terminal state names.
1117 Returns:
1118 Set of state names where terminal=True.
1119 """
1120 return {name for name, state in self.states.items() if state.terminal}
1122 def get_all_referenced_states(self) -> set[str]:
1123 """Get all state names referenced by transitions.
1125 This includes the initial state and all states referenced
1126 in routing configurations.
1128 Returns:
1129 Set of all referenced state names.
1130 """
1131 refs: set[str] = {self.initial}
1132 for state in self.states.values():
1133 refs.update(state.get_referenced_states())
1134 if self.on_max_iterations is not None:
1135 refs.add(self.on_max_iterations)
1136 return refs