From 12af482db463367a5ecead20e25d6a88767331d8 Mon Sep 17 00:00:00 2001
From: Muhammad Ali <117142933+muhammad-ali-e@users.noreply.github.com>
Date: Tue, 27 Oct 2026 07:18:51 +0000
Subject: [PATCH 1/3] [FIX] Cut repeated workflow-execution GET calls and stop
 loading file executions by default

Workers fetched the workflow-execution context about 8 times per document,
and every fetch loaded all of the execution's file-execution rows: the
backend defaults file_execution to true and the client only ever sent
"?file_execution=true", never false.

- Client always sends file_execution explicitly, defaults to false, and
  builds the query via params (include_cost + file_execution used to
  produce "?include_cost=true?file_execution=true").
- Callbacks still request the rows (subscription billing reads them) and
  hand the fetched context to the status step, which reused to make two
  more GETs (wall-clock time, expected files).
- Each file reuses the batch's execution context instead of fetching it,
  and its source/destination config lookups reuse it too.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WpxxVeu41XtPNkvP2Lm8H4
---
 workers/callback/tasks.py                     |  37 ++-
 workers/file_processing/tasks.py              |   5 +
 workers/shared/api/internal_client.py         |   3 +-
 workers/shared/clients/execution_client.py    |  14 +-
 workers/shared/models/file_processing.py      |   4 +
 workers/shared/processing/files/processor.py  |   3 +
 workers/shared/processing/files/time_utils.py |  33 ++-
 workers/shared/workflow/execution/service.py  |  64 +++--
 .../test_workflow_execution_fetch_reuse.py    | 268 ++++++++++++++++++
 9 files changed, 381 insertions(+), 50 deletions(-)
 create mode 100644 workers/tests/test_workflow_execution_fetch_reuse.py

diff --git a/workers/callback/tasks.py b/workers/callback/tasks.py
index 8259fff591..a4b794f039 100644
--- a/workers/callback/tasks.py
+++ b/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_data: 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_data: dict[str, Any] | None = None,
 ) -> 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_data: Execution context the callback already fetched. When
+            given, it is reused instead of fetching the execution twice more.
 
     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_data=execution_data,
     )
 
     # Update aggregated results with wall-clock time
@@ -299,19 +306,19 @@ def _determine_execution_status_unified(
     # 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
+        if execution_data is None:
+            execution_response = api_client.get_workflow_execution(execution_id)
+            if execution_response.success:
+                execution_data = execution_response.data
+            else:
+                logger.warning(
+                    f"Could not fetch execution details for {execution_id}: {execution_response.error}"
+                )
+        if execution_data is not None:
             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}")
 
@@ -728,8 +735,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 +770,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 +819,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_data = 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 +1489,7 @@ def _process_batch_callback_core(
                         api_client=context.api_client,
                         execution_id=context.execution_id,
                         organization_id=context.organization_id,
+                        execution_data=context.execution_data,
                     )
                 )
                 # Update workflow execution status using unified function
@@ -1668,8 +1679,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 +1754,7 @@ def process_batch_callback_api(
                         api_client=api_client,
                         execution_id=execution_id,
                         organization_id=organization_id,
+                        execution_data=execution_context,
                     )
                 )
 
diff --git a/workers/file_processing/tasks.py b/workers/file_processing/tasks.py
index 764953834b..53c9d559a3 100644
--- a/workers/file_processing/tasks.py
+++ b/workers/file_processing/tasks.py
@@ -691,6 +691,7 @@ def _process_individual_files(context: WorkflowContextData) -> WorkflowContextDa
     use_file_history = context.get_setting("use_file_history", True)
     api_client = context.organization_context.api_client
     workflow_execution = context.metadata["workflow_execution"]
+    execution_context = context.metadata.get("execution_context")
     pre_created_file_executions = context.pre_created_file_executions
     skipped_already_completed = context.metadata.get("skipped_already_completed", [])
     skipped_active_duplicate = context.metadata.get("skipped_active_duplicate", [])
@@ -865,6 +866,7 @@ def _process_individual_files(context: WorkflowContextData) -> WorkflowContextDa
             workflow_file_execution_id=workflow_file_execution_id,  # Pass pre-created ID
             workflow_file_execution_object=workflow_file_execution_object,  # Pass pre-created object
             workflow_logger=workflow_logger,  # Pass workflow logger for UI logging
+            execution_context=execution_context,
         )
 
         # Handle file processing result
@@ -1634,6 +1636,7 @@ def _process_file(
     workflow_file_execution_id: str = None,
     workflow_file_execution_object: Any = None,
     workflow_logger: Any = None,
+    execution_context: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Process a single file matching Django backend _process_file pattern.
 
@@ -1647,6 +1650,7 @@ def _process_file(
         file_hash: FileHashData instance with type-safe access
         api_client: Internal API client
         workflow_execution: Workflow execution context
+        execution_context: Full execution context the batch already fetched
 
     Returns:
         File execution result
@@ -1662,6 +1666,7 @@ def _process_file(
         workflow_file_execution_id=workflow_file_execution_id,
         workflow_file_execution_object=workflow_file_execution_object,
         workflow_logger=workflow_logger,
+        execution_context=execution_context,
     )
 
 
diff --git a/workers/shared/api/internal_client.py b/workers/shared/api/internal_client.py
index e832c609d0..c3fcf6ac0c 100644
--- a/workers/shared/api/internal_client.py
+++ b/workers/shared/api/internal_client.py
@@ -472,7 +472,7 @@ 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.
 
@@ -480,6 +480,7 @@ def get_workflow_execution(
             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
diff --git a/workers/shared/clients/execution_client.py b/workers/shared/clients/execution_client.py
index ef98f8daec..aa4c906c25 100644
--- a/workers/shared/clients/execution_client.py
+++ b/workers/shared/clients/execution_client.py
@@ -104,7 +104,7 @@ 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.
 
@@ -112,20 +112,24 @@ def get_workflow_execution(
             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(
diff --git a/workers/shared/models/file_processing.py b/workers/shared/models/file_processing.py
index af40624176..c83a8e2fcb 100644
--- a/workers/shared/models/file_processing.py
+++ b/workers/shared/models/file_processing.py
@@ -32,6 +32,7 @@ def __init__(
         workflow_logger: Any = None,  # Type as Any to avoid import dependency
         current_file_idx: int = 1,
         total_files: int = 1,
+        execution_context: dict[str, Any] | None = None,
     ):
         self.file_data = file_data
         self.file_hash = file_hash
@@ -42,6 +43,9 @@ def __init__(
         self.workflow_logger = workflow_logger
         self.current_file_idx = current_file_idx
         self.total_files = total_files
+        # Full execution context the batch already fetched (execution,
+        # source_config, ...). Reused per file instead of fetching it again.
+        self.execution_context = execution_context
 
         # Extract common identifiers
         self.execution_id = file_data.execution_id
diff --git a/workers/shared/processing/files/processor.py b/workers/shared/processing/files/processor.py
index 90009d3a55..6fbd30ba43 100644
--- a/workers/shared/processing/files/processor.py
+++ b/workers/shared/processing/files/processor.py
@@ -477,6 +477,7 @@ def process_file(
         workflow_file_execution_id: str = None,
         workflow_file_execution_object: Any = None,
         workflow_logger: WorkerWorkflowLogger = None,
+        execution_context: dict[str, Any] | None = None,
     ) -> FileProcessingResult:
         """Main orchestrator method that replaces the complex _process_file method.
 
@@ -497,6 +498,7 @@ def process_file(
             workflow_execution: Workflow execution context
             workflow_file_execution_id: Pre-created workflow file execution ID
             workflow_file_execution_object: Pre-created workflow file execution object
+            execution_context: Full execution context the batch already fetched
 
         Returns:
             FileProcessingResult dataclass
@@ -512,6 +514,7 @@ def process_file(
             workflow_logger=workflow_logger,
             current_file_idx=current_file_idx,
             total_files=total_files,
+            execution_context=execution_context,
         )
 
         logger.debug(
diff --git a/workers/shared/processing/files/time_utils.py b/workers/shared/processing/files/time_utils.py
index f045455272..cf83344661 100644
--- a/workers/shared/processing/files/time_utils.py
+++ b/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_data: dict[str, Any] | None = None,
     ) -> float:
         """Calculate wall-clock execution time with multiple fallback strategies.
 
@@ -28,6 +29,7 @@ def calculate_execution_time(
             execution_id: Workflow execution ID
             organization_id: Organization context
             fallback_results: List of file results for summing as fallback
+            execution_data: Already-fetched execution context to reuse, if any
 
         Returns:
             Execution time in seconds
@@ -35,7 +37,7 @@ def calculate_execution_time(
         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_data
             )
         except Exception as e:
             logger.error(f"Error calculating wall-clock time: {e}")
@@ -44,21 +46,30 @@ 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 execution_context is None:
+            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", {})
+        execution_data = execution_context.get("execution", {})
         if not execution_data:
             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")
 
@@ -67,7 +78,7 @@ def _get_wall_clock_time(
 
         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")
diff --git a/workers/shared/workflow/execution/service.py b/workers/shared/workflow/execution/service.py
index ebf8bce17c..1032569ba4 100644
--- a/workers/shared/workflow/execution/service.py
+++ b/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 context fetched (or handed in) for the file being processed,
+        # keyed by execution_id so the source/destination config lookups reuse it.
+        self._execution_context_cache: dict[str, dict] = {}
 
     def execute_workflow_for_file(
         self,
@@ -155,7 +158,12 @@ def execute_workflow_for_file(
                 raise ValueError("API client required for workflow execution")
 
             execution_context, tool_instances_data = self._get_workflow_execution_context(
-                execution_id, workflow_id, organization_id
+                execution_id,
+                workflow_id,
+                organization_id,
+                execution_context=getattr(
+                    file_processing_context, "execution_context", None
+                ),
             )
 
             workflow_context = self._get_workflow(workflow_id, organization_id)
@@ -554,21 +562,44 @@ def _cleanup_tool_execution_tracker(
             )
 
     def _get_workflow_execution_context(
-        self, execution_id: str, workflow_id: str, organization_id: str
+        self,
+        execution_id: str,
+        workflow_id: str,
+        organization_id: str,
+        execution_context: dict | None = None,
     ) -> 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.
+
+        ``execution_context`` is the context the batch already fetched; when
+        given, it is reused instead of fetching the execution again.
+        """
+        if not isinstance(execution_context, dict):
+            execution_context = self._fetch_execution_context(execution_id)
+        self._execution_context_cache[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."""
+        execution_context = self._execution_context_cache.get(execution_id)
+        if execution_context is None:
+            execution_context = self._fetch_execution_context(execution_id)
+            self._execution_context_cache[execution_id] = execution_context
+        return execution_context
 
     def _get_workflow(
         self, workflow_id: str, organization_id: str
@@ -1171,6 +1202,7 @@ def _copy_filesystem_file(
 
         from unstract.connectors.constants import Common
         from unstract.connectors.filesystems import connectors
+
         from unstract.filesystem import FileStorageType, FileSystem
 
         # Get workflow file storage
@@ -1706,12 +1738,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 +1764,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:
diff --git a/workers/tests/test_workflow_execution_fetch_reuse.py b/workers/tests/test_workflow_execution_fetch_reuse.py
new file mode 100644
index 0000000000..02b2d7b046
--- /dev/null
+++ b/workers/tests/test_workflow_execution_fetch_reuse.py
@@ -0,0 +1,268 @@
+"""GET workflow-execution: fetched once per step, file-execution rows only on demand.
+
+The internal endpoint defaults ``file_execution`` to true and then loads every
+file-execution row of the execution. Workers called it about 8 times per document,
+and the client never sent ``file_execution=false``, so every call paid for those
+rows (quadratic in the file count for an ETL run). These tests pin:
+
+* the client always sends ``file_execution`` explicitly and defaults it to false;
+* only the callback's context fetch asks for the rows (billing reads them);
+* the callback's status step reuses that fetch instead of two more GETs;
+* a file reuses the batch's context, and its source config reuses that too.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+import callback.tasks as cb
+import pytest
+from shared.clients.execution_client import ExecutionAPIClient
+from shared.processing.files.time_utils import WallClockTimeCalculator
+from shared.workflow.execution.service import WorkerWorkflowExecutionService
+from unstract.core.data_models import ExecutionStatus
+
+
+def _execution_context(**execution):
+    execution.setdefault("status", ExecutionStatus.EXECUTING.value)
+    execution.setdefault("workflow_id", "wf")
+    return {
+        "execution": execution,
+        "workflow_definition": {"workflow_id": "wf"},
+        "source_config": {"type": "FILESYSTEM", "connector_id": "c1"},
+        "destination_config": {"connection_type": "DATABASE"},
+        "organization_context": {"organization_id": "org-1"},
+        "file_executions": [],
+    }
+
+
+class TestClientQueryParams:
+    """``ExecutionAPIClient.get_workflow_execution`` builds its query via params."""
+
+    def _client(self):
+        client = ExecutionAPIClient.__new__(ExecutionAPIClient)
+        client.get = MagicMock(return_value={"execution": {}})
+        return client
+
+    def _params(self, client):
+        assert client.get.call_count == 1
+        url = client.get.call_args.args[0]
+        assert "?" not in url, "query must go through params, not the URL"
+        return client.get.call_args.kwargs["params"]
+
+    def test_default_sends_file_execution_false(self):
+        client = self._client()
+        client.get_workflow_execution("e1")
+        assert self._params(client) == {"file_execution": "false"}
+
+    def test_file_execution_true_is_sent(self):
+        client = self._client()
+        client.get_workflow_execution("e1", file_execution=True)
+        assert self._params(client) == {"file_execution": "true"}
+
+    def test_include_cost_combines_with_file_execution(self):
+        # Before: "?include_cost=true?file_execution=true" — the backend read
+        # include_cost as "true?file_execution=true" and skipped the cost.
+        client = self._client()
+        client.get_workflow_execution("e1", include_cost=True, file_execution=True)
+        assert self._params(client) == {
+            "file_execution": "true",
+            "include_cost": "true",
+        }
+
+    def test_organization_id_passed_through(self):
+        client = self._client()
+        client.get_workflow_execution("e1", organization_id="org-9")
+        assert client.get.call_args.kwargs["organization_id"] == "org-9"
+
+
+class TestCallbackFetchesOnce:
+    def _api(self, data):
+        api = MagicMock()
+        api.get_workflow_execution.return_value = MagicMock(success=True, data=data)
+        return api
+
+    def test_extract_requests_file_executions_and_keeps_context(self):
+        data = _execution_context(pipeline_id="p")
+        data["file_executions"] = [{"id": "f1", "status": "COMPLETED"}]
+        api = self._api(data)
+        task = MagicMock()
+        task.request.id = "t1"
+        with (
+            patch.object(cb, "create_api_client", return_value=api),
+            patch.object(cb, "_fetch_pipeline_name_from_api", return_value="pipe"),
+        ):
+            context = cb._extract_callback_parameters(
+                task, [], {"execution_id": "e1", "organization_id": "org-1"}
+            )
+
+        api.get_workflow_execution.assert_called_once_with("e1", file_execution=True)
+        assert context.execution_data is data
+        # Billing still sees the rows it commits usage for.
+        assert context.file_executions == data["file_executions"]
+
+    def test_status_determination_reuses_execution_data(self):
+        created = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+        data = _execution_context(created_at=created)
+        api = MagicMock()
+        api.get_workflow_execution.side_effect = AssertionError("no extra GET")
+
+        _, status, expected_files = cb._determine_execution_status_unified(
+            file_batch_results=[],
+            api_client=api,
+            execution_id="e1",
+            organization_id="org-1",
+            execution_data=data,
+        )
+
+        api.get_workflow_execution.assert_not_called()
+        # Unchanged semantics: total_files is read from the top level, as before.
+        assert expected_files == 0
+        assert status == ExecutionStatus.COMPLETED.value
+
+    def test_status_determination_without_data_still_fetches(self):
+        created = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+        api = self._api(_execution_context(created_at=created))
+
+        cb._determine_execution_status_unified(
+            file_batch_results=[],
+            api_client=api,
+            execution_id="e1",
+            organization_id="org-1",
+        )
+
+        assert api.get_workflow_execution.call_count == 2
+        for call in api.get_workflow_execution.call_args_list:
+            assert call.kwargs.get("file_execution", False) is False
+
+    def test_api_callback_passes_its_context_to_status_step(self):
+        data = _execution_context()
+        data["file_executions"] = [{"id": "f1", "status": "COMPLETED"}]
+        api = self._api(data)
+        task = MagicMock()
+        task.request.id = "t1"
+        seen = {}
+
+        def determine(**kwargs):
+            seen.update(kwargs)
+            raise RuntimeError("stop after status step")
+
+        with (
+            patch.object(cb, "create_api_client", return_value=api),
+            patch.object(
+                cb, "_fetch_pipeline_data_simplified", return_value=("api", "API")
+            ),
+            patch.object(
+                cb, "_determine_execution_status_unified", side_effect=determine
+            ),
+            pytest.raises(RuntimeError, match="stop after status step"),
+        ):
+            cb.process_batch_callback_api(
+                task, [], execution_id="e1", organization_id="org-1", pipeline_id="p"
+            )
+
+        api.get_workflow_execution.assert_called_once_with("e1", file_execution=True)
+        assert seen["execution_data"] is data
+
+
+class TestWallClockReuse:
+    def test_uses_given_context_without_fetching(self):
+        created = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+        api = MagicMock()
+        api.get_workflow_execution.side_effect = AssertionError("no extra GET")
+
+        elapsed = WallClockTimeCalculator.calculate_execution_time(
+            api, "e1", "org-1", [], execution_data=_execution_context(created_at=created)
+        )
+
+        assert 29 <= elapsed <= 60
+        api.get_workflow_execution.assert_not_called()
+
+    def test_falls_back_to_file_times_when_context_lacks_created_at(self):
+        api = MagicMock()
+        elapsed = WallClockTimeCalculator.calculate_execution_time(
+            api,
+            "e1",
+            "org-1",
+            [{"processing_time": 2.0}, {"processing_time": 3.0}],
+            execution_data=_execution_context(),
+        )
+        assert elapsed == pytest.approx(5.0)
+        api.get_workflow_execution.assert_not_called()
+
+
+class TestPerFileReuse:
+    def _service(self):
+        api = MagicMock()
+        api.get_workflow_execution.return_value = MagicMock(
+            success=True, data=_execution_context()
+        )
+        api.get_tool_instances_by_workflow.return_value = MagicMock(
+            tool_instances=[{"id": "t"}]
+        )
+        return WorkerWorkflowExecutionService(api_client=api), api
+
+    def test_batch_context_is_reused_for_the_file(self):
+        service, api = self._service()
+        batch_context = _execution_context(pipeline_id="p")
+
+        context, tools = service._get_workflow_execution_context(
+            "e1", "wf", "org-1", execution_context=batch_context
+        )
+
+        assert context is batch_context
+        assert tools == [{"id": "t"}]
+        api.get_workflow_execution.assert_not_called()
+        # The source config lookup later in the same file reuses it as well.
+        assert service._get_source_config("wf", "e1") == batch_context["source_config"]
+        assert (
+            service._get_destination_config("wf", "e1")
+            == batch_context["destination_config"]
+        )
+        api.get_workflow_execution.assert_not_called()
+
+    def test_without_batch_context_fetches_once_per_file(self):
+        service, api = self._service()
+
+        service._get_workflow_execution_context("e1", "wf", "org-1")
+        service._get_source_config("wf", "e1")
+        service._get_destination_config("wf", "e1")
+
+        api.get_workflow_execution.assert_called_once_with("e1")
+
+    def test_source_config_alone_fetches(self):
+        service, api = self._service()
+        assert service._get_source_config("wf", "e1") == {
+            "type": "FILESYSTEM",
+            "connector_id": "c1",
+        }
+        api.get_workflow_execution.assert_called_once_with("e1")
+
+    def test_failed_fetch_keeps_none_result(self):
+        service, api = self._service()
+        api.get_workflow_execution.return_value = MagicMock(success=False, error="x")
+        assert service._get_source_config("wf", "e1") is None
+
+    def test_file_context_carries_batch_context_to_the_service(self):
+        service, api = self._service()
+        batch_context = _execution_context(pipeline_id="p")
+        file_context = MagicMock()
+        file_context.execution_context = batch_context
+
+        with patch.object(
+            service,
+            "_get_workflow_execution_context",
+            wraps=service._get_workflow_execution_context,
+        ) as spy:
+            # Stop right after setup: workflow lookup is the next call.
+            api.get_workflow.side_effect = RuntimeError("stop")
+            service.execute_workflow_for_file(
+                file_processing_context=file_context,
+                organization_id="org-1",
+                workflow_id="wf",
+                execution_id="e1",
+            )
+
+        assert spy.call_args.kwargs["execution_context"] is batch_context
+        api.get_workflow_execution.assert_not_called()

From fbee2959f6b8eb2699bd6fbb0f0386ce4fb88fd4 Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Mon, 26 Oct 2026 09:43:41 +0000
Subject: [PATCH 2/3] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 workers/shared/workflow/execution/service.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/workers/shared/workflow/execution/service.py b/workers/shared/workflow/execution/service.py
index 1032569ba4..266fc31025 100644
--- a/workers/shared/workflow/execution/service.py
+++ b/workers/shared/workflow/execution/service.py
@@ -1202,7 +1202,6 @@ def _copy_filesystem_file(
 
         from unstract.connectors.constants import Common
         from unstract.connectors.filesystems import connectors
-
         from unstract.filesystem import FileStorageType, FileSystem
 
         # Get workflow file storage

From b7218e2e0900806a80a6faaf9e4359eadfbe7cf6 Mon Sep 17 00:00:00 2001
From: Muhammad Ali <117142933+muhammad-ali-e@users.noreply.github.com>
Date: Tue, 27 Oct 2026 07:18:52 +0000
Subject: [PATCH 3/3] UN-4514 [FIX] Address review: keep per-file context
 fresh, key its cache, harden fallbacks

- Drop reusing the batch's execution context per file: it would freeze the
  source connector settings (auth tokens) for the whole batch. Each file
  fetches once (no file-execution rows); its config lookups reuse that.
- Key the per-file cache by execution_id.
- Status step tolerates a missing context; wall-clock time refetches on
  an empty one.
- Name the full response execution_context throughout.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WpxxVeu41XtPNkvP2Lm8H4
---
 workers/callback/tasks.py                     |  37 ++---
 workers/file_processing/tasks.py              |   5 -
 workers/shared/models/file_processing.py      |   4 -
 workers/shared/processing/files/processor.py  |   3 -
 workers/shared/processing/files/time_utils.py |  14 +-
 workers/shared/workflow/execution/service.py  |  37 ++---
 .../test_workflow_execution_fetch_reuse.py    | 138 +++++++++---------
 7 files changed, 105 insertions(+), 133 deletions(-)

diff --git a/workers/callback/tasks.py b/workers/callback/tasks.py
index a4b794f039..9dbeb36e00 100644
--- a/workers/callback/tasks.py
+++ b/workers/callback/tasks.py
@@ -73,7 +73,7 @@ def __init__(self):
         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_data: dict[str, Any] | None = None
+        self.execution_context: dict[str, Any] | None = None
 
 
 def _initialize_performance_managers():
@@ -260,7 +260,7 @@ def _determine_execution_status_unified(
     api_client: InternalAPIClient,
     execution_id: str,
     organization_id: str,
-    execution_data: dict[str, Any] | None = None,
+    execution_context: dict[str, Any],
 ) -> tuple[dict[str, Any], str, int]:
     """Unified status determination logic with timeout detection for all callback types.
 
@@ -273,8 +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_data: Execution context the callback already fetched. When
-            given, it is reused instead of fetching the execution twice more.
+        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)
@@ -288,7 +288,7 @@ def _determine_execution_status_unified(
         execution_id,
         organization_id,
         aggregated_results.get("file_results", []),
-        execution_data=execution_data,
+        execution_context=execution_context,
     )
 
     # Update aggregated results with wall-clock time
@@ -304,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:
-        if execution_data is None:
-            execution_response = api_client.get_workflow_execution(execution_id)
-            if execution_response.success:
-                execution_data = execution_response.data
-            else:
-                logger.warning(
-                    f"Could not fetch execution details for {execution_id}: {execution_response.error}"
-                )
-        if execution_data is not None:
-            expected_files = execution_data.get("total_files", 0)
-            logger.info(
-                f"Expected files from execution details: {expected_files} for execution {execution_id}"
-            )
-    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)
@@ -819,7 +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_data = execution_data
+        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)
@@ -1489,7 +1476,7 @@ def _process_batch_callback_core(
                         api_client=context.api_client,
                         execution_id=context.execution_id,
                         organization_id=context.organization_id,
-                        execution_data=context.execution_data,
+                        execution_context=context.execution_context,
                     )
                 )
                 # Update workflow execution status using unified function
@@ -1754,7 +1741,7 @@ def process_batch_callback_api(
                         api_client=api_client,
                         execution_id=execution_id,
                         organization_id=organization_id,
-                        execution_data=execution_context,
+                        execution_context=execution_context,
                     )
                 )
 
diff --git a/workers/file_processing/tasks.py b/workers/file_processing/tasks.py
index 53c9d559a3..764953834b 100644
--- a/workers/file_processing/tasks.py
+++ b/workers/file_processing/tasks.py
@@ -691,7 +691,6 @@ def _process_individual_files(context: WorkflowContextData) -> WorkflowContextDa
     use_file_history = context.get_setting("use_file_history", True)
     api_client = context.organization_context.api_client
     workflow_execution = context.metadata["workflow_execution"]
-    execution_context = context.metadata.get("execution_context")
     pre_created_file_executions = context.pre_created_file_executions
     skipped_already_completed = context.metadata.get("skipped_already_completed", [])
     skipped_active_duplicate = context.metadata.get("skipped_active_duplicate", [])
@@ -866,7 +865,6 @@ def _process_individual_files(context: WorkflowContextData) -> WorkflowContextDa
             workflow_file_execution_id=workflow_file_execution_id,  # Pass pre-created ID
             workflow_file_execution_object=workflow_file_execution_object,  # Pass pre-created object
             workflow_logger=workflow_logger,  # Pass workflow logger for UI logging
-            execution_context=execution_context,
         )
 
         # Handle file processing result
@@ -1636,7 +1634,6 @@ def _process_file(
     workflow_file_execution_id: str = None,
     workflow_file_execution_object: Any = None,
     workflow_logger: Any = None,
-    execution_context: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Process a single file matching Django backend _process_file pattern.
 
@@ -1650,7 +1647,6 @@ def _process_file(
         file_hash: FileHashData instance with type-safe access
         api_client: Internal API client
         workflow_execution: Workflow execution context
-        execution_context: Full execution context the batch already fetched
 
     Returns:
         File execution result
@@ -1666,7 +1662,6 @@ def _process_file(
         workflow_file_execution_id=workflow_file_execution_id,
         workflow_file_execution_object=workflow_file_execution_object,
         workflow_logger=workflow_logger,
-        execution_context=execution_context,
     )
 
 
diff --git a/workers/shared/models/file_processing.py b/workers/shared/models/file_processing.py
index c83a8e2fcb..af40624176 100644
--- a/workers/shared/models/file_processing.py
+++ b/workers/shared/models/file_processing.py
@@ -32,7 +32,6 @@ def __init__(
         workflow_logger: Any = None,  # Type as Any to avoid import dependency
         current_file_idx: int = 1,
         total_files: int = 1,
-        execution_context: dict[str, Any] | None = None,
     ):
         self.file_data = file_data
         self.file_hash = file_hash
@@ -43,9 +42,6 @@ def __init__(
         self.workflow_logger = workflow_logger
         self.current_file_idx = current_file_idx
         self.total_files = total_files
-        # Full execution context the batch already fetched (execution,
-        # source_config, ...). Reused per file instead of fetching it again.
-        self.execution_context = execution_context
 
         # Extract common identifiers
         self.execution_id = file_data.execution_id
diff --git a/workers/shared/processing/files/processor.py b/workers/shared/processing/files/processor.py
index 6fbd30ba43..90009d3a55 100644
--- a/workers/shared/processing/files/processor.py
+++ b/workers/shared/processing/files/processor.py
@@ -477,7 +477,6 @@ def process_file(
         workflow_file_execution_id: str = None,
         workflow_file_execution_object: Any = None,
         workflow_logger: WorkerWorkflowLogger = None,
-        execution_context: dict[str, Any] | None = None,
     ) -> FileProcessingResult:
         """Main orchestrator method that replaces the complex _process_file method.
 
@@ -498,7 +497,6 @@ def process_file(
             workflow_execution: Workflow execution context
             workflow_file_execution_id: Pre-created workflow file execution ID
             workflow_file_execution_object: Pre-created workflow file execution object
-            execution_context: Full execution context the batch already fetched
 
         Returns:
             FileProcessingResult dataclass
@@ -514,7 +512,6 @@ def process_file(
             workflow_logger=workflow_logger,
             current_file_idx=current_file_idx,
             total_files=total_files,
-            execution_context=execution_context,
         )
 
         logger.debug(
diff --git a/workers/shared/processing/files/time_utils.py b/workers/shared/processing/files/time_utils.py
index cf83344661..2a568ddd5d 100644
--- a/workers/shared/processing/files/time_utils.py
+++ b/workers/shared/processing/files/time_utils.py
@@ -20,7 +20,7 @@ def calculate_execution_time(
         execution_id: str,
         organization_id: str,
         fallback_results: list[dict[str, Any]] = None,
-        execution_data: dict[str, Any] | None = None,
+        execution_context: dict[str, Any] | None = None,
     ) -> float:
         """Calculate wall-clock execution time with multiple fallback strategies.
 
@@ -29,7 +29,7 @@ def calculate_execution_time(
             execution_id: Workflow execution ID
             organization_id: Organization context
             fallback_results: List of file results for summing as fallback
-            execution_data: Already-fetched execution context to reuse, if any
+            execution_context: Already-fetched execution context to reuse, if any
 
         Returns:
             Execution time in seconds
@@ -37,7 +37,7 @@ def calculate_execution_time(
         try:
             # Primary: Get workflow execution start time from backend
             return WallClockTimeCalculator._get_wall_clock_time(
-                api_client, execution_id, organization_id, execution_data
+                api_client, execution_id, organization_id, execution_context
             )
         except Exception as e:
             logger.error(f"Error calculating wall-clock time: {e}")
@@ -56,7 +56,7 @@ def _get_wall_clock_time(
         ``execution_context`` is an already-fetched execution context; when given,
         no API call is made (created_at never changes).
         """
-        if execution_context is None:
+        if not execution_context:
             execution_response = api_client.get_workflow_execution(
                 execution_id, organization_id
             )
@@ -66,15 +66,15 @@ def _get_wall_clock_time(
             execution_context = execution_response.data
 
         # Extract execution data from the nested structure
-        execution_data = execution_context.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_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(
diff --git a/workers/shared/workflow/execution/service.py b/workers/shared/workflow/execution/service.py
index 266fc31025..c28e10b3e0 100644
--- a/workers/shared/workflow/execution/service.py
+++ b/workers/shared/workflow/execution/service.py
@@ -70,9 +70,9 @@ def __init__(self, api_client: InternalAPIClient = None):
         self.api_client = api_client
         self.logger = logger
         self._last_execution_error = None
-        # Execution context fetched (or handed in) for the file being processed,
-        # keyed by execution_id so the source/destination config lookups reuse it.
-        self._execution_context_cache: dict[str, dict] = {}
+        # (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,
@@ -158,12 +158,7 @@ def execute_workflow_for_file(
                 raise ValueError("API client required for workflow execution")
 
             execution_context, tool_instances_data = self._get_workflow_execution_context(
-                execution_id,
-                workflow_id,
-                organization_id,
-                execution_context=getattr(
-                    file_processing_context, "execution_context", None
-                ),
+                execution_id, workflow_id, organization_id
             )
 
             workflow_context = self._get_workflow(workflow_id, organization_id)
@@ -562,20 +557,15 @@ def _cleanup_tool_execution_tracker(
             )
 
     def _get_workflow_execution_context(
-        self,
-        execution_id: str,
-        workflow_id: str,
-        organization_id: str,
-        execution_context: dict | None = None,
+        self, execution_id: str, workflow_id: str, organization_id: str
     ) -> tuple[dict, list]:
         """Get workflow execution context and tool instances.
 
-        ``execution_context`` is the context the batch already fetched; when
-        given, it is reused instead of fetching the execution again.
+        The context is fetched fresh for each file (connector settings can change
+        during a long batch) and kept for this file's config lookups.
         """
-        if not isinstance(execution_context, dict):
-            execution_context = self._fetch_execution_context(execution_id)
-        self._execution_context_cache[execution_id] = execution_context
+        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,
@@ -595,10 +585,11 @@ def _fetch_execution_context(self, execution_id: str) -> dict:
 
     def _get_cached_execution_context(self, execution_id: str) -> dict:
         """Execution context already loaded for this file, else fetch it."""
-        execution_context = self._execution_context_cache.get(execution_id)
-        if execution_context is None:
-            execution_context = self._fetch_execution_context(execution_id)
-            self._execution_context_cache[execution_id] = execution_context
+        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(
diff --git a/workers/tests/test_workflow_execution_fetch_reuse.py b/workers/tests/test_workflow_execution_fetch_reuse.py
index 02b2d7b046..b3f9dc0bc2 100644
--- a/workers/tests/test_workflow_execution_fetch_reuse.py
+++ b/workers/tests/test_workflow_execution_fetch_reuse.py
@@ -8,7 +8,7 @@
 * the client always sends ``file_execution`` explicitly and defaults it to false;
 * only the callback's context fetch asks for the rows (billing reads them);
 * the callback's status step reuses that fetch instead of two more GETs;
-* a file reuses the batch's context, and its source config reuses that too.
+* a file fetches once, and its source config lookup reuses that fetch.
 """
 
 from __future__ import annotations
@@ -98,11 +98,11 @@ def test_extract_requests_file_executions_and_keeps_context(self):
             )
 
         api.get_workflow_execution.assert_called_once_with("e1", file_execution=True)
-        assert context.execution_data is data
+        assert context.execution_context is data
         # Billing still sees the rows it commits usage for.
         assert context.file_executions == data["file_executions"]
 
-    def test_status_determination_reuses_execution_data(self):
+    def test_status_determination_reuses_execution_context(self):
         created = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
         data = _execution_context(created_at=created)
         api = MagicMock()
@@ -113,7 +113,7 @@ def test_status_determination_reuses_execution_data(self):
             api_client=api,
             execution_id="e1",
             organization_id="org-1",
-            execution_data=data,
+            execution_context=data,
         )
 
         api.get_workflow_execution.assert_not_called()
@@ -121,21 +121,6 @@ def test_status_determination_reuses_execution_data(self):
         assert expected_files == 0
         assert status == ExecutionStatus.COMPLETED.value
 
-    def test_status_determination_without_data_still_fetches(self):
-        created = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
-        api = self._api(_execution_context(created_at=created))
-
-        cb._determine_execution_status_unified(
-            file_batch_results=[],
-            api_client=api,
-            execution_id="e1",
-            organization_id="org-1",
-        )
-
-        assert api.get_workflow_execution.call_count == 2
-        for call in api.get_workflow_execution.call_args_list:
-            assert call.kwargs.get("file_execution", False) is False
-
     def test_api_callback_passes_its_context_to_status_step(self):
         data = _execution_context()
         data["file_executions"] = [{"id": "f1", "status": "COMPLETED"}]
@@ -163,7 +148,7 @@ def determine(**kwargs):
             )
 
         api.get_workflow_execution.assert_called_once_with("e1", file_execution=True)
-        assert seen["execution_data"] is data
+        assert seen["execution_context"] is data
 
 
 class TestWallClockReuse:
@@ -173,7 +158,7 @@ def test_uses_given_context_without_fetching(self):
         api.get_workflow_execution.side_effect = AssertionError("no extra GET")
 
         elapsed = WallClockTimeCalculator.calculate_execution_time(
-            api, "e1", "org-1", [], execution_data=_execution_context(created_at=created)
+            api, "e1", "org-1", [], execution_context=_execution_context(created_at=created)
         )
 
         assert 29 <= elapsed <= 60
@@ -186,50 +171,49 @@ def test_falls_back_to_file_times_when_context_lacks_created_at(self):
             "e1",
             "org-1",
             [{"processing_time": 2.0}, {"processing_time": 3.0}],
-            execution_data=_execution_context(),
+            execution_context=_execution_context(),
         )
         assert elapsed == pytest.approx(5.0)
         api.get_workflow_execution.assert_not_called()
 
 
 class TestPerFileReuse:
+    """Each file fetches its execution context once, without file-execution rows.
+
+    It is fetched per file, not taken from the batch, so a long batch still sees
+    connector settings (auth tokens) as they are when each file starts.
+    """
+
     def _service(self):
         api = MagicMock()
         api.get_workflow_execution.return_value = MagicMock(
-            success=True, data=_execution_context()
+            success=True, data=_execution_context(pipeline_id="p")
         )
         api.get_tool_instances_by_workflow.return_value = MagicMock(
             tool_instances=[{"id": "t"}]
         )
         return WorkerWorkflowExecutionService(api_client=api), api
 
-    def test_batch_context_is_reused_for_the_file(self):
-        service, api = self._service()
-        batch_context = _execution_context(pipeline_id="p")
-
-        context, tools = service._get_workflow_execution_context(
-            "e1", "wf", "org-1", execution_context=batch_context
-        )
-
-        assert context is batch_context
-        assert tools == [{"id": "t"}]
-        api.get_workflow_execution.assert_not_called()
-        # The source config lookup later in the same file reuses it as well.
-        assert service._get_source_config("wf", "e1") == batch_context["source_config"]
-        assert (
-            service._get_destination_config("wf", "e1")
-            == batch_context["destination_config"]
-        )
-        api.get_workflow_execution.assert_not_called()
-
-    def test_without_batch_context_fetches_once_per_file(self):
+    def test_config_lookups_reuse_the_files_fetch(self):
         service, api = self._service()
 
-        service._get_workflow_execution_context("e1", "wf", "org-1")
-        service._get_source_config("wf", "e1")
-        service._get_destination_config("wf", "e1")
+        context, tools = service._get_workflow_execution_context("e1", "wf", "org-1")
+        source = service._get_source_config("wf", "e1")
+        destination = service._get_destination_config("wf", "e1")
 
         api.get_workflow_execution.assert_called_once_with("e1")
+        assert tools == [{"id": "t"}]
+        assert source == context["source_config"]
+        assert destination == context["destination_config"]
+
+    def test_each_file_fetches_fresh(self):
+        # One service per file (processor.py); a second file must not see the
+        # first file's context.
+        first, api = self._service()
+        first._get_workflow_execution_context("e1", "wf", "org-1")
+        second = WorkerWorkflowExecutionService(api_client=api)
+        second._get_workflow_execution_context("e1", "wf", "org-1")
+        assert api.get_workflow_execution.call_count == 2
 
     def test_source_config_alone_fetches(self):
         service, api = self._service()
@@ -244,25 +228,47 @@ def test_failed_fetch_keeps_none_result(self):
         api.get_workflow_execution.return_value = MagicMock(success=False, error="x")
         assert service._get_source_config("wf", "e1") is None
 
-    def test_file_context_carries_batch_context_to_the_service(self):
+    def test_failed_fetch_is_retried_by_a_later_lookup(self):
         service, api = self._service()
-        batch_context = _execution_context(pipeline_id="p")
-        file_context = MagicMock()
-        file_context.execution_context = batch_context
-
-        with patch.object(
-            service,
-            "_get_workflow_execution_context",
-            wraps=service._get_workflow_execution_context,
-        ) as spy:
-            # Stop right after setup: workflow lookup is the next call.
-            api.get_workflow.side_effect = RuntimeError("stop")
-            service.execute_workflow_for_file(
-                file_processing_context=file_context,
-                organization_id="org-1",
-                workflow_id="wf",
-                execution_id="e1",
-            )
+        api.get_workflow_execution.return_value = MagicMock(success=False, error="x")
+        assert service._get_source_config("wf", "e1") is None
+        api.get_workflow_execution.return_value = MagicMock(
+            success=True, data=_execution_context()
+        )
+        assert service._get_source_config("wf", "e1") is not None
 
-        assert spy.call_args.kwargs["execution_context"] is batch_context
-        api.get_workflow_execution.assert_not_called()
+    def test_cached_context_is_not_served_for_another_execution(self):
+        service, api = self._service()
+        service._get_workflow_execution_context("e1", "wf", "org-1")
+        service._get_source_config("wf", "e2")
+        assert [c.args[0] for c in api.get_workflow_execution.call_args_list] == [
+            "e1",
+            "e2",
+        ]
+
+
+class TestMissingContextFallbacks:
+    def test_status_step_tolerates_missing_context(self):
+        api = MagicMock()
+        api.get_workflow_execution.return_value = MagicMock(success=False, error="x")
+        _, status, expected_files = cb._determine_execution_status_unified(
+            file_batch_results=[],
+            api_client=api,
+            execution_id="e1",
+            organization_id="org-1",
+            execution_context=None,
+        )
+        assert expected_files == 0
+        assert status == ExecutionStatus.COMPLETED.value
+
+    def test_wall_clock_refetches_on_empty_context(self):
+        created = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+        api = MagicMock()
+        api.get_workflow_execution.return_value = MagicMock(
+            success=True, data=_execution_context(created_at=created)
+        )
+        elapsed = WallClockTimeCalculator.calculate_execution_time(
+            api, "e1", "org-1", [], execution_context={}
+        )
+        assert 29 <= elapsed <= 60
+        api.get_workflow_execution.assert_called_once()