From 215929d40090b181e8d9d7682e06980c74c81bfa Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Tue, 1 Sep 2026 12:09:48 +0530
Subject: [PATCH 1/8] UN-1924 [FIX] Reject unsupported files by sniffing MIME
 in API storage staging

API-deployment uploads were gated on the multipart Content-Type, which the
caller supplies and nothing verifies, with a fallback to
application/octet-stream that is itself in AllowedFileTypes. Any file passed
that check, reached the bucket, and failed at extraction with an error that
did not name the cause.

Detect the type from the file's own bytes with libmagic before writing
anything, matching the filesystem source path, and report a rejected file as
a failed entry in the API response instead of staging it under a placeholder
hash that later surfaced as an empty-file error.
---
 .../workflow_manager/endpoint_v2/source.py    |  61 +++++---
 .../tests/test_api_storage_mime_validation.py | 140 ++++++++++++++++++
 2 files changed, 179 insertions(+), 22 deletions(-)
 create mode 100644 backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py

diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py
index 23b0d04f2c..dcc7d746af 100644
--- a/backend/workflow_manager/endpoint_v2/source.py
+++ b/backend/workflow_manager/endpoint_v2/source.py
@@ -2,7 +2,6 @@
 import logging
 import os
 import shutil
-import uuid
 from collections.abc import Collection
 from hashlib import sha256
 from io import BytesIO
@@ -25,7 +24,11 @@
     SourceConstant,
     SourceKey,
 )
-from workflow_manager.endpoint_v2.dto import FileHash, SourceConfig
+from workflow_manager.endpoint_v2.dto import (
+    FileExecutionResult,
+    FileHash,
+    SourceConfig,
+)
 from workflow_manager.endpoint_v2.enums import AllowedFileTypes
 from workflow_manager.endpoint_v2.exceptions import (
     InvalidInputDirectory,
@@ -37,6 +40,7 @@
     UnsupportedMimeTypeError,
 )
 from workflow_manager.endpoint_v2.models import WorkflowEndpoint
+from workflow_manager.endpoint_v2.result_cache_utils import ResultCacheUtils
 from workflow_manager.file_execution.models import WorkflowFileExecution
 from workflow_manager.utils.workflow_log import WorkflowLog
 from workflow_manager.workflow_v2.enums import ExecutionStatus
@@ -69,6 +73,8 @@ class SourceConnector(BaseConnector):
     """
 
     READ_CHUNK_SIZE = 4194304  # Chunk size for reading files
+    # libmagic classifies from the leading bytes; reading more only costs memory.
+    MIME_DETECT_CHUNK_SIZE = 8192
 
     def __init__(
         self,
@@ -1187,6 +1193,22 @@ def load_file(self, input_file_path: str) -> tuple[str, BytesIO]:
 
         return os.path.basename(input_file_path), file_stream
 
+    @classmethod
+    def _detect_uploaded_file_mime_type(cls, file: UploadedFile) -> str:
+        """Detect an uploaded file's MIME type from its own bytes.
+
+        The multipart Content-Type is supplied by the caller and never verified,
+        so it cannot be used to decide what is allowed into API storage.
+        """
+        sample = file.read(cls.MIME_DETECT_CHUNK_SIZE)
+        file.seek(0)
+        if not sample:
+            # libmagic reports "application/x-empty" here, which would reject the
+            # file as an unsupported type. An empty upload is a distinct failure
+            # and is reported as such once staging hands off, so let it pass.
+            return AllowedFileTypes.OCTET_STREAM.value
+        return magic.from_buffer(sample, mime=True)
+
     @classmethod
     def add_input_file_to_api_storage(
         cls,
@@ -1228,30 +1250,25 @@ def add_input_file_to_api_storage(
             file_name = file.name
             destination_path = os.path.join(api_storage_dir, file_name)
 
-            mime_type = file.content_type
+            mime_type = cls._detect_uploaded_file_mime_type(file)
             logger.info(f"Detected MIME type: {mime_type} for file {file_name}")
-            if not mime_type:
-                logger.info(
-                    f"MIME type not found for file {file_name}, using default MIME type: {AllowedFileTypes.OCTET_STREAM.value}"
-                )
-                mime_type = AllowedFileTypes.OCTET_STREAM.value
 
             if not AllowedFileTypes.is_allowed(mime_type):
-                log_message = f"Skipping file '{file_name}' to stage due to unsupported MIME type '{mime_type}'"
-                workflow_log.log_info(logger=logger, message=log_message)
-                # Generate a clearly marked temporary hash to avoid reading the file content
-                # Helps to prevent duplicate entries in file executions
-                fake_hash = f"temp-hash-{uuid.uuid4().hex}"
-                file_hash = FileHash(
-                    file_path=destination_path,
-                    source_connection_type=connection_type,
-                    file_name=file_name,
-                    file_hash=fake_hash,
-                    is_executed=True,
-                    file_size=file.size,
-                    mime_type=mime_type,
+                log_message = (
+                    f"Rejecting file '{file_name}' with unsupported MIME type "
+                    f"'{mime_type}'"
+                )
+                workflow_log.log_error(logger=logger, message=log_message)
+                # Rejected files are never dispatched, so nothing downstream will
+                # report on them - surface the failure in the API response here.
+                ResultCacheUtils.update_api_results(
+                    workflow_id=workflow_id,
+                    execution_id=execution_id,
+                    api_result=FileExecutionResult(
+                        file=file_name,
+                        error=log_message,
+                    ),
                 )
-                file_hashes.update({file_name: file_hash})
                 continue
 
             file_system = FileSystem(FileStorageType.API_EXECUTION)
diff --git a/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
new file mode 100644
index 0000000000..2f6e3c4718
--- /dev/null
+++ b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
@@ -0,0 +1,140 @@
+"""MIME validation for files staged into API storage.
+
+``SourceConnector.add_input_file_to_api_storage`` is the single funnel through
+which API-deployment uploads reach the API storage bucket, so an unsupported
+file has to be rejected here or it reaches the extraction step and fails there
+with an error that does not name the real cause.
+
+Unit tests: the real classmethod runs with its DB/storage-touching
+collaborators patched on the imported module, so no database is needed. MIME
+detection itself is deliberately *not* patched — sniffing the bytes with
+libmagic is the behaviour under test.
+"""
+
+from unittest import mock
+from unittest.mock import MagicMock
+
+import pytest
+from django.core.files.uploadedfile import SimpleUploadedFile
+
+import workflow_manager.endpoint_v2.source as src_mod
+from workflow_manager.endpoint_v2.constants import ApiDeploymentResultStatus
+from workflow_manager.endpoint_v2.source import SourceConnector
+
+# Bytes chosen from what libmagic actually reports (verified against the pinned
+# python-magic): a PDF header sniffs application/pdf, an HTML document sniffs
+# text/html, which is absent from AllowedFileTypes.
+PDF_BYTES = b"%PDF-1.7\n1 0 obj\n<< /Type /Catalog >>\nendobj\n"
+HTML_BYTES = b"<!DOCTYPE html><html><body>hello</body></html>"
+
+
+API_STORAGE_DIR = "/api-storage/exec-1"
+
+
+@pytest.fixture
+def collaborators():
+    """Patch everything the staging loop touches except MIME detection."""
+    with (
+        mock.patch.multiple(
+            src_mod,
+            UserContext=mock.DEFAULT,
+            WorkflowLog=mock.DEFAULT,
+            Workflow=mock.DEFAULT,
+            FileSystem=mock.DEFAULT,
+            FileHistoryHelper=mock.DEFAULT,
+            ResultCacheUtils=mock.DEFAULT,
+        ) as mocks,
+        mock.patch.object(
+            SourceConnector,
+            "get_api_storage_dir_path",
+            return_value=API_STORAGE_DIR,
+        ),
+    ):
+        storage = MagicMock()
+        mocks["FileSystem"].return_value.get_file_storage.return_value = storage
+        mocks["storage"] = storage
+        yield mocks
+
+
+def _upload(name: str, content: bytes, declared: str) -> SimpleUploadedFile:
+    """An uploaded file whose declared Content-Type may not match its bytes."""
+    return SimpleUploadedFile(name, content, content_type=declared)
+
+
+def _stage(files):
+    return SourceConnector.add_input_file_to_api_storage(
+        pipeline_id="pipe-1",
+        workflow_id="wf-1",
+        execution_id="exec-1",
+        file_objs=files,
+    )
+
+
+def _staged_names(storage: MagicMock) -> set[str]:
+    """File names that actually had bytes written to API storage."""
+    return {
+        call.kwargs["path"].rsplit("/", 1)[-1] for call in storage.write.call_args_list
+    }
+
+
+def test_supported_file_is_staged(collaborators) -> None:
+    """A real PDF is staged and returned for dispatch."""
+    result = _stage([_upload("doc.pdf", PDF_BYTES, "application/pdf")])
+
+    assert set(result) == {"doc.pdf"}
+    assert result["doc.pdf"].mime_type == "application/pdf"
+    assert _staged_names(collaborators["storage"]) == {"doc.pdf"}
+
+
+def test_unsupported_bytes_rejected_despite_supported_declared_type(
+    collaborators,
+) -> None:
+    """The declared Content-Type must not decide what reaches the bucket.
+
+    An HTML file announced as application/pdf satisfies any header-based check,
+    so only sniffing the bytes keeps it out.
+    """
+    result = _stage([_upload("evil.pdf", HTML_BYTES, "application/pdf")])
+
+    # Never dispatched...
+    assert result == {}
+    # ...and never written to the bucket.
+    collaborators["storage"].write.assert_not_called()
+
+
+def test_rejection_is_reported_to_the_caller(collaborators) -> None:
+    """A rejected file gets its own failed entry in the API response."""
+    _stage([_upload("evil.pdf", HTML_BYTES, "application/pdf")])
+
+    collaborators["ResultCacheUtils"].update_api_results.assert_called_once()
+    api_result = collaborators["ResultCacheUtils"].update_api_results.call_args.kwargs[
+        "api_result"
+    ]
+    assert api_result.file == "evil.pdf"
+    # The message has to name the offending type, not a downstream symptom.
+    assert "text/html" in api_result.error
+    assert api_result.status == ApiDeploymentResultStatus.FAILED
+
+
+def test_missing_declared_type_falls_back_to_sniffed_type(collaborators) -> None:
+    """A supported file with no declared Content-Type is still staged.
+
+    The recorded type comes from the bytes, so an absent header neither blocks
+    the file nor degrades it to application/octet-stream.
+    """
+    result = _stage([_upload("doc.pdf", PDF_BYTES, "")])
+
+    assert result["doc.pdf"].mime_type == "application/pdf"
+
+
+def test_supported_files_survive_a_rejected_sibling(collaborators) -> None:
+    """One bad file does not fail the whole request."""
+    result = _stage(
+        [
+            _upload("good.pdf", PDF_BYTES, "application/pdf"),
+            _upload("evil.pdf", HTML_BYTES, "application/pdf"),
+        ]
+    )
+
+    assert set(result) == {"good.pdf"}
+    assert _staged_names(collaborators["storage"]) == {"good.pdf"}

From 7af47324878e9f90b9394028f2cfa8821f3d11ff Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Tue, 1 Sep 2026 13:06:30 +0530
Subject: [PATCH 2/8] UN-1924 [FIX] Terminalise an API execution when every
 file is rejected

Rejecting files at staging means the dispatch set can now be empty, which
reached a path that was previously unreachable: the API worker's
_unified_api_execution short-circuits an empty file set and returns
status COMPLETED without ever writing that status back, so the row kept the
status it was dispatched with and the caller polled a PENDING execution
forever.

Skip the dispatch entirely when staging yields nothing, marking the execution
COMPLETED and returning the per-file rejection entries, and make the worker's
own short-circuit persist the status so an empty set from any other caller
cannot strand an execution either.
---
 backend/api_v2/deployment_helper.py           | 21 ++++++
 .../api_v2/tests/test_deployment_helper.py    | 67 +++++++++++++++++--
 .../workflow_manager/workflow_v2/execution.py | 13 ++++
 workers/api-deployment/tasks.py               |  7 ++
 4 files changed, 103 insertions(+), 5 deletions(-)

diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
index 5fbec7999c..2a920dd05e 100644
--- a/backend/api_v2/deployment_helper.py
+++ b/backend/api_v2/deployment_helper.py
@@ -19,6 +19,7 @@
 from utils.constants import Account, CeleryQueue
 from utils.local_context import StateStore
 from workflow_manager.endpoint_v2.destination import DestinationConnector
+from workflow_manager.endpoint_v2.result_cache_utils import ResultCacheUtils
 from workflow_manager.endpoint_v2.source import SourceConnector
 from workflow_manager.workflow_v2.dto import ExecutionResponse
 from workflow_manager.workflow_v2.enums import ExecutionStatus
@@ -306,6 +307,26 @@ def execute_workflow(
                 )
             ).data
 
+        # Staging rejected every file, so there is nothing to dispatch. The worker
+        # short-circuits an empty file set without writing a status back, which
+        # would strand this execution in PENDING — terminalise it here instead.
+        if not hash_values_of_files:
+            WorkflowExecutionServiceHelper.update_execution_completed(str(execution_id))
+            APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id))
+            DestinationConnector.delete_api_storage_dir(
+                workflow_id=workflow_id, execution_id=execution_id
+            )
+            return APIExecutionResponseSerializer(
+                ExecutionResponse(
+                    workflow_id=workflow_id,
+                    execution_id=execution_id,
+                    execution_status=ExecutionStatus.COMPLETED.value,
+                    result=ResultCacheUtils.get_api_results(
+                        workflow_id=str(workflow_id), execution_id=str(execution_id)
+                    ),
+                )
+            ).data
+
         try:
             result = WorkflowHelper.execute_workflow_async(
                 workflow_id=workflow_id,
diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py
index 39b23e5b16..5c12a8cf72 100644
--- a/backend/api_v2/tests/test_deployment_helper.py
+++ b/backend/api_v2/tests/test_deployment_helper.py
@@ -38,8 +38,8 @@ def collaborators():
         mocks[
             "WorkflowExecutionServiceHelper"
         ].create_workflow_execution.return_value = execution_row
-        mocks["SourceConnector"].add_input_file_to_api_storage.side_effect = (
-            RuntimeError("boom")
+        mocks["SourceConnector"].add_input_file_to_api_storage.side_effect = RuntimeError(
+            "boom"
         )
         yield mocks
 
@@ -74,9 +74,9 @@ def test_staging_failure_marks_execution_error(collaborators) -> None:
 
 def test_staging_failure_cleanup_survives_db_marking_error(collaborators) -> None:
     """If marking the row ERROR itself raises, cleanup must still run (not propagate)."""
-    collaborators["WorkflowExecutionServiceHelper"].update_execution_err.side_effect = (
-        RuntimeError("db down")
-    )
+    collaborators[
+        "WorkflowExecutionServiceHelper"
+    ].update_execution_err.side_effect = RuntimeError("db down")
 
     # Must NOT raise — a failed error-marking should not break cleanup.
     dh.DeploymentHelper.execute_workflow(
@@ -89,3 +89,60 @@ def test_staging_failure_cleanup_survives_db_marking_error(collaborators) -> Non
     # Cleanup still runs even though error-marking raised.
     collaborators["APIDeploymentRateLimiter"].release_slot.assert_called_once()
     collaborators["DestinationConnector"].delete_api_storage_dir.assert_called_once()
+
+
+@pytest.fixture
+def staging_rejects_everything():
+    """Patch execute_workflow's collaborators; staging returns no dispatchable files."""
+    with mock.patch.multiple(
+        dh,
+        WorkflowExecutionServiceHelper=mock.DEFAULT,
+        SourceConnector=mock.DEFAULT,
+        DestinationConnector=mock.DEFAULT,
+        APIDeploymentRateLimiter=mock.DEFAULT,
+        WorkflowHelper=mock.DEFAULT,
+        ResultCacheUtils=mock.DEFAULT,
+        Tag=mock.DEFAULT,
+        logger=mock.DEFAULT,
+    ) as mocks:
+        execution_row = MagicMock()
+        execution_row.id = "exec-123"
+        mocks[
+            "WorkflowExecutionServiceHelper"
+        ].create_workflow_execution.return_value = execution_row
+        mocks["SourceConnector"].add_input_file_to_api_storage.return_value = {}
+        mocks["ResultCacheUtils"].get_api_results.return_value = [
+            {"file": "evil.pdf", "status": "Failed", "error": "unsupported MIME type"}
+        ]
+        yield mocks
+
+
+def test_all_files_rejected_completes_without_dispatch(
+    staging_rejects_everything,
+) -> None:
+    """A request whose every file is rejected must reach a terminal status.
+
+    The worker short-circuits an empty file set without writing a status back, so
+    dispatching one strands the execution in PENDING and the caller polls forever.
+    """
+    mocks = staging_rejects_everything
+    response = dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[],
+        timeout=-1,
+    )
+
+    # Nothing is dispatched...
+    mocks["WorkflowHelper"].execute_workflow_async.assert_not_called()
+    # ...the row is terminalised here instead of being left PENDING...
+    mocks[
+        "WorkflowExecutionServiceHelper"
+    ].update_execution_completed.assert_called_once_with("exec-123")
+    # ...the slot and staging dir are released...
+    mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once()
+    mocks["DestinationConnector"].delete_api_storage_dir.assert_called_once()
+    # ...and the caller still sees why each file failed.
+    assert response["execution_status"] == "COMPLETED"
+    assert response["result"][0]["file"] == "evil.pdf"
+    assert response["result"][0]["status"] == "Failed"
diff --git a/backend/workflow_manager/workflow_v2/execution.py b/backend/workflow_manager/workflow_v2/execution.py
index 213c81302b..8da761770a 100644
--- a/backend/workflow_manager/workflow_v2/execution.py
+++ b/backend/workflow_manager/workflow_v2/execution.py
@@ -387,6 +387,19 @@ def update_execution_err(execution_id: str, err_msg: str = "") -> WorkflowExecut
         except WorkflowExecution.DoesNotExist:
             logger.error(f"execution doesn't exist {execution_id}")
 
+    @staticmethod
+    def update_execution_completed(execution_id: str) -> WorkflowExecution | None:
+        """Terminalise an execution that finished without any work to dispatch."""
+        try:
+            execution = WorkflowExecution.objects.get(pk=execution_id)
+            # Same reason as update_execution_err: the model method owns the
+            # terminal-one-way guard, so this cannot revert an already-final row.
+            execution.update_execution(status=ExecutionStatus.COMPLETED)
+            return execution
+        except WorkflowExecution.DoesNotExist:
+            logger.error(f"execution doesn't exist {execution_id}")
+            return None
+
     @staticmethod
     def update_execution_task(execution_id: str, task_id: str) -> None:
         try:
diff --git a/workers/api-deployment/tasks.py b/workers/api-deployment/tasks.py
index a07ef7b7bd..c72c172678 100644
--- a/workers/api-deployment/tasks.py
+++ b/workers/api-deployment/tasks.py
@@ -222,6 +222,13 @@ def _unified_api_execution(
 
         if not converted_files:
             logger.warning("No valid files to process after conversion")
+            # Returning COMPLETED is not enough: without this write the row keeps
+            # whatever status it was dispatched with, and the caller polls forever.
+            api_client.update_workflow_execution_status(
+                execution_id=execution_id,
+                status=ExecutionStatus.COMPLETED.value,
+                total_files=0,
+            )
             return {
                 "execution_id": execution_id,
                 "status": "COMPLETED",

From 63d5237b07d6cdce7c8b9a4052f0bec9f6e1f8c0 Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Wed, 2 Sep 2026 14:54:59 +0530
Subject: [PATCH 3/8] UN-1924 [FIX] Address review findings on unsupported-file
 rejection

Resolving an 8 KiB sample alone rejected every legacy Office upload larger
than the window: libmagic reads .doc/.xls/.ppt through the OLE2 directory
sector at the end of the file, so the sample only ever showed the container
(application/x-ole-storage), which is not allow-listed. Reproduced on a
710 KB .doc and a 1.2 MB .xls, on the API path and the UI execute endpoint
alike. Container samples now escalate to a full-file classification, using
the upload's temp path when Django has spilled it to disk.

Also from review:

- Isolate the terminal status write on the all-rejected path so the rate
  limit slot and staging dir are released even if it raises, matching the
  staging-failure path above it.
- Report the execution's stored status instead of asserting COMPLETED; the
  row can be missing or the terminal guard can refuse the change, and
  claiming success only hides a stranded execution behind a 200.
- Write total_files/failed_files alongside the status, since a terminal row
  with a NULL failed_files reads as a clean success to is_failure_run and to
  run history.
- Distinguish an empty dispatch from a total conversion failure in the
  worker: convert_file_hash_data swallows per-file errors and returns {} for
  both, so the second was being reported as a zero-file success.
- Scope the guard comment on update_execution_completed to the PG transport;
  the legacy path applies the status unconditionally.

Tests: pin the short-circuit to the staging result rather than the upload
list (the previous test passed with the original bug reintroduced), cover
the cleanup-on-DB-error path, the container escalation, and the empty-upload
branch. Mutation-checked: reverting the escalation, gating the short-circuit
on file_objs, and dropping the cleanup isolation each fail the suite.
---
 backend/api_v2/deployment_helper.py           | 21 ++++-
 .../api_v2/tests/test_deployment_helper.py    | 66 +++++++++++++++-
 .../workflow_manager/endpoint_v2/source.py    | 33 +++++++-
 .../tests/test_api_storage_mime_validation.py | 79 +++++++++++++++++++
 .../workflow_manager/workflow_v2/execution.py | 35 +++++++-
 workers/api-deployment/tasks.py               | 27 ++++++-
 6 files changed, 249 insertions(+), 12 deletions(-)

diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
index 2a920dd05e..174d2092f9 100644
--- a/backend/api_v2/deployment_helper.py
+++ b/backend/api_v2/deployment_helper.py
@@ -311,16 +311,33 @@ def execute_workflow(
         # short-circuits an empty file set without writing a status back, which
         # would strand this execution in PENDING — terminalise it here instead.
         if not hash_values_of_files:
-            WorkflowExecutionServiceHelper.update_execution_completed(str(execution_id))
+            # Isolate the DB write the way the staging-failure path above does, so
+            # the rate limit slot and staging dir are released even if it raises.
+            execution = None
+            try:
+                execution = WorkflowExecutionServiceHelper.update_execution_completed(
+                    str(execution_id),
+                    total_files=len(file_objs),
+                    failed_files=len(file_objs),
+                )
+            except Exception:
+                logger.exception(f"Failed to mark execution {execution_id} as COMPLETED")
+
             APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id))
             DestinationConnector.delete_api_storage_dir(
                 workflow_id=workflow_id, execution_id=execution_id
             )
+            # Report the stored status rather than asserting COMPLETED: the row may
+            # be missing, or the terminal guard may have refused the change. Claiming
+            # success here would only hide the stranded execution behind a 200 that a
+            # follow-up GET /status then contradicts.
             return APIExecutionResponseSerializer(
                 ExecutionResponse(
                     workflow_id=workflow_id,
                     execution_id=execution_id,
-                    execution_status=ExecutionStatus.COMPLETED.value,
+                    execution_status=(
+                        execution.status if execution else ExecutionStatus.ERROR.value
+                    ),
                     result=ResultCacheUtils.get_api_results(
                         workflow_id=str(workflow_id), execution_id=str(execution_id)
                     ),
diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py
index 5c12a8cf72..05344e8b44 100644
--- a/backend/api_v2/tests/test_deployment_helper.py
+++ b/backend/api_v2/tests/test_deployment_helper.py
@@ -114,6 +114,11 @@ def staging_rejects_everything():
         mocks["ResultCacheUtils"].get_api_results.return_value = [
             {"file": "evil.pdf", "status": "Failed", "error": "unsupported MIME type"}
         ]
+        completed_row = MagicMock()
+        completed_row.status = "COMPLETED"
+        mocks[
+            "WorkflowExecutionServiceHelper"
+        ].update_execution_completed.return_value = completed_row
         yield mocks
 
 
@@ -126,19 +131,25 @@ def test_all_files_rejected_completes_without_dispatch(
     dispatching one strands the execution in PENDING and the caller polls forever.
     """
     mocks = staging_rejects_everything
+    # A non-empty upload whose staging result is empty. Passing [] instead would
+    # leave the branch satisfied by `not file_objs` too, and the original bug -
+    # dispatching a request whose files were all rejected - would pass this test.
     response = dh.DeploymentHelper.execute_workflow(
         organization_name="org",
         api=_api(),
-        file_objs=[],
+        file_objs=[MagicMock()],
         timeout=-1,
     )
 
     # Nothing is dispatched...
     mocks["WorkflowHelper"].execute_workflow_async.assert_not_called()
-    # ...the row is terminalised here instead of being left PENDING...
+    # ...the row is terminalised here instead of being left PENDING, and the
+    # counters are written so the run does not read back as a clean success...
     mocks[
         "WorkflowExecutionServiceHelper"
-    ].update_execution_completed.assert_called_once_with("exec-123")
+    ].update_execution_completed.assert_called_once_with(
+        "exec-123", total_files=1, failed_files=1
+    )
     # ...the slot and staging dir are released...
     mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once()
     mocks["DestinationConnector"].delete_api_storage_dir.assert_called_once()
@@ -146,3 +157,52 @@ def test_all_files_rejected_completes_without_dispatch(
     assert response["execution_status"] == "COMPLETED"
     assert response["result"][0]["file"] == "evil.pdf"
     assert response["result"][0]["status"] == "Failed"
+
+
+def test_files_staged_successfully_are_dispatched(staging_rejects_everything) -> None:
+    """The short-circuit must not fire when staging did return files.
+
+    Sibling to the test above: together they pin the branch to the staging result
+    rather than to the upload list.
+    """
+    mocks = staging_rejects_everything
+    mocks["SourceConnector"].add_input_file_to_api_storage.return_value = {
+        "good.pdf": MagicMock()
+    }
+
+    dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[MagicMock()],
+        timeout=-1,
+    )
+
+    mocks["WorkflowHelper"].execute_workflow_async.assert_called_once()
+    mocks["WorkflowExecutionServiceHelper"].update_execution_completed.assert_not_called()
+
+
+def test_all_files_rejected_cleanup_survives_db_marking_error(
+    staging_rejects_everything,
+) -> None:
+    """A failing status write must not strand the slot or the staging dir.
+
+    update_execution_completed only catches DoesNotExist, so a lock timeout or a
+    dropped connection propagates; without isolation the org's rate limit slot
+    stays held for its full TTL and throttles every other call for that org.
+    """
+    mocks = staging_rejects_everything
+    mocks[
+        "WorkflowExecutionServiceHelper"
+    ].update_execution_completed.side_effect = Exception("db is down")
+
+    response = dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[MagicMock()],
+        timeout=-1,
+    )
+
+    mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once()
+    mocks["DestinationConnector"].delete_api_storage_dir.assert_called_once()
+    # The row never reached COMPLETED, so the response must not claim it did.
+    assert response["execution_status"] == "ERROR"
diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py
index dcc7d746af..8bb1de81e4 100644
--- a/backend/workflow_manager/endpoint_v2/source.py
+++ b/backend/workflow_manager/endpoint_v2/source.py
@@ -73,8 +73,15 @@ class SourceConnector(BaseConnector):
     """
 
     READ_CHUNK_SIZE = 4194304  # Chunk size for reading files
-    # libmagic classifies from the leading bytes; reading more only costs memory.
+    # Most formats are identifiable from their leading bytes, so a small sample
+    # keeps the common path cheap.
     MIME_DETECT_CHUNK_SIZE = 8192
+    # These two carry the real format in a structure libmagic can only reach by
+    # reading the whole file: the OLE2 directory sector and the zip central
+    # directory both sit at the end. A sample of any size reports the container
+    # rather than the .doc/.xls/.ppt or .docx/.xlsx/.pptx inside it, so these
+    # must never be resolved from the sample alone.
+    CONTAINER_MIME_TYPES = frozenset({"application/x-ole-storage", "application/zip"})
 
     def __init__(
         self,
@@ -1207,7 +1214,29 @@ def _detect_uploaded_file_mime_type(cls, file: UploadedFile) -> str:
             # file as an unsupported type. An empty upload is a distinct failure
             # and is reported as such once staging hands off, so let it pass.
             return AllowedFileTypes.OCTET_STREAM.value
-        return magic.from_buffer(sample, mime=True)
+
+        mime_type = magic.from_buffer(sample, mime=True)
+        if mime_type not in cls.CONTAINER_MIME_TYPES:
+            return mime_type
+        return cls._detect_container_mime_type(file, fallback=mime_type)
+
+    @classmethod
+    def _detect_container_mime_type(cls, file: UploadedFile, fallback: str) -> str:
+        """Resolve a container format by classifying the file in full.
+
+        Django spills uploads over FILE_UPLOAD_MAX_MEMORY_SIZE to disk, so this
+        hands libmagic the path when there is one and only buffers the whole
+        upload for the in-memory case, where that ceiling already bounds it.
+        """
+        temporary_file_path = getattr(file, "temporary_file_path", None)
+        if temporary_file_path is not None:
+            return magic.from_file(temporary_file_path(), mime=True)
+
+        content = file.read()
+        file.seek(0)
+        if not content:
+            return fallback
+        return magic.from_buffer(content, mime=True)
 
     @classmethod
     def add_input_file_to_api_storage(
diff --git a/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
index 2f6e3c4718..ab4f304661 100644
--- a/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
+++ b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
@@ -138,3 +138,82 @@ def test_supported_files_survive_a_rejected_sibling(collaborators) -> None:
 
     assert set(result) == {"good.pdf"}
     assert _staged_names(collaborators["storage"]) == {"good.pdf"}
+
+
+def _ole2_like(total_size: int) -> bytes:
+    """An OLE2 compound file whose format markers sit past the sample window.
+
+    libmagic resolves .doc/.xls/.ppt through the OLE2 directory sector, which
+    lives at the end of the file. Only the container signature is visible in the
+    leading bytes, which is exactly the shape that made these files unstageable.
+    """
+    header = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 504
+    return header + b"\x00" * (total_size - len(header))
+
+
+def test_container_prefix_triggers_a_full_file_sniff(collaborators) -> None:
+    """A container type seen in the sample must not decide the verdict alone.
+
+    Pins the regression directly: an OLE2 upload sniffs application/x-ole-storage
+    from its first bytes, which is absent from AllowedFileTypes, so resolving from
+    the sample alone rejects every legacy Office file bigger than the window.
+
+    The sniff results are stubbed because libmagic's container reporting differs
+    between builds; what must hold everywhere is that an inconclusive sample is
+    escalated to the full file instead of being treated as a verdict.
+    """
+    ole_bytes = _ole2_like(SourceConnector.MIME_DETECT_CHUNK_SIZE * 4)
+    sniffs = ["application/x-ole-storage", "application/msword"]
+
+    with mock.patch.object(src_mod.magic, "from_buffer", side_effect=sniffs) as sniff:
+        result = _stage([_upload("legacy.doc", ole_bytes, "application/msword")])
+
+    # The sample verdict was inconclusive, so the whole file was classified...
+    assert sniff.call_count == 2
+    assert len(sniff.call_args_list[0].args[0]) == SourceConnector.MIME_DETECT_CHUNK_SIZE
+    assert len(sniff.call_args_list[1].args[0]) == len(ole_bytes)
+    # ...and the answer from the full file is what decides.
+    assert set(result) == {"legacy.doc"}
+    assert result["legacy.doc"].mime_type == "application/msword"
+
+
+def test_container_still_rejected_when_the_full_file_is_unsupported(
+    collaborators,
+) -> None:
+    """The full-file re-sniff widens the evidence, not the allow-list."""
+    ole_bytes = _ole2_like(SourceConnector.MIME_DETECT_CHUNK_SIZE * 4)
+    sniffs = ["application/x-ole-storage", "application/x-dosexec"]
+
+    with mock.patch.object(src_mod.magic, "from_buffer", side_effect=sniffs):
+        result = _stage([_upload("legacy.doc", ole_bytes, "application/msword")])
+
+    assert result == {}
+    collaborators["storage"].write.assert_not_called()
+
+
+def test_container_upload_is_not_consumed_by_detection(collaborators) -> None:
+    """Reading the whole file to classify it must still leave it stageable."""
+    ole_bytes = _ole2_like(SourceConnector.MIME_DETECT_CHUNK_SIZE * 4)
+    sniffs = ["application/x-ole-storage", "application/msword"]
+
+    with mock.patch.object(src_mod.magic, "from_buffer", side_effect=sniffs):
+        _stage([_upload("legacy.doc", ole_bytes, "application/msword")])
+
+    written = b"".join(
+        call.kwargs["data"] for call in collaborators["storage"].write.call_args_list
+    )
+    assert written == ole_bytes
+
+
+def test_empty_upload_is_staged_rather_than_called_unsupported(collaborators) -> None:
+    """An empty file must reach the downstream empty-file error, not a type error.
+
+    libmagic calls zero bytes application/x-empty, which is absent from
+    AllowedFileTypes; without the short-circuit an empty upload would be reported
+    as an unsupported type, which names the wrong cause.
+    """
+    result = _stage([_upload("empty.pdf", b"", "application/pdf")])
+
+    assert set(result) == {"empty.pdf"}
+    assert result["empty.pdf"].mime_type == "application/octet-stream"
+    collaborators["ResultCacheUtils"].update_api_results.assert_not_called()
diff --git a/backend/workflow_manager/workflow_v2/execution.py b/backend/workflow_manager/workflow_v2/execution.py
index 8da761770a..d84bccfa90 100644
--- a/backend/workflow_manager/workflow_v2/execution.py
+++ b/backend/workflow_manager/workflow_v2/execution.py
@@ -388,13 +388,40 @@ def update_execution_err(execution_id: str, err_msg: str = "") -> WorkflowExecut
             logger.error(f"execution doesn't exist {execution_id}")
 
     @staticmethod
-    def update_execution_completed(execution_id: str) -> WorkflowExecution | None:
-        """Terminalise an execution that finished without any work to dispatch."""
+    def update_execution_completed(
+        execution_id: str, total_files: int = 0, failed_files: int = 0
+    ) -> WorkflowExecution | None:
+        """Terminalise an execution that finished without any work to dispatch.
+
+        The counters must be written alongside the status: a terminal row whose
+        failed_files is NULL reads as a clean success to is_failure_run() and to
+        run history, which would hide a run whose files were all rejected.
+
+        Returns the row as persisted, so callers can see whether the status
+        actually changed rather than assuming it did.
+        """
         try:
             execution = WorkflowExecution.objects.get(pk=execution_id)
-            # Same reason as update_execution_err: the model method owns the
-            # terminal-one-way guard, so this cannot revert an already-final row.
+            # Same reason as update_execution_err: on the PG transport the model
+            # method owns the terminal-one-way guard, so a row the callback already
+            # finalized cannot be reverted. The legacy transport has no such guard.
             execution.update_execution(status=ExecutionStatus.COMPLETED)
+            execution.total_files = total_files
+            execution.successful_files = 0
+            execution.failed_files = failed_files
+            # Field-scoped, matching update_execution, so this cannot clobber the
+            # status write or anything a concurrent writer touched.
+            execution.save(
+                update_fields=[
+                    "total_files",
+                    "successful_files",
+                    "failed_files",
+                    "modified_at",
+                ]
+            )
+            # The guard may have refused the status change without raising; re-read
+            # so the returned row reflects what is actually stored.
+            execution.refresh_from_db()
             return execution
         except WorkflowExecution.DoesNotExist:
             logger.error(f"execution doesn't exist {execution_id}")
diff --git a/workers/api-deployment/tasks.py b/workers/api-deployment/tasks.py
index c72c172678..e913b0441d 100644
--- a/workers/api-deployment/tasks.py
+++ b/workers/api-deployment/tasks.py
@@ -221,7 +221,32 @@ def _unified_api_execution(
         converted_files = FileProcessingUtils.convert_file_hash_data(hash_values_of_files)
 
         if not converted_files:
-            logger.warning("No valid files to process after conversion")
+            # convert_file_hash_data swallows per-file errors and returns only what
+            # converted, so {} means "nothing was dispatched" OR "every file failed
+            # to convert". Reporting the second as COMPLETED would turn a total
+            # failure into a silent success with no results and no error.
+            if hash_values_of_files:
+                error_message = (
+                    f"None of the {len(hash_values_of_files)} dispatched files could "
+                    "be converted for processing"
+                )
+                logger.error(error_message)
+                api_client.update_workflow_execution_status(
+                    execution_id=execution_id,
+                    status=ExecutionStatus.ERROR.value,
+                    error_message=error_message,
+                    total_files=len(hash_values_of_files),
+                    successful_files=0,
+                    failed_files=len(hash_values_of_files),
+                )
+                return {
+                    "execution_id": execution_id,
+                    "status": "ERROR",
+                    "message": error_message,
+                    "files_processed": 0,
+                }
+
+            logger.warning("No files dispatched for this execution")
             # Returning COMPLETED is not enough: without this write the row keeps
             # whatever status it was dispatched with, and the caller polls forever.
             api_client.update_workflow_execution_status(

From 4b436a8683c78816611e145641f932da158ffdde Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Wed, 2 Sep 2026 14:57:51 +0530
Subject: [PATCH 4/8] UN-1924 [FIX] Release API rate limit slots by org id, not
 model instance

release_slot formats its argument into the Redis key, and acquire_slot built
that key from str(organization.organization_id). Passing the Organization
instance produced a different key, so the ZREM removed a non-member: it
returns 0 and raises nothing, leaving the slot held for the full TTL and
throttling every other API-deployment call for that org.

All three call sites in this module were affected, including the one added
for the all-rejected path. The two correct call sites in the codebase
(undispatched_sweep.py, models/execution.py) already pass the id string, and
the former carries a comment describing this exact trap.

The same instance-instead-of-id call remains in api_deployment_views.py and
in two places in mcp_server/tools/execution.py; those are outside this
change's surface and are left for a separate fix.
---
 backend/api_v2/deployment_helper.py            | 17 ++++++++++++++---
 backend/api_v2/tests/test_deployment_helper.py |  9 +++++++--
 2 files changed, 21 insertions(+), 5 deletions(-)

diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
index 174d2092f9..143544f906 100644
--- a/backend/api_v2/deployment_helper.py
+++ b/backend/api_v2/deployment_helper.py
@@ -294,7 +294,14 @@ def execute_workflow(
                 logger.exception(f"Failed to mark execution {execution_id} as ERROR")
 
             # Async job never started — release the rate limit slot and clean up.
-            APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id))
+            # str(...organization_id), NOT the model instance: release_slot formats
+            # its argument into the Redis key, and acquire_slot built that key from
+            # str(organization.organization_id). Passing the instance ZREMs a
+            # non-member — it returns 0 and raises nothing, so the slot silently
+            # stays held for the full TTL. Same trap as undispatched_sweep.py:245.
+            APIDeploymentRateLimiter.release_slot(
+                str(api.organization.organization_id), str(execution_id)
+            )
             DestinationConnector.delete_api_storage_dir(
                 workflow_id=workflow_id, execution_id=execution_id
             )
@@ -323,7 +330,9 @@ def execute_workflow(
             except Exception:
                 logger.exception(f"Failed to mark execution {execution_id} as COMPLETED")
 
-            APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id))
+            APIDeploymentRateLimiter.release_slot(
+                str(api.organization.organization_id), str(execution_id)
+            )
             DestinationConnector.delete_api_storage_dir(
                 workflow_id=workflow_id, execution_id=execution_id
             )
@@ -390,7 +399,9 @@ def execute_workflow(
             # Dispatch failures are marked ERROR internally by execute_workflow_async;
             # post-dispatch failures (enrichment/config) must not overwrite a running
             # execution's status, so only release the slot and clean up storage here.
-            APIDeploymentRateLimiter.release_slot(api.organization, str(execution_id))
+            APIDeploymentRateLimiter.release_slot(
+                str(api.organization.organization_id), str(execution_id)
+            )
 
             # Clean up storage
             DestinationConnector.delete_api_storage_dir(
diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py
index 05344e8b44..e167dcbeb8 100644
--- a/backend/api_v2/tests/test_deployment_helper.py
+++ b/backend/api_v2/tests/test_deployment_helper.py
@@ -48,6 +48,7 @@ def _api() -> MagicMock:
     api = MagicMock()
     api.workflow.id = "wf-1"
     api.id = "pipe-1"
+    api.organization.organization_id = "org-uuid-1"
     return api
 
 
@@ -150,8 +151,12 @@ def test_all_files_rejected_completes_without_dispatch(
     ].update_execution_completed.assert_called_once_with(
         "exec-123", total_files=1, failed_files=1
     )
-    # ...the slot and staging dir are released...
-    mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once()
+    # ...the slot and staging dir are released. The slot must be released by org
+    # id string: release_slot formats its argument into the Redis key, so passing
+    # the model instance removes a non-member and silently holds the slot.
+    mocks["APIDeploymentRateLimiter"].release_slot.assert_called_once_with(
+        "org-uuid-1", "exec-123"
+    )
     mocks["DestinationConnector"].delete_api_storage_dir.assert_called_once()
     # ...and the caller still sees why each file failed.
     assert response["execution_status"] == "COMPLETED"

From 9929f88d8e75fda8b3966c139e4c558206aa3152 Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Wed, 2 Sep 2026 15:02:39 +0530
Subject: [PATCH 5/8] UN-1924 [FIX] Fail one undetectable upload instead of the
 whole request

MIME detection reads the upload, so a broken stream raises inside the staging
loop and aborts every remaining file in the request. Rejection is already
per-file for an unsupported type; an unreadable one now behaves the same way.

The message says detection failed rather than naming a type, since an I/O
fault and an unsupported format need different follow-ups.
---
 .../workflow_manager/endpoint_v2/source.py    | 20 +++++++++++++++-
 .../tests/test_api_storage_mime_validation.py | 24 +++++++++++++++++++
 2 files changed, 43 insertions(+), 1 deletion(-)

diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py
index 8bb1de81e4..3c0e15da6a 100644
--- a/backend/workflow_manager/endpoint_v2/source.py
+++ b/backend/workflow_manager/endpoint_v2/source.py
@@ -1279,7 +1279,25 @@ def add_input_file_to_api_storage(
             file_name = file.name
             destination_path = os.path.join(api_storage_dir, file_name)
 
-            mime_type = cls._detect_uploaded_file_mime_type(file)
+            try:
+                mime_type = cls._detect_uploaded_file_mime_type(file)
+            except Exception:
+                # Detection reads the upload, so a broken stream raises here. Fail
+                # this one file instead of the whole request, and say that detection
+                # failed rather than blaming the file's type - an I/O fault and an
+                # unsupported format need different follow-ups.
+                log_message = (
+                    f"Rejecting file '{file_name}': could not determine its type"
+                )
+                logger.exception(log_message)
+                workflow_log.log_error(logger=logger, message=log_message)
+                ResultCacheUtils.update_api_results(
+                    workflow_id=workflow_id,
+                    execution_id=execution_id,
+                    api_result=FileExecutionResult(file=file_name, error=log_message),
+                )
+                continue
+
             logger.info(f"Detected MIME type: {mime_type} for file {file_name}")
 
             if not AllowedFileTypes.is_allowed(mime_type):
diff --git a/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
index ab4f304661..8bd5412d32 100644
--- a/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
+++ b/backend/workflow_manager/endpoint_v2/tests/test_api_storage_mime_validation.py
@@ -205,6 +205,30 @@ def test_container_upload_is_not_consumed_by_detection(collaborators) -> None:
     assert written == ole_bytes
 
 
+def test_undetectable_file_fails_alone(collaborators) -> None:
+    """A stream that cannot be read fails its own file, not the whole request."""
+    with mock.patch.object(
+        SourceConnector,
+        "_detect_uploaded_file_mime_type",
+        side_effect=[OSError("stream is gone"), "application/pdf"],
+    ):
+        result = _stage(
+            [
+                _upload("broken.pdf", PDF_BYTES, "application/pdf"),
+                _upload("good.pdf", PDF_BYTES, "application/pdf"),
+            ]
+        )
+
+    assert set(result) == {"good.pdf"}
+    api_result = collaborators["ResultCacheUtils"].update_api_results.call_args.kwargs[
+        "api_result"
+    ]
+    assert api_result.file == "broken.pdf"
+    # An I/O fault and an unsupported format need different follow-ups, so the
+    # message must not blame the file's type.
+    assert "could not determine its type" in api_result.error
+
+
 def test_empty_upload_is_staged_rather_than_called_unsupported(collaborators) -> None:
     """An empty file must reach the downstream empty-file error, not a type error.
 

From 0efd943b76ee31a2fccefac96b1b9d8233359022 Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Tue, 8 Sep 2026 02:09:03 +0530
Subject: [PATCH 6/8] UN-1924 [FIX] Acknowledge results and notify subscribers
 on an all-rejected run

The all-rejected early return reaches a terminal status without going through
WorkflowHelper, so it skipped two things the dispatch path does.

It serves the per-file results in its own response but never marked them
consumed, so a follow-up GET /status served them a second time with 200 where
the contract is 406. The synchronous dispatch path acknowledges at exactly
this point.

It also never reached PipelineUtils.update_pipeline_status, the only
dispatcher of API deployment notifications, so an all-rejected request
alerted nobody where a dispatched-and-failed run would have.

Both are wrapped so a failing webhook cannot turn a handled rejection into a
500, and the response is unchanged.

set_result_acknowledge loses its underscore: it is now called from another
module, so it is part of the contract rather than an internal detail.
---
 backend/api_v2/deployment_helper.py           | 25 ++++++++--
 .../api_v2/tests/test_deployment_helper.py    | 49 +++++++++++++++++++
 .../tests/test_pg_finalization_fixes.py       |  2 +-
 .../workflow_manager/workflow_v2/execution.py |  2 +-
 .../tests/test_record_dispatch_handle.py      |  2 +-
 .../workflow_v2/workflow_helper.py            |  6 +--
 6 files changed, 77 insertions(+), 9 deletions(-)

diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
index 143544f906..bc784b0fba 100644
--- a/backend/api_v2/deployment_helper.py
+++ b/backend/api_v2/deployment_helper.py
@@ -21,6 +21,7 @@
 from workflow_manager.endpoint_v2.destination import DestinationConnector
 from workflow_manager.endpoint_v2.result_cache_utils import ResultCacheUtils
 from workflow_manager.endpoint_v2.source import SourceConnector
+from workflow_manager.utils.pipeline_utils import PipelineUtils
 from workflow_manager.workflow_v2.dto import ExecutionResponse
 from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.execution import WorkflowExecutionServiceHelper
@@ -336,6 +337,26 @@ def execute_workflow(
             DestinationConnector.delete_api_storage_dir(
                 workflow_id=workflow_id, execution_id=execution_id
             )
+            api_results = ResultCacheUtils.get_api_results(
+                workflow_id=str(workflow_id), execution_id=str(execution_id)
+            )
+            if execution is not None:
+                try:
+                    # This response carries the results, so mark them consumed the way
+                    # the synchronous dispatch path does — otherwise a follow-up
+                    # GET /status serves them a second time with 200 instead of 406.
+                    WorkflowHelper.set_result_acknowledge(execution)
+                    # Terminalising here bypasses WorkflowHelper, which is what
+                    # normally notifies API deployment subscribers on a terminal
+                    # status. Without this an all-rejected run alerts nobody.
+                    PipelineUtils.update_pipeline_status(
+                        pipeline_id=pipeline_id, workflow_execution=execution
+                    )
+                except Exception:
+                    logger.exception(
+                        f"Post-completion handling failed for execution {execution_id}"
+                    )
+
             # Report the stored status rather than asserting COMPLETED: the row may
             # be missing, or the terminal guard may have refused the change. Claiming
             # success here would only hide the stranded execution behind a 200 that a
@@ -347,9 +368,7 @@ def execute_workflow(
                     execution_status=(
                         execution.status if execution else ExecutionStatus.ERROR.value
                     ),
-                    result=ResultCacheUtils.get_api_results(
-                        workflow_id=str(workflow_id), execution_id=str(execution_id)
-                    ),
+                    result=api_results,
                 )
             ).data
 
diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py
index e167dcbeb8..d05d845902 100644
--- a/backend/api_v2/tests/test_deployment_helper.py
+++ b/backend/api_v2/tests/test_deployment_helper.py
@@ -103,6 +103,7 @@ def staging_rejects_everything():
         APIDeploymentRateLimiter=mock.DEFAULT,
         WorkflowHelper=mock.DEFAULT,
         ResultCacheUtils=mock.DEFAULT,
+        PipelineUtils=mock.DEFAULT,
         Tag=mock.DEFAULT,
         logger=mock.DEFAULT,
     ) as mocks:
@@ -164,6 +165,54 @@ def test_all_files_rejected_completes_without_dispatch(
     assert response["result"][0]["status"] == "Failed"
 
 
+def test_all_files_rejected_acknowledges_and_notifies(
+    staging_rejects_everything,
+) -> None:
+    """The early return owes the caller what the dispatch path would have done.
+
+    It hands back the results in its own response and reaches a terminal status
+    without going through WorkflowHelper, so both the acknowledgement and the
+    subscriber notification have to happen here or they happen nowhere.
+    """
+    mocks = staging_rejects_everything
+    completed_row = mocks[
+        "WorkflowExecutionServiceHelper"
+    ].update_execution_completed.return_value
+
+    dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[MagicMock()],
+        timeout=-1,
+    )
+
+    # Results were served in this response, so a later GET /status must 406
+    # rather than serve them again.
+    mocks["WorkflowHelper"].set_result_acknowledge.assert_called_once_with(completed_row)
+    # PipelineUtils is the only dispatcher of API deployment notifications.
+    mocks["PipelineUtils"].update_pipeline_status.assert_called_once_with(
+        pipeline_id="pipe-1", workflow_execution=completed_row
+    )
+
+
+def test_all_files_rejected_still_responds_if_notification_fails(
+    staging_rejects_everything,
+) -> None:
+    """A failing webhook must not turn a handled rejection into a 500."""
+    mocks = staging_rejects_everything
+    mocks["PipelineUtils"].update_pipeline_status.side_effect = Exception("webhook down")
+
+    response = dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[MagicMock()],
+        timeout=-1,
+    )
+
+    assert response["execution_status"] == "COMPLETED"
+    assert response["result"][0]["file"] == "evil.pdf"
+
+
 def test_files_staged_successfully_are_dispatched(staging_rejects_everything) -> None:
     """The short-circuit must not fire when staging did return files.
 
diff --git a/backend/workflow_manager/execution/tests/test_pg_finalization_fixes.py b/backend/workflow_manager/execution/tests/test_pg_finalization_fixes.py
index d9cc5ba4da..25770e8c6e 100644
--- a/backend/workflow_manager/execution/tests/test_pg_finalization_fixes.py
+++ b/backend/workflow_manager/execution/tests/test_pg_finalization_fixes.py
@@ -633,7 +633,7 @@ def test_result_acknowledge_does_not_touch_status_or_counters(self):
         stale = self._stale(
             ex, ExecutionStatus.EXECUTING, successful_files=None, failed_files=None
         )
-        WorkflowHelper._set_result_acknowledge(stale)
+        WorkflowHelper.set_result_acknowledge(stale)
         ex.refresh_from_db()
         assert ex.status == ExecutionStatus.COMPLETED.value
         assert ex.successful_files == 1
diff --git a/backend/workflow_manager/workflow_v2/execution.py b/backend/workflow_manager/workflow_v2/execution.py
index d84bccfa90..03b44ed377 100644
--- a/backend/workflow_manager/workflow_v2/execution.py
+++ b/backend/workflow_manager/workflow_v2/execution.py
@@ -499,7 +499,7 @@ def update_execution_queue_message_id(
         # update_fields) re-runs _handle_execution_cache(), which would republish
         # this method's stale in-memory status to the Redis execution cache and can
         # clobber a status the worker has since advanced — the same reason
-        # _set_result_acknowledge uses a queryset .update(). This marker write must
+        # set_result_acknowledge uses a queryset .update(). This marker write must
         # touch ONLY the handle column and never the status/counters or the cache.
         updated = WorkflowExecution.objects.filter(pk=execution_id).update(
             queue_message_id=queue_message_id
diff --git a/backend/workflow_manager/workflow_v2/tests/test_record_dispatch_handle.py b/backend/workflow_manager/workflow_v2/tests/test_record_dispatch_handle.py
index ec40a18bb6..1bfc0baacb 100644
--- a/backend/workflow_manager/workflow_v2/tests/test_record_dispatch_handle.py
+++ b/backend/workflow_manager/workflow_v2/tests/test_record_dispatch_handle.py
@@ -105,7 +105,7 @@ class TestUpdateQueueMessageIdWriteShape:
     has advanced status, it reverts the cache to the stale value with no later
     corrector and the API-deployment sync-poll blocks to its full timeout. The
     codebase already uses the queryset-``.update()`` pattern for the same reason
-    in ``_set_result_acknowledge``; this pins it for the marker write too.
+    in ``set_result_acknowledge``; this pins it for the marker write too.
     """
 
     def test_write_uses_queryset_update_not_save(self):
diff --git a/backend/workflow_manager/workflow_v2/workflow_helper.py b/backend/workflow_manager/workflow_v2/workflow_helper.py
index 971fb7a8c1..42f3fe296c 100644
--- a/backend/workflow_manager/workflow_v2/workflow_helper.py
+++ b/backend/workflow_manager/workflow_v2/workflow_helper.py
@@ -413,7 +413,7 @@ def get_status_of_async_task(
             task_result = ResultCacheUtils.get_api_results(
                 workflow_id=str(execution.workflow.id), execution_id=execution_id
             )
-            cls._set_result_acknowledge(execution)
+            cls.set_result_acknowledge(execution)
 
         result_response = ExecutionResponse(
             workflow_id=str(execution.workflow.id),
@@ -425,7 +425,7 @@ def get_status_of_async_task(
         return result_response
 
     @staticmethod
-    def _set_result_acknowledge(execution: WorkflowExecution) -> None:
+    def set_result_acknowledge(execution: WorkflowExecution) -> None:
         """Mark the result as acknowledged and update the database.
 
         This method is called once the task has completed and its result is forgotten.
@@ -728,7 +728,7 @@ def execute_workflow_async(
                 task_result = ResultCacheUtils.get_api_results(
                     workflow_id=workflow_id, execution_id=execution_id
                 )
-                cls._set_result_acknowledge(workflow_execution)
+                cls.set_result_acknowledge(workflow_execution)
             else:
                 task_result = None
             return ExecutionResponse(

From b27ef4c40e00cf4ed0f1de93414dcdf7f7e7cd94 Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Tue, 8 Sep 2026 02:12:34 +0530
Subject: [PATCH 7/8] UN-1924 [FIX] Keep acknowledgement and notification
 independent

Sharing one try block meant a failed acknowledgement returned before the
notification ran, so an all-rejected execution could reach a terminal state
without alerting API deployment subscribers. They are independent
obligations; neither now depends on the other succeeding.
---
 backend/api_v2/deployment_helper.py           | 11 +++++++++-
 .../api_v2/tests/test_deployment_helper.py    | 22 +++++++++++++++++++
 2 files changed, 32 insertions(+), 1 deletion(-)

diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
index bc784b0fba..b6885e92f3 100644
--- a/backend/api_v2/deployment_helper.py
+++ b/backend/api_v2/deployment_helper.py
@@ -341,11 +341,20 @@ def execute_workflow(
                 workflow_id=str(workflow_id), execution_id=str(execution_id)
             )
             if execution is not None:
+                # Separate try blocks on purpose: these are independent obligations,
+                # and sharing one would let a failed acknowledgement silence the
+                # notification that subscribers depend on.
                 try:
                     # This response carries the results, so mark them consumed the way
                     # the synchronous dispatch path does — otherwise a follow-up
                     # GET /status serves them a second time with 200 instead of 406.
                     WorkflowHelper.set_result_acknowledge(execution)
+                except Exception:
+                    logger.exception(
+                        f"Failed to acknowledge results for execution {execution_id}"
+                    )
+
+                try:
                     # Terminalising here bypasses WorkflowHelper, which is what
                     # normally notifies API deployment subscribers on a terminal
                     # status. Without this an all-rejected run alerts nobody.
@@ -354,7 +363,7 @@ def execute_workflow(
                     )
                 except Exception:
                     logger.exception(
-                        f"Post-completion handling failed for execution {execution_id}"
+                        f"Failed to notify subscribers for execution {execution_id}"
                     )
 
             # Report the stored status rather than asserting COMPLETED: the row may
diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py
index d05d845902..39e70cda57 100644
--- a/backend/api_v2/tests/test_deployment_helper.py
+++ b/backend/api_v2/tests/test_deployment_helper.py
@@ -195,6 +195,28 @@ def test_all_files_rejected_acknowledges_and_notifies(
     )
 
 
+def test_notification_survives_a_failed_acknowledgement(
+    staging_rejects_everything,
+) -> None:
+    """Acknowledgement and notification are independent obligations.
+
+    Sharing one try block would let a failed acknowledgement silence the
+    notification that subscribers depend on.
+    """
+    mocks = staging_rejects_everything
+    mocks["WorkflowHelper"].set_result_acknowledge.side_effect = Exception("db is down")
+
+    response = dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[MagicMock()],
+        timeout=-1,
+    )
+
+    mocks["PipelineUtils"].update_pipeline_status.assert_called_once()
+    assert response["execution_status"] == "COMPLETED"
+
+
 def test_all_files_rejected_still_responds_if_notification_fails(
     staging_rejects_everything,
 ) -> None:

From 2bf3bb7d45e72c41425adb4c6e00407fd1294204 Mon Sep 17 00:00:00 2001
From: Ghost Jake <89829542+Deepak-Kesavan@users.noreply.github.com>
Date: Tue, 8 Sep 2026 02:21:22 +0530
Subject: [PATCH 8/8] UN-1924 [MISC] Drop formatter churn on two untouched
 assertions

Reformatted by a newer ruff than the v0.3.4 the pre-commit config pins, in
the opposite direction to how the file already reads. Restores both to the
committed formatting so the diff is additive.
---
 backend/api_v2/tests/test_deployment_helper.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/backend/api_v2/tests/test_deployment_helper.py b/backend/api_v2/tests/test_deployment_helper.py
index 39e70cda57..2e8c3ec7d3 100644
--- a/backend/api_v2/tests/test_deployment_helper.py
+++ b/backend/api_v2/tests/test_deployment_helper.py
@@ -38,8 +38,8 @@ def collaborators():
         mocks[
             "WorkflowExecutionServiceHelper"
         ].create_workflow_execution.return_value = execution_row
-        mocks["SourceConnector"].add_input_file_to_api_storage.side_effect = RuntimeError(
-            "boom"
+        mocks["SourceConnector"].add_input_file_to_api_storage.side_effect = (
+            RuntimeError("boom")
         )
         yield mocks
 
@@ -75,9 +75,9 @@ def test_staging_failure_marks_execution_error(collaborators) -> None:
 
 def test_staging_failure_cleanup_survives_db_marking_error(collaborators) -> None:
     """If marking the row ERROR itself raises, cleanup must still run (not propagate)."""
-    collaborators[
-        "WorkflowExecutionServiceHelper"
-    ].update_execution_err.side_effect = RuntimeError("db down")
+    collaborators["WorkflowExecutionServiceHelper"].update_execution_err.side_effect = (
+        RuntimeError("db down")
+    )
 
     # Must NOT raise — a failed error-marking should not break cleanup.
     dh.DeploymentHelper.execute_workflow(