From 7dc3126ff4e5e249a43d3ed293462c6dcca20227 Mon Sep 17 00:00:00 2001 From: Blackoutta <37723456+Blackoutta@users.noreply.github.com> Date: Wed, 15 Jul 2026 13:58:37 +0800 Subject: [PATCH] fix: preserve and display workflow retry details (#38854) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- api/core/app/workflow/layers/persistence.py | 45 +++- api/core/app/workflow/retry_history.py | 24 ++ api/core/helper/ssrf_proxy.py | 2 +- ..._api_workflow_node_execution_repository.py | 7 +- api/fields/workflow_run_fields.py | 1 + api/openapi/markdown/console-openapi.md | 1 + .../api_workflow_node_execution_repository.py | 8 +- ..._api_workflow_node_execution_repository.py | 11 +- api/services/rag_pipeline/rag_pipeline.py | 24 +- api/services/snippet_service.py | 10 +- .../workflow_node_execution_trace_service.py | 217 ++++++++++++++++++ api/services/workflow_run_service.py | 11 +- ..._api_workflow_node_execution_repository.py | 29 +++ .../test_persistence_inspector_publish.py | 3 +- .../app/workflow/test_persistence_layer.py | 111 +++++++++ .../unit_tests/core/helper/test_ssrf_proxy.py | 14 ++ .../logstore/repositories/__init__.py | 0 ..._api_workflow_node_execution_repository.py | 15 ++ .../rag_pipeline/test_rag_pipeline_service.py | 21 +- .../services/test_snippet_service.py | 12 +- ...t_workflow_node_execution_trace_service.py | 157 +++++++++++++ .../services/test_workflow_run_service.py | 20 +- oxlint-suppressions.json | 8 - .../generated/api/console/apps/types.gen.ts | 1 + .../generated/api/console/apps/zod.gen.ts | 1 + .../generated/api/console/rag/types.gen.ts | 1 + .../generated/api/console/rag/zod.gen.ts | 1 + .../api/console/snippets/types.gen.ts | 1 + .../generated/api/console/snippets/zod.gen.ts | 1 + .../__tests__/retry-result-panel.spec.tsx | 44 +++- .../run/retry-log/retry-result-panel.tsx | 31 +-- 31 files changed, 757 insertions(+), 75 deletions(-) create mode 100644 api/core/app/workflow/retry_history.py create mode 100644 api/services/workflow_node_execution_trace_service.py create mode 100644 api/tests/unit_tests/extensions/logstore/repositories/__init__.py create mode 100644 api/tests/unit_tests/extensions/logstore/repositories/test_logstore_api_workflow_node_execution_repository.py create mode 100644 api/tests/unit_tests/services/test_workflow_node_execution_trace_service.py 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( - , - ) + render() 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() 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 = ({ list, onBack }) => { +function RetryResultPanel({ list, onBack }: Props) { const { t } = useTranslation() return (
-
{ e.stopPropagation() e.nativeEvent.stopImmediatePropagation() onBack() }} > - + {t(($) => $['singleRun.back'], { ns: 'workflow' })} + +
+ {list.map((item, index) => ( + $['nodes.common.retry.retry'], { ns: 'workflow' })} ${index + 1}`, + }} + /> + ))}
- ({ - ...item, - title: `${t(($) => $['nodes.common.retry.retry'], { ns: 'workflow' })} ${index + 1}`, - }))} - className="bg-background-section-burn" - />
) }