From 89a52560b4d1dfa33f557b49e6af0f8e27097632 Mon Sep 17 00:00:00 2001 From: Hari John Kuriakose Date: Sat, 29 Aug 2026 23:09:04 +0530 Subject: [PATCH 1/6] UN-3016 [FIX] Record why an execution failed; stop dispatching skipped files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A customer (Moody's) execution ended in ERROR with no explanation. Verified against the prod row for 30b84e4a-8675-4e93-a235-8d5dbac89be8: status=ERROR, error_message='' (blank), total_files=1. Three distinct defects, fixed here: 1. Execution errors were recorded blank. _determine_execution_status_unified() decided ERROR purely from failure counts and returned no reason, so all three call sites passed error_message=None. WorkflowExecution.update_execution() guards on `if error:`, so error_message was never written. It now returns a reason summarised from the per-file errors already present in aggregated_results, capped to the CharField(256) that otherwise truncates silently. 2. A file skipped for an unsupported MIME type was still handed to a worker. The skip deliberately never wrote the file's bytes, but returned it with is_executed=True and a "temp-hash-" sentinel. Nothing downstream filters on is_executed, so a worker ran, failed on the missing file, and produced the opaque "Execution: ; Destination: " error seen in the incident row (that string is built by interpolating two FileNotFoundError paths, not by mangling a MIME message). Skipped files are now excluded from the staged set; if every file is skipped the request fails with a 400 naming the unsupported types instead of dispatching an empty execution. 3. The workers' AllowedFileTypes lacked XLSM while the backend had it. The two MIME lists are now identical, so a file the API accepts cannot be rejected again inside the worker. Tests: workers/tests/test_un3016_execution_error.py, 9 passing — covers the real incident error shape, the never-blank guarantee, the 256-char column fit, and a regression guard that no caller reintroduces error_message=None. Note: XLSM was absent from the backend enum at the time of the incident and has since been added, so the customer's .xlsm would be accepted today; the defects above remain for any other unsupported type. --- .../workflow_manager/endpoint_v2/source.py | 44 +++++--- workers/callback/tasks.py | 73 ++++++++++-- workers/shared/enums/file_types.py | 3 + workers/tests/test_un3016_execution_error.py | 106 ++++++++++++++++++ 4 files changed, 203 insertions(+), 23 deletions(-) create mode 100644 workers/tests/test_un3016_execution_error.py diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py index 23b0d04f2c..de6dfa783b 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, kept only to + # report them; they are never staged and never handed to a worker. + skipped_files: dict[str, 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: ; Destination: " 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[file_name] = mime_type continue file_system = FileSystem(FileStorageType.API_EXECUTION) @@ -1285,6 +1286,19 @@ 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: + details = ", ".join( + f"'{name}' ({mime})" for name, mime in skipped_files.items() + ) + raise UnsupportedMimeTypeError( + f"No files could be processed. Unsupported file type(s): {details}" + ) + return file_hashes @classmethod diff --git a/workers/callback/tasks.py b/workers/callback/tasks.py index 94297ccd42..e6844a0285 100644 --- a/workers/callback/tasks.py +++ b/workers/callback/tasks.py @@ -252,12 +252,59 @@ 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. +_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 {} + if not errors: + return f"All {total_files} file(s) failed." + + # Report the distinct reasons rather than repeating an identical message + # once per file; cap the detail so a large batch cannot bloat the column. + seen: list[str] = [] + for file_name, error in errors.items(): + detail = str(error).strip() if error else "" + if not detail: + continue + entry = f"{file_name}: {detail}" + if entry not in seen: + seen.append(entry) + + if not seen: + return f"All {total_files} file(s) failed." + + summary = f"All {total_files} file(s) failed. " + shown = seen[:_MAX_ERRORS_IN_SUMMARY] + summary += " | ".join(shown) + remaining = len(seen) - 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 +318,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 +377,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 +393,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 +411,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 +1526,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 +1541,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 +1596,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 +1786,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 +1802,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 +1863,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..a330213755 --- /dev/null +++ b/workers/tests/test_un3016_execution_error.py @@ -0,0 +1,106 @@ +"""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. + +The helper is extracted with `ast` rather than imported, because importing +workers.callback.tasks pulls in celery and the whole worker runtime. +""" + +import ast +from pathlib import Path + +import pytest + +_TASKS = Path(__file__).resolve().parents[1] / "callback" / "tasks.py" + + +def _load_helper(): + """Exec just the helper and its constants out of callback/tasks.py.""" + tree = ast.parse(_TASKS.read_text()) + ns: dict = {"Any": object} + for node in tree.body: + if isinstance(node, ast.Assign) and any( + getattr(t, "id", "").startswith(("_MAX_ERRORS", "_EXECUTION_ERROR")) + for t in node.targets + ): + exec(compile(ast.Module([node], []), "", "exec"), ns) + elif isinstance(node, ast.FunctionDef) and node.name == "_summarize_file_errors": + exec(compile(ast.Module([node], []), "", "exec"), ns) + return ns["_summarize_file_errors"], ns["_EXECUTION_ERROR_MAX_LENGTH"] + + +summarize, MAX_LEN = _load_helper() + + +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 test_status_function_returns_a_reason(): + """_determine_execution_status_unified must return a 4-tuple.""" + tree = ast.parse(_TASKS.read_text()) + fn = next( + n + for n in ast.walk(tree) + if isinstance(n, ast.FunctionDef) + and n.name == "_determine_execution_status_unified" + ) + returns = [n for n in ast.walk(fn) if isinstance(n, ast.Return)] + assert returns, "function must return" + assert all( + isinstance(r.value, ast.Tuple) and len(r.value.elts) == 4 for r in returns + ), "every return must carry (results, status, expected_files, error_message)" + + +def test_no_caller_passes_a_hardcoded_none_error(): + """Regression guard: the blank error_message was the UN-3016 defect.""" + source = _TASKS.read_text() + assert "error_message=None" not in source From c3aa791bc2d6fb2fdece0e695f5ed371c1879226 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:40:15 +0000 Subject: [PATCH 2/6] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- workers/callback/tasks.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/workers/callback/tasks.py b/workers/callback/tasks.py index e6844a0285..e55bb6a769 100644 --- a/workers/callback/tasks.py +++ b/workers/callback/tasks.py @@ -258,9 +258,7 @@ def _get_performance_stats() -> dict: _MAX_ERRORS_IN_SUMMARY = 3 -def _summarize_file_errors( - aggregated_results: dict[str, Any], total_files: int -) -> str: +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, From 9e1adc4ca8b204430640ae92c3720bbec2e210ae Mon Sep 17 00:00:00 2001 From: Hari John Kuriakose Date: Mon, 31 Aug 2026 00:40:47 +0530 Subject: [PATCH 3/6] UN-3016 [FIX] Clean up staging failures; scope the error_message guard Remediation of review findings on PR #2256. F1 (High): the new UnsupportedMimeTypeError raised by add_input_file_to_api_storage escaped the try/except in WorkflowViewSet.execute that owns delete_api_storage_dir, so a partial stage could leave written files behind. The staging call now has its own handler mirroring the one at the end of the method. The sibling caller (api_v2/deployment_helper.py:279) was already guarded. F2 (High): test_no_caller_passes_a_hardcoded_none_error asserted "error_message=None" not in the whole 1900-line tasks.py. It matched nothing at the sites it meant to guard and would trip on any unrelated keyword default. Now an AST check scoped to _process_batch_callback_core and process_batch_callback_api. Mutation-checked: reintroducing error_message=None at process_batch_callback_api fails the test with that call site's line number. F4 (Medium): _EXECUTION_ERROR_MAX_LENGTH now names its authority, EXECUTION_ERROR_LENGTH in backend workflow_v2/models/execution.py, following the convention in workflow_v2/undispatched_sweep.py. Adversarial verification also found two comments I had written asserting mechanisms the code does not have (the storage dir is a computed path, not a mkdir; the AST check does not catch positional/indirected None). Both false claims deleted rather than reworded. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NQhqNqCwFcXZZ7cQU6HxUE --- backend/workflow_manager/workflow_v2/views.py | 28 +++++++++++---- workers/callback/tasks.py | 4 +++ workers/tests/test_un3016_execution_error.py | 34 +++++++++++++++++-- 3 files changed, 56 insertions(+), 10 deletions(-) diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py index fefba8c21a..154c7aafaa 100644 --- a/backend/workflow_manager/workflow_v2/views.py +++ b/backend/workflow_manager/workflow_v2/views.py @@ -260,13 +260,27 @@ def execute( hashes_of_files: dict[str, FileHash] = {} if file_objs and execution_id and workflow_id: 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, - ) + # Staging sits outside the main try/except below, so it needs its own + # cleanup: a partial stage can leave already-written files behind. + # Mirrors the handler at the end of this method. + try: + 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, + ) + except Exception as exception: + logger.error( + f"Error while staging files for execution {execution_id}: " + f"{exception}", + exc_info=True, + ) + DestinationConnector.delete_api_storage_dir( + workflow_id=workflow_id, execution_id=execution_id + ) + raise try: workflow = self.get_workflow_by_id(workflow_id=workflow_id) diff --git a/workers/callback/tasks.py b/workers/callback/tasks.py index e55bb6a769..a96c4cfe8a 100644 --- a/workers/callback/tasks.py +++ b/workers/callback/tasks.py @@ -254,6 +254,10 @@ def _get_performance_stats() -> dict: # 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 diff --git a/workers/tests/test_un3016_execution_error.py b/workers/tests/test_un3016_execution_error.py index a330213755..835f7f4332 100644 --- a/workers/tests/test_un3016_execution_error.py +++ b/workers/tests/test_un3016_execution_error.py @@ -101,6 +101,34 @@ def test_status_function_returns_a_reason(): def test_no_caller_passes_a_hardcoded_none_error(): - """Regression guard: the blank error_message was the UN-3016 defect.""" - source = _TASKS.read_text() - assert "error_message=None" not in source + """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. + """ + tree = ast.parse(_TASKS.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)" + ) From af64650ca96d28f40864b299b3d20c1868c9c2b3 Mon Sep 17 00:00:00 2001 From: Hari John Kuriakose Date: Mon, 31 Aug 2026 00:57:02 +0530 Subject: [PATCH 4/6] UN-3016 [CLEANUP] Simplify per /simplify; test the value, not the source shape MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Behaviour-preserving cleanup, run once after the review verdict settled. - views.py: drop the second cleanup handler added in the previous commit and stage inside the existing try instead. That handler's guard is exactly the staging condition, so one handler now covers both paths rather than two copies of the same contract. - test_un3016_execution_error.py: import callback.tasks directly instead of extracting the helper with ast/exec. The docstring's claim that importing pulls in an unusable celery runtime is false — conftest loads .env.test before collection, and test_pg_callback_duplicate_guard.py already imports the module at module level. Verified by running the import under pytest. test_status_function_returns_a_reason is now behavioural: it calls _determine_execution_status_unified and asserts the reason is non-blank. The old version asserted only that every return was a 4-tuple, which would have passed with an always-None fourth element — i.e. it could not detect the very defect it was named for. Mutation-checked: forcing error_message = None now fails the test. - _summarize_file_errors: errors is keyed by file name, so entries are distinct by construction and the `entry not in seen` dedup could never fire. Removed, along with the duplicate early return it guarded. - source.py: skipped_files was a dict never used as a mapping; now a list of pre-formatted entries. Tests: 1298 passed, 132 skipped. test_pg_reaper.py deselected — it needs a live Postgres on 127.0.0.1:5432 and hangs identically on unmodified HEAD. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NQhqNqCwFcXZZ7cQU6HxUE --- .../workflow_manager/endpoint_v2/source.py | 14 +-- backend/workflow_manager/workflow_v2/views.py | 26 ++--- workers/callback/tasks.py | 30 ++--- workers/tests/test_un3016_execution_error.py | 103 +++++++++++------- 4 files changed, 90 insertions(+), 83 deletions(-) diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py index de6dfa783b..3498db05ba 100644 --- a/backend/workflow_manager/endpoint_v2/source.py +++ b/backend/workflow_manager/endpoint_v2/source.py @@ -1222,9 +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, kept only to - # report them; they are never staged and never handed to a worker. - skipped_files: dict[str, str] = {} + # 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 @@ -1252,7 +1252,7 @@ def add_input_file_to_api_storage( f"'{mime_type}'. It will not be processed." ) workflow_log.log_error(logger=logger, message=log_message) - skipped_files[file_name] = mime_type + skipped_files.append(f"'{file_name}' ({mime_type})") continue file_system = FileSystem(FileStorageType.API_EXECUTION) @@ -1292,11 +1292,9 @@ def add_input_file_to_api_storage( # which would otherwise finish as a vacuous success and leave the user # wondering why nothing happened. if skipped_files and not file_hashes: - details = ", ".join( - f"'{name}' ({mime})" for name, mime in skipped_files.items() - ) raise UnsupportedMimeTypeError( - f"No files could be processed. Unsupported file type(s): {details}" + "No files could be processed. Unsupported file type(s): " + + ", ".join(skipped_files) ) return file_hashes diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py index 154c7aafaa..a6ed5ae804 100644 --- a/backend/workflow_manager/workflow_v2/views.py +++ b/backend/workflow_manager/workflow_v2/views.py @@ -258,12 +258,14 @@ 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 - # Staging sits outside the main try/except below, so it needs its own - # cleanup: a partial stage can leave already-written files behind. - # Mirrors the handler at the end of this method. - try: + + 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, @@ -271,18 +273,6 @@ def execute( file_objs=file_objs, use_file_history=False, ) - except Exception as exception: - logger.error( - f"Error while staging files for execution {execution_id}: " - f"{exception}", - exc_info=True, - ) - DestinationConnector.delete_api_storage_dir( - workflow_id=workflow_id, execution_id=execution_id - ) - raise - - try: workflow = self.get_workflow_by_id(workflow_id=workflow_id) execution_response = self.execute_workflow( workflow=workflow, @@ -304,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 a96c4cfe8a..0deca72af1 100644 --- a/workers/callback/tasks.py +++ b/workers/callback/tasks.py @@ -270,27 +270,19 @@ def _summarize_file_errors(aggregated_results: dict[str, Any], total_files: int) errors are already aggregated as {file_name: error}; surface them here. """ errors: dict[str, Any] = aggregated_results.get("errors") or {} - if not errors: - return f"All {total_files} file(s) failed." - - # Report the distinct reasons rather than repeating an identical message - # once per file; cap the detail so a large batch cannot bloat the column. - seen: list[str] = [] - for file_name, error in errors.items(): - detail = str(error).strip() if error else "" - if not detail: - continue - entry = f"{file_name}: {detail}" - if entry not in seen: - seen.append(entry) - - if not seen: + # `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." - summary = f"All {total_files} file(s) failed. " - shown = seen[:_MAX_ERRORS_IN_SUMMARY] - summary += " | ".join(shown) - remaining = len(seen) - len(shown) + 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)" diff --git a/workers/tests/test_un3016_execution_error.py b/workers/tests/test_un3016_execution_error.py index 835f7f4332..81e5703b31 100644 --- a/workers/tests/test_un3016_execution_error.py +++ b/workers/tests/test_un3016_execution_error.py @@ -2,36 +2,27 @@ 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. +pin the summary helper that now supplies that reason, and assert the reason +survives the call that builds it. -The helper is extracted with `ast` rather than imported, because importing -workers.callback.tasks pulls in celery and the whole worker runtime. +conftest loads .env.test before collection, so importing callback.tasks here +works the same way it does in test_pg_callback_duplicate_guard.py. """ -import ast -from pathlib import Path +from unittest.mock import MagicMock, patch import pytest -_TASKS = Path(__file__).resolve().parents[1] / "callback" / "tasks.py" - - -def _load_helper(): - """Exec just the helper and its constants out of callback/tasks.py.""" - tree = ast.parse(_TASKS.read_text()) - ns: dict = {"Any": object} - for node in tree.body: - if isinstance(node, ast.Assign) and any( - getattr(t, "id", "").startswith(("_MAX_ERRORS", "_EXECUTION_ERROR")) - for t in node.targets - ): - exec(compile(ast.Module([node], []), "", "exec"), ns) - elif isinstance(node, ast.FunctionDef) and node.name == "_summarize_file_errors": - exec(compile(ast.Module([node], []), "", "exec"), ns) - return ns["_summarize_file_errors"], ns["_EXECUTION_ERROR_MAX_LENGTH"] - - -summarize, MAX_LEN = _load_helper() +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(): @@ -84,20 +75,48 @@ def test_fits_the_database_column(): 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(): - """_determine_execution_status_unified must return a 4-tuple.""" - tree = ast.parse(_TASKS.read_text()) - fn = next( - n - for n in ast.walk(tree) - if isinstance(n, ast.FunctionDef) - and n.name == "_determine_execution_status_unified" - ) - returns = [n for n in ast.walk(fn) if isinstance(n, ast.Return)] - assert returns, "function must return" - assert all( - isinstance(r.value, ast.Tuple) and len(r.value.elts) == 4 for r in returns - ), "every return must carry (results, status, expected_files, error_message)" + """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_no_caller_passes_a_hardcoded_none_error(): @@ -108,8 +127,16 @@ def test_no_caller_passes_a_hardcoded_none_error(): 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. """ - tree = ast.parse(_TASKS.read_text()) + 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) From 1ceb723a5ab50f388c780f9897e57d38f9b7f81b Mon Sep 17 00:00:00 2001 From: Hari John Kuriakose Date: Mon, 31 Aug 2026 01:14:49 +0530 Subject: [PATCH 5/6] UN-3016 [TEST] Cover the timeout ERROR branch's reason too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _determine_execution_status_unified marks ERROR from two places, and a blank error_message from either one is the UN-3016 defect. Only the failed_files == total_files branch was covered; the timeout branch (files expected, no batch result came back) had no test at all — mutating its error_message to None left the suite green. Adds test_timeout_failure_also_returns_a_reason, which mocks the api_client so get_workflow_execution reports total_files=3 with an empty file_batch_results, driving has_timeout_failure. Mutation-checked: forcing that branch's error_message to None now fails the test. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NQhqNqCwFcXZZ7cQU6HxUE --- workers/tests/test_un3016_execution_error.py | 31 ++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/workers/tests/test_un3016_execution_error.py b/workers/tests/test_un3016_execution_error.py index 81e5703b31..124460ffb8 100644 --- a/workers/tests/test_un3016_execution_error.py +++ b/workers/tests/test_un3016_execution_error.py @@ -119,6 +119,37 @@ def test_status_function_returns_a_reason(): 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. From e15e3e2aadc085fc2cb9b7015b3cf44c08c7a8f3 Mon Sep 17 00:00:00 2001 From: Hari John Kuriakose Date: Mon, 31 Aug 2026 23:08:03 +0530 Subject: [PATCH 6/6] UN-3016 [TEST] Cover the backend half: MIME skip and staging cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #2256's workers half had 8 tests; its backend half had none. Six tests close that gap, all DB-free (the ORM boundary and file storage are patched, so no Postgres is needed). source.py — add_input_file_to_api_storage: - partial skip returns only the supported files, and does NOT raise. A rejected file alongside a runnable one is simply absent from the mapping; the request proceeds with what can be run (UN-4055 tracks whether that should remain the behaviour, so this pins it rather than asserting a raise). - a total skip raises UnsupportedMimeTypeError naming every skipped file and its MIME type, instead of dispatching an empty execution that would report a vacuous success. - an empty request returns {} without raising: the guard is on something having been skipped, not merely on the mapping being empty. - an accepted file's FileHash carries the sha256 of the staged bytes. This is characterisation only — it held before the fix and no mutation of the fix makes it fail; it documents what a returned entry looks like, which is what makes the rejected file's absence meaningful. views.py — WorkflowViewSet.execute: - a staging failure reaches delete_api_storage_dir, so a partial stage does not leave written files behind. - a request that staged nothing does not attempt that cleanup. Files are real SimpleUploadedFile objects, not mocks: a MagicMock's content_type is never in AllowedFileTypes, so a mocked "supported" file would silently take the skip branch and the test would pass for the wrong reason. Mutation-checked, each mutation reverted and the restore verified on disk: - re-adding the rejected file to file_hashes with a temp hash -> partial-skip and total-skip tests fail - dropping "and not file_hashes" from the raise condition -> partial-skip test fails (it now raises on a request that has runnable files) - dropping "skipped_files and" from the raise condition -> empty-request test fails - deleting the raise block -> total-skip test fails - stripping the filenames from the error message -> total-skip test fails - moving staging back outside the try -> staging-failure test fails - dropping the has_uploads guard on cleanup -> no-cleanup test fails Five of the six tests are pinned by a mutation; the sha256 test is the characterisation case noted above. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NQhqNqCwFcXZZ7cQU6HxUE --- .../test_un3016_unsupported_mime_skip.py | 131 ++++++++++++++++++ .../test_un3016_execute_staging_cleanup.py | 96 +++++++++++++ 2 files changed, 227 insertions(+) create mode 100644 backend/workflow_manager/endpoint_v2/tests/test_un3016_unsupported_mime_skip.py create mode 100644 backend/workflow_manager/workflow_v2/tests/test_un3016_execute_staging_cleanup.py 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: ; Destination: " 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()