From 43a7f671ade89eb816c82fb8662a75f2081e8956 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Fri, 31 Jul 2026 16:28:00 -0400 Subject: [PATCH 01/12] feat(etl-uvicorn): install invocation-settings handling (0.1.0) --- CHANGELOG.md | 24 ++++ pyproject.toml | 4 +- test/api/test_invocation_settings.py | 103 ++++++++++++++++++ unstructured_platform_plugins/__version__.py | 2 +- .../etl_uvicorn/api_generator.py | 40 ++++++- .../etl_uvicorn/main.py | 10 ++ 6 files changed, 177 insertions(+), 6 deletions(-) create mode 100644 test/api/test_invocation_settings.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e3a14bd..adcff2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,27 @@ +## 0.1.0 + +* **The wrapper now installs the invocation-settings envelope handling itself.** Every wrapped app + gets the `utic-invocation-settings` ASGI middleware and a `/metadata` route at construction: the + reserved `invocation_settings` / `invocation_context` fields are handled outside the generated + handler schema, a sealed `dag_node_settings` member is decrypted with the configured private + key, and the resolved values are exposed request-scoped through + `current_invocation_settings()` / `current_invocation_context()`. Missing fields preserve the + existing fallback behavior; when `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` is enabled, + missing or plaintext settings fail closed. Repeated installation is safe: the middleware + installs once and the last `/metadata` registration wins. +* **New opt-in `invoke_with_sealed_dag_node_settings` advertisement.** 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 consumes sealed per-invocation settings. + 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. +* **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 diff --git a/pyproject.toml b/pyproject.toml index 4bb84aa..471440e 100644 --- a/pyproject.toml +++ b/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.3.0,<1.0.0", "opentelemetry-instrumentation-fastapi", "opentelemetry-exporter-otlp-proto-grpc", "dataclasses-json" diff --git a/test/api/test_invocation_settings.py b/test/api/test_invocation_settings.py new file mode 100644 index 0000000..856e17e --- /dev/null +++ b/test/api/test_invocation_settings.py @@ -0,0 +1,103 @@ +"""The wrapper-installed invocation-settings surface: /metadata and reserved-field binding.""" + +from typing import Optional + +from fastapi.testclient import TestClient +from pydantic import BaseModel +from utic_invocation_settings import current_invocation_settings + +from unstructured_platform_plugins.etl_uvicorn.api_generator import wrap_in_fastapi + + +class _Echo(BaseModel): + content: str + settings: Optional[dict] + + +def _echo_settings(content: str) -> _Echo: + return _Echo(content=content, settings=current_invocation_settings()) + + +def test_metadata_route_is_registered_with_default_capabilities(): + client = TestClient(wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin")) + + resp = client.get("/metadata") + + assert resp.status_code == 200 + payload = resp.json() + assert payload["identifier"] == "mock_plugin" + assert payload["capabilities"] == ["invocation_settings", "invocation_context"] + + +def test_sealed_capability_is_opt_in(): + client = TestClient( + wrap_in_fastapi( + func=_echo_settings, + plugin_id="mock_plugin", + invoke_with_sealed_dag_node_settings=True, + ) + ) + + payload = client.get("/metadata").json() + + assert "invoke_with_sealed_dag_node_settings" in payload["capabilities"] + + +def test_reserved_settings_field_binds_without_appearing_in_schema(): + app = wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin") + client = TestClient(app) + + resp = client.post( + "/invoke", + json={"content": "hello", "invocation_settings": {"model": "m"}}, + ) + + assert resp.status_code == 200 + output = resp.json()["output"] + assert output == {"content": "hello", "settings": {"model": "m"}} + # The wrapper does not add the reserved field to the generated handler input model. + openapi = app.openapi() + request_schema = openapi["paths"]["/invoke"]["post"]["requestBody"]["content"][ + "application/json" + ]["schema"] + schema_name = request_schema["$ref"].rsplit("/", 1)[-1] + properties = openapi["components"]["schemas"][schema_name]["properties"] + assert "invocation_settings" not in properties + + +def test_sync_function_sees_bound_settings_across_the_executor(): + # Sync functions run in an executor thread; the context must be copied there or the + # request-scoped binding would read as absent. + def sync_echo(content: str) -> _Echo: + return _Echo(content=content, settings=current_invocation_settings()) + + client = TestClient(wrap_in_fastapi(func=sync_echo, plugin_id="mock_plugin")) + + resp = client.post( + "/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}} + ) + + assert resp.json()["output"]["settings"] == {"model": "m"} + + +async def _async_echo(content: str) -> _Echo: + return _Echo(content=content, settings=current_invocation_settings()) + + +def test_async_function_sees_bound_settings(): + client = TestClient(wrap_in_fastapi(func=_async_echo, plugin_id="mock_plugin")) + + resp = client.post( + "/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}} + ) + + assert resp.json()["output"]["settings"] == {"model": "m"} + + +def test_absent_reserved_fields_bind_none(): + client = TestClient(wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin")) + + resp = client.post("/invoke", json={"content": "hello"}) + + assert resp.status_code == 200 + assert resp.json()["output"] == {"content": "hello", "settings": None} diff --git a/unstructured_platform_plugins/__version__.py b/unstructured_platform_plugins/__version__.py index d8f2458..2a08aec 100644 --- a/unstructured_platform_plugins/__version__.py +++ b/unstructured_platform_plugins/__version__.py @@ -1 +1 @@ -__version__ = "0.0.45" # pragma: no cover +__version__ = "0.1.0" # pragma: no cover diff --git a/unstructured_platform_plugins/etl_uvicorn/api_generator.py b/unstructured_platform_plugins/etl_uvicorn/api_generator.py index 4b7d8ce..c1ec1c1 100644 --- a/unstructured_platform_plugins/etl_uvicorn/api_generator.py +++ b/unstructured_platform_plugins/etl_uvicorn/api_generator.py @@ -1,4 +1,5 @@ import asyncio +import contextvars import hashlib import inspect import json @@ -14,6 +15,7 @@ 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 utic_invocation_settings import add_metadata_route, install_invocation_envelope from uvicorn.config import LOG_LEVELS from uvicorn.importer import import_from_string @@ -67,7 +69,13 @@ async def invoke_func(func: Callable, kwargs: Optional[dict[str, Any]] = None) - if inspect.iscoroutinefunction(func): return await func(**kwargs) else: - return await asyncio.get_event_loop().run_in_executor(None, partial(func, **kwargs)) + # run_in_executor does not propagate contextvars, so without copying the context a sync + # plugin would observe request-scoped bindings (current_invocation_settings and friends) + # as absent and could take an unintended fallback path. + ctx = contextvars.copy_context() + return await asyncio.get_event_loop().run_in_executor( + None, ctx.run, partial(func, **kwargs) + ) def check_precheck_func(precheck_func: Callable): @@ -117,9 +125,15 @@ def wrap_in_fastapi( func: Callable, plugin_id: str, precheck_func: Optional[Callable] = None, + invoke_with_sealed_dag_node_settings: bool = False, ) -> FastAPI: try: - return _wrap_in_fastapi(func=func, plugin_id=plugin_id, precheck_func=precheck_func) + return _wrap_in_fastapi( + func=func, + plugin_id=plugin_id, + precheck_func=precheck_func, + invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings, + ) except Exception as e: logger.error(f"failed to wrap function in FastAPI: {e}", exc_info=True) raise EtlApiException(e) from e @@ -129,6 +143,7 @@ def _wrap_in_fastapi( func: Callable, plugin_id: str, precheck_func: Optional[Callable] = None, + invoke_with_sealed_dag_node_settings: bool = False, ) -> FastAPI: if precheck_func is not None: check_precheck_func(precheck_func=precheck_func) @@ -347,6 +362,19 @@ async def get_id() -> str: except TypeError as e: raise TypeError(f"failed to validate function schema: {e}") from e + # The middleware handles the reserved /invoke fields (invocation_settings and + # invocation_context) outside the generated handler schema. It resolves sealed settings with + # the configured private key and exposes both values through request-scoped accessors. The + # sealed-settings capability remains opt-in because it asserts that the wrapped function + # consumes current_invocation_settings(), not merely that the host can resolve it. Repeated + # installation is safe: the middleware installs once and the last /metadata registration wins. + add_metadata_route( + fastapi_app, + identifier=plugin_id, + invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings, + ) + install_invocation_envelope(fastapi_app) + FastAPIInstrumentor.instrument_app( fastapi_app, tracer_provider=get_trace_provider(), meter_provider=get_metric_provider() ) @@ -361,6 +389,7 @@ def generate_fast_api( id_method: Optional[str] = None, precheck_str: Optional[str] = None, precheck_method: Optional[str] = None, + invoke_with_sealed_dag_node_settings: bool = False, ) -> FastAPI: instance = import_from_string(app) func = get_func(instance, method_name) @@ -379,4 +408,9 @@ def generate_fast_api( elif precheck_method: precheck_func = get_func(instance, precheck_method) - return wrap_in_fastapi(func=func, plugin_id=plugin_id, precheck_func=precheck_func) + return wrap_in_fastapi( + func=func, + plugin_id=plugin_id, + precheck_func=precheck_func, + invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings, + ) diff --git a/unstructured_platform_plugins/etl_uvicorn/main.py b/unstructured_platform_plugins/etl_uvicorn/main.py index 600e0c1..62b067f 100644 --- a/unstructured_platform_plugins/etl_uvicorn/main.py +++ b/unstructured_platform_plugins/etl_uvicorn/main.py @@ -56,6 +56,7 @@ def api_wrapper( plugin_id_method: Optional[str] = None, precheck_app: Optional[str] = None, precheck_app_method: Optional[str] = None, + sealed_dag_node_settings: bool = False, **kwargs, ): # Make sure logging is configured before the call to run() so any setup has the same format @@ -73,6 +74,7 @@ def api_wrapper( id_method=plugin_id_method, precheck_str=precheck_app, precheck_method=precheck_app_method, + invoke_with_sealed_dag_node_settings=sealed_dag_node_settings, ) # Explicitly map values that are manipulated in the original # call to run(), preventing **kwargs reference @@ -130,6 +132,14 @@ def api_wrapper( "If precheck-app not provided, assumes method " "lives on main class passes in.", ), + click.Option( + ["--sealed-dag-node-settings"], + is_flag=True, + default=False, + help="Advertise the invoke_with_sealed_dag_node_settings capability on " + "/metadata. Set only for a plugin that consumes per-invoke settings " + "through current_invocation_settings().", + ), ] ) return cmd From e2ddaaa0f3983b4ed9b5cea42752be67f0ffefcc Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Fri, 31 Jul 2026 22:49:51 -0400 Subject: [PATCH 02/12] refactor(etl-uvicorn): own the /invoke transport, not the contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves the ASGI middleware, the /metadata capability route and the request-scoped binding out of utic-invocation-settings and into this package, as unstructured_platform_plugins.invocation_settings. The split follows what the two halves actually are. The contract — which keys carry settings, how a sealed envelope is told from plaintext, what an absent field is allowed to mean — stays in the library: the absence rule is the tenant-confusion vector, and it belongs next to the crypto it governs and the threat model that describes it. Buffering a request body and registering a route do not. Two things this buys immediately: - No more duck-typing. Living in the library forced the middleware to reach into `app.router.routes` through getattr chains to avoid importing Starlette. Here fastapi is already a dependency, so route eviction and the ASGI signature are typed against the real thing. - The library goes back to cryptography + pydantic with no framework test dependencies at all; its packaging suite asserts that against the built wheel. Requires utic-invocation-settings >=0.4.0 for resolve_invocation_settings, http_status_for and the contract constants. Tests: 109 passed (84 + 25 ported transport tests), ruff clean. --- CHANGELOG.md | 32 +- pyproject.toml | 2 +- test/api/test_invocation_middleware.py | 452 ++++++++++++++++++ test/api/test_invocation_settings.py | 2 +- .../etl_uvicorn/api_generator.py | 5 +- .../invocation_settings.py | 243 ++++++++++ 6 files changed, 724 insertions(+), 12 deletions(-) create mode 100644 test/api/test_invocation_middleware.py create mode 100644 unstructured_platform_plugins/invocation_settings.py diff --git a/CHANGELOG.md b/CHANGELOG.md index adcff2a..8591439 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,14 +1,21 @@ ## 0.1.0 -* **The wrapper now installs the invocation-settings envelope handling itself.** Every wrapped app - gets the `utic-invocation-settings` ASGI middleware and a `/metadata` route at construction: the - reserved `invocation_settings` / `invocation_context` fields are handled outside the generated - handler schema, a sealed `dag_node_settings` member is decrypted with the configured private - key, and the resolved values are exposed request-scoped through - `current_invocation_settings()` / `current_invocation_context()`. Missing fields preserve the - existing fallback behavior; when `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` is enabled, - missing or plaintext settings fail closed. Repeated installation is safe: the middleware - installs once and the last `/metadata` registration wins. +* **This package now owns the `/invoke` transport for the reserved fields.** + `unstructured_platform_plugins.invocation_settings` holds the ASGI middleware, the `/metadata` + capability route, and the request-scoped binding. It sits on `utic-invocation-settings >=0.4.0`, + which owns the *contract* — which keys carry 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 body buffering and + route registration belong here, where a web framework is already a dependency. Nothing about the + wire format is decided in this repository. +* **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 middleware installs once and the last `/metadata` + registration wins. * **New opt-in `invoke_with_sealed_dag_node_settings` advertisement.** 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; @@ -16,6 +23,13 @@ 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 middleware 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 diff --git a/pyproject.toml b/pyproject.toml index 471440e..4950598 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ dependencies = [ "fastapi", "click", "unstructured-ingest", - "utic-invocation-settings>=0.3.0,<1.0.0", + "utic-invocation-settings>=0.4.0,<1.0.0", "opentelemetry-instrumentation-fastapi", "opentelemetry-exporter-otlp-proto-grpc", "dataclasses-json" diff --git a/test/api/test_invocation_middleware.py b/test/api/test_invocation_middleware.py new file mode 100644 index 0000000..2b8b125 --- /dev/null +++ b/test/api/test_invocation_middleware.py @@ -0,0 +1,452 @@ +"""The transport for the reserved /invoke fields: middleware, /metadata, request-scoped binding. + +The *contract* these exercise — which shapes carry settings, what absence means — is owned and +tested in `utic_invocation_settings`. What is tested here is delivery: that a raw body is read, +resolved, bound, and replayed intact, and that a payload which cannot be used fails the request +instead of reaching a handler as absence. +""" + +from __future__ import annotations + +import asyncio +import json +from base64 import b64decode, b64encode + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from fastapi import FastAPI +from fastapi.testclient import TestClient +from utic_invocation_settings import ( + DAG_NODE_SETTINGS_KEY, + REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, + default_resolver, + reset_workload_identity_cache, +) +from utic_invocation_settings.crypto import seal_settings + +from unstructured_platform_plugins.invocation_settings import ( + InvocationEnvelopeMiddleware, + add_metadata_route, + current_invocation_context, + current_invocation_settings, +) + +SENTINEL_SECRET = "sealed-settings-sentinel-secret" + + +@pytest.fixture(scope="session") +def private_key() -> rsa.RSAPrivateKey: + return rsa.generate_private_key(public_exponent=65537, key_size=3072) + + +@pytest.fixture(autouse=True) +def isolated_identity(monkeypatch): + """The identity memo and the resolver caches both outlive a test; an inherited env var or a + stale entry would make these order-dependent.""" + for var in ("WORKLOAD_IDENTITY_DIR", "INVOCATION_SETTINGS_KEY_DIR", + REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR): + monkeypatch.delenv(var, raising=False) + reset_workload_identity_cache() + default_resolver().clear_caches() + yield + reset_workload_identity_cache() + default_resolver().clear_caches() + + +@pytest.fixture +def key_dir(tmp_path, private_key, monkeypatch, isolated_identity): + (tmp_path / "tls.key").write_bytes( + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + monkeypatch.setenv("WORKLOAD_IDENTITY_DIR", str(tmp_path)) + reset_workload_identity_cache() + return tmp_path + + +def sealed_payload(private_key: rsa.RSAPrivateKey, settings: dict) -> dict: + return seal_settings(settings, private_key.public_key()).model_dump( + mode="json", exclude_none=True + ) + + +def tampered(sealed: dict) -> dict: + ciphertext = bytearray(b64decode(sealed["content_encryption"]["ciphertext"])) + ciphertext[0] ^= 0x01 + sealed["content_encryption"]["ciphertext"] = b64encode(bytes(ciphertext)).decode() + return sealed + + +class TestMetadataRoute: + def test_advertises_settings_and_context_capabilities(self): + app = FastAPI() + add_metadata_route(app, identifier="plugin.test") + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload == { + "api_version": "3", + "identifier": "plugin.test", + "capabilities": ["invocation_settings", "invocation_context"], + } + + def test_sealed_dag_node_settings_flag_advertises_capability(self): + app = FastAPI() + add_metadata_route(app, invoke_with_sealed_dag_node_settings=True) + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload["capabilities"] == [ + "invocation_settings", + "invocation_context", + "invoke_with_sealed_dag_node_settings", + ] + + def test_last_call_wins(self): + # A host wrapper registers /metadata with defaults at construction; the plugin's later call + # with the sealed capability must replace it, not be shadowed by route order. + app = FastAPI() + add_metadata_route(app, identifier="plugin.test") + add_metadata_route(app, identifier="plugin.test", invoke_with_sealed_dag_node_settings=True) + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert "invoke_with_sealed_dag_node_settings" in payload["capabilities"] + + def test_replaces_a_directly_registered_metadata_route(self): + # A route the app registered itself would otherwise win by route order and pin its stale + # payload. + app = FastAPI() + + @app.get("/metadata") + async def stale_metadata() -> dict: + return {"api_version": "3", "identifier": "stale", "capabilities": []} + + add_metadata_route(app, identifier="plugin.test", invoke_with_sealed_dag_node_settings=True) + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload["identifier"] == "plugin.test" + assert "invoke_with_sealed_dag_node_settings" in payload["capabilities"] + + +def _invoke_scope(path: str = "/invoke", method: str = "POST") -> dict: + return {"type": "http", "method": method, "path": path} + + +def _receive_for(body: bytes): + chunks = [ + {"type": "http.request", "body": body[: len(body) // 2], "more_body": True}, + {"type": "http.request", "body": body[len(body) // 2 :], "more_body": False}, + ] + + async def receive(): + return chunks.pop(0) + + return receive + + +async def _ok(send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"{}"}) + + +class _DownstreamApp: + """Records the replayed body and the envelope bound while handling.""" + + def __init__(self): + self.body = None + self.called = False + self.seen_settings = "unset" + self.seen_context = "unset" + + async def __call__(self, scope, receive, send): + body = b"" + while True: + message = await receive() + body += message.get("body", b"") + if not message.get("more_body"): + break + self.body = body + self.called = True + self.seen_settings = current_invocation_settings() + self.seen_context = current_invocation_context() + await _ok(send) + + +def _run_middleware(body: bytes, scope: dict | None = None) -> tuple[_DownstreamApp, list]: + downstream = _DownstreamApp() + middleware = InvocationEnvelopeMiddleware(downstream) + sent = [] + + async def send(message): + sent.append(message) + + asyncio.run(middleware(scope or _invoke_scope(), _receive_for(body), send)) + return downstream, sent + + +class TestInvocationEnvelopeMiddleware: + def test_binds_reserved_fields_and_replays_body(self): + body = json.dumps( + { + "element_dicts": "/in.json", + "invocation_settings": {"model": "m"}, + "invocation_context": {"schema_version": "1", "job_id": "job-1"}, + } + ).encode() + + downstream, sent = _run_middleware(body) + + assert downstream.body == body + assert downstream.seen_settings == {"model": "m"} + assert downstream.seen_context.job_id == "job-1" + assert sent[0]["status"] == 200 + + def test_absent_fields_bind_none(self): + downstream, _ = _run_middleware(json.dumps({"element_dicts": "/in.json"}).encode()) + + assert downstream.seen_settings is None + assert downstream.seen_context is None + + def test_non_dict_reserved_field_is_rejected(self): + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": "not-a-dict"}).encode() + ) + + assert downstream.body is None + assert sent[0]["status"] == 422 + assert b"invocation_settings" in sent[1]["body"] + + def test_context_with_an_unreadable_schema_version_is_rejected(self): + # Absence means "older caller, use the boot settings"; a context this plugin cannot read + # must not be downgraded to that. + downstream, sent = _run_middleware( + json.dumps({"invocation_context": {"schema_version": "99"}}).encode() + ) + + assert downstream.body is None + assert sent[0]["status"] == 422 + assert b"invocation_context" in sent[1]["body"] + + def test_malformed_context_is_rejected(self): + downstream, sent = _run_middleware( + json.dumps({"invocation_context": "not-an-object"}).encode() + ) + + assert downstream.body is None + assert sent[0]["status"] == 422 + + def test_non_invoke_requests_pass_through_untouched(self): + body = json.dumps({"invocation_settings": "not-a-dict"}).encode() + + downstream, sent = _run_middleware(body, scope=_invoke_scope(path="/schema", method="GET")) + + assert downstream.body == body + assert sent[0]["status"] == 200 + + def test_envelope_is_reset_after_request(self): + body = json.dumps({"invocation_settings": {"model": "m"}}).encode() + + async def scenario(): + middleware = InvocationEnvelopeMiddleware(_DownstreamApp()) + + async def send(_message): + pass + + await middleware(_invoke_scope(), _receive_for(body), send) + return current_invocation_settings(), current_invocation_context() + + assert asyncio.run(scenario()) == (None, None) + + def test_oversized_body_is_rejected(self): + body = json.dumps({"invocation_settings": {"pad": "x" * 64}}).encode() + + downstream = _DownstreamApp() + middleware = InvocationEnvelopeMiddleware(downstream, max_body_bytes=16) + sent = [] + + async def send(message): + sent.append(message) + + asyncio.run(middleware(_invoke_scope(), _receive_for(body), send)) + + assert downstream.body is None + assert sent[0]["status"] == 413 + + def test_drained_replay_proxies_disconnect(self): + body = json.dumps({"invocation_settings": {"model": "m"}}).encode() + + class _DisconnectWatcher: + def __init__(self): + self.saw_disconnect = False + + async def __call__(self, scope, receive, send): + while True: + message = await receive() + if message["type"] == "http.disconnect": + self.saw_disconnect = True + return + if not message.get("more_body"): + break + message = await receive() + self.saw_disconnect = message["type"] == "http.disconnect" + await _ok(send) + + chunks = [ + {"type": "http.request", "body": body, "more_body": False}, + {"type": "http.disconnect"}, + ] + + async def receive(): + return chunks.pop(0) + + watcher = _DisconnectWatcher() + middleware = InvocationEnvelopeMiddleware(watcher) + sent = [] + + async def send(message): + sent.append(message) + + asyncio.run(middleware(_invoke_scope(), receive, send)) + + assert watcher.saw_disconnect + + +class TestMiddlewareResolution: + """Sealed payloads through the middleware. The HTTP class comes from the library's blame + taxonomy, so only a caller-fixable fault is a 422.""" + + def test_sealed_envelope_binds_plaintext_settings(self, key_dir, private_key): + settings = {"api_key": SENTINEL_SECRET, "max_characters": 700} + sealed = sealed_payload(private_key, settings) + + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": sealed}).encode() + ) + + assert sent[0]["status"] == 200 + assert downstream.seen_settings == settings + + def test_composite_dag_node_settings_member_is_opened(self, key_dir, private_key): + settings = {"api_key": SENTINEL_SECRET, "max_characters": 700} + composite = {DAG_NODE_SETTINGS_KEY: sealed_payload(private_key, settings)} + + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": composite}).encode() + ) + + assert sent[0]["status"] == 200 + assert downstream.seen_settings == settings + + def test_plain_dict_settings_pass_through(self): + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": {"model": "m"}}).encode() + ) + + assert sent[0]["status"] == 200 + assert downstream.seen_settings == {"model": "m"} + + def test_non_envelope_member_fails_as_platform_error(self, key_dir): + composite = {DAG_NODE_SETTINGS_KEY: {"model": "m"}} + + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": composite}).encode() + ) + + assert sent[0]["status"] == 500 + assert "MalformedDagNodeSettingsError" in json.loads(sent[1]["body"])["detail"] + assert not downstream.called + + def test_undecryptable_envelope_fails_without_leaking_the_secret( + self, key_dir, private_key, caplog + ): + sealed = tampered(sealed_payload(private_key, {"api_key": SENTINEL_SECRET})) + + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": sealed}).encode() + ) + + assert sent[0]["status"] == 500 + detail = json.loads(sent[1]["body"])["detail"] + assert "DecryptionError" in detail + assert SENTINEL_SECRET not in caplog.text + assert SENTINEL_SECRET not in detail + assert not downstream.called + + def test_unmounted_identity_fails_as_platform_error(self, tmp_path, monkeypatch, private_key): + monkeypatch.setenv("WORKLOAD_IDENTITY_DIR", str(tmp_path / "missing")) + sealed = sealed_payload(private_key, {"api_key": SENTINEL_SECRET}) + + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": sealed}).encode() + ) + + assert sent[0]["status"] == 500 + assert "IdentityNotMountedError" in json.loads(sent[1]["body"])["detail"] + assert not downstream.called + + +class TestRequireSealedDagNodeSettings: + """A native pod's operator sets FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS in the same + decision that drops the init-secrets sidecar: without it, an invoke that arrived with no + envelope would fall back to a settings file that was never written.""" + + @pytest.fixture(autouse=True) + def _require_sealed(self, monkeypatch): + monkeypatch.setenv(REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, "true") + + def test_missing_settings_fail_as_platform_error(self): + downstream, sent = _run_middleware(json.dumps({"element_dicts": "/tmp/x.json"}).encode()) + + assert sent[0]["status"] == 500 + assert "SealedDagNodeSettingsRequiredError" in json.loads(sent[1]["body"])["detail"] + assert not downstream.called + + def test_plaintext_settings_fail_as_platform_error(self): + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": {"model": "m"}}).encode() + ) + + assert sent[0]["status"] == 500 + assert not downstream.called + + def test_sealed_settings_still_bind(self, key_dir, private_key): + settings = {"api_key": SENTINEL_SECRET} + composite = {DAG_NODE_SETTINGS_KEY: sealed_payload(private_key, settings)} + + downstream, sent = _run_middleware( + json.dumps({"invocation_settings": composite}).encode() + ) + + assert sent[0]["status"] == 200 + assert downstream.seen_settings == settings + + def test_bodyless_invoke_fails_as_platform_error(self): + # A bodyless invoke cannot carry the envelope; on a pod with no boot settings file it must + # not dispatch a handler with no settings source at all. + downstream, sent = _run_middleware(b"") + + assert sent[0]["status"] == 500 + assert not downstream.called + + def test_non_object_json_body_fails_as_platform_error(self): + downstream, sent = _run_middleware(json.dumps([{"element": 1}]).encode()) + + assert sent[0]["status"] == 500 + assert not downstream.called + + +def test_non_object_bodies_pass_through_when_sealed_settings_not_required(): + downstream, sent = _run_middleware(b"") + + assert sent[0]["status"] == 200 + assert downstream.seen_settings is None diff --git a/test/api/test_invocation_settings.py b/test/api/test_invocation_settings.py index 856e17e..5523b23 100644 --- a/test/api/test_invocation_settings.py +++ b/test/api/test_invocation_settings.py @@ -4,9 +4,9 @@ from fastapi.testclient import TestClient from pydantic import BaseModel -from utic_invocation_settings import current_invocation_settings from unstructured_platform_plugins.etl_uvicorn.api_generator import wrap_in_fastapi +from unstructured_platform_plugins.invocation_settings import current_invocation_settings class _Echo(BaseModel): diff --git a/unstructured_platform_plugins/etl_uvicorn/api_generator.py b/unstructured_platform_plugins/etl_uvicorn/api_generator.py index c1ec1c1..95da043 100644 --- a/unstructured_platform_plugins/etl_uvicorn/api_generator.py +++ b/unstructured_platform_plugins/etl_uvicorn/api_generator.py @@ -15,7 +15,6 @@ 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 utic_invocation_settings import add_metadata_route, install_invocation_envelope from uvicorn.config import LOG_LEVELS from uvicorn.importer import import_from_string @@ -28,6 +27,10 @@ get_schema_dict, map_inputs, ) +from unstructured_platform_plugins.invocation_settings import ( + add_metadata_route, + install_invocation_envelope, +) from unstructured_platform_plugins.schema import FileDataMeta, NewRecord, UsageData from unstructured_platform_plugins.schema.json_schema import ( schema_to_base_model, diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py new file mode 100644 index 0000000..73d0338 --- /dev/null +++ b/unstructured_platform_plugins/invocation_settings.py @@ -0,0 +1,243 @@ +"""Transport for the reserved `/invoke` fields: ASGI middleware, `/metadata`, request binding. + +The *contract* — which keys carry settings, how a sealed envelope is told from plaintext, and what +an absent field is allowed to mean — lives in `utic_invocation_settings.invoke`, next to the crypto +it governs. This module is the other half: getting the payload off the wire and the result to the +handler. It owns no policy; every decision about a payload it delegates. + +The reserved fields are a first-class HTTP contract independent of the generated input schema. They +never appear in a plugin's declared signature, so `wrap_in_fastapi` keeps producing a handler model +built purely from the wrapped function, and a plugin reads the fields through +`current_invocation_settings()` / `current_invocation_context()` instead. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Any, Optional + +from fastapi import FastAPI +from starlette.types import ASGIApp, Receive, Scope, Send +from utic_invocation_settings import ( + INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_CAPABILITY, + RESERVED_CONTEXT_KEY, + RESERVED_ENVELOPE_KEY, + InvocationContext, + InvocationSettingsError, + extract_context, + http_status_for, + resolve_invocation_settings, +) + +logger = logging.getLogger(__name__) + +_METADATA_PATH = "/metadata" +_INVOKE_PATH = "/invoke" + +# Bounds middleware body buffering; generous because batch invokes carry an array of file_data +# payloads. The framework buffers the same body afterward, so this cap is the only guard against +# unbounded-memory requests. +MAX_INVOKE_BODY_BYTES = 64 * 1024 * 1024 + +_INVOCATION: ContextVar[tuple[Optional[dict], Optional[InvocationContext]]] = ContextVar( + "invocation", default=(None, None) +) + + +def current_invocation_settings() -> Optional[dict]: + """Reserved `invocation_settings` field bound for the current request, if any. + + `None` means the field was genuinely absent — the only case in which a plugin may fall back to + its boot-time settings. A field that arrived and could not be opened never reaches a handler: + the middleware fails the request first. + """ + return _INVOCATION.get()[0] + + +def current_invocation_context() -> Optional[InvocationContext]: + """Reserved `invocation_context` field bound for the current request, if any.""" + return _INVOCATION.get()[1] + + +@contextmanager +def invocation_envelope( + invocation_settings: Optional[dict], invocation_context: Optional[InvocationContext] +) -> Iterator[None]: + """Bind the reserved /invoke fields for the current context.""" + token = _INVOCATION.set((invocation_settings, invocation_context)) + try: + yield + finally: + _INVOCATION.reset(token) + + +def add_metadata_route( + app: FastAPI, + identifier: Optional[str] = None, + invoke_with_sealed_dag_node_settings: bool = False, +) -> None: + """Register GET /metadata advertising the reserved /invoke fields this plugin accepts. + + `/metadata` is the plugin API spec's own discovery surface (`PluginMetadataOutput`): capability + flags are strings in its `capabilities` list, which is where the controller looks before + forwarding the reserved fields — no controller-private probe route. + `invoke_with_sealed_dag_node_settings` additionally advertises that the plugin can be invoked + with a sealed `dag_node_settings` member and open it itself. + + Last call wins: the payload lives on `app.state` and every call overwrites it, while the route + is registered once. A host wrapper may register with default capabilities at app construction + and a plugin can still declare the sealed capability afterwards, with no route-order dependence. + """ + capabilities = [RESERVED_ENVELOPE_KEY, RESERVED_CONTEXT_KEY] + if invoke_with_sealed_dag_node_settings: + capabilities.append(INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_CAPABILITY) + app.state.plugin_metadata_payload = { + "api_version": "3", + "identifier": identifier, + "capabilities": capabilities, + } + if getattr(app.state, "plugin_metadata_route_installed", False): + return + app.state.plugin_metadata_route_installed = True + + # A /metadata route registered by the application itself would win by route order and pin its + # own stale payload; drop it so the last add_metadata_route call is the one that answers. + app.router.routes = [r for r in app.router.routes if getattr(r, "path", None) != _METADATA_PATH] + + @app.get(_METADATA_PATH) + async def plugin_metadata() -> dict: + return app.state.plugin_metadata_payload + + +async def _send_json(send: Send, status_code: int, payload: dict) -> None: + body = json.dumps(payload).encode() + await send( + { + "type": "http.response.start", + "status": status_code, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode()), + ], + } + ) + await send({"type": "http.response.body", "body": body}) + + +class InvocationEnvelopeMiddleware: + """Extract the reserved envelope fields from the raw POST /invoke body. + + Pure ASGI rather than `BaseHTTPMiddleware`: the body has to be read before the framework parses + it and then replayed intact, which is exactly what the raw protocol allows and what a + request/response middleware would fight. + + A reserved field that is present but unusable fails the request rather than being treated as + absent, because absence is the signal to fall back to the boot-time settings file: degrading a + malformed field to absence would quietly answer a request configured for one tenant with + whatever the pod happened to boot with. Under `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` + there is no settings file to fall back to, so absent or plaintext settings fail too — as does + any /invoke whose body is not a JSON object, since such a body cannot carry the envelope a + native pod requires. A body over `max_body_bytes` is rejected with 413 before it can exhaust + memory. + """ + + def __init__(self, app: ASGIApp, max_body_bytes: int = MAX_INVOKE_BODY_BYTES): + self.app = app + self.max_body_bytes = max_body_bytes + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if ( + scope["type"] != "http" + or scope.get("method") != "POST" + or scope.get("path") != _INVOKE_PATH + ): + await self.app(scope, receive, send) + return + + messages = [] + buffered_bytes = 0 + while True: + message = await receive() + messages.append(message) + buffered_bytes += len(message.get("body", b"")) + if buffered_bytes > self.max_body_bytes: + await _send_json(send, 413, {"detail": "Request body too large"}) + return + if message["type"] != "http.request" or not message.get("more_body"): + break + body = b"".join(m.get("body", b"") for m in messages if m["type"] == "http.request") + + try: + parsed = json.loads(body) if body else None + except ValueError: + # Malformed JSON: forward unchanged so the framework returns its own error. + parsed = None + + invocation_context: Optional[InvocationContext] = None + raw_settings: Optional[dict[str, Any]] = 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}"}) + 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 + # policy, under which a bodyless or non-object invoke cannot carry the envelope a native + # pod requires and is a failure, not a fallback signal. + try: + # Off the event loop: a cold resolve is an RSA unwrap of a couple of milliseconds, and + # this middleware sits in front of every invoke on the pod. + invocation_settings = await asyncio.to_thread(resolve_invocation_settings, raw_settings) + except Exception as exc: + # 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__}"}, + ) + return + if isinstance(parsed, dict): + try: + invocation_context = extract_context(parsed) + except InvocationSettingsError as exc: + # Includes an unknown schema_version: a producer this plugin cannot read fails + # loudly here rather than running with silently absent identity. The message is + # truncated because it can embed request-controlled values. + logger.warning("rejecting invalid %s: %.200s", RESERVED_CONTEXT_KEY, exc) + await _send_json(send, 422, {"detail": f"Invalid field: {RESERVED_CONTEXT_KEY}"}) + return + + # The joined body and its parsed tree can be tens of MB and are not needed past this point; + # the framework re-buffers and re-parses the replayed messages downstream, so holding these + # through the handler would double peak memory. + del body, parsed, raw_settings + + async def replay() -> dict: + if messages: + return messages.pop(0) + # Buffer drained: proxy the original channel so downstream still observes + # http.disconnect. + return await receive() + + with invocation_envelope(invocation_settings, invocation_context): + await self.app(scope, replay, send) + + +def install_invocation_envelope(app: FastAPI) -> None: + """Install out-of-schema envelope extraction on a FastAPI app. + + Idempotent per app: a host wrapper may install at app construction while a plugin that predates + the wrapper's support still calls this itself, and a double install would buffer and replay the + request body twice. + """ + if getattr(app.state, "invocation_envelope_installed", False): + return + app.state.invocation_envelope_installed = True + app.add_middleware(InvocationEnvelopeMiddleware) From 657faefe20bdcfc895b07a004f50a8a465e7fc06 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:25:25 -0400 Subject: [PATCH 03/12] ci: move workflows to the Python 3.11 floor --- .github/workflows/ci.yml | 4 ++-- .github/workflows/release.yml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f70a33..c5b4fd7 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 2a6ad8e..f8fb085 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,7 +6,7 @@ on: - published env: - PYTHON_VERSION: "3.10" + PYTHON_VERSION: "3.11" jobs: release: From f0cd1497e9840e8dc11877744b79419eefdde0d6 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Tue, 4 Aug 2026 09:03:36 -0400 Subject: [PATCH 04/12] feat(etl-uvicorn): settings-scoped cache for per-invoke derived state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A plugin consuming current_invocation_settings() builds its handler per distinct settings payload instead of once at boot, and construction typically does network work (model resolution, prechecks). This gives that pattern one home next to the accessor that creates the need: settings_cache_key digests the canonical settings JSON so secret-bearing payloads are never raw keys, and SettingsScopedCache memoizes derived state bounded by both size and age — age matters because state built from since-rotated credentials must not outlive them on a quiet pod. Stdlib-only, so the package's dependency set is unchanged. --- test/api/test_settings_scoped_cache.py | 94 +++++++++++++++++++ .../invocation_settings.py | 71 +++++++++++++- 2 files changed, 163 insertions(+), 2 deletions(-) create mode 100644 test/api/test_settings_scoped_cache.py diff --git a/test/api/test_settings_scoped_cache.py b/test/api/test_settings_scoped_cache.py new file mode 100644 index 0000000..4820ebf --- /dev/null +++ b/test/api/test_settings_scoped_cache.py @@ -0,0 +1,94 @@ +from unittest.mock import MagicMock + +import pytest + +from unstructured_platform_plugins.invocation_settings import ( + SettingsScopedCache, + settings_cache_key, +) + + +class TestSettingsCacheKey: + def test_key_is_insensitive_to_key_order(self): + assert settings_cache_key({"a": 1, "b": 2}) == settings_cache_key({"b": 2, "a": 1}) + + def test_key_is_sensitive_to_values(self): + assert settings_cache_key({"a": 1}) != settings_cache_key({"a": 2}) + + def test_secret_values_do_not_appear_in_the_key(self): + secret = "sk-super-secret-credential" + key = settings_cache_key({"api_key": secret}) + assert secret not in key + + +class TestSettingsScopedCache: + def test_second_lookup_with_same_settings_does_not_rebuild(self): + cache = SettingsScopedCache() + build = MagicMock(return_value="handler") + + first = cache.get_or_build({"model": "a"}, build) + second = cache.get_or_build({"model": "a"}, build) + + assert first == second == "handler" + build.assert_called_once() + + def test_distinct_settings_build_distinct_values(self): + cache = SettingsScopedCache() + + first = cache.get_or_build({"model": "a"}, lambda: object()) + second = cache.get_or_build({"model": "b"}, lambda: object()) + + assert first is not second + + def test_entry_expires_after_ttl(self): + now = [0.0] + cache = SettingsScopedCache(ttl_seconds=10, clock=lambda: now[0]) + build = MagicMock(return_value="handler") + + cache.get_or_build({"model": "a"}, build) + now[0] = 11.0 + cache.get_or_build({"model": "a"}, build) + + assert build.call_count == 2 + + def test_entry_survives_within_ttl(self): + now = [0.0] + cache = SettingsScopedCache(ttl_seconds=10, clock=lambda: now[0]) + build = MagicMock(return_value="handler") + + cache.get_or_build({"model": "a"}, build) + now[0] = 9.0 + cache.get_or_build({"model": "a"}, build) + + build.assert_called_once() + + def test_size_bound_evicts_least_recently_used(self): + cache = SettingsScopedCache(maxsize=2) + builds = {name: MagicMock(return_value=name) for name in ("a", "b", "c")} + + cache.get_or_build({"model": "a"}, builds["a"]) + cache.get_or_build({"model": "b"}, builds["b"]) + # Refresh "a" so "b" is the eviction candidate when "c" lands. + cache.get_or_build({"model": "a"}, builds["a"]) + cache.get_or_build({"model": "c"}, builds["c"]) + + cache.get_or_build({"model": "a"}, builds["a"]) + cache.get_or_build({"model": "b"}, builds["b"]) + + builds["a"].assert_called_once() + assert builds["b"].call_count == 2 + + def test_clear_forces_rebuild(self): + cache = SettingsScopedCache() + build = MagicMock(return_value="handler") + + cache.get_or_build({"model": "a"}, build) + cache.clear() + cache.get_or_build({"model": "a"}, build) + + assert build.call_count == 2 + + @pytest.mark.parametrize("kwargs", [{"ttl_seconds": 0}, {"ttl_seconds": -1}, {"maxsize": 0}]) + def test_degenerate_bounds_are_rejected(self, kwargs): + with pytest.raises(ValueError): + SettingsScopedCache(**kwargs) diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index 73d0338..ded7c13 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/unstructured_platform_plugins/invocation_settings.py @@ -14,12 +14,16 @@ from __future__ import annotations import asyncio +import hashlib import json import logging -from collections.abc import Iterator +import threading +import time +from collections import OrderedDict +from collections.abc import Callable, Iterator, Mapping from contextlib import contextmanager from contextvars import ContextVar -from typing import Any, Optional +from typing import Any, Optional, TypeVar from fastapi import FastAPI from starlette.types import ASGIApp, Receive, Scope, Send @@ -36,6 +40,8 @@ logger = logging.getLogger(__name__) +T = TypeVar("T") + _METADATA_PATH = "/metadata" _INVOKE_PATH = "/invoke" @@ -241,3 +247,64 @@ def install_invocation_envelope(app: FastAPI) -> None: return app.state.invocation_envelope_installed = True app.add_middleware(InvocationEnvelopeMiddleware) + + +def settings_cache_key(invocation_settings: Mapping[str, Any]) -> str: + """Digest of the canonical settings JSON, safe as a cache key for secret-bearing payloads.""" + return hashlib.sha256(json.dumps(invocation_settings, sort_keys=True).encode()).hexdigest() + + +class SettingsScopedCache: + """Bind expensive derived state (clients, models, handlers) to the settings that built it. + + A plugin consuming ``current_invocation_settings()`` builds its handler per distinct settings + payload instead of once at boot, and construction typically does network work (model + resolution, prechecks), so results are memoized keyed by ``settings_cache_key``. Both bounds + matter under shared tenancy: size caps how many distinct payloads stay live, and age evicts + state built from credentials that may since have been rotated — eviction driven only by the + count of distinct payloads can take arbitrarily long on a quiet pod. + + Thread-safe for lookups and inserts. Concurrent misses for the same settings may build twice; + the extra build is wasted work, never wrong state. + """ + + def __init__( + self, + *, + ttl_seconds: float = 15 * 60, + maxsize: int = 32, + clock: Callable[[], float] = time.monotonic, + ) -> None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + if maxsize < 1: + raise ValueError("maxsize must be at least 1") + self._ttl_seconds = float(ttl_seconds) + self._maxsize = maxsize + self._clock = clock + self._lock = threading.Lock() + self._entries: OrderedDict[str, tuple[float, Any]] = OrderedDict() + + def get_or_build(self, invocation_settings: Mapping[str, Any], build: Callable[[], T]) -> T: + """Return the cached value for these settings, building it on a miss.""" + key = settings_cache_key(invocation_settings) + now = self._clock() + with self._lock: + entry = self._entries.get(key) + if entry is not None: + expires_at, value = entry + if now < expires_at: + self._entries.move_to_end(key) + return value + del self._entries[key] + value = build() + with self._lock: + self._entries[key] = (now + self._ttl_seconds, value) + self._entries.move_to_end(key) + while len(self._entries) > self._maxsize: + self._entries.popitem(last=False) + return value + + def clear(self) -> None: + with self._lock: + self._entries.clear() From a17268c56b96132dbbbaa88c668138bab0dc347d Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Tue, 4 Aug 2026 10:40:01 -0400 Subject: [PATCH 05/12] fix(etl-uvicorn): map context errors through the blame taxonomy An invocation_context with an unreadable schema_version is deployment skew between platform components; answering 422 let an upstream blame classifier pin it on the caller. Context failures now take their status from http_status_for like settings failures already did: malformed fields stay the caller's 422, version skew answers 500 with the class name only. Also documents the two capability tiers on /metadata: the unconditional strings are transport-level facts the middleware makes true for every wrapped app; invoke_with_sealed_dag_node_settings is the consumption claim and stays a per-plugin opt-in. --- test/api/test_invocation_middleware.py | 8 +++--- .../invocation_settings.py | 27 ++++++++++++++----- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/test/api/test_invocation_middleware.py b/test/api/test_invocation_middleware.py index 2b8b125..83b9ebf 100644 --- a/test/api/test_invocation_middleware.py +++ b/test/api/test_invocation_middleware.py @@ -226,16 +226,18 @@ def test_non_dict_reserved_field_is_rejected(self): assert sent[0]["status"] == 422 assert b"invocation_settings" in sent[1]["body"] - def test_context_with_an_unreadable_schema_version_is_rejected(self): + 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 - # must not be downgraded to that. + # must not be downgraded to that. And it is deployment skew, not a caller fault — a 422 + # would let an upstream blame classifier pin version skew on the customer. downstream, sent = _run_middleware( json.dumps({"invocation_context": {"schema_version": "99"}}).encode() ) assert downstream.body is None - assert sent[0]["status"] == 422 + assert sent[0]["status"] == 500 assert b"invocation_context" in sent[1]["body"] + assert b"UnsupportedContextVersionError" in sent[1]["body"] def test_malformed_context_is_rejected(self): downstream, sent = _run_middleware( diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index ded7c13..73288ea 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/unstructured_platform_plugins/invocation_settings.py @@ -92,8 +92,15 @@ def add_metadata_route( `/metadata` is the plugin API spec's own discovery surface (`PluginMetadataOutput`): capability flags are strings in its `capabilities` list, which is where the controller looks before forwarding the reserved fields — no controller-private probe route. - `invoke_with_sealed_dag_node_settings` additionally advertises that the plugin can be invoked - with a sealed `dag_node_settings` member and open it itself. + + The two tiers make different claims. `invocation_settings` / `invocation_context` are + *transport-level* facts, advertised unconditionally because the installed middleware makes + them true for every wrapped app: the reserved fields will be received, resolved, and bound — + or the request failed. They say nothing about whether the handler reads the binding. + `invoke_with_sealed_dag_node_settings` is the *consumption* claim — this plugin opens sealed + `dag_node_settings` itself and its handler acts on the result — and stays a per-plugin opt-in + set in the same change that makes it true, because it is the flag that invites the controller + to seal settings to this pod in place of any other settings source. Last call wins: the payload lives on `app.state` and every call overwrites it, while the route is registered once. A host wrapper may register with default capabilities at app construction @@ -213,11 +220,19 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: try: invocation_context = extract_context(parsed) except InvocationSettingsError as exc: - # Includes an unknown schema_version: a producer this plugin cannot read fails - # loudly here rather than running with silently absent identity. The message is - # truncated because it can embed request-controlled values. + # A context this plugin cannot read fails loudly here rather than running with + # silently absent identity. Status comes from the blame taxonomy: a malformed + # field is the caller's 422, but an unreadable schema_version is deployment skew + # between platform components and must not read as a caller fault. The log line is + # truncated because the message can embed request-controlled values. logger.warning("rejecting invalid %s: %.200s", RESERVED_CONTEXT_KEY, exc) - await _send_json(send, 422, {"detail": f"Invalid field: {RESERVED_CONTEXT_KEY}"}) + status = http_status_for(exc) + detail = ( + f"Invalid field: {RESERVED_CONTEXT_KEY}" + if status == 422 + else f"Unusable {RESERVED_CONTEXT_KEY}: {type(exc).__name__}" + ) + await _send_json(send, status, {"detail": detail}) return # The joined body and its parsed tree can be tens of MB and are not needed past this point; From 2d66570fde101aac149439785e5dbda44b8c90c5 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Tue, 4 Aug 2026 12:00:58 -0400 Subject: [PATCH 06/12] feat(etl-uvicorn): declare blame in failure responses instead of encoding it in status codes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- test/api/test_api.py | 33 +++++++++++++++++++ test/api/test_invocation_middleware.py | 10 ++++-- test/assets/exception_status_code.py | 12 +++++++ .../etl_uvicorn/api_generator.py | 8 ++++- .../invocation_settings.py | 22 +++++++++---- 5 files changed, 74 insertions(+), 11 deletions(-) diff --git a/test/api/test_api.py b/test/api/test_api.py index 83e2b23..f559333 100644 --- a/test/api/test_api.py +++ b/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] ) diff --git a/test/api/test_invocation_middleware.py b/test/api/test_invocation_middleware.py index 83b9ebf..9906d8c 100644 --- a/test/api/test_invocation_middleware.py +++ b/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( diff --git a/test/assets/exception_status_code.py b/test/assets/exception_status_code.py index 6e35997..f20cf93 100644 --- a/test/assets/exception_status_code.py +++ b/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") diff --git a/unstructured_platform_plugins/etl_uvicorn/api_generator.py b/unstructured_platform_plugins/etl_uvicorn/api_generator.py index 95da043..2426ecc 100644 --- a/unstructured_platform_plugins/etl_uvicorn/api_generator.py +++ b/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: diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index 73288ea..5adb52b 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/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; From 93c3f5446226406904cb96851055966db832fda2 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Tue, 4 Aug 2026 22:34:49 -0400 Subject: [PATCH 07/12] feat(etl-uvicorn): own the invocation-context model and the blame status spelling unstructured_platform_plugins.invocation_context holds the /invoke identity contract: InvocationContext, extract_context, dimensions, the reserved context key, the dimension fields, the supported versions, and UnsupportedContextVersionError. The context is protocol identity - no crypto, no secrets - so it ships 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. http_status_for - the HTTP spelling of the library's normative blame -> status rule - lives with the middleware that emits the responses. --- CHANGELOG.md | 21 ++- test/api/test_invocation_context.py | 153 +++++++++++++++++ .../invocation_context.py | 158 ++++++++++++++++++ .../invocation_settings.py | 35 +++- 4 files changed, 353 insertions(+), 14 deletions(-) create mode 100644 test/api/test_invocation_context.py create mode 100644 unstructured_platform_plugins/invocation_context.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8591439..b71f6b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,12 +2,21 @@ * **This package now owns the `/invoke` transport for the reserved fields.** `unstructured_platform_plugins.invocation_settings` holds the ASGI middleware, the `/metadata` - capability route, and the request-scoped binding. It sits on `utic-invocation-settings >=0.4.0`, - which owns the *contract* — which keys carry 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 body buffering and - route registration belong here, where a web framework is already a dependency. Nothing about the - wire format is decided in this repository. + capability route, the request-scoped binding, 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 body + buffering 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 diff --git a/test/api/test_invocation_context.py b/test/api/test_invocation_context.py new file mode 100644 index 0000000..f07a922 --- /dev/null +++ b/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 diff --git a/unstructured_platform_plugins/invocation_context.py b/unstructured_platform_plugins/invocation_context.py new file mode 100644 index 0000000..4cb898f --- /dev/null +++ b/unstructured_platform_plugins/invocation_context.py @@ -0,0 +1,158 @@ +"""The ``invocation_context`` companion to the settings envelope. + +Where ``invocation_settings`` carries *what* a plugin should be configured with, the context +carries *who* the invocation is for: the identity facets a shared-tenancy pod can no longer read +from its process environment. It travels in a second reserved, out-of-schema field of the +``/invoke`` body, extracted by the same middleware that resolves the settings field. + +The context is `/invoke` protocol identity, not settings security: it touches no crypto and no +secrets, and it evolves with the plugin protocol this package defines. The errors it raises come +from the shared ``InvocationSettingsError`` taxonomy so hosts classify context failures with the +same ``reason``/``blame`` machinery as settings failures. + +The model below is the **consumer** view of that contract, deliberately lenient: unknown keys are +preserved so a newer producer does not break an older plugin, and every identity field is optional +so a partially-populated context degrades to "less telemetry" rather than a failed invoke. The one +thing it is strict about is ``schema_version`` — that field exists to make an incompatible producer +detectable, which it can only do if somebody actually reads it. The payload is what carries the +version, not the route, so evolving the contract does not mean adding endpoints. +""" + +from __future__ import annotations + +from typing import Any, Mapping + +import pydantic +from utic_invocation_settings import Blame, InvocationSettingsError, MalformedEnvelopeError + +# Reserved key carrying the invocation context in the invoke request body. +RESERVED_CONTEXT_KEY = "invocation_context" + +# Context payload versions this package understands. Additive keys do not bump this; a change that +# would make an old consumer misread an existing key does. +SUPPORTED_CONTEXT_VERSIONS = frozenset({"1"}) + +# The identity facets that become telemetry dimensions. Shared rather than per-service policy: +# every hop on one invocation's path has to pick the same fields, or the same request is attributed +# differently depending on which component emitted the event. Excludes the batch fields, which +# describe the work rather than who it belongs to. +DIMENSION_FIELDS = ( + "invocation_id", + "tenant_id", + "org_id", + "job_id", + "workflow_id", + "attribution_id", + "dag_node_id", + "dag_node_type", + "dag_node_subtype", + "record_id", + "attempt", +) + +# Sentinel distinguishing a truly-absent reserved key from one present with a ``None`` value. +_ABSENT = object() + + +class UnsupportedContextVersionError(InvocationSettingsError): + """The ``invocation_context`` declares a ``schema_version`` this package does not understand. + + A producer upgrade this consumer cannot follow — deployment skew between platform components, + not a fault in the request. ``CONTENT`` (a 5xx) rather than ``CALLER``: contexts are produced + by the platform's own claim pipeline, and a 422 would make an upstream blame classifier pin a + version-skew failure on the customer. Loud at the first request rather than silently absent + telemetry dimensions later. + """ + + reason = "unsupported_context_version" + blame = Blame.CONTENT + + +class InvocationContext(pydantic.BaseModel): + """Request-scoped identity delivered alongside one claimed unit of work. + + ``extra="allow"`` keeps forward compatibility: fields added by a newer producer survive round + trips and stay reachable via ``model_extra`` instead of being silently dropped. + """ + + model_config = pydantic.ConfigDict(extra="allow") + + schema_version: str = "1" + + invocation_id: str | None = None + job_id: str | None = None + workflow_id: str | None = None + attribution_id: str | None = None + tenant_id: str | None = None + org_id: str | None = None + dag_node_id: str | None = None + dag_node_type: str | None = None + dag_node_subtype: str | None = None + record_id: str | None = None + attempt: int | None = None + job_created_timestamp: str | None = None + + # Added by the controller on the way to the plugin, not by the work API. The batch pair is + # index-aligned: entry i of `invocation_ids` is the invocation id of record i, or None where + # that record carried no context. There is deliberately no `work_dir` field: scratch space is + # the plugin's implementation detail (tempfile / uuid-named paths), not invoke-contract surface. + record_ids: list[str] | None = None + invocation_ids: list[str | None] | None = None + + @pydantic.field_validator("schema_version") + @classmethod + def _known_version(cls, value: str) -> str: + if value not in SUPPORTED_CONTEXT_VERSIONS: + raise ValueError( + f"unsupported invocation_context schema_version {value!r}; " + f"this package understands {sorted(SUPPORTED_CONTEXT_VERSIONS)}" + ) + return value + + +def extract_context(payload: Mapping[str, Any]) -> InvocationContext | None: + """Return the :class:`InvocationContext` from ``payload[RESERVED_CONTEXT_KEY]``. + + Returns ``None`` only when the reserved key is **absent** — the transitional signal that the + caller is an older controller. A present-but-invalid value fails closed rather than degrading + to "no context", because a context that silently vanishes takes a pod's tenant attribution with + it. + + A recognizable context carrying an unknown ``schema_version`` raises + :class:`UnsupportedContextVersionError` so a producer upgrade is loud at the first request + instead of showing up later as absent telemetry dimensions. + """ + raw = payload.get(RESERVED_CONTEXT_KEY, _ABSENT) + if raw is _ABSENT: + return None + if isinstance(raw, InvocationContext): + return raw + try: + return InvocationContext.model_validate(raw) + except pydantic.ValidationError as exc: + errors = exc.errors() + if any(error["loc"] == ("schema_version",) for error in errors): + raise UnsupportedContextVersionError( + f"unsupported invocation_context schema_version: " + f"{_reported_version(raw)!r}; expected one of {sorted(SUPPORTED_CONTEXT_VERSIONS)}" + ) from None + # `from None` so the pydantic error tree does not cross the domain-error boundary; a count + # plus the first message is enough signal. Mirrors the envelope extraction. + raise MalformedEnvelopeError( + f"invalid invocation_context: {exc.error_count()} validation error(s), " + f"first: {errors[0]['msg']}" + ) from None + + +def _reported_version(raw: Any) -> Any: + """The offending ``schema_version``, for the error message only. Never trusted.""" + return raw.get("schema_version") if isinstance(raw, Mapping) else None + + +def dimensions(context: InvocationContext | None) -> dict[str, Any]: + """The context's populated identity facets, ready to bind as telemetry dimensions.""" + if context is None: + return {} + return { + field: value for field in DIMENSION_FIELDS if (value := getattr(context, field)) is not None + } diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index 5adb52b..7b73d43 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/unstructured_platform_plugins/invocation_settings.py @@ -1,9 +1,13 @@ """Transport for the reserved `/invoke` fields: ASGI middleware, `/metadata`, request binding. -The *contract* — which keys carry settings, how a sealed envelope is told from plaintext, and what -an absent field is allowed to mean — lives in `utic_invocation_settings.invoke`, next to the crypto -it governs. This module is the other half: getting the payload off the wire and the result to the -handler. It owns no policy; every decision about a payload it delegates. +The *settings contract* — which key carries settings, how a sealed envelope is told from +plaintext, and what an absent field is allowed to mean — lives in +`utic_invocation_settings.invoke`, next to the crypto it governs; every decision about a settings +payload is delegated there. The *identity contract* — the `invocation_context` model — is +`/invoke` protocol rather than settings security and lives in this package's +`invocation_context` module. This module is the delivery mechanism for both: getting the payloads +off the wire and the results to the handler, and spelling the shared `blame` taxonomy as HTTP +statuses. The reserved fields are a first-class HTTP contract independent of the generated input schema. They never appear in a plugin's declared signature, so `wrap_in_fastapi` keeps producing a handler model @@ -29,18 +33,33 @@ from starlette.types import ASGIApp, Receive, Scope, Send from utic_invocation_settings import ( INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_CAPABILITY, - RESERVED_CONTEXT_KEY, RESERVED_ENVELOPE_KEY, - InvocationContext, + Blame, InvocationSettingsError, MalformedEnvelopeError, - extract_context, - http_status_for, resolve_invocation_settings, ) +from unstructured_platform_plugins.invocation_context import ( + RESERVED_CONTEXT_KEY, + InvocationContext, + extract_context, +) + logger = logging.getLogger(__name__) + +def http_status_for(error: BaseException) -> int: + """The HTTP status this transport answers for a failed resolution, from ``blame``. + + One rule, the one the library README states normatively: ``Blame.CALLER`` -> 422, everything + else -> 500. The line it draws is whether a different request would work. Sealing drift, an + envelope for another recipient and a broken local mount are all 5xx, which keeps a controller's + blame classification off the customer, whose request was fine. Anything that is not a + classified error is a 500: an unclassified failure is not the caller's. + """ + return 422 if getattr(error, "blame", None) is Blame.CALLER else 500 + T = TypeVar("T") _METADATA_PATH = "/metadata" From 8908808e45af9c7132e7cc4b076b16ea86f4bfb4 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Mon, 10 Aug 2026 14:10:29 -0400 Subject: [PATCH 08/12] refactor(etl-uvicorn): bind /invoke envelope without body replay (#75) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary - Replace raw `/invoke` body buffering and replay with a FastAPI dependency that reads Starlette's cached JSON parse. - Keep `InvokeBodyLimitMiddleware` below FastAPI as a streaming byte counter, so oversized bodies are rejected without a second buffer. - Install binding through the router's public dependency list before `POST /invoke` is registered; remove private `route.dependant` mutation. - Re-enter the captured invocation binding inside async-generator response iteration, so streaming plugins see settings on the repository's locked FastAPI 0.117.1 as well as newer FastAPI releases. - Preserve the explicit `invoke_with_sealed_dag_node_settings` capability opt-in and its wrapper, generator, and CLI arguments. - Align transport tests with the shared contract: sealed settings are accepted only at `invocation_settings.dag_node_settings`; a bare envelope fails closed. ## Why ASGI middleware has to consume the raw receive channel before FastAPI can parse it, which forced the old implementation to buffer, parse, and replay the request. Route-level extraction shares the framework's body and JSON caches instead. The initial dependency version still had three correctness problems: FastAPI 0.117.1 closed yield dependencies before `StreamingResponse` iteration, installation mutated FastAPI's private dependency graph after route registration, and unconditional sealed-capability advertisement conflated transport support with handler consumption. This revision fixes all three without raising the FastAPI floor. ## Impact - `install_invocation_envelope(app)` must run before `POST /invoke` is registered. It may run after unrelated routes such as `/metadata`, which preserves the hand-written plugin integration order. - The dependency is a path/method-aware no-op outside `POST /invoke`, including mixed-method routes and rooted deployments. - Malformed JSON on a declared FastAPI body model continues to use FastAPI's own validation response. ## Validation - `pytest -q` on FastAPI 0.117.1 / Starlette 0.48.0 with the current utic-invocation-settings 0.4.0 branch — 153 passed. - Focused transport tests on FastAPI 0.141.1 / Starlette 1.6.0 — 38 passed. - `ruff check .` — clean. - `ruff format --check` on changed Python files — clean. - `git diff --check` — clean. --- CHANGELOG.md | 33 +- test/api/test_invocation_envelope.py | 505 ++++++++++++++++++ test/api/test_invocation_middleware.py | 458 ---------------- test/api/test_invocation_settings.py | 24 +- .../etl_uvicorn/api_generator.py | 72 +-- .../invocation_settings.py | 310 ++++++----- 6 files changed, 774 insertions(+), 628 deletions(-) create mode 100644 test/api/test_invocation_envelope.py delete mode 100644 test/api/test_invocation_middleware.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b71f6b2..a6d828e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,13 +1,14 @@ ## 0.1.0 * **This package now owns the `/invoke` transport for the reserved fields.** - `unstructured_platform_plugins.invocation_settings` holds the ASGI middleware, the `/metadata` - capability route, the request-scoped binding, and `http_status_for` — the HTTP spelling of the + `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 body - buffering and route registration belong here, where a web framework is already a dependency. + 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`, @@ -23,17 +24,27 @@ 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 middleware installs once and the last `/metadata` + Repeated installation is safe: the dependency installs once and the last `/metadata` registration wins. -* **New opt-in `invoke_with_sealed_dag_node_settings` advertisement.** Pass +* **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 consumes sealed per-invocation settings. - 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. + 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 middleware fronts every invoke on the pod, so it is dispatched with + 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 diff --git a/test/api/test_invocation_envelope.py b/test/api/test_invocation_envelope.py new file mode 100644 index 0000000..5c243a5 --- /dev/null +++ b/test/api/test_invocation_envelope.py @@ -0,0 +1,505 @@ +"""The transport for the reserved /invoke fields: dependency binding, /metadata, the body cap. + +The *contract* these exercise — which shapes carry settings, what absence means — is owned and +tested in `utic_invocation_settings`. What is tested here is delivery: that the framework-parsed +body is resolved and bound for the handler, and that a payload which cannot be used fails the +request instead of reaching a handler as absence. +""" + +from __future__ import annotations + +import asyncio +import json +from base64 import b64decode, b64encode +from typing import Optional + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient +from utic_invocation_settings import ( + DAG_NODE_SETTINGS_KEY, + REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, + default_resolver, + reset_workload_identity_cache, +) +from utic_invocation_settings.crypto import seal_settings + +from unstructured_platform_plugins.invocation_settings import ( + InvokeBodyLimitMiddleware, + add_metadata_route, + current_invocation_context, + current_invocation_settings, + install_invocation_envelope, +) + +SENTINEL_SECRET = "sealed-settings-sentinel-secret" + + +@pytest.fixture(scope="session") +def private_key() -> rsa.RSAPrivateKey: + return rsa.generate_private_key(public_exponent=65537, key_size=3072) + + +@pytest.fixture(autouse=True) +def isolated_identity(monkeypatch): + """The identity memo and the resolver caches both outlive a test; an inherited env var or a + stale entry would make these order-dependent.""" + for var in ( + "WORKLOAD_IDENTITY_DIR", + "INVOCATION_SETTINGS_KEY_DIR", + REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, + ): + monkeypatch.delenv(var, raising=False) + reset_workload_identity_cache() + default_resolver().clear_caches() + yield + reset_workload_identity_cache() + default_resolver().clear_caches() + + +@pytest.fixture +def key_dir(tmp_path, private_key, monkeypatch, isolated_identity): + (tmp_path / "tls.key").write_bytes( + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + monkeypatch.setenv("WORKLOAD_IDENTITY_DIR", str(tmp_path)) + reset_workload_identity_cache() + return tmp_path + + +def sealed_payload(private_key: rsa.RSAPrivateKey, settings: dict) -> dict: + return seal_settings(settings, private_key.public_key()).model_dump( + mode="json", exclude_none=True + ) + + +def tampered(sealed: dict) -> dict: + ciphertext = bytearray(b64decode(sealed["content_encryption"]["ciphertext"])) + ciphertext[0] ^= 0x01 + sealed["content_encryption"]["ciphertext"] = b64encode(bytes(ciphertext)).decode() + return sealed + + +BASE_CAPABILITIES = [ + "invocation_settings", + "invocation_context", +] +ALL_CAPABILITIES = [ + *BASE_CAPABILITIES, + "invoke_with_sealed_dag_node_settings", +] + + +class TestMetadataRoute: + def test_advertises_transport_capabilities_by_default(self): + app = FastAPI() + add_metadata_route(app, identifier="plugin.test") + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload == { + "api_version": "3", + "identifier": "plugin.test", + "capabilities": BASE_CAPABILITIES, + } + + def test_sealed_consumption_capability_is_opt_in(self): + app = FastAPI() + add_metadata_route(app, invoke_with_sealed_dag_node_settings=True) + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload["capabilities"] == ALL_CAPABILITIES + + def test_last_call_wins(self): + # A host wrapper registers /metadata at construction; the plugin's later call with its own + # identifier must replace it, not be shadowed by route order. + app = FastAPI() + add_metadata_route(app, identifier="wrapper.default") + add_metadata_route(app, identifier="plugin.test", invoke_with_sealed_dag_node_settings=True) + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload["identifier"] == "plugin.test" + assert payload["capabilities"] == ALL_CAPABILITIES + + def test_replaces_a_directly_registered_metadata_route(self): + # A route the app registered itself would otherwise win by route order and pin its stale + # payload. + app = FastAPI() + + @app.get("/metadata") + async def stale_metadata() -> dict: + return {"api_version": "3", "identifier": "stale", "capabilities": []} + + add_metadata_route(app, identifier="plugin.test", invoke_with_sealed_dag_node_settings=True) + + with TestClient(app) as client: + payload = client.get("/metadata").json() + + assert payload["identifier"] == "plugin.test" + assert payload["capabilities"] == ALL_CAPABILITIES + + +class _Recorder: + """What the /invoke handler observed: the bound envelope, and whether it ran at all.""" + + def __init__(self): + self.called = False + self.seen_settings = "unset" + self.seen_context = "unset" + + +def _envelope_app(recorder: _Recorder, max_body_bytes: Optional[int] = None) -> FastAPI: + """A hand-rolled host app: the /invoke route reads the raw request, like a plugin that owns + its own route, so any body shape reaches the handler unless the dependency rejects it.""" + app = FastAPI() + if max_body_bytes is None: + install_invocation_envelope(app) + else: + install_invocation_envelope(app, max_body_bytes=max_body_bytes) + + @app.post("/invoke") + async def invoke(request: Request) -> dict: + recorder.called = True + recorder.seen_settings = current_invocation_settings() + recorder.seen_context = current_invocation_context() + return {} + + @app.get("/schema") + async def schema() -> dict: + recorder.called = True + return {} + + return app + + +def _post_invoke(payload, recorder: Optional[_Recorder] = None, **app_kwargs): + recorder = recorder if recorder is not None else _Recorder() + app = _envelope_app(recorder, **app_kwargs) + with TestClient(app, raise_server_exceptions=False) as client: + if isinstance(payload, bytes): + response = client.post("/invoke", content=payload) + else: + response = client.post("/invoke", json=payload) + return recorder, response + + +class TestInvocationEnvelopeBinding: + def test_binds_reserved_fields(self): + recorder, response = _post_invoke( + { + "element_dicts": "/in.json", + "invocation_settings": {"model": "m"}, + "invocation_context": {"schema_version": "1", "job_id": "job-1"}, + } + ) + + assert response.status_code == 200 + assert recorder.seen_settings == {"model": "m"} + assert recorder.seen_context.job_id == "job-1" + + def test_absent_fields_bind_none(self): + recorder, response = _post_invoke({"element_dicts": "/in.json"}) + + assert response.status_code == 200 + assert recorder.seen_settings is None + assert recorder.seen_context is None + + def test_non_dict_reserved_field_is_rejected(self): + recorder, response = _post_invoke({"invocation_settings": "not-a-dict"}) + + assert not recorder.called + assert response.status_code == 422 + body = response.json() + 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 + # must not be downgraded to that. And it is deployment skew, not a caller fault — a 422 + # would let an upstream blame classifier pin version skew on the customer. + recorder, response = _post_invoke({"invocation_context": {"schema_version": "99"}}) + + assert not recorder.called + assert response.status_code == 500 + body = response.json() + assert "invocation_context" in body["detail"] + assert "UnsupportedContextVersionError" in body["detail"] + assert body["reason"] == "unsupported_context_version" + + def test_malformed_context_is_rejected(self): + recorder, response = _post_invoke({"invocation_context": "not-an-object"}) + + assert not recorder.called + assert response.status_code == 422 + + def test_non_invoke_routes_are_untouched(self): + recorder = _Recorder() + app = _envelope_app(recorder) + + with TestClient(app) as client: + response = client.get("/schema") + + assert response.status_code == 200 + assert recorder.called + + def test_envelope_does_not_leak_between_requests(self): + recorder = _Recorder() + app = _envelope_app(recorder) + + with TestClient(app) as client: + client.post("/invoke", json={"invocation_settings": {"model": "m"}}) + client.post("/invoke", json={"element_dicts": "/in.json"}) + + assert recorder.seen_settings is None + assert recorder.seen_context is None + + def test_install_after_an_invoke_route_fails_loudly(self): + app = FastAPI() + + @app.post("/invoke") + async def invoke() -> dict: + return {} + + with pytest.raises(RuntimeError, match="POST /invoke"): + install_invocation_envelope(app) + + def test_install_after_metadata_but_before_invoke_is_supported(self): + recorder = _Recorder() + app = FastAPI() + add_metadata_route(app, invoke_with_sealed_dag_node_settings=True) + install_invocation_envelope(app) + + @app.post("/invoke") + async def invoke() -> dict: + recorder.seen_settings = current_invocation_settings() + return {} + + with TestClient(app) as client: + response = client.post("/invoke", json={"invocation_settings": {"model": "m"}}) + + assert response.status_code == 200 + assert recorder.seen_settings == {"model": "m"} + + def test_mixed_method_route_does_not_bind_or_parse_get(self): + recorder = _Recorder() + app = FastAPI() + install_invocation_envelope(app) + + @app.api_route("/invoke", methods=["GET", "POST"]) + async def invoke() -> dict: + recorder.seen_settings = current_invocation_settings() + return {} + + with TestClient(app) as client: + response = client.get("/invoke") + + assert response.status_code == 200 + assert recorder.seen_settings is None + + def test_root_path_does_not_prevent_invoke_binding(self): + recorder = _Recorder() + app = _envelope_app(recorder) + + with TestClient(app, root_path="/plugins/chunker") as client: + response = client.post("/invoke", json={"invocation_settings": {"model": "rooted"}}) + + assert response.status_code == 200 + assert recorder.seen_settings == {"model": "rooted"} + + def test_oversized_body_is_rejected(self): + recorder, response = _post_invoke( + {"invocation_settings": {"pad": "x" * 64}}, max_body_bytes=16 + ) + + assert not recorder.called + assert response.status_code == 413 + + +class TestInvokeBodyLimitMiddleware: + """ASGI-level behavior of the streaming byte counter: no buffering, disconnect pass-through.""" + + @staticmethod + def _run(middleware, scope, chunks) -> tuple[list, list]: + received = [] + sent = [] + + async def receive(): + return chunks.pop(0) + + async def send(message): + sent.append(message) + + async def downstream(scope, receive, send): + while True: + message = await receive() + received.append(message) + if message["type"] != "http.request" or not message.get("more_body"): + break + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"{}"}) + + middleware = middleware(downstream) + asyncio.run(middleware(scope, receive, send)) + return received, sent + + def test_chunked_body_over_the_cap_answers_413_and_cuts_downstream(self): + chunks = [ + {"type": "http.request", "body": b"x" * 10, "more_body": True}, + {"type": "http.request", "body": b"x" * 10, "more_body": False}, + ] + received, sent = self._run( + lambda app: InvokeBodyLimitMiddleware(app, max_body_bytes=16), + {"type": "http", "method": "POST", "path": "/invoke"}, + chunks, + ) + + assert received[-1] == {"type": "http.disconnect"} + assert sent[0]["status"] == 413 + + def test_client_disconnect_passes_through_uncounted(self): + chunks = [{"type": "http.disconnect"}] + received, sent = self._run( + lambda app: InvokeBodyLimitMiddleware(app, max_body_bytes=16), + {"type": "http", "method": "POST", "path": "/invoke"}, + chunks, + ) + + assert received == [{"type": "http.disconnect"}] + + def test_non_invoke_requests_are_not_capped(self): + chunks = [{"type": "http.request", "body": b"x" * 100, "more_body": False}] + received, sent = self._run( + lambda app: InvokeBodyLimitMiddleware(app, max_body_bytes=16), + {"type": "http", "method": "GET", "path": "/schema"}, + chunks, + ) + + assert sent[0]["status"] == 200 + + +class TestEnvelopeResolution: + """Sealed payloads through the binding dependency. The HTTP class comes from the library's + blame taxonomy, so only a caller-fixable fault is a 422.""" + + def test_bare_sealed_envelope_is_rejected(self, key_dir, private_key): + sealed = sealed_payload(private_key, {"api_key": SENTINEL_SECRET}) + + recorder, response = _post_invoke({"invocation_settings": sealed}) + + assert response.status_code == 500 + assert "MalformedDagNodeSettingsError" in response.json()["detail"] + assert not recorder.called + + def test_composite_dag_node_settings_member_is_opened(self, key_dir, private_key): + settings = {"api_key": SENTINEL_SECRET, "max_characters": 700} + composite = {DAG_NODE_SETTINGS_KEY: sealed_payload(private_key, settings)} + + recorder, response = _post_invoke({"invocation_settings": composite}) + + assert response.status_code == 200 + assert recorder.seen_settings == settings + + def test_plain_dict_settings_pass_through(self): + recorder, response = _post_invoke({"invocation_settings": {"model": "m"}}) + + assert response.status_code == 200 + assert recorder.seen_settings == {"model": "m"} + + def test_non_envelope_member_fails_as_platform_error(self, key_dir): + composite = {DAG_NODE_SETTINGS_KEY: {"model": "m"}} + + recorder, response = _post_invoke({"invocation_settings": composite}) + + assert response.status_code == 500 + assert "MalformedDagNodeSettingsError" in response.json()["detail"] + assert not recorder.called + + def test_undecryptable_envelope_fails_without_leaking_the_secret( + self, key_dir, private_key, caplog + ): + sealed = tampered(sealed_payload(private_key, {"api_key": SENTINEL_SECRET})) + composite = {DAG_NODE_SETTINGS_KEY: sealed} + + recorder, response = _post_invoke({"invocation_settings": composite}) + + assert response.status_code == 500 + detail = response.json()["detail"] + assert "DecryptionError" in detail + assert SENTINEL_SECRET not in caplog.text + assert SENTINEL_SECRET not in detail + assert not recorder.called + + def test_unmounted_identity_fails_as_platform_error(self, tmp_path, monkeypatch, private_key): + monkeypatch.setenv("WORKLOAD_IDENTITY_DIR", str(tmp_path / "missing")) + sealed = sealed_payload(private_key, {"api_key": SENTINEL_SECRET}) + composite = {DAG_NODE_SETTINGS_KEY: sealed} + + recorder, response = _post_invoke({"invocation_settings": composite}) + + assert response.status_code == 500 + assert "IdentityNotMountedError" in response.json()["detail"] + assert not recorder.called + + +class TestRequireSealedDagNodeSettings: + """A native pod's operator sets FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS in the same + decision that drops the init-secrets sidecar: without it, an invoke that arrived with no + envelope would fall back to a settings file that was never written.""" + + @pytest.fixture(autouse=True) + def _require_sealed(self, monkeypatch): + monkeypatch.setenv(REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, "true") + + def test_missing_settings_fail_as_platform_error(self): + recorder, response = _post_invoke({"element_dicts": "/tmp/x.json"}) + + assert response.status_code == 500 + assert "SealedDagNodeSettingsRequiredError" in response.json()["detail"] + assert not recorder.called + + def test_plaintext_settings_fail_as_platform_error(self): + recorder, response = _post_invoke({"invocation_settings": {"model": "m"}}) + + assert response.status_code == 500 + assert not recorder.called + + def test_sealed_settings_still_bind(self, key_dir, private_key): + settings = {"api_key": SENTINEL_SECRET} + composite = {DAG_NODE_SETTINGS_KEY: sealed_payload(private_key, settings)} + + recorder, response = _post_invoke({"invocation_settings": composite}) + + assert response.status_code == 200 + assert recorder.seen_settings == settings + + def test_bodyless_invoke_fails_as_platform_error(self): + # A bodyless invoke cannot carry the envelope; on a pod with no boot settings file it must + # not dispatch a handler with no settings source at all. + recorder, response = _post_invoke(b"") + + assert response.status_code == 500 + assert not recorder.called + + def test_non_object_json_body_fails_as_platform_error(self): + recorder, response = _post_invoke(json.dumps([{"element": 1}]).encode()) + + assert response.status_code == 500 + assert not recorder.called + + +def test_non_object_bodies_pass_through_when_sealed_settings_not_required(): + recorder, response = _post_invoke(b"") + + assert response.status_code == 200 + assert recorder.seen_settings is None diff --git a/test/api/test_invocation_middleware.py b/test/api/test_invocation_middleware.py deleted file mode 100644 index 9906d8c..0000000 --- a/test/api/test_invocation_middleware.py +++ /dev/null @@ -1,458 +0,0 @@ -"""The transport for the reserved /invoke fields: middleware, /metadata, request-scoped binding. - -The *contract* these exercise — which shapes carry settings, what absence means — is owned and -tested in `utic_invocation_settings`. What is tested here is delivery: that a raw body is read, -resolved, bound, and replayed intact, and that a payload which cannot be used fails the request -instead of reaching a handler as absence. -""" - -from __future__ import annotations - -import asyncio -import json -from base64 import b64decode, b64encode - -import pytest -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import rsa -from fastapi import FastAPI -from fastapi.testclient import TestClient -from utic_invocation_settings import ( - DAG_NODE_SETTINGS_KEY, - REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, - default_resolver, - reset_workload_identity_cache, -) -from utic_invocation_settings.crypto import seal_settings - -from unstructured_platform_plugins.invocation_settings import ( - InvocationEnvelopeMiddleware, - add_metadata_route, - current_invocation_context, - current_invocation_settings, -) - -SENTINEL_SECRET = "sealed-settings-sentinel-secret" - - -@pytest.fixture(scope="session") -def private_key() -> rsa.RSAPrivateKey: - return rsa.generate_private_key(public_exponent=65537, key_size=3072) - - -@pytest.fixture(autouse=True) -def isolated_identity(monkeypatch): - """The identity memo and the resolver caches both outlive a test; an inherited env var or a - stale entry would make these order-dependent.""" - for var in ("WORKLOAD_IDENTITY_DIR", "INVOCATION_SETTINGS_KEY_DIR", - REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR): - monkeypatch.delenv(var, raising=False) - reset_workload_identity_cache() - default_resolver().clear_caches() - yield - reset_workload_identity_cache() - default_resolver().clear_caches() - - -@pytest.fixture -def key_dir(tmp_path, private_key, monkeypatch, isolated_identity): - (tmp_path / "tls.key").write_bytes( - private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ) - ) - monkeypatch.setenv("WORKLOAD_IDENTITY_DIR", str(tmp_path)) - reset_workload_identity_cache() - return tmp_path - - -def sealed_payload(private_key: rsa.RSAPrivateKey, settings: dict) -> dict: - return seal_settings(settings, private_key.public_key()).model_dump( - mode="json", exclude_none=True - ) - - -def tampered(sealed: dict) -> dict: - ciphertext = bytearray(b64decode(sealed["content_encryption"]["ciphertext"])) - ciphertext[0] ^= 0x01 - sealed["content_encryption"]["ciphertext"] = b64encode(bytes(ciphertext)).decode() - return sealed - - -class TestMetadataRoute: - def test_advertises_settings_and_context_capabilities(self): - app = FastAPI() - add_metadata_route(app, identifier="plugin.test") - - with TestClient(app) as client: - payload = client.get("/metadata").json() - - assert payload == { - "api_version": "3", - "identifier": "plugin.test", - "capabilities": ["invocation_settings", "invocation_context"], - } - - def test_sealed_dag_node_settings_flag_advertises_capability(self): - app = FastAPI() - add_metadata_route(app, invoke_with_sealed_dag_node_settings=True) - - with TestClient(app) as client: - payload = client.get("/metadata").json() - - assert payload["capabilities"] == [ - "invocation_settings", - "invocation_context", - "invoke_with_sealed_dag_node_settings", - ] - - def test_last_call_wins(self): - # A host wrapper registers /metadata with defaults at construction; the plugin's later call - # with the sealed capability must replace it, not be shadowed by route order. - app = FastAPI() - add_metadata_route(app, identifier="plugin.test") - add_metadata_route(app, identifier="plugin.test", invoke_with_sealed_dag_node_settings=True) - - with TestClient(app) as client: - payload = client.get("/metadata").json() - - assert "invoke_with_sealed_dag_node_settings" in payload["capabilities"] - - def test_replaces_a_directly_registered_metadata_route(self): - # A route the app registered itself would otherwise win by route order and pin its stale - # payload. - app = FastAPI() - - @app.get("/metadata") - async def stale_metadata() -> dict: - return {"api_version": "3", "identifier": "stale", "capabilities": []} - - add_metadata_route(app, identifier="plugin.test", invoke_with_sealed_dag_node_settings=True) - - with TestClient(app) as client: - payload = client.get("/metadata").json() - - assert payload["identifier"] == "plugin.test" - assert "invoke_with_sealed_dag_node_settings" in payload["capabilities"] - - -def _invoke_scope(path: str = "/invoke", method: str = "POST") -> dict: - return {"type": "http", "method": method, "path": path} - - -def _receive_for(body: bytes): - chunks = [ - {"type": "http.request", "body": body[: len(body) // 2], "more_body": True}, - {"type": "http.request", "body": body[len(body) // 2 :], "more_body": False}, - ] - - async def receive(): - return chunks.pop(0) - - return receive - - -async def _ok(send): - await send({"type": "http.response.start", "status": 200, "headers": []}) - await send({"type": "http.response.body", "body": b"{}"}) - - -class _DownstreamApp: - """Records the replayed body and the envelope bound while handling.""" - - def __init__(self): - self.body = None - self.called = False - self.seen_settings = "unset" - self.seen_context = "unset" - - async def __call__(self, scope, receive, send): - body = b"" - while True: - message = await receive() - body += message.get("body", b"") - if not message.get("more_body"): - break - self.body = body - self.called = True - self.seen_settings = current_invocation_settings() - self.seen_context = current_invocation_context() - await _ok(send) - - -def _run_middleware(body: bytes, scope: dict | None = None) -> tuple[_DownstreamApp, list]: - downstream = _DownstreamApp() - middleware = InvocationEnvelopeMiddleware(downstream) - sent = [] - - async def send(message): - sent.append(message) - - asyncio.run(middleware(scope or _invoke_scope(), _receive_for(body), send)) - return downstream, sent - - -class TestInvocationEnvelopeMiddleware: - def test_binds_reserved_fields_and_replays_body(self): - body = json.dumps( - { - "element_dicts": "/in.json", - "invocation_settings": {"model": "m"}, - "invocation_context": {"schema_version": "1", "job_id": "job-1"}, - } - ).encode() - - downstream, sent = _run_middleware(body) - - assert downstream.body == body - assert downstream.seen_settings == {"model": "m"} - assert downstream.seen_context.job_id == "job-1" - assert sent[0]["status"] == 200 - - def test_absent_fields_bind_none(self): - downstream, _ = _run_middleware(json.dumps({"element_dicts": "/in.json"}).encode()) - - assert downstream.seen_settings is None - assert downstream.seen_context is None - - def test_non_dict_reserved_field_is_rejected(self): - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": "not-a-dict"}).encode() - ) - - assert downstream.body is None - assert sent[0]["status"] == 422 - 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 - # must not be downgraded to that. And it is deployment skew, not a caller fault — a 422 - # would let an upstream blame classifier pin version skew on the customer. - downstream, sent = _run_middleware( - json.dumps({"invocation_context": {"schema_version": "99"}}).encode() - ) - - assert downstream.body is None - assert sent[0]["status"] == 500 - 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( - json.dumps({"invocation_context": "not-an-object"}).encode() - ) - - assert downstream.body is None - assert sent[0]["status"] == 422 - - def test_non_invoke_requests_pass_through_untouched(self): - body = json.dumps({"invocation_settings": "not-a-dict"}).encode() - - downstream, sent = _run_middleware(body, scope=_invoke_scope(path="/schema", method="GET")) - - assert downstream.body == body - assert sent[0]["status"] == 200 - - def test_envelope_is_reset_after_request(self): - body = json.dumps({"invocation_settings": {"model": "m"}}).encode() - - async def scenario(): - middleware = InvocationEnvelopeMiddleware(_DownstreamApp()) - - async def send(_message): - pass - - await middleware(_invoke_scope(), _receive_for(body), send) - return current_invocation_settings(), current_invocation_context() - - assert asyncio.run(scenario()) == (None, None) - - def test_oversized_body_is_rejected(self): - body = json.dumps({"invocation_settings": {"pad": "x" * 64}}).encode() - - downstream = _DownstreamApp() - middleware = InvocationEnvelopeMiddleware(downstream, max_body_bytes=16) - sent = [] - - async def send(message): - sent.append(message) - - asyncio.run(middleware(_invoke_scope(), _receive_for(body), send)) - - assert downstream.body is None - assert sent[0]["status"] == 413 - - def test_drained_replay_proxies_disconnect(self): - body = json.dumps({"invocation_settings": {"model": "m"}}).encode() - - class _DisconnectWatcher: - def __init__(self): - self.saw_disconnect = False - - async def __call__(self, scope, receive, send): - while True: - message = await receive() - if message["type"] == "http.disconnect": - self.saw_disconnect = True - return - if not message.get("more_body"): - break - message = await receive() - self.saw_disconnect = message["type"] == "http.disconnect" - await _ok(send) - - chunks = [ - {"type": "http.request", "body": body, "more_body": False}, - {"type": "http.disconnect"}, - ] - - async def receive(): - return chunks.pop(0) - - watcher = _DisconnectWatcher() - middleware = InvocationEnvelopeMiddleware(watcher) - sent = [] - - async def send(message): - sent.append(message) - - asyncio.run(middleware(_invoke_scope(), receive, send)) - - assert watcher.saw_disconnect - - -class TestMiddlewareResolution: - """Sealed payloads through the middleware. The HTTP class comes from the library's blame - taxonomy, so only a caller-fixable fault is a 422.""" - - def test_sealed_envelope_binds_plaintext_settings(self, key_dir, private_key): - settings = {"api_key": SENTINEL_SECRET, "max_characters": 700} - sealed = sealed_payload(private_key, settings) - - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": sealed}).encode() - ) - - assert sent[0]["status"] == 200 - assert downstream.seen_settings == settings - - def test_composite_dag_node_settings_member_is_opened(self, key_dir, private_key): - settings = {"api_key": SENTINEL_SECRET, "max_characters": 700} - composite = {DAG_NODE_SETTINGS_KEY: sealed_payload(private_key, settings)} - - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": composite}).encode() - ) - - assert sent[0]["status"] == 200 - assert downstream.seen_settings == settings - - def test_plain_dict_settings_pass_through(self): - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": {"model": "m"}}).encode() - ) - - assert sent[0]["status"] == 200 - assert downstream.seen_settings == {"model": "m"} - - def test_non_envelope_member_fails_as_platform_error(self, key_dir): - composite = {DAG_NODE_SETTINGS_KEY: {"model": "m"}} - - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": composite}).encode() - ) - - assert sent[0]["status"] == 500 - assert "MalformedDagNodeSettingsError" in json.loads(sent[1]["body"])["detail"] - assert not downstream.called - - def test_undecryptable_envelope_fails_without_leaking_the_secret( - self, key_dir, private_key, caplog - ): - sealed = tampered(sealed_payload(private_key, {"api_key": SENTINEL_SECRET})) - - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": sealed}).encode() - ) - - assert sent[0]["status"] == 500 - detail = json.loads(sent[1]["body"])["detail"] - assert "DecryptionError" in detail - assert SENTINEL_SECRET not in caplog.text - assert SENTINEL_SECRET not in detail - assert not downstream.called - - def test_unmounted_identity_fails_as_platform_error(self, tmp_path, monkeypatch, private_key): - monkeypatch.setenv("WORKLOAD_IDENTITY_DIR", str(tmp_path / "missing")) - sealed = sealed_payload(private_key, {"api_key": SENTINEL_SECRET}) - - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": sealed}).encode() - ) - - assert sent[0]["status"] == 500 - assert "IdentityNotMountedError" in json.loads(sent[1]["body"])["detail"] - assert not downstream.called - - -class TestRequireSealedDagNodeSettings: - """A native pod's operator sets FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS in the same - decision that drops the init-secrets sidecar: without it, an invoke that arrived with no - envelope would fall back to a settings file that was never written.""" - - @pytest.fixture(autouse=True) - def _require_sealed(self, monkeypatch): - monkeypatch.setenv(REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_ENV_VAR, "true") - - def test_missing_settings_fail_as_platform_error(self): - downstream, sent = _run_middleware(json.dumps({"element_dicts": "/tmp/x.json"}).encode()) - - assert sent[0]["status"] == 500 - assert "SealedDagNodeSettingsRequiredError" in json.loads(sent[1]["body"])["detail"] - assert not downstream.called - - def test_plaintext_settings_fail_as_platform_error(self): - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": {"model": "m"}}).encode() - ) - - assert sent[0]["status"] == 500 - assert not downstream.called - - def test_sealed_settings_still_bind(self, key_dir, private_key): - settings = {"api_key": SENTINEL_SECRET} - composite = {DAG_NODE_SETTINGS_KEY: sealed_payload(private_key, settings)} - - downstream, sent = _run_middleware( - json.dumps({"invocation_settings": composite}).encode() - ) - - assert sent[0]["status"] == 200 - assert downstream.seen_settings == settings - - def test_bodyless_invoke_fails_as_platform_error(self): - # A bodyless invoke cannot carry the envelope; on a pod with no boot settings file it must - # not dispatch a handler with no settings source at all. - downstream, sent = _run_middleware(b"") - - assert sent[0]["status"] == 500 - assert not downstream.called - - def test_non_object_json_body_fails_as_platform_error(self): - downstream, sent = _run_middleware(json.dumps([{"element": 1}]).encode()) - - assert sent[0]["status"] == 500 - assert not downstream.called - - -def test_non_object_bodies_pass_through_when_sealed_settings_not_required(): - downstream, sent = _run_middleware(b"") - - assert sent[0]["status"] == 200 - assert downstream.seen_settings is None diff --git a/test/api/test_invocation_settings.py b/test/api/test_invocation_settings.py index 5523b23..2ec8410 100644 --- a/test/api/test_invocation_settings.py +++ b/test/api/test_invocation_settings.py @@ -1,5 +1,6 @@ """The wrapper-installed invocation-settings surface: /metadata and reserved-field binding.""" +import json from typing import Optional from fastapi.testclient import TestClient @@ -18,7 +19,7 @@ def _echo_settings(content: str) -> _Echo: return _Echo(content=content, settings=current_invocation_settings()) -def test_metadata_route_is_registered_with_default_capabilities(): +def test_metadata_route_is_registered_with_transport_capabilities(): client = TestClient(wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin")) resp = client.get("/metadata") @@ -73,9 +74,7 @@ def sync_echo(content: str) -> _Echo: client = TestClient(wrap_in_fastapi(func=sync_echo, plugin_id="mock_plugin")) - resp = client.post( - "/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}} - ) + resp = client.post("/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}}) assert resp.json()["output"]["settings"] == {"model": "m"} @@ -87,13 +86,24 @@ async def _async_echo(content: str) -> _Echo: def test_async_function_sees_bound_settings(): client = TestClient(wrap_in_fastapi(func=_async_echo, plugin_id="mock_plugin")) - resp = client.post( - "/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}} - ) + resp = client.post("/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}}) assert resp.json()["output"]["settings"] == {"model": "m"} +async def _stream_echo(content: str) -> _Echo: + yield _Echo(content=content, settings=current_invocation_settings()) + + +def test_async_generator_sees_bound_settings_during_stream_iteration(): + client = TestClient(wrap_in_fastapi(func=_stream_echo, plugin_id="mock_plugin")) + + resp = client.post("/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}}) + + line = json.loads(resp.text.strip()) + assert line["output"]["settings"] == {"model": "m"} + + def test_absent_reserved_fields_bind_none(): client = TestClient(wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin")) diff --git a/unstructured_platform_plugins/etl_uvicorn/api_generator.py b/unstructured_platform_plugins/etl_uvicorn/api_generator.py index 2426ecc..c0e242e 100644 --- a/unstructured_platform_plugins/etl_uvicorn/api_generator.py +++ b/unstructured_platform_plugins/etl_uvicorn/api_generator.py @@ -29,7 +29,10 @@ ) from unstructured_platform_plugins.invocation_settings import ( add_metadata_route, + current_invocation_context, + current_invocation_settings, install_invocation_envelope, + invocation_envelope, ) from unstructured_platform_plugins.schema import FileDataMeta, NewRecord, UsageData from unstructured_platform_plugins.schema.json_schema import ( @@ -154,6 +157,9 @@ def _wrap_in_fastapi( logger.debug(f"set static id response to: {plugin_id}") fastapi_app = FastAPI() + # Installation contributes a public router dependency, so it must happen before /invoke is + # registered. The dependency itself is a no-op for every other route. + install_invocation_envelope(fastapi_app) response_type = get_output_sig(func) filedata_meta_model = update_filedata_model(response_type) @@ -192,12 +198,33 @@ async def wrap_fn(func: Callable, kwargs: Optional[dict[str, Any]] = None) -> Re request_dict["message_channels"] = message_channels if "filedata_meta" in inspect.signature(func).parameters: request_dict["filedata_meta"] = filedata_meta + bound_settings = current_invocation_settings() + bound_context = current_invocation_context() try: if inspect.isasyncgenfunction(func): # Stream response if function is an async generator async def _stream_response(): - try: - async for output in func(**(request_dict or {})): + # FastAPI 0.117 closes yield dependencies before iterating a + # StreamingResponse. Re-enter the captured binding inside the generator so the + # plugin sees the right request regardless of dependency-cleanup timing. + with invocation_envelope(bound_settings, bound_context): + try: + async for output in func(**(request_dict or {})): + yield ( + InvokeResponse( + usage=usage, + message_channels=message_channels, + filedata_meta=filedata_meta_model.model_validate( + filedata_meta.model_dump() + ), + status_code=status.HTTP_200_OK, + output=output, + file_data=request_dict.get("file_data", None), + ).model_dump_json() + + "\n" + ) + except Exception as e: + logger.error(f"Failure streaming response: {e}", exc_info=True) yield ( InvokeResponse( usage=usage, @@ -205,31 +232,19 @@ async def _stream_response(): filedata_meta=filedata_meta_model.model_validate( filedata_meta.model_dump() ), - status_code=status.HTTP_200_OK, - output=output, - file_data=request_dict.get("file_data", None), + status_code=getattr(e, "status_code", None) + or status.HTTP_500_INTERNAL_SERVER_ERROR, + status_code_text=f"[{e.__class__.__name__}] {e}", ).model_dump_json() + "\n" ) - except Exception as e: - logger.error(f"Failure streaming response: {e}", exc_info=True) - yield ( - InvokeResponse( - usage=usage, - message_channels=message_channels, - filedata_meta=filedata_meta_model.model_validate( - filedata_meta.model_dump() - ), - status_code=getattr(e, "status_code", None) - or status.HTTP_500_INTERNAL_SERVER_ERROR, - status_code_text=f"[{e.__class__.__name__}] {e}", - ).model_dump_json() - + "\n" - ) return StreamingResponse(_stream_response(), media_type="application/x-ndjson") else: - output = await invoke_func(func=func, kwargs=request_dict) + # Keep execution-scoped binding explicit for the same reason as the streaming + # branch; nested binding is harmless while the route dependency is still active. + with invocation_envelope(bound_settings, bound_context): + output = await invoke_func(func=func, kwargs=request_dict) return InvokeResponse( usage=usage, message_channels=message_channels, @@ -311,9 +326,7 @@ async def run_job_with_body(request: BaseModel) -> ResponseType: @fastapi_app.post("/invoke", response_model=InvokeResponse) async def run_job(request: Optional[input_schema_model] = None) -> ResponseType: - return await run_job_with_body( - request if request is not None else input_schema_model() - ) + return await run_job_with_body(request if request is not None else input_schema_model()) elif input_schema_model.model_fields: @@ -371,18 +384,17 @@ async def get_id() -> str: except TypeError as e: raise TypeError(f"failed to validate function schema: {e}") from e - # The middleware handles the reserved /invoke fields (invocation_settings and + # The route dependency handles the reserved /invoke fields (invocation_settings and # invocation_context) outside the generated handler schema. It resolves sealed settings with - # the configured private key and exposes both values through request-scoped accessors. The - # sealed-settings capability remains opt-in because it asserts that the wrapped function - # consumes current_invocation_settings(), not merely that the host can resolve it. Repeated - # installation is safe: the middleware installs once and the last /metadata registration wins. + # the configured private key and exposes both values through request-scoped accessors. + # The sealed-settings capability remains opt-in because it asserts that the wrapped function + # consumes current_invocation_settings(), not merely that the host can resolve it. The binding + # dependency was installed before route registration; the last /metadata registration wins. add_metadata_route( fastapi_app, identifier=plugin_id, invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings, ) - install_invocation_envelope(fastapi_app) FastAPIInstrumentor.instrument_app( fastapi_app, tracer_provider=get_trace_provider(), meter_provider=get_metric_provider() diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index 7b73d43..c0e95e5 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/unstructured_platform_plugins/invocation_settings.py @@ -1,4 +1,4 @@ -"""Transport for the reserved `/invoke` fields: ASGI middleware, `/metadata`, request binding. +"""Transport for the reserved `/invoke` fields: request dependency, `/metadata`, body cap. The *settings contract* — which key carries settings, how a sealed envelope is told from plaintext, and what an absent field is allowed to mean — lives in @@ -9,10 +9,17 @@ off the wire and the results to the handler, and spelling the shared `blame` taxonomy as HTTP statuses. -The reserved fields are a first-class HTTP contract independent of the generated input schema. They -never appear in a plugin's declared signature, so `wrap_in_fastapi` keeps producing a handler model -built purely from the wrapped function, and a plugin reads the fields through +The reserved fields are a first-class HTTP contract independent of the generated input schema. +They never appear in a plugin's declared signature, so `wrap_in_fastapi` keeps producing a handler +model built purely from the wrapped function, and a plugin reads the fields through `current_invocation_settings()` / `current_invocation_context()` instead. + +Extraction runs in a route dependency (`bind_invocation_envelope`), so the `/invoke` body is +buffered and parsed exactly once: Starlette caches both the bytes and the parsed JSON on the +`Request`, and the dependency reads that cached parse. The request-size cap is the one concern +that must sit below the framework — neither Starlette nor uvicorn bounds request-body size — and +`InvokeBodyLimitMiddleware` enforces it by counting bytes as they stream through, without +buffering. """ from __future__ import annotations @@ -24,12 +31,14 @@ import threading import time from collections import OrderedDict -from collections.abc import Callable, Iterator, Mapping +from collections.abc import AsyncIterator, Callable, Iterator, Mapping from contextlib import contextmanager from contextvars import ContextVar from typing import Any, Optional, TypeVar -from fastapi import FastAPI +from fastapi import Depends, FastAPI, Request +from starlette.requests import ClientDisconnect +from starlette.responses import JSONResponse from starlette.types import ASGIApp, Receive, Scope, Send from utic_invocation_settings import ( INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_CAPABILITY, @@ -60,13 +69,14 @@ def http_status_for(error: BaseException) -> int: """ return 422 if getattr(error, "blame", None) is Blame.CALLER else 500 + T = TypeVar("T") _METADATA_PATH = "/metadata" _INVOKE_PATH = "/invoke" -# Bounds middleware body buffering; generous because batch invokes carry an array of file_data -# payloads. The framework buffers the same body afterward, so this cap is the only guard against +# Bounds the /invoke request body; generous because batch invokes carry an array of file_data +# payloads. Nothing below the framework buffers, so this cap is the only guard against # unbounded-memory requests. MAX_INVOKE_BODY_BYTES = 64 * 1024 * 1024 @@ -80,7 +90,7 @@ def current_invocation_settings() -> Optional[dict]: `None` means the field was genuinely absent — the only case in which a plugin may fall back to its boot-time settings. A field that arrived and could not be opened never reaches a handler: - the middleware fails the request first. + the binding dependency fails the request first. """ return _INVOCATION.get()[0] @@ -113,18 +123,14 @@ def add_metadata_route( flags are strings in its `capabilities` list, which is where the controller looks before forwarding the reserved fields — no controller-private probe route. - The two tiers make different claims. `invocation_settings` / `invocation_context` are - *transport-level* facts, advertised unconditionally because the installed middleware makes - them true for every wrapped app: the reserved fields will be received, resolved, and bound — - or the request failed. They say nothing about whether the handler reads the binding. - `invoke_with_sealed_dag_node_settings` is the *consumption* claim — this plugin opens sealed - `dag_node_settings` itself and its handler acts on the result — and stays a per-plugin opt-in - set in the same change that makes it true, because it is the flag that invites the controller - to seal settings to this pod in place of any other settings source. + `invocation_settings` and `invocation_context` are transport capabilities: installing the + dependency makes the host receive, resolve, and bind those fields. The sealed-settings + capability is stronger: it tells the controller that the plugin handler consumes the resolved + `dag_node_settings` in place of boot-time state, so it remains an explicit opt-in. Last call wins: the payload lives on `app.state` and every call overwrites it, while the route - is registered once. A host wrapper may register with default capabilities at app construction - and a plugin can still declare the sealed capability afterwards, with no route-order dependence. + is registered once. A host wrapper may register at app construction and a plugin can still + re-register with its own identifier afterwards, with no route-order dependence. """ capabilities = [RESERVED_ENVELOPE_KEY, RESERVED_CONTEXT_KEY] if invoke_with_sealed_dag_node_settings: @@ -147,6 +153,104 @@ async def plugin_metadata() -> dict: return app.state.plugin_metadata_payload +class UnusableInvocationEnvelope(Exception): + """A reserved /invoke field arrived but cannot be used. + + Raised by `bind_invocation_envelope` and answered by the handler + `install_invocation_envelope` registers, so the response shape — `detail` plus the library's + stable `reason` code as top-level siblings — stays what orchestrators parse, independent of + FastAPI's own error envelope. + """ + + def __init__(self, status_code: int, payload: dict): + super().__init__(payload.get("detail")) + self.status_code = status_code + self.payload = payload + + +async def _unusable_envelope_response( + _request: Request, exc: UnusableInvocationEnvelope +) -> JSONResponse: + return JSONResponse(status_code=exc.status_code, content=exc.payload) + + +async def bind_invocation_envelope(request: Request) -> AsyncIterator[None]: + """Resolve the reserved /invoke fields and bind them for the duration of the request. + + Runs as a route dependency, after the framework has read the body: `request.json()` is + Starlette-cached, so the parse is shared with the framework's own body handling. + A reserved field that is present but unusable fails the request rather than being + treated as absent, because absence is the signal to fall back to the boot-time settings file: + degrading a malformed field to absence would quietly answer a request configured for one + tenant with whatever the pod happened to boot with. Under + `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` there is no settings file to fall back to, + so absent or plaintext settings fail too — as does any /invoke whose body is not a JSON + object, since such a body cannot carry the envelope a native pod requires. + """ + # ASGI `path` is mount-relative and excludes a deployment root_path; request.url.path may + # include that prefix and would silently skip binding behind a rooted proxy deployment. + if request.method != "POST" or request.scope["path"] != _INVOKE_PATH: + yield + return + + try: + parsed = await request.json() + except ValueError: + # Empty or malformed body: the framework's own validation answers for the body itself; + # for the reserved fields it is absence, which resolve_invocation_settings still judges + # (absence is a failure on a native pod). + parsed = None + + raw_settings: Optional[Any] = None + if isinstance(parsed, dict): + raw_settings = parsed.get(RESERVED_ENVELOPE_KEY) + if RESERVED_ENVELOPE_KEY in parsed and not isinstance(raw_settings, dict): + raise UnusableInvocationEnvelope( + 422, + { + "detail": f"Invalid field: {RESERVED_ENVELOPE_KEY}", + "reason": MalformedEnvelopeError.reason, + }, + ) + try: + # 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. + invocation_settings = await asyncio.to_thread(resolve_invocation_settings, raw_settings) + except Exception as exc: + # 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__) + payload = {"detail": f"Unusable invocation settings: {type(exc).__name__}"} + reason = getattr(exc, "reason", None) + if isinstance(reason, str): + payload["reason"] = reason + raise UnusableInvocationEnvelope(http_status_for(exc), payload) from exc + + invocation_context: Optional[InvocationContext] = None + if isinstance(parsed, dict): + try: + invocation_context = extract_context(parsed) + except InvocationSettingsError as exc: + # A context this plugin cannot read fails loudly here rather than running with + # silently absent identity. Status comes from the blame taxonomy: a malformed + # field is the caller's 422, but an unreadable schema_version is deployment skew + # between platform components and must not read as a caller fault. The log line is + # truncated because the message can embed request-controlled values. + logger.warning("rejecting invalid %s: %.200s", RESERVED_CONTEXT_KEY, exc) + status = http_status_for(exc) + detail = ( + f"Invalid field: {RESERVED_CONTEXT_KEY}" + if status == 422 + else f"Unusable {RESERVED_CONTEXT_KEY}: {type(exc).__name__}" + ) + raise UnusableInvocationEnvelope( + status, {"detail": detail, "reason": exc.reason} + ) from exc + + with invocation_envelope(invocation_settings, invocation_context): + yield + + async def _send_json(send: Send, status_code: int, payload: dict) -> None: body = json.dumps(payload).encode() await send( @@ -162,21 +266,14 @@ async def _send_json(send: Send, status_code: int, payload: dict) -> None: await send({"type": "http.response.body", "body": body}) -class InvocationEnvelopeMiddleware: - """Extract the reserved envelope fields from the raw POST /invoke body. +class InvokeBodyLimitMiddleware: + """Reject a POST /invoke body over ``max_body_bytes`` with 413. - Pure ASGI rather than `BaseHTTPMiddleware`: the body has to be read before the framework parses - it and then replayed intact, which is exactly what the raw protocol allows and what a - request/response middleware would fight. - - A reserved field that is present but unusable fails the request rather than being treated as - absent, because absence is the signal to fall back to the boot-time settings file: degrading a - malformed field to absence would quietly answer a request configured for one tenant with - whatever the pod happened to boot with. Under `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` - there is no settings file to fall back to, so absent or plaintext settings fail too — as does - any /invoke whose body is not a JSON object, since such a body cannot carry the envelope a - native pod requires. A body over `max_body_bytes` is rejected with 413 before it can exhaust - memory. + Counts bytes as the framework consumes them; nothing is buffered here. When the count crosses + the cap the downstream read is answered with ``http.disconnect``, which aborts the framework's + body read before another byte is held, and the 413 is sent once the application has unwound. + This has to sit below the framework because neither Starlette nor uvicorn bounds request-body + size. """ def __init__(self, app: ASGIApp, max_body_bytes: int = MAX_INVOKE_BODY_BYTES): @@ -192,103 +289,72 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: await self.app(scope, receive, send) return - messages = [] - buffered_bytes = 0 - while True: + seen = 0 + exceeded = False + response_started = False + + async def counting_receive() -> dict: + nonlocal seen, exceeded message = await receive() - messages.append(message) - buffered_bytes += len(message.get("body", b"")) - if buffered_bytes > self.max_body_bytes: - await _send_json(send, 413, {"detail": "Request body too large"}) + if message["type"] == "http.request": + seen += len(message.get("body", b"")) + if seen > self.max_body_bytes: + exceeded = True + return {"type": "http.disconnect"} + return message + + async def guarded_send(message: dict) -> None: + nonlocal response_started + if exceeded and not response_started: + # A response computed after the body was cut is answering a truncated request; + # drop it so the 413 below is what the caller sees. A response that started + # before the cap tripped keeps streaming — its start is already on the wire. return - if message["type"] != "http.request" or not message.get("more_body"): - break - body = b"".join(m.get("body", b"") for m in messages if m["type"] == "http.request") + if message["type"] == "http.response.start": + response_started = True + await send(message) try: - parsed = json.loads(body) if body else None - except ValueError: - # Malformed JSON: forward unchanged so the framework returns its own error. - parsed = None - - invocation_context: Optional[InvocationContext] = None - raw_settings: Optional[dict[str, Any]] = 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}", - "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 - # policy, under which a bodyless or non-object invoke cannot carry the envelope a native - # pod requires and is a failure, not a fallback signal. - try: - # Off the event loop: a cold resolve is an RSA unwrap of a couple of milliseconds, and - # this middleware sits in front of every invoke on the pod. - invocation_settings = await asyncio.to_thread(resolve_invocation_settings, raw_settings) - except Exception as exc: - # 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__) - 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: - invocation_context = extract_context(parsed) - except InvocationSettingsError as exc: - # A context this plugin cannot read fails loudly here rather than running with - # silently absent identity. Status comes from the blame taxonomy: a malformed - # field is the caller's 422, but an unreadable schema_version is deployment skew - # between platform components and must not read as a caller fault. The log line is - # truncated because the message can embed request-controlled values. - logger.warning("rejecting invalid %s: %.200s", RESERVED_CONTEXT_KEY, exc) - status = http_status_for(exc) - detail = ( - f"Invalid field: {RESERVED_CONTEXT_KEY}" - if status == 422 - else f"Unusable {RESERVED_CONTEXT_KEY}: {type(exc).__name__}" - ) - 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; - # the framework re-buffers and re-parses the replayed messages downstream, so holding these - # through the handler would double peak memory. - del body, parsed, raw_settings - - async def replay() -> dict: - if messages: - return messages.pop(0) - # Buffer drained: proxy the original channel so downstream still observes - # http.disconnect. - return await receive() - - with invocation_envelope(invocation_settings, invocation_context): - await self.app(scope, replay, send) - - -def install_invocation_envelope(app: FastAPI) -> None: - """Install out-of-schema envelope extraction on a FastAPI app. - - Idempotent per app: a host wrapper may install at app construction while a plugin that predates - the wrapper's support still calls this itself, and a double install would buffer and replay the - request body twice. + await self.app(scope, counting_receive, guarded_send) + except ClientDisconnect: + if not exceeded: + raise + except Exception: + # The cut body stream can surface downstream as something other than + # ClientDisconnect; once the cap is the cause, the 413 below is the answer. + if not exceeded: + raise + if exceeded and not response_started: + await _send_json(send, 413, {"detail": "Request body too large"}) + + +def install_invocation_envelope(app: FastAPI, max_body_bytes: int = MAX_INVOKE_BODY_BYTES) -> None: + """Install reserved-field binding before registering a FastAPI app's routes. + + Adds `bind_invocation_envelope` through the router's public dependency list, installs the + body-size cap beneath the framework, and registers the failure response shape. The dependency + is a path/method-aware no-op outside POST /invoke, so routes registered after this call can all + inherit it without private dependency-graph mutation. Calling after routes have already been + registered raises rather than silently leaving those routes uncovered. + + Idempotent per app: a host wrapper may install at app construction while a plugin that + predates the wrapper's support still calls this itself, and a double install would resolve + settings twice per request. """ if getattr(app.state, "invocation_envelope_installed", False): return + if any( + getattr(route, "path", None) == _INVOKE_PATH + and "POST" in (getattr(route, "methods", None) or set()) + for route in app.router.routes + ): + raise RuntimeError( + "install_invocation_envelope must be called before the POST /invoke route is registered" + ) app.state.invocation_envelope_installed = True - app.add_middleware(InvocationEnvelopeMiddleware) + app.router.dependencies.append(Depends(bind_invocation_envelope)) + app.add_middleware(InvokeBodyLimitMiddleware, max_body_bytes=max_body_bytes) + app.add_exception_handler(UnusableInvocationEnvelope, _unusable_envelope_response) def settings_cache_key(invocation_settings: Mapping[str, Any]) -> str: From 69b6339e4c36939c1c5939491cc27302e3b08348 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:58:13 -0400 Subject: [PATCH 09/12] fix(etl-uvicorn): declare user blame on the streaming error path The non-streaming failure envelope derives blame from the UserError family; the async-generator error envelope omitted it, so a customer fault raised mid-stream read as unattributed. --- test/api/test_api.py | 24 +++++++++++++++++++ test/assets/exception_status_code.py | 13 ++++++++++ .../etl_uvicorn/api_generator.py | 1 + 3 files changed, 38 insertions(+) diff --git a/test/api/test_api.py b/test/api/test_api.py index f559333..79fd0f2 100644 --- a/test/api/test_api.py +++ b/test/api/test_api.py @@ -253,6 +253,30 @@ def test_non_user_failures_declare_no_blame(file_data): 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] ) diff --git a/test/assets/exception_status_code.py b/test/assets/exception_status_code.py index f20cf93..034f3af 100644 --- a/test/assets/exception_status_code.py +++ b/test/assets/exception_status_code.py @@ -1,6 +1,7 @@ """Test assets for testing exception handling with various status_code scenarios.""" from fastapi import HTTPException +from typing_extensions import TypedDict from unstructured_ingest.error import UnstructuredIngestError @@ -139,6 +140,18 @@ async def async_gen_function_raises_unstructured_ingest_error_with_none_status_c raise error +class PartialStreamResponse(TypedDict): + partial: str + + +async def async_gen_function_raises_user_error_mid_stream() -> PartialStreamResponse: + """Async generator that yields once, then raises UserError.""" + from unstructured_ingest.error import UserError + + yield PartialStreamResponse(partial="output") + raise UserError("Customer-owned resource rejected the request") + + def function_raises_user_error() -> None: from unstructured_ingest.error import UserError diff --git a/unstructured_platform_plugins/etl_uvicorn/api_generator.py b/unstructured_platform_plugins/etl_uvicorn/api_generator.py index c0e242e..6a4ad18 100644 --- a/unstructured_platform_plugins/etl_uvicorn/api_generator.py +++ b/unstructured_platform_plugins/etl_uvicorn/api_generator.py @@ -235,6 +235,7 @@ async def _stream_response(): status_code=getattr(e, "status_code", None) or status.HTTP_500_INTERNAL_SERVER_ERROR, status_code_text=f"[{e.__class__.__name__}] {e}", + blame="user" if isinstance(e, UserError) else None, ).model_dump_json() + "\n" ) From cba14014e0bd9eb55e6cd4fe2e5f3d9c2c2ca40f Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:58:38 -0400 Subject: [PATCH 10/12] fix(etl-uvicorn): match /invoke by the route path under a rooted deployment Per the ASGI spec, scope["path"] includes any deployment root_path; the router matches on get_route_path, which strips it. Comparing the raw path to /invoke made both envelope binding and the body cap silently skip on a spec-compliant rooted server, so settings read as absent and the boot-file fallback took over. TestClient(root_path=...) does not prefix path, so the regression tests build the spec-compliant scope themselves. --- test/api/test_invocation_envelope.py | 55 +++++++++++++++++++ .../invocation_settings.py | 9 +-- 2 files changed, 60 insertions(+), 4 deletions(-) diff --git a/test/api/test_invocation_envelope.py b/test/api/test_invocation_envelope.py index 5c243a5..0dd3231 100644 --- a/test/api/test_invocation_envelope.py +++ b/test/api/test_invocation_envelope.py @@ -183,6 +183,21 @@ async def schema() -> dict: return app +class _SpecCompliantRootShim: + """A rooted deployment as the ASGI spec describes it: ``path`` carries the mount prefix and + ``root_path`` names it. ``TestClient(root_path=...)`` sets only ``root_path`` without + prefixing ``path``, so it cannot produce this shape.""" + + def __init__(self, app, root: str): + self.app = app + self.root = root + + async def __call__(self, scope, receive, send): + if scope["type"] == "http": + scope = {**scope, "path": self.root + scope["path"], "root_path": self.root} + await self.app(scope, receive, send) + + def _post_invoke(payload, recorder: Optional[_Recorder] = None, **app_kwargs): recorder = recorder if recorder is not None else _Recorder() app = _envelope_app(recorder, **app_kwargs) @@ -317,6 +332,16 @@ def test_root_path_does_not_prevent_invoke_binding(self): assert response.status_code == 200 assert recorder.seen_settings == {"model": "rooted"} + def test_spec_compliant_root_path_scope_still_binds(self): + recorder = _Recorder() + app = _envelope_app(recorder) + + with TestClient(_SpecCompliantRootShim(app, "/plugins/chunker")) as client: + response = client.post("/invoke", json={"invocation_settings": {"model": "rooted"}}) + + assert response.status_code == 200 + assert recorder.seen_settings == {"model": "rooted"} + def test_oversized_body_is_rejected(self): recorder, response = _post_invoke( {"invocation_settings": {"pad": "x" * 64}}, max_body_bytes=16 @@ -325,6 +350,18 @@ def test_oversized_body_is_rejected(self): assert not recorder.called assert response.status_code == 413 + def test_spec_compliant_root_path_scope_still_caps_body(self): + recorder = _Recorder() + app = _envelope_app(recorder, max_body_bytes=16) + + with TestClient( + _SpecCompliantRootShim(app, "/plugins/chunker"), raise_server_exceptions=False + ) as client: + response = client.post("/invoke", json={"invocation_settings": {"pad": "x" * 64}}) + + assert not recorder.called + assert response.status_code == 413 + class TestInvokeBodyLimitMiddleware: """ASGI-level behavior of the streaming byte counter: no buffering, disconnect pass-through.""" @@ -367,6 +404,24 @@ def test_chunked_body_over_the_cap_answers_413_and_cuts_downstream(self): assert received[-1] == {"type": "http.disconnect"} assert sent[0]["status"] == 413 + def test_rooted_scope_over_the_cap_answers_413(self): + # Per the ASGI spec, `path` includes the deployment root_path; the cap must key on the + # mount-relative path, the same one the router matches. + chunks = [{"type": "http.request", "body": b"x" * 20, "more_body": False}] + received, sent = self._run( + lambda app: InvokeBodyLimitMiddleware(app, max_body_bytes=16), + { + "type": "http", + "method": "POST", + "path": "/plugins/chunker/invoke", + "root_path": "/plugins/chunker", + }, + chunks, + ) + + assert received[-1] == {"type": "http.disconnect"} + assert sent[0]["status"] == 413 + def test_client_disconnect_passes_through_uncounted(self): chunks = [{"type": "http.disconnect"}] received, sent = self._run( diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index c0e95e5..cd5a412 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/unstructured_platform_plugins/invocation_settings.py @@ -39,6 +39,7 @@ from fastapi import Depends, FastAPI, Request from starlette.requests import ClientDisconnect from starlette.responses import JSONResponse +from starlette.routing import get_route_path from starlette.types import ASGIApp, Receive, Scope, Send from utic_invocation_settings import ( INVOKE_WITH_SEALED_DAG_NODE_SETTINGS_CAPABILITY, @@ -187,9 +188,9 @@ async def bind_invocation_envelope(request: Request) -> AsyncIterator[None]: so absent or plaintext settings fail too — as does any /invoke whose body is not a JSON object, since such a body cannot carry the envelope a native pod requires. """ - # ASGI `path` is mount-relative and excludes a deployment root_path; request.url.path may - # include that prefix and would silently skip binding behind a rooted proxy deployment. - if request.method != "POST" or request.scope["path"] != _INVOKE_PATH: + # ASGI `path` includes any deployment root_path; get_route_path strips it, which is how the + # router itself matches, so binding fires exactly when the /invoke route does. + if request.method != "POST" or get_route_path(request.scope) != _INVOKE_PATH: yield return @@ -284,7 +285,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if ( scope["type"] != "http" or scope.get("method") != "POST" - or scope.get("path") != _INVOKE_PATH + or get_route_path(scope) != _INVOKE_PATH ): await self.app(scope, receive, send) return From dce3991b8a6cf81c36d7a864a05925ad32f2391d Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:59:22 -0400 Subject: [PATCH 11/12] fix(etl-uvicorn): reject a repeat envelope install with a different body cap The idempotence guard returned before reading max_body_bytes, so a second install asking for a different cap was silently ignored and the caller was left believing its limit was enforced. A same-value repeat stays a no-op. --- test/api/test_invocation_envelope.py | 18 ++++++++++++++++++ .../invocation_settings.py | 11 ++++++++++- 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/test/api/test_invocation_envelope.py b/test/api/test_invocation_envelope.py index 0dd3231..a2c880a 100644 --- a/test/api/test_invocation_envelope.py +++ b/test/api/test_invocation_envelope.py @@ -362,6 +362,24 @@ def test_spec_compliant_root_path_scope_still_caps_body(self): assert not recorder.called assert response.status_code == 413 + def test_repeat_install_with_the_same_cap_is_a_noop(self): + recorder = _Recorder() + app = _envelope_app(recorder, max_body_bytes=1024) + install_invocation_envelope(app, max_body_bytes=1024) + + with TestClient(app) as client: + response = client.post("/invoke", json={"element_dicts": "/in.json"}) + + assert response.status_code == 200 + assert recorder.called + + def test_repeat_install_with_a_different_cap_fails_loudly(self): + app = FastAPI() + install_invocation_envelope(app, max_body_bytes=16) + + with pytest.raises(ValueError, match="max_body_bytes"): + install_invocation_envelope(app, max_body_bytes=32) + class TestInvokeBodyLimitMiddleware: """ASGI-level behavior of the streaming byte counter: no buffering, disconnect pass-through.""" diff --git a/unstructured_platform_plugins/invocation_settings.py b/unstructured_platform_plugins/invocation_settings.py index cd5a412..680415f 100644 --- a/unstructured_platform_plugins/invocation_settings.py +++ b/unstructured_platform_plugins/invocation_settings.py @@ -340,9 +340,17 @@ def install_invocation_envelope(app: FastAPI, max_body_bytes: int = MAX_INVOKE_B Idempotent per app: a host wrapper may install at app construction while a plugin that predates the wrapper's support still calls this itself, and a double install would resolve - settings twice per request. + settings twice per request. A repeated call asking for a different ``max_body_bytes`` raises, + because the cap already installed cannot be changed and silently keeping the first value would + misrepresent the limit actually enforced. """ if getattr(app.state, "invocation_envelope_installed", False): + installed_max = app.state.invocation_envelope_max_body_bytes + if max_body_bytes != installed_max: + raise ValueError( + "install_invocation_envelope already installed with " + f"max_body_bytes={installed_max}; cannot reinstall with {max_body_bytes}" + ) return if any( getattr(route, "path", None) == _INVOKE_PATH @@ -353,6 +361,7 @@ def install_invocation_envelope(app: FastAPI, max_body_bytes: int = MAX_INVOKE_B "install_invocation_envelope must be called before the POST /invoke route is registered" ) app.state.invocation_envelope_installed = True + app.state.invocation_envelope_max_body_bytes = max_body_bytes app.router.dependencies.append(Depends(bind_invocation_envelope)) app.add_middleware(InvokeBodyLimitMiddleware, max_body_bytes=max_body_bytes) app.add_exception_handler(UnusableInvocationEnvelope, _unusable_envelope_response) From 6bbbe433e9f189dece2ebc155a222395b03c8624 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:59:22 -0400 Subject: [PATCH 12/12] chore(etl-uvicorn): describe context extraction as a route dependency The envelope has been extracted by a route dependency, not middleware, since the body-replay design was removed. --- unstructured_platform_plugins/invocation_context.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/unstructured_platform_plugins/invocation_context.py b/unstructured_platform_plugins/invocation_context.py index 4cb898f..2c456a6 100644 --- a/unstructured_platform_plugins/invocation_context.py +++ b/unstructured_platform_plugins/invocation_context.py @@ -3,7 +3,7 @@ Where ``invocation_settings`` carries *what* a plugin should be configured with, the context carries *who* the invocation is for: the identity facets a shared-tenancy pod can no longer read from its process environment. It travels in a second reserved, out-of-schema field of the -``/invoke`` body, extracted by the same middleware that resolves the settings field. +``/invoke`` body, extracted by the same route dependency that resolves the settings field. The context is `/invoke` protocol identity, not settings security: it touches no crypto and no secrets, and it evolves with the plugin protocol this package defines. The errors it raises come