diff --git a/api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py b/api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py
index 8e44bd68738..8efb59a8e9c 100644
--- a/api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py
+++ b/api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py
@@ -53,6 +53,7 @@ from services.rag_pipeline.pipeline_generate_service import PipelineGenerateServ
 from services.rag_pipeline.rag_pipeline import RagPipelineService
 from services.rag_pipeline.rag_pipeline_manage_service import RagPipelineManageService
 from services.rag_pipeline.rag_pipeline_transform_service import RagPipelineTransformService
+from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
 
@@ -781,7 +782,38 @@ class RagPipelineByIdApi(Resource):
             # Commit the transaction in the controller
             session.commit()
 
-        return workflow
+            return workflow
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @get_rag_pipeline
+    def delete(self, pipeline: Pipeline, workflow_id: str):
+        """
+        Delete a published workflow version that is not currently active on the pipeline.
+        """
+        if pipeline.workflow_id == workflow_id:
+            abort(400, description=f"Cannot delete workflow that is currently in use by pipeline '{pipeline.id}'")
+
+        workflow_service = WorkflowService()
+
+        with Session(db.engine) as session:
+            try:
+                workflow_service.delete_workflow(
+                    session=session,
+                    workflow_id=workflow_id,
+                    tenant_id=pipeline.tenant_id,
+                )
+                session.commit()
+            except WorkflowInUseError as e:
+                abort(400, description=str(e))
+            except DraftWorkflowDeletionError as e:
+                abort(400, description=str(e))
+            except ValueError as e:
+                raise NotFound(str(e))
+
+        return None, 204
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/processing/parameters")
diff --git a/api/tests/test_containers_integration_tests/conftest.py b/api/tests/test_containers_integration_tests/conftest.py
index 48bf3ca446d..be8a1c6aab0 100644
--- a/api/tests/test_containers_integration_tests/conftest.py
+++ b/api/tests/test_containers_integration_tests/conftest.py
@@ -32,6 +32,7 @@ from extensions.ext_database import db
 # Configure logging for test containers
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
+_TEST_SANDBOX_IMAGE = os.getenv("TEST_SANDBOX_IMAGE", "langgenius/dify-sandbox:0.2.12")
 
 DEFAULT_SANDBOX_TEST_IMAGE = "langgenius/dify-sandbox:0.2.14"
 SANDBOX_TEST_IMAGE_ENV = "DIFY_SANDBOX_TEST_IMAGE"
diff --git a/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py b/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py
index aa7c3c7fbdc..a3c0592d766 100644
--- a/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py
+++ b/api/tests/unit_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py
@@ -2,7 +2,7 @@ from datetime import datetime
 from unittest.mock import MagicMock, patch
 
 import pytest
-from werkzeug.exceptions import Forbidden, HTTPException, NotFound
+from werkzeug.exceptions import BadRequest, Forbidden, HTTPException, NotFound
 
 import services
 from controllers.console import console_ns
@@ -692,6 +692,57 @@ class TestRagPipelineByIdApi:
             result, status = method(api, pipeline, "w1")
             assert status == 400
 
+    def test_delete_success(self, app):
+        api = RagPipelineByIdApi()
+        method = unwrap(api.delete)
+
+        pipeline = MagicMock(tenant_id="t1", workflow_id="active-workflow", id="pipeline-1")
+
+        workflow_service = MagicMock()
+
+        session = MagicMock()
+        session_ctx = MagicMock()
+        session_ctx.__enter__.return_value = session
+        session_ctx.__exit__.return_value = None
+
+        fake_db = MagicMock()
+        fake_db.engine = MagicMock()
+
+        with (
+            app.test_request_context("/", method="DELETE"),
+            patch(
+                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.db",
+                fake_db,
+            ),
+            patch(
+                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.Session",
+                return_value=session_ctx,
+            ),
+            patch(
+                "controllers.console.datasets.rag_pipeline.rag_pipeline_workflow.WorkflowService",
+                return_value=workflow_service,
+            ),
+        ):
+            result = method(api, pipeline, "old-workflow")
+
+        workflow_service.delete_workflow.assert_called_once_with(
+            session=session,
+            workflow_id="old-workflow",
+            tenant_id="t1",
+        )
+        session.commit.assert_called_once()
+        assert result == (None, 204)
+
+    def test_delete_active_workflow_rejected(self, app):
+        api = RagPipelineByIdApi()
+        method = unwrap(api.delete)
+
+        pipeline = MagicMock(tenant_id="t1", workflow_id="active-workflow", id="pipeline-1")
+
+        with app.test_request_context("/", method="DELETE"):
+            with pytest.raises(BadRequest, match="currently in use by pipeline"):
+                method(api, pipeline, "active-workflow")
+
 
 class TestRagPipelineWorkflowLastRunApi:
     def test_last_run_success(self, app):