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

ee/codegen_integration/fixtures/simple_manual_trigger_workflow/display_data/simple_manual_trigger_workflow.json

@@ -154,7 +154,7 @@
     ],
     "edges": [
       {
-        "id": "cc598260-9e77-4774-99f1-b9152d925217",
+        "id": "3f69f314-b59e-462e-8d55-f75bf2f955dc",
         "source_node_id": "b3c8ab56-001f-4157-bbc2-4a7fe5ebf8c6",
         "source_handle_id": "eba8fd73-57ab-4d7b-8f75-b54dbe5fc8ba",
         "target_node_id": "67c2abb9-6c61-47d6-b222-49fd5442ba8f",

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/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/tests/workflow_serialization/test_multi_trigger_same_node_serialization.py

@@ -97,3 +97,114 @@ class TestWorkflow(BaseWorkflow[BaseInputs, BaseState]):
     assert slack_edge is not None, (
         f"Should have edge from Slack trigger ({slack_trigger_id}) " f"to ProcessNode ({process_node_id})"
     )
+
+
+def test_two_integration_triggers_same_node_unique_edge_ids():
+    """
+    Tests that when two IntegrationTriggers point to the same node,
+    each trigger edge gets a unique ID (not duplicates).
+    """
+
+    # GIVEN two different IntegrationTriggers
+    class SlackMessageTrigger(IntegrationTrigger):
+        message: str
+
+        class Config(IntegrationTrigger.Config):
+            provider = VellumIntegrationProviderType.COMPOSIO
+            integration_name = "SLACK"
+            slug = "slack_new_message"
+
+    class GithubPRTrigger(IntegrationTrigger):
+        pr_title: str
+
+        class Config(IntegrationTrigger.Config):
+            provider = VellumIntegrationProviderType.COMPOSIO
+            integration_name = "GITHUB"
+            slug = "github_pr_opened"
+
+    class ProcessNode(BaseNode):
+        pass
+
+    # AND a workflow where both triggers point to the same node
+    class TestWorkflow(BaseWorkflow[BaseInputs, BaseState]):
+        graph = {
+            SlackMessageTrigger >> ProcessNode,
+            GithubPRTrigger >> ProcessNode,
+        }
+
+    # WHEN we serialize the workflow
+    result = get_workflow_display(workflow_class=TestWorkflow).serialize()
+
+    # THEN we should have two triggers
+    triggers = result["triggers"]
+    assert isinstance(triggers, list)
+    assert len(triggers) == 2
+
+    def get_integration_name(trigger: dict) -> str:
+        exec_config = trigger.get("exec_config")
+        if isinstance(exec_config, dict):
+            name = exec_config.get("integration_name")
+            return str(name) if name else ""
+        return ""
+
+    slack_trigger = next(
+        (t for t in triggers if isinstance(t, dict) and get_integration_name(t) == "SLACK"),
+        None,
+    )
+    assert slack_trigger is not None
+    assert isinstance(slack_trigger, dict)
+    slack_trigger_id = slack_trigger["id"]
+
+    github_trigger = next(
+        (t for t in triggers if isinstance(t, dict) and get_integration_name(t) == "GITHUB"),
+        None,
+    )
+    assert github_trigger is not None
+    assert isinstance(github_trigger, dict)
+    github_trigger_id = github_trigger["id"]
+
+    # AND we should have edges from both triggers
+    workflow_raw_data = result["workflow_raw_data"]
+    assert isinstance(workflow_raw_data, dict)
+    edges = workflow_raw_data["edges"]
+    assert isinstance(edges, list)
+
+    nodes = workflow_raw_data["nodes"]
+    assert isinstance(nodes, list)
+    process_nodes = [n for n in nodes if isinstance(n, dict) and n.get("type") not in ("TERMINAL", "ENTRYPOINT")]
+    assert len(process_nodes) > 0
+    process_node = process_nodes[0]
+    assert isinstance(process_node, dict)
+    process_node_id = process_node["id"]
+
+    slack_edge = next(
+        (
+            e
+            for e in edges
+            if isinstance(e, dict)
+            and e.get("source_node_id") == slack_trigger_id
+            and e.get("target_node_id") == process_node_id
+        ),
+        None,
+    )
+    assert slack_edge is not None, "Should have edge from Slack trigger to ProcessNode"
+    assert isinstance(slack_edge, dict)
+
+    github_edge = next(
+        (
+            e
+            for e in edges
+            if isinstance(e, dict)
+            and e.get("source_node_id") == github_trigger_id
+            and e.get("target_node_id") == process_node_id
+        ),
+        None,
+    )
+    assert github_edge is not None, "Should have edge from GitHub trigger to ProcessNode"
+    assert isinstance(github_edge, dict)
+
+    # AND the edge IDs should be unique (not duplicates)
+    assert slack_edge["id"] != github_edge["id"], (
+        f"Edge IDs should be unique but both are {slack_edge['id']}. "
+        "Multiple triggers targeting the same node should have distinct edge IDs."
+    )

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

@@ -0,0 +1,55 @@
+"""Tests for partial WorkflowMetaDisplay override serialization."""
+
+from vellum.workflows import BaseWorkflow
+from vellum.workflows.inputs.base import BaseInputs
+from vellum.workflows.nodes.bases.base import BaseNode
+from vellum.workflows.state.base import BaseState
+from vellum_ee.workflows.display.base import WorkflowDisplayData, WorkflowDisplayDataViewport, WorkflowMetaDisplay
+from vellum_ee.workflows.display.workflows.base_workflow_display import BaseWorkflowDisplay
+from vellum_ee.workflows.display.workflows.get_vellum_workflow_display_class import get_workflow_display
+
+
+def test_triggerless_workflow_with_partial_display_override_creates_entrypoint():
+    """
+    Tests that a triggerless workflow with partial WorkflowMetaDisplay overrides
+    (only display_data set) still creates an ENTRYPOINT node with backfilled IDs.
+    """
+
+    # GIVEN a simple triggerless workflow
+    class ProcessNode(BaseNode):
+        pass
+
+    class TestWorkflow(BaseWorkflow[BaseInputs, BaseState]):
+        graph = ProcessNode
+
+    # AND a display class that only overrides display_data (viewport), not entrypoint IDs
+    class TestWorkflowDisplay(BaseWorkflowDisplay[TestWorkflow]):
+        workflow_display = WorkflowMetaDisplay(
+            display_data=WorkflowDisplayData(viewport=WorkflowDisplayDataViewport(x=100.0, y=200.0, zoom=1.5))
+        )
+
+    # WHEN we serialize the workflow
+    result = get_workflow_display(workflow_class=TestWorkflow).serialize()
+
+    # THEN the workflow should have an ENTRYPOINT node
+    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, "Triggerless workflow with partial overrides should have an ENTRYPOINT node"
+
+    # AND the ENTRYPOINT node should have valid IDs (not None)
+    entrypoint_node = entrypoint_nodes[0]
+    assert isinstance(entrypoint_node, dict)
+    assert entrypoint_node["id"] is not None, "ENTRYPOINT node should have a non-None id"
+    entrypoint_data = entrypoint_node["data"]
+    assert isinstance(entrypoint_data, dict)
+    source_handle_id = entrypoint_data["source_handle_id"]
+    assert source_handle_id is not None, "ENTRYPOINT node should have a non-None source_handle_id"
+
+    # AND there should be an edge from ENTRYPOINT to the process node
+    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, "Should have edges from ENTRYPOINT node"