Coverage for little_loops / fsm / fragments.py: 8%
115 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:27 -0500
1"""Fragment library resolution for FSM loops.
3Implements parse-time expansion of ``fragment:`` references in loop YAML files.
4Fragment libraries define named partial state definitions that any loop can import
5and reference, eliminating copy-paste duplication across loop files.
7Fragment resolution happens before ``FSMLoop.from_dict`` is called, so the engine
8never sees ``fragment:`` keys.
10Example loop YAML::
12 import:
13 - lib/common.yaml
15 states:
16 lint:
17 fragment: shell_exit # inherits action_type + evaluate from fragment
18 action: "ruff check ."
19 on_yes: done
20 on_no: fix
22Example library YAML (``lib/common.yaml``)::
24 fragments:
25 shell_exit:
26 action_type: shell
27 evaluate:
28 type: exit_code
29"""
31from __future__ import annotations
33from pathlib import Path
34from typing import Any
36import yaml
38_BUILTIN_LOOPS_DIR = Path(__file__).parent.parent / "loops"
41def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
42 """Deep-merge two dicts; override keys win at every nesting level.
44 For nested dicts, recursively merges instead of replacing. For all other
45 value types (str, int, bool, list, None), the override value wins outright.
46 Returns a new dict; neither input is mutated.
48 Args:
49 base: Base dict (e.g. a fragment definition).
50 override: Override dict (e.g. state-level fields).
52 Returns:
53 New merged dict with override taking precedence.
54 """
55 result = dict(base)
56 for key, value in override.items():
57 if key in result and isinstance(result[key], dict) and isinstance(value, dict):
58 result[key] = _deep_merge(result[key], value)
59 else:
60 result[key] = value
61 return result
64def resolve_fragments(raw_loop_dict: dict[str, Any], loop_dir: Path) -> dict[str, Any]:
65 """Load fragment libraries, merge namespaces, and expand ``fragment:`` references.
67 Resolution steps:
68 1. Load each path in ``import:`` (relative to ``loop_dir``), collecting all
69 named fragments. Later imports override earlier imports for the same name.
70 2. Merge the loop's own ``fragments:`` block on top (local overrides imports).
71 3. For each state that has a ``fragment:`` key, deep-merge the named fragment
72 into the state dict (state-level keys win), then remove the ``fragment:`` key.
74 Returns a new dict with all ``fragment:`` keys expanded. The ``import:`` and
75 ``fragments:`` keys remain in the returned dict so callers can access them for
76 display or validation purposes.
78 Args:
79 raw_loop_dict: Raw YAML dict loaded from a loop file.
80 loop_dir: Directory containing the loop file; import paths are resolved
81 relative to this directory.
83 Returns:
84 New dict with fragment references expanded.
86 Raises:
87 FileNotFoundError: If an ``import:`` path does not exist relative to ``loop_dir``.
88 ValueError: If a state references a ``fragment:`` name that is not defined in
89 any imported library or the local ``fragments:`` block.
90 """
91 # Step 1: load imported fragment libraries
92 imported_fragments: dict[str, dict[str, Any]] = {}
93 for import_path in raw_loop_dict.get("import", []):
94 lib_path = loop_dir / import_path
95 if not lib_path.exists():
96 builtin_path = _BUILTIN_LOOPS_DIR / import_path
97 if builtin_path.exists():
98 lib_path = builtin_path
99 else:
100 raise FileNotFoundError(
101 f"Fragment library not found: {import_path} "
102 f"(checked '{loop_dir / import_path}' and '{builtin_path}')"
103 )
104 with open(lib_path) as f:
105 lib_data = yaml.safe_load(f)
106 if isinstance(lib_data, dict):
107 for name, frag in lib_data.get("fragments", {}).items():
108 imported_fragments[name] = frag
110 # Step 2: merge local fragments: block on top (local wins)
111 all_fragments: dict[str, dict[str, Any]] = {
112 **imported_fragments,
113 **raw_loop_dict.get("fragments", {}),
114 }
116 result = dict(raw_loop_dict)
117 states: dict[str, Any] = dict(result.get("states", {}))
119 # Step 3: if no states reference a fragment, return early (no-op)
120 if not any(isinstance(s, dict) and s.get("fragment") is not None for s in states.values()):
121 return result
123 for state_name, state_dict in states.items():
124 if not isinstance(state_dict, dict):
125 continue
126 fragment_name = state_dict.get("fragment")
127 if fragment_name is None:
128 continue
129 if fragment_name not in all_fragments:
130 available = ", ".join(sorted(all_fragments))
131 raise ValueError(
132 f"State '{state_name}': fragment '{fragment_name}' not found. "
133 f"Available fragments: {available or '(none)'}"
134 )
135 # Deep merge: fragment is the base, state fields override
136 # Strip description before merge — it is metadata, not a state field
137 frag_copy = dict(all_fragments[fragment_name])
138 frag_copy.pop("description", None)
139 frag_parameters = frag_copy.pop("parameters", {}) # capture before deep merge
140 merged = _deep_merge(frag_copy, state_dict)
141 # Carry fragment metadata for runtime binding and validation
142 fragment_bindings = merged.pop("with", {}) # rename with: → fragment_bindings
143 merged["fragment_name"] = fragment_name # preserve for validation lookup
144 merged["fragment_bindings"] = fragment_bindings # per-state param bindings
145 if frag_parameters:
146 merged["fragment_parameters"] = frag_parameters # fragment's declared params
147 del merged["fragment"] # consume the fragment: key
148 states[state_name] = merged
150 result["states"] = states
151 return result
154def resolve_inheritance(
155 raw_loop_dict: dict[str, Any],
156 loop_dir: Path,
157 _seen: tuple[str, ...] = (),
158) -> dict[str, Any]:
159 """Resolve ``from:`` template inheritance by deep-merging parent into child.
161 A loop YAML with ``from: <name>`` inherits all top-level fields and states
162 from the named parent loop. The child's own fields override the parent's at
163 every nesting level (per :func:`_deep_merge` semantics): scalars and lists
164 are replaced by the child, dicts are merged recursively. The ``from:`` key
165 itself is stripped from the returned dict.
167 Parent lookup uses :func:`little_loops.cli.loop._helpers.resolve_loop_path`,
168 which searches ``loop_dir`` first then falls back to the bundled built-in
169 loops directory. Cycles in the ``from:`` chain raise ``ValueError`` with the
170 full chain path; missing parents raise ``FileNotFoundError``.
172 Resolution must run *before* :func:`resolve_fragments` and *before* the
173 required-fields check in :func:`load_and_validate`, so a child can omit
174 fields its parent provides (including ``initial`` and ``states``) and so a
175 parent's ``import:``/``fragments:`` blocks survive into the merged result.
177 Args:
178 raw_loop_dict: Raw YAML dict loaded from a loop file.
179 loop_dir: Directory containing the (child) loop file; parent names are
180 resolved relative to this directory.
181 _seen: Internal tuple of parent names already visited during recursion;
182 used for cycle detection.
184 Returns:
185 New dict with ``from:`` resolved and stripped. If the input has no
186 ``from:`` key, returns it unchanged.
188 Raises:
189 ValueError: If ``from:`` is not a string, the parent is not a YAML
190 mapping, or a cycle is detected in the inheritance chain.
191 FileNotFoundError: If the parent loop name cannot be resolved.
192 """
193 if "from" not in raw_loop_dict:
194 return raw_loop_dict
196 parent_name = raw_loop_dict["from"]
197 if not isinstance(parent_name, str):
198 raise ValueError(f"`from:` must be a string, got {type(parent_name).__name__}")
200 if parent_name in _seen:
201 chain = " -> ".join(_seen + (parent_name,))
202 raise ValueError(f"Circular `from:` chain: {chain}")
204 # Lazy import to avoid circular import at module load (cli.loop._helpers
205 # imports from fsm.* indirectly). Mirrors the pattern used in
206 # fsm/executor.py:410 for sub-loop calls.
207 from little_loops.cli.loop._helpers import resolve_loop_path
209 parent_path = resolve_loop_path(parent_name, loop_dir)
210 with open(parent_path) as f:
211 parent_data = yaml.safe_load(f)
213 if not isinstance(parent_data, dict):
214 raise ValueError(
215 f"Parent loop '{parent_name}' is not a YAML mapping (got {type(parent_data).__name__})"
216 )
218 parent_data = resolve_inheritance(parent_data, parent_path.parent, _seen + (parent_name,))
220 child_without_from = {k: v for k, v in raw_loop_dict.items() if k != "from"}
221 merged = _deep_merge(parent_data, child_without_from)
222 merged.pop("from", None)
223 return merged
226def resolve_flow(raw_loop_dict: dict[str, Any]) -> dict[str, Any]:
227 """Expand ``flow:`` linear shorthand into a verbose ``states:`` map.
229 A loop YAML with ``flow: [<state>, ...]`` declares an ordered linear chain
230 of states. Each entry is either a bare name (unconditional forward
231 transition) or a ternary ``name?yes_target:no_target`` (conditional
232 branching). The last entry is implicitly ``terminal: true``.
234 Optional ``state_defs:`` supplies prompt/action/evaluate bodies that are
235 deep-merged into the generated state skeletons. If both ``flow:`` and
236 ``states:`` are present, raises ``ValueError`` — the two are mutually
237 exclusive.
239 Resolution runs *after* :func:`resolve_inheritance` (so a child can
240 override a parent's states with its own ``flow:``) and *before* the
241 required-fields check in :func:`load_and_validate` (so the expanded
242 ``states:`` key satisfies the validator).
244 Args:
245 raw_loop_dict: Raw YAML dict loaded from a loop file.
247 Returns:
248 New dict with ``flow:`` expanded to ``states:`` and both ``flow:``
249 and ``state_defs:`` stripped. If the input has no ``flow:`` key,
250 returns it unchanged.
252 Raises:
253 ValueError: If ``flow:`` is not a list, is empty, or contains a
254 malformed ternary entry.
255 """
256 if "flow" not in raw_loop_dict:
257 return raw_loop_dict
259 # If both flow: and states: are present, flow: takes precedence. This
260 # handles the case where states was inherited via `from:` — the child's
261 # explicit flow: overrides the parent's states.
262 #
264 flow = raw_loop_dict["flow"]
265 if not isinstance(flow, list):
266 raise ValueError(f"'flow:' must be a list, got {type(flow).__name__}")
267 if len(flow) < 1:
268 raise ValueError("'flow:' must contain at least one state")
270 state_defs: dict[str, dict[str, Any]] = raw_loop_dict.get("state_defs", {})
271 if not isinstance(state_defs, dict):
272 state_defs = {}
274 generated_states: dict[str, dict[str, Any]] = {}
276 # Pre-parse all entries to extract state names (strip ternary suffixes)
277 def _parse_name(raw: str) -> str:
278 return raw.split("?", 1)[0] if "?" in raw else raw
280 parsed_names = [_parse_name(e) if isinstance(e, str) else e for e in flow]
282 for i, entry in enumerate(flow):
283 is_last = i == len(flow) - 1
284 next_name = parsed_names[i + 1] if not is_last else None
286 if not isinstance(entry, str):
287 raise ValueError(f"'flow:' entry {i} must be a string, got {type(entry).__name__}")
289 if "?" in entry:
290 # Ternary form: name?yes_target:no_target
291 parts = entry.split("?", 1)
292 state_name = parts[0]
293 targets = parts[1]
295 if ":" not in targets or targets.endswith(":") or targets.startswith(":"):
296 raise ValueError(
297 f"Malformed ternary in flow entry '{entry}': must be name?yes_target:no_target"
298 )
299 yes_target, no_target = targets.split(":", 1)
300 if not yes_target or not no_target:
301 raise ValueError(
302 f"Malformed ternary in flow entry '{entry}': "
303 f"both yes and no targets must be non-empty"
304 )
306 skeleton: dict[str, Any] = {"on_yes": yes_target, "on_no": no_target}
307 else:
308 state_name = entry
309 skeleton = {}
310 if next_name is not None:
311 skeleton["next"] = next_name
313 if is_last:
314 skeleton["terminal"] = True
316 # Deep-merge state_defs body into generated skeleton
317 if state_name in state_defs:
318 skeleton = _deep_merge(skeleton, state_defs[state_name])
320 generated_states[state_name] = skeleton
322 result = {k: v for k, v in raw_loop_dict.items() if k not in ("flow", "state_defs")}
323 result["states"] = generated_states
324 return result