AI/dify
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/langgenius/dify.git synced 2025-08-12 02:58:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: persist workflow execution status on partial success and failure (#20264)

Browse Source 
Signed-off-by: -LAN- <laipz8200@outlook.com>
This commit is contained in:
-LAN- 2025-05-27 11:24:08 +08:00 committed by GitHub
parent c57726a587
commit 55503ce771
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 21 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
api/core/workflow/workflow_cycle_manager.py
View File

@ -125,6 +125,7 @@ class WorkflowCycleManager:
)
)
self._workflow_execution_repository.save(workflow_execution)
return workflow_execution
def handle_workflow_run_partial_success(
@ -158,6 +159,7 @@ class WorkflowCycleManager:
)
)
self._workflow_execution_repository.save(execution)
return execution
def handle_workflow_run_failed(
@ -172,44 +174,45 @@ class WorkflowCycleManager:
trace_manager: Optional[TraceQueueManager] = None,
exceptions_count: int = 0,
) -> WorkflowExecution:
execution = self._get_workflow_execution_or_raise_error(workflow_run_id)
workflow_execution = self._get_workflow_execution_or_raise_error(workflow_run_id)
execution.status = WorkflowExecutionStatus(status.value)
execution.error_message = error_message
execution.total_tokens = total_tokens
execution.total_steps = total_steps
execution.finished_at = datetime.now(UTC).replace(tzinfo=None)
execution.exceptions_count = exceptions_count
workflow_execution.status = WorkflowExecutionStatus(status.value)
workflow_execution.error_message = error_message
workflow_execution.total_tokens = total_tokens
workflow_execution.total_steps = total_steps
workflow_execution.finished_at = datetime.now(UTC).replace(tzinfo=None)
workflow_execution.exceptions_count = exceptions_count
# Use the instance repository to find running executions for a workflow run
running_domain_executions = self._workflow_node_execution_repository.get_running_executions(
workflow_run_id=execution.id
running_node_executions = self._workflow_node_execution_repository.get_running_executions(
workflow_run_id=workflow_execution.id
)
# Update the domain models
now = datetime.now(UTC).replace(tzinfo=None)
for domain_execution in running_domain_executions:
if domain_execution.node_execution_id:
for node_execution in running_node_executions:
if node_execution.node_execution_id:
# Update the domain model
domain_execution.status = NodeExecutionStatus.FAILED
domain_execution.error = error_message
domain_execution.finished_at = now
domain_execution.elapsed_time = (now - domain_execution.created_at).total_seconds()
node_execution.status = NodeExecutionStatus.FAILED
node_execution.error = error_message
node_execution.finished_at = now
node_execution.elapsed_time = (now - node_execution.created_at).total_seconds()
# Update the repository with the domain model
self._workflow_node_execution_repository.save(domain_execution)
self._workflow_node_execution_repository.save(node_execution)
if trace_manager:
trace_manager.add_trace_task(
TraceTask(
TraceTaskName.WORKFLOW_TRACE,
workflow_execution=execution,
workflow_execution=workflow_execution,
conversation_id=conversation_id,
user_id=trace_manager.user_id,
)
)
return execution
self._workflow_execution_repository.save(workflow_execution)
return workflow_execution
def handle_node_execution_start(
self,