fix: base64_decode 缺少文件时回退重跑上一步

在工作流引擎中，当 base64_decode 因上一步未产出文件失败时，自动重跑前一节点最多 3 次并补充恢复日志，提升链路稳定性。

Made-with: Cursor

app/service/engine/workflow_engine.py

@@ -17,6 +17,10 @@ logger = logging.getLogger(__name__)
 
 
 class WorkflowEngine:
+    _BASE64_DECODE_TYPE = "base64_decode"
+    _BASE64_MISSING_FILES_ERROR = "No files provided for base64 decoding"
+    _MAX_PREVIOUS_STEP_RERUNS = 3
+
     def __init__(self, task_executor: TaskExecutor):
         self._task_executor = task_executor
         self._resolver = ParameterResolver()
@@ -38,14 +42,15 @@ class WorkflowEngine:
             logger.error("Failed to create workspace: %s", e)
             return WorkflowResult.failure(f"failed: {e}")
 
-        for idx, step_id in enumerate(workflow.order, 1):
+        for idx, step_id in enumerate(workflow.order):
             step = workflow.steps.get(step_id)
             if step is None:
                 logger.warning("Step '%s' not found in steps definition, skipping", step_id)
                 continue
 
+            step_no = idx + 1
             logger.info("------ Step %d/%d [%s] type=%s start ------",
-                         idx, total_steps, step.id, step.type)
+                         step_no, total_steps, step.id, step.type)
             logger.info("[%s] raw_inputs=%s", step.id, step.inputs)
 
             step_t0 = time.time()
@@ -55,6 +60,20 @@ class WorkflowEngine:
                 result = self._execute_step_with_retry(task_id, workspace, step, ctx)
                 step_elapsed = time.time() - step_t0
 
+                if result.is_failure():
+                    recovered_result = self._recover_base64_by_rerunning_previous_step(
+                        task_id=task_id,
+                        workspace=workspace,
+                        workflow=workflow,
+                        current_step_index=idx,
+                        current_step=step,
+                        ctx=ctx,
+                        failed_result=result,
+                    )
+                    if recovered_result is not None:
+                        result = recovered_result
+                        step_elapsed = time.time() - step_t0
+
                 if result.is_failure():
                     logger.error("[%s] FAILED in %.2fs after retries. Error: %s",
                                   step_id, step_elapsed, result.error_message)
@@ -137,6 +156,92 @@ class WorkflowEngine:
     ) -> dict[str, Any]:
         return {k: self._resolver.resolve(v, ctx) for k, v in inputs.items()} if inputs else {}
 
+    def _recover_base64_by_rerunning_previous_step(
+        self,
+        task_id: str,
+        workspace: str,
+        workflow: Workflow,
+        current_step_index: int,
+        current_step: Step,
+        ctx: ExecutionContext,
+        failed_result: TaskResult,
+    ) -> TaskResult | None:
+        if current_step.type != self._BASE64_DECODE_TYPE:
+            return None
+        if self._BASE64_MISSING_FILES_ERROR not in failed_result.error_message:
+            return None
+        if current_step_index <= 0:
+            logger.error(
+                "[%s] Missing files in base64_decode but no previous step exists",
+                current_step.id,
+            )
+            return None
+
+        previous_step_id = workflow.order[current_step_index - 1]
+        previous_step = workflow.steps.get(previous_step_id)
+        if previous_step is None:
+            logger.error(
+                "[%s] Missing files in base64_decode but previous step '%s' not found",
+                current_step.id,
+                previous_step_id,
+            )
+            return None
+
+        current_result = failed_result
+        for recover_attempt in range(1, self._MAX_PREVIOUS_STEP_RERUNS + 1):
+            logger.warning(
+                "[%s] Trigger fallback: rerun previous step [%s] (attempt %d/%d) "
+                "because base64_decode has no files",
+                current_step.id,
+                previous_step.id,
+                recover_attempt,
+                self._MAX_PREVIOUS_STEP_RERUNS,
+            )
+
+            previous_result = self._execute_step_with_retry(task_id, workspace, previous_step, ctx)
+            if previous_result.is_failure():
+                logger.warning(
+                    "[%s] Previous step rerun failed on attempt %d/%d: %s",
+                    previous_step.id,
+                    recover_attempt,
+                    self._MAX_PREVIOUS_STEP_RERUNS,
+                    previous_result.error_message,
+                )
+                continue
+
+            ctx.put_step_result(previous_step.id, previous_result.outputs)
+            logger.info(
+                "[%s] Previous step rerun succeeded on attempt %d/%d, rerunning current step",
+                previous_step.id,
+                recover_attempt,
+                self._MAX_PREVIOUS_STEP_RERUNS,
+            )
+
+            current_result = self._execute_step_with_retry(task_id, workspace, current_step, ctx)
+            if current_result.is_success():
+                logger.info(
+                    "[%s] Recovery succeeded after rerunning previous step [%s]",
+                    current_step.id,
+                    previous_step.id,
+                )
+                return current_result
+
+            if self._BASE64_MISSING_FILES_ERROR not in current_result.error_message:
+                logger.error(
+                    "[%s] Current step rerun failed with non-recoverable error: %s",
+                    current_step.id,
+                    current_result.error_message,
+                )
+                return current_result
+
+        logger.error(
+            "[%s] Recovery failed after %d attempts of rerunning previous step [%s]",
+            current_step.id,
+            self._MAX_PREVIOUS_STEP_RERUNS,
+            previous_step.id,
+        )
+        return current_result
+
 
 def _shorten(d: dict | None, max_len: int = 120) -> str:
     if d is None:

app/service/task/impl/base64_decode_task.py

@@ -21,6 +21,7 @@ class Base64DecodeTask(BaseTask):
             files: list[str] = files_obj if isinstance(files_obj, list) else []
 
             if not files:
+                logger.warning("[base64_decode] No files provided, waiting for workflow engine fallback")
                 return TaskResult.failure("No files provided for base64 decoding")
             if not param.workspace:
                 return TaskResult.failure("Workspace path is not provided")