Get the dynamic configuration for a node from its incoming edges. If multiple incoming edges have dynamic config, they must be identical (same type and parameters). Otherwise raises an error. Returns the dynamic config if found, or None.
(self, node: Node)
| 442 | |
| 443 | |
| 444 | def _get_dynamic_config_for_node(self, node: Node): |
| 445 | """Get the dynamic configuration for a node from its incoming edges. |
| 446 | |
| 447 | If multiple incoming edges have dynamic config, they must be identical |
| 448 | (same type and parameters). Otherwise raises an error. |
| 449 | |
| 450 | Returns the dynamic config if found, or None. |
| 451 | """ |
| 452 | from entity.configs.edge.dynamic_edge_config import DynamicEdgeConfig |
| 453 | |
| 454 | found_configs = [] # List of (source_node_id, dynamic_config) |
| 455 | |
| 456 | for predecessor in node.predecessors: |
| 457 | for edge_link in predecessor.iter_outgoing_edges(): |
| 458 | if edge_link.target is node and edge_link.dynamic_config is not None: |
| 459 | found_configs.append((predecessor.id, edge_link.dynamic_config)) |
| 460 | |
| 461 | if not found_configs: |
| 462 | return None |
| 463 | |
| 464 | if len(found_configs) == 1: |
| 465 | return found_configs[0][1] |
| 466 | |
| 467 | # Multiple dynamic configs found - verify they are consistent |
| 468 | first_source, first_config = found_configs[0] |
| 469 | for source_id, config in found_configs[1:]: |
| 470 | # Check type consistency |
| 471 | if config.type != first_config.type: |
| 472 | raise WorkflowExecutionError( |
| 473 | f"Node '{node.id}' has inconsistent dynamic configurations on incoming edges: " |
| 474 | f"edge from '{first_source}' has type '{first_config.type}', " |
| 475 | f"but edge from '{source_id}' has type '{config.type}'. " |
| 476 | f"All dynamic edges to the same node must use the same configuration." |
| 477 | ) |
| 478 | # Check split config consistency |
| 479 | if (config.split.type != first_config.split.type or |
| 480 | config.split.pattern != first_config.split.pattern or |
| 481 | config.split.json_path != first_config.split.json_path): |
| 482 | raise WorkflowExecutionError( |
| 483 | f"Node '{node.id}' has inconsistent split configurations on incoming edges: " |
| 484 | f"edges from '{first_source}' and '{source_id}' have different split settings. " |
| 485 | f"All dynamic edges to the same node must use the same configuration." |
| 486 | ) |
| 487 | # Check mode-specific config consistency |
| 488 | if config.max_parallel != first_config.max_parallel: |
| 489 | raise WorkflowExecutionError( |
| 490 | f"Node '{node.id}' has inconsistent max_parallel on incoming edges: " |
| 491 | f"edge from '{first_source}' has max_parallel={first_config.max_parallel}, " |
| 492 | f"but edge from '{source_id}' has max_parallel={config.max_parallel}." |
| 493 | ) |
| 494 | if config.type == "tree" and config.group_size != first_config.group_size: |
| 495 | raise WorkflowExecutionError( |
| 496 | f"Node '{node.id}' has inconsistent group_size on incoming edges: " |
| 497 | f"edge from '{first_source}' has group_size={first_config.group_size}, " |
| 498 | f"but edge from '{source_id}' has group_size={config.group_size}." |
| 499 | ) |
| 500 | |
| 501 | return first_config |
no test coverage detected