Coverage for little_loops / fsm / fragments.py: 30%

1"""Fragment library resolution for FSM loops. 

2 

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. 

6 

7Fragment resolution happens before ``FSMLoop.from_dict`` is called, so the engine 

8never sees ``fragment:`` keys. 

9 

10Example loop YAML:: 

11 

12 import: 

13 - lib/common.yaml 

14 

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 

21 

22Example library YAML (``lib/common.yaml``):: 

23 

24 fragments: 

25 shell_exit: 

26 action_type: shell 

27 evaluate: 

28 type: exit_code 

29""" 

30 

31from __future__ import annotations 

32 

33from pathlib import Path 

34from typing import Any 

35 

36import yaml 

37 

38_BUILTIN_LOOPS_DIR = Path(__file__).parent.parent / "loops" 

39 

40 

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. 

43 

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. 

47 

48 Args: 

49 base: Base dict (e.g. a fragment definition). 

50 override: Override dict (e.g. state-level fields). 

51 

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 

62 

63 

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. 

66 

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. 

73 

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. 

77 

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. 

82 

83 Returns: 

84 New dict with fragment references expanded. 

85 

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 

109 

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 } 

115 

116 result = dict(raw_loop_dict) 

117 states: dict[str, Any] = dict(result.get("states", {})) 

118 

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 

122 

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 merged = _deep_merge(frag_copy, state_dict) 

140 del merged["fragment"] # consume the fragment: key 

141 states[state_name] = merged 

142 

143 result["states"] = states 

144 return result