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.

Author: CyMule

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

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

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
+    }

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"