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