mirror of
https://github.com/langgenius/dify.git
synced 2026-07-20 09:38:32 +08:00
fix: preserve and display workflow retry details (#38854)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Blackoutta
GitHub
parent
7dd33929c7
commit
7dc3126ff4
@ -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,
|
||||
)
|
||||
|
||||
24
api/core/app/workflow/retry_history.py
Normal file
24
api/core/app/workflow/retry_history.py
Normal file
@ -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")
|
||||
@ -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(
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 |
|
||||
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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 ---
|
||||
|
||||
|
||||
217
api/services/workflow_node_execution_trace_service.py
Normal file
217
api/services/workflow_node_execution_trace_service.py
Normal file
@ -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)
|
||||
@ -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)
|
||||
|
||||
@ -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())
|
||||
|
||||
@ -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"}]
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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}]}
|
||||
@ -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(
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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
|
||||
@ -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,
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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(),
|
||||
})
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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(),
|
||||
})
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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(),
|
||||
})
|
||||
|
||||
@ -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'))
|
||||
|
||||
|
||||
@ -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>
|
||||
)
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user