diff --git a/api/controllers/console/app/workflow.py b/api/controllers/console/app/workflow.py
index 02167245be..8c451cd08c 100644
--- a/api/controllers/console/app/workflow.py
+++ b/api/controllers/console/app/workflow.py
@@ -16,6 +16,7 @@ from controllers.console.wraps import account_initialization_required, edit_perm
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
@@ -990,11 +991,13 @@ class DraftWorkflowTriggerRunApi(Resource):
             event = poller.poll()
             if not event:
                 return jsonable_encoder({"status": "waiting", "retry_in": LISTENING_RETRY_IN})
+            workflow_args = dict(event.workflow_args)
+            workflow_args[SKIP_PREPARE_USER_INPUTS_KEY] = True
             return helper.compact_generate_response(
                 AppGenerateService.generate(
                     app_model=app_model,
                     user=current_user,
-                    args=event.workflow_args,
+                    args=workflow_args,
                     invoke_from=InvokeFrom.DEBUGGER,
                     streaming=True,
                     root_node_id=node_id,
@@ -1145,10 +1148,12 @@ class DraftWorkflowTriggerRunAllApi(Resource):
             return jsonable_encoder({"status": "waiting", "retry_in": LISTENING_RETRY_IN})
 
         try:
+            workflow_args = dict(trigger_debug_event.workflow_args)
+            workflow_args[SKIP_PREPARE_USER_INPUTS_KEY] = True
             response = AppGenerateService.generate(
                 app_model=app_model,
                 user=current_user,
-                args=trigger_debug_event.workflow_args,
+                args=workflow_args,
                 invoke_from=InvokeFrom.DEBUGGER,
                 streaming=True,
                 root_node_id=trigger_debug_event.node_id,
diff --git a/api/core/app/apps/workflow/app_generator.py b/api/core/app/apps/workflow/app_generator.py
index 305f6cecd5..6c0a10d138 100644
--- a/api/core/app/apps/workflow/app_generator.py
+++ b/api/core/app/apps/workflow/app_generator.py
@@ -39,10 +39,16 @@ from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTrigger
 from models.enums import WorkflowRunTriggeredFrom
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
 
+SKIP_PREPARE_USER_INPUTS_KEY = "_skip_prepare_user_inputs"
+
 logger = logging.getLogger(__name__)
 
 
 class WorkflowAppGenerator(BaseAppGenerator):
+    @staticmethod
+    def _should_prepare_user_inputs(args: Mapping[str, Any]) -> bool:
+        return not bool(args.get(SKIP_PREPARE_USER_INPUTS_KEY))
+
     @overload
     def generate(
         self,
@@ -139,8 +145,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
             **extract_external_trace_id_from_args(args),
         }
         workflow_run_id = str(uuid.uuid4())
-        if triggered_from in (WorkflowRunTriggeredFrom.DEBUGGING, WorkflowRunTriggeredFrom.APP_RUN):
-            # start node get inputs
+        # for trigger debug run, not prepare user inputs 
+        if self._should_prepare_user_inputs(args):
             inputs = self._prepare_user_inputs(
                 user_inputs=inputs,
                 variables=app_config.variables,
diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py
new file mode 100644
index 0000000000..83ac3a5591
--- /dev/null
+++ b/api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py
@@ -0,0 +1,19 @@
+from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY, WorkflowAppGenerator
+
+
+def test_should_prepare_user_inputs_defaults_to_true():
+    args = {"inputs": {}}
+
+    assert WorkflowAppGenerator()._should_prepare_user_inputs(args)
+
+
+def test_should_prepare_user_inputs_skips_when_flag_truthy():
+    args = {"inputs": {}, SKIP_PREPARE_USER_INPUTS_KEY: True}
+
+    assert not WorkflowAppGenerator()._should_prepare_user_inputs(args)
+
+
+def test_should_prepare_user_inputs_keeps_validation_when_flag_false():
+    args = {"inputs": {}, SKIP_PREPARE_USER_INPUTS_KEY: False}
+
+    assert WorkflowAppGenerator()._should_prepare_user_inputs(args)