feat(etl-uvicorn): own the /invoke transport and invocation-context model (0.1.0)

.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,61 @@
+## 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* — which key carries settings, how a sealed envelope is told
+  from plaintext, 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, a sealed
+  `dag_node_settings` member is opened with this pod's mounted workload key, and the resolved
+  values are 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 sealed `dag_node_settings`. 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.** A cold resolve is an RSA unwrap of a couple of
+  milliseconds 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.
+* **Sync plugin functions now observe request-scoped context.** `invoke_func` copies the current
+  context into the executor thread; previously `run_in_executor` dropped contextvars, so a sync
+  function reading a request-scoped binding (such as `current_invocation_settings()`) would see
+  it as absent and could take an unintended fallback path.
+* **Python floor is now 3.11** (required by `utic-invocation-settings`).
+
 ## 0.0.45
 
 * **`/invoke` no longer demands a body from a plugin whose parameters are all optional.** A pydantic

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"

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,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]
 )

test/api/test_invocation_context.py

@@ -0,0 +1,153 @@
+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)
+
+
+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