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

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 

224 # Parse optional fallback suffixes 

225 # ${namespace.path:default=value} → use "value" on missing path 

226 # ${namespace.path?} → use "" on missing path 

227 default_value: str | None = None 

228 nullable = False 

229 

230 # Check for :default= first (so ? inside a default value is literal) 

231 if ":default=" in full_path: 

232 var_part, default_value = full_path.split(":default=", 1) 

233 if var_part.endswith("?"): 

234 raise InterpolationError( 

235 f"Ambiguous suffix: ${{{full_path}}} " 

236 "(?:default=... and ? are mutually exclusive)" 

237 ) 

238 full_path = var_part 

239 elif full_path.endswith("?"): 

240 nullable = True 

241 full_path = full_path[:-1] 

242 

243 if full_path == "messages": 

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

245 namespace, path = "messages", "" 

246 elif "." not in full_path: 

247 raise InterpolationError( 

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

249 ) 

250 else: 

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

252 

253 try: 

254 value = ctx.resolve(namespace, path) 

255 if value is None: 

256 return "" 

257 return str(value) 

258 except InterpolationError: 

259 if default_value is not None: 

260 return default_value 

261 if nullable: 

262 return "" 

263 raise 

264 

265 result = VARIABLE_PATTERN.sub(replace_var, result) 

266 

267 # Restore escaped sequences as literal ${ 

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

269 

270 return result 

271 

272 

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

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

275 

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

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

278 are recursively processed. 

279 

280 Args: 

281 obj: Dictionary to process 

282 ctx: Runtime context for resolution 

283 

284 Returns: 

285 New dictionary with interpolated string values 

286 

287 Raises: 

288 InterpolationError: If any variable resolution fails 

289 """ 

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

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

292 if isinstance(value, str): 

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

294 elif isinstance(value, dict): 

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

296 elif isinstance(value, list): 

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

298 else: 

299 result[key] = value 

300 return result 

301 

302 

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

304 """Interpolate string values in a list. 

305 

306 Args: 

307 items: List to process 

308 ctx: Runtime context for resolution 

309 

310 Returns: 

311 New list with interpolated string values 

312 """ 

313 result: list[Any] = [] 

314 for item in items: 

315 if isinstance(item, str): 

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

317 elif isinstance(item, dict): 

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

319 elif isinstance(item, list): 

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

321 else: 

322 result.append(item) 

323 return result 

324 

325 

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

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

328 

329 Args: 

330 ms: Duration in milliseconds 

331 

332 Returns: 

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

334 """ 

335 if ms < 1000: 

336 return f"{ms}ms" 

337 seconds = ms // 1000 

338 if seconds < 60: 

339 return f"{seconds}s" 

340 minutes = seconds // 60 

341 remaining_seconds = seconds % 60 

342 if remaining_seconds == 0: 

343 return f"{minutes}m" 

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