UN-4514 [FIX] Cut per-document workflow-execution fetches and stop loading every file-execution row on each one

workers/callback/tasks.py

@@ -71,6 +71,9 @@ def __init__(self):
         # here) — lets the PG duplicate guard skip a redelivered callback without a
         # second round-trip.
         self.execution_status: str | None = None
+        # The full execution context from that same fetch, reused by the status
+        # determination step instead of fetching the execution again.
+        self.execution_context: dict[str, Any] | None = None
 
 
 def _initialize_performance_managers():
@@ -257,6 +260,7 @@ def _determine_execution_status_unified(
     api_client: InternalAPIClient,
     execution_id: str,
     organization_id: str,
+    execution_context: dict[str, Any],
 ) -> tuple[dict[str, Any], str, int]:
     """Unified status determination logic with timeout detection for all callback types.
 
@@ -269,6 +273,8 @@ def _determine_execution_status_unified(
         api_client: Internal API client for workflow execution queries
         execution_id: Workflow execution ID
         organization_id: Organization context
+        execution_context: Execution context the callback already fetched,
+            reused here instead of fetching the execution again
 
     Returns:
         Tuple of (aggregated_results, final_status, expected_files)
@@ -282,6 +288,7 @@ def _determine_execution_status_unified(
         execution_id,
         organization_id,
         aggregated_results.get("file_results", []),
+        execution_context=execution_context,
     )
 
     # Update aggregated results with wall-clock time
@@ -297,23 +304,10 @@ def _determine_execution_status_unified(
         logger.warning(f"Execution time is 0! File results for execution {execution_id}")
 
     # Step 3: Get expected file count from workflow execution details for timeout detection
-    expected_files = 0
-    try:
-        execution_response = api_client.get_workflow_execution(
-            execution_id, file_execution=False
-        )
-        if execution_response.success:
-            execution_data = execution_response.data
-            expected_files = execution_data.get("total_files", 0)
-            logger.info(
-                f"Expected files from execution details: {expected_files} for execution {execution_id}"
-            )
-        else:
-            logger.warning(
-                f"Could not fetch execution details for {execution_id}: {execution_response.error}"
-            )
-    except Exception as e:
-        logger.warning(f"Could not fetch execution details for {execution_id}: {e}")
+    expected_files = (execution_context or {}).get("total_files", 0)
+    logger.info(
+        f"Expected files from execution details: {expected_files} for execution {execution_id}"
+    )
 
     # Step 4: Extract file processing metrics for status determination
     total_files_processed = aggregated_results.get("total_files_processed", 0)
@@ -728,8 +722,9 @@ def _extract_callback_parameters(
                 f"Fetching complete context from workflow execution {context.execution_id}"
             )
 
+            # file_execution=True: subscription billing reads context.file_executions.
             execution_response = api_client.get_workflow_execution(
-                context.execution_id, file_execution=False
+                context.execution_id, file_execution=True
             )
 
             if not execution_response.success:
@@ -762,9 +757,10 @@ def _extract_callback_parameters(
             temp_config = WorkerConfig()
             temp_api_client = InternalAPIClient(temp_config)
             try:
+                # file_execution=True: subscription billing reads context.file_executions.
                 execution_response = temp_api_client.get_workflow_execution(
                     context.execution_id,
-                    file_execution=False,
+                    file_execution=True,
                 )
 
                 if not execution_response.success:
@@ -810,6 +806,7 @@ def _extract_callback_parameters(
         ) or workflow_definition.get("workflow_id")
         # Status from the same fetch, for the PG duplicate guard (no extra call).
         context.execution_status = execution_info.get("status")
+        context.execution_context = execution_data
 
         # Use existing API detection from source_config (no additional API calls needed)
         is_api_deployment = source_config.get("is_api", False)
@@ -1479,6 +1476,7 @@ def _process_batch_callback_core(
                         api_client=context.api_client,
                         execution_id=context.execution_id,
                         organization_id=context.organization_id,
+                        execution_context=context.execution_context,
                     )
                 )
                 # Update workflow execution status using unified function
@@ -1668,8 +1666,9 @@ def process_batch_callback_api(
     logger.info(f"Created organization-scoped API client: {organization_id}")
 
     try:
+        # file_execution=True: subscription billing reads context.file_executions.
         execution_response = api_client.get_workflow_execution(
-            execution_id, file_execution=False
+            execution_id, file_execution=True
         )
         if not execution_response.success:
             raise RuntimeError(
@@ -1742,6 +1741,7 @@ def process_batch_callback_api(
                         api_client=api_client,
                         execution_id=execution_id,
                         organization_id=organization_id,
+                        execution_context=execution_context,
                     )
                 )
 

workers/shared/api/internal_client.py

@@ -472,14 +472,15 @@ def get_workflow_execution(
         execution_id: str | uuid.UUID,
         organization_id: str | None = None,
         include_cost: bool = False,
-        file_execution: bool = True,
+        file_execution: bool = False,
     ) -> WorkflowExecutionResponse:
         """Get workflow execution with context.
 
         Args:
             execution_id: Workflow execution ID
             organization_id: Optional organization ID override
             include_cost: Whether to include aggregated usage cost (expensive operation)
+            file_execution: Whether to include the execution's file-execution rows
 
         Returns:
             WorkflowExecutionResponse containing workflow execution data

workers/shared/clients/execution_client.py

@@ -104,28 +104,32 @@ def get_workflow_execution(
         execution_id: str | uuid.UUID,
         organization_id: str | None = None,
         include_cost: bool = False,
-        file_execution: bool = True,
+        file_execution: bool = False,
     ) -> ExecutionResponse:
         """Get workflow execution with context.
 
         Args:
             execution_id: Workflow execution ID
             organization_id: Optional organization ID override
             include_cost: Whether to include aggregated usage cost (expensive operation)
+            file_execution: Whether to include every file-execution row of the
+                execution (expensive for large executions; only the callback's
+                billing step needs them)
 
         Returns:
             ExecutionResponse containing workflow execution data
         """
         try:
-            # Build URL with optional cost parameter
             url = self._build_url("workflow_execution", f"{str(execution_id)}/")
+            # file_execution is always sent: the backend defaults it to true, so
+            # omitting it would load every file-execution row.
+            params = {"file_execution": "true" if file_execution else "false"}
             if include_cost:
-                url += "?include_cost=true"
-            if file_execution:
-                url += "?file_execution=true"
+                params["include_cost"] = "true"
 
             response = self.get(
                 url,
+                params=params,
                 organization_id=organization_id,
             )
             return ExecutionResponse.success_response(

workers/shared/processing/files/time_utils.py

@@ -20,6 +20,7 @@ def calculate_execution_time(
         execution_id: str,
         organization_id: str,
         fallback_results: list[dict[str, Any]] = None,
+        execution_context: dict[str, Any] | None = None,
     ) -> float:
         """Calculate wall-clock execution time with multiple fallback strategies.
 
@@ -28,14 +29,15 @@ def calculate_execution_time(
             execution_id: Workflow execution ID
             organization_id: Organization context
             fallback_results: List of file results for summing as fallback
+            execution_context: Already-fetched execution context to reuse, if any
 
         Returns:
             Execution time in seconds
         """
         try:
             # Primary: Get workflow execution start time from backend
             return WallClockTimeCalculator._get_wall_clock_time(
-                api_client, execution_id, organization_id
+                api_client, execution_id, organization_id, execution_context
             )
         except Exception as e:
             logger.error(f"Error calculating wall-clock time: {e}")
@@ -44,30 +46,39 @@ def calculate_execution_time(
 
     @staticmethod
     def _get_wall_clock_time(
-        api_client: InternalAPIClient, execution_id: str, organization_id: str
+        api_client: InternalAPIClient,
+        execution_id: str,
+        organization_id: str,
+        execution_context: dict[str, Any] | None = None,
     ) -> float:
-        """Get wall-clock time from execution created_at timestamp."""
-        execution_response = api_client.get_workflow_execution(
-            execution_id, organization_id
-        )
+        """Get wall-clock time from execution created_at timestamp.
+
+        ``execution_context`` is an already-fetched execution context; when given,
+        no API call is made (created_at never changes).
+        """
+        if not execution_context:
+            execution_response = api_client.get_workflow_execution(
+                execution_id, organization_id
+            )
 
-        if not (execution_response.success and execution_response.data):
-            raise ValueError("Failed to get execution data from API")
+            if not (execution_response.success and execution_response.data):
+                raise ValueError("Failed to get execution data from API")
+            execution_context = execution_response.data
 
         # Extract execution data from the nested structure
-        execution_data = execution_response.data.get("execution", {})
-        if not execution_data:
+        execution = execution_context.get("execution", {})
+        if not execution:
             logger.error(
-                f"No 'execution' key in API response. Available keys: {list(execution_response.data.keys())}"
+                f"No 'execution' key in API response. Available keys: {list(execution_context.keys())}"
             )
             raise ValueError("No execution data found in API response")
 
         # Get created_at from the execution data
-        created_at_str = execution_data.get("created_at")
+        created_at_str = execution.get("created_at")
 
         if not created_at_str:
             logger.error(
-                f"Missing timestamp field in API response. Available fields: {list(execution_response.data.keys())}"
+                f"Missing timestamp field in API response. Available fields: {list(execution_context.keys())}"
             )
             # Don't raise error, let it fall back to file timing calculation
             raise ValueError("No created_at timestamp found in execution data")

workers/shared/workflow/execution/service.py

@@ -70,6 +70,9 @@ def __init__(self, api_client: InternalAPIClient = None):
         self.api_client = api_client
         self.logger = logger
         self._last_execution_error = None
+        # (execution_id, context) of the file being processed, reused by the
+        # source/destination config lookups.
+        self._execution_context: tuple[str, dict] | None = None
 
     def execute_workflow_for_file(
         self,
@@ -556,19 +559,38 @@ def _cleanup_tool_execution_tracker(
     def _get_workflow_execution_context(
         self, execution_id: str, workflow_id: str, organization_id: str
     ) -> tuple[dict, list]:
-        """Get workflow execution context and tool instances."""
-        execution_response = self.api_client.get_workflow_execution(execution_id)
-        if not execution_response.success:
-            raise Exception(
-                f"Failed to get workflow execution: {execution_response.error}"
-            )
+        """Get workflow execution context and tool instances.
+
+        The context is fetched fresh for each file (connector settings can change
+        during a long batch) and kept for this file's config lookups.
+        """
+        execution_context = self._fetch_execution_context(execution_id)
+        self._execution_context = (str(execution_id), execution_context)
 
         tool_instances_response = self.api_client.get_tool_instances_by_workflow(
             workflow_id=workflow_id,
             organization_id=organization_id,
         )
 
-        return execution_response.data, tool_instances_response.tool_instances
+        return execution_context, tool_instances_response.tool_instances
+
+    def _fetch_execution_context(self, execution_id: str) -> dict:
+        """Fetch the execution context, without its file-execution rows."""
+        execution_response = self.api_client.get_workflow_execution(execution_id)
+        if not execution_response.success:
+            raise Exception(
+                f"Failed to get workflow execution: {execution_response.error}"
+            )
+        return execution_response.data
+
+    def _get_cached_execution_context(self, execution_id: str) -> dict:
+        """Execution context already loaded for this file, else fetch it."""
+        cached = self._execution_context
+        if cached is not None and cached[0] == str(execution_id):
+            return cached[1]
+        execution_context = self._fetch_execution_context(execution_id)
+        self._execution_context = (str(execution_id), execution_context)
+        return execution_context
 
     def _get_workflow(
         self, workflow_id: str, organization_id: str
@@ -1706,12 +1728,7 @@ def _get_destination_config(
         """Get destination configuration for the workflow via API."""
         try:
             # Get workflow execution context which includes destination config
-            execution_response = self.api_client.get_workflow_execution(execution_id)
-            if not execution_response.success:
-                raise Exception(
-                    f"Failed to get execution context: {execution_response.error}"
-                )
-            execution_context = execution_response.data
+            execution_context = self._get_cached_execution_context(execution_id)
             destination_config = execution_context.get("destination_config", {})
 
             if not destination_config:
@@ -1737,12 +1754,7 @@ def _get_source_config(
         """Get source configuration for the workflow via API."""
         try:
             # Get workflow execution context which includes source config
-            execution_response = self.api_client.get_workflow_execution(execution_id)
-            if not execution_response.success:
-                raise Exception(
-                    f"Failed to get execution context: {execution_response.error}"
-                )
-            execution_context = execution_response.data
+            execution_context = self._get_cached_execution_context(execution_id)
             source_config = execution_context.get("source_config", {})
 
             if not source_config: