Validate workflow status filter to prevent 500 on invalid enum value

The /workflow/fetch and /workflow/summary endpoints accepted a free-form
status query param and passed it straight into a query that casts to the
workflow_status PG enum (active/archived). Any other value — e.g. an
external caller passing 'published' (a workflow_definitions version state,
not a workflow status) — failed deep in Postgres as
InvalidTextRepresentationError, surfacing as an unhandled HTTP 500.

Add _validate_status_filter() to reject values outside WorkflowStatus with
a clean 422 before any DB query, for both the single and comma-separated
paths. Add route tests covering invalid, valid-single, comma-separated, and
mixed valid/invalid cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

api/routes/workflow.py

@@ -15,7 +15,7 @@ from api.db import db_client
 from api.db.agent_trigger_client import TriggerPathConflictError
 from api.db.models import UserModel
 from api.db.workflow_template_client import WorkflowTemplateClient
-from api.enums import CallType, PostHogEvent, StorageBackend
+from api.enums import CallType, PostHogEvent, StorageBackend, WorkflowStatus
 from api.schemas.ai_model_configuration import OrganizationAIModelConfigurationV2
 from api.schemas.workflow import WorkflowRunResponseSchema
 from api.sdk_expose import sdk_expose
@@ -578,6 +578,30 @@ async def get_workflow_count(
     )
 
 
+def _validate_status_filter(status: Optional[str]) -> List[str]:
+    """Parse and validate a workflow ``status`` query filter.
+
+    Accepts a single value or a comma-separated list. Returns the list of
+    validated status values (empty when no filter was supplied). Any value
+    outside the ``workflow_status`` enum raises 422 so the request fails as a
+    clean client error instead of a 500 from the Postgres enum cast.
+    """
+    if not status:
+        return []
+    allowed = {s.value for s in WorkflowStatus}
+    requested = [s.strip() for s in status.split(",") if s.strip()]
+    invalid = sorted({s for s in requested if s not in allowed})
+    if invalid:
+        raise HTTPException(
+            status_code=422,
+            detail=(
+                f"Invalid workflow status filter: {invalid}. "
+                f"Allowed values: {sorted(allowed)}."
+            ),
+        )
+    return requested
+
+
 @router.get(
     "/fetch",
     **sdk_expose(
@@ -597,21 +621,22 @@ async def get_workflows(
     Returns a lightweight response with only essential fields for listing.
     Use GET /workflow/fetch/{workflow_id} to get full workflow details.
     """
-    # Handle comma-separated status values
-    if status and "," in status:
-        # Split comma-separated values and fetch workflows for each status
-        status_list = [s.strip() for s in status.split(",")]
+    statuses = _validate_status_filter(status)
+    if statuses:
+        # Fetch workflows for each requested status and combine the results.
         all_workflows = []
-        for status_value in status_list:
-            workflows = await db_client.get_all_workflows_for_listing(
-                organization_id=user.selected_organization_id, status=status_value
+        for status_value in statuses:
+            all_workflows.extend(
+                await db_client.get_all_workflows_for_listing(
+                    organization_id=user.selected_organization_id,
+                    status=status_value,
+                )
             )
-            all_workflows.extend(workflows)
         workflows = all_workflows
     else:
-        # Single status or no status filter
+        # No status filter
         workflows = await db_client.get_all_workflows_for_listing(
-            organization_id=user.selected_organization_id, status=status
+            organization_id=user.selected_organization_id, status=None
         )
 
     # Get run counts for all workflows in a single query
@@ -820,10 +845,20 @@ async def get_workflows_summary(
     ),
 ) -> List[WorkflowSummaryResponse]:
     """Get minimal workflow information (id and name only) for all workflows"""
-    workflows = await db_client.get_all_workflows(
-        organization_id=user.selected_organization_id,
-        status=status,
-    )
+    statuses = _validate_status_filter(status)
+    if statuses:
+        workflows = []
+        for status_value in statuses:
+            workflows.extend(
+                await db_client.get_all_workflows(
+                    organization_id=user.selected_organization_id,
+                    status=status_value,
+                )
+            )
+    else:
+        workflows = await db_client.get_all_workflows(
+            organization_id=user.selected_organization_id, status=None
+        )
     return [
         WorkflowSummaryResponse(id=workflow.id, name=workflow.name)
         for workflow in workflows

api/tests/test_workflow_list_route.py

@@ -50,3 +50,70 @@ def test_workflow_fetch_list_includes_workflow_uuid():
             "workflow_uuid": workflow.workflow_uuid,
         }
     ]
+
+
+def test_workflow_fetch_invalid_status_returns_422_without_db_query():
+    """A status outside the workflow_status enum (e.g. 'published') must fail
+    as a clean 422 instead of a 500 from the Postgres enum cast."""
+    app = _make_test_app()
+    client = TestClient(app)
+
+    with patch("api.routes.workflow.db_client") as mock_db:
+        mock_db.get_all_workflows_for_listing = AsyncMock()
+        mock_db.get_workflow_run_counts = AsyncMock()
+
+        response = client.get("/workflow/fetch?status=published")
+
+    assert response.status_code == 422
+    assert "published" in response.json()["detail"]
+    # The invalid value must never reach the database layer.
+    mock_db.get_all_workflows_for_listing.assert_not_called()
+
+
+def test_workflow_fetch_valid_single_status_passes_through():
+    app = _make_test_app()
+    client = TestClient(app)
+
+    with patch("api.routes.workflow.db_client") as mock_db:
+        mock_db.get_all_workflows_for_listing = AsyncMock(return_value=[])
+        mock_db.get_workflow_run_counts = AsyncMock(return_value={})
+
+        response = client.get("/workflow/fetch?status=active")
+
+    assert response.status_code == 200
+    mock_db.get_all_workflows_for_listing.assert_awaited_once_with(
+        organization_id=11, status="active"
+    )
+
+
+def test_workflow_fetch_comma_separated_status_queries_each_value():
+    app = _make_test_app()
+    client = TestClient(app)
+
+    with patch("api.routes.workflow.db_client") as mock_db:
+        mock_db.get_all_workflows_for_listing = AsyncMock(return_value=[])
+        mock_db.get_workflow_run_counts = AsyncMock(return_value={})
+
+        response = client.get("/workflow/fetch?status=active,archived")
+
+    assert response.status_code == 200
+    assert mock_db.get_all_workflows_for_listing.await_count == 2
+    statuses = {
+        call.kwargs["status"]
+        for call in mock_db.get_all_workflows_for_listing.await_args_list
+    }
+    assert statuses == {"active", "archived"}
+
+
+def test_workflow_fetch_mixed_valid_and_invalid_status_returns_422():
+    app = _make_test_app()
+    client = TestClient(app)
+
+    with patch("api.routes.workflow.db_client") as mock_db:
+        mock_db.get_all_workflows_for_listing = AsyncMock()
+        mock_db.get_workflow_run_counts = AsyncMock()
+
+        response = client.get("/workflow/fetch?status=active,published")
+
+    assert response.status_code == 422
+    mock_db.get_all_workflows_for_listing.assert_not_called()