Recursively validate a list of steps.
(
steps: list[dict[str, Any]],
seen_ids: set[str],
errors: list[str],
)
| 243 | |
| 244 | |
| 245 | def _validate_steps( |
| 246 | steps: list[dict[str, Any]], |
| 247 | seen_ids: set[str], |
| 248 | errors: list[str], |
| 249 | ) -> None: |
| 250 | """Recursively validate a list of steps.""" |
| 251 | from . import STEP_REGISTRY |
| 252 | |
| 253 | for step_config in steps: |
| 254 | if not isinstance(step_config, dict): |
| 255 | errors.append(f"Step must be a mapping, got {type(step_config).__name__}.") |
| 256 | continue |
| 257 | |
| 258 | step_id = step_config.get("id") |
| 259 | if not step_id: |
| 260 | errors.append("Step is missing 'id' field.") |
| 261 | continue |
| 262 | |
| 263 | if ":" in step_id: |
| 264 | errors.append( |
| 265 | f"Step ID {step_id!r} contains ':' which is reserved " |
| 266 | f"for engine-generated nested IDs (parentId:childId)." |
| 267 | ) |
| 268 | |
| 269 | if step_id in seen_ids: |
| 270 | errors.append(f"Duplicate step ID {step_id!r}.") |
| 271 | seen_ids.add(step_id) |
| 272 | |
| 273 | # Determine step type |
| 274 | step_type = step_config.get("type", "command") |
| 275 | if step_type not in _get_valid_step_types(): |
| 276 | errors.append( |
| 277 | f"Step {step_id!r} has invalid type {step_type!r}." |
| 278 | ) |
| 279 | continue |
| 280 | |
| 281 | # Delegate to step-specific validation |
| 282 | step_impl = STEP_REGISTRY.get(step_type) |
| 283 | if step_impl: |
| 284 | step_errors = step_impl.validate(step_config) |
| 285 | errors.extend(step_errors) |
| 286 | |
| 287 | # Validate optional `continue_on_error` field. The engine honours |
| 288 | # this on any step that returns StepStatus.FAILED so the pipeline can route |
| 289 | # around the failure via a downstream `if` or `switch` (or a |
| 290 | # `gate` that surfaces the failure to the operator via message |
| 291 | # interpolation). The field must be a literal boolean — |
| 292 | # coercion from truthy strings is deliberately not supported so |
| 293 | # authoring mistakes surface at validation time rather than |
| 294 | # silently changing run semantics. |
| 295 | if "continue_on_error" in step_config: |
| 296 | coe = step_config["continue_on_error"] |
| 297 | if not isinstance(coe, bool): |
| 298 | errors.append( |
| 299 | f"Step {step_id!r}: 'continue_on_error' must be a " |
| 300 | f"boolean, got {type(coe).__name__}." |
| 301 | ) |
| 302 |
no test coverage detected