Coverage for little_loops / fsm / compilers.py: 100%
100 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-03-04 01:17 -0600
1"""Paradigm compilers for FSM loop generation.
3Each paradigm (goal, convergence, invariants, imperative) compiles to the
4universal FSM schema via deterministic template expansion. This provides
5a simple authoring experience while maintaining a single execution engine.
7Compilers transform high-level paradigm YAML specifications into FSMLoop
8instances that can be validated and executed. Each compiler is a pure
9function (~50-100 lines) that performs template expansion with variable
10substitution.
12Example usage:
13 >>> spec = {
14 ... "paradigm": "goal",
15 ... "goal": "No type errors in src/",
16 ... "tools": ["/ll:check-code types", "/ll:manage-issue bug fix"],
17 ... }
18 >>> fsm = compile_paradigm(spec)
19 >>> fsm.initial
20 'evaluate'
21"""
23from __future__ import annotations
25import re
26from typing import Any
28from little_loops.fsm.schema import (
29 EvaluateConfig,
30 FSMLoop,
31 RouteConfig,
32 StateConfig,
33)
36def _slugify(text: str, max_length: int = 50) -> str:
37 """Convert text to slug format for FSM names.
39 Args:
40 text: Text to convert to slug
41 max_length: Maximum length of the resulting slug
43 Returns:
44 Lowercase slug with hyphens, truncated to max_length
45 """
46 text = re.sub(r"[^\w\s-]", "", text)
47 text = re.sub(r"[-\s]+", "-", text)
48 return text.strip("-").lower()[:max_length]
51def _build_evaluate_config(evaluator_spec: dict[str, Any] | None) -> EvaluateConfig | None:
52 """Build EvaluateConfig from evaluator specification dict.
54 Args:
55 evaluator_spec: Optional evaluator configuration dict with keys:
56 - type: Evaluator type (exit_code, output_contains, output_numeric, llm_structured)
57 - pattern: Pattern string for output_contains
58 - operator: Comparison operator for output_numeric (eq, lt, gt, le, ge, ne)
59 - target: Target value for output_numeric
61 Returns:
62 EvaluateConfig instance if spec provided, None otherwise
63 """
64 if evaluator_spec is None:
65 return None
67 eval_type = evaluator_spec.get("type", "exit_code")
69 # For exit_code, we can return None to use the default behavior
70 if eval_type == "exit_code":
71 return None
73 return EvaluateConfig(
74 type=eval_type,
75 pattern=evaluator_spec.get("pattern"),
76 operator=evaluator_spec.get("operator"),
77 target=evaluator_spec.get("target"),
78 )
81def compile_paradigm(spec: dict[str, Any]) -> FSMLoop:
82 """Route to appropriate compiler based on paradigm field.
84 This is the main entry point for paradigm compilation. It examines
85 the 'paradigm' field in the spec and routes to the appropriate
86 compiler function.
88 Args:
89 spec: Paradigm specification dictionary. Must include a 'paradigm'
90 field (defaults to 'fsm' if not present).
92 Returns:
93 Compiled FSMLoop instance
95 Raises:
96 ValueError: If the paradigm is unknown
98 Example:
99 >>> spec = {"paradigm": "goal", "goal": "Clean", "tools": ["cmd"]}
100 >>> fsm = compile_paradigm(spec)
101 >>> fsm.paradigm
102 'goal'
103 """
104 paradigm = spec.get("paradigm", "fsm")
106 compilers = {
107 "goal": compile_goal,
108 "convergence": compile_convergence,
109 "invariants": compile_invariants,
110 "imperative": compile_imperative,
111 "fsm": _passthrough_fsm,
112 }
114 if paradigm not in compilers:
115 raise ValueError(
116 f"Unknown paradigm: '{paradigm}'. Must be one of: {', '.join(sorted(compilers.keys()))}"
117 )
119 return compilers[paradigm](spec)
122def _passthrough_fsm(spec: dict[str, Any]) -> FSMLoop:
123 """Pass through FSM spec directly (no compilation needed).
125 Args:
126 spec: FSM specification dictionary matching FSMLoop schema
128 Returns:
129 FSMLoop instance created from the spec
130 """
131 return FSMLoop.from_dict(spec)
134def compile_goal(spec: dict[str, Any]) -> FSMLoop:
135 """Compile goal paradigm to FSM.
137 Goal paradigm: evaluate → (success → done, failure → fix), fix → evaluate
139 The goal paradigm is the simplest: it repeatedly checks a condition
140 and applies a fix until the goal is achieved.
142 Input spec:
143 paradigm: goal
144 goal: "No type errors in src/"
145 tools:
146 - /ll:check-code types # Check tool (first)
147 - /ll:manage-issue bug fix # Fix tool (second, optional)
148 max_iterations: 50 # Optional, defaults to 50 (examples show 20 for brevity)
149 name: "my-goal" # Optional, auto-generated from goal
150 evaluator: # Optional evaluator config
151 type: output_contains
152 pattern: "Success"
154 Args:
155 spec: Goal paradigm specification dict
157 Returns:
158 Compiled FSMLoop instance with evaluate/fix/done states
160 Raises:
161 ValueError: If required fields are missing
162 """
163 # Validate required fields
164 required = ["goal", "tools"]
165 missing = [f for f in required if f not in spec]
166 if missing:
167 raise ValueError(f"Goal paradigm requires: {', '.join(missing)}")
169 if "tools" in spec and not spec["tools"]:
170 raise ValueError("Goal paradigm 'tools' requires at least one tool")
172 goal = spec["goal"]
173 tools = spec["tools"]
174 check_tool = tools[0]
175 fix_tool = tools[1] if len(tools) > 1 else tools[0]
177 name = spec.get("name", f"goal-{_slugify(goal)}")
179 # Extract evaluator config if provided
180 evaluate_config = _build_evaluate_config(spec.get("evaluator"))
182 states = {
183 "evaluate": StateConfig(
184 action=check_tool,
185 evaluate=evaluate_config,
186 on_success="done",
187 on_failure="fix",
188 on_error="fix",
189 ),
190 "fix": StateConfig(
191 action=fix_tool,
192 next="evaluate",
193 ),
194 "done": StateConfig(terminal=True),
195 }
197 return FSMLoop(
198 name=name,
199 paradigm="goal",
200 initial="evaluate",
201 states=states,
202 max_iterations=spec.get("max_iterations", 50),
203 backoff=spec.get("backoff"),
204 timeout=spec.get("timeout"),
205 on_handoff=spec.get("on_handoff", "pause"),
206 )
209def compile_convergence(spec: dict[str, Any]) -> FSMLoop:
210 """Compile convergence paradigm to FSM.
212 Convergence paradigm: measure → (target → done, progress → apply, stall → done),
213 apply → measure
215 The convergence paradigm drives a metric toward a target value. It measures
216 the current value, compares to the previous measurement, and routes based
217 on whether progress is being made.
219 Input spec:
220 paradigm: convergence
221 name: "reduce-lint-errors"
222 check: "ruff check src/ --output-format=json | jq '.count'"
223 toward: 0
224 using: "/ll:check-code fix"
225 tolerance: 0 # Optional, defaults to 0
227 Args:
228 spec: Convergence paradigm specification dict
230 Returns:
231 Compiled FSMLoop instance with measure/apply/done states
233 Raises:
234 ValueError: If required fields are missing
235 """
236 # Validate required fields
237 required = ["name", "check", "toward", "using"]
238 missing = [f for f in required if f not in spec]
239 if missing:
240 raise ValueError(f"Convergence paradigm requires: {', '.join(missing)}")
242 name = spec["name"]
243 metric_cmd = spec["check"]
244 target = spec["toward"]
245 fix_action = spec["using"]
246 tolerance = spec.get("tolerance", 0)
248 # Build context for variable interpolation
249 context = {
250 "metric_cmd": metric_cmd,
251 "target": target,
252 "tolerance": tolerance,
253 }
255 states = {
256 "measure": StateConfig(
257 action="${context.metric_cmd}",
258 capture="current_value",
259 evaluate=EvaluateConfig(
260 type="convergence",
261 target="${context.target}",
262 tolerance="${context.tolerance}",
263 previous="${prev.output}",
264 ),
265 route=RouteConfig(
266 routes={
267 "target": "done",
268 "progress": "apply",
269 "stall": "done",
270 }
271 ),
272 ),
273 "apply": StateConfig(
274 action=fix_action,
275 next="measure",
276 ),
277 "done": StateConfig(terminal=True),
278 }
280 return FSMLoop(
281 name=name,
282 paradigm="convergence",
283 initial="measure",
284 states=states,
285 context=context,
286 max_iterations=spec.get("max_iterations", 50),
287 backoff=spec.get("backoff"),
288 timeout=spec.get("timeout"),
289 )
292def compile_invariants(spec: dict[str, Any]) -> FSMLoop:
293 """Compile invariants paradigm to FSM.
295 Invariants paradigm: check_1 → (success → check_2, failure → fix_1),
296 fix_1 → check_1, ...
298 The invariants paradigm chains multiple constraints. Each constraint
299 is checked in sequence; if any fails, its fix is applied before
300 re-checking. When all pass, the loop can optionally restart (maintain mode).
302 Input spec:
303 paradigm: invariants
304 name: "code-quality-guardian"
305 constraints:
306 - name: "tests-pass"
307 check: "pytest"
308 fix: "/ll:manage-issue bug fix"
309 evaluator: # Optional per-constraint
310 type: output_contains
311 pattern: "passed"
312 - name: "lint-clean"
313 check: "ruff check src/"
314 fix: "/ll:check-code fix"
315 maintain: true # Optional, restarts after all valid
317 Args:
318 spec: Invariants paradigm specification dict
320 Returns:
321 Compiled FSMLoop instance with check/fix pairs and all_valid terminal
323 Raises:
324 ValueError: If required fields are missing or constraint is invalid
325 """
326 # Validate required fields
327 required = ["name", "constraints"]
328 missing = [f for f in required if f not in spec]
329 if missing:
330 raise ValueError(f"Invariants paradigm requires: {', '.join(missing)}")
332 if "constraints" in spec and not spec["constraints"]:
333 raise ValueError("Invariants paradigm 'constraints' requires at least one constraint")
335 name = spec["name"]
336 constraints = spec["constraints"]
337 maintain = spec.get("maintain", False)
339 # Validate each constraint
340 for i, constraint in enumerate(constraints):
341 if "name" not in constraint:
342 raise ValueError(f"Constraint {i} requires 'name' field")
343 if "check" not in constraint:
344 raise ValueError(f"Constraint '{constraint.get('name', i)}' requires 'check' field")
345 if "fix" not in constraint:
346 raise ValueError(f"Constraint '{constraint.get('name', i)}' requires 'fix' field")
348 states: dict[str, StateConfig] = {}
350 for i, constraint in enumerate(constraints):
351 check_state = f"check_{constraint['name']}"
352 fix_state = f"fix_{constraint['name']}"
354 # Next check state or terminal
355 next_check = (
356 f"check_{constraints[i + 1]['name']}" if i + 1 < len(constraints) else "all_valid"
357 )
359 # Extract evaluator config if provided for this constraint
360 evaluate_config = _build_evaluate_config(constraint.get("evaluator"))
362 states[check_state] = StateConfig(
363 action=constraint["check"],
364 evaluate=evaluate_config,
365 on_success=next_check,
366 on_failure=fix_state,
367 )
368 states[fix_state] = StateConfig(
369 action=constraint["fix"],
370 next=check_state,
371 )
373 # Terminal state with optional maintain loop-back
374 first_check = f"check_{constraints[0]['name']}"
375 states["all_valid"] = StateConfig(
376 terminal=True,
377 on_maintain=first_check if maintain else None,
378 )
380 return FSMLoop(
381 name=name,
382 paradigm="invariants",
383 initial=first_check,
384 states=states,
385 maintain=maintain,
386 max_iterations=spec.get("max_iterations", 50),
387 backoff=spec.get("backoff"),
388 timeout=spec.get("timeout"),
389 )
392def compile_imperative(spec: dict[str, Any]) -> FSMLoop:
393 """Compile imperative paradigm to FSM.
395 Imperative paradigm: step_0 → step_1 → ... → check_done →
396 (success → done, failure → step_0)
398 The imperative paradigm runs a sequence of steps, then checks an
399 exit condition. If the condition fails, the sequence restarts from
400 the beginning.
402 Input spec:
403 paradigm: imperative
404 name: "fix-all-types"
405 steps:
406 - /ll:check-code types
407 - /ll:manage-issue bug fix
408 until:
409 check: "mypy src/"
410 passes: true
411 evaluator: # Optional evaluator for exit condition
412 type: output_contains
413 pattern: "Success"
414 max_iterations: 50 # Optional, defaults to 50 (examples show 20 for brevity)
415 backoff: 2 # Seconds between iterations
417 Args:
418 spec: Imperative paradigm specification dict
420 Returns:
421 Compiled FSMLoop instance with step sequence and check_done
423 Raises:
424 ValueError: If required fields are missing
425 """
426 # Validate required fields
427 required = ["name", "steps", "until"]
428 missing = [f for f in required if f not in spec]
429 if missing:
430 raise ValueError(f"Imperative paradigm requires: {', '.join(missing)}")
432 if "steps" in spec and not spec["steps"]:
433 raise ValueError("Imperative paradigm 'steps' requires at least one step")
435 if "check" not in spec["until"]:
436 raise ValueError("Imperative paradigm 'until' requires 'check' field")
438 name = spec["name"]
439 steps = spec["steps"]
440 until_check = spec["until"]["check"]
442 # Extract evaluator config for exit condition if provided
443 evaluate_config = _build_evaluate_config(spec["until"].get("evaluator"))
445 states: dict[str, StateConfig] = {}
447 # Create step states
448 for i, step in enumerate(steps):
449 state_name = f"step_{i}"
450 next_state = f"step_{i + 1}" if i + 1 < len(steps) else "check_done"
452 states[state_name] = StateConfig(
453 action=step,
454 next=next_state,
455 )
457 # Create check_done state with optional evaluator
458 states["check_done"] = StateConfig(
459 action=until_check,
460 evaluate=evaluate_config,
461 on_success="done",
462 on_failure="step_0",
463 )
465 # Terminal state
466 states["done"] = StateConfig(terminal=True)
468 return FSMLoop(
469 name=name,
470 paradigm="imperative",
471 initial="step_0",
472 states=states,
473 max_iterations=spec.get("max_iterations", 50),
474 backoff=spec.get("backoff"),
475 timeout=spec.get("timeout"),
476 )