From 846188113e2b73bd118f206b86bb6544a6c54d86 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Wed, 14 Oct 2026 11:27:12 -0400 Subject: [PATCH 1/6] feat: Add invocation identity and scratch helpers --- CHANGELOG.md | 3 + test/api/test_invocation_scope.py | 124 ++++++++++++++++++ unstructured_platform_plugins/__init__.py | 13 ++ .../invocation_scope.py | 60 +++++++++ 4 files changed, 200 insertions(+) create mode 100644 test/api/test_invocation_scope.py create mode 100644 unstructured_platform_plugins/invocation_scope.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e2e8dd..ebbe347 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ ## 0.1.1 +* Add `InvocationIdentity`, `current_invocation_identity()` and `invocation_scratch_dir()` for + request-scoped identity and scratch paths, with environment and root-path fallback in boot mode. + * **A provider rate limit no longer reaches the wire as terminal.** `plugin_error_of` declared `retryable = false` for the whole legacy `UserError` family, `RateLimitError` included, so a 429 a consumer reads off the declared field fails the record on the first throttle instead of backing diff --git a/test/api/test_invocation_scope.py b/test/api/test_invocation_scope.py new file mode 100644 index 0000000..291c0f4 --- /dev/null +++ b/test/api/test_invocation_scope.py @@ -0,0 +1,124 @@ +import asyncio +from dataclasses import FrozenInstanceError, dataclass, fields +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from unstructured_platform_plugins import ( + InvocationIdentity, + current_invocation_identity, + invocation_scratch_dir, +) +from unstructured_platform_plugins.etl_uvicorn.api_generator import generate_fast_api +from unstructured_platform_plugins.invocation_context import InvocationContext +from unstructured_platform_plugins.invocation_settings import invocation_envelope + + +@pytest.mark.parametrize("bound", [False, True]) +def test_identity_field_fallback(monkeypatch, bound): + for field in fields(InvocationIdentity): + monkeypatch.setenv(field.name.upper(), f"env-{field.name}") + context = InvocationContext(job_id="request", tenant_id="") if bound else None + with invocation_envelope(None, context): + identity = current_invocation_identity() + assert identity.job_id == ("request" if bound else "env-job_id") + assert identity.tenant_id == ("" if bound else "env-tenant_id") + assert identity.org_id == "env-org_id" + with pytest.raises(FrozenInstanceError): + identity.job_id = "other" + + +def test_absent_identity(monkeypatch): + for field in fields(InvocationIdentity): + monkeypatch.delenv(field.name.upper(), raising=False) + assert all( + getattr(current_invocation_identity(), f.name) is None for f in fields(InvocationIdentity) + ) + + +def test_boot_path_and_creation(tmp_path): + root = tmp_path / "scratch" + assert invocation_scratch_dir(root, create=False) == root + assert not root.exists() + assert invocation_scratch_dir(str(root)) == root + assert root.is_dir() + + +@pytest.mark.parametrize("segment", ["../a", "..", ".", "a/b", "a\\b", "x" * 300, "☃"]) +def test_segments_are_safe_and_bounded(tmp_path, segment): + context = InvocationContext(job_id=segment, dag_node_id=segment, invocation_id=segment) + with invocation_envelope(None, context): + path = invocation_scratch_dir(tmp_path) + assert path.is_relative_to(tmp_path) + assert len(path.relative_to(tmp_path).parts) == 3 + assert all(p not in {".", ".."} and len(p) <= 96 for p in path.relative_to(tmp_path).parts) + assert path.is_dir() + + +def test_sanitization_does_not_alias(tmp_path): + paths = [] + for job_id in ["a/b", "a_b"]: + with invocation_envelope(None, InvocationContext(job_id=job_id, invocation_id="invoke")): + paths.append(invocation_scratch_dir(tmp_path, create=False)) + assert paths[0] != paths[1] + + +def test_single_record_and_batch_correlation(tmp_path): + with invocation_envelope( + None, + InvocationContext( + job_id="job", dag_node_id="node", invocation_id="invoke", record_id="record" + ), + ): + assert invocation_scratch_dir(tmp_path, create=False) == tmp_path / "job/node/invoke" + with invocation_envelope(None, InvocationContext(record_id="record")): + assert invocation_scratch_dir(tmp_path, create=False).name == "record" + with invocation_envelope(None, InvocationContext(record_id="record", record_ids=["record"])): + first = invocation_scratch_dir(tmp_path, create=False) + second = invocation_scratch_dir(tmp_path, create=False) + assert first != second + assert first.name != "record" + + +def test_concurrent_invocations_are_isolated(tmp_path): + async def invoke(job): + with invocation_envelope( + None, InvocationContext(job_id=job, dag_node_id="node", invocation_id="invoke") + ): + await asyncio.sleep(0) + return current_invocation_identity().job_id, invocation_scratch_dir(tmp_path) + + async def run(): + return await asyncio.gather(invoke("one"), invoke("two")) + + results = asyncio.run(run()) + assert results == [("one", tmp_path / "one/node/invoke"), ("two", tmp_path / "two/node/invoke")] + + +@dataclass +class ScratchOutput: + path: str + + +def test_streaming_retains_invocation_scope(tmp_path): + async def stream() -> ScratchOutput: + await asyncio.sleep(0) + yield ScratchOutput( + str(invocation_scratch_dir(tmp_path) / current_invocation_identity().job_id) + ) + + app = generate_fast_api(stream) + with TestClient(app) as client: + response = client.post( + "/invoke", + json={ + "invocation_context": { + "job_id": "job", + "dag_node_id": "node", + "invocation_id": "invoke", + } + }, + ) + assert response.status_code == 200 + assert str(Path(tmp_path) / "job/node/invoke/job") in response.text diff --git a/unstructured_platform_plugins/__init__.py b/unstructured_platform_plugins/__init__.py index e69de29..1115c35 100644 --- a/unstructured_platform_plugins/__init__.py +++ b/unstructured_platform_plugins/__init__.py @@ -0,0 +1,13 @@ +from unstructured_platform_plugins.invocation_scope import ( + InvocationIdentity, + current_invocation_identity, + invocation_scratch_dir, +) +from unstructured_platform_plugins.invocation_settings import current_invocation_context + +__all__ = [ + "InvocationIdentity", + "current_invocation_context", + "current_invocation_identity", + "invocation_scratch_dir", +] diff --git a/unstructured_platform_plugins/invocation_scope.py b/unstructured_platform_plugins/invocation_scope.py new file mode 100644 index 0000000..a4350da --- /dev/null +++ b/unstructured_platform_plugins/invocation_scope.py @@ -0,0 +1,60 @@ +"""Identity and filesystem scope for the current invocation.""" + +from __future__ import annotations + +import hashlib +import os +import re +from dataclasses import dataclass, fields +from pathlib import Path +from uuid import uuid4 + +from unstructured_platform_plugins.invocation_settings import current_invocation_context + + +@dataclass(frozen=True) +class InvocationIdentity: + job_id: str | None + tenant_id: str | None + org_id: str | None + workflow_id: str | None + attribution_id: str | None + dag_node_id: str | None + dag_node_type: str | None + dag_node_subtype: str | None + + +def current_invocation_identity() -> InvocationIdentity: + """Read bound identity, falling back to the environment for each absent field.""" + context = current_invocation_context() + values = {} + for field in fields(InvocationIdentity): + value = getattr(context, field.name, None) + values[field.name] = os.getenv(field.name.upper()) if value is None else value + return InvocationIdentity(**values) + + +def _segment(value: str | None) -> str: + if not value: + return "unknown" + safe = re.sub(r"[^a-zA-Z0-9_.-]", "_", value) + if safe in {".", ".."} or safe != value or len(safe) > 96: + digest = hashlib.sha256(value.encode()).hexdigest()[:16] + safe = f"{safe[:79].strip('.') or 'unknown'}-{digest}" + return safe + + +def invocation_scratch_dir(root: str | os.PathLike[str], *, create: bool = True) -> Path: + """Return an invocation directory, or the unchanged root when no context is bound.""" + path = Path(root) + context = current_invocation_context() + if context is not None: + identity = current_invocation_identity() + invocation_id = context.invocation_id + if invocation_id is None and context.record_ids is None and context.invocation_ids is None: + invocation_id = context.record_id + path = path / _segment(identity.job_id) / _segment(identity.dag_node_id) + path = path / _segment(invocation_id or uuid4().hex) + if create: + path.mkdir(parents=True, exist_ok=True) + return path From b7281e115fc1fde551136f1525ba6232f99c0bc6 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Wed, 14 Oct 2026 12:09:19 -0400 Subject: [PATCH 2/6] fix: Fall back to record_id when invocation_id is empty --- test/api/test_invocation_scope.py | 2 ++ unstructured_platform_plugins/invocation_scope.py | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/test/api/test_invocation_scope.py b/test/api/test_invocation_scope.py index 291c0f4..3b2bc89 100644 --- a/test/api/test_invocation_scope.py +++ b/test/api/test_invocation_scope.py @@ -74,6 +74,8 @@ def test_single_record_and_batch_correlation(tmp_path): assert invocation_scratch_dir(tmp_path, create=False) == tmp_path / "job/node/invoke" with invocation_envelope(None, InvocationContext(record_id="record")): assert invocation_scratch_dir(tmp_path, create=False).name == "record" + with invocation_envelope(None, InvocationContext(invocation_id="", record_id="record")): + assert invocation_scratch_dir(tmp_path, create=False).name == "record" with invocation_envelope(None, InvocationContext(record_id="record", record_ids=["record"])): first = invocation_scratch_dir(tmp_path, create=False) second = invocation_scratch_dir(tmp_path, create=False) diff --git a/unstructured_platform_plugins/invocation_scope.py b/unstructured_platform_plugins/invocation_scope.py index a4350da..b231c19 100644 --- a/unstructured_platform_plugins/invocation_scope.py +++ b/unstructured_platform_plugins/invocation_scope.py @@ -51,7 +51,7 @@ def invocation_scratch_dir(root: str | os.PathLike[str], *, create: bool = True) if context is not None: identity = current_invocation_identity() invocation_id = context.invocation_id - if invocation_id is None and context.record_ids is None and context.invocation_ids is None: + if not invocation_id and context.record_ids is None and context.invocation_ids is None: invocation_id = context.record_id path = path / _segment(identity.job_id) / _segment(identity.dag_node_id) path = path / _segment(invocation_id or uuid4().hex) From 86f9c1b891ab142203036ba3582500367246f574 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Wed, 14 Oct 2026 12:25:06 -0400 Subject: [PATCH 3/6] docs: Describe invocation scope helpers in the 0.1.1 changelog --- CHANGELOG.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ebbe347..271a1ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,9 @@ ## 0.1.1 -* Add `InvocationIdentity`, `current_invocation_identity()` and `invocation_scratch_dir()` for - request-scoped identity and scratch paths, with environment and root-path fallback in boot mode. +* **Plugins can read job identity and scratch paths per request.** `current_invocation_identity()` + returns an `InvocationIdentity` from the bound invocation context, falling back to the + environment for each absent field. `invocation_scratch_dir(root)` returns + `root///` while a context is bound and `root` unchanged otherwise. * **A provider rate limit no longer reaches the wire as terminal.** `plugin_error_of` declared `retryable = false` for the whole legacy `UserError` family, `RateLimitError` included, so a 429 From 552ab72dac803abd6fff06fe3eba50d6a2636c6c Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Wed, 14 Oct 2026 13:03:29 -0400 Subject: [PATCH 4/6] fix: Keep bound invocation identity authoritative --- CHANGELOG.md | 4 ++-- test/api/test_invocation_scope.py | 6 ++++-- unstructured_platform_plugins/invocation_scope.py | 9 ++++++--- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 271a1ea..7297f3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,8 @@ ## 0.1.1 * **Plugins can read job identity and scratch paths per request.** `current_invocation_identity()` - returns an `InvocationIdentity` from the bound invocation context, falling back to the - environment for each absent field. `invocation_scratch_dir(root)` returns + returns an `InvocationIdentity` exclusively from the bound invocation context; absent fields + are `None`. It falls back to the environment only when no context is bound. `invocation_scratch_dir(root)` returns `root///` while a context is bound and `root` unchanged otherwise. * **A provider rate limit no longer reaches the wire as terminal.** `plugin_error_of` declared diff --git a/test/api/test_invocation_scope.py b/test/api/test_invocation_scope.py index 3b2bc89..239f737 100644 --- a/test/api/test_invocation_scope.py +++ b/test/api/test_invocation_scope.py @@ -16,7 +16,7 @@ @pytest.mark.parametrize("bound", [False, True]) -def test_identity_field_fallback(monkeypatch, bound): +def test_identity_context_is_authoritative(monkeypatch, bound): for field in fields(InvocationIdentity): monkeypatch.setenv(field.name.upper(), f"env-{field.name}") context = InvocationContext(job_id="request", tenant_id="") if bound else None @@ -24,7 +24,9 @@ def test_identity_field_fallback(monkeypatch, bound): identity = current_invocation_identity() assert identity.job_id == ("request" if bound else "env-job_id") assert identity.tenant_id == ("" if bound else "env-tenant_id") - assert identity.org_id == "env-org_id" + for field in fields(InvocationIdentity): + if field.name not in {"job_id", "tenant_id"}: + assert getattr(identity, field.name) == (None if bound else f"env-{field.name}") with pytest.raises(FrozenInstanceError): identity.job_id = "other" diff --git a/unstructured_platform_plugins/invocation_scope.py b/unstructured_platform_plugins/invocation_scope.py index b231c19..7acffbf 100644 --- a/unstructured_platform_plugins/invocation_scope.py +++ b/unstructured_platform_plugins/invocation_scope.py @@ -25,12 +25,15 @@ class InvocationIdentity: def current_invocation_identity() -> InvocationIdentity: - """Read bound identity, falling back to the environment for each absent field.""" + """Read identity from the bound context, or the environment when no context is bound.""" context = current_invocation_context() values = {} for field in fields(InvocationIdentity): - value = getattr(context, field.name, None) - values[field.name] = os.getenv(field.name.upper()) if value is None else value + values[field.name] = ( + getattr(context, field.name, None) + if context is not None + else os.getenv(field.name.upper()) + ) return InvocationIdentity(**values) From 53d9d6aea92772bf166fcfa421fb34591ca08464 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:33:42 -0400 Subject: [PATCH 5/6] fix(invocation): Prevent scratch directory aliases --- test/api/test_invocation_scope.py | 9 +++++++-- unstructured_platform_plugins/invocation_scope.py | 10 +++++----- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/test/api/test_invocation_scope.py b/test/api/test_invocation_scope.py index 239f737..ec31711 100644 --- a/test/api/test_invocation_scope.py +++ b/test/api/test_invocation_scope.py @@ -60,10 +60,15 @@ def test_segments_are_safe_and_bounded(tmp_path, segment): def test_sanitization_does_not_alias(tmp_path): paths = [] - for job_id in ["a/b", "a_b"]: + for job_id in [None, "unknown", "~missing", "a/b", "a_b"]: with invocation_envelope(None, InvocationContext(job_id=job_id, invocation_id="invoke")): paths.append(invocation_scratch_dir(tmp_path, create=False)) - assert paths[0] != paths[1] + encoded_job_id = paths[3].relative_to(tmp_path).parts[0] + with invocation_envelope( + None, InvocationContext(job_id=encoded_job_id, invocation_id="invoke") + ): + paths.append(invocation_scratch_dir(tmp_path, create=False)) + assert len(set(paths)) == len(paths) def test_single_record_and_batch_correlation(tmp_path): diff --git a/unstructured_platform_plugins/invocation_scope.py b/unstructured_platform_plugins/invocation_scope.py index 7acffbf..0ee7bce 100644 --- a/unstructured_platform_plugins/invocation_scope.py +++ b/unstructured_platform_plugins/invocation_scope.py @@ -39,12 +39,12 @@ def current_invocation_identity() -> InvocationIdentity: def _segment(value: str | None) -> str: if not value: - return "unknown" + return "~missing" safe = re.sub(r"[^a-zA-Z0-9_.-]", "_", value) - if safe in {".", ".."} or safe != value or len(safe) > 96: - digest = hashlib.sha256(value.encode()).hexdigest()[:16] - safe = f"{safe[:79].strip('.') or 'unknown'}-{digest}" - return safe + if safe == value and safe not in {".", ".."} and len(safe) <= 96: + return safe + digest = hashlib.sha256(value.encode()).hexdigest() + return f"~{safe[:30].strip('.') or 'unknown'}-{digest}" def invocation_scratch_dir(root: str | os.PathLike[str], *, create: bool = True) -> Path: From 0ebcdb790713e86a965f4bd48c974177b821bbd2 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:18:06 -0400 Subject: [PATCH 6/6] refactor(invocation): Import scope helpers from their module --- test/api/test_invocation_scope.py | 6 +++--- unstructured_platform_plugins/__init__.py | 13 ------------- 2 files changed, 3 insertions(+), 16 deletions(-) diff --git a/test/api/test_invocation_scope.py b/test/api/test_invocation_scope.py index ec31711..b3e857d 100644 --- a/test/api/test_invocation_scope.py +++ b/test/api/test_invocation_scope.py @@ -5,13 +5,13 @@ import pytest from fastapi.testclient import TestClient -from unstructured_platform_plugins import ( +from unstructured_platform_plugins.etl_uvicorn.api_generator import generate_fast_api +from unstructured_platform_plugins.invocation_context import InvocationContext +from unstructured_platform_plugins.invocation_scope import ( InvocationIdentity, current_invocation_identity, invocation_scratch_dir, ) -from unstructured_platform_plugins.etl_uvicorn.api_generator import generate_fast_api -from unstructured_platform_plugins.invocation_context import InvocationContext from unstructured_platform_plugins.invocation_settings import invocation_envelope diff --git a/unstructured_platform_plugins/__init__.py b/unstructured_platform_plugins/__init__.py index 1115c35..e69de29 100644 --- a/unstructured_platform_plugins/__init__.py +++ b/unstructured_platform_plugins/__init__.py @@ -1,13 +0,0 @@ -from unstructured_platform_plugins.invocation_scope import ( - InvocationIdentity, - current_invocation_identity, - invocation_scratch_dir, -) -from unstructured_platform_plugins.invocation_settings import current_invocation_context - -__all__ = [ - "InvocationIdentity", - "current_invocation_context", - "current_invocation_identity", - "invocation_scratch_dir", -]