diff --git a/api/core/app/workflow/layers/persistence.py b/api/core/app/workflow/layers/persistence.py
index 1708b39b7c2..52887765c02 100644
--- a/api/core/app/workflow/layers/persistence.py
+++ b/api/core/app/workflow/layers/persistence.py
@@ -15,6 +15,7 @@ from datetime import datetime
 from typing import Any, Union, override
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
+from core.app.workflow.retry_history import RETRY_HISTORY_PROCESS_DATA_KEY, WorkflowNodeRetryAttempt
 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
@@ -257,6 +258,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         domain_execution = self._get_node_execution(event.id)
         domain_execution.status = WorkflowNodeExecutionStatus.RETRY
         domain_execution.error = 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)
         _inspector_publish_node_changed(
@@ -356,6 +358,46 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         self._node_sequence += 1
         return self._node_sequence
 
+    def _append_retry_history(self, execution: WorkflowNodeExecution, event: NodeRunRetryEvent) -> None:
+        """Append a validated full attempt before repository truncation or offload."""
+        finished_at = naive_utc_now()
+        process_data = dict(execution.process_data or {})
+        raw_history = process_data.get(RETRY_HISTORY_PROCESS_DATA_KEY)
+        history = list(raw_history) if isinstance(raw_history, list) else []
+        projected_outputs = project_node_outputs_for_workflow_run(
+            node_type=execution.node_type,
+            inputs=event.node_run_result.inputs,
+            outputs=event.node_run_result.outputs,
+        )
+        attempt = WorkflowNodeRetryAttempt(
+            retry_index=event.retry_index,
+            inputs=event.node_run_result.inputs,
+            process_data=event.node_run_result.process_data,
+            outputs=projected_outputs,
+            error=event.node_run_result.error or event.error,
+            elapsed_time=max((finished_at - event.start_at).total_seconds(), 0.0),
+            execution_metadata={key.value: value for key, value in event.node_run_result.metadata.items()},
+            created_at=int(event.start_at.timestamp()),
+            finished_at=int(finished_at.timestamp()),
+        )
+        history.append(attempt.model_dump(mode="json"))
+        process_data[RETRY_HISTORY_PROCESS_DATA_KEY] = history
+        execution.process_data = process_data
+
+    @staticmethod
+    def _merge_retry_history(
+        existing_process_data: Mapping[str, Any] | None,
+        next_process_data: Mapping[str, Any] | None,
+    ) -> Mapping[str, Any] | None:
+        """Keep internal retry history while replacing node-specific Process Data."""
+        raw_history = (existing_process_data or {}).get(RETRY_HISTORY_PROCESS_DATA_KEY)
+        if not isinstance(raw_history, list) or not raw_history:
+            return next_process_data
+
+        merged_process_data = dict(next_process_data or {})
+        merged_process_data[RETRY_HISTORY_PROCESS_DATA_KEY] = raw_history
+        return merged_process_data
+
     def _populate_completion_statistics(self, execution: WorkflowExecution, *, update_finished: bool = True) -> None:
         if update_finished:
             execution.finished_at = naive_utc_now()
@@ -391,9 +433,10 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
                 inputs=node_result.inputs,
                 outputs=node_result.outputs,
             )
+            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=node_result.process_data,
+                process_data=process_data,
                 outputs=projected_outputs,
                 metadata=node_result.metadata,
             )
diff --git a/api/core/app/workflow/retry_history.py b/api/core/app/workflow/retry_history.py
new file mode 100644
index 00000000000..75060c55fd1
--- /dev/null
+++ b/api/core/app/workflow/retry_history.py
@@ -0,0 +1,24 @@
+"""Validated internal payloads for persisted workflow node retry history."""
+
+from collections.abc import Mapping
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+RETRY_HISTORY_PROCESS_DATA_KEY = "__dify_retry_history"
+
+
+class WorkflowNodeRetryAttempt(BaseModel):
+    """Complete data captured for one failed node attempt before a retry."""
+
+    retry_index: int = Field(gt=0)
+    inputs: Mapping[str, Any]
+    process_data: Mapping[str, Any]
+    outputs: Mapping[str, Any]
+    error: str
+    elapsed_time: float = Field(ge=0)
+    execution_metadata: Mapping[str, Any]
+    created_at: int
+    finished_at: int
+
+    model_config = ConfigDict(extra="forbid")
diff --git a/api/core/helper/ssrf_proxy.py b/api/core/helper/ssrf_proxy.py
index f2a7cac1bd4..86a7f9cfa75 100644
--- a/api/core/helper/ssrf_proxy.py
+++ b/api/core/helper/ssrf_proxy.py
@@ -200,7 +200,7 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
                         f"The URL may point to a private or local network address. "
                     )
 
-            if response.status_code not in STATUS_FORCELIST:
+            if response.status_code not in STATUS_FORCELIST or max_retries == 0:
                 return response
             else:
                 logger.warning(
diff --git a/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py b/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py
index 27dfb8b0cea..6a2223697e3 100644
--- a/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py
+++ b/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py
@@ -7,7 +7,7 @@ WorkflowNodeExecutionModel operations using Aliyun SLS LogStore.
 
 import logging
 import time
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Any, override
 
@@ -130,6 +130,11 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep
         logger.debug("LogstoreAPIWorkflowNodeExecutionRepository.__init__: initializing")
         self.logstore_client = AliyunLogStore()
 
+    @override
+    def load_full_process_data(self, execution: WorkflowNodeExecutionModel) -> Mapping[str, Any] | None:
+        """Return complete Process Data already stored inline by LogStore."""
+        return execution.process_data_dict
+
     @override
     def get_node_last_execution(
         self,
diff --git a/api/fields/workflow_run_fields.py b/api/fields/workflow_run_fields.py
index 53cdfa234f9..1cc0aa6491d 100644
--- a/api/fields/workflow_run_fields.py
+++ b/api/fields/workflow_run_fields.py
@@ -200,6 +200,7 @@ class WorkflowRunNodeExecutionResponse(ResponseModel):
     inputs_truncated: bool | None = None
     outputs_truncated: bool | None = None
     process_data_truncated: bool | None = None
+    retry_index: int | None = Field(default=None, exclude_if=lambda value: value is None)
 
     @field_validator("status", mode="before")
     @classmethod
diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md
index 76993e0cbec..2a4f2a404d6 100644
--- a/api/openapi/markdown/console-openapi.md
+++ b/api/openapi/markdown/console-openapi.md
@@ -23887,6 +23887,7 @@ Lifecycle state for an asynchronous archive download request.
 | predecessor_node_id | string |  | No |
 | process_data |  |  | No |
 | process_data_truncated | boolean |  | No |
+| retry_index | integer |  | No |
 | status | string |  | No |
 | title | string |  | No |
 
diff --git a/api/repositories/api_workflow_node_execution_repository.py b/api/repositories/api_workflow_node_execution_repository.py
index 3595ea33f07..bd53e9ec4f1 100644
--- a/api/repositories/api_workflow_node_execution_repository.py
+++ b/api/repositories/api_workflow_node_execution_repository.py
@@ -9,10 +9,10 @@ The service repository handles operations that require access to database-specif
 tenant_id, app_id, triggered_from, etc., which are not part of the core domain model.
 """
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Protocol
+from typing import Any, Protocol
 
 from sqlalchemy.orm import Session
 
@@ -62,6 +62,10 @@ class DifyAPIWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository, Pr
     - Supports cleanup and maintenance operations
     """
 
+    def load_full_process_data(self, execution: WorkflowNodeExecutionModel) -> Mapping[str, Any] | None:
+        """Return full Process Data, loading offloaded content when necessary."""
+        ...
+
     def get_node_last_execution(
         self,
         tenant_id: str,
diff --git a/api/repositories/sqlalchemy_api_workflow_node_execution_repository.py b/api/repositories/sqlalchemy_api_workflow_node_execution_repository.py
index 0d8f2c09aba..14c4716da67 100644
--- a/api/repositories/sqlalchemy_api_workflow_node_execution_repository.py
+++ b/api/repositories/sqlalchemy_api_workflow_node_execution_repository.py
@@ -6,14 +6,15 @@ using SQLAlchemy 2.0 style queries for WorkflowNodeExecutionModel operations.
 """
 
 import json
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from datetime import datetime
-from typing import Protocol, cast, override
+from typing import Any, Protocol, cast, override
 
 from sqlalchemy import asc, delete, desc, func, select
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session, sessionmaker
 
+from extensions.ext_storage import storage
 from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionOffload
 from repositories.api_workflow_node_execution_repository import (
@@ -65,6 +66,12 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
         """
         self._session_maker = session_maker
 
+    @override
+    def load_full_process_data(self, execution: WorkflowNodeExecutionModel) -> Mapping[str, Any] | None:
+        """Load Process Data from external storage when the execution was offloaded."""
+        with self._session_maker() as session:
+            return execution.load_full_process_data(session, storage)
+
     @override
     def get_node_last_execution(
         self,
diff --git a/api/services/rag_pipeline/rag_pipeline.py b/api/services/rag_pipeline/rag_pipeline.py
index 8bd3918eb15..d59ac19b593 100644
--- a/api/services/rag_pipeline/rag_pipeline.py
+++ b/api/services/rag_pipeline/rag_pipeline.py
@@ -29,7 +29,7 @@ from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasour
 from core.datasource.website_crawl.website_crawl_plugin import WebsiteCrawlDatasourcePlugin
 from core.helper import marketplace
 from core.rag.entities import DatasourceCompletedEvent, DatasourceErrorEvent, DatasourceProcessingEvent
-from core.repositories.factory import DifyCoreRepositoryFactory, OrderConfig
+from core.repositories.factory import DifyCoreRepositoryFactory
 from core.repositories.sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository
 from core.workflow.node_factory import LATEST_VERSION, get_node_type_classes_mapping
 from core.workflow.system_variables import (
@@ -82,6 +82,10 @@ from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError,
 from services.rag_pipeline.pipeline_template.pipeline_template_factory import PipelineTemplateRetrievalFactory
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
 from services.workflow_draft_variable_service import DraftVariableSaver, DraftVarLoader
+from services.workflow_node_execution_trace_service import (
+    WorkflowNodeExecutionTrace,
+    assemble_workflow_node_execution_traces,
+)
 from services.workflow_ref_service import WorkflowRef
 from services.workflow_restore import apply_published_workflow_snapshot_to_draft
 
@@ -1196,7 +1200,7 @@ class RagPipelineService:
         pipeline: Pipeline,
         run_id: str,
         user: Account | EndUser,
-    ) -> list[WorkflowNodeExecutionModel]:
+    ) -> list[WorkflowNodeExecutionTrace]:
         """
         Get workflow run node execution list
         """
@@ -1208,20 +1212,12 @@ class RagPipelineService:
         if not workflow_run:
             return []
 
-        # Use the repository to get the node execution
-        repository = SQLAlchemyWorkflowNodeExecutionRepository(
-            session_factory=db.engine, app_id=pipeline.id, user=user, triggered_from=None
-        )
-
-        # Use the repository to get the node executions with ordering
-        order_config = OrderConfig(order_by=["created_at"], order_direction="asc")
-        node_executions = repository.get_db_models_by_workflow_run(
+        node_executions = self._node_execution_service_repo.get_executions_by_workflow_run(
+            tenant_id=pipeline.tenant_id,
+            app_id=pipeline.id,
             workflow_run_id=run_id,
-            order_config=order_config,
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.RAG_PIPELINE_RUN,
         )
-
-        return list(node_executions)
+        return assemble_workflow_node_execution_traces(node_executions, self._node_execution_service_repo)
 
     @classmethod
     def publish_customized_pipeline_template(
diff --git a/api/services/snippet_service.py b/api/services/snippet_service.py
index 23b133bf481..c23895a4d60 100644
--- a/api/services/snippet_service.py
+++ b/api/services/snippet_service.py
@@ -37,6 +37,10 @@ from models.workflow import (
 from repositories.factory import DifyAPIRepositoryFactory
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
 from services.tag_service import TagService
+from services.workflow_node_execution_trace_service import (
+    WorkflowNodeExecutionTrace,
+    assemble_workflow_node_execution_traces,
+)
 from services.workflow_restore import apply_published_workflow_snapshot_to_draft
 
 logger = logging.getLogger(__name__)
@@ -896,13 +900,13 @@ class SnippetService:
         *,
         snippet: CustomizedSnippet,
         run_id: str,
-    ) -> Sequence[WorkflowNodeExecutionModel]:
+    ) -> list[WorkflowNodeExecutionTrace]:
         """
         Get workflow run node execution list.
 
         :param snippet: CustomizedSnippet instance
         :param run_id: Workflow run ID
-        :return: List of WorkflowNodeExecutionModel
+        :return: Public terminal and retry trace records
         """
         workflow_run = self.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
         if not workflow_run:
@@ -914,7 +918,7 @@ class SnippetService:
             workflow_run_id=workflow_run.id,
         )
 
-        return node_executions
+        return assemble_workflow_node_execution_traces(node_executions, self._node_execution_service_repo)
 
     # --- Node Execution Operations ---
 
diff --git a/api/services/workflow_node_execution_trace_service.py b/api/services/workflow_node_execution_trace_service.py
new file mode 100644
index 00000000000..4d028318c81
--- /dev/null
+++ b/api/services/workflow_node_execution_trace_service.py
@@ -0,0 +1,217 @@
+"""Expand persisted node executions into public terminal and retry trace records."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping, Sequence
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, ValidationError
+
+from core.app.workflow.retry_history import RETRY_HISTORY_PROCESS_DATA_KEY, WorkflowNodeRetryAttempt
+from libs.helper import to_timestamp
+from models.workflow import WorkflowNodeExecutionModel
+from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowNodeExecutionRepository
+from services.variable_truncator import VariableTruncator
+
+logger = logging.getLogger(__name__)
+
+
+class WorkflowNodeExecutionTrace(BaseModel):
+    """Read model preserving node-execution identity for terminal and virtual retry traces."""
+
+    id: str
+    tenant_id: str
+    app_id: str
+    workflow_id: str
+    triggered_from: str
+    workflow_run_id: str | None = None
+    index: int | None = None
+    predecessor_node_id: str | None = None
+    node_execution_id: str | None = None
+    node_id: str | None = None
+    node_type: str | None = None
+    title: str | None = None
+    inputs: Mapping[str, Any] | None = None
+    process_data: Mapping[str, Any] | None = None
+    outputs: Mapping[str, Any] | None = None
+    status: str | None = None
+    error: str | None = None
+    elapsed_time: float | None = None
+    execution_metadata: Mapping[str, Any] | None = None
+    extras: Any = None
+    created_at: int | None = None
+    created_by_role: str | None = None
+    created_by: str
+    created_by_account: Any = None
+    created_by_end_user: Any = None
+    finished_at: int | None = None
+    inputs_truncated: bool = False
+    outputs_truncated: bool = False
+    process_data_truncated: bool = False
+    retry_index: int | None = None
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+
+def assemble_workflow_node_execution_traces(
+    executions: Sequence[WorkflowNodeExecutionModel],
+    repository: DifyAPIWorkflowNodeExecutionRepository,
+) -> list[WorkflowNodeExecutionTrace]:
+    """Expand valid persisted retry attempts before each terminal execution."""
+    traces: list[WorkflowNodeExecutionTrace] = []
+    for execution in executions:
+        traces.extend(_expand_execution(execution, repository))
+    return traces
+
+
+def _expand_execution(
+    execution: WorkflowNodeExecutionModel,
+    repository: DifyAPIWorkflowNodeExecutionRepository,
+) -> list[WorkflowNodeExecutionTrace]:
+    full_process_data = _load_full_process_data(execution, repository)
+    retry_attempts = _parse_retry_attempts(execution, full_process_data)
+    terminal_metadata = execution.execution_metadata_dict
+    traces = [_retry_trace(execution, attempt, terminal_metadata) for attempt in retry_attempts]
+    traces.append(_terminal_trace(execution))
+    return traces
+
+
+def _load_full_process_data(
+    execution: WorkflowNodeExecutionModel,
+    repository: DifyAPIWorkflowNodeExecutionRepository,
+) -> Mapping[str, Any] | None:
+    try:
+        return repository.load_full_process_data(execution)
+    except Exception:
+        logger.warning(
+            "Failed to load full Process Data for workflow run %s node execution %s; using inline preview",
+            execution.workflow_run_id,
+            execution.id,
+            exc_info=True,
+        )
+        return execution.process_data_dict
+
+
+def _parse_retry_attempts(
+    execution: WorkflowNodeExecutionModel,
+    process_data: Mapping[str, Any] | None,
+) -> list[WorkflowNodeRetryAttempt]:
+    raw_history = (process_data or {}).get(RETRY_HISTORY_PROCESS_DATA_KEY)
+    if not isinstance(raw_history, list):
+        return []
+
+    attempts: list[WorkflowNodeRetryAttempt] = []
+    seen_indices: set[int] = set()
+    for raw_attempt in raw_history:
+        try:
+            attempt = WorkflowNodeRetryAttempt.model_validate(raw_attempt)
+        except ValidationError:
+            logger.warning(
+                "Skipping malformed retry history for workflow run %s node execution %s",
+                execution.workflow_run_id,
+                execution.id,
+                exc_info=True,
+            )
+            continue
+        if attempt.retry_index in seen_indices:
+            logger.warning(
+                "Skipping duplicate retry index %s for workflow run %s node execution %s",
+                attempt.retry_index,
+                execution.workflow_run_id,
+                execution.id,
+            )
+            continue
+        seen_indices.add(attempt.retry_index)
+        attempts.append(attempt)
+
+    return sorted(attempts, key=lambda attempt: attempt.retry_index)
+
+
+def _retry_trace(
+    execution: WorkflowNodeExecutionModel,
+    attempt: WorkflowNodeRetryAttempt,
+    terminal_metadata: Mapping[str, Any],
+) -> WorkflowNodeExecutionTrace:
+    truncator = VariableTruncator.default()
+    inputs, inputs_truncated = truncator.truncate_variable_mapping(attempt.inputs)
+    process_data, process_data_truncated = truncator.truncate_variable_mapping(attempt.process_data)
+    outputs, outputs_truncated = truncator.truncate_variable_mapping(attempt.outputs)
+    execution_metadata = {**terminal_metadata, **attempt.execution_metadata}
+    return WorkflowNodeExecutionTrace(
+        id=f"{execution.id}:retry:{attempt.retry_index}",
+        tenant_id=execution.tenant_id,
+        app_id=execution.app_id,
+        workflow_id=execution.workflow_id,
+        triggered_from=_enum_value(execution.triggered_from) or "",
+        workflow_run_id=execution.workflow_run_id,
+        index=execution.index,
+        predecessor_node_id=execution.predecessor_node_id,
+        node_execution_id=execution.node_execution_id,
+        node_id=execution.node_id,
+        node_type=execution.node_type,
+        title=execution.title,
+        inputs=inputs,
+        process_data=process_data,
+        outputs=outputs,
+        status="retry",
+        error=attempt.error,
+        elapsed_time=attempt.elapsed_time,
+        execution_metadata=execution_metadata,
+        extras=execution.extras,
+        created_at=attempt.created_at,
+        created_by_role=_enum_value(execution.created_by_role),
+        created_by=execution.created_by,
+        created_by_account=execution.created_by_account,
+        created_by_end_user=execution.created_by_end_user,
+        finished_at=attempt.finished_at,
+        inputs_truncated=inputs_truncated,
+        outputs_truncated=outputs_truncated,
+        process_data_truncated=process_data_truncated,
+        retry_index=attempt.retry_index,
+    )
+
+
+def _terminal_trace(execution: WorkflowNodeExecutionModel) -> WorkflowNodeExecutionTrace:
+    process_data = execution.process_data_dict
+    if process_data is not None and RETRY_HISTORY_PROCESS_DATA_KEY in process_data:
+        process_data = dict(process_data)
+        process_data.pop(RETRY_HISTORY_PROCESS_DATA_KEY, None)
+    return WorkflowNodeExecutionTrace(
+        id=execution.id,
+        tenant_id=execution.tenant_id,
+        app_id=execution.app_id,
+        workflow_id=execution.workflow_id,
+        triggered_from=_enum_value(execution.triggered_from) or "",
+        workflow_run_id=execution.workflow_run_id,
+        index=execution.index,
+        predecessor_node_id=execution.predecessor_node_id,
+        node_execution_id=execution.node_execution_id,
+        node_id=execution.node_id,
+        node_type=execution.node_type,
+        title=execution.title,
+        inputs=execution.inputs_dict,
+        process_data=process_data,
+        outputs=execution.outputs_dict,
+        status=_enum_value(execution.status),
+        error=execution.error,
+        elapsed_time=execution.elapsed_time,
+        execution_metadata=execution.execution_metadata_dict,
+        extras=execution.extras,
+        created_at=to_timestamp(execution.created_at),
+        created_by_role=_enum_value(execution.created_by_role),
+        created_by=execution.created_by,
+        created_by_account=execution.created_by_account,
+        created_by_end_user=execution.created_by_end_user,
+        finished_at=to_timestamp(execution.finished_at),
+        inputs_truncated=execution.inputs_truncated,
+        outputs_truncated=execution.outputs_truncated,
+        process_data_truncated=execution.process_data_truncated,
+    )
+
+
+def _enum_value(value: Enum | str | None) -> str | None:
+    if value is None or isinstance(value, str):
+        return value
+    return str(value.value)
diff --git a/api/services/workflow_run_service.py b/api/services/workflow_run_service.py
index 1ff0b59514f..0c5c0876b19 100644
--- a/api/services/workflow_run_service.py
+++ b/api/services/workflow_run_service.py
@@ -1,5 +1,4 @@
 import threading
-from collections.abc import Sequence
 from typing import TypedDict
 
 from sqlalchemy import Engine, select
@@ -13,12 +12,15 @@ from models import (
     App,
     EndUser,
     Message,
-    WorkflowNodeExecutionModel,
     WorkflowRun,
     WorkflowRunTriggeredFrom,
 )
 from repositories.api_workflow_run_repository import APIWorkflowRunRepository
 from repositories.factory import DifyAPIRepositoryFactory
+from services.workflow_node_execution_trace_service import (
+    WorkflowNodeExecutionTrace,
+    assemble_workflow_node_execution_traces,
+)
 
 
 class WorkflowRunListArgs(TypedDict, total=False):
@@ -174,7 +176,7 @@ class WorkflowRunService:
         app_model: App,
         run_id: str,
         user: Account | EndUser,
-    ) -> Sequence[WorkflowNodeExecutionModel]:
+    ) -> list[WorkflowNodeExecutionTrace]:
         """
         Get workflow run node execution list
         """
@@ -191,8 +193,9 @@ class WorkflowRunService:
         if tenant_id is None:
             raise ValueError("User tenant_id cannot be None")
 
-        return self._node_execution_service_repo.get_executions_by_workflow_run(
+        node_executions = self._node_execution_service_repo.get_executions_by_workflow_run(
             tenant_id=tenant_id,
             app_id=app_model.id,
             workflow_run_id=run_id,
         )
+        return assemble_workflow_node_execution_traces(node_executions, self._node_execution_service_repo)
diff --git a/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_api_workflow_node_execution_repository.py b/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_api_workflow_node_execution_repository.py
index 641399c7f94..7fbffd48feb 100644
--- a/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_api_workflow_node_execution_repository.py
+++ b/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_api_workflow_node_execution_repository.py
@@ -5,6 +5,7 @@ from __future__ import annotations
 from datetime import timedelta
 from uuid import uuid4
 
+import pytest
 from sqlalchemy import Engine, delete
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -61,6 +62,34 @@ def _create_node_execution(
 
 
 class TestDifyAPISQLAlchemyWorkflowNodeExecutionRepository:
+    def test_load_full_process_data_returns_inline_mapping(self, db_session_with_containers: Session) -> None:
+        execution = WorkflowNodeExecutionModel(
+            process_data='{"request": "inline"}',
+            offload_data=[],
+        )
+        engine = db_session_with_containers.get_bind()
+        assert isinstance(engine, Engine)
+        repository = DifyAPISQLAlchemyWorkflowNodeExecutionRepository(sessionmaker(bind=engine, expire_on_commit=False))
+
+        assert repository.load_full_process_data(execution) == {"request": "inline"}
+
+    def test_load_full_process_data_reads_offload(
+        self,
+        db_session_with_containers: Session,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        execution = WorkflowNodeExecutionModel(process_data="{}", offload_data=[])
+        monkeypatch.setattr(
+            WorkflowNodeExecutionModel,
+            "load_full_process_data",
+            lambda _self, _session, _storage: {"__dify_retry_history": [{"retry_index": 1}]},
+        )
+        engine = db_session_with_containers.get_bind()
+        assert isinstance(engine, Engine)
+        repository = DifyAPISQLAlchemyWorkflowNodeExecutionRepository(sessionmaker(bind=engine, expire_on_commit=False))
+
+        assert repository.load_full_process_data(execution) == {"__dify_retry_history": [{"retry_index": 1}]}
+
     def test_get_executions_by_workflow_run_keeps_paused_records(self, db_session_with_containers: Session) -> None:
         tenant_id = str(uuid4())
         app_id = str(uuid4())
diff --git a/api/tests/unit_tests/core/app/workflow/layers/test_persistence_inspector_publish.py b/api/tests/unit_tests/core/app/workflow/layers/test_persistence_inspector_publish.py
index da8d6c5089d..e23f4a81814 100644
--- a/api/tests/unit_tests/core/app/workflow/layers/test_persistence_inspector_publish.py
+++ b/api/tests/unit_tests/core/app/workflow/layers/test_persistence_inspector_publish.py
@@ -150,8 +150,9 @@ def test_node_started_publishes_running(layer, capture_publishes):
     assert capture_publishes["node"] == [{"workflow_run_id": "run-1", "node_id": "agent-1", "status": "running"}]
 
 
-def test_node_retry_publishes_retry(layer, capture_publishes):
+def test_node_retry_publishes_retry(layer, capture_publishes, monkeypatch: pytest.MonkeyPatch):
     _seed_node_execution(layer, exec_id="exec-1", node_id="agent-1")
+    monkeypatch.setattr(layer, "_append_retry_history", lambda *args: None)
     event = MagicMock(id="exec-1", error="rate limit")
     layer._handle_node_retry(event)
     assert capture_publishes["node"] == [{"workflow_run_id": "run-1", "node_id": "agent-1", "status": "retry"}]
diff --git a/api/tests/unit_tests/core/app/workflow/test_persistence_layer.py b/api/tests/unit_tests/core/app/workflow/test_persistence_layer.py
index c8efaec33dc..31b9bb28940 100644
--- a/api/tests/unit_tests/core/app/workflow/test_persistence_layer.py
+++ b/api/tests/unit_tests/core/app/workflow/test_persistence_layer.py
@@ -14,6 +14,7 @@ from graphon.entities.pause_reason import SchedulingPause
 from graphon.enums import (
     BuiltinNodeTypes,
     WorkflowExecutionStatus,
+    WorkflowNodeExecutionMetadataKey,
     WorkflowNodeExecutionStatus,
     WorkflowType,
 )
@@ -330,6 +331,116 @@ class TestWorkflowPersistenceLayer:
         layer._handle_node_retry(retry_event)
         assert node_repo.saved_exec_data
 
+    def test_retry_history_is_preserved_after_node_succeeds(self):
+        layer, _, node_repo, _ = _make_layer()
+        layer._handle_graph_run_started()
+        started_at = _naive_utc_now()
+        layer._handle_node_started(
+            NodeRunStartedEvent(
+                id="exec",
+                node_id="node",
+                node_type=BuiltinNodeTypes.HTTP_REQUEST,
+                node_title="HTTP",
+                start_at=started_at,
+            )
+        )
+
+        for retry_index in (1, 2):
+            layer._handle_node_retry(
+                NodeRunRetryEvent(
+                    id="exec",
+                    node_id="node",
+                    node_type=BuiltinNodeTypes.HTTP_REQUEST,
+                    node_title="HTTP",
+                    start_at=started_at,
+                    error=f"attempt {retry_index} failed",
+                    retry_index=retry_index,
+                    node_run_result=NodeRunResult(
+                        inputs={"attempt": retry_index},
+                        process_data={"request": f"attempt-{retry_index}"},
+                        outputs={"status_code": 500, "body": f"failure-{retry_index}"},
+                        metadata={WorkflowNodeExecutionMetadataKey.ITERATION_ID: "iteration-1"},
+                    ),
+                )
+            )
+
+        layer._handle_node_succeeded(
+            NodeRunSucceededEvent(
+                id="exec",
+                node_id="node",
+                node_type=BuiltinNodeTypes.HTTP_REQUEST,
+                start_at=started_at,
+                node_run_result=NodeRunResult(
+                    inputs={"attempt": 3},
+                    process_data={"request": "successful-attempt"},
+                    outputs={"status_code": 200, "body": "ok"},
+                    metadata={},
+                ),
+            )
+        )
+
+        saved_execution = node_repo.saved_exec_data[-1]
+        assert saved_execution.status == WorkflowNodeExecutionStatus.SUCCEEDED
+        assert saved_execution.process_data["request"] == "successful-attempt"
+        retry_history = saved_execution.process_data["__dify_retry_history"]
+        assert [attempt["retry_index"] for attempt in retry_history] == [1, 2]
+        assert retry_history[0]["inputs"] == {"attempt": 1}
+        assert retry_history[0]["process_data"] == {"request": "attempt-1"}
+        assert retry_history[0]["outputs"] == {"status_code": 500, "body": "failure-1"}
+        assert retry_history[0]["error"] == "attempt 1 failed"
+        assert retry_history[0]["execution_metadata"] == {"iteration_id": "iteration-1"}
+
+    @pytest.mark.parametrize(
+        ("event_type", "expected_status"),
+        [
+            (NodeRunFailedEvent, WorkflowNodeExecutionStatus.FAILED),
+            (NodeRunExceptionEvent, WorkflowNodeExecutionStatus.EXCEPTION),
+        ],
+    )
+    def test_retry_history_is_preserved_after_terminal_error(self, event_type, expected_status):
+        layer, _, node_repo, _ = _make_layer()
+        layer._handle_graph_run_started()
+        started_at = _naive_utc_now()
+        layer._handle_node_started(
+            NodeRunStartedEvent(
+                id="exec",
+                node_id="node",
+                node_type=BuiltinNodeTypes.LLM,
+                node_title="LLM",
+                start_at=started_at,
+            )
+        )
+        layer._handle_node_retry(
+            NodeRunRetryEvent(
+                id="exec",
+                node_id="node",
+                node_type=BuiltinNodeTypes.LLM,
+                node_title="LLM",
+                start_at=started_at,
+                error="retry failed",
+                retry_index=1,
+                node_run_result=NodeRunResult(outputs={"attempt": 1}),
+            )
+        )
+
+        terminal_event = event_type(
+            id="exec",
+            node_id="node",
+            node_type=BuiltinNodeTypes.LLM,
+            start_at=started_at,
+            error="terminal failure",
+            node_run_result=NodeRunResult(process_data={"terminal": True}),
+        )
+        if expected_status == WorkflowNodeExecutionStatus.FAILED:
+            layer._handle_node_failed(terminal_event)
+        else:
+            layer._handle_node_exception(terminal_event)
+
+        saved_execution = node_repo.saved_exec_data[-1]
+        assert saved_execution.status == expected_status
+        assert saved_execution.process_data["terminal"] is True
+        assert saved_execution.process_data["__dify_retry_history"][0]["retry_index"] == 1
+
     def test_handle_node_result_events_update_execution(self):
         layer, _, node_repo, _ = _make_layer()
         layer._handle_graph_run_started()
diff --git a/api/tests/unit_tests/core/helper/test_ssrf_proxy.py b/api/tests/unit_tests/core/helper/test_ssrf_proxy.py
index 0659765da40..07ce9813e00 100644
--- a/api/tests/unit_tests/core/helper/test_ssrf_proxy.py
+++ b/api/tests/unit_tests/core/helper/test_ssrf_proxy.py
@@ -42,6 +42,20 @@ def test_retry_exceed_max_retries(mock_get_client):
     assert str(e.value) == f"Reached maximum retries ({SSRF_DEFAULT_MAX_RETRIES - 1}) for URL http://example.com"
 
 
+@patch("core.helper.ssrf_proxy._get_ssrf_client", autospec=True)
+def test_force_list_response_returns_when_retries_disabled(mock_get_client):
+    mock_client = MagicMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 500
+    mock_client.request.return_value = mock_response
+    mock_get_client.return_value = mock_client
+
+    response = make_request("GET", "http://example.com", max_retries=0)
+
+    assert response is mock_response
+    mock_client.request.assert_called_once()
+
+
 def test_build_ssrf_client_passes_ssl_verify_to_proxy_mount_transports():
     mock_client = MagicMock()
     http_transport = MagicMock()
diff --git a/api/tests/unit_tests/extensions/logstore/repositories/__init__.py b/api/tests/unit_tests/extensions/logstore/repositories/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/api/tests/unit_tests/extensions/logstore/repositories/test_logstore_api_workflow_node_execution_repository.py b/api/tests/unit_tests/extensions/logstore/repositories/test_logstore_api_workflow_node_execution_repository.py
new file mode 100644
index 00000000000..913e8ea6722
--- /dev/null
+++ b/api/tests/unit_tests/extensions/logstore/repositories/test_logstore_api_workflow_node_execution_repository.py
@@ -0,0 +1,15 @@
+from unittest.mock import patch
+
+from extensions.logstore.repositories.logstore_api_workflow_node_execution_repository import (
+    LogstoreAPIWorkflowNodeExecutionRepository,
+)
+from models.workflow import WorkflowNodeExecutionModel
+
+
+def test_load_full_process_data_returns_logstore_mapping() -> None:
+    with patch("extensions.logstore.repositories.logstore_api_workflow_node_execution_repository.AliyunLogStore"):
+        repository = LogstoreAPIWorkflowNodeExecutionRepository(session_maker=None)
+    execution = WorkflowNodeExecutionModel()
+    execution.process_data = '{"__dify_retry_history": [{"retry_index": 1}]}'
+
+    assert repository.load_full_process_data(execution) == {"__dify_retry_history": [{"retry_index": 1}]}
diff --git a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_service.py b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_service.py
index 0d74b3abf9c..ba4c8d6f540 100644
--- a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_service.py
+++ b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_service.py
@@ -1347,22 +1347,33 @@ def test_get_rag_pipeline_workflow_run_node_executions_empty_when_run_missing(
     assert result == []
 
 
-def test_get_rag_pipeline_workflow_run_node_executions_returns_sorted_executions(
+def test_get_rag_pipeline_workflow_run_node_executions_assembles_configured_repository_results(
     mocker: MockerFixture, rag_pipeline_service: RagPipelineServiceTestContext
 ) -> None:
     pipeline = _make_pipeline()
     mocker.patch.object(
         rag_pipeline_service.service, "get_rag_pipeline_workflow_run", return_value=SimpleNamespace(id="run-1")
     )
-    repo = mocker.Mock()
-    repo.get_db_models_by_workflow_run.return_value = ["n1", "n2"]
-    mocker.patch("services.rag_pipeline.rag_pipeline.SQLAlchemyWorkflowNodeExecutionRepository", return_value=repo)
+    node_repo = rag_pipeline_service.service._node_execution_service_repo
+    expected_executions = ["n1", "n2"]
+    expected_traces = ["retry-n1", "n1", "n2"]
+    node_repo.get_executions_by_workflow_run = mocker.Mock(return_value=expected_executions)
+    mock_assemble = mocker.patch(
+        "services.rag_pipeline.rag_pipeline.assemble_workflow_node_execution_traces",
+        return_value=expected_traces,
+    )
 
     result = rag_pipeline_service.service.get_rag_pipeline_workflow_run_node_executions(
         pipeline=pipeline, run_id="run-1", user=_make_account()
     )
 
-    assert result == ["n1", "n2"]
+    assert result == expected_traces
+    node_repo.get_executions_by_workflow_run.assert_called_once_with(
+        tenant_id=pipeline.tenant_id,
+        app_id=pipeline.id,
+        workflow_run_id="run-1",
+    )
+    mock_assemble.assert_called_once_with(expected_executions, node_repo)
 
 
 def test_get_recommended_plugins_returns_empty_when_no_active_plugins(
diff --git a/api/tests/unit_tests/services/test_snippet_service.py b/api/tests/unit_tests/services/test_snippet_service.py
index 731d09017ed..b3c6cdcbf6c 100644
--- a/api/tests/unit_tests/services/test_snippet_service.py
+++ b/api/tests/unit_tests/services/test_snippet_service.py
@@ -655,7 +655,7 @@ def test_delete_archived_workflow_run_files_removes_prefixed_objects(monkeypatch
     archive_storage.delete_object.assert_called_once_with("tenant-1/app_id=snippet-1/run.json")
 
 
-def test_workflow_run_queries_delegate_to_repositories() -> None:
+def test_workflow_run_queries_delegate_to_repositories(monkeypatch: pytest.MonkeyPatch) -> None:
     service = SnippetService.__new__(SnippetService)
     workflow_run_repo = SimpleNamespace(
         get_paginated_workflow_runs=Mock(return_value=SimpleNamespace(data=[])),
@@ -668,12 +668,13 @@ def test_workflow_run_queries_delegate_to_repositories() -> None:
     service._workflow_run_repo = workflow_run_repo
     service._node_execution_service_repo = node_execution_repo
     snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+    expected_traces = [SimpleNamespace(id="node-execution-1:retry:1"), SimpleNamespace(id="node-execution-1")]
+    mock_assemble = Mock(return_value=expected_traces)
+    monkeypatch.setattr("services.snippet_service.assemble_workflow_node_execution_traces", mock_assemble)
 
     assert service.get_snippet_workflow_runs(snippet=snippet, args={"limit": "5", "last_id": "run-0"}).data == []
     assert service.get_snippet_workflow_run(snippet=snippet, run_id="run-1").id == "run-1"
-    assert service.get_snippet_workflow_run_node_executions(snippet=snippet, run_id="run-1")[0].id == (
-        "node-execution-1"
-    )
+    assert service.get_snippet_workflow_run_node_executions(snippet=snippet, run_id="run-1") == expected_traces
     assert (
         service.get_snippet_node_last_run(
             snippet=snippet,
@@ -693,6 +694,9 @@ def test_workflow_run_queries_delegate_to_repositories() -> None:
         app_id="snippet-1",
         workflow_run_id="run-1",
     )
+    mock_assemble.assert_called_once_with(
+        node_execution_repo.get_executions_by_workflow_run.return_value, node_execution_repo
+    )
     node_execution_repo.get_node_last_execution.assert_called_once_with(
         tenant_id="tenant-1",
         app_id="snippet-1",
diff --git a/api/tests/unit_tests/services/test_workflow_node_execution_trace_service.py b/api/tests/unit_tests/services/test_workflow_node_execution_trace_service.py
new file mode 100644
index 00000000000..fd540734ccf
--- /dev/null
+++ b/api/tests/unit_tests/services/test_workflow_node_execution_trace_service.py
@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from typing import cast
+from unittest.mock import MagicMock
+
+from core.app.workflow.retry_history import RETRY_HISTORY_PROCESS_DATA_KEY, WorkflowNodeRetryAttempt
+from fields.workflow_run_fields import WorkflowRunNodeExecutionListResponse
+from graphon.enums import WorkflowNodeExecutionStatus
+from models.enums import CreatorUserRole
+from models.workflow import WorkflowNodeExecutionModel
+from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowNodeExecutionRepository
+from services.workflow_node_execution_trace_service import assemble_workflow_node_execution_traces
+
+
+def _retry_attempt(retry_index: int, **overrides: object) -> dict[str, object]:
+    values = {
+        "retry_index": retry_index,
+        "inputs": {"attempt": retry_index},
+        "process_data": {"request": f"attempt-{retry_index}"},
+        "outputs": {"status_code": 500, "body": f"failure-{retry_index}"},
+        "error": f"attempt {retry_index} failed",
+        "elapsed_time": float(retry_index),
+        "execution_metadata": {},
+        "created_at": 1_700_000_000 + retry_index,
+        "finished_at": 1_700_000_010 + retry_index,
+    }
+    values.update(overrides)
+    return WorkflowNodeRetryAttempt.model_validate(values).model_dump(mode="json")
+
+
+def _execution(process_data: dict[str, object]) -> WorkflowNodeExecutionModel:
+    return cast(
+        WorkflowNodeExecutionModel,
+        SimpleNamespace(
+            id="exec-1",
+            tenant_id="tenant-1",
+            app_id="app-1",
+            workflow_id="workflow-1",
+            triggered_from="workflow-run",
+            workflow_run_id="run-1",
+            index=3,
+            predecessor_node_id="previous-node",
+            node_execution_id="node-execution-1",
+            node_id="node-1",
+            node_type="http-request",
+            title="HTTP Request",
+            inputs_dict={"attempt": 3},
+            process_data_dict=process_data,
+            outputs_dict={"status_code": 200, "body": "ok"},
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            error=None,
+            elapsed_time=3.5,
+            execution_metadata_dict={"iteration_id": "iteration-1"},
+            extras={},
+            created_at=datetime(2023, 11, 14, tzinfo=UTC),
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by="account-1",
+            created_by_account=None,
+            created_by_end_user=None,
+            finished_at=datetime(2023, 11, 14, 0, 0, 4, tzinfo=UTC),
+            inputs_truncated=False,
+            outputs_truncated=False,
+            process_data_truncated=False,
+        ),
+    )
+
+
+def _repository(full_process_data: dict[str, object] | None) -> DifyAPIWorkflowNodeExecutionRepository:
+    repository = MagicMock()
+    repository.load_full_process_data.return_value = full_process_data
+    return cast(DifyAPIWorkflowNodeExecutionRepository, repository)
+
+
+def test_assemble_expands_retry_history_before_terminal_trace() -> None:
+    process_data = {
+        "request": "successful-attempt",
+        RETRY_HISTORY_PROCESS_DATA_KEY: [_retry_attempt(2), _retry_attempt(1)],
+    }
+    execution = _execution(process_data)
+    repository = _repository(process_data)
+
+    traces = assemble_workflow_node_execution_traces([execution], repository)
+
+    assert [trace.id for trace in traces] == ["exec-1:retry:1", "exec-1:retry:2", "exec-1"]
+    assert [trace.status for trace in traces] == ["retry", "retry", "succeeded"]
+    assert traces[0].retry_index == 1
+    assert traces[0].outputs == {"status_code": 500, "body": "failure-1"}
+    assert traces[0].execution_metadata == {"iteration_id": "iteration-1"}
+    assert RETRY_HISTORY_PROCESS_DATA_KEY not in traces[-1].process_data
+    for trace in traces:
+        assert trace.tenant_id == "tenant-1"
+        assert trace.app_id == "app-1"
+        assert trace.workflow_id == "workflow-1"
+        assert trace.triggered_from == "workflow-run"
+        assert trace.workflow_run_id == "run-1"
+        assert trace.node_execution_id == "node-execution-1"
+        assert trace.created_by == "account-1"
+    repository.load_full_process_data.assert_called_once_with(execution)
+
+
+def test_assemble_keeps_old_execution_without_retry_history() -> None:
+    process_data = {"request": "terminal"}
+    execution = _execution(process_data)
+
+    traces = assemble_workflow_node_execution_traces([execution], _repository(process_data))
+
+    assert len(traces) == 1
+    assert traces[0].id == "exec-1"
+    assert traces[0].process_data == process_data
+
+
+def test_assemble_skips_malformed_and_duplicate_retry_attempts() -> None:
+    process_data = {
+        RETRY_HISTORY_PROCESS_DATA_KEY: [
+            _retry_attempt(2),
+            {"retry_index": 0},
+            _retry_attempt(1),
+            _retry_attempt(1, error="duplicate"),
+        ]
+    }
+
+    traces = assemble_workflow_node_execution_traces([_execution(process_data)], _repository(process_data))
+
+    assert [trace.retry_index for trace in traces[:-1]] == [1, 2]
+    assert traces[0].error == "attempt 1 failed"
+
+
+def test_assemble_truncates_retry_attempt_fields() -> None:
+    process_data = {RETRY_HISTORY_PROCESS_DATA_KEY: [_retry_attempt(1, outputs={"body": "x" * 2_000_000})]}
+
+    traces = assemble_workflow_node_execution_traces([_execution(process_data)], _repository(process_data))
+
+    assert traces[0].outputs_truncated is True
+    assert traces[-1].id == "exec-1"
+
+
+def test_assemble_falls_back_to_inline_process_data_when_loader_fails() -> None:
+    process_data = {RETRY_HISTORY_PROCESS_DATA_KEY: [_retry_attempt(1)]}
+    execution = _execution(process_data)
+    repository = _repository(None)
+    repository.load_full_process_data.side_effect = OSError("storage unavailable")
+
+    traces = assemble_workflow_node_execution_traces([execution], repository)
+
+    assert [trace.id for trace in traces] == ["exec-1:retry:1", "exec-1"]
+
+
+def test_virtual_trace_validates_through_node_execution_response() -> None:
+    process_data = {RETRY_HISTORY_PROCESS_DATA_KEY: [_retry_attempt(1)]}
+    traces = assemble_workflow_node_execution_traces([_execution(process_data)], _repository(process_data))
+
+    response = WorkflowRunNodeExecutionListResponse.model_validate({"data": traces}, from_attributes=True)
+
+    assert response.data[0].retry_index == 1
+    assert response.model_dump(mode="json")["data"][0]["retry_index"] == 1
diff --git a/api/tests/unit_tests/services/test_workflow_run_service.py b/api/tests/unit_tests/services/test_workflow_run_service.py
index 2c69a742f15..c21b764a9cd 100644
--- a/api/tests/unit_tests/services/test_workflow_run_service.py
+++ b/api/tests/unit_tests/services/test_workflow_run_service.py
@@ -247,17 +247,21 @@ class TestWorkflowRunServiceQueries:
         monkeypatch.setattr(service_module, "EndUser", FakeEndUser)
         user = cast(EndUser, FakeEndUser(tenant_id="tenant-end-user"))
         app_model = _app_model(id="app-1")
-        expected = [SimpleNamespace(id="exec-1")]
-        node_repo.get_executions_by_workflow_run.return_value = expected
+        expected_executions = [SimpleNamespace(id="exec-1")]
+        expected_traces = [SimpleNamespace(id="exec-1:retry:1")]
+        node_repo.get_executions_by_workflow_run.return_value = expected_executions
+        mock_assemble = MagicMock(return_value=expected_traces)
+        monkeypatch.setattr(service_module, "assemble_workflow_node_execution_traces", mock_assemble)
 
         result = service.get_workflow_run_node_executions(app_model=app_model, run_id="run-1", user=user)
 
-        assert result == expected
+        assert result == expected_traces
         node_repo.get_executions_by_workflow_run.assert_called_once_with(
             tenant_id="tenant-end-user",
             app_id="app-1",
             workflow_run_id="run-1",
         )
+        mock_assemble.assert_called_once_with(expected_executions, node_repo)
 
     def test_get_workflow_run_node_executions_should_use_account_current_tenant_id(
         self,
@@ -269,17 +273,21 @@ class TestWorkflowRunServiceQueries:
         monkeypatch.setattr(service, "get_workflow_run", MagicMock(return_value=SimpleNamespace(id="run-1")))
         app_model = _app_model(id="app-1")
         user = _account(current_tenant_id="tenant-account")
-        expected = [SimpleNamespace(id="exec-1"), SimpleNamespace(id="exec-2")]
-        node_repo.get_executions_by_workflow_run.return_value = expected
+        expected_executions = [SimpleNamespace(id="exec-1"), SimpleNamespace(id="exec-2")]
+        expected_traces = [SimpleNamespace(id="exec-1:retry:1"), SimpleNamespace(id="exec-1")]
+        node_repo.get_executions_by_workflow_run.return_value = expected_executions
+        mock_assemble = MagicMock(return_value=expected_traces)
+        monkeypatch.setattr(service_module, "assemble_workflow_node_execution_traces", mock_assemble)
 
         result = service.get_workflow_run_node_executions(app_model=app_model, run_id="run-1", user=user)
 
-        assert result == expected
+        assert result == expected_traces
         node_repo.get_executions_by_workflow_run.assert_called_once_with(
             tenant_id="tenant-account",
             app_id="app-1",
             workflow_run_id="run-1",
         )
+        mock_assemble.assert_called_once_with(expected_executions, node_repo)
 
     def test_get_workflow_run_node_executions_should_raise_when_resolved_tenant_id_is_none(
         self,
diff --git a/oxlint-suppressions.json b/oxlint-suppressions.json
index d11268cad75..f47db0b2adb 100644
--- a/oxlint-suppressions.json
+++ b/oxlint-suppressions.json
@@ -6222,14 +6222,6 @@
       "count": 2
     }
   },
-  "web/app/components/workflow/run/retry-log/retry-result-panel.tsx": {
-    "jsx_a11y/click-events-have-key-events": {
-      "count": 1
-    },
-    "jsx_a11y/no-static-element-interactions": {
-      "count": 1
-    }
-  },
   "web/app/components/workflow/run/special-result-panel.tsx": {
     "jsx_a11y/click-events-have-key-events": {
       "count": 1
diff --git a/packages/contracts/generated/api/console/apps/types.gen.ts b/packages/contracts/generated/api/console/apps/types.gen.ts
index ad0272aae93..afc4c5edf03 100644
--- a/packages/contracts/generated/api/console/apps/types.gen.ts
+++ b/packages/contracts/generated/api/console/apps/types.gen.ts
@@ -1144,6 +1144,7 @@ export type WorkflowRunNodeExecutionResponse = {
   predecessor_node_id?: string | null
   process_data?: unknown
   process_data_truncated?: boolean | null
+  retry_index?: number | null
   status?: string | null
   title?: string | null
 }
diff --git a/packages/contracts/generated/api/console/apps/zod.gen.ts b/packages/contracts/generated/api/console/apps/zod.gen.ts
index 8faa637f586..5625ee8c10e 100644
--- a/packages/contracts/generated/api/console/apps/zod.gen.ts
+++ b/packages/contracts/generated/api/console/apps/zod.gen.ts
@@ -1667,6 +1667,7 @@ export const zWorkflowRunNodeExecutionResponse = z.object({
   predecessor_node_id: z.string().nullish(),
   process_data: z.unknown().optional(),
   process_data_truncated: z.boolean().nullish(),
+  retry_index: z.int().nullish(),
   status: z.string().nullish(),
   title: z.string().nullish(),
 })
diff --git a/packages/contracts/generated/api/console/rag/types.gen.ts b/packages/contracts/generated/api/console/rag/types.gen.ts
index baa0181f214..2777d150024 100644
--- a/packages/contracts/generated/api/console/rag/types.gen.ts
+++ b/packages/contracts/generated/api/console/rag/types.gen.ts
@@ -240,6 +240,7 @@ export type WorkflowRunNodeExecutionResponse = {
   predecessor_node_id?: string | null
   process_data?: unknown
   process_data_truncated?: boolean | null
+  retry_index?: number | null
   status?: string | null
   title?: string | null
 }
diff --git a/packages/contracts/generated/api/console/rag/zod.gen.ts b/packages/contracts/generated/api/console/rag/zod.gen.ts
index 77ba7a76c1b..3b9a4ad4a71 100644
--- a/packages/contracts/generated/api/console/rag/zod.gen.ts
+++ b/packages/contracts/generated/api/console/rag/zod.gen.ts
@@ -407,6 +407,7 @@ export const zWorkflowRunNodeExecutionResponse = z.object({
   predecessor_node_id: z.string().nullish(),
   process_data: z.unknown().optional(),
   process_data_truncated: z.boolean().nullish(),
+  retry_index: z.int().nullish(),
   status: z.string().nullish(),
   title: z.string().nullish(),
 })
diff --git a/packages/contracts/generated/api/console/snippets/types.gen.ts b/packages/contracts/generated/api/console/snippets/types.gen.ts
index c32ec0cd610..8fb46ea24ba 100644
--- a/packages/contracts/generated/api/console/snippets/types.gen.ts
+++ b/packages/contracts/generated/api/console/snippets/types.gen.ts
@@ -207,6 +207,7 @@ export type WorkflowRunNodeExecutionResponse = {
   predecessor_node_id?: string | null
   process_data?: unknown
   process_data_truncated?: boolean | null
+  retry_index?: number | null
   status?: string | null
   title?: string | null
 }
diff --git a/packages/contracts/generated/api/console/snippets/zod.gen.ts b/packages/contracts/generated/api/console/snippets/zod.gen.ts
index 37e7525736e..06b8dac4e6f 100644
--- a/packages/contracts/generated/api/console/snippets/zod.gen.ts
+++ b/packages/contracts/generated/api/console/snippets/zod.gen.ts
@@ -243,6 +243,7 @@ export const zWorkflowRunNodeExecutionResponse = z.object({
   predecessor_node_id: z.string().nullish(),
   process_data: z.unknown().optional(),
   process_data_truncated: z.boolean().nullish(),
+  retry_index: z.int().nullish(),
   status: z.string().nullish(),
   title: z.string().nullish(),
 })
diff --git a/web/app/components/workflow/run/retry-log/__tests__/retry-result-panel.spec.tsx b/web/app/components/workflow/run/retry-log/__tests__/retry-result-panel.spec.tsx
index 8c5a45f254c..0b3d8191c20 100644
--- a/web/app/components/workflow/run/retry-log/__tests__/retry-result-panel.spec.tsx
+++ b/web/app/components/workflow/run/retry-log/__tests__/retry-result-panel.spec.tsx
@@ -46,19 +46,45 @@ describe('RetryResultPanel', () => {
     vi.clearAllMocks()
   })
 
-  it('renders retry titles through the real tracing panel and triggers the back action', async () => {
-    const user = userEvent.setup()
-    const onBack = vi.fn()
+  it('should render every retry attempt with its details', async () => {
+    const attempts = [
+      createTrace({
+        id: 'retry-1',
+        status: NodeRunningStatus.Retry,
+        inputs: { attempt: 1 },
+        process_data: { request: 'attempt-1' },
+        outputs: { status_code: 500 },
+        error: 'first failure',
+        expand: true,
+      }),
+      createTrace({
+        id: 'retry-2',
+        status: NodeRunningStatus.Retry,
+        inputs: { attempt: 2 },
+        process_data: { request: 'attempt-2' },
+        outputs: { status_code: 503 },
+        error: 'second failure',
+        expand: true,
+      }),
+    ]
 
-    render(
-      <RetryResultPanel
-        list={[createTrace({ id: 'retry-1' }), createTrace({ id: 'retry-2' })]}
-        onBack={onBack}
-      />,
-    )
+    render(<RetryResultPanel list={attempts} onBack={vi.fn()} />)
 
     expect(screen.getByText('workflow.nodes.common.retry.retry 1')).toBeInTheDocument()
     expect(screen.getByText('workflow.nodes.common.retry.retry 2')).toBeInTheDocument()
+    expect(await screen.findByText('first failure')).toBeInTheDocument()
+    expect(screen.getByText('second failure')).toBeInTheDocument()
+    expect(screen.getByText(/attempt-1/)).toBeInTheDocument()
+    expect(screen.getByText(/attempt-2/)).toBeInTheDocument()
+    expect(screen.getByText(/500/)).toBeInTheDocument()
+    expect(screen.getByText(/503/)).toBeInTheDocument()
+  })
+
+  it('should trigger the back action when back is clicked', async () => {
+    const user = userEvent.setup()
+    const onBack = vi.fn()
+
+    render(<RetryResultPanel list={[]} onBack={onBack} />)
 
     await user.click(screen.getByText('workflow.singleRun.back'))
 
diff --git a/web/app/components/workflow/run/retry-log/retry-result-panel.tsx b/web/app/components/workflow/run/retry-log/retry-result-panel.tsx
index 539ba0210b0..2060f093d45 100644
--- a/web/app/components/workflow/run/retry-log/retry-result-panel.tsx
+++ b/web/app/components/workflow/run/retry-log/retry-result-panel.tsx
@@ -1,40 +1,43 @@
 'use client'
 
-import type { FC } from 'react'
 import type { NodeTracing } from '@/types/workflow'
-import { RiArrowLeftLine } from '@remixicon/react'
 import { memo } from 'react'
 import { useTranslation } from 'react-i18next'
-import TracingPanel from '../tracing-panel'
+import NodePanel from '../node'
 
 type Props = {
   readonly list: NodeTracing[]
   readonly onBack: () => void
 }
 
-const RetryResultPanel: FC<Props> = ({ list, onBack }) => {
+function RetryResultPanel({ list, onBack }: Props) {
   const { t } = useTranslation()
 
   return (
     <div>
-      <div
-        className="flex h-8 cursor-pointer items-center bg-components-panel-bg px-4 system-sm-medium text-text-accent-secondary"
+      <button
+        type="button"
+        className="flex h-8 w-full cursor-pointer items-center bg-components-panel-bg px-4 system-sm-medium text-text-accent-secondary outline-hidden focus-visible:ring-2 focus-visible:ring-state-accent-solid"
         onClick={(e) => {
           e.stopPropagation()
           e.nativeEvent.stopImmediatePropagation()
           onBack()
         }}
       >
-        <RiArrowLeftLine className="mr-1 size-4" />
+        <span aria-hidden className="mr-1 i-ri-arrow-left-line size-4" />
         {t(($) => $['singleRun.back'], { ns: 'workflow' })}
+      </button>
+      <div className="bg-background-section-burn py-2">
+        {list.map((item, index) => (
+          <NodePanel
+            key={`${item.id}:${item.retry_index ?? item.created_at}`}
+            nodeInfo={{
+              ...item,
+              title: `${t(($) => $['nodes.common.retry.retry'], { ns: 'workflow' })} ${index + 1}`,
+            }}
+          />
+        ))}
       </div>
-      <TracingPanel
-        list={list.map((item, index) => ({
-          ...item,
-          title: `${t(($) => $['nodes.common.retry.retry'], { ns: 'workflow' })} ${index + 1}`,
-        }))}
-        className="bg-background-section-burn"
-      />
     </div>
   )
 }