[APO-2321] Make entrypoint node optional in Python SDK serialization

Co-Authored-By: [email protected] <[email protected]>

Author: devin-ai-integration[bot]

ee/vellum_ee/workflows/display/base.py

@@ -49,9 +49,9 @@ class WorkflowDisplayData(UniversalBaseModel):
 
 @dataclass
 class WorkflowMetaDisplay:
-    entrypoint_node_id: UUID
-    entrypoint_node_source_handle_id: UUID
-    entrypoint_node_display: NodeDisplayData = Field(default_factory=NodeDisplayData)
+    entrypoint_node_id: Optional[UUID] = None
+    entrypoint_node_source_handle_id: Optional[UUID] = None
+    entrypoint_node_display: Optional[NodeDisplayData] = None
     display_data: WorkflowDisplayData = field(default_factory=WorkflowDisplayData)
 
     @classmethod

ee/vellum_ee/workflows/display/workflows/base_workflow_display.py

@@ -249,8 +249,33 @@ def serialize(self) -> JsonObject:
         manual_trigger_edges = [edge for edge in trigger_edges if issubclass(edge.trigger_class, ManualTrigger)]
         has_manual_trigger = len(manual_trigger_edges) > 0
 
+        # Determine which nodes have explicit non-trigger entrypoints in the graph
+        # We need this early to decide whether to create an ENTRYPOINT node
+        non_trigger_entrypoint_nodes: Set[Type[BaseNode]] = set()
+        for subgraph in self._workflow.get_subgraphs():
+            if any(True for _ in subgraph.trigger_edges):
+                continue
+            for entrypoint in subgraph.entrypoints:
+                try:
+                    non_trigger_entrypoint_nodes.add(get_unadorned_node(entrypoint))
+                except Exception:
+                    continue
+
+        # Check if workflow has only non-manual triggers (IntegrationTrigger/ScheduleTrigger)
+        has_only_non_manual_triggers = len(trigger_edges) > 0 and not has_manual_trigger
+
+        # We need an ENTRYPOINT node if:
+        # 1. There's a ManualTrigger, OR
+        # 2. There are no triggers at all (backward compatibility), OR
+        # 3. There are non-trigger entrypoints (nodes that need ENTRYPOINT edges alongside triggers)
+        # We skip ENTRYPOINT node only when there are ONLY non-manual triggers with no regular entrypoints
+        needs_entrypoint_node = (
+            has_manual_trigger or not has_only_non_manual_triggers or len(non_trigger_entrypoint_nodes) > 0
+        )
+
         entrypoint_node_id: Optional[UUID] = None
         entrypoint_node_source_handle_id: Optional[UUID] = None
+        entrypoint_node_display = self.display_context.workflow_display.entrypoint_node_display
 
         if has_manual_trigger:
             # ManualTrigger: use trigger ID for ENTRYPOINT node (backward compatibility)
@@ -267,28 +292,30 @@ def serialize(self) -> JsonObject:
                     "label": "Entrypoint Node",
                     "source_handle_id": str(entrypoint_node_source_handle_id),
                 },
-                "display_data": self.display_context.workflow_display.entrypoint_node_display.dict(),
+                "display_data": entrypoint_node_display.dict() if entrypoint_node_display else NodeDisplayData().dict(),
                 "base": None,
                 "definition": None,
             }
-        else:
-            # All other cases: use workflow_display ENTRYPOINT node
+        elif needs_entrypoint_node:
+            # Non-trigger entrypoints exist: use workflow_display ENTRYPOINT node
             entrypoint_node_id = self.display_context.workflow_display.entrypoint_node_id
             entrypoint_node_source_handle_id = self.display_context.workflow_display.entrypoint_node_source_handle_id
 
-            serialized_nodes[entrypoint_node_id] = {
-                "id": str(entrypoint_node_id),
-                "type": "ENTRYPOINT",
-                "inputs": [],
-                "data": {
-                    "label": "Entrypoint Node",
-                    "source_handle_id": str(entrypoint_node_source_handle_id),
-                },
-                "display_data": self.display_context.workflow_display.entrypoint_node_display.dict(),
-                "base": None,
-                "definition": None,
-            }
-        # else: has_only_integration_trigger without explicit entrypoint - no ENTRYPOINT node needed
+            if entrypoint_node_id is not None and entrypoint_node_source_handle_id is not None:
+                display_data = entrypoint_node_display.dict() if entrypoint_node_display else NodeDisplayData().dict()
+                serialized_nodes[entrypoint_node_id] = {
+                    "id": str(entrypoint_node_id),
+                    "type": "ENTRYPOINT",
+                    "inputs": [],
+                    "data": {
+                        "label": "Entrypoint Node",
+                        "source_handle_id": str(entrypoint_node_source_handle_id),
+                    },
+                    "display_data": display_data,
+                    "base": None,
+                    "definition": None,
+                }
+        # else: only non-manual triggers with no regular entrypoints - no ENTRYPOINT node needed
 
         # Add all the nodes in the workflows
         for node in self._workflow.get_all_nodes():
@@ -395,19 +422,8 @@ def serialize(self) -> JsonObject:
             except Exception:
                 continue
 
-        # Determine which nodes have explicit non-trigger entrypoints in the graph
-        non_trigger_entrypoint_nodes: Set[Type[BaseNode]] = set()
-        for subgraph in self._workflow.get_subgraphs():
-            # If the subgraph contains trigger edges, its entrypoints were derived from triggers
-            if any(True for _ in subgraph.trigger_edges):
-                continue
-            for entrypoint in subgraph.entrypoints:
-                try:
-                    non_trigger_entrypoint_nodes.add(get_unadorned_node(entrypoint))
-                except Exception:
-                    continue
-
         # Add edges from entrypoint first to preserve expected ordering
+        # Note: non_trigger_entrypoint_nodes was computed earlier to determine if we need an ENTRYPOINT node
 
         for target_node, entrypoint_display in self.display_context.entrypoint_displays.items():
             unadorned_target_node = get_unadorned_node(target_node)
@@ -1048,9 +1064,12 @@ def _generate_entrypoint_display(
         target_node_display = node_displays[entrypoint_target]
         target_node_id = target_node_display.node_id
 
-        edge_display = edge_display_overrides or self._generate_edge_display_from_source(
-            entrypoint_node_id, target_node_id
-        )
+        if edge_display_overrides:
+            edge_display = edge_display_overrides
+        elif entrypoint_node_id is not None:
+            edge_display = self._generate_edge_display_from_source(entrypoint_node_id, target_node_id)
+        else:
+            edge_display = EdgeDisplay(id=uuid4_from_hash(f"{self.workflow_id}|id|{target_node_id}"))
 
         return EntrypointDisplay(id=entrypoint_id, edge_display=edge_display)