diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py
index 23b0d04f2c..3498db05ba 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
@@ -1223,6 +1222,9 @@ def add_input_file_to_api_storage(
         workflow: Workflow = Workflow.objects.get(id=workflow_id)
         file_hashes: dict[str, FileHash] = {}
         unique_file_hashes: set[str] = set()
+        # UN-3016: files rejected for an unsupported MIME type, pre-formatted
+        # for the error below; they are never staged and never handed to a worker.
+        skipped_files: list[str] = []
         connection_type = WorkflowEndpoint.ConnectionType.API
         for file in file_objs:
             file_name = file.name
@@ -1237,21 +1239,20 @@ def add_input_file_to_api_storage(
                 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,
+                # UN-3016: the file is deliberately NOT staged (its bytes are never
+                # written), so it must not be handed to a worker either. Previously
+                # it was returned with is_executed=True and a temp hash; nothing
+                # downstream filters on is_executed, so the worker ran anyway, failed
+                # on the missing file, and the execution died with an opaque
+                # "Execution: <path>; Destination: <path>" error instead of a clear
+                # "skipped, unsupported type" outcome. Excluding it here keeps the
+                # skip a skip.
+                log_message = (
+                    f"Skipping file '{file_name}': unsupported file type "
+                    f"'{mime_type}'. It will not be processed."
                 )
-                file_hashes.update({file_name: file_hash})
+                workflow_log.log_error(logger=logger, message=log_message)
+                skipped_files.append(f"'{file_name}' ({mime_type})")
                 continue
 
             file_system = FileSystem(FileStorageType.API_EXECUTION)
@@ -1285,6 +1286,17 @@ def add_input_file_to_api_storage(
                 mime_type=mime_type,
             )
             file_hashes.update({file_name: file_hash})
+
+        # UN-3016: if every uploaded file was rejected there is nothing to run.
+        # Fail loudly with the reason instead of dispatching an empty execution,
+        # which would otherwise finish as a vacuous success and leave the user
+        # wondering why nothing happened.
+        if skipped_files and not file_hashes:
+            raise UnsupportedMimeTypeError(
+                "No files could be processed. Unsupported file type(s): "
+                + ", ".join(skipped_files)
+            )
+
         return file_hashes
 
     @classmethod
diff --git a/backend/workflow_manager/endpoint_v2/tests/test_un3016_unsupported_mime_skip.py b/backend/workflow_manager/endpoint_v2/tests/test_un3016_unsupported_mime_skip.py
new file mode 100644
index 0000000000..55e307e87a
--- /dev/null
+++ b/backend/workflow_manager/endpoint_v2/tests/test_un3016_unsupported_mime_skip.py
@@ -0,0 +1,131 @@
+"""UN-3016: a file rejected for an unsupported MIME type must not reach a worker.
+
+``SourceConnector.add_input_file_to_api_storage`` deliberately does not stage
+the bytes of a file whose MIME type is not allowed. It used to return that file
+anyway, with ``is_executed=True`` and a placeholder hash; nothing downstream
+filters on ``is_executed``, so a worker picked the file up, failed on the file
+that was never written, and the whole execution died with an opaque
+"Execution: <path>; Destination: <path>" message. These tests pin the fix: the
+rejected file is excluded from the returned mapping, the supported files in the
+same request still go through, and an all-rejected request fails loudly with a
+message that names what was rejected.
+
+DB-free by construction: the ORM boundary (``Workflow.objects.get``) and the
+file storage are patched, so nothing here needs Postgres. Files are real
+``SimpleUploadedFile`` objects rather than mocks — a mock's ``content_type`` is
+not in ``AllowedFileTypes``, so a mocked "supported" file would silently take
+the skip branch and the test would pass for the wrong reason.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import django
+import pytest
+from django.apps import apps
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test")
+if not apps.ready:
+    django.setup()
+
+from django.core.files.uploadedfile import SimpleUploadedFile  # noqa: E402
+from workflow_manager.endpoint_v2.exceptions import (  # noqa: E402
+    UnsupportedMimeTypeError,
+)
+from workflow_manager.endpoint_v2.source import SourceConnector  # noqa: E402
+
+SUPPORTED_MIME = "application/pdf"
+UNSUPPORTED_MIME = "application/x-msdownload"
+
+
+def _upload(name: str, content_type: str) -> SimpleUploadedFile:
+    """A real uploaded file: gives .name, .content_type, .size and .chunks()."""
+    return SimpleUploadedFile(name, b"some bytes", content_type=content_type)
+
+
+def _stage(file_objs: list[SimpleUploadedFile]) -> dict:
+    """Call the staging helper with every external boundary patched out."""
+    with (
+        patch("workflow_manager.endpoint_v2.source.UserContext") as mock_user_context,
+        patch("workflow_manager.endpoint_v2.source.WorkflowLog"),
+        patch("workflow_manager.endpoint_v2.source.Workflow") as mock_workflow,
+        patch("workflow_manager.endpoint_v2.source.FileSystem"),
+        patch.object(
+            SourceConnector,
+            "get_api_storage_dir_path",
+            return_value="unstract/api/org/exec-1",
+        ),
+    ):
+        mock_user_context.get_organization_identifier.return_value = "org"
+        mock_workflow.objects.get.return_value = MagicMock()
+        return SourceConnector.add_input_file_to_api_storage(
+            pipeline_id="pipeline-1",
+            workflow_id="workflow-1",
+            execution_id="exec-1",
+            file_objs=file_objs,
+        )
+
+
+def test_partial_skip_returns_only_the_supported_files():
+    """The core fix: a rejected file is absent from the returned mapping, and
+    the supported file alongside it is unaffected.
+
+    Also pins that a partial skip does NOT raise — rejecting some files while
+    others are runnable proceeds with what can be run (UN-4055 tracks whether
+    that should stay the behaviour).
+    """
+    file_hashes = _stage(
+        [
+            _upload("good.pdf", SUPPORTED_MIME),
+            _upload("bad.exe", UNSUPPORTED_MIME),
+        ]
+    )
+
+    assert "bad.exe" not in file_hashes
+    assert "good.pdf" in file_hashes
+    assert len(file_hashes) == 1
+    assert file_hashes["good.pdf"].mime_type == SUPPORTED_MIME
+    assert file_hashes["good.pdf"].file_name == "good.pdf"
+
+
+def test_supported_file_is_staged_with_a_real_hash():
+    """An accepted file's FileHash carries the sha256 of the bytes that were
+    staged.
+
+    Characterisation only: this held before the fix too, and no mutation of the
+    fix makes it fail. It is here to document what a returned entry looks like,
+    which is what makes the rejected file's absence elsewhere meaningful.
+    """
+    file_hashes = _stage([_upload("good.pdf", SUPPORTED_MIME)])
+
+    assert set(file_hashes) == {"good.pdf"}
+    file_hash = file_hashes["good.pdf"].file_hash
+    assert len(file_hash) == 64
+    assert all(c in "0123456789abcdef" for c in file_hash)
+
+
+def test_total_skip_raises_naming_the_skipped_files():
+    """When nothing survives the filter there is nothing to run: fail with the
+    reason instead of dispatching an empty execution that reports success.
+    """
+    with pytest.raises(UnsupportedMimeTypeError) as excinfo:
+        _stage(
+            [
+                _upload("bad.exe", UNSUPPORTED_MIME),
+                _upload("worse.dll", UNSUPPORTED_MIME),
+            ]
+        )
+
+    message = str(excinfo.value)
+    assert "bad.exe" in message
+    assert "worse.dll" in message
+    assert UNSUPPORTED_MIME in message
+
+
+def test_no_files_at_all_returns_empty_without_raising():
+    """An empty request has no skipped files, so it is not an unsupported-type
+    failure — the raise is guarded on there being something skipped.
+    """
+    assert _stage([]) == {}
diff --git a/backend/workflow_manager/workflow_v2/tests/test_un3016_execute_staging_cleanup.py b/backend/workflow_manager/workflow_v2/tests/test_un3016_execute_staging_cleanup.py
new file mode 100644
index 0000000000..e6f96ddb0b
--- /dev/null
+++ b/backend/workflow_manager/workflow_v2/tests/test_un3016_execute_staging_cleanup.py
@@ -0,0 +1,96 @@
+"""UN-3016: a staging failure must clean up the API storage directory.
+
+``WorkflowViewSet.execute`` stages uploaded files before running the workflow.
+Staging can now fail part-way — ``add_input_file_to_api_storage`` raises
+``UnsupportedMimeTypeError`` when every uploaded file is rejected, and it may
+have written some files before reaching that point. The staging call therefore
+sits inside the ``try`` whose handler calls ``delete_api_storage_dir``, and that
+handler's guard (``has_uploads``) is exactly the condition under which staging
+ran at all. These tests pin both halves: cleanup happens when staging fails,
+and cleanup is not attempted for a request that never staged anything.
+
+DB-free: the serializer, the workflow lookup and both connectors are patched,
+so ``execute`` is exercised as pure control flow.
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import ExitStack
+from unittest.mock import MagicMock, patch
+
+import django
+import pytest
+from django.apps import apps
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings.test")
+if not apps.ready:
+    django.setup()
+
+from workflow_manager.endpoint_v2.exceptions import (  # noqa: E402
+    UnsupportedMimeTypeError,
+)
+from workflow_manager.workflow_v2.views import WorkflowViewSet  # noqa: E402
+
+WORKFLOW_ID = "workflow-1"
+EXECUTION_ID = "exec-1"
+
+VIEWS = "workflow_manager.workflow_v2.views"
+
+
+def _request(with_files: bool) -> MagicMock:
+    request = MagicMock()
+    request.FILES.getlist.return_value = [MagicMock()] if with_files else []
+    return request
+
+
+def _patched_serializer(stack):
+    """Patch the serializer so execute() gets ids without parsing a payload."""
+    serializer_cls = stack.enter_context(patch(f"{VIEWS}.ExecuteWorkflowSerializer"))
+    serializer = serializer_cls.return_value
+    serializer.get_workflow_id.return_value = WORKFLOW_ID
+    serializer.get_execution_id.return_value = EXECUTION_ID
+    serializer.get_execution_action.return_value = None
+    return serializer
+
+
+def test_staging_failure_deletes_the_api_storage_dir():
+    """A staging failure leaves already-written files behind unless the handler
+    cleans up, so the failure must reach ``delete_api_storage_dir``.
+    """
+    with ExitStack() as stack:
+        _patched_serializer(stack)
+        source = stack.enter_context(patch(f"{VIEWS}.SourceConnector"))
+        destination = stack.enter_context(patch(f"{VIEWS}.DestinationConnector"))
+        source.add_input_file_to_api_storage.side_effect = UnsupportedMimeTypeError(
+            "No files could be processed. Unsupported file type(s): 'bad.exe'"
+        )
+
+        with pytest.raises(UnsupportedMimeTypeError):
+            WorkflowViewSet().execute(_request(with_files=True))
+
+        destination.delete_api_storage_dir.assert_called_once_with(
+            workflow_id=WORKFLOW_ID, execution_id=EXECUTION_ID
+        )
+
+
+def test_no_cleanup_when_the_request_staged_nothing():
+    """A request with no uploads never created a storage dir; a later failure
+    must not try to delete one.
+    """
+    with ExitStack() as stack:
+        _patched_serializer(stack)
+        stack.enter_context(patch(f"{VIEWS}.SourceConnector"))
+        destination = stack.enter_context(patch(f"{VIEWS}.DestinationConnector"))
+        stack.enter_context(
+            patch.object(
+                WorkflowViewSet,
+                "get_workflow_by_id",
+                side_effect=RuntimeError("boom"),
+            )
+        )
+
+        with pytest.raises(RuntimeError):
+            WorkflowViewSet().execute(_request(with_files=False))
+
+        destination.delete_api_storage_dir.assert_not_called()
diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py
index fefba8c21a..a6ed5ae804 100644
--- a/backend/workflow_manager/workflow_v2/views.py
+++ b/backend/workflow_manager/workflow_v2/views.py
@@ -258,17 +258,21 @@ def execute(
         use_file_history: bool = True
 
         hashes_of_files: dict[str, FileHash] = {}
-        if file_objs and execution_id and workflow_id:
+        has_uploads = bool(file_objs and execution_id and workflow_id)
+        if has_uploads:
             use_file_history = False
-            hashes_of_files = SourceConnector.add_input_file_to_api_storage(
-                pipeline_id=pipeline_guid,
-                workflow_id=workflow_id,
-                execution_id=execution_id,
-                file_objs=file_objs,
-                use_file_history=False,
-            )
 
         try:
+            # Staged inside this try so the handler below cleans up after a
+            # partial stage: its guard is exactly this staging condition.
+            if has_uploads:
+                hashes_of_files = SourceConnector.add_input_file_to_api_storage(
+                    pipeline_id=pipeline_guid,
+                    workflow_id=workflow_id,
+                    execution_id=execution_id,
+                    file_objs=file_objs,
+                    use_file_history=False,
+                )
             workflow = self.get_workflow_by_id(workflow_id=workflow_id)
             execution_response = self.execute_workflow(
                 workflow=workflow,
@@ -290,7 +294,7 @@ def execute(
             )
         except Exception as exception:
             logger.error(f"Error while executing workflow: {exception}", exc_info=True)
-            if file_objs and execution_id and workflow_id:
+            if has_uploads:
                 DestinationConnector.delete_api_storage_dir(
                     workflow_id=workflow_id, execution_id=execution_id
                 )
diff --git a/workers/callback/tasks.py b/workers/callback/tasks.py
index 94297ccd42..0deca72af1 100644
--- a/workers/callback/tasks.py
+++ b/workers/callback/tasks.py
@@ -252,12 +252,53 @@ def _get_performance_stats() -> dict:
     return stats
 
 
+# WorkflowExecution.error_message is a CharField(256) that truncates SILENTLY,
+# so the summary is capped here instead of losing its tail in the database.
+# Must match EXECUTION_ERROR_LENGTH in
+# backend/workflow_manager/workflow_v2/models/execution.py; workers cannot
+# import backend models across the service boundary, so the value is duplicated
+# deliberately (same convention as backend/.../workflow_v2/undispatched_sweep.py).
+_EXECUTION_ERROR_MAX_LENGTH = 256
+_MAX_ERRORS_IN_SUMMARY = 3
+
+
+def _summarize_file_errors(aggregated_results: dict[str, Any], total_files: int) -> str:
+    """Build an execution-level reason from the per-file errors.
+
+    The execution row previously recorded ERROR with a blank error_message,
+    leaving users with a failed run and no explanation (UN-3016). The per-file
+    errors are already aggregated as {file_name: error}; surface them here.
+    """
+    errors: dict[str, Any] = aggregated_results.get("errors") or {}
+    # `errors` is keyed by file name, so every entry is distinct already; the
+    # cap is what keeps a large batch from bloating the column.
+    entries = [
+        f"{file_name}: {str(error).strip()}"
+        for file_name, error in errors.items()
+        if error and str(error).strip()
+    ]
+    if not entries:
+        return f"All {total_files} file(s) failed."
+
+    shown = entries[:_MAX_ERRORS_IN_SUMMARY]
+    summary = f"All {total_files} file(s) failed. " + " | ".join(shown)
+    remaining = len(entries) - len(shown)
+    if remaining > 0:
+        summary += f" | (+{remaining} more)"
+
+    if len(summary) > _EXECUTION_ERROR_MAX_LENGTH:
+        # Trim with an explicit ellipsis so a cut message is visibly incomplete
+        # rather than silently losing its tail in the database.
+        summary = summary[: _EXECUTION_ERROR_MAX_LENGTH - 3] + "..."
+    return summary
+
+
 def _determine_execution_status_unified(
     file_batch_results: list[dict[str, Any]],
     api_client: InternalAPIClient,
     execution_id: str,
     organization_id: str,
-) -> tuple[dict[str, Any], str, int]:
+) -> tuple[dict[str, Any], str, int, str | None]:
     """Unified status determination logic with timeout detection for all callback types.
 
     This function combines the logic from both process_batch_callback_api and
@@ -271,7 +312,9 @@ def _determine_execution_status_unified(
         organization_id: Organization context
 
     Returns:
-        Tuple of (aggregated_results, final_status, expected_files)
+        Tuple of (aggregated_results, final_status, expected_files, error_message).
+        error_message is None unless final_status is ERROR, in which case it
+        summarises why so the execution row records a reason instead of a blank.
     """
     # Step 1: Aggregate results from all file batches using existing helper
     aggregated_results = aggregate_file_batch_results(file_batch_results)
@@ -328,9 +371,15 @@ def _determine_execution_status_unified(
         total_files == 0 and total_files_processed == 0 and expected_files > 0
     )
 
+    error_message: str | None = None
+
     if has_timeout_failure:
         # Timeout or complete failure - mark as ERROR
         final_status = ExecutionStatus.ERROR.value
+        error_message = (
+            f"Expected {expected_files} file(s) but none were processed "
+            f"(likely a timeout or worker failure)."
+        )
         logger.error(
             f"Execution {execution_id} failed - expected {expected_files} files "
             f"but processed 0 (likely timeout/failure)"
@@ -338,6 +387,7 @@ def _determine_execution_status_unified(
     elif failed_files > 0 and failed_files == total_files:
         # ALL processed files failed - mark as ERROR
         final_status = ExecutionStatus.ERROR.value
+        error_message = _summarize_file_errors(aggregated_results, total_files)
         logger.error(f"Execution {execution_id} failed - all {total_files} files failed")
     else:
         # Some or all files succeeded, or legitimate empty batch - mark as COMPLETED
@@ -355,7 +405,7 @@ def _determine_execution_status_unified(
                 f"Execution {execution_id} completed successfully - {successful_files} files processed"
             )
 
-    return aggregated_results, final_status, expected_files
+    return aggregated_results, final_status, expected_files, error_message
 
 
 def _update_execution_status_unified(
@@ -1470,7 +1520,7 @@ def _process_batch_callback_core(
 
             try:
                 # Use unified status determination with timeout detection (shared with API callback)
-                aggregated_results, execution_status, expected_files = (
+                aggregated_results, execution_status, expected_files, status_error = (
                     _determine_execution_status_unified(
                         file_batch_results=results,
                         api_client=context.api_client,
@@ -1485,7 +1535,7 @@ def _process_batch_callback_core(
                     final_status=execution_status,
                     aggregated_results=aggregated_results,
                     organization_id=context.organization_id,
-                    error_message=None,
+                    error_message=status_error,
                     is_pg=is_pg,
                 )
                 # Handle pipeline updates using unified function (non-API deployment)
@@ -1540,7 +1590,7 @@ def _process_batch_callback_core(
                         workflow_id=context.workflow_id,
                         pipeline_name=context.pipeline_name,
                         pipeline_type=context.pipeline_type,
-                        error_message=None,
+                        error_message=status_error,
                     )
                     callback_result["notification_result"] = notification_result
                 except Exception as notif_error:
@@ -1730,7 +1780,7 @@ def process_batch_callback_api(
                         pipeline_type = PipelineType.API.value
 
                 # Use unified status determination with timeout detection
-                aggregated_results, execution_status, expected_files = (
+                aggregated_results, execution_status, expected_files, status_error = (
                     _determine_execution_status_unified(
                         file_batch_results=file_batch_results,
                         api_client=api_client,
@@ -1746,6 +1796,7 @@ def process_batch_callback_api(
                     final_status=execution_status,
                     aggregated_results=aggregated_results,
                     organization_id=organization_id,
+                    error_message=status_error,
                     is_pg=is_pg,
                 )
 
@@ -1806,7 +1857,7 @@ def process_batch_callback_api(
                     workflow_id=workflow_id,
                     pipeline_name=pipeline_name,
                     pipeline_type=pipeline_type,
-                    error_message=None,
+                    error_message=status_error,
                 )
 
                 callback_result = {
diff --git a/workers/shared/enums/file_types.py b/workers/shared/enums/file_types.py
index f49b7880e3..a353e11b9b 100644
--- a/workers/shared/enums/file_types.py
+++ b/workers/shared/enums/file_types.py
@@ -37,6 +37,9 @@ class AllowedFileTypes(Enum):
     DOC = "application/msword"
     XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
     XLS = "application/vnd.ms-excel"
+    # Kept in step with the backend's AllowedFileTypes (UN-3016): the two lists
+    # must agree or a file the API accepts is rejected again inside the worker.
+    XLSM = "application/vnd.ms-excel.sheet.macroenabled.12"
     PPTX = "application/vnd.openxmlformats-officedocument.presentationml.presentation"
     PPT = "application/vnd.ms-powerpoint"
 
diff --git a/workers/tests/test_un3016_execution_error.py b/workers/tests/test_un3016_execution_error.py
new file mode 100644
index 0000000000..124460ffb8
--- /dev/null
+++ b/workers/tests/test_un3016_execution_error.py
@@ -0,0 +1,192 @@
+"""UN-3016: an ERROR execution must record WHY it failed.
+
+The execution row used to be written with status=ERROR and a blank
+error_message, so a failed run gave the user no reason at all. These tests
+pin the summary helper that now supplies that reason, and assert the reason
+survives the call that builds it.
+
+conftest loads .env.test before collection, so importing callback.tasks here
+works the same way it does in test_pg_callback_duplicate_guard.py.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import callback.tasks as _tasks_module
+from callback.tasks import (
+    _EXECUTION_ERROR_MAX_LENGTH as MAX_LEN,
+)
+from callback.tasks import (
+    _determine_execution_status_unified,
+)
+from callback.tasks import (
+    _summarize_file_errors as summarize,
+)
+
+
+def test_real_un3016_error_is_surfaced():
+    """The actual Moody's failure shape must produce a non-blank reason."""
+    aggregated = {
+        "errors": {
+            "Villa Bella.xlsm": (
+                "Workflow error: Execution: unstract/api/org_x/e/x.xlsm; "
+                "Destination: unstract/execution/org_x/e/METADATA.json"
+            )
+        }
+    }
+    result = summarize(aggregated, 1)
+    assert result.strip()
+    assert "Villa Bella.xlsm" in result
+    assert "Workflow error" in result
+
+
+@pytest.mark.parametrize(
+    "aggregated",
+    [
+        {},
+        {"errors": {}},
+        {"errors": {"a.pdf": "", "b.pdf": None}},
+    ],
+)
+def test_never_returns_blank(aggregated):
+    """Whatever the input, the execution must never get an empty reason."""
+    assert summarize(aggregated, 2).strip()
+
+
+def test_reports_each_failing_file():
+    result = summarize({"errors": {"a.pdf": "boom", "b.pdf": "bang"}}, 2)
+    assert "a.pdf" in result
+    assert "b.pdf" in result
+
+
+def test_caps_number_of_reported_errors():
+    errors = {f"f{i}.pdf": f"err{i}" for i in range(10)}
+    result = summarize({"errors": errors}, 10)
+    assert "more)" in result
+    assert result.count("f9.pdf") == 0  # beyond the cap
+
+
+def test_fits_the_database_column():
+    """error_message is CharField(256) and truncates SILENTLY."""
+    errors = {f"file_{i}_{'x' * 80}.pdf": "y" * 200 for i in range(5)}
+    result = summarize({"errors": errors}, 5)
+    assert len(result) <= MAX_LEN
+    assert result.endswith("...")
+
+
+def _all_failed_batch():
+    """One batch, one file, that file errored — the UN-3016 shape."""
+    return [
+        {
+            "total_files": 1,
+            "successful_files": 0,
+            "failed_files": 1,
+            "execution_time": 1.0,
+            "file_results": [
+                {
+                    "status": "error",
+                    "file_name": "Villa Bella.xlsm",
+                    "error": "Workflow error: Execution: unstract/api/org_x/e/x.xlsm",
+                }
+            ],
+        }
+    ]
+
+
+def test_status_function_returns_a_reason():
+    """The real call must hand back a non-blank reason, not just a 4-tuple.
+
+    This is what the defect actually was: the tuple gained a fourth slot but
+    an always-None fourth slot would still leave the execution row blank, so
+    assert on the value rather than on the shape.
+    """
+    api_client = MagicMock()
+    with patch(
+        "callback.tasks.WallClockTimeCalculator.calculate_execution_time",
+        return_value=1.0,
+    ):
+        _, final_status, _, error_message = _determine_execution_status_unified(
+            file_batch_results=_all_failed_batch(),
+            api_client=api_client,
+            execution_id="e-1",
+            organization_id="org-1",
+        )
+
+    assert final_status == "ERROR"
+    assert error_message, "an ERROR execution must carry a reason (UN-3016)"
+    assert "Villa Bella.xlsm" in error_message
+    assert len(error_message) <= MAX_LEN
+
+
+def test_timeout_failure_also_returns_a_reason():
+    """The other ERROR branch must carry a reason too.
+
+    _determine_execution_status_unified marks ERROR from two places; a blank
+    error_message from either one is the UN-3016 defect. This covers the
+    timeout branch: files were expected but no batch result came back at all.
+    """
+    api_client = MagicMock()
+    api_client.get_workflow_execution.return_value.success = True
+    api_client.get_workflow_execution.return_value.data = {"total_files": 3}
+
+    with patch(
+        "callback.tasks.WallClockTimeCalculator.calculate_execution_time",
+        return_value=0.0,
+    ):
+        _, final_status, expected_files, error_message = (
+            _determine_execution_status_unified(
+                file_batch_results=[],
+                api_client=api_client,
+                execution_id="e-2",
+                organization_id="org-1",
+            )
+        )
+
+    assert final_status == "ERROR"
+    assert expected_files == 3
+    assert error_message, "a timed-out execution must carry a reason (UN-3016)"
+    assert "3" in error_message
+    assert len(error_message) <= MAX_LEN
+
+
+def test_no_caller_passes_a_hardcoded_none_error():
+    """Regression guard: the blank error_message was the UN-3016 defect.
+
+    Scoped to the two callback bodies that consume the status tuple, and matched
+    on the AST rather than on the source text, so an unrelated keyword default
+    elsewhere in the module cannot trip it. Detects the literal `error_message=None`
+    keyword only — a positional None, an indirected variable, or a `**kwargs`
+    splat would pass; all three defect sites were the literal form.
+
+    Source-shape rather than behavioural because it guards the *call sites*:
+    the behavioural cover for the value itself is
+    test_status_function_returns_a_reason above.
+    """
+    import ast
+    from pathlib import Path
+
+    tasks_py = Path(_tasks_module.__file__)
+    tree = ast.parse(tasks_py.read_text())
+    callers = [
+        n
+        for n in ast.walk(tree)
+        if isinstance(n, ast.FunctionDef)
+        and n.name in {"_process_batch_callback_core", "process_batch_callback_api"}
+    ]
+    assert len(callers) == 2, "both callback entry points must exist"
+
+    offenders = [
+        f"{fn.name}:{kw.value.lineno}"
+        for fn in callers
+        for call in ast.walk(fn)
+        if isinstance(call, ast.Call)
+        for kw in call.keywords
+        if kw.arg == "error_message"
+        and isinstance(kw.value, ast.Constant)
+        and kw.value.value is None
+    ]
+    assert not offenders, (
+        f"error_message=None reintroduced at {offenders}; the execution row would "
+        "record ERROR with no reason (UN-3016)"
+    )