feat(etl-uvicorn): give plugins request-scoped settings and invocation context

.github/workflows/ci.yml

@@ -16,7 +16,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ "3.10", "3.11", "3.12", "3.13" ]
+        python-version: [ "3.11", "3.12", "3.13" ]
     steps:
       - uses: actions/checkout@v3
 
@@ -47,7 +47,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ "3.10", "3.11", "3.12", "3.13" ]
+        python-version: [ "3.11", "3.12", "3.13" ]
     steps:
       - uses: actions/checkout@v3
 

.github/workflows/release.yml

@@ -6,7 +6,7 @@ on:
       - published
 
 env:
-  PYTHON_VERSION: "3.10"
+  PYTHON_VERSION: "3.11"
 
 jobs:
   release:

CHANGELOG.md

@@ -1,3 +1,59 @@
+## 0.1.0
+
+* **This package now owns the `/invoke` transport for the reserved fields.**
+  `unstructured_platform_plugins.invocation_settings` holds the `/invoke` binding dependency and
+  body cap, the `/metadata` capability route, the request-scoped accessors, and `http_status_for`
+  — the HTTP spelling of the
+  library's normative `blame` → status rule. It sits on `utic-invocation-settings >=0.4.0`, which
+  owns the *settings contract* — including the field-set carrier as the only accepted sealed
+  `/invoke` shape, independent sealed-field resolution, and what an absent field is allowed to
+  mean. That split is deliberate: the
+  absence rule is a security decision and belongs next to the crypto it governs, while request
+  handling and route registration belong here, where a web framework is already a dependency.
+  Nothing about the sealed-settings wire format is decided in this repository.
+* **This package now owns the `invocation_context` identity model.**
+  `unstructured_platform_plugins.invocation_context` holds `InvocationContext`,
+  `extract_context`, `dimensions`, `RESERVED_CONTEXT_KEY`, `DIMENSION_FIELDS`,
+  `SUPPORTED_CONTEXT_VERSIONS` and `UnsupportedContextVersionError`. The context is `/invoke`
+  protocol identity — no crypto, no secrets — so it lives with the plugin protocol. Its errors
+  subclass the shared `InvocationSettingsError` taxonomy, so hosts classify context failures with
+  the same `reason`/`blame` machinery as settings failures. This module is the public home for
+  the surface `utic-invocation-settings 0.2.x` carried and its `0.3.0` removed.
+* **Every wrapped app installs it at construction.** The reserved `invocation_settings` /
+  `invocation_context` fields are handled outside the generated handler schema, the opaque
+  settings payload is delegated to `utic-invocation-settings`, and only the final resolved mapping
+  is exposed through `current_invocation_settings()` / `current_invocation_context()`.
+  An absent field preserves the existing fallback behaviour; under
+  `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` missing or plaintext settings fail closed.
+  Repeated installation is safe: the dependency installs once and the last `/metadata`
+  registration wins.
+* **Extraction is a route dependency.** `bind_invocation_envelope` reads the body the framework
+  buffered and parsed (`request.json()` is Starlette-cached), so the `/invoke` body is held and
+  decoded exactly once per request. `install_invocation_envelope` contributes that path-aware
+  dependency through the router's public dependency list before `/invoke` is registered; no
+  private FastAPI dependency graph is mutated. It also registers the failure response shape and
+  installs `InvokeBodyLimitMiddleware`, a streaming byte counter that answers 413 over the cap
+  without buffering. Async-generator plugins explicitly re-enter the captured request binding
+  inside response iteration, so streaming stays correct independently of FastAPI's yield-dependency
+  cleanup timing.
+* **Sealed settings consumption remains opt-in.** Pass
+  `invoke_with_sealed_dag_node_settings=True` to `wrap_in_fastapi` / `generate_fast_api` (or
+  `--sealed-dag-node-settings` on the CLI) only for a plugin that consumes per-invoke settings;
+  it advertises that the application accepts and acts on the versioned field-set carrier.
+  Transport support alone continues to advertise only `invocation_settings` and
+  `invocation_context`. A
+  plugin that serves a custom `/metadata` payload must register it via `add_metadata_route` (which
+  replaces the wrapper's route) — a plain `@app.get("/metadata")` added after construction is
+  shadowed by the wrapper's earlier registration.
+* **Resolution runs off the event loop.** Resolution may perform blocking cryptography for
+  independently sealed fields and this dependency fronts every invoke on the pod, so it is
+  dispatched with `asyncio.to_thread` rather than blocking the loop.
+* **Failures map through the library's blame taxonomy**, not a flat 500: only a caller-fixable
+  fault answers 422. Sealing drift, an envelope addressed to another recipient and a broken local
+  mount are all 5xx, which keeps the controller's blame classification off the customer. Responses
+  carry the error's class name and never its message, which can embed request-controlled values.
+* **Python floor is now 3.11** (required by `utic-invocation-settings`).
+
 ## 0.0.46
 
 * **Carry preflight failure categories through standard `/precheck` responses.**

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "unstructured_platform_plugins"
 description = "Wrapper to convert arbitrary code into a uvicorn/fastapi implementation for Unstructured Platform"
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 classifiers = [
     "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
@@ -10,7 +10,6 @@ classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
@@ -24,6 +23,7 @@ dependencies = [
     "fastapi",
     "click",
     "unstructured-ingest",
+    "utic-invocation-settings>=0.4.0,<1.0.0",
     "opentelemetry-instrumentation-fastapi",
     "opentelemetry-exporter-otlp-proto-grpc",
     "dataclasses-json"
@@ -86,3 +86,7 @@ fail_under = 15
 
 [tool.hatch.build.targets.sdist]
     packages = ["/unstructured_platform_plugins"]
+
+[tool.uv.sources]
+# Temporary source pin until field-atomic utic-invocation-settings 0.4.0 is published.
+utic-invocation-settings = { git = "https://github.com/Unstructured-IO/utic-public-libs", subdirectory = "libs/utic-invocation-settings", rev = "05d7bcc2b7720694904113b827c6e70fe565fc22" }

test/api/test_api.py

@@ -26,6 +26,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
 
@@ -222,6 +223,62 @@ 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
+
+
+def test_streaming_user_error_declares_user_blame():
+    """The streaming error envelope carries the same blame derivation as the non-streaming path."""
+    from test.assets.exception_status_code import (
+        async_gen_function_raises_user_error_mid_stream as test_fn,
+    )
+
+    client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin"))
+
+    resp = client.post("/invoke", json={"file_data": mock_file_data[0].model_dump()})
+
+    assert resp.status_code == 200
+    assert resp.headers["content-type"] == "application/x-ndjson"
+
+    import json
+
+    lines = resp.content.decode().strip().split("\n")
+    assert len(lines) == 2  # One yielded item, then the error envelope
+
+    assert InvokeResponse.model_validate(json.loads(lines[0])).blame is None
+    error_response = InvokeResponse.model_validate(json.loads(lines[1]))
+    assert error_response.status_code >= 400
+    assert error_response.blame == "user"
+
+
 @pytest.mark.parametrize(
     "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
 )
@@ -608,9 +665,26 @@ def test_precheck_reports_failure_category_from_raised_error():
     body = resp.json()
     assert body["status_code"] == 403
     assert body["failure_category"] == "AUTH_PERMISSION_DENIED"
+    # A plain exception is not the customer's to fix.
+    assert body["blame"] is None
     assert "credential rejected" in body["status_code_text"]
 
 
+def test_precheck_declares_user_blame_like_invoke_does():
+    from unstructured_ingest.error import UserError
+
+    def user_fault_precheck() -> None:
+        raise UserError("bad credentials")
+
+    client = TestClient(
+        wrap_in_fastapi(func=_no_params, plugin_id="mock_plugin", precheck_func=user_fault_precheck)
+    )
+
+    body = client.get("/precheck").json()
+
+    assert body["blame"] == "user"
+
+
 def test_precheck_success_has_no_failure_category():
     client = TestClient(
         wrap_in_fastapi(func=_no_params, plugin_id="mock_plugin", precheck_func=_passing_precheck)

test/api/test_invocation_context.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import pytest
+from utic_invocation_settings import (
+    Blame,
+    DecryptionError,
+    IdentityNotMountedError,
+    KeyNotFoundError,
+    MalformedDagNodeSettingsError,
+    MalformedEnvelopeError,
+    SealedDagNodeSettingsRequiredError,
+)
+
+from unstructured_platform_plugins.invocation_context import (
+    RESERVED_CONTEXT_KEY,
+    InvocationContext,
+    UnsupportedContextVersionError,
+    dimensions,
+    extract_context,
+)
+from unstructured_platform_plugins.invocation_settings import http_status_for
+
+VALID = {
+    "schema_version": "1",
+    "invocation_id": "inv-1",
+    "job_id": "job-1",
+    "tenant_id": "tenant-1",
+    "dag_node_id": "node-1",
+    "dag_node_type": "chunker",
+    "record_id": "rec-1",
+    "attempt": 2,
+}
+
+
+def test_extracts_identity_fields_from_body():
+    context = extract_context({"file_data": {"path": "x"}, RESERVED_CONTEXT_KEY: VALID})
+    assert context is not None
+    assert context.tenant_id == "tenant-1"
+    assert context.attempt == 2
+
+
+def test_absent_key_returns_none():
+    assert extract_context({"file_data": {"path": "x"}}) is None
+
+
+def test_accepts_already_parsed():
+    context = InvocationContext(**VALID)
+    assert extract_context({RESERVED_CONTEXT_KEY: context}) is context
+
+
+def test_present_but_null_fails_closed():
+    # Same rule as the envelope: a context that silently vanishes takes tenant attribution with it.
+    with pytest.raises(MalformedEnvelopeError):
+        extract_context({RESERVED_CONTEXT_KEY: None})
+
+
+def test_present_but_not_an_object_fails_closed():
+    with pytest.raises(MalformedEnvelopeError):
+        extract_context({RESERVED_CONTEXT_KEY: "tenant-1"})
+
+
+def test_unknown_schema_version_is_rejected_by_its_own_error():
+    with pytest.raises(UnsupportedContextVersionError) as exc:
+        extract_context({RESERVED_CONTEXT_KEY: {**VALID, "schema_version": "2"}})
+    assert "'2'" in str(exc.value)
+
+
+@pytest.mark.parametrize("version", [None, 2, ["1"]])
+def test_a_mistyped_schema_version_is_malformed_not_version_skew(version):
+    # Only a well-typed version string this consumer does not know reads as deployment skew;
+    # a wrong-typed field is the caller's malformed context like any other field.
+    with pytest.raises(MalformedEnvelopeError):
+        extract_context({RESERVED_CONTEXT_KEY: {**VALID, "schema_version": version}})
+
+
+def test_misaligned_batch_lists_are_rejected():
+    # invocation_ids is read positionally against record_ids, so a length mismatch would let
+    # every id past the gap name the wrong record.
+    with pytest.raises(MalformedEnvelopeError):
+        extract_context(
+            {
+                RESERVED_CONTEXT_KEY: {
+                    **VALID,
+                    "record_ids": ["r1", "r2"],
+                    "invocation_ids": ["inv-1"],
+                }
+            }
+        )
+
+
+def test_partial_context_is_accepted():
+    # A producer that populates only some identity facets degrades to less telemetry, not a
+    # failed invoke.
+    context = extract_context({RESERVED_CONTEXT_KEY: {"schema_version": "1", "job_id": "job-1"}})
+    assert context is not None
+    assert context.job_id == "job-1"
+    assert context.tenant_id is None
+
+
+def test_unknown_fields_survive_for_forward_compatibility():
+    context = extract_context({RESERVED_CONTEXT_KEY: {**VALID, "future_field": "keep me"}})
+    assert context is not None
+    assert context.model_extra["future_field"] == "keep me"
+
+
+def test_batch_fields_are_index_aligned():
+    # The controller emits one invocation id per record, using None where a record carried no
+    # context, so entry i always describes record i.
+    context = extract_context(
+        {
+            RESERVED_CONTEXT_KEY: {
+                **VALID,
+                "record_ids": ["rec-1", "rec-2", "rec-3"],
+                "invocation_ids": ["inv-1", None, "inv-3"],
+            }
+        }
+    )
+    assert context is not None
+    assert len(context.record_ids) == len(context.invocation_ids)
+    assert dict(zip(context.record_ids, context.invocation_ids))["rec-2"] is None
+
+
+class TestDimensions:
+    def test_returns_populated_identity_facets(self):
+        context = InvocationContext(**VALID)
+
+        assert dimensions(context) == {
+            "invocation_id": "inv-1",
+            "job_id": "job-1",
+            "tenant_id": "tenant-1",
+            "dag_node_id": "node-1",
+            "dag_node_type": "chunker",
+            "record_id": "rec-1",
+            "attempt": 2,
+        }
+
+    def test_excludes_batch_fields(self):
+        # These describe the work, not who it belongs to, and would blow up dimension cardinality.
+        context = InvocationContext.model_validate(
+            {**VALID, "record_ids": ["a"], "invocation_ids": ["b"]}
+        )
+
+        assert not {"record_ids", "invocation_ids"} & set(dimensions(context))
+
+    def test_unknown_producer_fields_are_not_promoted_to_dimensions(self):
+        context = InvocationContext.model_validate({**VALID, "future_field": "value"})
+
+        assert "future_field" not in dimensions(context)
+
+    def test_absent_context_yields_no_dimensions(self):
+        assert dimensions(None) == {}
+
+
+class TestHttpStatusFor:
+    """The transport's spelling of the library's normative `blame` -> status rule."""
+
+    def test_caller_blame_is_the_only_422(self):
+        assert http_status_for(MalformedEnvelopeError("x")) == 422
+        assert MalformedEnvelopeError.blame is Blame.CALLER
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            DecryptionError("x"),
+            KeyNotFoundError("x"),
+            IdentityNotMountedError("x"),
+            SealedDagNodeSettingsRequiredError("x"),
+            MalformedDagNodeSettingsError("x"),
+            UnsupportedContextVersionError("x"),
+        ],
+    )
+    def test_everything_else_is_5xx(self, error):
+        assert http_status_for(error) == 500
+
+    def test_an_unclassified_exception_is_not_blamed_on_the_caller(self):
+        assert http_status_for(RuntimeError("boom")) == 500