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,60 @@
+## 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.5.0`, which
+  owns the *settings contract* — including the v2 document 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.
+* **The wrapper consumes the ratified `invocation_context` contract locally.**
+  The field model, reserved key, supported versions, and dimension allow-list are generated from
+  `https://schemas.u10d.dev/invocation-context/v1.json`. The handwritten
+  `unstructured_platform_plugins.invocation_context` adapter retains transport error mapping and
+  the equal-length batch invariant that JSON Schema cannot express. The ratified
+  `https://schemas.u10d.dev/errors/audience/v1.json` vocabulary also replaces the redundant
+  top-level `blame` response field: a legacy `UserError` now carries a complete `plugin_error`
+  metadata object with `audience=user`.
+* **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_INVOCATION_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
+  can optionally install `InvokeBodyLimitMiddleware`, a streaming byte counter that answers 413
+  over a host-selected cap without buffering. The cap is disabled by default so a wrapper upgrade
+  cannot impose an unvalidated fleet-wide request limit. 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_v2=True` to `wrap_in_fastapi` / `generate_fast_api` (or
+  `--sealed-dag-node-settings-v2` on the CLI) only for a plugin that consumes per-invoke settings;
+  it advertises that the application accepts and acts on the versioned v2 document.
+  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.5.0,<1.0.0",
     "opentelemetry-instrumentation-fastapi",
     "opentelemetry-exporter-otlp-proto-grpc",
     "dataclasses-json"

scripts/generate_invocation_contracts.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Generate local Python bindings from the ratified invocation schema IDs."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+OUTPUT_DIR = ROOT / "unstructured_platform_plugins" / "generated"
+API_ROOT = "repos/Unstructured-IO/schemas-experimental/contents"
+
+CONTRACTS = {
+    "invocation_context_v1.py": {
+        "schema": f"{API_ROOT}/schemas/ratified-types/invocation-context/v1.json?ref=main",
+        "typeviz": f"{API_ROOT}/web/public/typeviz/invocation-context-v1.json?ref=main",
+        "schema_id": "https://schemas.u10d.dev/invocation-context/v1.json",
+        "root_type": "InvocationContext",
+    },
+    "error_audience_v1.py": {
+        "schema": f"{API_ROOT}/schemas/ratified-types/errors/audience/v1.json?ref=main",
+        "typeviz": f"{API_ROOT}/web/public/typeviz/errors-audience-v1.json?ref=main",
+        "schema_id": "https://schemas.u10d.dev/errors/audience/v1.json",
+        "root_type": "ErrorAudience",
+    },
+}
+
+
+def _load_json(api_path: str) -> dict[str, Any]:
+    completed = subprocess.run(
+        ["gh", "api", "-H", "Accept: application/vnd.github.raw+json", api_path],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return json.loads(completed.stdout)
+
+
+def _python_binding(sidecar: dict[str, Any]) -> str:
+    for binding in sidecar["bindings"]:
+        if binding["key"] == "python":
+            return binding["code"].rstrip() + "\n"
+    raise ValueError("typeviz sidecar has no Python binding")
+
+
+def _render(spec: dict[str, str]) -> str:
+    schema = _load_json(spec["schema"])
+    sidecar = _load_json(spec["typeviz"])
+    if schema["$id"] != spec["schema_id"]:
+        raise ValueError(f"unexpected schema id: {schema['$id']}")
+    if sidecar["root_type_name"] != spec["root_type"]:
+        raise ValueError(f"unexpected root type: {sidecar['root_type_name']}")
+
+    canonical = json.dumps(schema, sort_keys=True, separators=(",", ":")).encode()
+    schema_hash = hashlib.sha256(canonical).hexdigest()
+    generated = _python_binding(sidecar)
+    if spec["root_type"] == "InvocationContext":
+        generated = generated.replace(
+            "from typing import Any, Literal", "from typing import Literal"
+        )
+    else:
+        generated = generated.replace("\nfrom pydantic import BaseModel, Field\n", "")
+    provenance = (
+        "\n# Generation provenance used by drift tests and reviewers.\n"
+        f"SCHEMA_ID = {schema['$id']!r}\n"
+        f"SCHEMA_SHA256 = {schema_hash!r}\n"
+    )
+    if spec["root_type"] == "InvocationContext":
+        policy = schema["x-unstructured-version-policy"]
+        provenance += (
+            f"RESERVED_CONTEXT_KEY = {schema['x-unstructured-reserved-key']!r}\n"
+            f"SUPPORTED_CONTEXT_VERSIONS = frozenset({policy['supported']!r})\n"
+            f"DIMENSION_FIELDS = {tuple(schema['x-unstructured-dimension-fields'])!r}\n"
+        )
+    return (
+        "# Generated by scripts/generate_invocation_contracts.py; do not edit by hand.\n"
+        "# ruff: noqa: E501\n"
+        f"# Source: {spec['schema_id']}\n"
+        + generated
+        + provenance
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--check", action="store_true")
+    args = parser.parse_args()
+
+    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    stale: list[str] = []
+    for filename, spec in CONTRACTS.items():
+        output = OUTPUT_DIR / filename
+        rendered = _render(spec)
+        if args.check:
+            if not output.exists() or output.read_text() != rendered:
+                stale.append(str(output.relative_to(ROOT)))
+        else:
+            output.write_text(rendered)
+
+    if stale:
+        print("stale generated invocation contracts:", *stale, sep="\n  ", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

test/api/test_api.py

@@ -18,14 +18,24 @@
     UsageData,
     wrap_in_fastapi,
 )
+from unstructured_platform_plugins.generated.error_audience_v1 import ErrorAudience
 from unstructured_platform_plugins.schema.filedata_meta import FileDataMeta
 
 
+class PluginErrorMetadata(BaseModel):
+    error_type: str
+    error_reason: str
+    dependency: Optional[str] = None
+    audience: Optional[ErrorAudience] = None
+    retryable: bool = False
+
+
 class InvokeResponse(BaseModel):
     usage: list[UsageData]
     status_code: int
     filedata_meta: FileDataMeta
     status_code_text: Optional[str] = None
+    plugin_error: Optional[PluginErrorMetadata] = None
     output: Optional[Any] = None
     file_data: Optional[Union[FileData, BatchFileData]] = None
 
@@ -222,6 +232,67 @@ 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_canonical_user_audience(file_data):
+    """Only the UserError family declares a user-actionable plugin error."""
+    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.plugin_error is not None
+    assert invoke_response.plugin_error.audience is ErrorAudience.USER
+    assert invoke_response.plugin_error.error_type == "configuration"
+    assert invoke_response.plugin_error.error_reason == "invalid_input"
+    assert invoke_response.plugin_error.retryable is False
+
+
+@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_plugin_error(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.plugin_error is None
+
+
+def test_streaming_user_error_declares_user_audience():
+    """The streaming error envelope carries the same audience 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])).plugin_error is None
+    error_response = InvokeResponse.model_validate(json.loads(lines[1]))
+    assert error_response.status_code >= 400
+    assert error_response.plugin_error is not None
+    assert error_response.plugin_error.audience is ErrorAudience.USER
+
+
 @pytest.mark.parametrize(
     "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
 )
@@ -608,9 +679,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["plugin_error"] is None
     assert "credential rejected" in body["status_code_text"]
 
 
+def test_precheck_declares_user_audience_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["plugin_error"]["audience"] == "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)