Fix: backfill entrypoint defaults and skip ENTRYPOINT for trigger-only workflows

- Backfill entrypoint IDs in _generate_workflow_meta_display when overrides omit them
- Skip ENTRYPOINT node when all graph branches are sourced from triggers
- Update tests to match new spec for IntegrationTrigger-only workflows

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

Author: devin-ai-integration[bot]

ee/vellum_ee/workflows/display/tests/workflow_serialization/test_integration_trigger_serialization.py

@@ -187,8 +187,9 @@ class TestWorkflow(BaseWorkflow[BaseInputs, BaseState]):
 
 
 def test_integration_trigger_no_entrypoint_node():
-    """IntegrationTrigger workflows now create ENTRYPOINT nodes and route edges through them."""
+    """IntegrationTrigger-only workflows should NOT have an ENTRYPOINT node when all branches are trigger-sourced."""
 
+    # GIVEN an IntegrationTrigger workflow where all branches are sourced from the trigger
     class SlackMessageTrigger(IntegrationTrigger):
         message: str
 
@@ -203,39 +204,32 @@ class ProcessNode(BaseNode):
     class TestWorkflow(BaseWorkflow[BaseInputs, BaseState]):
         graph = SlackMessageTrigger >> ProcessNode
 
+    # WHEN we serialize the workflow
     result = get_workflow_display(workflow_class=TestWorkflow).serialize()
 
-    # Get trigger ID
+    # THEN the trigger should be serialized
     triggers = result["triggers"]
     assert isinstance(triggers, list)
     assert len(triggers) == 1
     trigger = triggers[0]
     assert isinstance(trigger, dict)
     trigger_id = trigger["id"]
 
-    # Verify ENTRYPOINT node exists
+    # AND there should be NO ENTRYPOINT node (all branches are trigger-sourced)
     workflow_raw_data = result["workflow_raw_data"]
     assert isinstance(workflow_raw_data, dict)
     nodes = workflow_raw_data["nodes"]
     assert isinstance(nodes, list)
     entrypoint_nodes = [n for n in nodes if isinstance(n, dict) and n.get("type") == "ENTRYPOINT"]
-    assert len(entrypoint_nodes) == 1, "IntegrationTrigger workflows should have an ENTRYPOINT node"
-
-    entrypoint_node = entrypoint_nodes[0]
-    assert isinstance(entrypoint_node, dict)
-    entrypoint_node_id = entrypoint_node["id"]
+    assert len(entrypoint_nodes) == 0, "IntegrationTrigger-only workflows should NOT have an ENTRYPOINT node"
 
+    # AND edges should use trigger ID as source_node_id
     edges = workflow_raw_data["edges"]
     assert isinstance(edges, list)
-    entrypoint_edges = [e for e in edges if isinstance(e, dict) and e.get("source_node_id") == entrypoint_node_id]
-    assert len(entrypoint_edges) == 0
-
-    # Verify edges use trigger ID as sourceNodeId (not ENTRYPOINT)
     trigger_edges = [e for e in edges if isinstance(e, dict) and e.get("source_node_id") == trigger_id]
     assert len(trigger_edges) > 0, "Should have edges from trigger ID"
 
-    # Verify the edge connects trigger to first node
-    # ProcessNode should be the only non-terminal, non-entrypoint node
+    # AND the edge should connect trigger to the process node
     process_nodes = [n for n in nodes if isinstance(n, dict) and n.get("type") not in ("TERMINAL", "ENTRYPOINT")]
     assert len(process_nodes) > 0, "Should have at least one process node"
     process_node = process_nodes[0]

ee/vellum_ee/workflows/display/tests/workflow_serialization/test_integration_trigger_with_entrypoint_node_id.py

@@ -16,9 +16,13 @@
 def test_integration_trigger_with_explicit_entrypoint_node_id():
     """
     Tests that a workflow with an IntegrationTrigger and explicit entrypoint_node_id
-    creates an ENTRYPOINT node using the explicit ID (not the trigger's ID).
+    does NOT create an ENTRYPOINT node when all branches are trigger-sourced.
+
+    Even with explicit IDs provided, if all graph branches originate from triggers,
+    the ENTRYPOINT node should be skipped.
     """
 
+    # GIVEN an IntegrationTrigger workflow with explicit entrypoint_node_id
     class SlackMessageTrigger(IntegrationTrigger):
         message: str
 
@@ -53,8 +57,10 @@ class TestWorkflowDisplay(BaseWorkflowDisplay[TestWorkflow]):
             )
         }
 
+    # WHEN we serialize the workflow
     result: dict = get_workflow_display(workflow_class=TestWorkflow).serialize()
 
+    # THEN the trigger should be serialized
     assert "workflow_raw_data" in result
     workflow_raw_data = result["workflow_raw_data"]
     assert isinstance(workflow_raw_data, dict)
@@ -67,22 +73,19 @@ class TestWorkflowDisplay(BaseWorkflowDisplay[TestWorkflow]):
     assert isinstance(trigger, dict)
     trigger_id = trigger["id"]
 
+    # AND there should be NO ENTRYPOINT node (all branches are trigger-sourced)
     nodes = workflow_raw_data["nodes"]
     assert isinstance(nodes, list)
 
     entrypoint_nodes = [n for n in nodes if isinstance(n, dict) and n.get("type") == "ENTRYPOINT"]
 
-    assert (
-        len(entrypoint_nodes) == 1
-    ), "IntegrationTrigger workflows with explicit entrypoint_node_id should create an ENTRYPOINT node"
-
-    entrypoint_node = entrypoint_nodes[0]
-    assert isinstance(entrypoint_node, dict)
-    entrypoint_node_id = entrypoint_node["id"]
-
-    assert entrypoint_node_id == str(explicit_entrypoint_node_id), (
-        f"Entrypoint node ID should be {explicit_entrypoint_node_id} from workflow_display, "
-        f"not {trigger_id} from trigger"
+    assert len(entrypoint_nodes) == 0, (
+        "IntegrationTrigger-only workflows should NOT create an ENTRYPOINT node "
+        "even with explicit entrypoint_node_id, since all branches are trigger-sourced"
     )
 
-    assert entrypoint_node_id != trigger_id, "Entrypoint node should not use the trigger's ID"
+    # AND edges should use trigger ID as source_node_id
+    edges = workflow_raw_data["edges"]
+    assert isinstance(edges, list)
+    trigger_edges = [e for e in edges if isinstance(e, dict) and e.get("source_node_id") == trigger_id]
+    assert len(trigger_edges) > 0, "Should have edges from trigger ID"

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

@@ -250,7 +250,7 @@ def serialize(self) -> JsonObject:
         has_manual_trigger = len(manual_trigger_edges) > 0
 
         # Determine which nodes have explicit non-trigger entrypoints in the graph
-        # This is used later to decide whether to skip entrypoint edges for nodes with triggers
+        # This is used to decide whether to create an ENTRYPOINT node and skip entrypoint edges
         non_trigger_entrypoint_nodes: Set[Type[BaseNode]] = set()
         for subgraph in self._workflow.get_subgraphs():
             if any(True for _ in subgraph.trigger_edges):
@@ -261,6 +261,14 @@ def serialize(self) -> JsonObject:
                 except Exception:
                     continue
 
+        # Determine if we need an ENTRYPOINT node:
+        # - ManualTrigger: always need ENTRYPOINT (backward compatibility)
+        # - No triggers: always need ENTRYPOINT (traditional workflows)
+        # - Non-trigger entrypoints exist: need ENTRYPOINT for those branches
+        # - Only non-manual triggers with no regular entrypoints: skip ENTRYPOINT
+        has_triggers = len(trigger_edges) > 0
+        needs_entrypoint_node = has_manual_trigger or not has_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
@@ -284,9 +292,8 @@ def serialize(self) -> JsonObject:
                 "base": None,
                 "definition": None,
             }
-        else:
-            # Non-manual trigger or no triggers: use workflow_display ENTRYPOINT node if IDs are present
-            # This allows the ENTRYPOINT node to be optional - if IDs are None, no ENTRYPOINT node is created
+        elif needs_entrypoint_node:
+            # No triggers or 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
 
@@ -304,6 +311,7 @@ def serialize(self) -> JsonObject:
                     "base": None,
                     "definition": None,
                 }
+        # else: only non-manual triggers with no regular entrypoints - skip ENTRYPOINT node
 
         # Add all the nodes in the workflows
         for node in self._workflow.get_all_nodes():
@@ -991,16 +999,33 @@ def display_context(self) -> WorkflowDisplayContext:
         )
 
     def _generate_workflow_meta_display(self) -> WorkflowMetaDisplay:
+        defaults = WorkflowMetaDisplay.get_default(self._workflow)
         overrides = self.workflow_display
-        if overrides:
-            return WorkflowMetaDisplay(
-                entrypoint_node_id=overrides.entrypoint_node_id,
-                entrypoint_node_source_handle_id=overrides.entrypoint_node_source_handle_id,
-                entrypoint_node_display=overrides.entrypoint_node_display,
-                display_data=overrides.display_data,
-            )
 
-        return WorkflowMetaDisplay.get_default(self._workflow)
+        if not overrides:
+            return defaults
+
+        # Merge overrides with defaults - if override provides None, fall back to default
+        entrypoint_node_id = (
+            overrides.entrypoint_node_id if overrides.entrypoint_node_id is not None else defaults.entrypoint_node_id
+        )
+        entrypoint_node_source_handle_id = (
+            overrides.entrypoint_node_source_handle_id
+            if overrides.entrypoint_node_source_handle_id is not None
+            else defaults.entrypoint_node_source_handle_id
+        )
+        entrypoint_node_display = (
+            overrides.entrypoint_node_display
+            if overrides.entrypoint_node_display is not None
+            else defaults.entrypoint_node_display
+        )
+
+        return WorkflowMetaDisplay(
+            entrypoint_node_id=entrypoint_node_id,
+            entrypoint_node_source_handle_id=entrypoint_node_source_handle_id,
+            entrypoint_node_display=entrypoint_node_display,
+            display_data=overrides.display_data,
+        )
 
     def _generate_workflow_input_display(
         self, workflow_input: WorkflowInputReference, overrides: Optional[WorkflowInputsDisplay] = None