feat: reserved invocation_settings envelope field on generated /invoke

Every wrap_in_fastapi-generated /invoke now accepts an optional
invocation_settings object as a reserved envelope field. It is kept out
of the function's /schema (no namespace collision with data inputs) and
delivered either as a kwarg when the wrapped function declares an
invocation_settings parameter, or via the get_invocation_settings()
ContextVar accessor otherwise (context propagated across the executor
thread). The ContextVar is set even on the kwarg path so helper code
deep in a plugin's call stack can always use the accessor.

GET /capabilities advertises support so callers (e.g. the ETL
controller) can gate injection without schema sniffing. The signature
check for the kwarg path is hoisted to wrap time; requests pay a
boolean, not reflection.

This lets one standing plugin process serve work items with differing
node settings (warm pools, worker fleets, per-invoke reconfiguration)
instead of freezing settings from a boot-time file.

75 tests passed (5 new in test/test_invocation_settings.py).

Author: CyMule

test/test_invocation_settings.py

@@ -0,0 +1,63 @@
+from typing import Any, Optional
+
+from fastapi.testclient import TestClient
+
+from unstructured_platform_plugins.etl_uvicorn.api_generator import wrap_in_fastapi
+from unstructured_platform_plugins.etl_uvicorn.invocation_context import (
+    get_invocation_settings,
+)
+from pydantic import BaseModel
+
+
+class Out(BaseModel):
+    content: str
+    settings_seen: Optional[dict[str, Any]] = None
+
+
+def fn_declares(content: str, invocation_settings: Optional[dict[str, Any]] = None) -> Out:
+    return Out(content=content, settings_seen=invocation_settings)
+
+
+def fn_uses_context(content: str) -> Out:
+    return Out(content=content, settings_seen=get_invocation_settings())
+
+
+def test_declared_param_receives_settings():
+    client = TestClient(wrap_in_fastapi(func=fn_declares, plugin_id="t1"))
+    resp = client.post(
+        "/invoke",
+        json={"content": "x", "invocation_settings": {"strategy": "fast"}},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["output"]["settings_seen"] == {"strategy": "fast"}
+
+
+def test_context_accessor_receives_settings():
+    client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t2"))
+    resp = client.post(
+        "/invoke",
+        json={"content": "x", "invocation_settings": {"k": 1}},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["output"]["settings_seen"] == {"k": 1}
+
+
+def test_omitted_settings_is_none_and_context_resets():
+    client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t3"))
+    with_settings = client.post("/invoke", json={"content": "x", "invocation_settings": {"k": 1}})
+    without = client.post("/invoke", json={"content": "x"})
+    assert with_settings.json()["output"]["settings_seen"] == {"k": 1}
+    assert without.json()["output"]["settings_seen"] is None
+
+
+def test_capabilities_endpoint():
+    client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t4"))
+    resp = client.get("/capabilities")
+    assert resp.status_code == 200
+    assert resp.json()["invocation_settings"] is True
+
+
+def test_schema_unpolluted_by_envelope_field():
+    client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t5"))
+    schema = client.get("/schema").json()
+    assert "invocation_settings" not in schema["inputs"].get("properties", {})

unstructured_platform_plugins/etl_uvicorn/api_generator.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 import hashlib
 import inspect
 import json
@@ -18,6 +19,9 @@
 from uvicorn.importer import import_from_string
 
 from unstructured_platform_plugins.etl_uvicorn.otel import get_metric_provider, get_trace_provider
+from unstructured_platform_plugins.etl_uvicorn.invocation_context import (
+    _invocation_settings_var,
+)
 from unstructured_platform_plugins.etl_uvicorn.utils import (
     get_func,
     get_input_schema,
@@ -67,7 +71,12 @@ 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))
+        # copy_context() so ContextVars (e.g. invocation_settings) reach the
+        # worker thread; run_in_executor does not propagate them by itself.
+        ctx = contextvars.copy_context()
+        return await asyncio.get_event_loop().run_in_executor(
+            None, partial(ctx.run, partial(func, **kwargs))
+        )
 
 
 def check_precheck_func(precheck_func: Callable):
@@ -149,8 +158,22 @@ class InvokeResponse(BaseModel):
         output: Optional[response_type] = None
         message_channels: MessageChannels = Field(default_factory=MessageChannels)
 
-    input_schema = get_input_schema(func, omit=["usage", "filedata_meta", "message_channels"])
+    input_schema = get_input_schema(
+        func, omit=["usage", "filedata_meta", "message_channels", "invocation_settings"]
+    )
     input_schema_model = schema_to_base_model(input_schema)
+    # First-class per-invoke settings: every generated /invoke accepts an optional
+    # invocation_settings object, independent of the wrapped function's signature.
+    # Delivery: passed as a kwarg when the function declares the parameter,
+    # otherwise exposed through get_invocation_settings() for the call's duration.
+    input_schema_model = create_model(
+        "InvokeEnvelope",
+        __base__=input_schema_model,
+        invocation_settings=(Optional[dict[str, Any]], None),
+    )
+    func_declares_invocation_settings = (
+        "invocation_settings" in inspect.signature(func).parameters
+    )
 
     logging.getLogger("etl_uvicorn.fastapi")
 
@@ -261,6 +284,10 @@ async def run_job(request: input_schema_model) -> ResponseType:
             log_func_and_body(func=func, body=request.json())
             # Create dictionary from pydantic model while preserving underlying types
             request_dict = {f: getattr(request, f) for f in request.model_fields}
+            invocation_settings = request_dict.pop("invocation_settings", None)
+            if func_declares_invocation_settings:
+                request_dict["invocation_settings"] = invocation_settings
+            settings_token = _invocation_settings_var.set(invocation_settings)
             # Make sure nested classes get instantiated correctly
             if "file_data" in request_dict:
                 request_dict["file_data"] = file_data_from_dict(
@@ -269,7 +296,10 @@ async def run_job(request: input_schema_model) -> ResponseType:
             map_inputs(func=func, raw_inputs=request_dict)
             if logger.level == LOG_LEVELS.get("trace", logging.NOTSET):
                 logger.log(level=logger.level, msg=f"passing inputs to function: {request_dict}")
-            return await wrap_fn(func=func, kwargs=request_dict)
+            try:
+                return await wrap_fn(func=func, kwargs=request_dict)
+            finally:
+                _invocation_settings_var.reset(settings_token)
 
     else:
 
@@ -315,6 +345,10 @@ async def run_precheck() -> InvokePrecheckResponse:
     async def get_id() -> str:
         return plugin_id
 
+    @fastapi_app.get("/capabilities")
+    async def get_capabilities() -> dict[str, Any]:
+        return {"invocation_settings": True}
+
     # Run initial schema validation
     try:
         asyncio.run(get_schema())

unstructured_platform_plugins/etl_uvicorn/invocation_context.py

@@ -0,0 +1,20 @@
+"""Per-invocation context for first-class /invoke envelope fields.
+
+``invocation_settings`` is a reserved, always-accepted field of every generated
+``/invoke`` request. Functions that declare an ``invocation_settings`` parameter
+receive it as a kwarg; all other code (including plugins with hand-rolled apps
+that adopt the same contract) can read it for the duration of the call via
+:func:`get_invocation_settings`.
+"""
+
+from contextvars import ContextVar
+from typing import Any, Optional
+
+_invocation_settings_var: ContextVar[Optional[dict[str, Any]]] = ContextVar(
+    "invocation_settings", default=None
+)
+
+
+def get_invocation_settings() -> Optional[dict[str, Any]]:
+    """The invocation_settings of the in-flight /invoke call, if any."""
+    return _invocation_settings_var.get()