feat(etl-uvicorn): declare blame in failure responses instead of encoding it in status codes

Status codes carry transport semantics for the immediate caller and cannot
also carry business blame: a 422 for a malformed reserved field (composed
by the platform) and a 422 for a customer's unreadable file are different
faults wearing the same number. Failure responses now say whose fault it
is explicitly:

- the invoke envelope gains an optional `blame`, set to "user" only when
  the plugin raised the UserError family — a fault in something the
  customer owns. Absent means not-the-customer's: an orchestrator must
  never infer customer fault from the status class alone.
- middleware error bodies carry the invocation-settings taxonomy `reason`
  code alongside `detail`, so an orchestrator can recognize a
  platform-composed payload failure whatever status answered the hop.

Author: CyMule

test/api/test_api.py

@@ -24,6 +24,7 @@ class InvokeResponse(BaseModel):
     status_code: int
     filedata_meta: FileDataMeta
     status_code_text: Optional[str] = None
+    blame: Optional[str] = None
     output: Optional[Any] = None
     file_data: Optional[Union[FileData, BatchFileData]] = None
 
@@ -220,6 +221,38 @@ def test_http_exception_handling(file_data):
     assert invoke_response.status_code_text == "Not found"
 
 
+@pytest.mark.parametrize(
+    "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
+)
+def test_user_error_declares_user_blame(file_data):
+    """Only the UserError family may claim the failure is the customer's to fix."""
+    from test.assets.exception_status_code import function_raises_user_error as test_fn
+
+    client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin"))
+
+    resp = client.post("/invoke", json={"file_data": file_data.model_dump()})
+    invoke_response = InvokeResponse.model_validate(resp.json())
+
+    assert invoke_response.status_code >= 400
+    assert invoke_response.blame == "user"
+
+
+@pytest.mark.parametrize(
+    "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
+)
+def test_non_user_failures_declare_no_blame(file_data):
+    """Anything undeclared is not the customer's: an orchestrator must not infer customer fault
+    from the status code, which also carries transport semantics."""
+    from test.assets.exception_status_code import function_raises_provider_error as test_fn
+
+    client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin"))
+
+    resp = client.post("/invoke", json={"file_data": file_data.model_dump()})
+    invoke_response = InvokeResponse.model_validate(resp.json())
+
+    assert invoke_response.blame is None
+
+
 @pytest.mark.parametrize(
     "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
 )

test/api/test_invocation_middleware.py

@@ -224,7 +224,9 @@ def test_non_dict_reserved_field_is_rejected(self):
 
         assert downstream.body is None
         assert sent[0]["status"] == 422
-        assert b"invocation_settings" in sent[1]["body"]
+        body = json.loads(sent[1]["body"])
+        assert "invocation_settings" in body["detail"]
+        assert body["reason"] == "malformed_envelope"
 
     def test_context_with_an_unreadable_schema_version_fails_as_platform_error(self):
         # Absence means "older caller, use the boot settings"; a context this plugin cannot read
@@ -236,8 +238,10 @@ def test_context_with_an_unreadable_schema_version_fails_as_platform_error(self)
 
         assert downstream.body is None
         assert sent[0]["status"] == 500
-        assert b"invocation_context" in sent[1]["body"]
-        assert b"UnsupportedContextVersionError" in sent[1]["body"]
+        body = json.loads(sent[1]["body"])
+        assert "invocation_context" in body["detail"]
+        assert "UnsupportedContextVersionError" in body["detail"]
+        assert body["reason"] == "unsupported_context_version"
 
     def test_malformed_context_is_rejected(self):
         downstream, sent = _run_middleware(

test/assets/exception_status_code.py

@@ -137,3 +137,15 @@ async def async_gen_function_raises_unstructured_ingest_error_with_none_status_c
     error = UnstructuredIngestError("Async gen test UnstructuredIngestError with None status_code")
     error.status_code = None
     raise error
+
+
+def function_raises_user_error() -> None:
+    from unstructured_ingest.error import UserError
+
+    raise UserError("Customer-owned resource rejected the request")
+
+
+def function_raises_provider_error() -> None:
+    from unstructured_ingest.error import ProviderError
+
+    raise ProviderError("Upstream provider failed")

unstructured_platform_plugins/etl_uvicorn/api_generator.py

@@ -14,7 +14,7 @@
 from starlette.responses import RedirectResponse
 from typing_extensions import deprecated
 from unstructured_ingest.data_types.file_data import BatchFileData, FileData, file_data_from_dict
-from unstructured_ingest.error import UnstructuredIngestError
+from unstructured_ingest.error import UnstructuredIngestError, UserError
 from uvicorn.config import LOG_LEVELS
 from uvicorn.importer import import_from_string
 
@@ -164,6 +164,11 @@ class InvokeResponse(BaseModel):
         file_data: Optional[FileDataType] = None
         filedata_meta: Optional[filedata_meta_model] = None
         status_code_text: Optional[str] = None
+        # Who must act on a failure: "user" only when the plugin raised the UserError family —
+        # a fault in something the customer owns (their file, their credentials, their provider).
+        # Absent means not-the-customer's: an orchestrator must never infer customer fault from
+        # the status code alone, which also carries transport semantics.
+        blame: Optional[str] = None
         output: Optional[response_type] = None
         message_channels: MessageChannels = Field(default_factory=MessageChannels)
 
@@ -258,6 +263,7 @@ async def _stream_response():
                 filedata_meta=filedata_meta_model.model_validate(filedata_meta.model_dump()),
                 status_code=exc.status_code or status.HTTP_500_INTERNAL_SERVER_ERROR,
                 status_code_text=str(exc),
+                blame="user" if isinstance(exc, UserError) else None,
                 file_data=request_dict.get("file_data", None),
             )
         except Exception as invoke_error:

unstructured_platform_plugins/invocation_settings.py

@@ -33,6 +33,7 @@
     RESERVED_ENVELOPE_KEY,
     InvocationContext,
     InvocationSettingsError,
+    MalformedEnvelopeError,
     extract_context,
     http_status_for,
     resolve_invocation_settings,
@@ -196,7 +197,14 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if isinstance(parsed, dict):
             raw_settings = parsed.get(RESERVED_ENVELOPE_KEY)
             if RESERVED_ENVELOPE_KEY in parsed and not isinstance(raw_settings, dict):
-                await _send_json(send, 422, {"detail": f"Invalid field: {RESERVED_ENVELOPE_KEY}"})
+                await _send_json(
+                    send,
+                    422,
+                    {
+                        "detail": f"Invalid field: {RESERVED_ENVELOPE_KEY}",
+                        "reason": MalformedEnvelopeError.reason,
+                    },
+                )
                 return
         # Resolved even when the field — or the whole JSON object — is absent:
         # resolve_invocation_settings owns the FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS
@@ -210,11 +218,11 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             # Class name only — never envelope contents, and never the exception's own message,
             # which can embed request-controlled values.
             logger.warning("unusable %s payload: %s", RESERVED_ENVELOPE_KEY, type(exc).__name__)
-            await _send_json(
-                send,
-                http_status_for(exc),
-                {"detail": f"Unusable invocation settings: {type(exc).__name__}"},
-            )
+            body = {"detail": f"Unusable invocation settings: {type(exc).__name__}"}
+            reason = getattr(exc, "reason", None)
+            if isinstance(reason, str):
+                body["reason"] = reason
+            await _send_json(send, http_status_for(exc), body)
             return
         if isinstance(parsed, dict):
             try:
@@ -232,7 +240,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                     if status == 422
                     else f"Unusable {RESERVED_CONTEXT_KEY}: {type(exc).__name__}"
                 )
-                await _send_json(send, status, {"detail": detail})
+                await _send_json(send, status, {"detail": detail, "reason": exc.reason})
                 return
 
         # The joined body and its parsed tree can be tens of MB and are not needed past this point;