Method _determine_start_nodes

workflow/graph_manager.py:273–319

Determine the effective set of start nodes (explicit only).

(self)

        }

    def _determine_start_nodes(self) -> None:
        """Determine the effective set of start nodes (explicit only)."""
        definition = self.graph.config.definition
        explicit_ordered = list(definition.start_nodes)
        explicit_set = set(explicit_ordered)

        # if explicit_ordered and not self.graph.has_cycles:
        #     raise ConfigError(
        #         "start nodes can only be specified for graphs that contain cycles",
        #         extend_path(definition.path, "start"),
        #     )

        if explicit_set:
            cycle_path = extend_path(definition.path, "start")
            for node_id in explicit_ordered:
                if node_id not in self.graph.nodes:
                    raise ConfigError(
                        f"start node '{node_id}' not defined in nodes",
                        cycle_path,
                    )

                cycle_id = self.cycle_manager.node_to_cycle.get(node_id)
                if cycle_id is None:
                    continue
                cycle_info = self.cycle_manager.cycles.get(cycle_id)
                if cycle_info is None:
                    raise ConfigError(
                        f"cycle data missing for start node '{node_id}'",
                        cycle_path,
                    )

                if cycle_info.configured_entry_node and cycle_info.configured_entry_node != node_id:
                    raise ConfigError(
                        f"cycle '{cycle_id}' already has start node '{cycle_info.configured_entry_node}'",
                        cycle_path,
                    )

                cycle_info.configured_entry_node = node_id

        if not explicit_ordered:
            raise ConfigError(
                "Unable to determine a start node for this graph. Configure at least one Start Node via Configure Graph > Advanced Settings > Start Node > input node ID.",
                extend_path(definition.path, "start"),
            )

        self.graph.start_nodes = explicit_ordered
        self.graph.explicit_start_nodes = explicit_ordered

    def _warn_on_untriggerable_nodes(self) -> None:
        """Emit warnings for nodes that cannot be triggered by any predecessor."""