fix(api): validate selected workflow export semantics

api/controllers/inner_api/app/dsl.py

@@ -5,6 +5,8 @@ to attribute the created app; workspace/membership validation is done by the
 Go admin-api caller.
 """
 
+from uuid import UUID
+
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
@@ -20,6 +22,7 @@ from models import Account, App
 from models.account import AccountStatus
 from services.app_dsl_service import AppDslService
 from services.entities.dsl_entities import ImportMode, ImportStatus
+from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 
 
 class InnerAppDSLImportPayload(BaseModel):
@@ -84,22 +87,46 @@ class EnterpriseAppDSLExport(Resource):
         "enterprise_app_dsl_export",
         responses={
             200: "Export successful",
-            404: "App not found",
+            400: "Invalid workflow ID or unpublished workflow version",
+            404: "App or workflow version not found",
         },
     )
     def get(self, app_id: str):
         """Export an app's DSL as YAML."""
         include_secret = request.args.get("include_secret", "false").lower() == "true"
+        workflow_id = request.args.get("workflow_id")
 
         app_model = db.session.get(App, app_id)
         if not app_model:
             return {"message": "app not found"}, 404
 
-        data = AppDslService.export_dsl(
-            app_model=app_model,
-            session=db.session(),
-            include_secret=include_secret,
-        )
+        if not workflow_id:
+            data = AppDslService.export_dsl(
+                app_model=app_model,
+                session=db.session(),
+                include_secret=include_secret,
+            )
+        else:
+            try:
+                workflow_id = str(UUID(workflow_id))
+            except ValueError:
+                return {
+                    "code": "invalid_workflow_id",
+                    "message": "workflow_id must be a valid UUID",
+                    "status": 400,
+                }, 400
+
+            try:
+                data = AppDslService.export_dsl(
+                    app_model=app_model,
+                    session=db.session(),
+                    include_secret=include_secret,
+                    workflow_id=workflow_id,
+                )
+            except WorkflowNotFoundError as exc:
+                return {"code": "workflow_version_not_found", "message": str(exc), "status": 404}, 404
+            except IsDraftWorkflowError as exc:
+                return {"code": "workflow_version_not_published", "message": str(exc), "status": 400}, 400
 
         return {"data": data}, 200
 

api/services/app_dsl_service.py

@@ -649,10 +649,13 @@ class AppDslService:
         Append workflow export data
         :param export_data: export data
         :param app_model: App instance
+        :param workflow_id: Optional published workflow version to export
         """
         workflow_service = WorkflowService()
         workflow = workflow_service.get_draft_workflow(app_model, workflow_id, session=session)
         if not workflow:
+            if workflow_id:
+                raise WorkflowNotFoundError(f"Workflow version not found. Workflow ID: {workflow_id}.")
             raise WorkflowNotFoundError("Missing draft workflow configuration, please check.")
 
         workflow_dict = workflow.to_dict(include_secret=include_secret)

api/tests/test_containers_integration_tests/services/test_app_dsl_service.py

@@ -1305,7 +1305,7 @@ class TestAppDslService:
 
         with pytest.raises(
             WorkflowNotFoundError,
-            match="Missing draft workflow configuration, please check.",
+            match="Workflow version not found. Workflow ID:",
         ):
             AppDslService.export_dsl(
                 app, include_secret=False, workflow_id=str(uuid4()), session=db_session_with_containers

api/tests/unit_tests/controllers/inner_api/app/test_dsl.py

@@ -20,6 +20,7 @@ from controllers.inner_api.app.dsl import (
 )
 from models.account import AccountStatus
 from services.app_dsl_service import Import, ImportStatus
+from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 
 
 class TestInnerAppDSLImportPayload:
@@ -239,6 +240,125 @@ class TestEnterpriseAppDSLExport:
         assert status_code == 200
         mock_dsl_cls.export_dsl.assert_called_once_with(app_model=mock_app, session=ANY, include_secret=True)
 
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    @patch("controllers.inner_api.app.dsl.db")
+    def test_export_selected_workflow_forwards_canonical_uuid(self, mock_db, mock_dsl_cls, api_instance, app: Flask):
+        mock_app = MagicMock()
+        mock_db.session.get.return_value = mock_app
+        mock_dsl_cls.export_dsl.return_value = "yaml-data"
+        workflow_id = "F1FD7266-56FC-45C7-9D81-A72CD5A1B4F6"
+
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context(f"?workflow_id={workflow_id}"):
+            body, status_code = unwrapped(api_instance, app_id="app-123")
+
+        assert status_code == 200
+        assert body["data"] == "yaml-data"
+        mock_dsl_cls.export_dsl.assert_called_once_with(
+            app_model=mock_app,
+            session=ANY,
+            include_secret=False,
+            workflow_id="f1fd7266-56fc-45c7-9d81-a72cd5a1b4f6",
+        )
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    @patch("controllers.inner_api.app.dsl.db")
+    def test_export_selected_workflow_with_secret(self, mock_db, mock_dsl_cls, api_instance, app: Flask):
+        mock_app = MagicMock()
+        mock_db.session.get.return_value = mock_app
+        mock_dsl_cls.export_dsl.return_value = "yaml-data"
+        workflow_id = "f1fd7266-56fc-45c7-9d81-a72cd5a1b4f6"
+
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context(f"?include_secret=true&workflow_id={workflow_id}"):
+            body, status_code = unwrapped(api_instance, app_id="app-123")
+
+        assert status_code == 200
+        assert body["data"] == "yaml-data"
+        mock_dsl_cls.export_dsl.assert_called_once_with(
+            app_model=mock_app,
+            session=ANY,
+            include_secret=True,
+            workflow_id=workflow_id,
+        )
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    @patch("controllers.inner_api.app.dsl.db")
+    def test_export_rejects_invalid_selected_workflow_id(self, mock_db, mock_dsl_cls, api_instance, app: Flask):
+        mock_db.session.get.return_value = MagicMock()
+
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context("?workflow_id=not-a-uuid"):
+            body, status_code = unwrapped(api_instance, app_id="app-123")
+
+        assert status_code == 400
+        assert body == {
+            "code": "invalid_workflow_id",
+            "message": "workflow_id must be a valid UUID",
+            "status": 400,
+        }
+        mock_dsl_cls.export_dsl.assert_not_called()
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    @patch("controllers.inner_api.app.dsl.db")
+    def test_export_selected_missing_workflow_returns_404(self, mock_db, mock_dsl_cls, api_instance, app: Flask):
+        mock_db.session.get.return_value = MagicMock()
+        mock_dsl_cls.export_dsl.side_effect = WorkflowNotFoundError("selected workflow not found")
+        workflow_id = "f1fd7266-56fc-45c7-9d81-a72cd5a1b4f6"
+
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context(f"?workflow_id={workflow_id}"):
+            body, status_code = unwrapped(api_instance, app_id="app-123")
+
+        assert status_code == 404
+        assert body == {
+            "code": "workflow_version_not_found",
+            "message": "selected workflow not found",
+            "status": 404,
+        }
+        mock_dsl_cls.export_dsl.assert_called_once_with(
+            app_model=ANY,
+            session=ANY,
+            include_secret=False,
+            workflow_id=workflow_id,
+        )
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    @patch("controllers.inner_api.app.dsl.db")
+    def test_export_selected_draft_workflow_returns_400(self, mock_db, mock_dsl_cls, api_instance, app: Flask):
+        mock_db.session.get.return_value = MagicMock()
+        mock_dsl_cls.export_dsl.side_effect = IsDraftWorkflowError("selected workflow is a draft")
+        workflow_id = "f1fd7266-56fc-45c7-9d81-a72cd5a1b4f6"
+
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context(f"?workflow_id={workflow_id}"):
+            body, status_code = unwrapped(api_instance, app_id="app-123")
+
+        assert status_code == 400
+        assert body == {
+            "code": "workflow_version_not_published",
+            "message": "selected workflow is a draft",
+            "status": 400,
+        }
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    @patch("controllers.inner_api.app.dsl.db")
+    def test_export_without_selected_workflow_preserves_workflow_error(
+        self, mock_db, mock_dsl_cls, api_instance, app: Flask
+    ):
+        mock_app = MagicMock()
+        mock_db.session.get.return_value = mock_app
+        mock_dsl_cls.export_dsl.side_effect = WorkflowNotFoundError(
+            "Missing draft workflow configuration, please check."
+        )
+
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context():
+            with pytest.raises(WorkflowNotFoundError, match="Missing draft workflow configuration"):
+                unwrapped(api_instance, app_id="app-123")
+
+        mock_dsl_cls.export_dsl.assert_called_once_with(app_model=mock_app, session=ANY, include_secret=False)
+
     @patch("controllers.inner_api.app.dsl.db")
     def test_export_app_not_found_returns_404(self, mock_db, api_instance, app: Flask):
         mock_db.session.get.return_value = None

api/tests/unit_tests/services/test_app_dsl_service.py

@@ -9,6 +9,7 @@ from models import App, AppMode
 from models.model import AppModelConfig, IconType
 from services.app_dsl_service import AppDslService
 from services.entities.dsl_entities import ImportStatus
+from services.errors.app import WorkflowNotFoundError
 
 
 @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
@@ -166,3 +167,20 @@ def test_export_dsl_loads_model_config_and_annotation_reply_with_request_session
     session.get.assert_called_once_with(AppModelConfig, "config-1")
     load_annotation_reply_config.assert_called_once_with(session, "app-1")
     app_model_config.to_dict.assert_called_once_with(annotation_reply=annotation_reply)
+
+
+def test_append_workflow_export_data_reports_missing_selected_workflow(monkeypatch: pytest.MonkeyPatch) -> None:
+    workflow_id = "11111111-1111-4111-8111-111111111111"
+    workflow_service = Mock()
+    workflow_service.get_draft_workflow.return_value = None
+    monkeypatch.setattr("services.app_dsl_service.WorkflowService", Mock(return_value=workflow_service))
+    app = cast(App, SimpleNamespace(id="app-1", tenant_id="tenant-1"))
+
+    with pytest.raises(WorkflowNotFoundError, match=f"Workflow version not found. Workflow ID: {workflow_id}"):
+        AppDslService._append_workflow_export_data(
+            export_data={},
+            app_model=app,
+            include_secret=False,
+            session=Mock(),
+            workflow_id=workflow_id,
+        )

api/tests/unit_tests/services/test_workflow_service.py

@@ -359,6 +359,31 @@ class TestWorkflowService:
 
         assert result is None
 
+    @pytest.mark.parametrize(
+        ("tenant_id", "app_id"),
+        [("other-tenant", "app-123"), ("tenant-456", "other-app")],
+    )
+    def test_get_published_workflow_by_id_rejects_foreign_workflow(
+        self,
+        tenant_id: str,
+        app_id: str,
+        workflow_service: WorkflowService,
+        sqlite_session: Session,
+    ):
+        app = TestWorkflowAssociatedDataFactory.create_app()
+        workflow = TestWorkflowAssociatedDataFactory.create_workflow(
+            workflow_id="workflow-123",
+            tenant_id=tenant_id,
+            app_id=app_id,
+            version="v1",
+        )
+        sqlite_session.add(workflow)
+        sqlite_session.commit()
+
+        result = workflow_service.get_published_workflow_by_id(app, workflow.id, session=sqlite_session)
+
+        assert result is None
+
     def test_get_published_workflow_success(self, workflow_service: WorkflowService, sqlite_session: Session):
         """Test get_published_workflow returns published workflow."""
         workflow_id = "workflow-123"