Coverage for little_loops / fsm / interpolation.py: 24%

1"""Variable interpolation for FSM loop definitions. 

2 

3This module provides runtime variable substitution using ${namespace.path} 

4syntax. Variables are resolved against an InterpolationContext that holds 

5runtime state including user context, captured values, and metadata. 

6 

7Supported namespaces: 

8 context: User-defined variables from FSM context block 

9 captured: Values stored via capture: in previous states 

10 prev: Previous state's result (shorthand) 

11 result: Current evaluation result 

12 state: Current state metadata (name, iteration) 

13 loop: Loop-level metadata (name, started_at, elapsed_ms, elapsed) 

14 env: Environment variables 

15 messages: Shared append-only message log (${messages}, ${messages.last(N)}, ${messages.summary}) 

16 param: Per-state parameter bindings for fragment references 

17""" 

18 

19from __future__ import annotations 

20 

21import os 

22import re 

23from dataclasses import dataclass, field 

24from typing import Any 

25 

26# Pre-compiled patterns for performance 

27VARIABLE_PATTERN = re.compile(r"\$\{([^}]+)\}") 

28ESCAPED_PATTERN = re.compile(r"\$\$\{") 

29ESCAPED_PLACEHOLDER = "\x00ESCAPED\x00" 

30 

31 

32class InterpolationError(Exception): 

33 """Raised when variable interpolation fails.""" 

34 

35 pass 

36 

37 

38@dataclass 

39class InterpolationContext: 

40 """Runtime context for variable resolution. 

41 

42 Holds all namespace data needed to resolve ${namespace.path} variables 

43 during FSM execution. 

44 

45 Attributes: 

46 context: User-defined variables from FSM context block 

47 captured: Stored action results {varname: {output, stderr, exit_code, duration_ms}} 

48 prev: Previous state result or None if first state 

49 result: Current evaluation result or None 

50 state_name: Current state name 

51 iteration: Current loop iteration (1-based) 

52 loop_name: FSM loop name 

53 started_at: ISO timestamp when loop started 

54 elapsed_ms: Milliseconds since loop started 

55 param: Per-state parameter bindings for fragment references (resolved from fragment_bindings) 

56 """ 

57 

58 context: dict[str, Any] = field(default_factory=dict) 

59 captured: dict[str, dict[str, Any]] = field(default_factory=dict) 

60 prev: dict[str, Any] | None = None 

61 result: dict[str, Any] | None = None 

62 state_name: str = "" 

63 iteration: int = 1 

64 loop_name: str = "" 

65 started_at: str = "" 

66 elapsed_ms: int = 0 

67 messages: list[str] = field(default_factory=list) 

68 messages_summary: str = "" 

69 param: dict[str, Any] = field(default_factory=dict) 

70 

71 def resolve(self, namespace: str, path: str) -> Any: 

72 """Resolve a namespace.path reference to its value. 

73 

74 Args: 

75 namespace: The namespace identifier (context, captured, etc.) 

76 path: The dot-separated path within the namespace 

77 

78 Returns: 

79 The resolved value 

80 

81 Raises: 

82 InterpolationError: If namespace unknown or path not found 

83 """ 

84 if namespace == "context": 

85 return self._get_nested(self.context, path, "context") 

86 elif namespace == "captured": 

87 return self._get_nested(self.captured, path, "captured") 

88 elif namespace == "prev": 

89 if self.prev is None: 

90 raise InterpolationError("No previous state result available") 

91 return self._get_nested(self.prev, path, "prev") 

92 elif namespace == "result": 

93 if self.result is None: 

94 raise InterpolationError("No evaluation result available") 

95 return self._get_nested(self.result, path, "result") 

96 elif namespace == "state": 

97 return self._get_state_value(path) 

98 elif namespace == "loop": 

99 return self._get_loop_value(path) 

100 elif namespace == "env": 

101 value = os.environ.get(path) 

102 if value is None: 

103 raise InterpolationError(f"Environment variable '{path}' not set") 

104 return value 

105 elif namespace == "messages": 

106 return self._get_messages_value(path) 

107 elif namespace == "param": 

108 return self._get_nested(self.param, path, "param") 

109 else: 

110 raise InterpolationError(f"Unknown namespace: {namespace}") 

111 

112 def _get_nested(self, obj: dict[str, Any], path: str, namespace: str) -> Any: 

113 """Get nested value from dict using dot notation. 

114 

115 Args: 

116 obj: Dictionary to traverse 

117 path: Dot-separated path (e.g., "errors.output") 

118 namespace: Namespace name for error messages 

119 

120 Returns: 

121 The value at the path 

122 

123 Raises: 

124 InterpolationError: If path not found 

125 """ 

126 parts = path.split(".") 

127 current: Any = obj 

128 for i, part in enumerate(parts): 

129 if isinstance(current, dict) and part in current: 

130 current = current[part] 

131 else: 

132 traversed = ".".join(parts[: i + 1]) 

133 raise InterpolationError(f"Path '{traversed}' not found in {namespace}") 

134 return current 

135 

136 def _get_state_value(self, key: str) -> Any: 

137 """Get state metadata value. 

138 

139 Args: 

140 key: State property name (name or iteration) 

141 

142 Returns: 

143 The state property value 

144 

145 Raises: 

146 InterpolationError: If key unknown 

147 """ 

148 if key == "name": 

149 return self.state_name 

150 elif key == "iteration": 

151 return self.iteration 

152 else: 

153 raise InterpolationError(f"Unknown state property: {key}") 

154 

155 def _get_messages_value(self, path: str) -> str: 

156 """Get value from the shared messages log. 

157 

158 Args: 

159 path: Empty string or "output" for full log; "last(N)" for last N entries; 

160 "summary" for the pre-computed summary string. 

161 

162 Returns: 

163 The resolved messages string 

164 

165 Raises: 

166 InterpolationError: If path is unrecognised 

167 """ 

168 if not path or path == "output": 

169 return "\n".join(self.messages) 

170 m = re.match(r"^last\((\d+)\)$", path) 

171 if m: 

172 n = int(m.group(1)) 

173 return "\n".join(self.messages[-n:]) 

174 if path == "summary": 

175 return self.messages_summary 

176 raise InterpolationError(f"Unknown messages property: {path!r}") 

177 

178 def _get_loop_value(self, key: str) -> Any: 

179 """Get loop metadata value. 

180 

181 Args: 

182 key: Loop property name 

183 

184 Returns: 

185 The loop property value 

186 

187 Raises: 

188 InterpolationError: If key unknown 

189 """ 

190 if key == "name": 

191 return self.loop_name 

192 elif key == "started_at": 

193 return self.started_at 

194 elif key == "elapsed_ms": 

195 return self.elapsed_ms 

196 elif key == "elapsed": 

197 return _format_duration(self.elapsed_ms) 

198 else: 

199 raise InterpolationError(f"Unknown loop property: {key}") 

200 

201 

202def interpolate(template: str, ctx: InterpolationContext) -> str: 

203 """Replace ${namespace.path} variables in template string. 

204 

205 Resolves variables at runtime against the provided context. 

206 Handles $${...} escaping (becomes literal ${...}). 

207 

208 Args: 

209 template: String containing variable references 

210 ctx: Runtime context for resolution 

211 

212 Returns: 

213 String with all variables resolved 

214 

215 Raises: 

216 InterpolationError: If variable format invalid or value not found 

217 """ 

218 # Replace escaped sequences with placeholder 

219 result = ESCAPED_PATTERN.sub(ESCAPED_PLACEHOLDER, template) 

220 

221 def replace_var(match: re.Match[str]) -> str: 

222 full_path = match.group(1) 

223 if full_path == "messages": 

224 # Bare ${messages} is shorthand for the full message log 

225 namespace, path = "messages", "" 

226 elif "." not in full_path: 

227 raise InterpolationError( 

228 f"Invalid variable: ${{{full_path}}} (expected namespace.path)" 

229 ) 

230 else: 

231 namespace, path = full_path.split(".", 1) 

232 value = ctx.resolve(namespace, path) 

233 # Convert to string, handling empty values 

234 if value is None: 

235 return "" 

236 return str(value) 

237 

238 result = VARIABLE_PATTERN.sub(replace_var, result) 

239 

240 # Restore escaped sequences as literal ${ 

241 result = result.replace(ESCAPED_PLACEHOLDER, "${") 

242 

243 return result 

244 

245 

246def interpolate_dict(obj: dict[str, Any], ctx: InterpolationContext) -> dict[str, Any]: 

247 """Recursively interpolate all string values in a dict. 

248 

249 Only string values are interpolated. Non-string values (int, float, 

250 bool, None) are passed through unchanged. Nested dicts and lists 

251 are recursively processed. 

252 

253 Args: 

254 obj: Dictionary to process 

255 ctx: Runtime context for resolution 

256 

257 Returns: 

258 New dictionary with interpolated string values 

259 

260 Raises: 

261 InterpolationError: If any variable resolution fails 

262 """ 

263 result: dict[str, Any] = {} 

264 for key, value in obj.items(): 

265 if isinstance(value, str): 

266 result[key] = interpolate(value, ctx) 

267 elif isinstance(value, dict): 

268 result[key] = interpolate_dict(value, ctx) 

269 elif isinstance(value, list): 

270 result[key] = _interpolate_list(value, ctx) 

271 else: 

272 result[key] = value 

273 return result 

274 

275 

276def _interpolate_list(items: list[Any], ctx: InterpolationContext) -> list[Any]: 

277 """Interpolate string values in a list. 

278 

279 Args: 

280 items: List to process 

281 ctx: Runtime context for resolution 

282 

283 Returns: 

284 New list with interpolated string values 

285 """ 

286 result: list[Any] = [] 

287 for item in items: 

288 if isinstance(item, str): 

289 result.append(interpolate(item, ctx)) 

290 elif isinstance(item, dict): 

291 result.append(interpolate_dict(item, ctx)) 

292 elif isinstance(item, list): 

293 result.append(_interpolate_list(item, ctx)) 

294 else: 

295 result.append(item) 

296 return result 

297 

298 

299def _format_duration(ms: int) -> str: 

300 """Format milliseconds as human-readable duration. 

301 

302 Args: 

303 ms: Duration in milliseconds 

304 

305 Returns: 

306 Formatted string like "500ms", "30s", or "2m 15s" 

307 """ 

308 if ms < 1000: 

309 return f"{ms}ms" 

310 seconds = ms // 1000 

311 if seconds < 60: 

312 return f"{seconds}s" 

313 minutes = seconds // 60 

314 remaining_seconds = seconds % 60 

315 if remaining_seconds == 0: 

316 return f"{minutes}m" 

317 return f"{minutes}m {remaining_seconds}s"