Add block labels validation in run_blocks (#4500)

skyvern/exceptions.py

@@ -867,9 +867,12 @@ class NoElementFound(SkyvernException):
         super().__init__("No element found.")
 
 
-class OutputParameterNotFound(SkyvernException):
+class OutputParameterNotFound(SkyvernHTTPException):
     def __init__(self, block_label: str, workflow_permanent_id: str) -> None:
-        super().__init__(f"Output parameter for {block_label} not found in workflow {workflow_permanent_id}")
+        super().__init__(
+            f"Output parameter for {block_label} not found in workflow {workflow_permanent_id}",
+            status_code=status.HTTP_400_BAD_REQUEST,
+        )
 
 
 class AzureBaseError(SkyvernException):

skyvern/forge/sdk/routes/agent_protocol.py

@@ -1476,6 +1476,12 @@ async def run_block(
     # LOG.critical("REMOVING BROWSER SESSION ID")
     # block_run_request.browser_session_id = None
 
+    await block_service.validate_block_labels(
+        workflow_permanent_id=block_run_request.workflow_id,
+        organization_id=organization.organization_id,
+        block_labels=block_run_request.block_labels,
+    )
+
     workflow_run = await block_service.ensure_workflow_run(
         organization=organization,
         template=template,

skyvern/services/block_service.py

@@ -16,6 +16,28 @@ from skyvern.services import workflow_service
 LOG = structlog.get_logger()
 
 
+async def validate_block_labels(
+    workflow_permanent_id: str,
+    organization_id: str,
+    block_labels: list[str],
+) -> None:
+    """
+    Validate that all block labels exist in the workflow.
+    This should be called BEFORE creating a workflow run to prevent orphaned runs.
+    """
+    workflow = await app.DATABASE.get_workflow_by_permanent_id(
+        workflow_permanent_id=workflow_permanent_id,
+        organization_id=organization_id,
+    )
+
+    if not workflow:
+        raise WorkflowNotFound(workflow_permanent_id=workflow_permanent_id)
+
+    for block_label in block_labels:
+        if not workflow.get_output_parameter(block_label):
+            raise OutputParameterNotFound(block_label=block_label, workflow_permanent_id=workflow_permanent_id)
+
+
 async def ensure_workflow_run(
     organization: Organization,
     template: bool,