UN-2901 [FIX] Prevent invalid status updates (EXECUTING/ERROR) from duplicate file processing runs (#1606)

* UN-2901 [FIX] Prevent invalid status updates (EXECUTING/ERROR) from duplicate file processing runs

Fixes race condition where late-arriving workers overwrite COMPLETED status
with invalid EXECUTING or ERROR states, causing files to appear failed/stuck
even though processing succeeded.

Changes:
- FileAPIClient: Fixed URL construction and method call bugs
- Fresh DB validation: Check current status before updating to EXECUTING
- Grace period optimization: Early exit when duplicate detected during tool polling
- File count accuracy: Include skipped files in total_files calculation

Impact: Files now correctly maintain COMPLETED status; no duplicate processing

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* coderabbit commnets addressed

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: muhammad-ali-e

unstract/core/src/unstract/core/data_models.py

@@ -861,10 +861,10 @@ class WorkflowFileExecutionData:
     id: str | uuid.UUID
     workflow_execution_id: str | uuid.UUID
     file_name: str
-    file_path: str
     file_size: int
     file_hash: str
     status: str = ExecutionStatus.PENDING.value
+    file_path: str | None = None
     provider_file_uuid: str | None = None
     mime_type: str = ""
     fs_metadata: dict[str, Any] = field(default_factory=dict)
@@ -883,8 +883,7 @@ def __post_init__(self):
         # Validate required fields
         if not self.file_name:
             raise ValueError("file_name is required")
-        if not self.file_path:
-            raise ValueError("file_path is required")
+        # file_path is optional - None for API workflows or when fetching status only
         # file_hash can be empty initially - gets populated during file processing with SHA256 hash
 
     def to_dict(self) -> dict[str, Any]:
@@ -904,7 +903,7 @@ def from_dict(cls, data: dict[str, Any]) -> "WorkflowFileExecutionData":
             id=data["id"],
             workflow_execution_id=data["workflow_execution_id"],
             file_name=data["file_name"],
-            file_path=data["file_path"],
+            file_path=data.get("file_path"),
             file_size=data.get("file_size", 0),
             file_hash=data["file_hash"],
             status=data.get("status", ExecutionStatus.PENDING.value),

unstract/tool-sandbox/src/unstract/tool_sandbox/helper.py

@@ -160,6 +160,29 @@ def poll_tool_status(
                 ).total_seconds()
 
                 if not_found_duration < POLL_NOT_FOUND_GRACE_PERIOD:
+                    # RACE CONDITION OPTIMIZATION: Check if another worker completed this file
+                    # This avoids waiting full grace period when work is already done
+                    tracker = FileExecutionStatusTracker()
+                    current_tracker_data = tracker.get_data(
+                        self.execution_id, file_execution_id
+                    )
+                    if (
+                        current_tracker_data
+                        and current_tracker_data.stage_status.stage
+                        == FileExecutionStage.COMPLETED
+                    ):
+                        logger.warning(
+                            f"File execution COMPLETED by another worker during NOT_FOUND grace period - "
+                            f"duplicate run detected after {not_found_duration:.1f}s for execution_id: {self.execution_id}, file_execution_id: {file_execution_id}"
+                        )
+                        # Break with ERROR to stop further processing (destination, finalization)
+                        # This is not a real error - the file was successfully processed by another worker
+                        response = self._create_run_response(
+                            status=RunnerContainerRunStatus.ERROR,
+                            error="File already completed by another worker (duplicate run avoided)",
+                        )
+                        break
+
                     logger.debug(
                         f"Within polling grace period ({not_found_duration:.1f}s/{POLL_NOT_FOUND_GRACE_PERIOD}s) - "
                         f"continuing poll for execution_id: {self.execution_id}, file_execution_id: {file_execution_id}"
@@ -300,11 +323,6 @@ def call_tool_handler(
                     file_execution_data=file_execution_data,
                 )
                 self._update_stage_status_for_tool_execution(file_execution_id, response)
-                self._update_stage_status(
-                    status=FileExecutionStageStatus.IN_PROGRESS,
-                    stage=FileExecutionStage.FINALIZATION,
-                    file_execution_id=file_execution_id,
-                )
             elif stage.is_before(FileExecutionStage.TOOL_EXECUTION):
                 self._update_stage_status(
                     status=FileExecutionStageStatus.SUCCESS,
@@ -319,11 +337,6 @@ def call_tool_handler(
                     retry_count=retry_count,
                 )
                 self._update_stage_status_for_tool_execution(file_execution_id, response)
-                self._update_stage_status(
-                    status=FileExecutionStageStatus.IN_PROGRESS,
-                    stage=FileExecutionStage.FINALIZATION,
-                    file_execution_id=file_execution_id,
-                )
             else:
                 logger.warning(
                     f"File execution data stage {file_execution_data.stage_status.stage} is after tool execution for execution_id: {self.execution_id} and file_execution_id: {file_execution_id}"

unstract/workflow-execution/src/unstract/workflow_execution/execution_file_handler.py

@@ -128,6 +128,28 @@ def add_metadata_to_volume(
         metadata_path = self.metadata_file
         if not metadata_path:
             raise FileMetadataJsonNotFound()
+
+        # Check if metadata file already exists - skip to avoid overwriting tool data
+        # This prevents Worker 2 from destroying tool_metadata written by Worker 1's tool container
+        try:
+            self.get_workflow_metadata()
+            logger.info(
+                f"Metadata file already exists for file_execution_id {file_execution_id}. "
+                f"Skipping creation to preserve tool data and avoid race conditions."
+            )
+            return  # Exit early - don't touch existing file
+        except (FileMetadataJsonNotFound, FileNotFoundError):
+            # Normal case - metadata file doesn't exist yet, proceed with creation
+            logger.info(
+                f"Creating new metadata file for file_execution_id {file_execution_id}"
+            )
+        except Exception:
+            # Unexpected errors only (e.g., permission issues, S3 connection errors)
+            logger.exception(
+                f"Error checking metadata existence for file_execution_id {file_execution_id}. "
+                f"Proceeding with creation."
+            )
+
         filename = os.path.basename(input_file_path)
         content = {
             MetaDataKey.SOURCE_NAME: filename,