feat: app deployment v2 (#39829)

Co-authored-by: zhangx1n <zhangxin@dify.ai>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/tests/test_containers_integration_tests/services/test_app_dsl_service.py

@@ -1318,7 +1318,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/console/app/test_app_response_models.py

@@ -1004,7 +1004,12 @@ def test_app_detail_api_attaches_current_user_permission_keys(app, app_module, u
                         overrides=[
                             app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                 resource_id="app-1",
-                                permission_keys=["app.acl.view_layout", "app.acl.edit", "app.acl.monitor"],
+                                permission_keys=[
+                                    "app.acl.view_layout",
+                                    "app.acl.edit",
+                                    "app.acl.deploy",
+                                    "app.acl.monitor",
+                                ],
                             )
                         ]
                     )
@@ -1026,7 +1031,12 @@ def test_app_detail_api_attaches_current_user_permission_keys(app, app_module, u
 
     get_app.assert_called_once_with(app_obj, session=unbound_session)
     get_permissions.assert_called_once_with("tenant-1", "acct-1", app_id="app-1", session=unbound_session)
-    assert resp["permission_keys"] == ["app.acl.view_layout", "app.acl.edit", "app.acl.monitor"]
+    assert resp["permission_keys"] == [
+        "app.acl.view_layout",
+        "app.acl.edit",
+        "app.acl.deploy",
+        "app.acl.monitor",
+    ]
 
 
 def test_app_copy_api_attaches_permission_keys(app, app_module, sqlite_session: Session, sqlite_engine: Engine):

api/tests/unit_tests/controllers/console/workspace/test_rbac.py

@@ -55,6 +55,27 @@ class TestCurrentIds:
             assert rbac_mod._current_ids() == ("tenant-1", "acct-1")
 
 
+class TestMyPermissions:
+    def test_returns_app_deploy_permission(self, app):
+        permissions = rbac_mod.svc.MyPermissionsResponse(
+            app=rbac_mod.svc.ResourcePermissionSnapshot(
+                default_permission_keys=["app.acl.deploy"],
+            )
+        )
+        with (
+            app.test_request_context("/workspaces/current/rbac/my-permissions"),
+            patch("controllers.console.workspace.rbac._current_ids", return_value=("tenant-1", "acct-1")),
+            patch(
+                "controllers.console.workspace.rbac.svc.RBACService.MyPermissions.get",
+                return_value=permissions,
+            ) as mock_get,
+        ):
+            response = inspect.unwrap(rbac_mod.RBACMyPermissionsApi.get)(rbac_mod.RBACMyPermissionsApi())
+
+        assert response["app"]["default_permission_keys"] == ["app.acl.deploy"]
+        mock_get.assert_called_once()
+
+
 class TestAccessMatrixAccountNames:
     def test_hydrates_missing_account_names(self):
         items = [

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

@@ -28,6 +28,7 @@ from models import Account, App
 from models.account import AccountStatus
 from models.model import AppMode, IconType
 from services.app_dsl_service import Import, ImportStatus
+from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 
 
 def _persist_app(session: Session) -> App:
@@ -235,6 +236,15 @@ class TestEnterpriseAppDSLExport:
     Uses inspect.unwrap() to bypass auth/setup decorators.
     """
 
+    def test_export_documents_query_parameters(self):
+        params = EnterpriseAppDSLExport.get.__apidoc__["params"]
+
+        assert params["include_secret"]["in"] == "query"
+        assert params["include_secret"]["type"] == "boolean"
+        assert params["workflow_id"]["in"] == "query"
+        assert params["workflow_id"]["type"] == "string"
+        assert params["workflow_id"]["format"] == "uuid"
+
     @pytest.fixture
     def api_instance(self):
         return EnterpriseAppDSLExport()
@@ -293,6 +303,160 @@ class TestEnterpriseAppDSLExport:
         assert call_kwargs["session"] is scoped_db()
         assert call_kwargs["include_secret"] is True
 
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    def test_export_selected_workflow_forwards_canonical_uuid(
+        self,
+        mock_dsl_cls,
+        api_instance,
+        app: Flask,
+        sqlite_session: Session,
+        scoped_db,
+    ):
+        app_model = _persist_app(sqlite_session)
+        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_model.id)
+
+        assert status_code == 200
+        assert body["data"] == "yaml-data"
+        call_kwargs = mock_dsl_cls.export_dsl.call_args.kwargs
+        assert call_kwargs["app_model"].id == app_model.id
+        assert call_kwargs["session"] is scoped_db()
+        assert call_kwargs["include_secret"] is False
+        assert call_kwargs["workflow_id"] == "f1fd7266-56fc-45c7-9d81-a72cd5a1b4f6"
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    def test_export_selected_workflow_with_secret(
+        self,
+        mock_dsl_cls,
+        api_instance,
+        app: Flask,
+        sqlite_session: Session,
+        scoped_db,
+    ):
+        app_model = _persist_app(sqlite_session)
+        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_model.id)
+
+        assert status_code == 200
+        assert body["data"] == "yaml-data"
+        call_kwargs = mock_dsl_cls.export_dsl.call_args.kwargs
+        assert call_kwargs["app_model"].id == app_model.id
+        assert call_kwargs["session"] is scoped_db()
+        assert call_kwargs["include_secret"] is True
+        assert call_kwargs["workflow_id"] == workflow_id
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    def test_export_rejects_invalid_selected_workflow_id(
+        self,
+        mock_dsl_cls,
+        api_instance,
+        app: Flask,
+        scoped_db,
+    ):
+        assert scoped_db() is not None
+        unwrapped = inspect.unwrap(api_instance.get)
+        with app.test_request_context("?workflow_id=not-a-uuid"):
+            body, status_code = unwrapped(api_instance, app_id=str(uuid4()))
+
+        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")
+    def test_export_selected_missing_workflow_returns_404(
+        self,
+        mock_dsl_cls,
+        api_instance,
+        app: Flask,
+        sqlite_session: Session,
+        scoped_db,
+    ):
+        app_model = _persist_app(sqlite_session)
+        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_model.id)
+
+        assert status_code == 404
+        assert body == {
+            "code": "workflow_version_not_found",
+            "message": "selected workflow not found",
+            "status": 404,
+        }
+        call_kwargs = mock_dsl_cls.export_dsl.call_args.kwargs
+        assert call_kwargs["app_model"].id == app_model.id
+        assert call_kwargs["session"] is scoped_db()
+        assert call_kwargs["include_secret"] is False
+        assert call_kwargs["workflow_id"] == workflow_id
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    def test_export_selected_draft_workflow_returns_400(
+        self,
+        mock_dsl_cls,
+        api_instance,
+        app: Flask,
+        sqlite_session: Session,
+        scoped_db,
+    ):
+        app_model = _persist_app(sqlite_session)
+        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_model.id)
+
+        assert status_code == 400
+        assert body == {
+            "code": "workflow_version_not_published",
+            "message": "selected workflow is a draft",
+            "status": 400,
+        }
+        call_kwargs = mock_dsl_cls.export_dsl.call_args.kwargs
+        assert call_kwargs["app_model"].id == app_model.id
+        assert call_kwargs["session"] is scoped_db()
+        assert call_kwargs["include_secret"] is False
+        assert call_kwargs["workflow_id"] == workflow_id
+
+    @patch("controllers.inner_api.app.dsl.AppDslService")
+    def test_export_without_selected_workflow_preserves_workflow_error(
+        self,
+        mock_dsl_cls,
+        api_instance,
+        app: Flask,
+        sqlite_session: Session,
+        scoped_db,
+    ):
+        app_model = _persist_app(sqlite_session)
+        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_model.id)
+
+        call_kwargs = mock_dsl_cls.export_dsl.call_args.kwargs
+        assert call_kwargs["app_model"].id == app_model.id
+        assert call_kwargs["session"] is scoped_db()
+        assert call_kwargs["include_secret"] is False
+        assert "workflow_id" not in call_kwargs
+
     def test_export_app_not_found_returns_404(self, api_instance, app: Flask, scoped_db):
         assert scoped_db() is not None
         unwrapped = inspect.unwrap(api_instance.get)

api/tests/unit_tests/services/enterprise/test_rbac_service.py

@@ -636,6 +636,10 @@ class TestMyPermissions:
         assert not any(key.startswith("billing.") for key in out.workspace.permission_keys)
         if role == "editor":
             assert "app.acl.log_and_annotation" in out.app.default_permission_keys
+        if role in {"owner", "admin", "editor"}:
+            assert "app.acl.deploy" in out.app.default_permission_keys
+        else:
+            assert "app.acl.deploy" not in out.app.default_permission_keys
 
     @pytest.mark.parametrize(
         ("role", "expected_snippet_keys"),

api/tests/unit_tests/services/test_app_dsl_service.py

@@ -15,6 +15,7 @@ from models.workflow import Workflow
 from services.app_dsl_service import AppDslService, PendingData
 from services.entities.dsl_entities import ImportStatus
 from services.errors.account import NoPermissionError
+from services.errors.app import WorkflowNotFoundError
 
 
 def test_extract_workflow_dependencies_uses_llm_environment_variable_provider(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -407,3 +408,20 @@ def test_import_app_reraises_permission_denial_instead_of_failed_result(
         )
 
     assert not unbound_session.in_transaction()
+
+
+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

@@ -11,6 +11,7 @@ This test suite covers:
 
 import json
 import uuid
+from datetime import datetime, timedelta
 from types import SimpleNamespace
 from typing import Any, cast
 from unittest.mock import ANY, MagicMock, patch, sentinel
@@ -356,6 +357,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"
@@ -1057,6 +1083,104 @@ class TestWorkflowService:
         assert result.marked_comment == "Initial release"
         assert retirement_candidates == set()
 
+    def test_publish_workflow_numbers_versions_from_one(
+        self, workflow_service: WorkflowService, sqlite_session: Session
+    ):
+        """
+        Test publish_workflow assigns an app-scoped version number starting at #1.
+
+        The number is what users see when a version carries no name, so it has to be
+        stable and monotonic per app rather than derived from list position.
+        """
+        app = TestWorkflowAssociatedDataFactory.create_app()
+        account = TestWorkflowAssociatedDataFactory.create_account()
+        graph = TestWorkflowAssociatedDataFactory.create_valid_workflow_graph()
+
+        draft = TestWorkflowAssociatedDataFactory.create_workflow(version=Workflow.VERSION_DRAFT, graph=graph)
+        sqlite_session.add(draft)
+        sqlite_session.commit()
+
+        with (
+            patch("services.workflow_service.app_published_workflow_was_updated"),
+            patch(
+                "services.workflow_service.dify_config.DEPLOYMENT_EDITION",
+                DeploymentEdition.COMMUNITY,
+            ),
+        ):
+            first, _ = workflow_service.publish_workflow(session=sqlite_session, app_model=app, account=account)
+            second, _ = workflow_service.publish_workflow(session=sqlite_session, app_model=app, account=account)
+
+        assert first.version_number == 1
+        assert second.version_number == 2
+
+    def test_publish_workflow_does_not_reuse_a_deleted_version_number(
+        self, workflow_service: WorkflowService, sqlite_session: Session
+    ):
+        """
+        Test version numbers are never handed out twice, even after a version is deleted.
+
+        Deployment records and audit logs refer to versions by number, so reusing one
+        would make two different workflows share an identity.
+        """
+        app = TestWorkflowAssociatedDataFactory.create_app()
+        account = TestWorkflowAssociatedDataFactory.create_account()
+        graph = TestWorkflowAssociatedDataFactory.create_valid_workflow_graph()
+
+        draft = TestWorkflowAssociatedDataFactory.create_workflow(version=Workflow.VERSION_DRAFT, graph=graph)
+        sqlite_session.add(draft)
+        sqlite_session.commit()
+
+        with (
+            patch("services.workflow_service.app_published_workflow_was_updated"),
+            patch(
+                "services.workflow_service.dify_config.DEPLOYMENT_EDITION",
+                DeploymentEdition.COMMUNITY,
+            ),
+        ):
+            published, _ = workflow_service.publish_workflow(session=sqlite_session, app_model=app, account=account)
+            sqlite_session.flush()
+            sqlite_session.delete(published)
+            sqlite_session.flush()
+
+            republished, _ = workflow_service.publish_workflow(session=sqlite_session, app_model=app, account=account)
+
+        assert republished.version_number == 2
+
+    def test_publish_workflow_numbers_each_app_independently(
+        self, workflow_service: WorkflowService, sqlite_session: Session
+    ):
+        """
+        Test the counter is scoped per app rather than global.
+
+        Every app starts its own sequence at #1; a busy neighbour must not advance it.
+        """
+        account = TestWorkflowAssociatedDataFactory.create_account()
+        graph = TestWorkflowAssociatedDataFactory.create_valid_workflow_graph()
+
+        published: list[Workflow] = []
+        for app_id in ("app-first", "app-second"):
+            app = TestWorkflowAssociatedDataFactory.create_app(app_id=app_id)
+            draft = TestWorkflowAssociatedDataFactory.create_workflow(
+                workflow_id=f"draft-{app_id}",
+                app_id=app_id,
+                version=Workflow.VERSION_DRAFT,
+                graph=graph,
+            )
+            sqlite_session.add(draft)
+            sqlite_session.commit()
+
+            with (
+                patch("services.workflow_service.app_published_workflow_was_updated"),
+                patch(
+                    "services.workflow_service.dify_config.DEPLOYMENT_EDITION",
+                    DeploymentEdition.COMMUNITY,
+                ),
+            ):
+                workflow, _ = workflow_service.publish_workflow(session=sqlite_session, app_model=app, account=account)
+            published.append(workflow)
+
+        assert [workflow.version_number for workflow in published] == [1, 1]
+
     def test_publish_workflow_no_draft_raises_error(self, workflow_service: WorkflowService, sqlite_session: Session):
         """
         Test publish_workflow raises error when no draft exists.
@@ -1180,6 +1304,44 @@ class TestWorkflowService:
         assert len(workflows) == 5
         assert has_more is False
 
+    def test_get_all_published_workflow_lists_the_draft_first(
+        self, workflow_service: WorkflowService, sqlite_session: Session
+    ):
+        """
+        Test the draft heads the version list no matter how old it is.
+
+        A draft is created together with its app and its `created_at` is never refreshed,
+        so ordering purely by publish time would put it last — off the first page entirely
+        once the app has accumulated enough published versions.
+        """
+        app = TestWorkflowAssociatedDataFactory.create_app(workflow_id="workflow-3")
+        app_created_at = datetime(2026, 1, 1)
+
+        sqlite_session.add(
+            TestWorkflowAssociatedDataFactory.create_workflow(
+                workflow_id="workflow-draft",
+                version=Workflow.VERSION_DRAFT,
+                created_at=app_created_at,
+            )
+        )
+        sqlite_session.add_all(
+            [
+                TestWorkflowAssociatedDataFactory.create_workflow(
+                    workflow_id=f"workflow-{i}",
+                    version=f"2026-02-0{i} 00:00:00",
+                    created_at=app_created_at + timedelta(days=i),
+                )
+                for i in range(1, 4)
+            ]
+        )
+        sqlite_session.commit()
+
+        workflows, _ = workflow_service.get_all_published_workflow(
+            session=sqlite_session, app_model=app, page=1, limit=2, user_id=None
+        )
+
+        assert [workflow.id for workflow in workflows] == ["workflow-draft", "workflow-3"]
+
     def test_get_all_published_workflow_has_more(self, workflow_service: WorkflowService, sqlite_session: Session):
         """
         Test get_all_published_workflow indicates has_more when results exceed limit.