Compile registered edge condition types into callable evaluators.
(self)
| 342 | return self.outputs |
| 343 | |
| 344 | def _prepare_edge_conditions(self) -> None: |
| 345 | """Compile registered edge condition types into callable evaluators.""" |
| 346 | context = ConditionFactoryContext(function_manager=self.function_manager, log_manager=self.log_manager) |
| 347 | processor_context = PayloadProcessorFactoryContext( |
| 348 | function_manager=self.edge_processor_function_manager, |
| 349 | log_manager=self.log_manager, |
| 350 | ) |
| 351 | for node in self.graph.nodes.values(): |
| 352 | for edge_link in node.iter_outgoing_edges(): |
| 353 | condition_config = edge_link.condition_config |
| 354 | if not isinstance(condition_config, EdgeConditionConfig): |
| 355 | raw_value = edge_link.config.get("condition", "true") |
| 356 | condition_config = EdgeConditionConfig.from_dict(raw_value, path=f"{node.path}.edges") |
| 357 | edge_link.condition_config = condition_config |
| 358 | try: |
| 359 | manager = build_edge_condition_manager(condition_config, context, self._get_execution_context()) |
| 360 | except Exception as exc: # pragma: no cover - defensive logging |
| 361 | error_msg = f"Failed to prepare condition '{condition_config.display_label()}': {exc}" |
| 362 | self.log_manager.error(error_msg) |
| 363 | logger = get_server_logger() |
| 364 | logger.log_exception(exc, error_msg, condition_type=condition_config.type) |
| 365 | raise WorkflowExecutionError(error_msg) from exc |
| 366 | edge_link.condition_manager = manager |
| 367 | label = getattr(manager, "label", None) or condition_config.display_label() |
| 368 | metadata = getattr(manager, "metadata", {}) or {} |
| 369 | edge_link.condition = label |
| 370 | edge_link.condition_metadata = metadata |
| 371 | edge_link.condition_type = condition_config.type |
| 372 | |
| 373 | process_config = edge_link.process_config |
| 374 | if process_config: |
| 375 | try: |
| 376 | processor = build_edge_payload_processor(process_config, processor_context) |
| 377 | except Exception as exc: # pragma: no cover |
| 378 | error_msg = ( |
| 379 | f"Failed to prepare processor '{process_config.display_label()}': {exc}" |
| 380 | ) |
| 381 | self.log_manager.error(error_msg) |
| 382 | logger = get_server_logger() |
| 383 | logger.log_exception(exc, error_msg, processor_type=process_config.type) |
| 384 | raise WorkflowExecutionError(error_msg) from exc |
| 385 | edge_link.payload_processor = processor |
| 386 | edge_link.process_type = process_config.type |
| 387 | edge_link.process_metadata = getattr(processor, "metadata", {}) or {} |
| 388 | processor_label = getattr(processor, "label", None) |
| 389 | if processor_label: |
| 390 | edge_link.config["process_label"] = processor_label |
| 391 | else: |
| 392 | edge_link.payload_processor = None |
| 393 | edge_link.process_metadata = {} |
| 394 | edge_link.process_type = None |
| 395 | |
| 396 | def _process_edge_output( |
| 397 | self, |
no test coverage detected