Coverage for src / invariant / executor.py: 94.63%
149 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-10-25 08:17 +0000
1"""Executor: The runtime engine for executing DAGs."""
3from collections.abc import Iterable
4from decimal import Decimal
5from typing import TYPE_CHECKING, Any
7from invariant.cacheable import is_cacheable
8from invariant.expressions import resolve_params
9from invariant.graph import Graph
10from invariant.hashing import hash_manifest
11from invariant.invocation import invoke_op
12from invariant.node import Node, SubGraphNode, SwitchNode
14if TYPE_CHECKING:
15 from invariant.registry import OpRegistry
16 from invariant.store.base import ArtifactStore
19class Executor:
20 """Runtime engine for executing DAGs.
22 Manages the two-phase execution:
23 - Phase 1: Context Resolution (Graph -> Manifest)
24 - Phase 2: Action Execution (Manifest -> Artifact)
25 """
27 def __init__(
28 self,
29 registry: "OpRegistry",
30 store: "ArtifactStore",
31 *,
32 cache: bool = True,
33 ) -> None:
34 """Initialize Executor.
36 Args:
37 registry: OpRegistry for looking up operations.
38 store: ArtifactStore for caching artifacts.
39 cache: Disable manifest hashing and artifact-store access for one-shot
40 execution. Node identity memoisation and validation still apply.
41 """
42 self.registry = registry
43 self.store = store
44 self.cache = cache
46 def execute(
47 self,
48 graph: Graph,
49 outputs: Iterable[str],
50 context: dict[str, Any] | None = None,
51 ) -> dict[str, Any]:
52 """Execute requested graph outputs and return their artifacts.
54 Execution is demand-driven from the requested output roots. Unreachable
55 vertices and inactive SwitchNode branches are not dependency-resolved,
56 registry-validated, cache-checked, or executed.
58 Args:
59 graph: Dictionary mapping node IDs to graph vertices.
60 outputs: Iterable of graph node IDs to produce.
61 context: Optional dictionary of external dependencies (values not in graph).
62 These are injected as artifacts available to any node that declares
63 them in deps.
65 Returns:
66 Dictionary mapping requested output IDs to their resulting artifacts.
68 Raises:
69 ValueError: If requested outputs or active paths are invalid.
70 """
71 output_ids = self._normalize_outputs(outputs, graph)
72 artifacts_by_node, _ = self._execute_requested_outputs(
73 graph,
74 output_ids,
75 context=context,
76 )
77 return {output: artifacts_by_node[output] for output in output_ids}
79 def _execute_requested_outputs(
80 self,
81 graph: Graph,
82 outputs: tuple[str, ...],
83 context: dict[str, Any] | None = None,
84 uncacheable_context_keys: set[str] | None = None,
85 ) -> tuple[dict[str, Any], set[str]]:
86 """Demand-execute the active paths for requested graph outputs."""
87 context = context or {}
88 uncacheable_nodes = set(uncacheable_context_keys or set())
89 artifacts_by_node: dict[str, Any] = {}
90 visiting: list[str] = []
91 visiting_set: set[str] = set()
93 def resolve_artifact(node_id: str) -> Any:
94 if node_id in artifacts_by_node:
95 return artifacts_by_node[node_id]
97 if node_id in graph:
98 if node_id in visiting_set:
99 cycle = " -> ".join([*visiting, node_id])
100 raise ValueError(f"Graph contains cycles on active path: {cycle}")
102 visiting.append(node_id)
103 visiting_set.add(node_id)
104 try:
105 node = graph[node_id]
106 for dep_id in node.deps:
107 resolve_artifact(dep_id)
109 depends_on_uncacheable = any(
110 dep_id in uncacheable_nodes for dep_id in node.deps
111 )
113 if isinstance(node, SwitchNode):
114 self._validate_switch_targets(node, node_id, graph)
115 target_id = self._select_switch_target(
116 node,
117 node_id,
118 artifacts_by_node,
119 )
120 if target_id not in graph:
121 raise ValueError(
122 f"SwitchNode '{node_id}' targets '{target_id}' "
123 "which doesn't exist in graph"
124 )
125 resolve_artifact(target_id)
126 artifacts_by_node[node_id] = artifacts_by_node[target_id]
127 if depends_on_uncacheable or target_id in uncacheable_nodes:
128 uncacheable_nodes.add(node_id)
129 elif isinstance(node, SubGraphNode):
130 manifest = self._build_manifest(
131 node,
132 node_id,
133 graph,
134 artifacts_by_node,
135 )
136 inner_uncacheable_context_keys = (
137 set(manifest.keys()) if depends_on_uncacheable else set()
138 )
139 inner_results, inner_uncacheable_nodes = (
140 self._execute_requested_outputs(
141 node.graph,
142 (node.output,),
143 context=manifest,
144 uncacheable_context_keys=inner_uncacheable_context_keys,
145 )
146 )
147 artifacts_by_node[node_id] = inner_results[node.output]
148 if (
149 depends_on_uncacheable
150 or node.output in inner_uncacheable_nodes
151 ):
152 uncacheable_nodes.add(node_id)
153 else:
154 manifest = self._build_manifest(
155 node,
156 node_id,
157 graph,
158 artifacts_by_node,
159 )
160 artifact = self._execute_node(
161 node,
162 node_id,
163 manifest,
164 depends_on_uncacheable,
165 uncacheable_nodes,
166 )
167 artifacts_by_node[node_id] = artifact
168 finally:
169 visiting.pop()
170 visiting_set.remove(node_id)
172 return artifacts_by_node[node_id]
174 if node_id in context:
175 value = context[node_id]
176 if not is_cacheable(value):
177 raise ValueError(
178 f"Context value for '{node_id}' is not cacheable, "
179 f"got {type(value)}"
180 )
181 artifacts_by_node[node_id] = value
182 return value
184 raise ValueError(
185 f"Node depends on '{node_id}' but it doesn't exist in graph "
186 f"or context. Available: graph={sorted(graph)}, "
187 f"context={sorted(context)}"
188 )
190 for output in outputs:
191 resolve_artifact(output)
192 return artifacts_by_node, uncacheable_nodes
194 def _normalize_outputs(
195 self,
196 outputs: Iterable[str],
197 graph: Graph,
198 ) -> tuple[str, ...]:
199 """Validate and freeze requested output IDs."""
200 if isinstance(outputs, (str, bytes)):
201 raise ValueError(
202 "outputs must be an iterable of node IDs, not str or bytes"
203 )
205 try:
206 output_ids = tuple(outputs)
207 except TypeError as exc:
208 raise ValueError("outputs must be an iterable of node IDs") from exc
210 if not output_ids:
211 raise ValueError("outputs must not be empty")
213 seen: set[str] = set()
214 for output in output_ids:
215 if not isinstance(output, str) or not output:
216 raise ValueError("outputs must contain non-empty strings")
217 if output in seen:
218 raise ValueError(f"outputs contains duplicate node ID '{output}'")
219 seen.add(output)
220 if output not in graph:
221 raise ValueError(
222 f"Output node '{output}' is not in graph. "
223 f"Available: {sorted(graph)}"
224 )
226 return output_ids
228 def _validate_switch_targets(
229 self,
230 node: SwitchNode,
231 node_id: str,
232 graph: Graph,
233 ) -> None:
234 """Validate branch targets for an active SwitchNode."""
235 targets = list(node.cases.values())
236 if node.default is not None:
237 targets.append(node.default)
238 for target in targets:
239 if target not in graph:
240 raise ValueError(
241 f"SwitchNode '{node_id}' targets '{target}' which doesn't "
242 "exist in graph"
243 )
245 def _build_manifest(
246 self,
247 node: Node | SubGraphNode,
248 node_id: str,
249 graph: Graph,
250 artifacts_by_node: dict[str, Any],
251 ) -> dict[str, Any]:
252 """Build the input manifest for a node (Phase 1).
254 The manifest is built entirely from resolved params. Dependencies are NOT
255 injected into the manifest directly - they are only available for ref()/cel()
256 resolution within params.
258 Args:
259 node: The node to build manifest for.
260 node_id: The ID of the node.
261 graph: The full graph (for reference).
262 artifacts_by_node: Already computed artifacts for upstream nodes.
264 Returns:
265 The manifest dictionary mapping parameter names to resolved values.
266 """
267 # Collect dependency artifacts for ref()/cel() resolution
268 dependencies: dict[str, Any] = {}
269 for dep_id in node.deps:
270 if dep_id not in artifacts_by_node:
271 raise ValueError(
272 f"Node '{node_id}' depends on '{dep_id}' but artifact not found. "
273 "This should not happen after active dependency resolution."
274 )
275 dependencies[dep_id] = artifacts_by_node[dep_id]
277 # Manifest = resolved params only. No dependency injection.
278 # ref() and cel() markers in params are resolved using dependencies.
279 return resolve_params(node.params, dependencies)
281 def _resolve_selector(
282 self,
283 node: SwitchNode,
284 node_id: str,
285 artifacts_by_node: dict[str, Any],
286 ) -> Any:
287 """Resolve a SwitchNode selector from its declared deps."""
288 dependencies: dict[str, Any] = {}
289 for dep_id in node.deps:
290 if dep_id not in artifacts_by_node:
291 raise ValueError(
292 f"SwitchNode '{node_id}' depends on '{dep_id}' but artifact "
293 "not found"
294 )
295 dependencies[dep_id] = artifacts_by_node[dep_id]
296 return resolve_params({"selector": node.selector}, dependencies)["selector"]
298 def _select_switch_target(
299 self,
300 node: SwitchNode,
301 node_id: str,
302 artifacts_by_node: dict[str, Any],
303 ) -> str:
304 """Resolve a SwitchNode selector and return the chosen target node ID."""
305 selector_value = self._resolve_selector(node, node_id, artifacts_by_node)
306 case_key = self._normalize_switch_key(selector_value, node_id)
307 if case_key in node.cases:
308 return node.cases[case_key]
309 if node.default is not None:
310 return node.default
311 available = ", ".join(sorted(node.cases))
312 raise ValueError(
313 f"SwitchNode '{node_id}' selector resolved to {case_key!r}, "
314 f"which has no case. Available cases: {available}"
315 )
317 def _normalize_switch_key(self, value: Any, node_id: str) -> str:
318 """Normalize selector results to switch case keys."""
319 if isinstance(value, str):
320 return value
321 if isinstance(value, bool):
322 return "true" if value else "false"
323 if value is None:
324 return "null"
325 if isinstance(value, int):
326 return str(value)
327 if isinstance(value, Decimal):
328 return str(value)
329 raise ValueError(
330 f"SwitchNode '{node_id}' selector returned unsupported "
331 f"{type(value).__name__}; expected str, bool, null, int, or Decimal"
332 )
334 def _execute_node(
335 self,
336 node: Node,
337 node_id: str,
338 manifest: dict[str, Any],
339 depends_on_uncacheable: bool,
340 uncacheable_nodes: set[str],
341 ) -> Any:
342 """Execute one Node using the existing cache semantics."""
343 if not self.registry.has(node.op_name):
344 raise ValueError(
345 f"Node '{node_id}' references unregistered op '{node.op_name}'"
346 )
348 should_cache = self.cache and node.cache and not depends_on_uncacheable
349 if not should_cache:
350 op = self.registry.get(node.op_name)
351 artifact = self._invoke_op(op, node.op_name, manifest)
352 uncacheable_nodes.add(node_id)
353 return artifact
355 digest = hash_manifest(manifest)
356 if self.store.exists(node.op_name, digest):
357 return self.store.get(node.op_name, digest)
359 op = self.registry.get(node.op_name)
360 artifact = self._invoke_op(op, node.op_name, manifest)
361 self.store.put(node.op_name, digest, artifact)
362 return artifact
364 def _invoke_op(self, op: Any, op_name: str, manifest: dict[str, Any]) -> Any:
365 """Invoke an operation with kwargs dispatch and return validation.
367 Args:
368 op: The callable operation to invoke.
369 op_name: The name of the operation (for error messages).
370 manifest: The manifest dictionary mapping parameter names to values.
372 Returns:
373 The operation result (native type or ICacheable domain type).
375 Raises:
376 ValueError: If required parameters are missing.
377 TypeError: If return value is not cacheable.
378 """
379 return invoke_op(op, op_name, manifest)