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: diff --git a/CHANGELOG.md b/CHANGELOG.md index e3a14bd..a6d828e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,61 @@ +## 0.1.0 + +* **This package now owns the `/invoke` transport for the reserved fields.** + `unstructured_platform_plugins.invocation_settings` holds the `/invoke` binding dependency and + body cap, the `/metadata` capability route, the request-scoped accessors, and `http_status_for` + — the HTTP spelling of the + library's normative `blame` → status rule. It sits on `utic-invocation-settings >=0.4.0`, + which owns the *settings contract* — which key carries settings, how a sealed envelope is told + from plaintext, and what an absent field is allowed to mean. That split is deliberate: the + absence rule is a security decision and belongs next to the crypto it governs, while request + handling and route registration belong here, where a web framework is already a dependency. + Nothing about the sealed-settings wire format is decided in this repository. +* **This package now owns the `invocation_context` identity model.** + `unstructured_platform_plugins.invocation_context` holds `InvocationContext`, + `extract_context`, `dimensions`, `RESERVED_CONTEXT_KEY`, `DIMENSION_FIELDS`, + `SUPPORTED_CONTEXT_VERSIONS` and `UnsupportedContextVersionError`. The context is `/invoke` + protocol identity — no crypto, no secrets — so it lives with the plugin protocol. Its errors + subclass the shared `InvocationSettingsError` taxonomy, so hosts classify context failures with + the same `reason`/`blame` machinery as settings failures. This module is the public home for + the surface `utic-invocation-settings 0.2.x` carried and its `0.3.0` removed. +* **Every wrapped app installs it at construction.** The reserved `invocation_settings` / + `invocation_context` fields are handled outside the generated handler schema, a sealed + `dag_node_settings` member is opened with this pod's mounted workload key, and the resolved + values are exposed through `current_invocation_settings()` / `current_invocation_context()`. + An absent field preserves the existing fallback behaviour; under + `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` missing or plaintext settings fail closed. + Repeated installation is safe: the dependency installs once and the last `/metadata` + registration wins. +* **Extraction is a route dependency.** `bind_invocation_envelope` reads the body the framework + buffered and parsed (`request.json()` is Starlette-cached), so the `/invoke` body is held and + decoded exactly once per request. `install_invocation_envelope` contributes that path-aware + dependency through the router's public dependency list before `/invoke` is registered; no + private FastAPI dependency graph is mutated. It also registers the failure response shape and + installs `InvokeBodyLimitMiddleware`, a streaming byte counter that answers 413 over the cap + without buffering. Async-generator plugins explicitly re-enter the captured request binding + inside response iteration, so streaming stays correct independently of FastAPI's yield-dependency + cleanup timing. +* **Sealed settings consumption remains opt-in.** Pass + `invoke_with_sealed_dag_node_settings=True` to `wrap_in_fastapi` / `generate_fast_api` (or + `--sealed-dag-node-settings` on the CLI) only for a plugin that consumes per-invoke settings; + it advertises that the application accepts and acts on sealed `dag_node_settings`. Transport + support alone continues to advertise only `invocation_settings` and `invocation_context`. A + plugin that serves a custom `/metadata` payload must register it via `add_metadata_route` (which + replaces the wrapper's route) — a plain `@app.get("/metadata")` added after construction is + shadowed by the wrapper's earlier registration. +* **Resolution runs off the event loop.** A cold resolve is an RSA unwrap of a couple of + milliseconds and this dependency fronts every invoke on the pod, so it is dispatched with + `asyncio.to_thread` rather than blocking the loop. +* **Failures map through the library's blame taxonomy**, not a flat 500: only a caller-fixable + fault answers 422. Sealing drift, an envelope addressed to another recipient and a broken local + mount are all 5xx, which keeps the controller's blame classification off the customer. Responses + carry the error's class name and never its message, which can embed request-controlled values. +* **Sync plugin functions now observe request-scoped context.** `invoke_func` copies the current + context into the executor thread; previously `run_in_executor` dropped contextvars, so a sync + function reading a request-scoped binding (such as `current_invocation_settings()`) would see + it as absent and could take an unintended fallback path. +* **Python floor is now 3.11** (required by `utic-invocation-settings`). + ## 0.0.45 * **`/invoke` no longer demands a body from a plugin whose parameters are all optional.** A pydantic diff --git a/pyproject.toml b/pyproject.toml index 4bb84aa..4950598 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.4.0,<1.0.0", "opentelemetry-instrumentation-fastapi", "opentelemetry-exporter-otlp-proto-grpc", "dataclasses-json" diff --git a/test/api/test_api.py b/test/api/test_api.py index 83e2b23..79fd0f2 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,62 @@ def test_http_exception_handling(file_data): assert invoke_response.status_code_text == "Not found" +@pytest.mark.parametrize( + "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data] +) +def test_user_error_declares_user_blame(file_data): + """Only the UserError family may claim the failure is the customer's to fix.""" + from test.assets.exception_status_code import function_raises_user_error as test_fn + + client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin")) + + resp = client.post("/invoke", json={"file_data": file_data.model_dump()}) + invoke_response = InvokeResponse.model_validate(resp.json()) + + assert invoke_response.status_code >= 400 + assert invoke_response.blame == "user" + + +@pytest.mark.parametrize( + "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data] +) +def test_non_user_failures_declare_no_blame(file_data): + """Anything undeclared is not the customer's: an orchestrator must not infer customer fault + from the status code, which also carries transport semantics.""" + from test.assets.exception_status_code import function_raises_provider_error as test_fn + + client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin")) + + resp = client.post("/invoke", json={"file_data": file_data.model_dump()}) + invoke_response = InvokeResponse.model_validate(resp.json()) + + assert invoke_response.blame is None + + +def test_streaming_user_error_declares_user_blame(): + """The streaming error envelope carries the same blame derivation as the non-streaming path.""" + from test.assets.exception_status_code import ( + async_gen_function_raises_user_error_mid_stream as test_fn, + ) + + client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin")) + + resp = client.post("/invoke", json={"file_data": mock_file_data[0].model_dump()}) + + assert resp.status_code == 200 + assert resp.headers["content-type"] == "application/x-ndjson" + + import json + + lines = resp.content.decode().strip().split("\n") + assert len(lines) == 2 # One yielded item, then the error envelope + + assert InvokeResponse.model_validate(json.loads(lines[0])).blame is None + error_response = InvokeResponse.model_validate(json.loads(lines[1])) + assert error_response.status_code >= 400 + assert error_response.blame == "user" + + @pytest.mark.parametrize( "file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data] ) 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/test/api/test_invocation_envelope.py b/test/api/test_invocation_envelope.py new file mode 100644 index 0000000..a2c880a --- /dev/null +++ b/test/api/test_invocation_envelope.py @@ -0,0 +1,578 @@ +"""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 + + +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) + 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_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 + ) + + 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 + + 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.""" + + @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_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( + 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_settings.py b/test/api/test_invocation_settings.py new file mode 100644 index 0000000..2ec8410 --- /dev/null +++ b/test/api/test_invocation_settings.py @@ -0,0 +1,113 @@ +"""The wrapper-installed invocation-settings surface: /metadata and reserved-field binding.""" + +import json +from typing import Optional + +from fastapi.testclient import TestClient +from pydantic import BaseModel + +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): + 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_transport_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"} + + +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")) + + resp = client.post("/invoke", json={"content": "hello"}) + + assert resp.status_code == 200 + assert resp.json()["output"] == {"content": "hello", "settings": None} 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/test/assets/exception_status_code.py b/test/assets/exception_status_code.py index 6e35997..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 @@ -137,3 +138,27 @@ 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 + + +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 + + 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/__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..6a4ad18 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 @@ -13,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 @@ -26,6 +27,13 @@ get_schema_dict, map_inputs, ) +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 ( schema_to_base_model, @@ -67,7 +75,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 +131,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 +149,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) @@ -136,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) @@ -146,6 +170,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) @@ -169,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, @@ -182,31 +232,20 @@ 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}", + blame="user" if isinstance(e, UserError) else None, ).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, @@ -240,6 +279,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: @@ -287,9 +327,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: @@ -347,6 +385,18 @@ async def get_id() -> str: except TypeError as e: raise TypeError(f"failed to validate function schema: {e}") from e + # 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. 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, + ) + FastAPIInstrumentor.instrument_app( fastapi_app, tracer_provider=get_trace_provider(), meter_provider=get_metric_provider() ) @@ -361,6 +411,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 +430,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 diff --git a/unstructured_platform_plugins/invocation_context.py b/unstructured_platform_plugins/invocation_context.py new file mode 100644 index 0000000..2c456a6 --- /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 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 +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 new file mode 100644 index 0000000..680415f --- /dev/null +++ b/unstructured_platform_plugins/invocation_settings.py @@ -0,0 +1,428 @@ +"""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 +`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 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 + +import asyncio +import hashlib +import json +import logging +import threading +import time +from collections import OrderedDict +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 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, + RESERVED_ENVELOPE_KEY, + Blame, + InvocationSettingsError, + MalformedEnvelopeError, + 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" +_INVOKE_PATH = "/invoke" + +# 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 + +_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 binding dependency 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. + + `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 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: + 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 + + +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` 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 + + 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( + { + "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 InvokeBodyLimitMiddleware: + """Reject a POST /invoke body over ``max_body_bytes`` with 413. + + 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): + 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 get_route_path(scope) != _INVOKE_PATH + ): + await self.app(scope, receive, send) + return + + seen = 0 + exceeded = False + response_started = False + + async def counting_receive() -> dict: + nonlocal seen, exceeded + message = await receive() + 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.response.start": + response_started = True + await send(message) + + try: + 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. 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 + 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.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) + + +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()