test: complete workflow test doubles with env/conversation variables

The secret-scrub wiring reads workflow.environment_variables /
conversation_variables when building the response converter; unit tests that
mock the workflow as a bare SimpleNamespace must provide these attributes
(empty = no secrets). Fixes the workflow task-pipeline and HITL contract tests.

api/controllers/console/app/app.py

@@ -5,6 +5,7 @@ from datetime import datetime
 from typing import Any, Literal
 
 from flask import request
+from flask_login import current_user
 from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, ValidationInfo, computed_field, field_validator, model_validator
 from sqlalchemy import select
@@ -889,6 +890,10 @@ class AppCopyApi(Resource):
         # The role of the current user in the ta table must be admin, owner, or editor
         args = CopyAppPayload.model_validate(console_ns.payload or {})
 
+        # The copy endpoint exports DSL server-side and immediately re-imports it within the
+        # same tenant — the YAML is never returned to the caller, so there is no secret
+        # exfiltration risk.  The assert_secret_export_allowed guard is only needed on the
+        # /export endpoint where plaintext is sent to the client.
         with Session(db.engine, expire_on_commit=False) as session:
             import_service = AppDslService(session)
             yaml_content = import_service.export_dsl(app_model=app_model, session=session, include_secret=True)
@@ -960,6 +965,11 @@ class AppExportApi(Resource):
         """Export app"""
         args = AppExportQuery.model_validate(request.args.to_dict(flat=True))
 
+        AppDslService.assert_secret_export_allowed(
+            include_secret=args.include_secret,
+            current_role=current_user.current_role,
+        )
+
         response = AppExportResponse(
             data=AppDslService.export_dsl(
                 app_model=app_model,

api/core/app/apps/advanced_chat/app_runner.py

@@ -32,6 +32,7 @@ from core.moderation.input_moderation import InputModeration
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import get_default_root_node_id
 from core.workflow.nodes.agent_v2.session_cleanup_layer import build_workflow_agent_session_cleanup_layer
+from core.workflow.secret_scrub import collect_workflow_secret_values
 from core.workflow.system_variables import (
     build_bootstrap_variables,
     build_system_variables,
@@ -260,6 +261,10 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 workflow_type=WorkflowType(self._workflow.type),
                 version=self._workflow.version,
                 graph_data=self._workflow.graph_dict,
+                secret_values=collect_workflow_secret_values(
+                    self._workflow.environment_variables,
+                    self._workflow.conversation_variables,
+                ),
             ),
             workflow_execution_repository=self._workflow_execution_repository,
             workflow_node_execution_repository=self._workflow_node_execution_repository,

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -76,12 +76,14 @@ from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from core.workflow.file_reference import resolve_file_record_id
 from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
+from core.workflow.secret_scrub import collect_workflow_secret_values
 from core.workflow.system_variables import build_system_variables
 from graphon.enums import WorkflowExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMUsage
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from graphon.nodes import BuiltinNodeTypes
 from graphon.runtime import GraphRuntimeState
+from graphon.variables import VariableBase
 from libs.datetime_utils import naive_utc_now
 from models import Account, Conversation, EndUser, Message, MessageFile
 from models.enums import CreatorUserRole, MessageFileBelongsTo, MessageStatus
@@ -97,6 +99,8 @@ class WorkflowSnapshot:
     id: str
     tenant_id: str
     features_dict: Mapping[str, Any]
+    environment_variables: tuple[VariableBase, ...]
+    conversation_variables: tuple[VariableBase, ...]
 
     @classmethod
     def from_workflow(cls, workflow: Workflow) -> "WorkflowSnapshot":
@@ -104,6 +108,8 @@ class WorkflowSnapshot:
             id=workflow.id,
             tenant_id=workflow.tenant_id,
             features_dict=dict(workflow.features_dict),
+            environment_variables=tuple(workflow.environment_variables),
+            conversation_variables=tuple(workflow.conversation_variables),
         )
 
 
@@ -188,6 +194,10 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             application_generate_entity=application_generate_entity,
             user=user,
             system_variables=self._workflow_system_variables,
+            secret_values=collect_workflow_secret_values(
+                workflow.environment_variables,
+                workflow.conversation_variables,
+            ),
         )
 
         self._task_state = WorkflowTaskState()
@@ -747,7 +757,8 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         )
 
         with self._database_session() as session:
-            err_event = QueueErrorEvent(error=ValueError(f"Run failed: {event.error}"))
+            redacted_error = self._workflow_response_converter._redact(event.error)
+            err_event = QueueErrorEvent(error=ValueError(f"Run failed: {redacted_error}"))
             err = self._base_task_pipeline.handle_error(event=err_event, session=session, message_id=self._message_id)
 
         yield workflow_finish_resp

api/core/app/apps/common/workflow_response_converter.py

@@ -4,7 +4,7 @@ import time
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, NewType, TypedDict, Union
+from typing import Any, NewType, TypedDict, Union, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -61,6 +61,7 @@ from core.workflow.human_input_policy import (
     resolve_human_input_pause_reason_inputs,
 )
 from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
+from core.workflow.secret_scrub import redact_secret_values
 from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
@@ -130,10 +131,12 @@ class WorkflowResponseConverter:
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
         user: Union[Account, EndUser],
         system_variables: Sequence[Variable],
+        secret_values: tuple[str, ...] = (),
     ):
         self._application_generate_entity = application_generate_entity
         self._user = user
         self._system_variables = system_variables_to_mapping(system_variables)
+        self._secret_values = secret_values
         self._workflow_inputs = self._prepare_workflow_inputs()
 
         # Disable truncation for SERVICE_API calls to keep backward compatibility.
@@ -234,6 +237,9 @@ class WorkflowResponseConverter:
         converter = WorkflowRuntimeTypeConverter()
         return converter.to_json_encodable(outputs)
 
+    def _redact[T](self, value: T) -> T:
+        return cast("T", redact_secret_values(value, self._secret_values))
+
     def workflow_start_to_stream_response(
         self,
         *,
@@ -279,7 +285,7 @@ class WorkflowResponseConverter:
         elapsed_time = (finished_at - started_at).total_seconds()
 
         outputs_mapping = graph_runtime_state.outputs or {}
-        encoded_outputs = WorkflowRuntimeTypeConverter().to_json_encodable(outputs_mapping)
+        encoded_outputs = self._redact(WorkflowRuntimeTypeConverter().to_json_encodable(outputs_mapping))
 
         created_by: CreatedByDict | dict[str, object] = {}
         user = self._user
@@ -304,7 +310,7 @@ class WorkflowResponseConverter:
                 workflow_id=workflow_id,
                 status=status,
                 outputs=encoded_outputs,
-                error=error,
+                error=self._redact(error),
                 elapsed_time=elapsed_time,
                 total_tokens=graph_runtime_state.total_tokens,
                 total_steps=graph_runtime_state.node_run_steps,
@@ -331,7 +337,7 @@ class WorkflowResponseConverter:
             )
         paused_at = naive_utc_now()
         elapsed_time = (paused_at - started_at).total_seconds()
-        encoded_outputs = self._encode_outputs(event.outputs) or {}
+        encoded_outputs = self._redact(self._encode_outputs(event.outputs) or {})
         if self._application_generate_entity.invoke_from == InvokeFrom.SERVICE_API:
             encoded_outputs = {}
         variable_pool = graph_runtime_state.variable_pool
@@ -575,9 +581,9 @@ class WorkflowResponseConverter:
         finished_at = event.finished_at or naive_utc_now()
         elapsed_time = (finished_at - start_at).total_seconds()
 
-        inputs, inputs_truncated = self._truncate_mapping(event.inputs)
-        process_data, process_data_truncated = self._truncate_mapping(event.process_data)
-        encoded_outputs = self._encode_outputs(event.outputs)
+        inputs, inputs_truncated = self._truncate_mapping(self._redact(event.inputs))
+        process_data, process_data_truncated = self._truncate_mapping(self._redact(event.process_data))
+        encoded_outputs = self._redact(self._encode_outputs(event.outputs))
         outputs, outputs_truncated = self._truncate_mapping(encoded_outputs)
         metadata = self._merge_metadata(event.execution_metadata, snapshot)
 
@@ -608,9 +614,9 @@ class WorkflowResponseConverter:
                 outputs=outputs,
                 outputs_truncated=outputs_truncated,
                 status=status,
-                error=error_message,
+                error=self._redact(error_message),
                 elapsed_time=elapsed_time,
-                execution_metadata=metadata,
+                execution_metadata=self._redact(metadata),
                 created_at=int(start_at.timestamp()),
                 finished_at=int(finished_at.timestamp()),
                 files=self.fetch_files_from_node_outputs(event.outputs or {}),
@@ -635,9 +641,9 @@ class WorkflowResponseConverter:
         finished_at = naive_utc_now()
         elapsed_time = (finished_at - event.start_at).total_seconds()
 
-        inputs, inputs_truncated = self._truncate_mapping(event.inputs)
-        process_data, process_data_truncated = self._truncate_mapping(event.process_data)
-        encoded_outputs = self._encode_outputs(event.outputs)
+        inputs, inputs_truncated = self._truncate_mapping(self._redact(event.inputs))
+        process_data, process_data_truncated = self._truncate_mapping(self._redact(event.process_data))
+        encoded_outputs = self._redact(self._encode_outputs(event.outputs))
         outputs, outputs_truncated = self._truncate_mapping(encoded_outputs)
         metadata = self._merge_metadata(event.execution_metadata, snapshot)
 
@@ -657,9 +663,9 @@ class WorkflowResponseConverter:
                 outputs=outputs,
                 outputs_truncated=outputs_truncated,
                 status=WorkflowNodeExecutionStatus.RETRY,
-                error=event.error,
+                error=self._redact(event.error),
                 elapsed_time=elapsed_time,
-                execution_metadata=metadata,
+                execution_metadata=self._redact(metadata),
                 created_at=int(snapshot.start_at.timestamp()),
                 finished_at=int(finished_at.timestamp()),
                 files=self.fetch_files_from_node_outputs(event.outputs or {}),
@@ -723,9 +729,9 @@ class WorkflowResponseConverter:
     ) -> IterationNodeCompletedStreamResponse:
         json_converter = WorkflowRuntimeTypeConverter()
 
-        new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(event.inputs or {})
+        new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(self._redact(event.inputs or {}))
         new_outputs, outputs_truncated = self._truncator.truncate_variable_mapping(
-            json_converter.to_json_encodable(event.outputs) or {}
+            self._redact(json_converter.to_json_encodable(event.outputs) or {})
         )
         return IterationNodeCompletedStreamResponse(
             task_id=task_id,
@@ -747,7 +753,7 @@ class WorkflowResponseConverter:
                 error=None,
                 elapsed_time=(naive_utc_now() - event.start_at).total_seconds(),
                 total_tokens=(lambda x: x if isinstance(x, int) else 0)(event.metadata.get("total_tokens", 0)),
-                execution_metadata=event.metadata,
+                execution_metadata=self._redact(event.metadata),
                 finished_at=int(time.time()),
                 steps=event.steps,
             ),
@@ -805,9 +811,9 @@ class WorkflowResponseConverter:
         event: QueueLoopCompletedEvent,
     ) -> LoopNodeCompletedStreamResponse:
         json_converter = WorkflowRuntimeTypeConverter()
-        new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(event.inputs or {})
+        new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(self._redact(event.inputs or {}))
         new_outputs, outputs_truncated = self._truncator.truncate_variable_mapping(
-            json_converter.to_json_encodable(event.outputs) or {}
+            self._redact(json_converter.to_json_encodable(event.outputs) or {})
         )
         return LoopNodeCompletedStreamResponse(
             task_id=task_id,
@@ -829,7 +835,7 @@ class WorkflowResponseConverter:
                 error=None,
                 elapsed_time=(naive_utc_now() - event.start_at).total_seconds(),
                 total_tokens=(lambda x: x if isinstance(x, int) else 0)(event.metadata.get("total_tokens", 0)),
-                execution_metadata=event.metadata,
+                execution_metadata=self._redact(event.metadata),
                 finished_at=int(time.time()),
                 steps=event.steps,
             ),

api/core/app/apps/pipeline/pipeline_runner.py

@@ -18,6 +18,7 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.db.session_factory import create_session
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import DifyGraphInitContext, DifyNodeFactory, get_default_root_node_id
+from core.workflow.secret_scrub import collect_workflow_secret_values
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
@@ -225,6 +226,10 @@ class PipelineRunner(WorkflowBasedAppRunner):
                 workflow_type=WorkflowType(workflow.type),
                 version=workflow.version,
                 graph_data=workflow.graph_dict,
+                secret_values=collect_workflow_secret_values(
+                    workflow.environment_variables,
+                    workflow.conversation_variables,
+                ),
             ),
             workflow_execution_repository=self._workflow_execution_repository,
             workflow_node_execution_repository=self._workflow_node_execution_repository,

api/core/app/apps/workflow/app_runner.py

@@ -15,6 +15,7 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import get_default_root_node_id
 from core.workflow.nodes.agent_v2.session_cleanup_layer import build_workflow_agent_session_cleanup_layer
+from core.workflow.secret_scrub import collect_workflow_secret_values
 from core.workflow.snippet_start import get_compatible_start_aliases
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
@@ -190,6 +191,10 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 workflow_type=WorkflowType(self._workflow.type),
                 version=self._workflow.version,
                 graph_data=self._workflow.graph_dict,
+                secret_values=collect_workflow_secret_values(
+                    self._workflow.environment_variables,
+                    self._workflow.conversation_variables,
+                ),
             ),
             workflow_execution_repository=self._workflow_execution_repository,
             workflow_node_execution_repository=self._workflow_node_execution_repository,

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -61,6 +61,7 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.workflow.secret_scrub import collect_workflow_secret_values
 from core.workflow.system_variables import build_system_variables
 from extensions.ext_database import db
 from graphon.entities import WorkflowStartReason
@@ -120,6 +121,10 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             application_generate_entity=application_generate_entity,
             user=user,
             system_variables=self._workflow_system_variables,
+            secret_values=collect_workflow_secret_values(
+                self._workflow.environment_variables,
+                self._workflow.conversation_variables,
+            ),
         )
         self._graph_runtime_state: GraphRuntimeState | None = self._base_task_pipeline.queue_manager.graph_runtime_state
 

api/core/app/workflow/layers/persistence.py

@@ -12,7 +12,7 @@ state.
 from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Union, override
+from typing import Any, Union, cast, override
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
 from core.app.workflow.retry_history import RETRY_HISTORY_PROCESS_DATA_KEY, WorkflowNodeRetryAttempt
@@ -20,6 +20,7 @@ from core.helper.trace_id_helper import ParentTraceContext
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.secret_scrub import redact_secret_values
 from core.workflow.system_variables import SystemVariableKey
 from core.workflow.variable_prefixes import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
@@ -64,6 +65,7 @@ class PersistenceWorkflowInfo:
     workflow_type: WorkflowType
     version: str
     graph_data: Mapping[str, Any]
+    secret_values: tuple[str, ...] = ()
 
 
 @dataclass(slots=True)
@@ -164,7 +166,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
 
     def _handle_graph_run_succeeded(self, event: GraphRunSucceededEvent) -> None:
         execution = self._get_workflow_execution()
-        execution.outputs = event.outputs
+        execution.outputs = self._redact(event.outputs)
         execution.status = WorkflowExecutionStatus.SUCCEEDED
         self._populate_completion_statistics(execution)
 
@@ -174,7 +176,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
 
     def _handle_graph_run_partial_succeeded(self, event: GraphRunPartialSucceededEvent) -> None:
         execution = self._get_workflow_execution()
-        execution.outputs = event.outputs
+        execution.outputs = self._redact(event.outputs)
         execution.status = WorkflowExecutionStatus.PARTIAL_SUCCEEDED
         execution.exceptions_count = event.exceptions_count
         self._populate_completion_statistics(execution)
@@ -186,7 +188,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
     def _handle_graph_run_failed(self, event: GraphRunFailedEvent) -> None:
         execution = self._get_workflow_execution()
         execution.status = WorkflowExecutionStatus.FAILED
-        execution.error_message = event.error
+        execution.error_message = self._redact(event.error)
         execution.exceptions_count = event.exceptions_count
         self._populate_completion_statistics(execution)
 
@@ -198,7 +200,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
     def _handle_graph_run_aborted(self, event: GraphRunAbortedEvent) -> None:
         execution = self._get_workflow_execution()
         execution.status = WorkflowExecutionStatus.STOPPED
-        execution.error_message = event.reason or "Workflow execution aborted"
+        execution.error_message = self._redact(event.reason or "Workflow execution aborted")
         self._populate_completion_statistics(execution)
 
         self._fail_running_node_executions(error_message=execution.error_message or "")
@@ -209,7 +211,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
     def _handle_graph_run_paused(self, event: GraphRunPausedEvent) -> None:
         execution = self._get_workflow_execution()
         execution.status = WorkflowExecutionStatus.PAUSED
-        execution.outputs = event.outputs
+        execution.outputs = self._redact(event.outputs)
         self._populate_completion_statistics(execution, update_finished=False)
 
         self._workflow_execution_repository.save(execution)
@@ -257,7 +259,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
     def _handle_node_retry(self, event: NodeRunRetryEvent) -> None:
         domain_execution = self._get_node_execution(event.id)
         domain_execution.status = WorkflowNodeExecutionStatus.RETRY
-        domain_execution.error = event.error
+        domain_execution.error = cast(str, self._redact(event.error))
         self._append_retry_history(domain_execution, event)
         self._workflow_node_execution_repository.save(domain_execution)
         self._workflow_node_execution_repository.save_execution_data(domain_execution)
@@ -425,7 +427,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         domain_execution.elapsed_time = max((actual_finished_at - start_at).total_seconds(), 0.0)
 
         if error:
-            domain_execution.error = error
+            domain_execution.error = self._redact(error)
 
         if update_outputs:
             projected_outputs = project_node_outputs_for_workflow_run(
@@ -435,10 +437,10 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             )
             process_data = self._merge_retry_history(domain_execution.process_data, node_result.process_data)
             domain_execution.update_from_mapping(
-                inputs=node_result.inputs,
-                process_data=process_data,
-                outputs=projected_outputs,
-                metadata=node_result.metadata,
+                inputs=self._redact(node_result.inputs),
+                process_data=self._redact(process_data),
+                outputs=self._redact(projected_outputs),
+                metadata=self._redact(node_result.metadata),
             )
 
         self._workflow_node_execution_repository.save(domain_execution)
@@ -446,10 +448,11 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
 
     def _fail_running_node_executions(self, *, error_message: str) -> None:
         now = naive_utc_now()
+        redacted_error = self._redact(error_message)
         for execution in self._node_execution_cache.values():
             if execution.status == WorkflowNodeExecutionStatus.RUNNING:
                 execution.status = WorkflowNodeExecutionStatus.FAILED
-                execution.error = error_message
+                execution.error = redacted_error
                 execution.finished_at = now
                 execution.elapsed_time = max((now - execution.created_at).total_seconds(), 0.0)
                 self._workflow_node_execution_repository.save(execution)
@@ -481,6 +484,9 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         )
         self._trace_manager.add_trace_task(trace_task)
 
+    def _redact[T](self, value: T) -> T:
+        return cast("T", redact_secret_values(value, self._workflow_info.secret_values))
+
     def _system_variables(self) -> Mapping[str, Any]:
         runtime_state = self.graph_runtime_state
         return runtime_state.variable_pool.get_by_prefix(SYSTEM_VARIABLE_NODE_ID)

api/core/workflow/secret_scrub.py

@@ -0,0 +1,87 @@
+"""Value-matching backstop redactor for workflow run logs.
+
+Known limitations: secrets transformed (encoded, truncated, …) before logging
+won't match; mapping *keys* are not scrubbed; the registry is collected from
+workflow-definition secrets only and may not cover all runtime credential
+sources.
+"""
+
+from collections.abc import Iterable, Mapping
+
+from graphon.variables import SecretVariable, VariableBase
+
+SECRET_PLACEHOLDER = "[REDACTED_SECRET]"
+# Values shorter than this are redacted only on an exact whole-string match;
+# substring-redacting a short value would corrupt unrelated text containing it.
+_MIN_SUBSTRING_LENGTH = 8
+
+
+def collect_workflow_secret_values(
+    environment_variables: Iterable[VariableBase],
+    conversation_variables: Iterable[VariableBase],
+) -> tuple[str, ...]:
+    """Collect the plaintext of every secret env/conversation variable.
+
+    Returns:
+        A de-duplicated tuple of non-empty secret string values, in first-seen order.
+    """
+    values: list[str] = []
+    for variable in (*environment_variables, *conversation_variables):
+        if isinstance(variable, SecretVariable) and isinstance(variable.value, str) and variable.value:
+            values.append(variable.value)
+    return tuple(dict.fromkeys(values))
+
+
+def _split_values(secret_values: Iterable[str]) -> tuple[tuple[str, ...], frozenset[str]]:
+    """Partition secrets into substring-redactable (long) and exact-match-only (short) sets."""
+    substrings: list[str] = []
+    exacts: set[str] = set()
+    for value in secret_values:
+        if not value:
+            continue
+        if len(value) >= _MIN_SUBSTRING_LENGTH:
+            substrings.append(value)
+        else:
+            exacts.add(value)
+    # Sort longest-first so a longer secret containing a shorter one is matched first,
+    # preventing the shorter secret from fragmenting a longer overlapping one.
+    deduped = list(dict.fromkeys(substrings))
+    return tuple(sorted(deduped, key=len, reverse=True)), frozenset(exacts)
+
+
+def redact_secret_values(value: object, secret_values: Iterable[str]) -> object:
+    """Return ``value`` unchanged when there is nothing to redact, otherwise a copy with secrets replaced.
+
+    Returns:
+        ``value`` unchanged (same object) when there are no secrets to redact;
+        otherwise a structurally-identical copy with secrets replaced by
+        ``SECRET_PLACEHOLDER``.
+    """
+    substrings, exacts = _split_values(secret_values)
+    if not substrings and not exacts:
+        return value
+
+    def redact_text(text: str) -> str:
+        if text in exacts:
+            return SECRET_PLACEHOLDER
+        for secret in substrings:
+            text = text.replace(secret, SECRET_PLACEHOLDER)
+        return text
+
+    def visit(item: object) -> object:
+        if isinstance(item, str):
+            return redact_text(item)
+        if isinstance(item, Mapping):
+            # Keys are field/variable names, not secret plaintext — intentionally not scrubbed.
+            return {key: visit(nested) for key, nested in item.items()}
+        if isinstance(item, list):
+            return [visit(nested) for nested in item]
+        if isinstance(item, tuple):
+            return tuple(visit(nested) for nested in item)
+        if isinstance(item, set):
+            return {visit(nested) for nested in item}
+        if isinstance(item, frozenset):
+            return frozenset(visit(nested) for nested in item)
+        return item
+
+    return visit(value)

api/services/app_dsl_service.py

@@ -14,6 +14,7 @@ from packaging.version import parse as parse_version
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
 from constants.dsl_version import CURRENT_APP_DSL_VERSION
@@ -37,6 +38,7 @@ from graphon.nodes.question_classifier.entities import QuestionClassifierNodeDat
 from graphon.nodes.tool.entities import ToolNodeData
 from libs.datetime_utils import naive_utc_now
 from models import Account, App, AppMode
+from models.account import TenantAccountRole
 from models.model import AppModelConfig, AppModelConfigDict, IconType, load_annotation_reply_config
 from models.workflow import Workflow
 from services.agent.dsl_service import AgentDslService, AgentPackage
@@ -395,6 +397,12 @@ class AppDslService:
             leaked_dependencies=leaked_dependencies,
         )
 
+    @staticmethod
+    def assert_secret_export_allowed(*, include_secret: bool, current_role: TenantAccountRole | None) -> None:
+        """Only workspace managers (owner/admin) may export secrets in a DSL."""
+        if include_secret and not TenantAccountRole.is_privileged_role(current_role):
+            raise Forbidden("Only workspace managers can export secrets.")
+
     def _create_or_update_app(
         self,
         *,

api/tests/unit_tests/controllers/service_api/app/test_hitl_service_api.py

@@ -433,7 +433,13 @@ class TestHitlServiceApi:
         )
         pipeline = AdvancedChatAppGenerateTaskPipeline(
             application_generate_entity=application_generate_entity,
-            workflow=SimpleNamespace(id="workflow-id", tenant_id="tenant", features_dict={}),
+            workflow=SimpleNamespace(
+                id="workflow-id",
+                tenant_id="tenant",
+                features_dict={},
+                environment_variables=[],
+                conversation_variables=[],
+            ),
             queue_manager=SimpleNamespace(invoke_from=InvokeFrom.WEB_APP, graph_runtime_state=None),
             conversation=SimpleNamespace(id="conv-id", mode=AppMode.ADVANCED_CHAT),
             message=SimpleNamespace(
@@ -526,7 +532,13 @@ class TestHitlServiceApi:
         )
         pipeline = WorkflowAppGenerateTaskPipeline(
             application_generate_entity=application_generate_entity,
-            workflow=SimpleNamespace(id="workflow-id", tenant_id="tenant", features_dict={}),
+            workflow=SimpleNamespace(
+                id="workflow-id",
+                tenant_id="tenant",
+                features_dict={},
+                environment_variables=[],
+                conversation_variables=[],
+            ),
             queue_manager=SimpleNamespace(invoke_from=InvokeFrom.WEB_APP, graph_runtime_state=None),
             user=SimpleNamespace(id="user", session_id="session"),
             stream=False,

api/tests/unit_tests/core/app/apps/advanced_chat/test_app_generator.py

@@ -401,7 +401,14 @@ class TestAdvancedChatAppGeneratorInternals:
             workflow_run_id="run-id",
         )
 
-        workflow = SimpleNamespace(id="wf-1", tenant_id="tenant", features={"feature": True}, features_dict={})
+        workflow = SimpleNamespace(
+            id="wf-1",
+            tenant_id="tenant",
+            features={"feature": True},
+            features_dict={},
+            environment_variables=[],
+            conversation_variables=[],
+        )
         conversation = SimpleNamespace(id="conv-1", mode=AppMode.ADVANCED_CHAT, override_model_configs=None)
         message = SimpleNamespace(
             id="msg-1",
@@ -506,7 +513,14 @@ class TestAdvancedChatAppGeneratorInternals:
             workflow_run_id="run-id",
         )
 
-        workflow = SimpleNamespace(id="wf-2", tenant_id="tenant", features={}, features_dict={})
+        workflow = SimpleNamespace(
+            id="wf-2",
+            tenant_id="tenant",
+            features={},
+            features_dict={},
+            environment_variables=[],
+            conversation_variables=[],
+        )
         conversation = SimpleNamespace(id="conv-2", mode=AppMode.ADVANCED_CHAT, override_model_configs=None)
         message = SimpleNamespace(
             id="msg-2",
@@ -981,7 +995,13 @@ class TestAdvancedChatAppGeneratorInternals:
         with pytest.raises(GenerateTaskStoppedError):
             generator._handle_advanced_chat_response(
                 application_generate_entity=application_generate_entity,
-                workflow=WorkflowSnapshot(id="wf", tenant_id="tenant", features_dict={}),
+                workflow=WorkflowSnapshot(
+                    id="wf",
+                    tenant_id="tenant",
+                    features_dict={},
+                    environment_variables=(),
+                    conversation_variables=(),
+                ),
                 queue_manager=SimpleNamespace(),
                 conversation=ConversationSnapshot(id="conv", mode=AppMode.ADVANCED_CHAT),
                 message=MessageSnapshot(
@@ -1029,7 +1049,13 @@ class TestAdvancedChatAppGeneratorInternals:
             with pytest.raises(ValueError, match="other error"):
                 generator._handle_advanced_chat_response(
                     application_generate_entity=application_generate_entity,
-                    workflow=WorkflowSnapshot(id="wf", tenant_id="tenant", features_dict={}),
+                    workflow=WorkflowSnapshot(
+                        id="wf",
+                        tenant_id="tenant",
+                        features_dict={},
+                        environment_variables=(),
+                        conversation_variables=(),
+                    ),
                     queue_manager=SimpleNamespace(),
                     conversation=ConversationSnapshot(id="conv", mode=AppMode.ADVANCED_CHAT),
                     message=MessageSnapshot(

api/tests/unit_tests/core/app/apps/advanced_chat/test_generate_task_pipeline.py

@@ -170,7 +170,13 @@ def test_resume_appends_chunks_to_paused_answer() -> None:
     user = EndUser()
     user.id = "user-1"
     user.session_id = "session-1"
-    workflow = pipeline_module.WorkflowSnapshot(id="workflow-1", tenant_id="tenant-1", features_dict={})
+    workflow = pipeline_module.WorkflowSnapshot(
+        id="workflow-1",
+        tenant_id="tenant-1",
+        features_dict={},
+        environment_variables=(),
+        conversation_variables=(),
+    )
 
     pipeline = pipeline_module.AdvancedChatAppGenerateTaskPipeline(
         application_generate_entity=application_generate_entity,

api/tests/unit_tests/core/app/apps/advanced_chat/test_generate_task_pipeline_core.py

@@ -92,7 +92,13 @@ def _make_pipeline():
         answer="",
     )
     conversation = ConversationSnapshot(id="conv-id", mode=AppMode.ADVANCED_CHAT)
-    workflow = WorkflowSnapshot(id="workflow-id", tenant_id="tenant", features_dict={})
+    workflow = WorkflowSnapshot(
+        id="workflow-id",
+        tenant_id="tenant",
+        features_dict={},
+        environment_variables=(),
+        conversation_variables=(),
+    )
     user = EndUser(tenant_id="tenant", type="session", name="tester", session_id="session")
 
     pipeline = AdvancedChatAppGenerateTaskPipeline(

api/tests/unit_tests/core/app/apps/common/test_workflow_response_converter.py

@@ -1,8 +1,22 @@
 from collections.abc import Mapping, Sequence
+from unittest.mock import Mock
 
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
+from core.app.entities.queue_entities import (
+    QueueNodeFailedEvent,
+    QueueNodeRetryEvent,
+    QueueNodeStartedEvent,
+    QueueNodeSucceededEvent,
+)
+from core.workflow.secret_scrub import SECRET_PLACEHOLDER
+from core.workflow.system_variables import build_system_variables
+from graphon.entities import WorkflowStartReason
+from graphon.enums import BuiltinNodeTypes, WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey
 from graphon.file import FILE_MODEL_IDENTITY, File, FileTransferMethod, FileType
 from graphon.variables.segments import ArrayFileSegment, FileSegment
+from libs.datetime_utils import naive_utc_now
+from models import Account
 
 
 class TestWorkflowResponseConverterFetchFilesFromVariableValue:
@@ -256,3 +270,187 @@ class TestWorkflowResponseConverterFetchFilesFromVariableValue:
         assert len(result) == 2
         assert result[0]["id"] == "complex_file"
         assert result[1]["id"] == "complex_obj"
+
+
+class TestWorkflowResponseConverterSecretRedaction:
+    """Test that WorkflowResponseConverter redacts secret values from SSE stream responses."""
+
+    def _make_converter(self, secret_values: tuple[str, ...] = ()) -> WorkflowResponseConverter:
+        mock_entity = Mock(spec=WorkflowAppGenerateEntity)
+        mock_app_config = Mock()
+        mock_app_config.tenant_id = "test-tenant-id"
+        mock_entity.invoke_from = InvokeFrom.WEB_APP
+        mock_entity.app_config = mock_app_config
+        mock_entity.inputs = {}
+
+        mock_user = Mock(spec=Account)
+        mock_user.id = "test-user-id"
+        mock_user.name = "Test User"
+        mock_user.email = "test@example.com"
+
+        system_variables = build_system_variables(workflow_id="wf-id", workflow_execution_id="run-id")
+        return WorkflowResponseConverter(
+            application_generate_entity=mock_entity,
+            user=mock_user,
+            system_variables=system_variables,
+            secret_values=secret_values,
+        )
+
+    def _prime_converter(self, converter: WorkflowResponseConverter) -> QueueNodeStartedEvent:
+        """Seed workflow run id and node snapshot so finish events are processable."""
+        converter.workflow_start_to_stream_response(
+            task_id="t1",
+            workflow_run_id="run-id",
+            workflow_id="wf-id",
+            reason=WorkflowStartReason.INITIAL,
+        )
+        start_event = QueueNodeStartedEvent(
+            node_execution_id="exec-1",
+            node_id="node-1",
+            node_title="Test Node",
+            node_type=BuiltinNodeTypes.CODE,
+            start_at=naive_utc_now(),
+            in_iteration_id=None,
+            in_loop_id=None,
+            provider_type="built-in",
+            provider_id="code",
+        )
+        converter.workflow_node_start_to_stream_response(event=start_event, task_id="t1")
+        return start_event
+
+    def test_node_finish_redacts_inputs_process_data_outputs(self):
+        """node-finish stream response must replace secret values with SECRET_PLACEHOLDER."""
+        secret = "supersecretvalue123"
+        converter = self._make_converter(secret_values=(secret,))
+        start_event = self._prime_converter(converter)
+
+        succeeded_event = QueueNodeSucceededEvent(
+            node_execution_id=start_event.node_execution_id,
+            node_id="node-1",
+            node_type=BuiltinNodeTypes.CODE,
+            start_at=naive_utc_now(),
+            inputs={"api_key": secret},
+            process_data={"raw_response": f"token={secret}"},
+            outputs={"result": f"hello {secret} world"},
+        )
+
+        response = converter.workflow_node_finish_to_stream_response(event=succeeded_event, task_id="t1")
+
+        assert response is not None
+        # Secret must be replaced in all three fields
+        assert SECRET_PLACEHOLDER in str(response.data.inputs)
+        assert secret not in str(response.data.inputs)
+        assert SECRET_PLACEHOLDER in str(response.data.process_data)
+        assert secret not in str(response.data.process_data)
+        assert SECRET_PLACEHOLDER in str(response.data.outputs)
+        assert secret not in str(response.data.outputs)
+
+    def test_workflow_finish_redacts_outputs(self):
+        """workflow_finish stream response must replace secret values with SECRET_PLACEHOLDER."""
+        secret = "supersecretvalue123"
+        converter = self._make_converter(secret_values=(secret,))
+        self._prime_converter(converter)
+
+        mock_graph_runtime_state = Mock()
+        mock_graph_runtime_state.outputs = {"result": f"token={secret}"}
+        mock_graph_runtime_state.total_tokens = 42
+        mock_graph_runtime_state.node_run_steps = 1
+
+        response = converter.workflow_finish_to_stream_response(
+            task_id="t1",
+            workflow_id="wf-id",
+            status=WorkflowExecutionStatus.SUCCEEDED,
+            graph_runtime_state=mock_graph_runtime_state,
+        )
+
+        assert response is not None
+        assert SECRET_PLACEHOLDER in str(response.data.outputs)
+        assert secret not in str(response.data.outputs)
+
+    def test_node_finish_redacts_error_and_execution_metadata(self):
+        """node-finish stream response must redact secret from error string and execution_metadata."""
+        secret = "supersecretvalue123"
+        converter = self._make_converter(secret_values=(secret,))
+        start_event = self._prime_converter(converter)
+
+        failed_event = QueueNodeFailedEvent(
+            node_execution_id=start_event.node_execution_id,
+            node_id="node-1",
+            node_type=BuiltinNodeTypes.CODE,
+            start_at=naive_utc_now(),
+            inputs={},
+            process_data={},
+            outputs={},
+            error=f"API call failed with token={secret}",
+            execution_metadata={
+                WorkflowNodeExecutionMetadataKey.TOOL_INFO: {"credential": secret},
+            },
+        )
+
+        response = converter.workflow_node_finish_to_stream_response(event=failed_event, task_id="t1")
+
+        assert response is not None
+        # error field must be scrubbed
+        assert response.data.error is not None
+        assert secret not in response.data.error
+        assert SECRET_PLACEHOLDER in response.data.error
+        # execution_metadata must be scrubbed
+        assert response.data.execution_metadata is not None
+        assert secret not in str(response.data.execution_metadata)
+        assert SECRET_PLACEHOLDER in str(response.data.execution_metadata)
+
+    def test_node_retry_redacts_error_and_execution_metadata(self):
+        """node-retry stream response must redact secret from error string and execution_metadata."""
+        secret = "supersecretvalue123"
+        converter = self._make_converter(secret_values=(secret,))
+        start_event = self._prime_converter(converter)
+
+        retry_event = QueueNodeRetryEvent(
+            node_execution_id=start_event.node_execution_id,
+            node_id="node-1",
+            node_title="Test Node",
+            node_type=BuiltinNodeTypes.CODE,
+            start_at=naive_utc_now(),
+            provider_type="built-in",
+            provider_id="code",
+            inputs={},
+            process_data={},
+            outputs={},
+            error=f"Retry because of token={secret}",
+            execution_metadata={WorkflowNodeExecutionMetadataKey.TOOL_INFO: {"credential": secret}},
+            retry_index=1,
+        )
+
+        response = converter.workflow_node_retry_to_stream_response(event=retry_event, task_id="t1")
+
+        assert response is not None
+        assert response.data.error is not None
+        assert secret not in response.data.error
+        assert SECRET_PLACEHOLDER in response.data.error
+        assert response.data.execution_metadata is not None
+        assert secret not in str(response.data.execution_metadata)
+        assert SECRET_PLACEHOLDER in str(response.data.execution_metadata)
+
+    def test_workflow_finish_redacts_error(self):
+        """workflow_finish stream response must redact secret from error string."""
+        secret = "supersecretvalue123"
+        converter = self._make_converter(secret_values=(secret,))
+        self._prime_converter(converter)
+
+        mock_graph_runtime_state = Mock()
+        mock_graph_runtime_state.outputs = {}
+        mock_graph_runtime_state.total_tokens = 0
+        mock_graph_runtime_state.node_run_steps = 1
+
+        response = converter.workflow_finish_to_stream_response(
+            task_id="t1",
+            workflow_id="wf-id",
+            status=WorkflowExecutionStatus.FAILED,
+            graph_runtime_state=mock_graph_runtime_state,
+            error=f"Workflow failed: bad credential={secret}",
+        )
+
+        assert response is not None
+        assert response.data.error is not None
+        assert secret not in response.data.error
+        assert SECRET_PLACEHOLDER in response.data.error

api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py

@@ -79,7 +79,13 @@ def _make_pipeline():
         extras={},
         call_depth=0,
     )
-    workflow = SimpleNamespace(id="workflow-id", tenant_id="tenant", features_dict={})
+    workflow = SimpleNamespace(
+        id="workflow-id",
+        tenant_id="tenant",
+        features_dict={},
+        environment_variables=[],
+        conversation_variables=[],
+    )
     user = SimpleNamespace(id="user", session_id="session")
 
     pipeline = WorkflowAppGenerateTaskPipeline(
@@ -505,7 +511,13 @@ class TestWorkflowGenerateTaskPipeline:
             extras={},
             call_depth=0,
         )
-        workflow = SimpleNamespace(id="workflow-id", tenant_id="tenant", features_dict={})
+        workflow = SimpleNamespace(
+            id="workflow-id",
+            tenant_id="tenant",
+            features_dict={},
+            environment_variables=[],
+            conversation_variables=[],
+        )
         queue_manager = SimpleNamespace(invoke_from=InvokeFrom.WEB_APP, graph_runtime_state=None)
         end_user = EndUser(tenant_id="tenant", type="session", name="user", session_id="session-id")
         end_user.id = "end-user-id"

api/tests/unit_tests/core/app/workflow/layers/test_persistence.py

@@ -9,10 +9,11 @@ from core.app.workflow.layers.persistence import (
     _NodeRuntimeSnapshot,
 )
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus, WorkflowType
+from graphon.graph_events import GraphRunFailedEvent, GraphRunSucceededEvent
 from graphon.node_events import NodeRunResult
 
 
-def _build_layer() -> WorkflowPersistenceLayer:
+def _build_layer(secret_values: tuple[str, ...] = ()) -> WorkflowPersistenceLayer:
     application_generate_entity = Mock()
     application_generate_entity.inputs = {}
 
@@ -23,6 +24,7 @@ def _build_layer() -> WorkflowPersistenceLayer:
             workflow_type=WorkflowType.WORKFLOW,
             version="1",
             graph_data={},
+            secret_values=secret_values,
         ),
         workflow_execution_repository=Mock(),
         workflow_node_execution_repository=Mock(),
@@ -97,3 +99,122 @@ def test_update_node_execution_projects_start_outputs() -> None:
         outputs={"question": "hello"},
         metadata={},
     )
+
+
+def test_update_node_execution_redacts_secret_values() -> None:
+    from core.workflow.secret_scrub import SECRET_PLACEHOLDER
+
+    layer = _build_layer(secret_values=("supersecretvalue123",))
+    node_execution = Mock()
+    node_execution.id = "node-exec-secret"
+    node_execution.node_type = BuiltinNodeTypes.START
+    node_execution.created_at = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC).replace(tzinfo=None)
+    node_execution.update_from_mapping = Mock()
+
+    layer._node_snapshots[node_execution.id] = _NodeRuntimeSnapshot(
+        node_id="start",
+        title="Start",
+        predecessor_node_id=None,
+        iteration_id=None,
+        loop_id=None,
+        created_at=node_execution.created_at,
+    )
+
+    layer._update_node_execution(
+        node_execution,
+        NodeRunResult(
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            inputs={"api_key": "Bearer supersecretvalue123"},
+            outputs={"question": "hello"},
+            process_data={"token": "supersecretvalue123"},
+        ),
+        WorkflowNodeExecutionStatus.SUCCEEDED,
+    )
+
+    kwargs = node_execution.update_from_mapping.call_args.kwargs
+    assert kwargs["inputs"] == {"api_key": f"Bearer {SECRET_PLACEHOLDER}"}
+    assert kwargs["process_data"] == {"token": SECRET_PLACEHOLDER}
+    assert "supersecretvalue123" not in str(kwargs)
+
+
+def test_handle_graph_run_succeeded_redacts_outputs(monkeypatch: pytest.MonkeyPatch) -> None:
+    from core.workflow.secret_scrub import SECRET_PLACEHOLDER
+
+    layer = _build_layer(secret_values=("supersecretvalue123",))
+
+    # Provide a real-attribute-storing mock as the current workflow execution.
+    execution = Mock()
+    layer._workflow_execution = execution
+
+    # Stub out _populate_completion_statistics — it accesses graph_runtime_state,
+    # which requires a bound GraphEngine; irrelevant to the redaction assertion.
+    layer._populate_completion_statistics = Mock()
+
+    # Silence the inspector side-effect; _enqueue_trace_task is already a no-op
+    # because trace_manager defaults to None in _build_layer.
+    monkeypatch.setattr(
+        "core.app.workflow.layers.persistence._inspector_publish_workflow_completed",
+        Mock(),
+    )
+
+    event = GraphRunSucceededEvent(outputs={"result": "Bearer supersecretvalue123"})
+    layer._handle_graph_run_succeeded(event)
+
+    assert execution.outputs == {"result": f"Bearer {SECRET_PLACEHOLDER}"}
+    assert "supersecretvalue123" not in str(execution.outputs)
+
+
+def test_update_node_execution_redacts_error_field() -> None:
+    """Secrets in node error strings must be redacted before persistence."""
+    from core.workflow.secret_scrub import SECRET_PLACEHOLDER
+
+    layer = _build_layer(secret_values=("supersecretvalue123",))
+    node_execution = Mock()
+    node_execution.id = "node-exec-error"
+    node_execution.node_type = BuiltinNodeTypes.START
+    node_execution.created_at = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC).replace(tzinfo=None)
+    node_execution.update_from_mapping = Mock()
+
+    layer._node_snapshots[node_execution.id] = _NodeRuntimeSnapshot(
+        node_id="code",
+        title="Code",
+        predecessor_node_id=None,
+        iteration_id=None,
+        loop_id=None,
+        created_at=node_execution.created_at,
+    )
+
+    layer._update_node_execution(
+        node_execution,
+        NodeRunResult(status=WorkflowNodeExecutionStatus.FAILED),
+        WorkflowNodeExecutionStatus.FAILED,
+        error="API call failed: token=supersecretvalue123 rejected",
+    )
+
+    assert SECRET_PLACEHOLDER in node_execution.error
+    assert "supersecretvalue123" not in node_execution.error
+
+
+def test_handle_graph_run_failed_redacts_error_message(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Secrets in workflow-level error messages must be redacted before persistence."""
+    from core.workflow.secret_scrub import SECRET_PLACEHOLDER
+
+    layer = _build_layer(secret_values=("supersecretvalue123",))
+
+    execution = Mock()
+    layer._workflow_execution = execution
+
+    layer._populate_completion_statistics = Mock()
+    # Stub _fail_running_node_executions to avoid iterating empty cache with redact calls.
+    layer._fail_running_node_executions = Mock()
+
+    monkeypatch.setattr(
+        "core.app.workflow.layers.persistence._inspector_publish_workflow_completed",
+        Mock(),
+    )
+
+    event = GraphRunFailedEvent(error="Upstream error: key=supersecretvalue123 invalid")
+    layer._handle_graph_run_failed(event)
+
+    assert SECRET_PLACEHOLDER in execution.error_message
+    assert "supersecretvalue123" not in execution.error_message

api/tests/unit_tests/core/workflow/test_secret_scrub.py

@@ -0,0 +1,60 @@
+from core.workflow.secret_scrub import (
+    SECRET_PLACEHOLDER,
+    collect_workflow_secret_values,
+    redact_secret_values,
+)
+from graphon.variables import SecretVariable, StringVariable
+
+SECRET = "supersecretvalue123"
+
+
+def test_empty_registry_is_a_no_op() -> None:
+    value = {"a": SECRET, "b": [SECRET]}
+    result = redact_secret_values(value, ())
+    assert result is value
+
+
+def test_long_secret_is_substring_redacted() -> None:
+    result = redact_secret_values({"h": f"Bearer {SECRET}"}, (SECRET,))
+    assert result == {"h": f"Bearer {SECRET_PLACEHOLDER}"}
+
+
+def test_short_secret_only_exact_match() -> None:
+    # "abc" (<8) must not corrupt unrelated text that merely contains it
+    result = redact_secret_values({"x": "abcdef", "y": "abc"}, ("abc",))
+    assert result == {"x": "abcdef", "y": SECRET_PLACEHOLDER}
+
+
+def test_redaction_recurses_nested_structures() -> None:
+    value = {"list": [f"k={SECRET}"], "tuple": (SECRET,), "n": 5}
+    result = redact_secret_values(value, (SECRET,))
+    assert result == {"list": [f"k={SECRET_PLACEHOLDER}"], "tuple": (SECRET_PLACEHOLDER,), "n": 5}
+
+
+def test_collect_gathers_env_and_conversation_secrets_only() -> None:
+    env = [SecretVariable(value="env-secret-123", name="API_KEY"), StringVariable(value="plain", name="p")]
+    conv = [SecretVariable(value="conv-secret-456", name="TOKEN")]
+    assert collect_workflow_secret_values(env, conv) == ("env-secret-123", "conv-secret-456")
+
+
+def test_collect_deduplicates_and_skips_empty() -> None:
+    env = [
+        SecretVariable(value="dup", name="a"),
+        SecretVariable(value="dup", name="b"),
+        SecretVariable(value="", name="c"),
+    ]
+    assert collect_workflow_secret_values(env, []) == ("dup",)
+
+
+def test_overlapping_secret_longer_wins() -> None:
+    # When a short secret is a substring of a longer secret,
+    # the longer secret must be fully masked without being fragmented.
+    long_secret = "supersecretvalue123"
+    short_secret = "supersecret"  # prefix of long_secret, len >= _MIN_SUBSTRING_LENGTH
+    text = f"token={long_secret}"
+
+    # Both secrets registered; longer-first ordering prevents the short one from
+    # splitting the long one before it can be matched.
+    result = redact_secret_values(text, (short_secret, long_secret))
+    # The full long secret must be replaced by exactly one placeholder.
+    assert result == f"token={SECRET_PLACEHOLDER}"

api/tests/unit_tests/services/test_app_dsl_service_secret_export.py

@@ -0,0 +1,30 @@
+import pytest
+from werkzeug.exceptions import Forbidden
+
+from models.account import TenantAccountRole
+from services.app_dsl_service import AppDslService
+
+
+def test_non_secret_export_allowed_for_any_role() -> None:
+    # Must not raise for a normal member exporting without secrets.
+    AppDslService.assert_secret_export_allowed(include_secret=False, current_role=TenantAccountRole.NORMAL)
+
+
+def test_secret_export_allowed_for_privileged_roles() -> None:
+    AppDslService.assert_secret_export_allowed(include_secret=True, current_role=TenantAccountRole.OWNER)
+    AppDslService.assert_secret_export_allowed(include_secret=True, current_role=TenantAccountRole.ADMIN)
+
+
+def test_secret_export_forbidden_for_non_privileged_role() -> None:
+    with pytest.raises(Forbidden):
+        AppDslService.assert_secret_export_allowed(include_secret=True, current_role=TenantAccountRole.NORMAL)
+
+
+def test_secret_export_forbidden_for_none_role() -> None:
+    with pytest.raises(Forbidden):
+        AppDslService.assert_secret_export_allowed(include_secret=True, current_role=None)
+
+
+def test_secret_export_forbidden_for_editor_role() -> None:
+    with pytest.raises(Forbidden):
+        AppDslService.assert_secret_export_allowed(include_secret=True, current_role=TenantAccountRole.EDITOR)